Compare commits

..

13 Commits

Author SHA1 Message Date
ZiWei
e4e4bfbe20 Merge branch 'dev' into feature/organic-extraction 2026-02-04 15:47:47 +08:00
Xuwznln
84a8223173 adapt to new edge format 2026-02-03 23:22:38 +08:00
Xuwznln
e8d1263488 workflow upload & prcxi transfer liquid 2026-02-03 18:10:32 +08:00
Xuwznln
380b39100d lh liquid 2026-02-03 15:15:57 +08:00
ZiWei
64c748d921 Merge branch 'vibe/dev' into feature/organic-extraction 2026-02-03 10:39:44 +08:00
ZiWei
15ff0e9d30 feat: add Bioyond deck imports to resource registration 2026-02-03 10:28:51 +08:00
ZiWei
f8a52860ad Add BIOYOND deck imports and update JSON configurations with new UUIDs for various components 2026-02-03 10:25:47 +08:00
Xuwznln
e30c01d54e Dev backward (#228)
* Workbench example, adjust log level, and ci check (#220)

* TestLatency Return Value Example & gitignore update

* Adjust log level & Add workbench virtual example & Add not action decorator & Add check_mode &

* Add CI Check

* CI Check Fix 1

* CI Check Fix 2

* CI Check Fix 3

* CI Check Fix 4

* CI Check Fix 5

* Upgrade to py 3.11.14; ros 0.7; unilabos 0.10.16

* Update to ROS2 Humble 0.7

* Fix Build 1

* Fix Build 2

* Fix Build 3

* Fix Build 4

* Fix Build 5

* Fix Build 6

* Fix Build 7

* ci(deps): bump actions/configure-pages from 4 to 5 (#222)

Bumps [actions/configure-pages](https://github.com/actions/configure-pages) from 4 to 5.
- [Release notes](https://github.com/actions/configure-pages/releases)
- [Commits](https://github.com/actions/configure-pages/compare/v4...v5)

---
updated-dependencies:
- dependency-name: actions/configure-pages
  dependency-version: '5'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* ci(deps): bump actions/upload-artifact from 4 to 6 (#224)

Bumps [actions/upload-artifact](https://github.com/actions/upload-artifact) from 4 to 6.
- [Release notes](https://github.com/actions/upload-artifact/releases)
- [Commits](https://github.com/actions/upload-artifact/compare/v4...v6)

---
updated-dependencies:
- dependency-name: actions/upload-artifact
  dependency-version: '6'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* ci(deps): bump actions/upload-pages-artifact from 3 to 4 (#225)

Bumps [actions/upload-pages-artifact](https://github.com/actions/upload-pages-artifact) from 3 to 4.
- [Release notes](https://github.com/actions/upload-pages-artifact/releases)
- [Commits](https://github.com/actions/upload-pages-artifact/compare/v3...v4)

---
updated-dependencies:
- dependency-name: actions/upload-pages-artifact
  dependency-version: '4'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* ci(deps): bump actions/checkout from 4 to 6 (#223)

Bumps [actions/checkout](https://github.com/actions/checkout) from 4 to 6.
- [Release notes](https://github.com/actions/checkout/releases)
- [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md)
- [Commits](https://github.com/actions/checkout/compare/v4...v6)

---
updated-dependencies:
- dependency-name: actions/checkout
  dependency-version: '6'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* Fix Build 8

* Fix Build 9

* Fix Build 10

* Fix Build 11

* Fix Build 12

* Fix Build 13

* v0.10.17

(cherry picked from commit 176de521b4)

* CI Check use production mode

* Fix OT2 & ReAdd Virtual Devices

* add msg goal

* transfer liquid handles

* gather query

* add unilabos_class

* Support root node change pos

* save class name when deserialize & protocol execute test

* fix upload workflow json

* workflow upload & set liquid fix & add set liquid with plate

* speed up registry load

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: hanhua@dp.tech <2509856570@qq.com>
2026-02-02 23:57:13 +08:00
ZiWei
37ec49f318 Refactor Bioyond resource handling: update warehouse mapping retrieval, add TipBox support, and improve liquid tracking logic. Migrate TipBox creation to bottle_carriers.py for better structure. 2026-01-29 16:31:14 +08:00
ZiWei
6bf57f18c1 Collaboration With Cursor 2026-01-29 11:29:38 +08:00
ZiWei
c4a3be1498 feat: enhance separation_step logic with polling thread management and error handling 2026-01-27 12:37:09 +08:00
ZiWei
e11070315d feat: add separation_step with sensor-motor linkage 2026-01-26 23:34:47 +08:00
ZiWei
50ebcad9d7 feat: add ZDT_X42 motor and XKC sensor drivers 2026-01-22 15:07:32 +08:00
31 changed files with 4674 additions and 268 deletions

View File

@@ -46,7 +46,7 @@ requirements:
- jinja2 - jinja2
- requests - requests
- uvicorn - uvicorn
- opcua - opcua # [not osx]
- pyserial - pyserial
- pandas - pandas
- pymodbus - pymodbus

View File

@@ -0,0 +1,328 @@
---
description: 设备驱动开发规范
globs: ["unilabos/devices/**/*.py"]
---
# 设备驱动开发规范
## 目录结构
```
unilabos/devices/
├── virtual/ # 虚拟设备(用于测试)
│ ├── virtual_stirrer.py
│ └── virtual_centrifuge.py
├── liquid_handling/ # 液体处理设备
├── balance/ # 天平设备
├── hplc/ # HPLC设备
├── pump_and_valve/ # 泵和阀门
├── temperature/ # 温度控制设备
├── workstation/ # 工作站(组合设备)
└── ...
```
## 设备类完整模板
```python
import asyncio
import logging
import time as time_module
from typing import Dict, Any, Optional
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
class MyDevice:
"""
设备类描述
Attributes:
device_id: 设备唯一标识
config: 设备配置字典
data: 设备状态数据
"""
_ros_node: BaseROS2DeviceNode
def __init__(
self,
device_id: str = None,
config: Dict[str, Any] = None,
**kwargs
):
"""
初始化设备
Args:
device_id: 设备ID
config: 配置字典
**kwargs: 其他参数
"""
# 兼容不同调用方式
if device_id is None and 'id' in kwargs:
device_id = kwargs.pop('id')
if config is None and 'config' in kwargs:
config = kwargs.pop('config')
self.device_id = device_id or "unknown_device"
self.config = config or {}
self.data = {}
# 从config读取参数
self.port = self.config.get('port') or kwargs.get('port', 'COM1')
self._max_value = self.config.get('max_value', 1000.0)
# 初始化日志
self.logger = logging.getLogger(f"MyDevice.{self.device_id}")
self.logger.info(f"设备 {self.device_id} 已创建")
def post_init(self, ros_node: BaseROS2DeviceNode):
"""
ROS节点注入 - 在ROS节点创建后调用
Args:
ros_node: ROS2设备节点实例
"""
self._ros_node = ros_node
async def initialize(self) -> bool:
"""
初始化设备 - 连接硬件、设置初始状态
Returns:
bool: 初始化是否成功
"""
self.logger.info(f"初始化设备 {self.device_id}")
try:
# 执行硬件初始化
# await self._connect_hardware()
# 设置初始状态
self.data.update({
"status": "待机",
"is_running": False,
"current_value": 0.0,
})
self.logger.info(f"设备 {self.device_id} 初始化完成")
return True
except Exception as e:
self.logger.error(f"初始化失败: {e}")
self.data["status"] = f"错误: {e}"
return False
async def cleanup(self) -> bool:
"""
清理设备 - 断开连接、释放资源
Returns:
bool: 清理是否成功
"""
self.logger.info(f"清理设备 {self.device_id}")
self.data.update({
"status": "离线",
"is_running": False,
})
return True
# ==================== 设备动作 ====================
async def execute_action(
self,
param1: float,
param2: str = "",
**kwargs
) -> bool:
"""
执行设备动作
Args:
param1: 参数1
param2: 参数2可选
Returns:
bool: 动作是否成功
"""
# 类型转换和验证
try:
param1 = float(param1)
except (ValueError, TypeError) as e:
self.logger.error(f"参数类型错误: {e}")
return False
# 参数验证
if param1 > self._max_value:
self.logger.error(f"参数超出范围: {param1} > {self._max_value}")
return False
self.logger.info(f"执行动作: param1={param1}, param2={param2}")
# 更新状态
self.data.update({
"status": "运行中",
"is_running": True,
})
# 执行动作(带进度反馈)
duration = 10.0 # 秒
start_time = time_module.time()
while True:
elapsed = time_module.time() - start_time
remaining = max(0, duration - elapsed)
progress = min(100, (elapsed / duration) * 100)
self.data.update({
"status": f"运行中: {progress:.0f}%",
"remaining_time": remaining,
})
if remaining <= 0:
break
await self._ros_node.sleep(1.0)
# 完成
self.data.update({
"status": "完成",
"is_running": False,
})
self.logger.info("动作执行完成")
return True
# ==================== 状态属性 ====================
@property
def status(self) -> str:
"""设备状态 - 自动发布为ROS Topic"""
return self.data.get("status", "未知")
@property
def is_running(self) -> bool:
"""是否正在运行"""
return self.data.get("is_running", False)
@property
def current_value(self) -> float:
"""当前值"""
return self.data.get("current_value", 0.0)
# ==================== 辅助方法 ====================
def get_device_info(self) -> Dict[str, Any]:
"""获取设备信息"""
return {
"device_id": self.device_id,
"status": self.status,
"is_running": self.is_running,
"current_value": self.current_value,
}
def __str__(self) -> str:
return f"MyDevice({self.device_id}: {self.status})"
```
## 关键规则
### 1. 参数处理
所有动作方法的参数都可能以字符串形式传入,必须进行类型转换:
```python
async def my_action(self, value: float, **kwargs) -> bool:
# 始终进行类型转换
try:
value = float(value)
except (ValueError, TypeError) as e:
self.logger.error(f"参数类型错误: {e}")
return False
```
### 2. vessel 参数处理
vessel 参数可能是字符串ID或字典
```python
def extract_vessel_id(vessel: Union[str, dict]) -> str:
if isinstance(vessel, dict):
return vessel.get("id", "")
return str(vessel) if vessel else ""
```
### 3. 状态更新
使用 `self.data` 字典存储状态,属性读取状态:
```python
# 更新状态
self.data["status"] = "运行中"
self.data["current_speed"] = 300.0
# 读取状态(通过属性)
@property
def status(self) -> str:
return self.data.get("status", "待机")
```
### 4. 异步等待
使用 ROS 节点的 sleep 方法:
```python
# 正确
await self._ros_node.sleep(1.0)
# 避免(除非在纯 Python 测试环境)
await asyncio.sleep(1.0)
```
### 5. 进度反馈
长时间运行的操作需要提供进度反馈:
```python
while remaining > 0:
progress = (elapsed / total_time) * 100
self.data["status"] = f"运行中: {progress:.0f}%"
self.data["remaining_time"] = remaining
await self._ros_node.sleep(1.0)
```
## 虚拟设备
虚拟设备用于测试和演示,放在 `unilabos/devices/virtual/` 目录:
- 类名以 `Virtual` 开头
- 文件名以 `virtual_` 开头
- 模拟真实设备的行为和时序
- 使用表情符号增强日志可读性(可选)
## 工作站设备
工作站是组合多个设备的复杂设备:
```python
from unilabos.devices.workstation.workstation_base import WorkstationBase
class MyWorkstation(WorkstationBase):
"""组合工作站"""
async def execute_workflow(self, workflow: Dict[str, Any]) -> bool:
"""执行工作流"""
pass
```
## 设备注册
设备类开发完成后,需要在注册表中注册:
1. 创建/编辑 `unilabos/registry/devices/my_category.yaml`
2. 添加设备配置(参考 `virtual_device.yaml`
3. 运行 `--complete_registry` 自动生成 schema

View File

@@ -0,0 +1,240 @@
---
description: 协议编译器开发规范
globs: ["unilabos/compile/**/*.py"]
---
# 协议编译器开发规范
## 概述
协议编译器负责将高级实验操作(如 Stir、Add、Filter编译为设备可执行的动作序列。
## 文件命名
- 位置: `unilabos/compile/`
- 命名: `{operation}_protocol.py`
- 示例: `stir_protocol.py`, `add_protocol.py`, `filter_protocol.py`
## 协议函数模板
```python
from typing import List, Dict, Any, Union
import networkx as nx
import logging
from .utils.unit_parser import parse_time_input
from .utils.vessel_parser import extract_vessel_id
logger = logging.getLogger(__name__)
def generate_{operation}_protocol(
G: nx.DiGraph,
vessel: Union[str, dict],
param1: Union[str, float] = "0",
param2: float = 0.0,
**kwargs
) -> List[Dict[str, Any]]:
"""
生成{操作}协议序列
Args:
G: 物理拓扑图 (NetworkX DiGraph)
vessel: 容器ID或Resource字典
param1: 参数1支持字符串单位如 "5 min"
param2: 参数2
**kwargs: 其他参数
Returns:
List[Dict]: 动作序列
Raises:
ValueError: 参数无效时
"""
# 1. 提取 vessel_id
vessel_id = extract_vessel_id(vessel)
# 2. 验证参数
if not vessel_id:
raise ValueError("vessel 参数不能为空")
if vessel_id not in G.nodes():
raise ValueError(f"容器 '{vessel_id}' 不存在于系统中")
# 3. 解析参数(支持单位)
parsed_param1 = parse_time_input(param1) # "5 min" -> 300.0
# 4. 查找设备
device_id = find_connected_device(G, vessel_id, device_type="my_device")
# 5. 生成动作序列
action_sequence = []
action = {
"device_id": device_id,
"action_name": "my_action",
"action_kwargs": {
"vessel": {"id": vessel_id}, # 始终使用字典格式
"param1": float(parsed_param1),
"param2": float(param2),
}
}
action_sequence.append(action)
logger.info(f"生成协议: {len(action_sequence)} 个动作")
return action_sequence
def find_connected_device(
G: nx.DiGraph,
vessel_id: str,
device_type: str = ""
) -> str:
"""
查找与容器相连的设备
Args:
G: 拓扑图
vessel_id: 容器ID
device_type: 设备类型关键字
Returns:
str: 设备ID
"""
# 查找所有匹配类型的设备
device_nodes = []
for node in G.nodes():
node_class = G.nodes[node].get('class', '') or ''
if device_type.lower() in node_class.lower():
device_nodes.append(node)
# 检查连接
if vessel_id and device_nodes:
for device in device_nodes:
if G.has_edge(device, vessel_id) or G.has_edge(vessel_id, device):
return device
# 返回第一个可用设备
if device_nodes:
return device_nodes[0]
# 默认设备
return f"{device_type}_1"
```
## 关键规则
### 1. vessel 参数处理
vessel 参数可能是字符串或字典,需要统一处理:
```python
def extract_vessel_id(vessel: Union[str, dict]) -> str:
"""提取vessel_id"""
if isinstance(vessel, dict):
# 可能是 {"id": "xxx"} 或完整 Resource 对象
return vessel.get("id", list(vessel.values())[0].get("id", ""))
return str(vessel) if vessel else ""
```
### 2. action_kwargs 中的 vessel
始终使用 `{"id": vessel_id}` 格式传递 vessel
```python
# 正确
"action_kwargs": {
"vessel": {"id": vessel_id}, # 字符串ID包装为字典
}
# 避免
"action_kwargs": {
"vessel": vessel_resource, # 不要传递完整 Resource 对象
}
```
### 3. 单位解析
使用 `parse_time_input` 解析时间参数:
```python
from .utils.unit_parser import parse_time_input
# 支持格式: "5 min", "1 h", "300", "1.5 hours"
time_seconds = parse_time_input("5 min") # -> 300.0
time_seconds = parse_time_input(120) # -> 120.0
time_seconds = parse_time_input("1 h") # -> 3600.0
```
### 4. 参数验证
所有参数必须进行验证和类型转换:
```python
# 验证范围
if speed < 10.0 or speed > 1500.0:
logger.warning(f"速度 {speed} 超出范围,修正为 300")
speed = 300.0
# 类型转换
param = float(param) if not isinstance(param, (int, float)) else param
```
### 5. 日志记录
使用项目日志记录器:
```python
logger = logging.getLogger(__name__)
def generate_protocol(...):
logger.info(f"开始生成协议...")
logger.debug(f"参数: vessel={vessel_id}, time={time}")
logger.warning(f"参数修正: {old_value} -> {new_value}")
```
## 便捷函数
为常用操作提供便捷函数:
```python
def stir_briefly(G: nx.DiGraph, vessel: Union[str, dict],
speed: float = 300.0) -> List[Dict[str, Any]]:
"""短时间搅拌30秒"""
return generate_stir_protocol(G, vessel, time="30", stir_speed=speed)
def stir_vigorously(G: nx.DiGraph, vessel: Union[str, dict],
time: str = "5 min") -> List[Dict[str, Any]]:
"""剧烈搅拌"""
return generate_stir_protocol(G, vessel, time=time, stir_speed=800.0)
```
## 测试函数
每个协议文件应包含测试函数:
```python
def test_{operation}_protocol():
"""测试协议生成"""
# 测试参数处理
vessel_dict = {"id": "flask_1", "name": "反应瓶1"}
vessel_id = extract_vessel_id(vessel_dict)
assert vessel_id == "flask_1"
# 测试单位解析
time_s = parse_time_input("5 min")
assert time_s == 300.0
if __name__ == "__main__":
test_{operation}_protocol()
```
## 现有协议参考
- `stir_protocol.py` - 搅拌操作
- `add_protocol.py` - 添加物料
- `filter_protocol.py` - 过滤操作
- `heatchill_protocol.py` - 加热/冷却
- `separate_protocol.py` - 分离操作
- `evaporate_protocol.py` - 蒸发操作

View File

@@ -0,0 +1,319 @@
---
description: 注册表配置规范 (YAML)
globs: ["unilabos/registry/**/*.yaml"]
---
# 注册表配置规范
## 概述
注册表使用 YAML 格式定义设备和资源类型,是 Uni-Lab-OS 的核心配置系统。
## 目录结构
```
unilabos/registry/
├── devices/ # 设备类型注册
│ ├── virtual_device.yaml
│ ├── liquid_handler.yaml
│ └── ...
├── device_comms/ # 通信设备配置
│ ├── communication_devices.yaml
│ └── modbus_ioboard.yaml
└── resources/ # 资源类型注册
├── bioyond/
├── organic/
├── opentrons/
└── ...
```
## 设备注册表格式
### 基本结构
```yaml
device_type_id:
# 基本信息
description: "设备描述"
version: "1.0.0"
category:
- category_name
icon: "icon_device.webp"
# 类配置
class:
module: "unilabos.devices.my_module:MyClass"
type: python
# 状态类型(属性 -> ROS消息类型
status_types:
status: String
temperature: Float64
is_running: Bool
# 动作映射
action_value_mappings:
action_name:
type: UniLabJsonCommand # 或 UniLabJsonCommandAsync
goal: {}
feedback: {}
result: {}
schema: {...}
handles: {}
```
### action_value_mappings 详细格式
```yaml
action_value_mappings:
# 同步动作
my_sync_action:
type: UniLabJsonCommand
goal:
param1: param1
param2: param2
feedback: {}
result:
success: success
message: message
goal_default:
param1: 0.0
param2: ""
handles: {}
placeholder_keys:
device_param: unilabos_devices # 设备选择器
resource_param: unilabos_resources # 资源选择器
schema:
title: "动作名称参数"
description: "动作描述"
type: object
properties:
goal:
type: object
properties:
param1:
type: number
param2:
type: string
required:
- param1
feedback: {}
result:
type: object
properties:
success:
type: boolean
message:
type: string
required:
- goal
# 异步动作
my_async_action:
type: UniLabJsonCommandAsync
goal: {}
feedback:
progress: progress
current_status: status
result:
success: success
schema: {...}
```
### 自动生成的动作
以 `auto-` 开头的动作由系统自动生成:
```yaml
action_value_mappings:
auto-initialize:
type: UniLabJsonCommandAsync
goal: {}
feedback: {}
result: {}
schema: {...}
auto-cleanup:
type: UniLabJsonCommandAsync
goal: {}
feedback: {}
result: {}
schema: {...}
```
### handles 配置
用于工作流编辑器中的数据流连接:
```yaml
handles:
input:
- handler_key: "input_resource"
data_type: "resource"
label: "输入资源"
data_source: "handle"
data_key: "resources"
output:
- handler_key: "output_labware"
data_type: "resource"
label: "输出器皿"
data_source: "executor"
data_key: "created_resource.@flatten"
```
## 资源注册表格式
```yaml
resource_type_id:
description: "资源描述"
version: "1.0.0"
category:
- category_name
icon: ""
handles: []
init_param_schema: {}
class:
module: "unilabos.resources.my_module:MyResource"
type: pylabrobot # 或 python
```
### PyLabRobot 资源示例
```yaml
BIOYOND_Electrolyte_6VialCarrier:
category:
- bottle_carriers
- bioyond
class:
module: "unilabos.resources.bioyond.bottle_carriers:BIOYOND_Electrolyte_6VialCarrier"
type: pylabrobot
version: "1.0.0"
```
## 状态类型映射
Python 类型到 ROS 消息类型的映射:
| Python 类型 | ROS 消息类型 |
|------------|-------------|
| `str` | `String` |
| `bool` | `Bool` |
| `int` | `Int64` |
| `float` | `Float64` |
| `list` | `String` (序列化) |
| `dict` | `String` (序列化) |
## 自动完善注册表
使用 `--complete_registry` 参数自动生成 schema
```bash
python -m unilabos.app.main --complete_registry
```
这会:
1. 扫描设备类的方法签名
2. 自动生成 `auto-` 前缀的动作
3. 生成 JSON Schema
4. 更新 YAML 文件
## 验证规则
1. **device_type_id** 必须唯一
2. **module** 路径必须正确可导入
3. **status_types** 的类型必须是有效的 ROS 消息类型
4. **schema** 必须是有效的 JSON Schema
## 示例:完整设备配置
```yaml
virtual_stirrer:
category:
- virtual_device
description: "虚拟搅拌器设备"
version: "1.0.0"
icon: "icon_stirrer.webp"
handles: []
init_param_schema: {}
class:
module: "unilabos.devices.virtual.virtual_stirrer:VirtualStirrer"
type: python
status_types:
status: String
operation_mode: String
current_speed: Float64
is_stirring: Bool
remaining_time: Float64
action_value_mappings:
auto-initialize:
type: UniLabJsonCommandAsync
goal: {}
feedback: {}
result: {}
schema:
title: "initialize参数"
type: object
properties:
goal:
type: object
properties: {}
feedback: {}
result: {}
required:
- goal
stir:
type: UniLabJsonCommandAsync
goal:
stir_time: stir_time
stir_speed: stir_speed
settling_time: settling_time
feedback:
current_speed: current_speed
remaining_time: remaining_time
result:
success: success
goal_default:
stir_time: 60.0
stir_speed: 300.0
settling_time: 30.0
handles: {}
schema:
title: "stir参数"
description: "搅拌操作"
type: object
properties:
goal:
type: object
properties:
stir_time:
type: number
description: "搅拌时间(秒)"
stir_speed:
type: number
description: "搅拌速度RPM"
settling_time:
type: number
description: "沉降时间(秒)"
required:
- stir_time
- stir_speed
feedback:
type: object
properties:
current_speed:
type: number
remaining_time:
type: number
result:
type: object
properties:
success:
type: boolean
required:
- goal
```

View File

@@ -0,0 +1,233 @@
---
description: ROS 2 集成开发规范
globs: ["unilabos/ros/**/*.py", "**/*_node.py"]
---
# ROS 2 集成开发规范
## 概述
Uni-Lab-OS 使用 ROS 2 作为设备通信中间件,基于 rclpy 实现。
## 核心组件
### BaseROS2DeviceNode
设备节点基类,提供:
- ROS Topic 自动发布(状态属性)
- Action Server 自动创建(设备动作)
- 资源管理服务
- 异步任务调度
```python
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
```
### 消息转换器
```python
from unilabos.ros.msgs.message_converter import (
convert_to_ros_msg,
convert_from_ros_msg_with_mapping,
msg_converter_manager,
ros_action_to_json_schema,
ros_message_to_json_schema,
)
```
## 设备与 ROS 集成
### post_init 方法
设备类必须实现 `post_init` 方法接收 ROS 节点:
```python
class MyDevice:
_ros_node: BaseROS2DeviceNode
def post_init(self, ros_node: BaseROS2DeviceNode):
"""ROS节点注入"""
self._ros_node = ros_node
```
### 状态属性发布
设备的 `@property` 属性会自动发布为 ROS Topic
```python
class MyDevice:
@property
def temperature(self) -> float:
return self._temperature
# 自动发布到 /{namespace}/temperature Topic
```
### Topic 配置装饰器
```python
from unilabos.utils.decorator import topic_config
class MyDevice:
@property
@topic_config(period=1.0, print_publish=False, qos=10)
def fast_data(self) -> float:
"""高频数据 - 每秒发布一次"""
return self._fast_data
@property
@topic_config(period=5.0)
def slow_data(self) -> str:
"""低频数据 - 每5秒发布一次"""
return self._slow_data
```
### 订阅装饰器
```python
from unilabos.utils.decorator import subscribe
class MyDevice:
@subscribe(topic="/external/sensor_data", qos=10)
def on_sensor_data(self, msg):
"""订阅外部Topic"""
self._sensor_value = msg.data
```
## 异步操作
### 使用 ROS 节点睡眠
```python
# 推荐使用ROS节点的睡眠方法
await self._ros_node.sleep(1.0)
# 不推荐直接使用asyncio可能导致回调阻塞
await asyncio.sleep(1.0)
```
### 获取事件循环
```python
from unilabos.ros.x.rclpyx import get_event_loop
loop = get_event_loop()
```
## 消息类型
### unilabos_msgs 包
```python
from unilabos_msgs.msg import Resource
from unilabos_msgs.srv import (
ResourceAdd,
ResourceDelete,
ResourceUpdate,
ResourceList,
SerialCommand,
)
from unilabos_msgs.action import SendCmd
```
### Resource 消息结构
```python
Resource:
id: str
name: str
category: str
type: str
parent: str
children: List[str]
config: str # JSON字符串
data: str # JSON字符串
sample_id: str
pose: Pose
```
## 日志适配器
```python
from unilabos.utils.log import info, debug, warning, error, trace
class MyDevice:
def __init__(self):
# 创建设备专属日志器
self.logger = logging.getLogger(f"MyDevice.{self.device_id}")
```
ROSLoggerAdapter 同时向自定义日志和 ROS 日志发送消息。
## Action Server
设备动作自动创建为 ROS Action Server
```yaml
# 在注册表中配置
action_value_mappings:
my_action:
type: UniLabJsonCommandAsync # 异步Action
goal: {...}
feedback: {...}
result: {...}
```
### Action 类型
- **UniLabJsonCommand**: 同步动作
- **UniLabJsonCommandAsync**: 异步动作支持feedback
## 服务客户端
```python
from rclpy.client import Client
# 调用其他节点的服务
response = await self._ros_node.call_service(
service_name="/other_node/service",
request=MyServiceRequest(...)
)
```
## 命名空间
设备节点使用命名空间隔离:
```
/{device_id}/ # 设备命名空间
/{device_id}/status # 状态Topic
/{device_id}/temperature # 温度Topic
/{device_id}/my_action # 动作Server
```
## 调试
### 查看 Topic
```bash
ros2 topic list
ros2 topic echo /{device_id}/status
```
### 查看 Action
```bash
ros2 action list
ros2 action info /{device_id}/my_action
```
### 查看 Service
```bash
ros2 service list
ros2 service call /{device_id}/resource_list unilabos_msgs/srv/ResourceList
```
## 最佳实践
1. **状态属性命名**: 使用蛇形命名法snake_case
2. **Topic 频率**: 根据数据变化频率调整,避免过高频率
3. **Action 反馈**: 长时间操作提供进度反馈
4. **错误处理**: 使用 try-except 捕获并记录错误
5. **资源清理**: 在 cleanup 方法中正确清理资源

View File

@@ -0,0 +1,357 @@
---
description: 测试开发规范
globs: ["tests/**/*.py", "**/test_*.py"]
---
# 测试开发规范
## 目录结构
```
tests/
├── __init__.py
├── devices/ # 设备测试
│ └── liquid_handling/
│ └── test_transfer_liquid.py
├── resources/ # 资源测试
│ ├── test_bottle_carrier.py
│ └── test_resourcetreeset.py
├── ros/ # ROS消息测试
│ └── msgs/
│ ├── test_basic.py
│ ├── test_conversion.py
│ └── test_mapping.py
└── workflow/ # 工作流测试
└── merge_workflow.py
```
## 测试框架
使用 pytest 作为测试框架:
```bash
# 运行所有测试
pytest tests/
# 运行特定测试文件
pytest tests/resources/test_bottle_carrier.py
# 运行特定测试函数
pytest tests/resources/test_bottle_carrier.py::test_bottle_carrier
# 显示详细输出
pytest -v tests/
# 显示打印输出
pytest -s tests/
```
## 测试文件模板
```python
import pytest
from typing import List, Dict, Any
# 导入被测试的模块
from unilabos.resources.bioyond.bottle_carriers import (
BIOYOND_Electrolyte_6VialCarrier,
)
from unilabos.resources.bioyond.bottles import (
BIOYOND_PolymerStation_Solid_Vial,
)
class TestBottleCarrier:
"""BottleCarrier 测试类"""
def setup_method(self):
"""每个测试方法前执行"""
self.carrier = BIOYOND_Electrolyte_6VialCarrier("test_carrier")
def teardown_method(self):
"""每个测试方法后执行"""
pass
def test_carrier_creation(self):
"""测试载架创建"""
assert self.carrier.name == "test_carrier"
assert len(self.carrier.sites) == 6
def test_bottle_placement(self):
"""测试瓶子放置"""
bottle = BIOYOND_PolymerStation_Solid_Vial("test_bottle")
# 测试逻辑...
assert bottle.name == "test_bottle"
def test_standalone_function():
"""独立测试函数"""
result = some_function()
assert result is True
# 参数化测试
@pytest.mark.parametrize("input,expected", [
("5 min", 300.0),
("1 h", 3600.0),
("120", 120.0),
(60, 60.0),
])
def test_time_parsing(input, expected):
"""测试时间解析"""
from unilabos.compile.utils.unit_parser import parse_time_input
assert parse_time_input(input) == expected
# 异常测试
def test_invalid_input_raises_error():
"""测试无效输入抛出异常"""
with pytest.raises(ValueError) as exc_info:
invalid_function("bad_input")
assert "invalid" in str(exc_info.value).lower()
# 跳过条件测试
@pytest.mark.skipif(
not os.environ.get("ROS_DISTRO"),
reason="需要ROS环境"
)
def test_ros_feature():
"""需要ROS环境的测试"""
pass
```
## 设备测试
### 虚拟设备测试
```python
import pytest
import asyncio
from unittest.mock import MagicMock, AsyncMock
from unilabos.devices.virtual.virtual_stirrer import VirtualStirrer
class TestVirtualStirrer:
"""VirtualStirrer 测试"""
@pytest.fixture
def stirrer(self):
"""创建测试用搅拌器"""
device = VirtualStirrer(
device_id="test_stirrer",
config={"max_speed": 1500.0, "min_speed": 50.0}
)
# Mock ROS节点
mock_node = MagicMock()
mock_node.sleep = AsyncMock(return_value=None)
device.post_init(mock_node)
return device
@pytest.mark.asyncio
async def test_initialize(self, stirrer):
"""测试初始化"""
result = await stirrer.initialize()
assert result is True
assert stirrer.status == "待机中"
@pytest.mark.asyncio
async def test_stir_action(self, stirrer):
"""测试搅拌动作"""
await stirrer.initialize()
result = await stirrer.stir(
stir_time=5.0,
stir_speed=300.0,
settling_time=2.0
)
assert result is True
assert stirrer.operation_mode == "Completed"
@pytest.mark.asyncio
async def test_stir_invalid_speed(self, stirrer):
"""测试无效速度"""
await stirrer.initialize()
# 速度超出范围
result = await stirrer.stir(
stir_time=5.0,
stir_speed=2000.0, # 超过max_speed
settling_time=0.0
)
assert result is False
assert "错误" in stirrer.status
```
### 异步测试配置
```python
# conftest.py
import pytest
import asyncio
@pytest.fixture(scope="session")
def event_loop():
"""创建事件循环"""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
```
## 资源测试
```python
import pytest
from unilabos.resources.resource_tracker import (
ResourceTreeSet,
ResourceTreeInstance,
)
def test_resource_tree_creation():
"""测试资源树创建"""
tree_set = ResourceTreeSet()
# 添加资源
resource = {"id": "res_1", "name": "Resource 1"}
tree_set.add_resource(resource)
# 验证
assert len(tree_set.all_nodes) == 1
assert tree_set.get_resource("res_1") is not None
def test_resource_tree_merge():
"""测试资源树合并"""
local_set = ResourceTreeSet()
remote_set = ResourceTreeSet()
# 设置数据...
local_set.merge_remote_resources(remote_set)
# 验证合并结果...
```
## ROS 消息测试
```python
import pytest
from unilabos.ros.msgs.message_converter import (
convert_to_ros_msg,
convert_from_ros_msg_with_mapping,
msg_converter_manager,
)
def test_message_conversion():
"""测试消息转换"""
# Python -> ROS
python_data = {"id": "test", "value": 42}
ros_msg = convert_to_ros_msg(python_data, MyMsgType)
assert ros_msg.id == "test"
assert ros_msg.value == 42
# ROS -> Python
result = convert_from_ros_msg_with_mapping(ros_msg, mapping)
assert result["id"] == "test"
```
## 协议测试
```python
import pytest
import networkx as nx
from unilabos.compile.stir_protocol import (
generate_stir_protocol,
extract_vessel_id,
)
@pytest.fixture
def topology_graph():
"""创建测试拓扑图"""
G = nx.DiGraph()
G.add_node("flask_1", **{"class": "flask"})
G.add_node("stirrer_1", **{"class": "virtual_stirrer"})
G.add_edge("stirrer_1", "flask_1")
return G
def test_generate_stir_protocol(topology_graph):
"""测试搅拌协议生成"""
actions = generate_stir_protocol(
G=topology_graph,
vessel="flask_1",
time="5 min",
stir_speed=300.0
)
assert len(actions) == 1
assert actions[0]["device_id"] == "stirrer_1"
assert actions[0]["action_name"] == "stir"
def test_extract_vessel_id():
"""测试vessel_id提取"""
# 字典格式
assert extract_vessel_id({"id": "flask_1"}) == "flask_1"
# 字符串格式
assert extract_vessel_id("flask_2") == "flask_2"
# 空值
assert extract_vessel_id("") == ""
```
## 测试标记
```python
# 慢速测试
@pytest.mark.slow
def test_long_running():
pass
# 需要网络
@pytest.mark.network
def test_network_call():
pass
# 需要ROS
@pytest.mark.ros
def test_ros_feature():
pass
```
运行特定标记的测试:
```bash
pytest -m "not slow" # 排除慢速测试
pytest -m ros # 仅ROS测试
```
## 覆盖率
```bash
# 生成覆盖率报告
pytest --cov=unilabos tests/
# HTML报告
pytest --cov=unilabos --cov-report=html tests/
```
## 最佳实践
1. **测试命名**: `test_{功能}_{场景}_{预期结果}`
2. **独立性**: 每个测试独立运行,不依赖其他测试
3. **Mock外部依赖**: 使用 unittest.mock 模拟外部服务
4. **参数化**: 使用 `@pytest.mark.parametrize` 减少重复代码
5. **fixtures**: 使用 fixtures 共享测试设置
6. **断言清晰**: 每个断言只验证一件事

View File

@@ -0,0 +1,353 @@
---
description: Uni-Lab-OS 实验室自动化平台开发规范 - 核心规则
globs: ["**/*.py", "**/*.yaml", "**/*.json"]
---
# Uni-Lab-OS 项目开发规范
## 项目概述
Uni-Lab-OS 是一个实验室自动化操作系统,用于连接和控制各种实验设备,实现实验工作流的自动化和标准化。
## 技术栈
- **Python 3.11** - 核心开发语言
- **ROS 2** - 设备通信中间件 (rclpy)
- **Conda/Mamba** - 包管理 (robostack-staging, conda-forge)
- **FastAPI** - Web API 服务
- **WebSocket** - 实时通信
- **NetworkX** - 拓扑图管理
- **YAML** - 配置和注册表定义
- **PyLabRobot** - 实验室自动化库集成
- **pytest** - 测试框架
- **asyncio** - 异步编程
## 项目结构
```
unilabos/
├── app/ # 应用入口、Web服务、后端
├── compile/ # 协议编译器 (stir, add, filter 等)
├── config/ # 配置管理
├── devices/ # 设备驱动 (真实/虚拟)
├── device_comms/ # 设备通信协议
├── device_mesh/ # 3D网格和可视化
├── registry/ # 设备和资源类型注册表 (YAML)
├── resources/ # 资源定义
├── ros/ # ROS 2 集成
├── utils/ # 工具函数
└── workflow/ # 工作流管理
```
## 代码规范
### Python 风格
1. **类型注解**:所有函数必须使用类型注解
```python
def transfer_liquid(
source: str,
destination: str,
volume: float,
**kwargs
) -> List[Dict[str, Any]]:
```
2. **Docstring**:使用 Google 风格的文档字符串
```python
def initialize(self) -> bool:
"""
初始化设备
Returns:
bool: 初始化是否成功
"""
```
3. **导入顺序**
- 标准库
- 第三方库
- ROS 相关 (rclpy, unilabos_msgs)
- 项目内部模块
### 异步编程
1. 设备操作方法使用 `async def`
2. 使用 `await self._ros_node.sleep()` 而非 `asyncio.sleep()`
3. 长时间运行操作需提供进度反馈
```python
async def stir(self, stir_time: float, stir_speed: float, **kwargs) -> bool:
"""执行搅拌操作"""
start_time = time_module.time()
while True:
elapsed = time_module.time() - start_time
remaining = max(0, stir_time - elapsed)
self.data.update({
"remaining_time": remaining,
"status": f"搅拌中: {stir_speed} RPM"
})
if remaining <= 0:
break
await self._ros_node.sleep(1.0)
return True
```
### 日志规范
使用项目自定义日志系统:
```python
from unilabos.utils.log import logger, info, debug, warning, error, trace
# 在设备类中使用
self.logger = logging.getLogger(f"DeviceName.{self.device_id}")
self.logger.info("设备初始化完成")
```
## 设备驱动开发
### 设备类结构
```python
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
class MyDevice:
"""设备驱动类"""
_ros_node: BaseROS2DeviceNode
def __init__(self, device_id: str = None, config: Dict[str, Any] = None, **kwargs):
self.device_id = device_id or "unknown_device"
self.config = config or {}
self.data = {} # 设备状态数据
def post_init(self, ros_node: BaseROS2DeviceNode):
"""ROS节点注入"""
self._ros_node = ros_node
async def initialize(self) -> bool:
"""初始化设备"""
pass
async def cleanup(self) -> bool:
"""清理设备"""
pass
# 状态属性 - 自动发布为 ROS Topic
@property
def status(self) -> str:
return self.data.get("status", "待机")
```
### 状态属性装饰器
```python
from unilabos.utils.decorator import topic_config
class MyDevice:
@property
@topic_config(period=1.0, qos=10) # 每秒发布一次
def temperature(self) -> float:
return self._temperature
```
### 虚拟设备
虚拟设备放置在 `unilabos/devices/virtual/` 目录下,命名为 `virtual_*.py`
## 注册表配置
### 设备注册表 (YAML)
位置: `unilabos/registry/devices/*.yaml`
```yaml
my_device_type:
category:
- my_category
description: "设备描述"
version: "1.0.0"
class:
module: "unilabos.devices.my_device:MyDevice"
type: python
status_types:
status: String
temperature: Float64
action_value_mappings:
auto-initialize:
type: UniLabJsonCommandAsync
goal: {}
feedback: {}
result: {}
schema: {...}
```
### 资源注册表 (YAML)
位置: `unilabos/registry/resources/**/*.yaml`
```yaml
my_container:
category:
- container
class:
module: "unilabos.resources.my_resource:MyContainer"
type: pylabrobot
version: "1.0.0"
```
## 协议编译器
位置: `unilabos/compile/*_protocol.py`
### 协议生成函数模板
```python
from typing import List, Dict, Any, Union
import networkx as nx
def generate_my_protocol(
G: nx.DiGraph,
vessel: Union[str, dict],
param1: float = 0.0,
**kwargs
) -> List[Dict[str, Any]]:
"""
生成操作协议序列
Args:
G: 物理拓扑图
vessel: 容器ID或字典
param1: 参数1
Returns:
List[Dict]: 动作序列
"""
# 提取vessel_id
vessel_id = vessel if isinstance(vessel, str) else vessel.get("id", "")
# 查找设备
device_id = find_connected_device(G, vessel_id)
# 生成动作
action_sequence = [{
"device_id": device_id,
"action_name": "my_action",
"action_kwargs": {
"vessel": {"id": vessel_id},
"param1": float(param1)
}
}]
return action_sequence
```
## 测试规范
### 测试文件位置
- 单元测试: `tests/` 目录
- 设备测试: `tests/devices/`
- 资源测试: `tests/resources/`
- ROS消息测试: `tests/ros/msgs/`
### 测试命名
```python
# tests/devices/my_device/test_my_device.py
import pytest
def test_device_initialization():
"""测试设备初始化"""
pass
def test_device_action():
"""测试设备动作"""
pass
```
## 错误处理
```python
from unilabos.utils.exception import UniLabException
try:
result = await device.execute_action()
except ValueError as e:
self.logger.error(f"参数错误: {e}")
self.data["status"] = "错误: 参数无效"
return False
except Exception as e:
self.logger.error(f"执行失败: {e}")
raise
```
## 配置管理
```python
from unilabos.config.config import BasicConfig, HTTPConfig
# 读取配置
port = BasicConfig.port
is_host = BasicConfig.is_host_mode
# 配置文件: local_config.py
```
## 常用工具
### 单例模式
```python
from unilabos.utils.decorator import singleton
@singleton
class MyManager:
pass
```
### 类型检查
```python
from unilabos.utils.type_check import NoAliasDumper
yaml.dump(data, f, Dumper=NoAliasDumper)
```
### 导入管理
```python
from unilabos.utils.import_manager import get_class
device_class = get_class("unilabos.devices.my_device:MyDevice")
```
## Git 提交规范
提交信息格式:
```
<type>(<scope>): <subject>
<body>
```
类型:
- `feat`: 新功能
- `fix`: 修复bug
- `docs`: 文档更新
- `refactor`: 重构
- `test`: 测试相关
- `chore`: 构建/工具相关
示例:
```
feat(devices): 添加虚拟搅拌器设备
- 实现VirtualStirrer类
- 支持定时搅拌和持续搅拌模式
- 添加速度验证逻辑
```

188
.cursorignore Normal file
View File

@@ -0,0 +1,188 @@
# ============================================================
# Uni-Lab-OS Cursor Ignore 配置,控制 Cursor AI 的文件索引范围
# ============================================================
# ==================== 敏感配置文件 ====================
# 本地配置(可能包含密钥)
**/local_config.py
test_config.py
local_test*.py
# 环境变量和密钥
.env
.env.*
**/.certs/
*.pem
*.key
credentials.json
secrets.yaml
# ==================== 二进制和 3D 模型文件 ====================
# 3D 模型文件(无需索引)
*.stl
*.dae
*.glb
*.gltf
*.obj
*.fbx
*.blend
# URDF/Xacro 机器人描述文件大型XML
*.xacro
# 图片文件
*.png
*.jpg
*.jpeg
*.gif
*.webp
*.ico
*.svg
*.bmp
# 压缩包
*.zip
*.tar
*.tar.gz
*.tgz
*.bz2
*.rar
*.7z
# ==================== Python 生成文件 ====================
__pycache__/
*.py[cod]
*$py.class
*.so
*.pyd
*.egg
*.egg-info/
.eggs/
dist/
build/
*.manifest
*.spec
# ==================== IDE 和编辑器 ====================
.idea/
.vscode/
*.swp
*.swo
*~
.#*
# ==================== 测试和覆盖率 ====================
.pytest_cache/
.coverage
.coverage.*
htmlcov/
.tox/
.nox/
coverage.xml
*.cover
# ==================== 虚拟环境 ====================
.venv/
venv/
env/
ENV/
# ==================== ROS 2 生成文件 ====================
# ROS 构建目录
build/
install/
log/
logs/
devel/
# ROS 消息生成
msg_gen/
srv_gen/
msg/*Action.msg
msg/*ActionFeedback.msg
msg/*ActionGoal.msg
msg/*ActionResult.msg
msg/*Feedback.msg
msg/*Goal.msg
msg/*Result.msg
msg/_*.py
srv/_*.py
build_isolated/
devel_isolated/
# ROS 动态配置
*.cfgc
/cfg/cpp/
/cfg/*.py
# ==================== 项目特定目录 ====================
# 工作数据目录
unilabos_data/
# 临时和输出目录
temp/
output/
cursor_docs/
configs/
# 文档构建
docs/_build/
/site
# ==================== 大型数据文件 ====================
# 点云数据
*.pcd
# GraphML 图形文件
*.graphml
# 日志文件
*.log
# 数据库
*.sqlite3
*.db
# Jupyter 检查点
.ipynb_checkpoints/
# ==================== 设备网格资源 ====================
# 3D 网格文件目录(包含大量 STL/DAE 文件)
unilabos/device_mesh/devices/**/*.stl
unilabos/device_mesh/devices/**/*.dae
unilabos/device_mesh/resources/**/*.stl
unilabos/device_mesh/resources/**/*.glb
unilabos/device_mesh/resources/**/*.xacro
# RViz 配置
*.rviz
# ==================== 系统文件 ====================
.DS_Store
Thumbs.db
desktop.ini
# ==================== 锁文件 ====================
poetry.lock
Pipfile.lock
pdm.lock
package-lock.json
yarn.lock
# ==================== 类型检查缓存 ====================
.mypy_cache/
.dmypy.json
.pytype/
.pyre/
pyrightconfig.json
# ==================== 其他 ====================
# Catkin
CATKIN_IGNORE
# Eclipse/Qt
.project
.cproject
CMakeLists.txt.user
*.user
qtcreator-*

View File

@@ -1,15 +1,11 @@
from __future__ import annotations from __future__ import annotations
import asyncio
import time import time
import traceback import traceback
from collections import Counter from collections import Counter
from typing import List, Sequence, Optional, Literal, Union, Iterator, Dict, Any, Callable, Set, cast from typing import List, Sequence, Optional, Literal, Union, Iterator, Dict, Any, Callable, Set, cast
from typing_extensions import TypedDict
from pylabrobot.liquid_handling import LiquidHandler, LiquidHandlerBackend, LiquidHandlerChatterboxBackend, Strictness from pylabrobot.liquid_handling import LiquidHandler, LiquidHandlerBackend, LiquidHandlerChatterboxBackend, Strictness
from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
from unilabos.devices.liquid_handling.laiyu.backend.laiyu_v_backend import UniLiquidHandlerLaiyuBackend
from pylabrobot.liquid_handling.liquid_handler import TipPresenceProbingMethod from pylabrobot.liquid_handling.liquid_handler import TipPresenceProbingMethod
from pylabrobot.liquid_handling.standard import GripDirection from pylabrobot.liquid_handling.standard import GripDirection
from pylabrobot.resources import ( from pylabrobot.resources import (
@@ -28,14 +24,27 @@ from pylabrobot.resources import (
Tip, Tip,
) )
from unilabos.registry.placeholder_type import ResourceSlot
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
from unilabos.resources.resource_tracker import ResourceTreeSet
class SimpleReturn(TypedDict): class SimpleReturn(TypedDict):
samples: list samples: List[List[ResourceDict]]
volumes: list volumes: List[float]
class SetLiquidReturn(TypedDict):
wells: List[List[ResourceDict]]
volumes: List[float]
class SetLiquidFromPlateReturn(TypedDict):
plate: List[List[ResourceDict]]
wells: List[List[ResourceDict]]
volumes: List[float]
class TransferLiquidReturn(TypedDict):
sources: List[List[ResourceDict]]
targets: List[List[ResourceDict]]
class SetLiquidReturn(TypedDict): class SetLiquidReturn(TypedDict):
@@ -678,40 +687,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
well.set_liquids([(liquid_name, volume)]) # type: ignore well.set_liquids([(liquid_name, volume)]) # type: ignore
res_volumes.append(volume) res_volumes.append(volume)
return SetLiquidReturn( return SimpleReturn(samples=res_samples, volumes=res_volumes)
wells=ResourceTreeSet.from_plr_resources(wells, known_newly_created=False).dump(), volumes=res_volumes # type: ignore
)
@classmethod
def set_liquid_from_plate(
cls, plate: ResourceSlot, well_names: list[str], liquid_names: list[str], volumes: list[float]
) -> SetLiquidFromPlateReturn:
"""Set the liquid in wells of a plate by well names (e.g., A1, A2, B3).
如果 liquid_names 和 volumes 为空,但 plate 和 well_names 不为空,直接返回 plate 和 wells。
"""
# 根据 well_names 获取对应的 Well 对象
wells = [plate.get_well(name) for name in well_names]
res_volumes = []
# 如果 liquid_names 和 volumes 都为空,直接返回
if not liquid_names and not volumes:
return SetLiquidFromPlateReturn(
plate=ResourceTreeSet.from_plr_resources([plate], known_newly_created=False).dump(), # type: ignore
wells=ResourceTreeSet.from_plr_resources(wells, known_newly_created=False).dump(), # type: ignore
volumes=res_volumes,
)
for well, liquid_name, volume in zip(wells, liquid_names, volumes):
well.set_liquids([(liquid_name, volume)]) # type: ignore
res_volumes.append(volume)
return SetLiquidFromPlateReturn(
plate=ResourceTreeSet.from_plr_resources([plate], known_newly_created=False).dump(), # type: ignore
wells=ResourceTreeSet.from_plr_resources(wells, known_newly_created=False).dump(), # type: ignore
volumes=res_volumes,
)
# --------------------------------------------------------------- # ---------------------------------------------------------------
# REMOVE LIQUID -------------------------------------------------- # REMOVE LIQUID --------------------------------------------------
# --------------------------------------------------------------- # ---------------------------------------------------------------
@@ -1111,7 +1087,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
mix_liquid_height: Optional[float] = None, mix_liquid_height: Optional[float] = None,
delays: Optional[List[int]] = None, delays: Optional[List[int]] = None,
none_keys: List[str] = [], none_keys: List[str] = [],
): ) -> TransferLiquidReturn:
"""Transfer liquid with automatic mode detection. """Transfer liquid with automatic mode detection.
Supports three transfer modes: Supports three transfer modes:
@@ -1251,6 +1227,11 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
"Supported modes: 1->N, N->1, or N->N." "Supported modes: 1->N, N->1, or N->N."
) )
return TransferLiquidReturn(
sources=ResourceTreeSet.from_plr_resources(list(sources), known_newly_created=False).dump(), # type: ignore
targets=ResourceTreeSet.from_plr_resources(list(targets), known_newly_created=False).dump(), # type: ignore
)
async def _transfer_one_to_one( async def _transfer_one_to_one(
self, self,
sources: Sequence[Container], sources: Sequence[Container],

View File

@@ -52,6 +52,7 @@ from unilabos.devices.liquid_handling.liquid_handler_abstract import (
SimpleReturn, SimpleReturn,
SetLiquidReturn, SetLiquidReturn,
SetLiquidFromPlateReturn, SetLiquidFromPlateReturn,
TransferLiquidReturn,
) )
from unilabos.registry.placeholder_type import ResourceSlot from unilabos.registry.placeholder_type import ResourceSlot
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
@@ -154,25 +155,29 @@ class PRCXI9300Plate(Plate):
**kwargs, **kwargs,
): ):
# 如果 ordered_items 不为 None直接使用 # 如果 ordered_items 不为 None直接使用
items = None
ordering_param = None
if ordered_items is not None: if ordered_items is not None:
items = ordered_items items = ordered_items
elif ordering is not None: elif ordering is not None:
# 检查 ordering 中的值是否是字符串(从 JSON 反序列化时的情况) # 检查 ordering 中的值是否是字符串(从 JSON 反序列化时的情况)
# 如果是字符串,说明这是位置名称,需要让 Plate 自己创建 Well 对象 # 如果是字符串,说明这是位置名称,需要让 Plate 自己创建 Well 对象
# 我们只传递位置信息(键),不传递值,使用 ordering 参数 # 我们只传递位置信息(键),不传递值,使用 ordering 参数
if ordering and isinstance(next(iter(ordering.values()), None), str): if ordering:
values = list(ordering.values())
value = values[0]
if isinstance(value, str):
# ordering 的值是字符串,只使用键(位置信息)创建新的 OrderedDict # ordering 的值是字符串,只使用键(位置信息)创建新的 OrderedDict
# 传递 ordering 参数而不是 ordered_items让 Plate 自己创建 Well 对象 # 传递 ordering 参数而不是 ordered_items让 Plate 自己创建 Well 对象
items = None items = None
# 使用 ordering 参数,只包含位置信息(键) # 使用 ordering 参数,只包含位置信息(键)
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys()) ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
elif value is None:
ordering_param = ordering
else: else:
# ordering 的值已经是对象,可以直接使用 # ordering 的值已经是对象,可以直接使用
items = ordering items = ordering
ordering_param = None ordering_param = None
else:
items = None
ordering_param = None
# 根据情况传递不同的参数 # 根据情况传递不同的参数
if items is not None: if items is not None:
@@ -590,7 +595,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
return super().set_liquid(wells, liquid_names, volumes) return super().set_liquid(wells, liquid_names, volumes)
def set_liquid_from_plate( def set_liquid_from_plate(
self, plate: ResourceSlot, well_names: list[str], liquid_names: list[str], volumes: list[float] self, plate: List[ResourceSlot], well_names: list[str], liquid_names: list[str], volumes: list[float]
) -> SetLiquidFromPlateReturn: ) -> SetLiquidFromPlateReturn:
return super().set_liquid_from_plate(plate, well_names, liquid_names, volumes) return super().set_liquid_from_plate(plate, well_names, liquid_names, volumes)
@@ -713,7 +718,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
mix_liquid_height: Optional[float] = None, mix_liquid_height: Optional[float] = None,
delays: Optional[List[int]] = None, delays: Optional[List[int]] = None,
none_keys: List[str] = [], none_keys: List[str] = [],
): ) -> TransferLiquidReturn:
return await super().transfer_liquid( return await super().transfer_liquid(
sources, sources,
targets, targets,

View File

@@ -0,0 +1,376 @@
# -*- coding: utf-8 -*-
"""
ZDT X42 Closed-Loop Stepper Motor Driver
RS485 Serial Communication via USB-Serial Converter
- Baudrate: 115200
"""
import serial
import time
import threading
import struct
import logging
from typing import Optional, Any
try:
from unilabos.device_comms.universal_driver import UniversalDriver
except ImportError:
class UniversalDriver:
def __init__(self, *args, **kwargs):
self.logger = logging.getLogger(self.__class__.__name__)
def execute_command_from_outer(self, command: Any): pass
from serial.rs485 import RS485Settings
class ZDTX42Driver(UniversalDriver):
"""
ZDT X42 闭环步进电机驱动器
支持功能:
- 速度模式运行
- 位置模式运行 (相对/绝对)
- 位置读取和清零
- 使能/禁用控制
通信协议:
- 帧格式: [设备ID] [功能码] [数据...] [校验位=0x6B]
- 响应长度根据功能码决定
"""
def __init__(
self,
port: str,
baudrate: int = 115200,
device_id: int = 1,
timeout: float = 0.5,
debug: bool = False
):
"""
初始化 ZDT X42 电机驱动
Args:
port: 串口设备路径
baudrate: 波特率 (默认 115200)
device_id: 设备地址 (1-255)
timeout: 通信超时时间(秒)
debug: 是否启用调试输出
"""
super().__init__()
self.id = device_id
self.debug = debug
self.lock = threading.RLock()
self.status = "idle" # 对应注册表中的 status (str)
self.position = 0 # 对应注册表中的 position (int)
try:
self.ser = serial.Serial(
port=port,
baudrate=baudrate,
timeout=timeout,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE
)
# 启用 RS485 模式
try:
self.ser.rs485_mode = RS485Settings(
rts_level_for_tx=True,
rts_level_for_rx=False
)
except Exception:
pass # RS485 模式是可选的
self.logger.info(
f"ZDT X42 Motor connected: {port} "
f"(Baud: {baudrate}, ID: {device_id})"
)
# 自动使能电机,确保初始状态可运动
self.enable(True)
# 启动背景轮询线程,确保 position 实时刷新
self._stop_event = threading.Event()
self._polling_thread = threading.Thread(
target=self._update_loop,
name=f"ZDTPolling_{port}",
daemon=True
)
self._polling_thread.start()
except Exception as e:
self.logger.error(f"Failed to open serial port {port}: {e}")
self.ser = None
def _update_loop(self):
"""背景循环读取电机位置"""
while not self._stop_event.is_set():
try:
self.get_position()
except Exception as e:
if self.debug:
self.logger.error(f"Polling error: {e}")
time.sleep(1.0) # 每1秒刷新一次位置数据
def _send(self, func_code: int, payload: list) -> bytes:
"""
发送指令并接收响应
Args:
func_code: 功能码
payload: 数据负载 (list of bytes)
Returns:
响应数据 (bytes)
"""
if not self.ser:
self.logger.error("Serial port not available")
return b""
with self.lock:
# 清空输入缓冲区
self.ser.reset_input_buffer()
# 构建消息: [ID] [功能码] [数据...] [校验位=0x6B]
message = bytes([self.id, func_code] + payload + [0x6B])
# 发送
self.ser.write(message)
# 根据功能码决定响应长度
# 查询类指令返回 10 字节,控制类指令返回 4 字节
read_len = 10 if func_code in [0x31, 0x32, 0x35, 0x24, 0x27] else 4
response = self.ser.read(read_len)
# 调试输出
if self.debug:
sent_hex = message.hex().upper()
recv_hex = response.hex().upper() if response else 'TIMEOUT'
print(f"[ID {self.id}] TX: {sent_hex} → RX: {recv_hex}")
return response
def enable(self, on: bool = True) -> bool:
"""
使能/禁用电机
Args:
on: True=使能(锁轴), False=禁用(松轴)
Returns:
是否成功
"""
state = 1 if on else 0
resp = self._send(0xF3, [0xAB, state, 0])
return len(resp) >= 4
def move_speed(
self,
speed_rpm: int,
direction: str = "CW",
acceleration: int = 10
) -> bool:
"""
速度模式运行
Args:
speed_rpm: 转速 (RPM)
direction: 方向 ("CW"=顺时针, "CCW"=逆时针)
acceleration: 加速度 (0-255)
Returns:
是否成功
"""
dir_val = 0 if direction.upper() in ["CW", "顺时针"] else 1
speed_bytes = struct.pack('>H', int(speed_rpm))
self.status = f"moving@{speed_rpm}rpm"
resp = self._send(0xF6, [dir_val, speed_bytes[0], speed_bytes[1], acceleration, 0])
return len(resp) >= 4
def move_position(
self,
pulses: int,
speed_rpm: int,
direction: str = "CW",
acceleration: int = 10,
absolute: bool = False
) -> bool:
"""
位置模式运行
Args:
pulses: 脉冲数
speed_rpm: 转速 (RPM)
direction: 方向 ("CW"=顺时针, "CCW"=逆时针)
acceleration: 加速度 (0-255)
absolute: True=绝对位置, False=相对位置
Returns:
是否成功
"""
dir_val = 0 if direction.upper() in ["CW", "顺时针"] else 1
speed_bytes = struct.pack('>H', int(speed_rpm))
self.status = f"moving_to_{pulses}"
pulse_bytes = struct.pack('>I', int(pulses))
abs_flag = 1 if absolute else 0
payload = [
dir_val,
speed_bytes[0], speed_bytes[1],
acceleration,
pulse_bytes[0], pulse_bytes[1], pulse_bytes[2], pulse_bytes[3],
abs_flag,
0
]
resp = self._send(0xFD, payload)
return len(resp) >= 4
def stop(self) -> bool:
"""
停止电机
Returns:
是否成功
"""
self.status = "idle"
resp = self._send(0xFE, [0x98, 0])
return len(resp) >= 4
def rotate_quarter(self, speed_rpm: int = 60, direction: str = "CW") -> bool:
"""
电机旋转 1/4 圈 (阻塞式)
假设电机细分为 3200 脉冲/圈1/4 圈 = 800 脉冲
"""
pulses = 800
success = self.move_position(pulses=pulses, speed_rpm=speed_rpm, direction=direction, absolute=False)
if success:
# 计算预估旋转时间并进行阻塞等待 (Time = revolutions / (RPM/60))
# 1/4 rev / (RPM/60) = 15.0 / RPM
estimated_time = 15.0 / max(1, speed_rpm)
time.sleep(estimated_time + 0.5) # 额外给 0.5 秒缓冲
self.status = "idle"
return success
def wait_time(self, duration_s: float) -> bool:
"""
等待指定时间 (秒)
"""
self.logger.info(f"Waiting for {duration_s} seconds...")
time.sleep(duration_s)
return True
def set_zero(self) -> bool:
"""
清零当前位置
Returns:
是否成功
"""
resp = self._send(0x0A, [])
return len(resp) >= 4
def get_position(self) -> Optional[int]:
"""
读取当前位置 (脉冲数)
Returns:
当前位置脉冲数,失败返回 None
"""
resp = self._send(0x32, [])
if len(resp) >= 8:
# 响应格式: [ID] [Func] [符号位] [数值4字节] [校验]
sign = resp[2] # 0=正, 1=负
value = struct.unpack('>I', resp[3:7])[0]
self.position = -value if sign == 1 else value
if self.debug:
print(f"[Position] Raw: {resp.hex().upper()}, Parsed: {self.position}")
return self.position
self.logger.warning("Failed to read position")
return None
def close(self):
"""关闭串口连接并停止线程"""
if hasattr(self, '_stop_event'):
self._stop_event.set()
if self.ser and self.ser.is_open:
self.ser.close()
self.logger.info("Serial port closed")
# ============================================================
# 测试和调试代码
# ============================================================
def test_motor():
"""基础功能测试"""
logging.basicConfig(level=logging.INFO)
print("="*60)
print("ZDT X42 电机驱动测试")
print("="*60)
driver = ZDTX42Driver(
port="/dev/tty.usbserial-3110",
baudrate=115200,
device_id=2,
debug=True
)
if not driver.ser:
print("❌ 串口打开失败")
return
try:
# 测试 1: 读取位置
print("\n[1] 读取当前位置")
pos = driver.get_position()
print(f"✓ 当前位置: {pos} 脉冲")
# 测试 2: 使能
print("\n[2] 使能电机")
driver.enable(True)
time.sleep(0.3)
print("✓ 电机已锁定")
# 测试 3: 相对位置运动
print("\n[3] 相对位置运动 (1000脉冲)")
driver.move_position(pulses=1000, speed_rpm=60, direction="CW")
time.sleep(2)
pos = driver.get_position()
print(f"✓ 新位置: {pos}")
# 测试 4: 速度运动
print("\n[4] 速度模式 (30RPM, 3秒)")
driver.move_speed(speed_rpm=30, direction="CW")
time.sleep(3)
driver.stop()
pos = driver.get_position()
print(f"✓ 停止后位置: {pos}")
# 测试 5: 禁用
print("\n[5] 禁用电机")
driver.enable(False)
print("✓ 电机已松开")
print("\n" + "="*60)
print("✅ 测试完成")
print("="*60)
except Exception as e:
print(f"\n❌ 测试失败: {e}")
import traceback
traceback.print_exc()
finally:
driver.close()
if __name__ == "__main__":
test_motor()

View File

@@ -623,6 +623,119 @@ class ChinweDevice(UniversalDriver):
time.sleep(duration) time.sleep(duration)
return True return True
def separation_step(self, motor_id: int = 5, speed: int = 60, pulses: int = 700,
max_cycles: int = 0, timeout: int = 300) -> bool:
"""
分液步骤 - 液位传感器与电机联动
当液位传感器检测到"有液"时,电机顺时针旋转指定脉冲数
当液位传感器检测到"无液"时,电机逆时针旋转指定脉冲数
:param motor_id: 电机ID (必须在初始化时配置的motor_ids中)
:param speed: 电机转速 (RPM)
:param pulses: 每次旋转的脉冲数 (默认700约为1/4圈,假设3200脉冲/圈)
:param max_cycles: 最大执行循环次数 (0=无限制,默认0)
:param timeout: 整体超时时间 (秒)
:return: 成功返回True,超时或失败返回False
"""
motor_id = int(motor_id)
speed = int(speed)
pulses = int(pulses)
max_cycles = int(max_cycles)
timeout = int(timeout)
# 检查电机是否存在
if motor_id not in self.motors:
self.logger.error(f"Motor {motor_id} not found in configured motors: {list(self.motors.keys())}")
return False
# 检查传感器是否可用
if not self.sensor:
self.logger.error("Sensor not initialized")
return False
motor = self.motors[motor_id]
# 停止轮询线程,避免与 separation_step 同时读取传感器造成串口冲突
self.logger.info("Stopping polling thread for separation_step...")
self._stop_event.set()
if self._poll_thread and self._poll_thread.is_alive():
self._poll_thread.join(timeout=2.0)
# 使能电机
self.logger.info(f"Enabling motor {motor_id}...")
motor.enable(True)
time.sleep(0.2)
self.logger.info(f"Starting separation step: motor_id={motor_id}, speed={speed} RPM, "
f"pulses={pulses}, max_cycles={max_cycles}, timeout={timeout}s")
# 记录上一次的液位状态
last_level = None
cycle_count = 0
start_time = time.time()
error_count = 0
try:
while True:
# 检查超时
if time.time() - start_time > timeout:
self.logger.warning(f"Separation step timeout after {timeout} seconds")
return False
# 检查循环次数限制
if max_cycles > 0 and cycle_count >= max_cycles:
self.logger.info(f"Separation step completed: reached max_cycles={max_cycles}")
return True
# 读取传感器数据
data = self.sensor.read_level()
if data is None:
error_count += 1
if error_count > 5:
self.logger.warning("Sensor read failed multiple times, retrying...")
error_count = 0
time.sleep(0.5)
continue
error_count = 0
current_level = data['level']
rssi = data['rssi']
# 检测状态变化 (包括首次检测)
if current_level != last_level:
cycle_count += 1
if current_level:
# 有液 -> 电机顺时针旋转
self.logger.info(f"[Cycle {cycle_count}] Liquid detected (RSSI={rssi}), "
f"rotating motor {motor_id} clockwise {pulses} pulses")
motor.run_position(pulses=pulses, speed_rpm=speed, direction=0, absolute=False)
# 等待电机完成 (预估时间)
estimated_time = 15.0 / max(1, speed)
time.sleep(estimated_time + 0.5)
else:
# 无液 -> 电机逆时针旋转
self.logger.info(f"[Cycle {cycle_count}] No liquid detected (RSSI={rssi}), "
f"rotating motor {motor_id} counter-clockwise {pulses} pulses")
motor.run_position(pulses=pulses, speed_rpm=speed, direction=1, absolute=False)
# 等待电机完成 (预估时间)
estimated_time = 15.0 / max(1, speed)
time.sleep(estimated_time + 0.5)
# 更新状态
last_level = current_level
# 轮询间隔
time.sleep(0.1)
finally:
# 恢复轮询线程
self.logger.info("Restarting polling thread...")
self._start_polling()
def execute_command_from_outer(self, command_dict: Dict[str, Any]) -> bool: def execute_command_from_outer(self, command_dict: Dict[str, Any]) -> bool:
"""支持标准 JSON 指令调用""" """支持标准 JSON 指令调用"""
return super().execute_command_from_outer(command_dict) return super().execute_command_from_outer(command_dict)

View File

@@ -0,0 +1,379 @@
# -*- coding: utf-8 -*-
"""
XKC RS485 液位传感器 (Modbus RTU)
说明:
1. 遵循 Modbus-RTU 协议。
2. 数据寄存器: 0x0001 (液位状态, 1=有液, 0=无液), 0x0002 (RSSI 信号强度)。
3. 地址寄存器: 0x0004 (可读写, 范围 1-254)。
4. 波特率寄存器: 0x0005 (可写, 代码表见 change_baudrate 方法)。
"""
import struct
import threading
import time
import logging
import serial
from typing import Optional, Dict, Any, List
from unilabos.device_comms.universal_driver import UniversalDriver
class TransportManager:
"""
统一通信管理类。
仅支持 串口 (Serial/有线) 连接。
"""
def __init__(self, port: str, baudrate: int = 9600, timeout: float = 3.0, logger=None):
self.port = port
self.baudrate = baudrate
self.timeout = timeout
self.logger = logger
self.lock = threading.RLock() # 线程锁,确保多设备共用一个连接时不冲突
self.serial = None
self._connect_serial()
def _connect_serial(self):
try:
self.serial = serial.Serial(
port=self.port,
baudrate=self.baudrate,
timeout=self.timeout
)
except Exception as e:
raise ConnectionError(f"Serial open failed: {e}")
def close(self):
"""关闭连接"""
if self.serial and self.serial.is_open:
self.serial.close()
def clear_buffer(self):
"""清空缓冲区 (Thread-safe)"""
with self.lock:
if self.serial:
self.serial.reset_input_buffer()
def write(self, data: bytes):
"""发送原始字节"""
with self.lock:
if self.serial:
self.serial.write(data)
def read(self, size: int) -> bytes:
"""读取指定长度字节"""
if self.serial:
return self.serial.read(size)
return b''
class XKCSensorDriver(UniversalDriver):
"""XKC RS485 液位传感器 (Modbus RTU)"""
def __init__(self, port: str, baudrate: int = 9600, device_id: int = 6,
threshold: int = 300, timeout: float = 3.0, debug: bool = False):
super().__init__()
self.port = port
self.baudrate = baudrate
self.device_id = device_id
self.threshold = threshold
self.timeout = timeout
self.debug = debug
self.level = False
self.rssi = 0
self.status = {"level": self.level, "rssi": self.rssi}
try:
self.transport = TransportManager(port, baudrate, timeout, logger=self.logger)
self.logger.info(f"XKCSensorDriver connected to {port} (ID: {device_id})")
except Exception as e:
self.logger.error(f"Failed to connect XKCSensorDriver: {e}")
self.transport = None
# 启动背景轮询线程,确保 status 实时刷新
self._stop_event = threading.Event()
self._polling_thread = threading.Thread(
target=self._update_loop,
name=f"XKCPolling_{port}",
daemon=True
)
if self.transport:
self._polling_thread.start()
def _update_loop(self):
"""背景循环读取传感器数据"""
while not self._stop_event.is_set():
try:
self.read_level()
except Exception as e:
if self.debug:
self.logger.error(f"Polling error: {e}")
time.sleep(2.0) # 每2秒刷新一次数据
def _crc(self, data: bytes) -> bytes:
crc = 0xFFFF
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 0x0001: crc = (crc >> 1) ^ 0xA001
else: crc >>= 1
return struct.pack('<H', crc)
def read_level(self) -> Optional[Dict[str, Any]]:
"""
读取液位。
返回: {'level': bool, 'rssi': int}
"""
if not self.transport:
return None
with self.transport.lock:
self.transport.clear_buffer()
# Modbus Read Registers: 01 03 00 01 00 02 CRC
payload = struct.pack('>HH', 0x0001, 0x0002)
msg = struct.pack('BB', self.device_id, 0x03) + payload
msg += self._crc(msg)
if self.debug:
self.logger.info(f"TX (ID {self.device_id}): {msg.hex().upper()}")
self.transport.write(msg)
# Read header
h = self.transport.read(3) # Addr, Func, Len
if self.debug:
self.logger.info(f"RX Header: {h.hex().upper()}")
if len(h) < 3: return None
length = h[2]
# Read body + CRC
body = self.transport.read(length + 2)
if self.debug:
self.logger.info(f"RX Body+CRC: {body.hex().upper()}")
if len(body) < length + 2:
# Firmware bug fix specific to some modules
if len(body) == 4 and length == 4:
pass
else:
return None
data = body[:-2]
# 根据手册说明:
# 寄存器 0x0001 (data[0:2]): 液位状态 (00 01 为有液, 00 00 为无液)
# 寄存器 0x0002 (data[2:4]): 信号强度 RSSI
hw_level = False
rssi = 0
if len(data) >= 4:
hw_level = ((data[0] << 8) | data[1]) == 1
rssi = (data[2] << 8) | data[3]
elif len(data) == 2:
# 兼容模式: 某些老固件可能只返回 1 个寄存器
rssi = (data[0] << 8) | data[1]
hw_level = rssi > self.threshold
else:
return None
# 最终判定: 优先使用硬件层级的 level 判定,但 RSSI 阈值逻辑作为补充/校验
# 注意: 如果用户显式设置了 THRESHOLD我们可以在逻辑中做权衡
self.level = hw_level or (rssi > self.threshold)
self.rssi = rssi
result = {
'level': self.level,
'rssi': self.rssi
}
self.status = result
return result
def wait_level(self, target_state: bool, timeout: float = 60.0) -> bool:
"""
等待液位达到目标状态 (阻塞式)
"""
self.logger.info(f"Waiting for level: {target_state}")
start_time = time.time()
while (time.time() - start_time) < timeout:
res = self.read_level()
if res and res.get('level') == target_state:
return True
time.sleep(0.5)
self.logger.warning(f"Wait level timeout ({timeout}s)")
return False
def wait_for_liquid(self, target_state: bool, timeout: float = 120.0) -> bool:
"""
实时检测电导率(RSSI)并等待用户指定的“有液”或“无液”状态。
一旦检测到符合目标状态,立即返回。
Args:
target_state: True 为“有液”, False 为“无液”
timeout: 最大等待时间(秒)
"""
state_str = "有液" if target_state else "无液"
self.logger.info(f"开始实时检测电导率,等待状态: {state_str} (超时: {timeout}s)")
start_time = time.time()
while (time.time() - start_time) < timeout:
res = self.read_level() # 内部已更新 self.level 和 self.rssi
if res:
current_level = res.get('level')
current_rssi = res.get('rssi')
if current_level == target_state:
self.logger.info(f"✅ 检测到目标状态: {state_str} (当前电导率/RSSI: {current_rssi})")
return True
if self.debug:
self.logger.debug(f"当前状态: {'有液' if current_level else '无液'}, RSSI: {current_rssi}")
time.sleep(0.2) # 高频采样
self.logger.warning(f"❌ 等待 {state_str} 状态超时 ({timeout}s)")
return False
def set_threshold(self, threshold: int):
"""设置液位判定阈值"""
self.threshold = int(threshold)
self.logger.info(f"Threshold updated to: {self.threshold}")
def change_device_id(self, new_id: int) -> bool:
"""
修改设备的 Modbus 从站地址。
寄存器: 0x0004, 功能码: 0x06
"""
if not (1 <= new_id <= 254):
self.logger.error(f"Invalid device ID: {new_id}. Must be 1-254.")
return False
self.logger.info(f"Changing device ID from {self.device_id} to {new_id}")
success = self._write_single_register(0x0004, new_id)
if success:
self.device_id = new_id # 更新内存中的地址
self.logger.info(f"Device ID update command sent successfully (target {new_id}).")
return success
def change_baudrate(self, baud_code: int) -> bool:
"""
更改通讯波特率 (寄存器: 0x0005)。
设置成功后传感器 LED 会闪烁,通常无数据返回。
波特率代码对照表 (16进制):
05: 2400
06: 4800
07: 9600 (默认)
08: 14400
09: 19200
0A: 28800
0C: 57600
0D: 115200
0E: 128000
0F: 256000
"""
self.logger.info(f"Sending baudrate change command (Code: {baud_code:02X})")
# 写入寄存器 0x0005
self._write_single_register(0x0005, baud_code)
self.logger.info("Baudrate change command executed. Device LED should flash. Please update connection settings.")
return True
def factory_reset(self) -> bool:
"""
恢复出厂设置 (通过广播地址 FF)。
设置地址为 01逻辑为向 0x0004 写入 0x0002
"""
self.logger.info("Sending factory reset command via broadcast address FF...")
# 广播指令通常无回显
self._write_single_register(0x0004, 0x0002, slave_id=0xFF)
self.logger.info("Factory reset command sent. Device address should be 01 now.")
return True
def _write_single_register(self, reg_addr: int, value: int, slave_id: Optional[int] = None) -> bool:
"""内部辅助函数: Modbus 功能码 06 写单个寄存器"""
if not self.transport: return False
target_id = slave_id if slave_id is not None else self.device_id
msg = struct.pack('BBHH', target_id, 0x06, reg_addr, value)
msg += self._crc(msg)
with self.transport.lock:
self.transport.clear_buffer()
if self.debug:
self.logger.info(f"TX Write (Reg {reg_addr:#06x}): {msg.hex().upper()}")
self.transport.write(msg)
# 广播地址、波特率修改或厂家特定指令可能无回显
if target_id == 0xFF or reg_addr == 0x0005:
time.sleep(0.5)
return True
# 等待返回 (正常应返回相同报文)
resp = self.transport.read(len(msg))
if self.debug:
self.logger.info(f"RX Write Response: {resp.hex().upper()}")
return resp == msg
def close(self):
if self.transport:
self.transport.close()
if __name__ == "__main__":
# 快速实例化测试
import logging
# 减少冗余日志,仅显示重要信息
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
# 硬件配置 (根据实际情况修改)
TEST_PORT = "/dev/tty.usbserial-3110"
SLAVE_ID = 1
THRESHOLD = 300
print("\n" + "="*50)
print(f" XKC RS485 传感器独立测试程序")
print(f" 端口: {TEST_PORT} | 地址: {SLAVE_ID} | 阈值: {THRESHOLD}")
print("="*50)
sensor = XKCSensorDriver(port=TEST_PORT, device_id=SLAVE_ID, threshold=THRESHOLD, debug=False)
try:
if sensor.transport:
print(f"\n开始实时连续采样测试 (持续 15 秒)...")
print(f"按 Ctrl+C 可提前停止\n")
start_time = time.time()
duration = 15
count = 0
while time.time() - start_time < duration:
count += 1
res = sensor.read_level()
if res:
rssi = res['rssi']
level = res['level']
status_str = "【有液】" if level else "【无液】"
# 使用 \r 实现单行刷新显示 (或者不刷,直接打印历史)
# 为了方便查看变化,我们直接打印
elapsed = time.time() - start_time
print(f" [{elapsed:4.1f}s] 采样 {count:<3}: 电导率/RSSI = {rssi:<5} | 判定结果: {status_str}")
else:
print(f" [{time.time()-start_time:4.1f}s] 采样 {count:<3}: 通信失败 (无响应)")
time.sleep(0.5) # 每秒采样 2 次
print(f"\n--- 15 秒采样测试完成 (总计 {count} 次) ---")
# [3] 测试动态修改阈值
print(f"\n[3] 动态修改阈值演示...")
new_threshold = 400
sensor.set_threshold(new_threshold)
res = sensor.read_level()
if res:
print(f" 采样 (当前阈值={new_threshold}): 电导率/RSSI = {res['rssi']:<5} | 判定结果: {'【有液】' if res['level'] else '【无液】'}")
sensor.set_threshold(THRESHOLD) # 还原
except KeyboardInterrupt:
print("\n[!] 用户中断测试")
except Exception as e:
print(f"\n[!] 测试运行出错: {e}")
finally:
sensor.close()
print("\n--- 测试程序已退出 ---\n")

View File

@@ -258,7 +258,7 @@ class BioyondResourceSynchronizer(ResourceSynchronizer):
logger.info(f"[同步→Bioyond] 物料不存在于 Bioyond将创建新物料并入库") logger.info(f"[同步→Bioyond] 物料不存在于 Bioyond将创建新物料并入库")
# 第1步从配置中获取仓库配置 # 第1步从配置中获取仓库配置
warehouse_mapping = self.bioyond_config.get("warehouse_mapping", {}) warehouse_mapping = self.workstation.bioyond_config.get("warehouse_mapping", {})
# 确定目标仓库名称 # 确定目标仓库名称
parent_name = None parent_name = None

View File

@@ -317,6 +317,47 @@ separator.chinwe:
- port - port
type: object type: object
type: UniLabJsonCommand type: UniLabJsonCommand
separation_step:
goal:
max_cycles: 0
motor_id: 5
pulses: 700
speed: 60
timeout: 300
handles: {}
schema:
description: 分液步骤 - 液位传感器与电机联动 (有液→顺时针, 无液→逆时针)
properties:
goal:
properties:
max_cycles:
default: 0
description: 最大循环次数 (0=无限制)
type: integer
motor_id:
default: '5'
description: 选择电机
enum:
- '4'
- '5'
title: '注: 4=搅拌, 5=旋钮'
type: string
pulses:
default: 700
description: 每次旋转脉冲数 (约1/4圈)
type: integer
speed:
default: 60
description: 电机转速 (RPM)
type: integer
timeout:
default: 300
description: 超时时间 (秒)
type: integer
required:
- motor_id
type: object
type: UniLabJsonCommand
wait_sensor_level: wait_sensor_level:
goal: goal:
target_state: 有液 target_state: 有液

View File

@@ -638,7 +638,7 @@ liquid_handler:
placeholder_keys: {} placeholder_keys: {}
result: {} result: {}
schema: schema:
description: 吸头迭代函数。用于自动管理和切换吸头架中的吸头,实现批量实验中的吸头自动分配和追踪。该函数监控吸头使用状态,自动切换到下一个可用吸头位置,确保实验流程的连续性。适用于高通量实验、批量处理、自动化流水线等需要大量吸头管理的应用场景。 description: 吸头迭代函数。用于自动管理和切换枪头盒中的吸头,实现批量实验中的吸头自动分配和追踪。该函数监控吸头使用状态,自动切换到下一个可用吸头位置,确保实验流程的连续性。适用于高通量实验、批量处理、自动化流水线等需要大量吸头管理的应用场景。
properties: properties:
feedback: {} feedback: {}
goal: goal:
@@ -712,6 +712,43 @@ liquid_handler:
title: set_group参数 title: set_group参数
type: object type: object
type: UniLabJsonCommand type: UniLabJsonCommand
auto-set_liquid_from_plate:
feedback: {}
goal: {}
goal_default:
liquid_names: null
plate: null
volumes: null
well_names: null
handles: {}
placeholder_keys: {}
result: {}
schema:
description: ''
properties:
feedback: {}
goal:
properties:
liquid_names:
type: string
plate:
type: string
volumes:
type: string
well_names:
type: string
required:
- plate
- well_names
- liquid_names
- volumes
type: object
result: {}
required:
- goal
title: set_liquid_from_plate参数
type: object
type: UniLabJsonCommand
auto-set_tiprack: auto-set_tiprack:
feedback: {} feedback: {}
goal: {} goal: {}
@@ -721,7 +758,7 @@ liquid_handler:
placeholder_keys: {} placeholder_keys: {}
result: {} result: {}
schema: schema:
description: 吸头架设置函数。用于配置和初始化液体处理系统的吸头架信息,包括吸头架位置、类型、容量等参数。该函数建立吸头资源管理系统,为后续的吸头选择和使用提供基础配置。适用于系统初始化、吸头架更换、实验配置等需要吸头资源管理的操作场景。 description: 枪头盒设置函数。用于配置和初始化液体处理系统的枪头盒信息,包括枪头盒位置、类型、容量等参数。该函数建立吸头资源管理系统,为后续的吸头选择和使用提供基础配置。适用于系统初始化、枪头盒更换、实验配置等需要吸头资源管理的操作场景。
properties: properties:
feedback: {} feedback: {}
goal: goal:
@@ -4093,32 +4130,32 @@ liquid_handler:
- 0 - 0
handles: handles:
input: input:
- data_key: liquid - data_key: sources
data_source: handle data_source: handle
data_type: resource data_type: resource
handler_key: sources handler_key: sources
label: sources label: 待移动液体
- data_key: liquid - data_key: targets
data_source: executor
data_type: resource
handler_key: targets
label: targets
- data_key: liquid
data_source: executor
data_type: resource
handler_key: tip_rack
label: tip_rack
output:
- data_key: liquid
data_source: handle data_source: handle
data_type: resource data_type: resource
handler_key: targets
label: 转移目标
- data_key: tip_racks
data_source: handle
data_type: resource
handler_key: tip_rack
label: 枪头盒
output:
- data_key: sources.@flatten
data_source: executor
data_type: resource
handler_key: sources_out handler_key: sources_out
label: sources label: 移液后源孔
- data_key: liquid - data_key: targets.@flatten
data_source: executor data_source: executor
data_type: resource data_type: resource
handler_key: targets_out handler_key: targets_out
label: targets label: 移液后目标孔
placeholder_keys: placeholder_keys:
sources: unilabos_resources sources: unilabos_resources
targets: unilabos_resources targets: unilabos_resources
@@ -5114,19 +5151,34 @@ liquid_handler.biomek:
- 0 - 0
handles: handles:
input: input:
- data_key: liquid - data_key: sources
data_source: handle data_source: handle
data_type: resource data_type: resource
handler_key: liquid-input handler_key: sources
io_type: target io_type: target
label: Liquid Input label: 待移动液体
- data_key: targets
data_source: handle
data_type: resource
handler_key: targets
label: 转移目标
- data_key: tip_racks
data_source: handle
data_type: resource
handler_key: tip_rack
label: 枪头盒
output: output:
- data_key: liquid - data_key: sources.@flatten
data_source: executor data_source: executor
data_type: resource data_type: resource
handler_key: liquid-output handler_key: sources_out
io_type: source io_type: source
label: Liquid Output label: 移液后源孔
- data_key: targets.@flatten
data_source: executor
data_type: resource
handler_key: targets_out
label: 移液后目标孔
placeholder_keys: placeholder_keys:
sources: unilabos_resources sources: unilabos_resources
targets: unilabos_resources targets: unilabos_resources
@@ -9451,6 +9503,7 @@ liquid_handler.prcxi:
type: string type: string
type: array type: array
plate: plate:
items:
properties: properties:
category: category:
type: string type: string
@@ -9523,6 +9576,8 @@ liquid_handler.prcxi:
- data - data
title: plate title: plate
type: object type: object
title: plate
type: array
volumes: volumes:
items: items:
type: number type: number
@@ -9544,8 +9599,7 @@ liquid_handler.prcxi:
title: Plate title: Plate
type: array type: array
volumes: volumes:
items: items: {}
type: number
title: Volumes title: Volumes
type: array type: array
wells: wells:
@@ -9922,18 +9976,18 @@ liquid_handler.prcxi:
data_source: handle data_source: handle
data_type: resource data_type: resource
handler_key: tip_rack_identifier handler_key: tip_rack_identifier
label: 头盒 label: 头盒
output: output:
- data_key: liquid - data_key: sources.@flatten
data_source: handle data_source: executor
data_type: resource data_type: resource
handler_key: sources_out handler_key: sources_out
label: sources label: 移液后源孔
- data_key: liquid - data_key: targets.@flatten
data_source: executor data_source: executor
data_type: resource data_type: resource
handler_key: targets_out handler_key: targets_out
label: targets label: 移液后目标孔
placeholder_keys: placeholder_keys:
sources: unilabos_resources sources: unilabos_resources
targets: unilabos_resources targets: unilabos_resources

View File

@@ -0,0 +1,286 @@
motor.zdt_x42:
category:
- motor
class:
action_value_mappings:
auto-enable:
feedback: {}
goal: {}
goal_default:
'on': true
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 使能或禁用电机。使能后电机进入锁轴状态,可接收运动指令;禁用后电机进入松轴状态。
properties:
feedback: {}
goal:
properties:
'on':
default: true
type: boolean
required: []
type: object
result: {}
required:
- goal
title: enable参数
type: object
type: UniLabJsonCommand
auto-get_position:
feedback: {}
goal: {}
goal_default: {}
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 获取当前电机脉冲位置。
properties:
feedback: {}
goal:
properties: {}
required: []
type: object
result:
properties:
position:
type: integer
type: object
required:
- goal
title: get_position参数
type: object
type: UniLabJsonCommand
auto-move_position:
feedback: {}
goal: {}
goal_default:
absolute: false
acceleration: 10
direction: CW
pulses: 1000
speed_rpm: 60
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 位置模式运行。控制电机移动到指定脉冲位置或相对于当前位置移动指定脉冲数。
properties:
feedback: {}
goal:
properties:
absolute:
default: false
type: boolean
acceleration:
default: 10
maximum: 255
minimum: 0
type: integer
direction:
default: CW
enum:
- CW
- CCW
type: string
pulses:
default: 1000
type: integer
speed_rpm:
default: 60
minimum: 0
type: integer
required:
- pulses
- speed_rpm
type: object
result: {}
required:
- goal
title: move_position参数
type: object
type: UniLabJsonCommand
auto-move_speed:
feedback: {}
goal: {}
goal_default:
acceleration: 10
direction: CW
speed_rpm: 60
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 速度模式运行。控制电机以指定转速和方向持续转动。
properties:
feedback: {}
goal:
properties:
acceleration:
default: 10
maximum: 255
minimum: 0
type: integer
direction:
default: CW
enum:
- CW
- CCW
type: string
speed_rpm:
default: 60
minimum: 0
type: integer
required:
- speed_rpm
type: object
result: {}
required:
- goal
title: move_speed参数
type: object
type: UniLabJsonCommand
auto-rotate_quarter:
feedback: {}
goal: {}
goal_default:
direction: CW
speed_rpm: 60
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 电机旋转 1/4 圈 (阻塞式)。
properties:
feedback: {}
goal:
properties:
direction:
default: CW
enum:
- CW
- CCW
type: string
speed_rpm:
default: 60
minimum: 1
type: integer
required: []
type: object
result: {}
required:
- goal
title: rotate_quarter参数
type: object
type: UniLabJsonCommand
auto-set_zero:
feedback: {}
goal: {}
goal_default: {}
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 将当前电机位置设为零点。
properties:
feedback: {}
goal:
properties: {}
required: []
type: object
result: {}
required:
- goal
title: set_zero参数
type: object
type: UniLabJsonCommand
auto-stop:
feedback: {}
goal: {}
goal_default: {}
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 立即停止电机运动。
properties:
feedback: {}
goal:
properties: {}
required: []
type: object
result: {}
required:
- goal
title: stop参数
type: object
type: UniLabJsonCommand
auto-wait_time:
feedback: {}
goal: {}
goal_default:
duration_s: 1.0
handles: {}
placeholder_keys: {}
result: {}
schema:
description: 等待指定时间 (秒)。
properties:
feedback: {}
goal:
properties:
duration_s:
default: 1.0
minimum: 0
type: number
required:
- duration_s
type: object
result: {}
required:
- goal
title: wait_time参数
type: object
type: UniLabJsonCommand
module: unilabos.devices.motor.ZDT_X42:ZDTX42Driver
status_types:
position: int
status: str
type: python
config_info: []
description: ZDT X42 闭环步进电机驱动。支持速度运行、精确位置控制、位置查询和清零功能。适用于各种需要精确运动控制的实验室自动化场景。
handles: []
icon: ''
init_param_schema:
config:
properties:
baudrate:
default: 115200
type: integer
debug:
default: false
type: boolean
device_id:
default: 1
type: integer
port:
type: string
timeout:
default: 0.5
type: number
required:
- port
type: object
data:
properties:
position:
type: integer
status:
type: string
required:
- status
- position
type: object
version: 1.0.0

View File

@@ -0,0 +1,148 @@
sensor.xkc_rs485:
category:
- sensor
- separator
class:
action_value_mappings:
auto-change_baudrate:
goal:
baud_code: 7
handles: {}
schema:
description: '更改通讯波特率 (设置成功后无返回,且需手动切换波特率重连)。代码表 (16进制): 05=2400, 06=4800,
07=9600, 08=14400, 09=19200, 0A=28800, 0C=57600, 0D=115200, 0E=128000,
0F=256000'
properties:
goal:
properties:
baud_code:
description: '波特率代码 (例如: 7 为 9600, 13 即 0x0D 为 115200)'
type: integer
required:
- baud_code
type: object
type: UniLabJsonCommand
auto-change_device_id:
goal:
new_id: 1
handles: {}
schema:
description: 修改传感器的 Modbus 从站地址
properties:
goal:
properties:
new_id:
description: 新的从站地址 (1-254)
maximum: 254
minimum: 1
type: integer
required:
- new_id
type: object
type: UniLabJsonCommand
auto-factory_reset:
goal: {}
handles: {}
schema:
description: 恢复出厂设置 (地址重置为 01)
properties:
goal:
type: object
type: UniLabJsonCommand
auto-read_level:
goal: {}
handles: {}
schema:
description: 直接读取当前液位及信号强度
properties:
goal:
type: object
type: object
type: UniLabJsonCommand
auto-set_threshold:
goal:
threshold: 300
handles: {}
schema:
description: 设置液位判定阈值
properties:
goal:
properties:
threshold:
type: integer
required:
- threshold
type: object
type: UniLabJsonCommand
auto-wait_for_liquid:
goal:
target_state: true
timeout: 120
handles: {}
schema:
description: 实时检测电导率(RSSI)并等待用户指定的状态
properties:
goal:
properties:
target_state:
default: true
description: 目标状态 (True=有液, False=无液)
type: boolean
timeout:
default: 120
description: 超时时间 (秒)
required:
- target_state
type: object
type: UniLabJsonCommand
auto-wait_level:
goal:
level: true
timeout: 10
handles: {}
schema:
description: 等待液位达到目标状态
properties:
goal:
properties:
level:
type: boolean
timeout:
type: number
required:
- level
type: object
type: UniLabJsonCommand
module: unilabos.devices.separator.xkc_sensor:XKCSensorDriver
status_types:
level: bool
rssi: int
type: python
config_info: []
description: XKC RS485 非接触式液位传感器 (Modbus RTU)
handles: []
icon: ''
init_param_schema:
config:
properties:
baudrate:
default: 9600
type: integer
debug:
default: false
type: boolean
device_id:
default: 1
type: integer
port:
type: string
threshold:
default: 300
type: integer
timeout:
default: 3.0
type: number
required:
- port
type: object
version: 1.0.0

View File

@@ -46,3 +46,16 @@ BIOYOND_PolymerStation_8StockCarrier:
init_param_schema: {} init_param_schema: {}
registry_type: resource registry_type: resource
version: 1.0.0 version: 1.0.0
BIOYOND_PolymerStation_TipBox:
category:
- bottle_carriers
- tip_racks
class:
module: unilabos.resources.bioyond.bottle_carriers:BIOYOND_PolymerStation_TipBox
type: pylabrobot
description: BIOYOND_PolymerStation_TipBox (4x6布局24个枪头孔位)
handles: []
icon: ''
init_param_schema: {}
registry_type: resource
version: 1.0.0

View File

@@ -82,14 +82,3 @@ BIOYOND_PolymerStation_Solution_Beaker:
icon: '' icon: ''
init_param_schema: {} init_param_schema: {}
version: 1.0.0 version: 1.0.0
BIOYOND_PolymerStation_TipBox:
category:
- bottles
- tip_boxes
class:
module: unilabos.resources.bioyond.bottles:BIOYOND_PolymerStation_TipBox
type: pylabrobot
handles: []
icon: ''
init_param_schema: {}
version: 1.0.0

View File

@@ -1,4 +1,4 @@
from pylabrobot.resources import create_homogeneous_resources, Coordinate, ResourceHolder, create_ordered_items_2d from pylabrobot.resources import create_homogeneous_resources, Coordinate, ResourceHolder, create_ordered_items_2d, Container
from unilabos.resources.itemized_carrier import BottleCarrier from unilabos.resources.itemized_carrier import BottleCarrier
from unilabos.resources.bioyond.bottles import ( from unilabos.resources.bioyond.bottles import (
@@ -9,6 +9,28 @@ from unilabos.resources.bioyond.bottles import (
BIOYOND_PolymerStation_Reagent_Bottle, BIOYOND_PolymerStation_Reagent_Bottle,
BIOYOND_PolymerStation_Flask, BIOYOND_PolymerStation_Flask,
) )
def BIOYOND_PolymerStation_Tip(name: str, size_x: float = 8.0, size_y: float = 8.0, size_z: float = 50.0) -> Container:
"""创建单个枪头资源
Args:
name: 枪头名称
size_x: 枪头宽度 (mm)
size_y: 枪头长度 (mm)
size_z: 枪头高度 (mm)
Returns:
Container: 枪头容器
"""
return Container(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
category="tip",
model="BIOYOND_PolymerStation_Tip",
)
# 命名约定:试剂瓶-Bottle烧杯-Beaker烧瓶-Flask,小瓶-Vial # 命名约定:试剂瓶-Bottle烧杯-Beaker烧瓶-Flask,小瓶-Vial
@@ -322,3 +344,88 @@ def BIOYOND_Electrolyte_1BottleCarrier(name: str) -> BottleCarrier:
carrier.num_items_z = 1 carrier.num_items_z = 1
carrier[0] = BIOYOND_PolymerStation_Solution_Beaker(f"{name}_beaker_1") carrier[0] = BIOYOND_PolymerStation_Solution_Beaker(f"{name}_beaker_1")
return carrier return carrier
def BIOYOND_PolymerStation_TipBox(
name: str,
size_x: float = 127.76, # 枪头盒宽度
size_y: float = 85.48, # 枪头盒长度
size_z: float = 100.0, # 枪头盒高度
barcode: str = None,
) -> BottleCarrier:
"""创建4×6枪头盒 (24个枪头) - 使用 BottleCarrier 结构
Args:
name: 枪头盒名称
size_x: 枪头盒宽度 (mm)
size_y: 枪头盒长度 (mm)
size_z: 枪头盒高度 (mm)
barcode: 条形码
Returns:
BottleCarrier: 包含24个枪头孔位的枪头盒载架
布局说明:
- 4行×6列 (A-D, 1-6)
- 枪头孔位间距: 18mm (x方向) × 18mm (y方向)
- 起始位置居中对齐
- 索引顺序: 列优先 (0=A1, 1=B1, 2=C1, 3=D1, 4=A2, ...)
"""
# 枪头孔位参数
num_cols = 6 # 1-6 (x方向)
num_rows = 4 # A-D (y方向)
tip_diameter = 8.0 # 枪头孔位直径
tip_spacing_x = 18.0 # 列间距 (增加到18mm更宽松)
tip_spacing_y = 18.0 # 行间距 (增加到18mm更宽松)
# 计算起始位置 (居中对齐)
total_width = (num_cols - 1) * tip_spacing_x + tip_diameter
total_height = (num_rows - 1) * tip_spacing_y + tip_diameter
start_x = (size_x - total_width) / 2
start_y = (size_y - total_height) / 2
# 使用 create_ordered_items_2d 创建孔位
# create_ordered_items_2d 返回的 key 是数字索引: 0, 1, 2, ...
# 顺序是列优先: 先y后x (即 0=A1, 1=B1, 2=C1, 3=D1, 4=A2, 5=B2, ...)
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=num_cols,
num_items_y=num_rows,
dx=start_x,
dy=start_y,
dz=5.0,
item_dx=tip_spacing_x,
item_dy=tip_spacing_y,
size_x=tip_diameter,
size_y=tip_diameter,
size_z=50.0, # 枪头深度
)
# 更新 sites 中每个 ResourceHolder 的名称
for k, v in sites.items():
v.name = f"{name}_{v.name}"
# 创建枪头盒载架
# 注意:不设置 category使用默认的 "bottle_carrier",这样前端会显示为完整的矩形载架
tip_box = BottleCarrier(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
sites=sites, # 直接使用数字索引的 sites
model="BIOYOND_PolymerStation_TipBox",
)
# 设置自定义属性
tip_box.barcode = barcode
tip_box.tip_count = 24 # 4行×6列
tip_box.num_items_x = num_cols
tip_box.num_items_y = num_rows
tip_box.num_items_z = 1
# ⭐ 枪头盒不需要放入子资源
# 与其他 carrier 不同,枪头盒在 Bioyond 中是一个整体
# 不需要追踪每个枪头的状态,保持为空的 ResourceHolder 即可
# 这样前端会显示24个空槽位可以用于放置枪头
return tip_box

View File

@@ -116,7 +116,9 @@ def BIOYOND_PolymerStation_TipBox(
size_z: float = 100.0, # 枪头盒高度 size_z: float = 100.0, # 枪头盒高度
barcode: str = None, barcode: str = None,
): ):
"""创建4×6枪头盒 (24个枪头) """创建4×6枪头盒 (24个枪头) - 使用 BottleCarrier 结构
注意:此函数已弃用,请使用 bottle_carriers.py 中的版本
Args: Args:
name: 枪头盒名称 name: 枪头盒名称
@@ -126,55 +128,11 @@ def BIOYOND_PolymerStation_TipBox(
barcode: 条形码 barcode: 条形码
Returns: Returns:
TipBoxCarrier: 包含24个枪头孔位的枪头盒 BottleCarrier: 包含24个枪头孔位的枪头盒载架
""" """
from pylabrobot.resources import Container, Coordinate # 重定向到 bottle_carriers.py 中的实现
from unilabos.resources.bioyond.bottle_carriers import BIOYOND_PolymerStation_TipBox as TipBox_Carrier
# 创建枪头盒容器 return TipBox_Carrier(name=name, size_x=size_x, size_y=size_y, size_z=size_z, barcode=barcode)
tip_box = Container(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
category="tip_rack",
model="BIOYOND_PolymerStation_TipBox_4x6",
)
# 设置自定义属性
tip_box.barcode = barcode
tip_box.tip_count = 24 # 4行×6列
tip_box.num_items_x = 6 # 6列
tip_box.num_items_y = 4 # 4行
# 创建24个枪头孔位 (4行×6列)
# 假设孔位间距为 9mm
tip_spacing_x = 9.0 # 列间距
tip_spacing_y = 9.0 # 行间距
start_x = 14.38 # 第一个孔位的x偏移
start_y = 11.24 # 第一个孔位的y偏移
for row in range(4): # A, B, C, D
for col in range(6): # 1-6
spot_name = f"{chr(65 + row)}{col + 1}" # A1, A2, ..., D6
x = start_x + col * tip_spacing_x
y = start_y + row * tip_spacing_y
# 创建枪头孔位容器
tip_spot = Container(
name=spot_name,
size_x=8.0, # 单个枪头孔位大小
size_y=8.0,
size_z=size_z - 10.0, # 略低于盒子高度
category="tip_spot",
)
# 添加到枪头盒
tip_box.assign_child_resource(
tip_spot,
location=Coordinate(x=x, y=y, z=0)
)
return tip_box
def BIOYOND_PolymerStation_Flask( def BIOYOND_PolymerStation_Flask(

View File

@@ -151,12 +151,40 @@ def canonicalize_links_ports(links: List[Dict[str, Any]], resource_tree_set: Res
""" """
# 构建 id 到 uuid 的映射 # 构建 id 到 uuid 的映射
id_to_uuid: Dict[str, str] = {} id_to_uuid: Dict[str, str] = {}
uuid_to_id: Dict[str, str] = {}
for node in resource_tree_set.all_nodes: for node in resource_tree_set.all_nodes:
id_to_uuid[node.res_content.id] = node.res_content.uuid id_to_uuid[node.res_content.id] = node.res_content.uuid
uuid_to_id[node.res_content.uuid] = node.res_content.id
# 第三遍处理:为每个 link 添加 source_uuid 和 target_uuid
for link in links:
source_id = link.get("source")
target_id = link.get("target")
# 添加 source_uuid
if source_id and source_id in id_to_uuid:
link["source_uuid"] = id_to_uuid[source_id]
# 添加 target_uuid
if target_id and target_id in id_to_uuid:
link["target_uuid"] = id_to_uuid[target_id]
source_uuid = link.get("source_uuid")
target_uuid = link.get("target_uuid")
# 添加 source_uuid
if source_uuid and source_uuid in uuid_to_id:
link["source"] = uuid_to_id[source_uuid]
# 添加 target_uuid
if target_uuid and target_uuid in uuid_to_id:
link["target"] = uuid_to_id[target_uuid]
# 第一遍处理将字符串类型的port转换为字典格式 # 第一遍处理将字符串类型的port转换为字典格式
for link in links: for link in links:
port = link.get("port") port = link.get("port")
if port is None:
continue
if link.get("type", "physical") == "physical": if link.get("type", "physical") == "physical":
link["type"] = "fluid" link["type"] = "fluid"
if isinstance(port, int): if isinstance(port, int):
@@ -179,13 +207,15 @@ def canonicalize_links_ports(links: List[Dict[str, Any]], resource_tree_set: Res
link["port"] = {link["source"]: None, link["target"]: None} link["port"] = {link["source"]: None, link["target"]: None}
# 构建边字典,键为(source节点, target节点)值为对应的port信息 # 构建边字典,键为(source节点, target节点)值为对应的port信息
edges = {(link["source"], link["target"]): link["port"] for link in links} edges = {(link["source"], link["target"]): link["port"] for link in links if link.get("port")}
# 第二遍处理填充反向边的dest信息 # 第二遍处理填充反向边的dest信息
delete_reverses = [] delete_reverses = []
for i, link in enumerate(links): for i, link in enumerate(links):
s, t = link["source"], link["target"] s, t = link["source"], link["target"]
current_port = link["port"] current_port = link.get("port")
if current_port is None:
continue
if current_port.get(t) is None: if current_port.get(t) is None:
reverse_key = (t, s) reverse_key = (t, s)
reverse_port = edges.get(reverse_key) reverse_port = edges.get(reverse_key)
@@ -200,20 +230,6 @@ def canonicalize_links_ports(links: List[Dict[str, Any]], resource_tree_set: Res
current_port[t] = current_port[s] current_port[t] = current_port[s]
# 删除已被使用反向端口信息的反向边 # 删除已被使用反向端口信息的反向边
standardized_links = [link for i, link in enumerate(links) if i not in delete_reverses] standardized_links = [link for i, link in enumerate(links) if i not in delete_reverses]
# 第三遍处理:为每个 link 添加 source_uuid 和 target_uuid
for link in standardized_links:
source_id = link.get("source")
target_id = link.get("target")
# 添加 source_uuid
if source_id and source_id in id_to_uuid:
link["source_uuid"] = id_to_uuid[source_id]
# 添加 target_uuid
if target_id and target_id in id_to_uuid:
link["target_uuid"] = id_to_uuid[target_id]
return standardized_links return standardized_links
@@ -284,6 +300,8 @@ def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]
edge["sourceHandle"] = port[source] edge["sourceHandle"] = port[source]
elif "source_port" in edge: elif "source_port" in edge:
edge["sourceHandle"] = edge.pop("source_port") edge["sourceHandle"] = edge.pop("source_port")
elif "source_handle" in edge:
edge["sourceHandle"] = edge.pop("source_handle")
else: else:
typ = edge.get("type") typ = edge.get("type")
if typ == "communication": if typ == "communication":
@@ -292,6 +310,8 @@ def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]
edge["targetHandle"] = port[target] edge["targetHandle"] = port[target]
elif "target_port" in edge: elif "target_port" in edge:
edge["targetHandle"] = edge.pop("target_port") edge["targetHandle"] = edge.pop("target_port")
elif "target_handle" in edge:
edge["targetHandle"] = edge.pop("target_handle")
else: else:
typ = edge.get("type") typ = edge.get("type")
if typ == "communication": if typ == "communication":
@@ -759,6 +779,9 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
bottle = plr_material[number] = initialize_resource( bottle = plr_material[number] = initialize_resource(
{"name": f'{detail["name"]}_{number}', "class": reverse_type_mapping[typeName][0]}, resource_type=ResourcePLR {"name": f'{detail["name"]}_{number}', "class": reverse_type_mapping[typeName][0]}, resource_type=ResourcePLR
) )
# 只有具有 tracker 属性的容器才设置液体信息(如 Bottle, Well
# ResourceHolder 等不支持液体追踪的容器跳过
if hasattr(bottle, "tracker"):
bottle.tracker.liquids = [ bottle.tracker.liquids = [
(detail["name"], float(detail.get("quantity", 0)) if detail.get("quantity") else 0) (detail["name"], float(detail.get("quantity", 0)) if detail.get("quantity") else 0)
] ]
@@ -770,6 +793,8 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
# 只对有 capacity 属性的容器(液体容器)处理液体追踪 # 只对有 capacity 属性的容器(液体容器)处理液体追踪
if hasattr(plr_material, 'capacity'): if hasattr(plr_material, 'capacity'):
bottle = plr_material[0] if plr_material.capacity > 0 else plr_material bottle = plr_material[0] if plr_material.capacity > 0 else plr_material
# 确保 bottle 有 tracker 属性才设置液体信息
if hasattr(bottle, "tracker"):
bottle.tracker.liquids = [ bottle.tracker.liquids = [
(material["name"], float(material.get("quantity", 0)) if material.get("quantity") else 0) (material["name"], float(material.get("quantity", 0)) if material.get("quantity") else 0)
] ]
@@ -801,24 +826,29 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
wh_name = loc.get("whName") wh_name = loc.get("whName")
logger.debug(f"[物料位置] {unique_name} 尝试放置到 warehouse: {wh_name} (Bioyond坐标: x={loc.get('x')}, y={loc.get('y')}, z={loc.get('z')})") logger.debug(f"[物料位置] {unique_name} 尝试放置到 warehouse: {wh_name} (Bioyond坐标: x={loc.get('x')}, y={loc.get('y')}, z={loc.get('z')})")
# Bioyond坐标映射 (重要!): x→行(1=A,2=B...), y→列(1=01,2=02...), z→层(通常=1)
# 必须在warehouse映射之前先获取坐标以便后续调整
x = loc.get("x", 1) # 行号 (1-based: 1=A, 2=B, 3=C, 4=D)
y = loc.get("y", 1) # 列号 (1-based: 1=01, 2=02, 3=03...)
z = loc.get("z", 1) # 层号 (1-based, 通常为1)
# 特殊处理: Bioyond的"堆栈1"需要映射到"堆栈1左"或"堆栈1右" # 特殊处理: Bioyond的"堆栈1"需要映射到"堆栈1左"或"堆栈1右"
# 根据列号(x)判断: 1-4映射到左侧, 5-8映射到右侧 # 根据列号(y)判断: 1-4映射到左侧, 5-8映射到右侧
if wh_name == "堆栈1": if wh_name == "堆栈1":
x_val = loc.get("x", 1) if 1 <= y <= 4:
if 1 <= x_val <= 4:
wh_name = "堆栈1左" wh_name = "堆栈1左"
elif 5 <= x_val <= 8: elif 5 <= y <= 8:
wh_name = "堆栈1右" wh_name = "堆栈1右"
y = y - 4 # 调整列号: 5-8映射到1-4
else: else:
logger.warning(f"物料 {material['name']} 的列号 x={x_val} 超出范围无法映射到堆栈1左或堆栈1右") logger.warning(f"物料 {material['name']} 的列号 y={y} 超出范围无法映射到堆栈1左或堆栈1右")
continue continue
# 特殊处理: Bioyond的"站内Tip盒堆栈"也需要进行拆分映射 # 特殊处理: Bioyond的"站内Tip盒堆栈"也需要进行拆分映射
if wh_name == "站内Tip盒堆栈": if wh_name == "站内Tip盒堆栈":
y_val = loc.get("y", 1) if y == 1:
if y_val == 1:
wh_name = "站内Tip盒堆栈(右)" wh_name = "站内Tip盒堆栈(右)"
elif y_val in [2, 3]: elif y in [2, 3]:
wh_name = "站内Tip盒堆栈(左)" wh_name = "站内Tip盒堆栈(左)"
y = y - 1 # 调整列号,因为左侧仓库对应的 Bioyond y=2 实际上是它的第1列 y = y - 1 # 调整列号,因为左侧仓库对应的 Bioyond y=2 实际上是它的第1列
@@ -826,15 +856,6 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
warehouse = deck.warehouses[wh_name] warehouse = deck.warehouses[wh_name]
logger.debug(f"[Warehouse匹配] 找到warehouse: {wh_name} (容量: {warehouse.capacity}, 行×列: {warehouse.num_items_x}×{warehouse.num_items_y})") logger.debug(f"[Warehouse匹配] 找到warehouse: {wh_name} (容量: {warehouse.capacity}, 行×列: {warehouse.num_items_x}×{warehouse.num_items_y})")
# Bioyond坐标映射 (重要!): x→行(1=A,2=B...), y→列(1=01,2=02...), z→层(通常=1)
x = loc.get("x", 1) # 行号 (1-based: 1=A, 2=B, 3=C, 4=D)
y = loc.get("y", 1) # 列号 (1-based: 1=01, 2=02, 3=03...)
z = loc.get("z", 1) # 层号 (1-based, 通常为1)
# 如果是右侧堆栈,需要调整列号 (5→1, 6→2, 7→3, 8→4)
if wh_name == "堆栈1右":
y = y - 4 # 将5-8映射到1-4
# 特殊处理竖向warehouse站内试剂存放堆栈、测量小瓶仓库 # 特殊处理竖向warehouse站内试剂存放堆栈、测量小瓶仓库
# 这些warehouse使用 vertical-col-major 布局 # 这些warehouse使用 vertical-col-major 布局
if wh_name in ["站内试剂存放堆栈", "测量小瓶仓库(测密度)"]: if wh_name in ["站内试剂存放堆栈", "测量小瓶仓库(测密度)"]:

View File

@@ -18,3 +18,9 @@ def register():
from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
from unilabos.devices.liquid_handling.laiyu.backend.laiyu_v_backend import UniLiquidHandlerLaiyuBackend from unilabos.devices.liquid_handling.laiyu.backend.laiyu_v_backend import UniLiquidHandlerLaiyuBackend
# noinspection PyUnresolvedReferences
from unilabos.resources.bioyond.decks import (
BIOYOND_PolymerReactionStation_Deck,
BIOYOND_PolymerPreparationStation_Deck,
BIOYOND_YB_Deck,
)

View File

@@ -341,6 +341,7 @@ class ResourceTreeSet(object):
"deck": "deck", "deck": "deck",
"tip_rack": "tip_rack", "tip_rack": "tip_rack",
"tip_spot": "tip_spot", "tip_spot": "tip_spot",
"tip": "tip", # 添加 tip 类型支持
"tube": "tube", "tube": "tube",
"bottle_carrier": "bottle_carrier", "bottle_carrier": "bottle_carrier",
} }

View File

@@ -1581,7 +1581,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
f"转换ResourceSlot列表参数 {arg_name} 失败: {e}\n{traceback.format_exc()}" f"转换ResourceSlot列表参数 {arg_name} 失败: {e}\n{traceback.format_exc()}"
) )
raise JsonCommandInitError(f"ResourceSlot列表参数转换失败: {arg_name}") raise JsonCommandInitError(f"ResourceSlot列表参数转换失败: {arg_name}")
# todo: 默认反报送
return function(**function_args) return function(**function_args)
except KeyError as ex: except KeyError as ex:
raise JsonCommandInitError( raise JsonCommandInitError(
@@ -1614,8 +1614,8 @@ class BaseROS2DeviceNode(Node, Generic[T]):
timeout = 30.0 timeout = 30.0
elapsed = 0.0 elapsed = 0.0
while not future.done() and elapsed < timeout: while not future.done() and elapsed < timeout:
time.sleep(0.05) time.sleep(0.02)
elapsed += 0.05 elapsed += 0.02
if not future.done(): if not future.done():
raise Exception(f"资源查询超时: {uuids_list}") raise Exception(f"资源查询超时: {uuids_list}")

View File

@@ -807,7 +807,7 @@ class HostNode(BaseROS2DeviceNode):
assign_sample_id(action_kwargs) assign_sample_id(action_kwargs)
goal_msg = convert_to_ros_msg(action_client._action_type.Goal(), action_kwargs) goal_msg = convert_to_ros_msg(action_client._action_type.Goal(), action_kwargs)
self.lab_logger().info(f"[Host Node] Sending goal for {action_id}: {str(goal_msg)[:1000]}") # self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {str(goal_msg)[:1000]}")
self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {action_kwargs}") self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {action_kwargs}")
self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {goal_msg}") self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {goal_msg}")
action_client.wait_for_server() action_client.wait_for_server()
@@ -1180,7 +1180,7 @@ class HostNode(BaseROS2DeviceNode):
""" """
更新节点信息回调 更新节点信息回调
""" """
# self.lab_logger().info(f"[Host Node] Node info update request received: {request}") self.lab_logger().trace(f"[Host Node] Node info update request received: {request}")
try: try:
from unilabos.app.communication import get_communication_client from unilabos.app.communication import get_communication_client
from unilabos.app.web.client import HTTPClient, http_client from unilabos.app.web.client import HTTPClient, http_client

View File

@@ -0,0 +1,795 @@
{
"nodes": [
{
"id": "PRCXI",
"name": "PRCXI",
"type": "device",
"class": "liquid_handler.prcxi",
"parent": "",
"pose": {
"size": {
"width": 562,
"height": 394,
"depth": 0
}
},
"config": {
"axis": "Left",
"deck": {
"_resource_type": "unilabos.devices.liquid_handling.prcxi.prcxi:PRCXI9300Deck",
"_resource_child_name": "PRCXI_Deck"
},
"host": "10.20.30.184",
"port": 9999,
"debug": true,
"setup": true,
"is_9320": true,
"timeout": 10,
"matrix_id": "5de524d0-3f95-406c-86dd-f83626ebc7cb",
"simulator": true,
"channel_num": 2
},
"data": {
"reset_ok": true
},
"schema": {},
"description": "",
"model": null,
"position": {
"x": 0,
"y": 240,
"z": 0
}
},
{
"id": "PRCXI_Deck",
"name": "PRCXI_Deck",
"children": [],
"parent": "PRCXI",
"type": "deck",
"class": "",
"position": {
"x": 10,
"y": 10,
"z": 0
},
"config": {
"type": "PRCXI9300Deck",
"size_x": 542,
"size_y": 374,
"size_z": 0,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "deck",
"barcode": null
},
"data": {}
},
{
"id": "T1",
"name": "T1",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 0,
"y": 288,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T1",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T2",
"name": "T2",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 138,
"y": 288,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T2",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T3",
"name": "T3",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 276,
"y": 288,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T3",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T4",
"name": "T4",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 414,
"y": 288,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T4",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T5",
"name": "T5",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 0,
"y": 192,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T5",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T6",
"name": "T6",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 138,
"y": 192,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T6",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T7",
"name": "T7",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 276,
"y": 192,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T7",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T8",
"name": "T8",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 414,
"y": 192,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T8",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T9",
"name": "T9",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 0,
"y": 96,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T9",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T10",
"name": "T10",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 138,
"y": 96,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T10",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T11",
"name": "T11",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 276,
"y": 96,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T11",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T12",
"name": "T12",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 414,
"y": 96,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T12",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T13",
"name": "T13",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 0,
"y": 0,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T13",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T14",
"name": "T14",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 138,
"y": 0,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T14",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T15",
"name": "T15",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 276,
"y": 0,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T15",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
},
{
"id": "T16",
"name": "T16",
"children": [],
"parent": "PRCXI_Deck",
"type": "plate",
"class": "",
"position": {
"x": 414,
"y": 0,
"z": 0
},
"config": {
"type": "PRCXI9300Container",
"size_x": 127,
"size_y": 85.5,
"size_z": 10,
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "plate",
"model": null,
"barcode": null,
"ordering": {},
"sites": [
{
"label": "T16",
"visible": true,
"position": { "x": 0, "y": 0, "z": 0 },
"size": { "width": 128.0, "height": 86, "depth": 0 },
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack"
]
}
]
},
"data": {}
}
],
"edges": []
}

View File

@@ -0,0 +1,29 @@
{
"nodes": [
{
"id": "Liquid_Sensor_1",
"name": "XKC Sensor",
"children": [],
"parent": null,
"type": "device",
"class": "sensor.xkc_rs485",
"position": {
"x": 0,
"y": 0,
"z": 0
},
"config": {
"port": "/dev/tty.usbserial-3110",
"baudrate": 9600,
"device_id": 1,
"threshold": 300,
"timeout": 3.0
},
"data": {
"level": false,
"rssi": 0
}
}
],
"links": []
}

View File

@@ -0,0 +1,28 @@
{
"nodes": [
{
"id": "ZDT_Motor",
"name": "ZDT Motor",
"children": [],
"parent": null,
"type": "device",
"class": "motor.zdt_x42",
"position": {
"x": 0,
"y": 0,
"z": 0
},
"config": {
"port": "/dev/tty.usbserial-3110",
"baudrate": 115200,
"device_id": 1,
"debug": true
},
"data": {
"position": 0,
"status": "idle"
}
}
],
"links": []
}

View File

@@ -19,7 +19,9 @@
第一步: 按 slot 去重创建 create_resource 节点(创建板子) 第一步: 按 slot 去重创建 create_resource 节点(创建板子)
-------------------------------------------------------------------------------- --------------------------------------------------------------------------------
- 首先创建一个 Group 节点type="Group", minimized=true用于包含所有 create_resource 节点
- 遍历所有 reagent按 slot 去重,为每个唯一的 slot 创建一个板子 - 遍历所有 reagent按 slot 去重,为每个唯一的 slot 创建一个板子
- 所有 create_resource 节点的 parent_uuid 指向 Group 节点minimized=true
- 生成参数: - 生成参数:
res_id: plate_slot_{slot} res_id: plate_slot_{slot}
device_id: /PRCXI device_id: /PRCXI
@@ -29,11 +31,13 @@
- 输出端口: labware用于连接 set_liquid_from_plate - 输出端口: labware用于连接 set_liquid_from_plate
- 控制流: create_resource 之间通过 ready 端口串联 - 控制流: create_resource 之间通过 ready 端口串联
示例: slot=1, slot=4 -> 创建 2 个 create_resource 节点 示例: slot=1, slot=4 -> 创建 1 个 Group + 2 个 create_resource 节点
第二步: 为每个 reagent 创建 set_liquid_from_plate 节点(设置液体) 第二步: 为每个 reagent 创建 set_liquid_from_plate 节点(设置液体)
-------------------------------------------------------------------------------- --------------------------------------------------------------------------------
- 首先创建一个 Group 节点type="Group", minimized=true用于包含所有 set_liquid_from_plate 节点
- 遍历所有 reagent为每个试剂创建 set_liquid_from_plate 节点 - 遍历所有 reagent为每个试剂创建 set_liquid_from_plate 节点
- 所有 set_liquid_from_plate 节点的 parent_uuid 指向 Group 节点minimized=true
- 生成参数: - 生成参数:
plate: [](通过连接传递,来自 create_resource 的 labware plate: [](通过连接传递,来自 create_resource 的 labware
well_names: ["A1", "A3", "A5"](来自 reagent 的 well 数组) well_names: ["A1", "A3", "A5"](来自 reagent 的 well 数组)
@@ -76,6 +80,13 @@ transfer_liquid:
输入: sources -> sources_identifier, targets -> targets_identifier 输入: sources -> sources_identifier, targets -> targets_identifier
输出: sources -> sources_out, targets -> targets_out 输出: sources -> sources_out, targets -> targets_out
==================== 设备名配置 (device_name) ====================
每个节点都有 device_name 字段,指定在哪个设备上执行:
- create_resource: device_name = "host_node"(固定)
- set_liquid_from_plate: device_name = "PRCXI"(可配置,见 DEVICE_NAME_DEFAULT
- transfer_liquid 等动作: device_name = "PRCXI"(可配置,见 DEVICE_NAME_DEFAULT
==================== 校验规则 ==================== ==================== 校验规则 ====================
- 检查 sources/targets 是否在 reagent 中定义 - 检查 sources/targets 是否在 reagent 中定义
@@ -97,6 +108,13 @@ Json = Dict[str, Any]
# ==================== 默认配置 ==================== # ==================== 默认配置 ====================
# 设备名配置
DEVICE_NAME_HOST = "host_node" # create_resource 固定在 host_node 上执行
DEVICE_NAME_DEFAULT = "PRCXI" # transfer_liquid, set_liquid_from_plate 等动作的默认设备名
# 节点类型
NODE_TYPE_DEFAULT = "ILab" # 所有节点的默认类型
# create_resource 节点默认参数 # create_resource 节点默认参数
CREATE_RESOURCE_DEFAULTS = { CREATE_RESOURCE_DEFAULTS = {
"device_id": "/PRCXI", "device_id": "/PRCXI",
@@ -367,6 +385,21 @@ def build_protocol_graph(
"res_id": res_id, "res_id": res_id,
} }
# 创建 Group 节点,包含所有 create_resource 节点
group_node_id = str(uuid.uuid4())
G.add_node(
group_node_id,
name="Resources Group",
type="Group",
parent_uuid="",
lab_node_type="Device",
template_name="",
resource_name="",
footer="",
minimized=True,
param=None,
)
# 为每个唯一的 slot 创建 create_resource 节点 # 为每个唯一的 slot 创建 create_resource 节点
res_index = 0 res_index = 0
last_create_resource_id = None last_create_resource_id = None
@@ -383,6 +416,10 @@ def build_protocol_graph(
description=f"Create plate on slot {slot}", description=f"Create plate on slot {slot}",
lab_node_type="Labware", lab_node_type="Labware",
footer="create_resource-host_node", footer="create_resource-host_node",
device_name=DEVICE_NAME_HOST,
type=NODE_TYPE_DEFAULT,
parent_uuid=group_node_id, # 指向 Group 节点
minimized=True, # 折叠显示
param={ param={
"res_id": res_id, "res_id": res_id,
"device_id": CREATE_RESOURCE_DEFAULTS["device_id"], "device_id": CREATE_RESOURCE_DEFAULTS["device_id"],
@@ -400,6 +437,21 @@ def build_protocol_graph(
last_create_resource_id = node_id last_create_resource_id = node_id
# ==================== 第二步:为每个 reagent 创建 set_liquid_from_plate 节点 ==================== # ==================== 第二步:为每个 reagent 创建 set_liquid_from_plate 节点 ====================
# 创建 Group 节点,包含所有 set_liquid_from_plate 节点
set_liquid_group_id = str(uuid.uuid4())
G.add_node(
set_liquid_group_id,
name="SetLiquid Group",
type="Group",
parent_uuid="",
lab_node_type="Device",
template_name="",
resource_name="",
footer="",
minimized=True,
param=None,
)
set_liquid_index = 0 set_liquid_index = 0
last_set_liquid_id = last_create_resource_id # set_liquid_from_plate 连接在 create_resource 之后 last_set_liquid_id = last_create_resource_id # set_liquid_from_plate 连接在 create_resource 之后
@@ -430,6 +482,10 @@ def build_protocol_graph(
description=f"Set liquid: {labware_id}", description=f"Set liquid: {labware_id}",
lab_node_type="Reagent", lab_node_type="Reagent",
footer="set_liquid_from_plate-liquid_handler.prcxi", footer="set_liquid_from_plate-liquid_handler.prcxi",
device_name=DEVICE_NAME_DEFAULT,
type=NODE_TYPE_DEFAULT,
parent_uuid=set_liquid_group_id, # 指向 Group 节点
minimized=True, # 折叠显示
param={ param={
"plate": [], # 通过连接传递 "plate": [], # 通过连接传递
"well_names": wells, # 孔位名数组,如 ["A1", "A3", "A5"] "well_names": wells, # 孔位名数组,如 ["A1", "A3", "A5"]
@@ -544,9 +600,11 @@ def build_protocol_graph(
if param_key in params: if param_key in params:
params[param_key] = [] params[param_key] = []
# 更新 step 的 paramfooter # 更新 step 的 paramfooter、device_name 和 type
step_copy = step.copy() step_copy = step.copy()
step_copy["param"] = params step_copy["param"] = params
step_copy["device_name"] = DEVICE_NAME_DEFAULT # 动作节点使用默认设备名
step_copy["type"] = NODE_TYPE_DEFAULT # 节点类型
# 如果有警告,修改 footer 添加警告标记(警告放前面) # 如果有警告,修改 footer 添加警告标记(警告放前面)
if warnings: if warnings: