open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/deepmodeling/Uni-Lab-OS synced 2026-04-27 23:20:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Update workstation reference and templates with new PLC integration details and enhanced workflow mappings

Browse Source
This commit is contained in:
ZiWei
2026-03-11 14:09:46 +08:00
parent 92bfb069d5
commit fe501c965f
3 changed files with 1064 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files

506
.cursor/skills/add-workstation/SKILL.md
View File

@@ -5,254 +5,89 @@ description: Guide for adding new workstations to Uni-Lab-OS (接入新工作站
# Uni-Lab-OS 工作站接入指南
工作站（workstation）是组合多个子设备的大型设备，拥有独立的物料管理系统（PLR Deck）和工作流引擎。本指南覆盖从需求分析到验证的全流程。
工作站是组合多个子设备的大型设备，拥有独立的物料管理系统（PLR Deck）和工作流引擎。
> **前置知识**：工作站接入基于 `docs/ai_guides/add_device.md` 的通用设备接入框架，但有显著差异。阅读本指南前无需先读通用指南。
> **完整代码模板**见 [templates.md](templates.md)，**高级模式**见 [reference.md](reference.md)。
## 第一步：确定工作站类型
向用户确认以下信息：
**Q1: 工作站的业务场景？**
向用户确认：
| 类型 | 基类 | 适用场景 | 示例 |
|------|------|----------|------|
| **Protocol 工作站** | `ProtocolNode` | 标准化学操作协议（过滤、转移、加热等） | FilterProtocolStation |
| **外部系统工作站** | `WorkstationBase` | 与外部 LIMS/MES 系统对接，有专属 API | BioyondStation |
| **硬件控制工作站** | `WorkstationBase` | 直接控制 PLC/硬件，无外部系统 | CoinCellAssembly |
| **Protocol** | `ProtocolNode` | 标准化学操作协议 | FilterProtocolStation |
| **外部系统** | `WorkstationBase` | 对接 LIMS/MES API | BioyondStation |
| **硬件控制** | `WorkstationBase` | 直接控制 PLC/硬件 | CoinCellAssembly |
**Q2: 工作站英文名称？**（如 `my_reaction_station`）
**Q3: 与外部系统的交互方式？**
| 方式 | 适用场景 | 需要的配置 |
|------|----------|-----------|
| 无外部系统 | Protocol 工作站、纯硬件控制 | 无 |
| HTTP API | LIMS/MES 系统（如 Bioyond） | `api_host`, `api_key` |
| Modbus TCP | PLC 控制 | `address`, `port` |
| OPC UA | 工业设备 | `url` |
**Q4: 子设备组成？**
- 列出所有子设备（如反应器、泵、阀、传感器等）
- 哪些是已有设备类型？哪些需要新增？
- 子设备之间的硬件代理关系（如泵通过串口设备通信）
**Q5: 物料管理需求？**
- 是否需要 Deck（物料面板）？
- 物料类型（plate、tip_rack、bottle 等）
- 是否需要与外部物料系统同步？
还需确认：
- 英文名称、通信方式（HTTP/Modbus/OPC UA/无）
- 子设备组成（哪些已有、哪些新增、硬件代理关系）
- 物料需求（是否需要 Deck、物料类型、是否需外部同步）
---
## 第二步：理解工作站架构
工作站与普通设备的核心差异：
| 维度 | 普通设备 | 工作站 |
|------|---------|--------|
| 基类 | 无（纯 Python 类） | `WorkstationBase` 或 `ProtocolNode` |
| 基类 | 纯 Python 类 | `WorkstationBase` / `ProtocolNode` |
| ROS 节点 | `BaseROS2DeviceNode` | `ROS2WorkstationNode` |
| 状态管理 | `self.data` 字典 | 通常不用 `self.data`，用 `@property` 直接访问 |
| 子设备 | 无 | `children` 列表，通过 `self._children` 访问 |
| 状态管理 | `self.data` 字典 | `@property` 直接访问 |
| 子设备 | 无 | `self._children` / `self._ros_node.sub_devices` |
| 物料 | 无 | `self.deck`（PLR Deck） |
| 图文件角色 | `parent: null` 或 `parent: "<station>"` | `parent: null`，含 `children` 和 `deck` |
### 继承体系
`WorkstationBase` (ABC) → `ProtocolNode` (通用协议) / `BioyondWorkstation` (→ ReactionStation, DispensingStation) / `CoinCellAssemblyWorkstation` (硬件控制)
```
WorkstationBase (ABC)
├── BioyondWorkstation ← HTTP RPC + 资源同步
│ ├── BioyondReactionStation
│ └── BioyondDispensingStation
├── CoinCellAssemblyWorkstation ← Modbus/PLC
└── ProtocolNode ← 标准化学协议
```
### ROS 层
### 子设备初始化流程
`ROS2WorkstationNode` 额外负责：初始化 children 子设备节点、为子设备创建 ActionClient、配置硬件代理、为 protocol_type 创建协议 ActionServer。
`ROS2WorkstationNode.__init__` → 遍历 `children`（type=="device"）→ `initialize_device_from_dict()` → 存入 `sub_devices` → 为每个动作创建 `ActionClient` → 识别通信设备（`serial_*`/`io_*`）→ `_setup_hardware_proxy()`
---
## 第三步：创建驱动文件
文件路径：`unilabos/devices/workstation/<station_name>/<station_name>.py`
路径：`unilabos/devices/workstation/<station_name>/<station_name>.py`
### 模板 A：基于外部系统的工作站
根据类型选择模板（完整代码见 [templates.md](templates.md)）：
适用于与 LIMS/MES 等外部系统对接的场景。
| 类型 | 模板 | 关键要素 |
|------|------|---------|
| 外部系统 | Template A | `config` 接收 API 配置，`post_init` 启动 RPC/HTTP 服务 |
| 硬件控制 | Template B | `TCPClient` + CSV 寄存器映射，`use_node()` 读写 |
| Protocol | Template C | 直接使用 `ProtocolNode`，通常不需要自定义类 |
```python
import logging
from typing import Dict, Any, Optional, List
from pylabrobot.resources import Deck
from unilabos.devices.workstation.workstation_base import WorkstationBase
try:
from unilabos.ros.nodes.presets.workstation import ROS2WorkstationNode
except ImportError:
ROS2WorkstationNode = None
class MyWorkstation(WorkstationBase):
"""工作站描述"""
_ros_node: "ROS2WorkstationNode"
def __init__(
self,
config: dict = None,
deck: Optional[Deck] = None,
protocol_type: list = None,
**kwargs,
):
super().__init__(deck=deck, **kwargs)
self.config = config or {}
self.logger = logging.getLogger(f"MyWorkstation")
# 外部系统连接配置
self.api_host = self.config.get("api_host", "")
self.api_key = self.config.get("api_key", "")
# 工作站业务状态（不同于 self.data 模式）
self._status = "Idle"
def post_init(self, ros_node: "ROS2WorkstationNode") -> None:
super().post_init(ros_node)
# 在这里启动后台服务、连接监控等
# ============ 子设备访问 ============
def _get_child_device(self, device_id: str):
"""通过 ID 获取子设备节点"""
return self._children.get(device_id)
# ============ 动作方法 ============
async def scheduler_start(self, **kwargs) -> Dict[str, Any]:
"""启动调度器"""
return {"success": True}
async def create_order(self, json_str: str, **kwargs) -> Dict[str, Any]:
"""创建工单"""
return {"success": True}
# ============ 属性 ============
@property
def workflow_sequence(self) -> str:
return "[]"
@property
def material_info(self) -> str:
return "{}"
```
### 模板 B：基于硬件控制的工作站
适用于直接与 PLC/硬件通信的场景。
```python
import logging
from typing import Dict, Any, Optional
from pylabrobot.resources import Deck
from unilabos.devices.workstation.workstation_base import WorkstationBase
try:
from unilabos.ros.nodes.presets.workstation import ROS2WorkstationNode
except ImportError:
ROS2WorkstationNode = None
class MyHardwareWorkstation(WorkstationBase):
"""硬件控制工作站"""
_ros_node: "ROS2WorkstationNode"
def __init__(
self,
config: dict = None,
deck: Optional[Deck] = None,
address: str = "192.168.1.100",
port: str = "502",
debug_mode: bool = False,
*args,
**kwargs,
):
super().__init__(deck=deck, *args, **kwargs)
self.config = config or {}
self.address = address
self.port = int(port)
self.debug_mode = debug_mode
self.logger = logging.getLogger("MyHardwareWorkstation")
# 初始化通信客户端
if not debug_mode:
from unilabos.device_comms.modbus_plc.client import ModbusTcpClient
self.client = ModbusTcpClient(host=self.address, port=self.port)
else:
self.client = None
def post_init(self, ros_node: "ROS2WorkstationNode") -> None:
super().post_init(ros_node)
# ============ 硬件读写 ============
def _read_register(self, name: str):
"""读取 Modbus 寄存器"""
if self.debug_mode:
return 0
# 实际读取逻辑
pass
# ============ 动作方法 ============
async def start_process(self, **kwargs) -> Dict[str, Any]:
"""启动加工流程"""
return {"success": True}
async def stop_process(self, **kwargs) -> Dict[str, Any]:
"""停止加工流程"""
return {"success": True}
# ============ 属性（从硬件实时读取）============
@property
def sys_status(self) -> str:
return str(self._read_register("SYS_STATUS"))
```
### 模板 C：Protocol 工作站
适用于标准化学操作协议的场景，直接使用 `ProtocolNode`。
```python
from typing import List, Optional
from pylabrobot.resources import Resource as PLRResource
from unilabos.devices.workstation.workstation_base import ProtocolNode
class MyProtocolStation(ProtocolNode):
"""Protocol 工作站 — 使用标准化学操作协议"""
def __init__(
self,
protocol_type: List[str],
deck: Optional[PLRResource] = None,
*args,
**kwargs,
):
super().__init__(protocol_type=protocol_type, deck=deck, *args, **kwargs)
```
> Protocol 工作站通常不需要自定义驱动类，直接使用 `ProtocolNode` 并在注册表和图文件中配置 `protocol_type` 即可。
**所有模板的 `__init__` 必须接受 `deck` 和 `**kwargs`。**
---
## 第四步：创建子设备驱动（如需要）
## 第四步：创建子设备（如需要）
工作站的子设备本身是独立设备。按 `docs/ai_guides/add_device.md` 的标准流程创建。
子设备是独立设备，有自己的驱动类和注册表。完整模板见 [templates.md § 子设备模板](templates.md)。
子设备的关键约束：
- 在图文件中 `parent` 指向工作站 ID
- 图文件中在工作站的 `children` 数组里列出
- 如需硬件代理，在子设备的 `config.hardware_interface.name` 指向通信设备 ID
### 关键要点
1. **驱动类**：普通 Python 类，`self.data` 预填所有属性
2. **注册表**：`category` 包含工作站标识，`auto-` 前缀动作不创建 ActionClient
3. **图文件**：`parent` 指向工作站 ID，`type: "device"`
4. **代码访问**：`self._children.get("reactor_1").driver_instance`
### 硬件代理模式
当子设备需要通过通信设备（串口/IO）通信时：
1. 通信设备 ID 必须以 `serial_` 或 `io_` 开头
2. 子设备注册表中声明 `hardware_interface: {name, read, write}`
3. 子设备实例的 `name` 属性值 = 通信设备 ID
4. ROS 节点自动将通信设备的 read/write 方法注入到子设备上
---
@@ -260,7 +95,7 @@ class MyProtocolStation(ProtocolNode):
路径：`unilabos/registry/devices/<station_name>.yaml`
### 最小配置
**最小配置（`--complete_registry` 自动补全）：**
```yaml
my_workstation:
@@ -271,92 +106,57 @@ my_workstation:
type: python
```
启动时 `--complete_registry` 自动补全 `status_types` 和 `action_value_mappings`。
### 完整配置参考
```yaml
my_workstation:
description: "我的工作站"
version: "1.0.0"
category:
- workstation
- my_category
class:
module: unilabos.devices.workstation.my_station.my_station:MyWorkstation
type: python
status_types:
workflow_sequence: String
material_info: String
action_value_mappings:
scheduler_start:
type: UniLabJsonCommandAsync
goal: {}
result:
success: success
create_order:
type: UniLabJsonCommandAsync
goal:
json_str: json_str
result:
success: success
init_param_schema:
config:
type: object
deck:
type: object
protocol_type:
type: array
```
### 子设备注册表
子设备有独立的注册表文件，需要在 `category` 中包含工作站标识：
```yaml
my_reactor:
category:
- reactor
- my_workstation
class:
module: unilabos.devices.workstation.my_station.my_reactor:MyReactor
type: python
```
**完整配置**见 [templates.md § 注册表完整配置](templates.md)。
---
## 第六步：配置 Deck 资源（如需要）
## 第六步：配置物料系统（如需要）
如果工作站有物料管理需求，需要定义 Deck 类。
物料层级：`Deck` → `WareHouse` → `ResourceHolder` (site) → `BottleCarrier` → `Bottle`
### 使用已有 Deck 类
### 快速流程
查看 `unilabos/resources/` 目录下是否有适用的 Deck 类。
1. **创建 Bottle**（`unilabos/resources/<project>/bottles.py`）— 工厂函数，返回 `Bottle` 实例
2. **创建 Carrier**（`.../bottle_carriers.py`）— 工厂函数，用 `create_ordered_items_2d` 定义槽位
3. **创建 WareHouse**（`.../warehouses.py`）— 用 `warehouse_factory()` 创建堆栈
4. **创建 Deck**（`.../decks.py`）— 继承 `pylabrobot.resources.Deck`，`setup()` 中放置 WareHouse
5. **注册表**（`unilabos/registry/resources/<project>/`）— `class.type: pylabrobot`
6. **PLR 扩展**（`unilabos/resources/plr_additional_res_reg.py`）— 导入新 Deck 类
### 创建自定义 Deck
完整代码模板见 [templates.md § 物料资源模板](templates.md)。
在 `unilabos/resources/<category>/decks.py` 中定义：
### 图文件中的 Deck 配置
```python
from pylabrobot.resources import Deck
from pylabrobot.resources.coordinate import Coordinate
工作站节点引用 Deck：
def MyStation_Deck(name: str = "MyStation_Deck") -> Deck:
deck = Deck(name=name, size_x=2700.0, size_y=1080.0, size_z=1500.0)
# 在 deck 上定义子资源位置（carrier、plate 等）
return deck
```json
"deck": {
"data": {
"_resource_child_name": "my_deck",
"_resource_type": "unilabos.resources.my_project.decks:MyStation_Deck"
}
}
```
在 `unilabos/resources/<category>/` 下注册或通过注册表引用。
Deck 子节点：
```json
{
"id": "my_deck",
"parent": "my_station",
"type": "deck",
"class": "MyStation_Deck",
"config": {"type": "MyStation_Deck", "setup": true, "rotation": {"x": 0, "y": 0, "z": 0, "type": "Rotation"}}
}
```
> **`_resource_child_name`** 必须与 Deck 节点的 `id` 一致。
---
## 第七步：配置图文件
图文件路径：`unilabos/test/experiments/<station_name>.json`
### 完整结构
路径：`unilabos/test/experiments/<station_name>.json`
```json
{
@@ -364,53 +164,19 @@ def MyStation_Deck(name: str = "MyStation_Deck") -> Deck:
{
"id": "my_station",
"name": "my_station",
"children": ["my_deck", "sub_device_1", "sub_device_2"],
"children": ["my_deck", "sub_device_1"],
"parent": null,
"type": "device",
"class": "my_workstation",
"position": {"x": 0, "y": 0, "z": 0},
"config": {
"api_host": "http://192.168.1.100:8080",
"api_key": "YOUR_KEY"
},
"deck": {
"data": {
"_resource_child_name": "my_deck",
"_resource_type": "unilabos.resources.my_module.decks:MyStation_Deck"
}
},
"size_x": 2700.0,
"size_y": 1080.0,
"size_z": 1500.0,
"config": {},
"deck": {"data": {"_resource_child_name": "my_deck", "_resource_type": "...decks:MyStation_Deck"}},
"size_x": 2700.0, "size_y": 1080.0, "size_z": 1500.0,
"protocol_type": [],
"data": {}
},
{
"id": "my_deck",
"name": "my_deck",
"children": [],
"parent": "my_station",
"type": "deck",
"class": "MyStation_Deck",
"position": {"x": 0, "y": 0, "z": 0},
"config": {
"type": "MyStation_Deck",
"setup": true,
"rotation": {"x": 0, "y": 0, "z": 0, "type": "Rotation"}
},
"data": {}
},
{
"id": "sub_device_1",
"name": "sub_device_1",
"children": [],
"parent": "my_station",
"type": "device",
"class": "sub_device_registry_name",
"position": {"x": 100, "y": 0, "z": 0},
"config": {},
"data": {}
}
{"id": "my_deck", "parent": "my_station", "type": "deck", "class": "MyStation_Deck", "config": {"type": "MyStation_Deck", "setup": true}},
{"id": "sub_device_1", "parent": "my_station", "type": "device", "class": "sub_device_class", "config": {}}
]
}
```
@@ -419,82 +185,76 @@ def MyStation_Deck(name: str = "MyStation_Deck") -> Deck:
| 字段 | 说明 |
|------|------|
| `id` | 节点唯一标识，与 `children` 数组中的引用一致 |
| `children` | 包含 deck ID 和所有子设备 ID |
| `parent` | 工作站节点为 `null`；子设备/deck 指向工作站 ID |
| `type` | 工作站和子设备为 `"device"`；deck 为 `"deck"` |
| `class` | 对应注册表中的设备名 |
| `deck.data._resource_child_name` | 必须与 deck 节点的 `id` 一致 |
| `deck.data._resource_type` | Deck 工厂函数的完整 Python 路径 |
| `protocol_type` | Protocol 工作站填入协议名列表；否则为 `[]` |
| `config` | 传入驱动 `__init__` 的 `config` 参数 |
| `parent` | 工作站为 `null`；子设备/deck 指向工作站 ID |
| `type` | 工作站和子设备 `"device"`；deck 为 `"deck"` |
| `class` | 注册表中的设备名 |
| `protocol_type` | Protocol 工作站填协议名列表；否则 `[]` |
| `config` | 传入 `__init__` 的 `config` 参数 |
### Config 字段速查
| 字段 | 外部系统 | PLC/硬件 | 说明 |
|------|---------|---------|------|
| `api_host` / `api_key` | ✅ | — | 外部 API 连接 |
| `address` / `port` | — | ✅ | PLC 地址（init 参数，非 config 内） |
| `workflow_mappings` | ✅ | — | 工作流名 → 外部 UUID |
| `material_type_mappings` | ✅ | — | PLR 资源类 → 外部物料类型 |
| `warehouse_mapping` | ✅ | — | 仓库 → 外部 UUID + 库位 UUID |
| `http_service_config` | ✅ | — | HTTP 回调 host/port |
> 完整 Config 结构详见 [reference.md § 2](reference.md)
---
## 第八步：验证
```bash
# 1. 模块可导入
python -c "from unilabos.devices.workstation.<name>.<name> import <ClassName>"
# 2. 注册表补全
unilab -g <graph>.json --complete_registry
# 3. 启动测试
unilab -g <graph>.json
```
---
## 高级模式
实现外部系统对接型工作站时，详见 [reference.md](reference.md)：RPC 客户端、HTTP 回调服务、连接监控、Config 结构模式（material_type_mappings / warehouse_mapping / workflow_mappings）、ResourceSynchronizer、update_resource、工作流序列、站间物料转移、post_init 完整模式。
---
## 关键规则
1. **`__init__` 必须接受 `deck` 和 `**kwargs`** — `WorkstationBase.__init__` 需要 `deck` 参数
2. **通过 `self._children` 访问子设备** — 不要自行维护子设备引用
3. **`post_init` 中启动后台服务** — 不要在 `__init__` 中启动网络连接
4. **异步方法使用 `await self._ros_node.sleep()`** — 禁止 `time.sleep()` 和 `asyncio.sleep()`
5. **子设备在图文件中声明** — 不在驱动代码中创建子设备实例
6. **`deck` 配置中的 `_resource_child_name` 必须与 deck 节点 ID 一致**
7. **Protocol 工作站优先使用 `ProtocolNode`** — 不需要自定义类
1. `__init__` 必须接受 `deck` 和 `**kwargs`
2. 通过 `self._children` 访问子设备，不自行维护引用
3. `post_init` 中启动后台服务，不在 `__init__` 中启动网络连接
4. 异步方法使用 `await self._ros_node.sleep()`，禁止 `time.sleep()` / `asyncio.sleep()`
5. 子设备在图文件中声明，不在驱动代码中创建
6. `_resource_child_name` 必须与 deck 节点 ID 一致
7. Protocol 工作站优先使用 `ProtocolNode`
8. 通信设备 ID 以 `serial_` 或 `io_` 开头
---
## 工作流清单
```
工作站接入进度：
- [ ] 1. 确定工作站类型（Protocol / 外部系统 / 硬件控制）
- [ ] 1. 确定类型（Protocol / 外部系统 / 硬件控制）
- [ ] 2. 确认子设备组成和物料需求
- [ ] 3. 创建工作站驱动 unilabos/devices/workstation/<name>/<name>.py
- [ ] 4. 创建子设备驱动（如需要，按 add_device.md 流程）
- [ ] 5. 创建注册表 unilabos/registry/devices/<name>.yaml
- [ ] 6. 创建/选择 Deck 资源类（如需要）
- [ ] 7. 配置图文件 unilabos/test/experiments/<name>.json
- [ ] 8. 验证：可导入 + 注册表补全 + 启动测试
- [ ] 3. 创建工作站驱动
- [ ] 4. 创建子设备驱动 + 注册表（如需要）
- [ ] 5. 创建工作站注册表
- [ ] 6. 创建物料资源 Bottle→Carrier→WareHouse→Deck（如需要）
- [ ] 7. 注册 PLR 扩展（Deck 类需要）
- [ ] 8. 配置图文件
- [ ] 9. 验证
```
---
## 现有工作站参考
## 参考资源
| 工作站 | 注册表名 | 驱动类 | 类型 |
|--------|----------|--------|------|
| Protocol 通用 | `workstation` | `ProtocolNode` | Protocol |
| Bioyond 反应站 | `reaction_station.bioyond` | `BioyondReactionStation` | 外部系统 |
| Bioyond 配液站 | `bioyond_dispensing_station` | `BioyondDispensingStation` | 外部系统 |
| 纽扣电池组装 | `coincellassemblyworkstation_device` | `CoinCellAssemblyWorkstation` | 硬件控制 |
- **代码模板**：[templates.md](templates.md) — 驱动模板 A/B/C、子设备、注册表、物料资源
- **高级模式**：[reference.md](reference.md) — 外部系统集成、Config 模式、资源同步、PLC 框架、端到端案例
- **现有工作站**：
### 参考文件路径
- 基类: `unilabos/devices/workstation/workstation_base.py`
- Bioyond 基类: `unilabos/devices/workstation/bioyond_studio/station.py`
- 反应站: `unilabos/devices/workstation/bioyond_studio/reaction_station/reaction_station.py`
- 配液站: `unilabos/devices/workstation/bioyond_studio/dispensing_station/dispensing_station.py`
- 纽扣电池: `unilabos/devices/workstation/coin_cell_assembly/coin_cell_assembly.py`
- ROS 节点: `unilabos/ros/nodes/presets/workstation.py`
- 图文件: `unilabos/test/experiments/reaction_station_bioyond.json`, `dispensing_station_bioyond.json`
| 工作站 | 注册表名 | 类型 | 驱动路径 |
|--------|----------|------|---------|
| Bioyond 反应站 | `reaction_station.bioyond` | 外部系统 | `bioyond_studio/reaction_station/` |
| Bioyond 配液站 | `bioyond_dispensing_station` | 外部系统 | `bioyond_studio/dispensing_station/` |
| 纽扣电池组装 | `coincellassemblyworkstation_device` | 硬件控制 | `coin_cell_assembly/` |
| Protocol 通用 | `workstation` | Protocol | `workstation_base.py` |

485
.cursor/skills/add-workstation/reference.md
View File

@@ -1,6 +1,6 @@
# 工作站高级模式参考
本文件是 SKILL.md 的补充，包含外部系统集成、物料同步、配置结构等高级模式。
本文件是 SKILL.md 的补充，包含外部系统集成、物料同步、PLC 框架、硬件代理等高级模式。
Agent 在需要实现这些功能时按需阅读。
---
@@ -116,7 +116,6 @@ class ConnectionMonitor:
def _monitor_loop(self):
while self._running:
try:
# 调用外部系统接口检测连接
self.workstation.hardware_interface.ping()
status = "online"
except Exception:
@@ -210,6 +209,35 @@ class ConnectionMonitor:
}
```
### 2.7 工作流到工序名映射
```json
{
"workflow_to_section_map": {
"reactor_taken_in": "反应器放入",
"reactor_taken_out": "反应器取出",
"Solid_feeding_vials": "固体投料-小瓶"
}
}
```
### 2.8 动作名称映射
```json
{
"action_names": {
"reactor_taken_in": {
"config": "通量-配置",
"stirring": "反应模块-开始搅拌"
},
"solid_feeding_vials": {
"feeding": "粉末加样模块-投料",
"observe": "反应模块-观察搅拌结果"
}
}
}
```
---
## 3. 资源同步机制
@@ -246,14 +274,25 @@ class MyResourceSynchronizer(ResourceSynchronizer):
return True
```
### 3.2 update_resource — 上传资源树到云端
### 3.2 资源树回调
Bioyond 工作站注册了资源树变更回调，实现与外部系统的自动同步：
| 回调名 | 触发时机 | 外部操作 |
|--------|---------|---------|
| `resource_tree_add` | PLR Deck 中添加资源 | 入库到外部系统 |
| `resource_tree_remove` | PLR Deck 中移除资源 | 出库 |
| `resource_tree_transfer` | 创建物料（不入库） | 创建外部物料记录 |
| `resource_tree_update` | 资源位置移动 | 更新外部系统库位 |
### 3.3 update_resource — 上传资源树到云端
将 PLR Deck 序列化后通过 ROS 服务上传。典型使用场景：
```python
# 在 post_init 中上传初始 deck
from unilabos.ros.nodes.base_device_node import ROS2DeviceNode
# 在 post_init 中上传初始 deck
ROS2DeviceNode.run_async_func(
self._ros_node.update_resource, True,
**{"resources": [self.deck]}
@@ -315,15 +354,11 @@ async def transfer_materials_to_another_station(
"""将物料转移到另一个工作站"""
target_node = self._children.get(target_device_id)
if not target_node:
# 通过 ROS 节点查找非子设备的目标站
pass
for group in transfer_groups:
resource = self.find_resource_by_name(group["resource_name"])
# 从本站 deck 移除
resource.unassign()
# 调用目标站的接收方法
# ...
return {"success": True, "transferred": len(transfer_groups)}
```
@@ -369,3 +404,437 @@ def post_init(self, ros_node):
# 5. 初始化资源同步器（可选）
self.resource_synchronizer = MyResourceSynchronizer(self, self.rpc_client)
```
---
## 7. PLC/Modbus 完整框架
### 7.1 寄存器映射 CSV 格式
PLC 工作站使用 CSV 文件定义寄存器映射表。路径通常为工作站目录下的 `<name>.csv`。
**CSV 列定义：**
| 列名 | 说明 | 值示例 |
|------|------|--------|
| `Name` | 寄存器节点名称（代码中引用的唯一标识） | `COIL_SYS_START_CMD` |
| `DataType` | 数据类型 | `BOOL`, `INT16`, `FLOAT32` |
| `InitValue` | 初始值（可选） | — |
| `Comment` | 注释（可选） | — |
| `Attribute` | 自定义属性（可选） | — |
| `DeviceType` | Modbus 设备类型 | `coil`, `hold_register`, `input_register`, `discrete_inputs` |
| `Address` | Modbus 地址 | `8010`, `11000` |
**CSV 示例：**
```csv
Name,DataType,InitValue,Comment,Attribute,DeviceType,Address,
COIL_SYS_START_CMD,BOOL,,系统启动命令,,coil,8010,
COIL_SYS_STOP_CMD,BOOL,,系统停止命令,,coil,8020,
COIL_SYS_RESET_CMD,BOOL,,系统复位命令,,coil,8030,
REG_MSG_ELECTROLYTE_VOLUME,INT16,,电解液体积,,hold_register,11004,
REG_DATA_OPEN_CIRCUIT_VOLTAGE,FLOAT32,,开路电压,,hold_register,10002,
REG_DATA_AXIS_X_POS,FLOAT32,,X轴位置,,hold_register,10004,
```
**命名约定：**
- 线圈：`COIL_` 前缀（读写布尔量）
- 保持寄存器：`REG_MSG_`（消息/命令寄存器）、`REG_DATA_`（数据/状态寄存器）
- `_CMD` 后缀：写入命令
- `_STATUS` 后缀：读取状态
### 7.2 TCPClient 初始化
```python
from unilabos.device_comms.modbus_plc.client import TCPClient, BaseClient
from unilabos.device_comms.modbus_plc.modbus import DataType, WorderOrder
# 创建 Modbus TCP 客户端
modbus_client = TCPClient(addr="192.168.1.100", port=502)
modbus_client.client.connect()
# 从 CSV 加载寄存器映射
import os
csv_path = os.path.join(os.path.dirname(__file__), 'register_map.csv')
nodes = BaseClient.load_csv(csv_path)
client = modbus_client.register_node_list(nodes)
```
### 7.3 寄存器读写操作
```python
# 读取线圈（布尔值）
result, err = client.use_node('COIL_SYS_START_STATUS').read(1)
is_started = result[0] if not err else False
# 写入线圈
client.use_node('COIL_SYS_START_CMD').write(True)
# 读取保持寄存器（INT16）
result, err = client.use_node('REG_DATA_ASSEMBLY_COIN_CELL_NUM').read(1)
# 读取保持寄存器（FLOAT32，需要 2 个寄存器）
result, err = client.use_node('REG_DATA_OPEN_CIRCUIT_VOLTAGE').read(2)
# 写入保持寄存器（FLOAT32）
client.use_node('REG_MSG_ELECTROLYTE_VOLUME').write(
100.0,
data_type=DataType.FLOAT32,
word_order=WorderOrder.LITTLE,
)
```
**FLOAT32 字节序注意：** 许多 PLC 使用 Big Byte Order + Little Word Order，需要交换两个 16 位寄存器的顺序。参考 `coin_cell_assembly.py` 中的 `_decode_float32_correct` 函数。
### 7.4 ModbusWorkflow 生命周期
PLC 工作站的动作通过 `ModbusWorkflow` + `WorkflowAction` 组织，每个动作有 4 个生命周期阶段：
```python
from unilabos.device_comms.modbus_plc.client import ModbusWorkflow, WorkflowAction
# 定义动作的生命周期函数
def my_init(use_node):
"""初始化：设置参数"""
use_node('REG_MSG_ELECTROLYTE_VOLUME').write(
100.0, data_type=DataType.FLOAT32, word_order=WorderOrder.LITTLE
)
return True
def my_start(use_node):
"""启动：触发动作并轮询等待完成"""
use_node('COIL_SYS_START_CMD').write(True)
while True:
result, err = use_node('COIL_SYS_START_STATUS').read(1)
if not err and result[0]:
break
time.sleep(0.5)
return True
def my_stop(use_node):
"""停止：复位触发信号"""
use_node('COIL_SYS_START_CMD').write(False)
return True
def my_cleanup(use_node):
"""清理：无论成功失败都执行"""
use_node('COIL_SYS_RESET_CMD').write(True)
# 组合成工作流
workflow = ModbusWorkflow(
name="我的加工流程",
actions=[
WorkflowAction(init=my_init, start=my_start, stop=my_stop, cleanup=my_cleanup)
],
)
# 执行
client.run_modbus_workflow(workflow)
```
**生命周期执行顺序：** `init` → `start` → `stop` → `cleanup`（cleanup 始终执行，即使前序步骤失败）
### 7.5 PLC 工作站中的握手循环
纽扣电池组装站的典型 PLC 交互模式（信息交换握手）：
```python
async def _send_msg_to_plc(self, data: dict):
"""向 PLC 发送消息并等待确认"""
# 1. 写入数据寄存器
for key, value in data.items():
self._write_register(key, value)
# 2. 发送"消息已准备好"信号
self._write_coil('COIL_UNILAB_SEND_MSG_SUCC_CMD', True)
# 3. 等待 PLC 读取确认
while not self._read_coil('COIL_REQUEST_REC_MSG_STATUS'):
await self._ros_node.sleep(0.3)
# 4. 撤销发送信号
self._write_coil('COIL_UNILAB_SEND_MSG_SUCC_CMD', False)
async def _recv_msg_from_plc(self) -> dict:
"""等待 PLC 发送消息"""
# 1. 等待 PLC 发送信号
while not self._read_coil('COIL_REQUEST_SEND_MSG_STATUS'):
await self._ros_node.sleep(0.3)
# 2. 读取数据寄存器
data = {}
for key in self._recv_registers:
data[key] = self._read_register(key)
# 3. 发送"已收到"确认
self._write_coil('COIL_UNILAB_REC_MSG_SUCC_CMD', True)
# 4. 等待 PLC 撤销发送信号
while self._read_coil('COIL_REQUEST_SEND_MSG_STATUS'):
await self._ros_node.sleep(0.3)
# 5. 撤销确认信号
self._write_coil('COIL_UNILAB_REC_MSG_SUCC_CMD', False)
return data
```
### 7.6 JSON 驱动的 PLC 工作流
PLC 工作站还支持通过 JSON 描述工作流，无需编写 Python 代码。使用 `BaseClient.execute_procedure_from_json`：
```json
{
"register_node_list_from_csv_path": {"path": "register_map.csv"},
"create_flow": [
{
"name": "初始化系统",
"action": [
{
"address_function_to_create": [
{"func_name": "write_start", "node_name": "COIL_SYS_START_CMD", "mode": "write", "value": true},
{"func_name": "read_status", "node_name": "COIL_SYS_START_STATUS", "mode": "read", "value": 1}
],
"create_init_function": null,
"create_start_function": {
"func_name": "start_sys",
"write_functions": ["write_start"],
"condition_functions": ["read_status"],
"stop_condition_expression": "read_status[0]"
},
"create_stop_function": {"func_name": "stop_start", "node_name": "COIL_SYS_START_CMD", "mode": "write", "value": false},
"create_cleanup_function": null
}
]
}
],
"execute_flow": ["初始化系统"]
}
```
参考：`unilabos/device_comms/modbus_plc/client.py`（`ExecuteProcedureJson` 类型定义）
---
## 8. 端到端案例 Walkthrough：Bioyond 反应站
以 Bioyond 反应站为例，展示从零接入一个带物料输入的外部系统工作站的完整过程。
### 8.1 需求
- **类型**：外部系统工作站（与 Bioyond LIMS 系统对接）
- **通信**：HTTP API（RPC 客户端 + HTTP 回调服务）
- **子设备**：5 个反应器（reactor_1 ~ reactor_5）
- **物料**：反应器、试剂瓶、烧杯、样品板、小瓶、枪头盒 → 6 种 WareHouse → 1 个 Deck
### 8.2 文件结构
```
unilabos/
├── devices/workstation/bioyond_studio/
│ ├── station.py # BioyondWorkstation 基类
│ ├── bioyond_rpc.py # RPC 客户端
│ └── reaction_station/
│ └── reaction_station.py # BioyondReactionStation + BioyondReactor
├── resources/bioyond/
│ ├── bottles.py # Bottle 工厂函数（8 种）
│ ├── bottle_carriers.py # Carrier 工厂函数（8 种）
│ ├── warehouses.py # WareHouse 工厂函数（6 种）
│ └── decks.py # BIOYOND_PolymerReactionStation_Deck
├── registry/
│ ├── devices/reaction_station_bioyond.yaml
│ └── resources/bioyond/
│ ├── bottles.yaml
│ ├── bottle_carriers.yaml
│ └── decks.yaml
└── test/experiments/reaction_station_bioyond.json
```
### 8.3 继承链
```
WorkstationBase
└── BioyondWorkstation # 通用 Bioyond 逻辑
├── __init__(config, deck, protocol_type)
├── post_init() → 启动连接监控 + HTTP 服务 + 上传 deck
├── BioyondResourceSynchronizer # 物料双向同步
└── BioyondReactionStation # 反应站特化
├── reactor_taken_in() # 反应器放入工作流
├── solid_feeding_vials() # 固体投料
├── liquid_feeding_solvents() # 液体投料
└── workflow_sequence @property # 工作流序列状态
```
### 8.4 物料资源层级（反应站实例）
```
BIOYOND_PolymerReactionStation_Deck (2700×1080×1500mm)
├── 堆栈1左 (WareHouse 4x4) ← Coordinate(-200, 400, 0)
│ ├── A01 → BottleCarrier → Reactor
│ ├── A02 → BottleCarrier → Reactor
│ └── ...（共 16 槽位）
├── 堆栈1右 (WareHouse 4x4, col_offset=4) ← Coordinate(350, 400, 0)
│ ├── A05 → BottleCarrier → Reactor
│ └── ...
├── 站内试剂存放堆栈 (WareHouse 1x2) ← Coordinate(1050, 400, 0)
│ ├── A01 → 1BottleCarrier → Bottle
│ └── A02 → 1BottleCarrier → Bottle
├── 测量小瓶仓库 (WareHouse 3x2) ← Coordinate(...)
├── 站内Tip盒堆栈(左) (WareHouse, removed_positions)
└── 站内Tip盒堆栈(右) (WareHouse)
```
### 8.5 图文件关键结构
```json
{
"nodes": [
{
"id": "reaction_station_bioyond",
"children": ["Bioyond_Deck", "reactor_1", "reactor_2", "reactor_3", "reactor_4", "reactor_5"],
"parent": null,
"type": "device",
"class": "reaction_station.bioyond",
"config": {
"api_key": "DE9BDDA0",
"api_host": "http://172.21.103.36:45388",
"workflow_mappings": {
"reactor_taken_out": "3a16081e-...",
"reactor_taken_in": "3a160df6-..."
},
"material_type_mappings": {
"BIOYOND_PolymerStation_Reactor": ["反应器", "3a14233b-..."],
"BIOYOND_PolymerStation_1BottleCarrier": ["试剂瓶", "3a14233b-..."]
},
"warehouse_mapping": {
"堆栈1左": {
"uuid": "3a14aa17-...",
"site_uuids": {"A01": "3a14aa17-...", "A02": "3a14aa17-..."}
}
},
"http_service_config": {
"http_service_host": "127.0.0.1",
"http_service_port": 8080
}
},
"deck": {
"data": {
"_resource_child_name": "Bioyond_Deck",
"_resource_type": "unilabos.resources.bioyond.decks:BIOYOND_PolymerReactionStation_Deck"
}
},
"size_x": 2700.0,
"size_y": 1080.0,
"size_z": 2500.0,
"protocol_type": [],
"data": {}
},
{
"id": "Bioyond_Deck",
"parent": "reaction_station_bioyond",
"type": "deck",
"class": "BIOYOND_PolymerReactionStation_Deck",
"config": {"type": "BIOYOND_PolymerReactionStation_Deck", "setup": true}
},
{
"id": "reactor_1",
"parent": "reaction_station_bioyond",
"type": "device",
"class": "reaction_station.reactor",
"position": {"x": 1150, "y": 300, "z": 0},
"config": {}
}
]
}
```
### 8.6 初始化时序
```
1. ROS2WorkstationNode.__init__
├── 创建 BioyondReactionStation 实例（__init__）
├── 加载 Deck（BIOYOND_PolymerReactionStation_Deck, setup=true → 创建 6 个 WareHouse）
├── 初始化 reactor_1~5（BioyondReactor 实例）→ sub_devices
└── 为每个 reactor 创建 ActionClient
2. BioyondReactionStation.post_init(ros_node)
├── 初始化 BioyondV1RPC（HTTP 客户端）
├── 初始化 BioyondResourceSynchronizer
├── 启动 ConnectionMonitor（30s 轮询）
├── 启动 WorkstationHTTPService（接收回调）
├── sync_from_external()（从 Bioyond 拉取物料到 Deck）
└── update_resource([self.deck])（上传 Deck 到云端）
```
### 8.7 物料同步流程
```
外部入库：
Bioyond API → stock_material() → 获取物料列表
→ resource_bioyond_to_plr() → 转为 PLR Bottle/Carrier
→ deck.warehouses["堆栈1左"]["A01"] = carrier
→ update_resource([deck])
外部变更回调：
Bioyond POST /report/material_change
→ WorkstationHTTPService 接收
→ process_material_change_report()
→ 更新 Deck 中的资源
→ update_resource([affected_resource])
```
### 8.8 工作站动作执行流程（以 reactor_taken_in 为例）
```python
async def reactor_taken_in(self, assign_material_name, cutoff, temperature, **kwargs):
# 1. 从 config 获取工作流 UUID
workflow_id = self.config["workflow_mappings"]["reactor_taken_in"]
# 2. 构建工序参数
sections = self._build_sections(temperature, cutoff, ...)
# 3. 合并到工作流序列
self._workflow_sequence.append({"name": "reactor_taken_in", ...})
# 4. 调用外部系统创建工单
result = self.hardware_interface.create_order(order_data)
# 5. 等待外部系统完成（通过 HTTP 回调通知）
# process_order_finish_report 被回调时更新状态
return {"success": True}
```
---
## 9. 现有工作站 Config 结构完整对比
| 特性 | BioyondReactionStation | BioyondDispensingStation | CoinCellAssemblyWorkstation |
|------|----------------------|------------------------|-----------------------------|
| **继承** | BioyondWorkstation | BioyondWorkstation | WorkstationBase (直接) |
| **通信方式** | HTTP RPC | HTTP RPC | Modbus TCP |
| **`__init__` 签名** | `(config, deck, protocol_type, **kwargs)` | `(config, deck, protocol_type, **kwargs)` | `(config, deck, address, port, debug_mode, **kwargs)` |
| **子设备** | 5 个 BioyondReactor | 无 | 无 |
| **Deck** | BioyondReactionDeck (6 个 WareHouse) | BioyondDispensingDeck | CoincellDeck |
| **物料同步** | BioyondResourceSynchronizer (双向) | BioyondResourceSynchronizer (双向) | 无（本地 PLR） |
| **status_types** | `workflow_sequence: str` | 空 | 18 个属性 (sys_status, 传感器数据等) |
| **动作风格** | 语义化 (reactor_taken_in, ...) | 语义化 (compute_experiment_design, ...) | PLC 操作 (func_pack_device_init, ...) |
| **post_init** | 连接监控 + HTTP 服务 + 资源同步 + 上传 deck | 继承父类 | 上传 deck |
| **工作流管理** | workflow_mappings → 合并序列 → create_order | batch_create → wait_for_reports | PLC 握手循环 |
### Config 字段对比
| 字段 | 反应站 | 配液站 | 纽扣电池 |
|------|--------|--------|---------|
| `api_host` | ✅ | ✅ | — |
| `api_key` | ✅ | ✅ | — |
| `workflow_mappings` | ✅ (8 个工作流) | — | — |
| `material_type_mappings` | ✅ (8 种物料) | ✅ | — |
| `warehouse_mapping` | ✅ (6 个仓库) | ✅ (3 个仓库) | — |
| `workflow_to_section_map` | ✅ | — | — |
| `action_names` | ✅ | — | — |
| `http_service_config` | ✅ | — | — |
| `material_default_parameters` | ✅ | — | — |
| `address` (init 参数) | — | — | ✅ |
| `port` (init 参数) | — | — | ✅ |
| `debug_mode` (init 参数) | — | — | ✅ |

454
.cursor/skills/add-workstation/templates.md Normal file
View File

@@ -0,0 +1,454 @@
# 工作站代码模板
本文件包含 SKILL.md 引用的所有代码模板。Agent 根据需要按需阅读。
---
## Template A：外部系统工作站
```python
import logging
from typing import Dict, Any, Optional, List
from pylabrobot.resources import Deck
from unilabos.devices.workstation.workstation_base import WorkstationBase
try:
from unilabos.ros.nodes.presets.workstation import ROS2WorkstationNode
except ImportError:
ROS2WorkstationNode = None
class MyWorkstation(WorkstationBase):
_ros_node: "ROS2WorkstationNode"
def __init__(
self,
config: dict = None,
deck: Optional[Deck] = None,
protocol_type: list = None,
**kwargs,
):
super().__init__(deck=deck, **kwargs)
self.config = config or {}
self.logger = logging.getLogger(f"MyWorkstation")
self.api_host = self.config.get("api_host", "")
self.api_key = self.config.get("api_key", "")
self._status = "Idle"
def post_init(self, ros_node: "ROS2WorkstationNode") -> None:
super().post_init(ros_node)
def _get_child_device(self, device_id: str):
return self._children.get(device_id)
async def scheduler_start(self, **kwargs) -> Dict[str, Any]:
return {"success": True}
async def create_order(self, json_str: str, **kwargs) -> Dict[str, Any]:
return {"success": True}
@property
def workflow_sequence(self) -> str:
return "[]"
@property
def material_info(self) -> str:
return "{}"
```
---
## Template B：PLC/Modbus 硬件控制工作站
```python
import os
import logging
from typing import Dict, Any, Optional
from pylabrobot.resources import Deck
from unilabos.devices.workstation.workstation_base import WorkstationBase
from unilabos.device_comms.modbus_plc.client import (
TCPClient, BaseClient, ModbusWorkflow, WorkflowAction,
)
from unilabos.device_comms.modbus_plc.modbus import (
Base as ModbusNodeBase, DataType, WorderOrder,
)
try:
from unilabos.ros.nodes.presets.workstation import ROS2WorkstationNode
except ImportError:
ROS2WorkstationNode = None
class MyHardwareWorkstation(WorkstationBase):
_ros_node: "ROS2WorkstationNode"
def __init__(
self,
config: dict = None,
deck: Optional[Deck] = None,
address: str = "192.168.1.100",
port: str = "502",
debug_mode: bool = False,
*args,
**kwargs,
):
super().__init__(deck=deck, *args, **kwargs)
self.config = config or {}
self.address = address
self.port = int(port)
self.debug_mode = debug_mode
self.logger = logging.getLogger("MyHardwareWorkstation")
if not debug_mode:
modbus_client = TCPClient(addr=self.address, port=self.port)
modbus_client.client.connect()
csv_path = os.path.join(os.path.dirname(__file__), 'register_map.csv')
self.nodes = BaseClient.load_csv(csv_path)
self.client = modbus_client.register_node_list(self.nodes)
else:
self.client = None
def _read_coil(self, name: str) -> bool:
if self.debug_mode:
return False
result, err = self.client.use_node(name).read(1)
return result[0] if not err else False
def _write_coil(self, name: str, value: bool):
if not self.debug_mode:
self.client.use_node(name).write(value)
def _read_register(self, name: str, data_type: DataType = DataType.INT16):
if self.debug_mode:
return 0
result, err = self.client.use_node(name).read(
2 if data_type == DataType.FLOAT32 else 1
)
return result if not err else 0
def _write_register(self, name: str, value, data_type: DataType = DataType.FLOAT32):
if not self.debug_mode:
self.client.use_node(name).write(
value, data_type=data_type, word_order=WorderOrder.LITTLE
)
async def start_process(self, **kwargs) -> Dict[str, Any]:
self._write_coil('COIL_SYS_START_CMD', True)
return {"success": True}
async def stop_process(self, **kwargs) -> Dict[str, Any]:
self._write_coil('COIL_SYS_STOP_CMD', True)
return {"success": True}
@property
def sys_status(self) -> str:
return str(self._read_coil("COIL_SYS_START_STATUS"))
```
### PLC 寄存器映射 CSV 格式
```csv
Name,DataType,InitValue,Comment,Attribute,DeviceType,Address,
COIL_SYS_START_CMD,BOOL,,系统启动命令,,coil,8010,
COIL_SYS_STOP_CMD,BOOL,,系统停止命令,,coil,8020,
REG_MSG_ELECTROLYTE_VOLUME,INT16,,电解液体积,,hold_register,11004,
REG_DATA_OPEN_CIRCUIT_VOLTAGE,FLOAT32,,开路电压,,hold_register,10002,
```
命名约定：`COIL_` 线圈 / `REG_MSG_` 命令寄存器 / `REG_DATA_` 数据寄存器 / `_CMD` 写入 / `_STATUS` 读取
---
## Template C：Protocol 工作站
```python
from typing import List, Optional
from pylabrobot.resources import Resource as PLRResource
from unilabos.devices.workstation.workstation_base import ProtocolNode
class MyProtocolStation(ProtocolNode):
def __init__(
self,
protocol_type: List[str],
deck: Optional[PLRResource] = None,
*args,
**kwargs,
):
super().__init__(protocol_type=protocol_type, deck=deck, *args, **kwargs)
```
> 通常不需要自定义类，直接在注册表和图文件中配置 `ProtocolNode` + `protocol_type` 即可。
---
## 子设备模板
### 驱动类
```python
class MyReactor:
def __init__(self, **kwargs):
self.data = {
"temperature": 0.0,
"status": "Idle",
}
async def update_metrics(self, metrics_json: str, **kwargs):
import json
metrics = json.loads(metrics_json)
self.data["temperature"] = metrics.get("temperature", 0.0)
self.data["status"] = metrics.get("status", "Idle")
return {"success": True}
```
### 注册表
```yaml
reaction_station.reactor:
category:
- reactor
- my_workstation
class:
module: unilabos.devices.workstation.my_station.my_station:MyReactor
type: python
status_types:
temperature: Float64
status: String
action_value_mappings:
auto-update_metrics:
type: UniLabJsonCommandAsync
goal:
metrics_json: json_str
result:
success: success
```
> `auto-` 前缀动作不创建 ActionClient，仅供工作站驱动内部调用。
### 图文件节点
```json
{
"id": "reactor_1",
"name": "reactor_1",
"children": [],
"parent": "my_station",
"type": "device",
"class": "reaction_station.reactor",
"position": {"x": 1150, "y": 300, "z": 0},
"config": {},
"data": {}
}
```
### 代码中访问子设备
```python
child_node = self._children.get("reactor_1")
child_node.driver_instance.update_metrics(data)
child = self._ros_node.sub_devices.get("reactor_1")
child.driver_instance.data["temperature"]
```
### 硬件代理配置
通信设备节点（ID 以 `serial_` 或 `io_` 开头）：
```json
{
"id": "serial_port_1",
"parent": "my_station",
"type": "device",
"class": "serial_device",
"config": {
"hardware_interface": {
"name": "hardware_interface_name",
"read": "read_method",
"write": "write_method"
}
}
}
```
子设备注册表中声明代理：
```yaml
my_sub_device:
class:
hardware_interface:
name: hardware_interface_name # 属性名，值为通信设备 ID
read: read_from_device # 注入的读方法
write: write_to_device # 注入的写方法
```
ROS 节点初始化时自动检测 `name` 属性值是否为另一子设备 ID，若是则注入通信设备的 read/write 方法。
---
## 注册表完整配置
```yaml
my_workstation:
description: "我的工作站"
version: "1.0.0"
category:
- workstation
- my_category
class:
module: unilabos.devices.workstation.my_station.my_station:MyWorkstation
type: python
status_types:
workflow_sequence: String
material_info: String
action_value_mappings:
scheduler_start:
type: UniLabJsonCommandAsync
goal: {}
result:
success: success
create_order:
type: UniLabJsonCommandAsync
goal:
json_str: json_str
result:
success: success
init_param_schema:
config:
type: object
deck:
type: object
protocol_type:
type: array
```
---
## 物料资源模板
### Bottle 工厂函数
```python
from unilabos.resources.itemized_carrier import Bottle
def My_Reagent_Bottle(
name: str,
diameter: float = 70.0,
height: float = 120.0,
max_volume: float = 500000.0, # 单位 μL（500mL = 500000）
barcode: str = None,
) -> Bottle:
return Bottle(
name=name, diameter=diameter, height=height,
max_volume=max_volume, barcode=barcode,
model="My_Reagent_Bottle",
)
```
### BottleCarrier 工厂函数
```python
from pylabrobot.resources import ResourceHolder
from pylabrobot.resources.carrier import create_ordered_items_2d
from unilabos.resources.itemized_carrier import BottleCarrier
def My_6SlotCarrier(name: str) -> BottleCarrier:
sites = create_ordered_items_2d(
klass=ResourceHolder,
num_items_x=3, num_items_y=2,
dx=10.0, dy=10.0, dz=5.0,
item_dx=42.0, item_dy=35.0,
size_x=20.0, size_y=20.0, size_z=50.0,
)
carrier = BottleCarrier(
name=name, size_x=146.0, size_y=80.0, size_z=55.0,
sites=sites, model="My_6SlotCarrier",
)
carrier.num_items_x = 3
carrier.num_items_y = 2
carrier.num_items_z = 1
return carrier
```
### WareHouse 工厂函数
```python
from unilabos.resources.warehouse import warehouse_factory
def my_warehouse_4x4(name: str) -> "WareHouse":
return warehouse_factory(
name=name,
num_items_x=4, num_items_y=4, num_items_z=1,
dx=137.0, dy=96.0, dz=120.0,
item_dx=137.0, item_dy=125.0, item_dz=10.0,
resource_size_x=127.0, resource_size_y=85.0, resource_size_z=100.0,
model="my_warehouse_4x4",
)
```
### Deck 类
```python
from pylabrobot.resources import Deck, Coordinate
class MyStation_Deck(Deck):
def __init__(
self,
name: str = "MyStation_Deck",
size_x: float = 2700.0,
size_y: float = 1080.0,
size_z: float = 1500.0,
category: str = "deck",
setup: bool = False,
) -> None:
super().__init__(name=name, size_x=size_x, size_y=size_y, size_z=size_z)
if setup:
self.setup()
def setup(self) -> None:
self.warehouses = {
"堆栈A": my_warehouse_4x4("堆栈A"),
"堆栈B": my_warehouse_4x4("堆栈B"),
}
self.warehouse_locations = {
"堆栈A": Coordinate(-200.0, 400.0, 0.0),
"堆栈B": Coordinate(2350.0, 400.0, 0.0),
}
for wh_name, wh in self.warehouses.items():
self.assign_child_resource(wh, location=self.warehouse_locations[wh_name])
```
### 资源注册表
```yaml
# unilabos/registry/resources/<project>/bottles.yaml
My_Reagent_Bottle:
category: [bottles]
class:
module: unilabos.resources.my_project.bottles:My_Reagent_Bottle
type: pylabrobot
# unilabos/registry/resources/<project>/decks.yaml
MyStation_Deck:
category: [deck]
class:
module: unilabos.resources.my_project.decks:MyStation_Deck
type: pylabrobot
registry_type: resource
```
### PLR 扩展注册
新 Deck 类需要在 `unilabos/resources/plr_additional_res_reg.py` 中导入：
```python
def register():
from unilabos.resources.my_project.decks import MyStation_Deck
```