Compare commits

..

9 Commits

Author SHA1 Message Date
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
Xuwznln
56eb7e2ab4 speed up registry load 2026-02-02 20:01:04 +08:00
Xuwznln
23ce145f74 workflow upload & set liquid fix & add set liquid with plate 2026-02-02 18:23:33 +08:00
Xuwznln
b0da149252 fix upload workflow json 2026-02-02 17:19:07 +08:00
Xuwznln
07c9e6f0fe save class name when deserialize & protocol execute test 2026-02-02 16:05:17 +08:00
Xuwznln
ccec6b9d77 Support root node change pos 2026-02-02 12:03:19 +08:00
hanhua@dp.tech
dadfdf3d8d add unilabos_class 2026-01-30 18:07:53 +08:00
18 changed files with 3208 additions and 1008 deletions

View File

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

View File

@@ -439,6 +439,9 @@ unilab --ak your_ak --sk your_sk -g test/experiments/mock_devices/mock_all.json
1. 访问 Web 界面,进入"仪器耗材"模块
2. 在"仪器设备"区域找到并添加上述设备
3. 在"物料耗材"区域找到并添加容器
4. 在workstation中配置protocol_type包含PumpTransferProtocol
![添加Protocol类型](image/add_protocol.png)
![物料列表](image/material.png)

Binary file not shown.

After

Width:  |  Height:  |  Size: 81 KiB

213
tests/workflow/test.json Normal file
View File

@@ -0,0 +1,213 @@
{
"workflow": [
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines",
"targets": "Liquid_1",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines",
"targets": "Liquid_2",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines",
"targets": "Liquid_3",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines_2",
"targets": "Liquid_4",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines_2",
"targets": "Liquid_5",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines_2",
"targets": "Liquid_6",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines_3",
"targets": "dest_set",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines_3",
"targets": "dest_set_2",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
},
{
"action": "transfer_liquid",
"action_args": {
"sources": "cell_lines_3",
"targets": "dest_set_3",
"asp_vol": 100.0,
"dis_vol": 74.75,
"asp_flow_rate": 94.0,
"dis_flow_rate": 95.5
}
}
],
"reagent": {
"Liquid_1": {
"slot": 1,
"well": [
"A4",
"A7",
"A10"
],
"labware": "rep 1"
},
"Liquid_4": {
"slot": 1,
"well": [
"A4",
"A7",
"A10"
],
"labware": "rep 1"
},
"dest_set": {
"slot": 1,
"well": [
"A4",
"A7",
"A10"
],
"labware": "rep 1"
},
"Liquid_2": {
"slot": 2,
"well": [
"A3",
"A5",
"A8"
],
"labware": "rep 2"
},
"Liquid_5": {
"slot": 2,
"well": [
"A3",
"A5",
"A8"
],
"labware": "rep 2"
},
"dest_set_2": {
"slot": 2,
"well": [
"A3",
"A5",
"A8"
],
"labware": "rep 2"
},
"Liquid_3": {
"slot": 3,
"well": [
"A4",
"A6",
"A10"
],
"labware": "rep 3"
},
"Liquid_6": {
"slot": 3,
"well": [
"A4",
"A6",
"A10"
],
"labware": "rep 3"
},
"dest_set_3": {
"slot": 3,
"well": [
"A4",
"A6",
"A10"
],
"labware": "rep 3"
},
"cell_lines": {
"slot": 4,
"well": [
"A1",
"A3",
"A5"
],
"labware": "DRUG + YOYO-MEDIA"
},
"cell_lines_2": {
"slot": 4,
"well": [
"A1",
"A3",
"A5"
],
"labware": "DRUG + YOYO-MEDIA"
},
"cell_lines_3": {
"slot": 4,
"well": [
"A1",
"A3",
"A5"
],
"labware": "DRUG + YOYO-MEDIA"
}
}
}

View File

@@ -359,9 +359,7 @@ class HTTPClient:
Returns:
Dict: API响应数据包含 code 和 data (uuid, name)
"""
# target_lab_uuid 暂时使用默认值,后续由后端根据 ak/sk 获取
payload = {
"target_lab_uuid": "28c38bb0-63f6-4352-b0d8-b5b8eb1766d5",
"name": name,
"data": {
"workflow_uuid": workflow_uuid,

View File

@@ -1,15 +1,11 @@
from __future__ import annotations
import asyncio
import time
import traceback
from collections import Counter
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 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.standard import GripDirection
from pylabrobot.resources import (
@@ -27,22 +23,48 @@ from pylabrobot.resources import (
Trash,
Tip,
)
from typing_extensions import TypedDict
from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
from unilabos.registry.placeholder_type import ResourceSlot
from unilabos.resources.resource_tracker import ResourceTreeSet, ResourceDict
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
class SimpleReturn(TypedDict):
samples: list
volumes: list
samples: List[List[ResourceDict]]
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 LiquidHandlerMiddleware(LiquidHandler):
def __init__(self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool = False, channel_num: int = 8, **kwargs):
def __init__(
self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool = False, channel_num: int = 8, **kwargs
):
self._simulator = simulator
self.channel_num = channel_num
self.pending_liquids_dict = {}
joint_config = kwargs.get("joint_config", None)
if simulator:
if joint_config:
self._simulate_backend = UniLiquidHandlerRvizBackend(channel_num, kwargs["total_height"],
joint_config=joint_config, lh_device_id=deck.name)
self._simulate_backend = UniLiquidHandlerRvizBackend(
channel_num, kwargs["total_height"], joint_config=joint_config, lh_device_id=deck.name
)
else:
self._simulate_backend = LiquidHandlerChatterboxBackend(channel_num)
self._simulate_handler = LiquidHandlerAbstract(self._simulate_backend, deck, False)
@@ -159,7 +181,9 @@ class LiquidHandlerMiddleware(LiquidHandler):
if not offsets or (isinstance(offsets, list) and len(offsets) != len(use_channels)):
offsets = [Coordinate.zero()] * len(use_channels)
if self._simulator:
return await self._simulate_handler.discard_tips(use_channels, allow_nonzero_volume, offsets, **backend_kwargs)
return await self._simulate_handler.discard_tips(
use_channels, allow_nonzero_volume, offsets, **backend_kwargs
)
await super().discard_tips(use_channels, allow_nonzero_volume, offsets, **backend_kwargs)
self.pending_liquids_dict = {}
return
@@ -180,7 +204,6 @@ class LiquidHandlerMiddleware(LiquidHandler):
**backend_kwargs,
):
if self._simulator:
return await self._simulate_handler.aspirate(
resources,
@@ -208,15 +231,16 @@ class LiquidHandlerMiddleware(LiquidHandler):
res_samples = []
res_volumes = []
for resource, volume, channel in zip(resources, vols, use_channels):
res_samples.append({"name": resource.name, "sample_uuid": resource.unilabos_extra.get("sample_uuid", None)})
res_samples.append(
{"name": resource.name, "sample_uuid": resource.unilabos_extra.get("sample_uuid", None)}
)
res_volumes.append(volume)
self.pending_liquids_dict[channel] = {
"sample_uuid": resource.unilabos_extra.get("sample_uuid", None),
"volume": volume
"volume": volume,
}
return SimpleReturn(samples=res_samples, volumes=res_volumes)
async def dispense(
self,
resources: Sequence[Container],
@@ -578,10 +602,18 @@ class LiquidHandlerMiddleware(LiquidHandler):
class LiquidHandlerAbstract(LiquidHandlerMiddleware):
"""Extended LiquidHandler with additional operations."""
support_touch_tip = True
_ros_node: BaseROS2DeviceNode
def __init__(self, backend: LiquidHandlerBackend, deck: Deck, simulator: bool=False, channel_num:int = 8, total_height:float = 310):
def __init__(
self,
backend: LiquidHandlerBackend,
deck: Deck,
simulator: bool = False,
channel_num: int = 8,
total_height: float = 310,
):
"""Initialize a LiquidHandler.
Args:
@@ -605,6 +637,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
module_name = ".".join(components[:-1])
try:
import importlib
mod = importlib.import_module(module_name)
except ImportError:
mod = None
@@ -614,6 +647,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
# Try pylabrobot style import (if available)
try:
import pylabrobot
backend_cls = getattr(pylabrobot, type_str, None)
except Exception:
backend_cls = None
@@ -631,16 +665,67 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
self._ros_node = ros_node
@classmethod
def set_liquid(cls, wells: list[Well], liquid_names: list[str], volumes: list[float]) -> SimpleReturn:
"""Set the liquid in a well."""
res_samples = []
def set_liquid(cls, wells: list[Well], liquid_names: list[str], volumes: list[float]) -> SetLiquidReturn:
"""Set the liquid in a well.
如果 liquid_names 和 volumes 为空,但 wells 不为空,直接返回 wells。
"""
res_volumes = []
# 如果 liquid_names 和 volumes 都为空,直接返回 wells
if not liquid_names and not volumes:
return SetLiquidReturn(
wells=ResourceTreeSet.from_plr_resources(wells, known_newly_created=False).dump(), volumes=res_volumes # type: ignore
)
for well, liquid_name, volume in zip(wells, liquid_names, volumes):
well.set_liquids([(liquid_name, volume)]) # type: ignore
res_samples.append({"name": well.name, "sample_uuid": well.unilabos_extra.get("sample_uuid", None)})
res_volumes.append(volume)
return SimpleReturn(samples=res_samples, volumes=res_volumes)
return SetLiquidReturn(
wells=ResourceTreeSet.from_plr_resources(wells, known_newly_created=False).dump(), volumes=res_volumes # type: ignore
)
def set_liquid_from_plate(
self, plate: List[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。
"""
if isinstance(plate, list): # 未来移除
plate = plate[0]
assert issubclass(plate.__class__, Plate), "plate must be a Plate"
plate: Plate = cast(Plate, plate)
# 根据 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)
task = ROS2DeviceNode.run_async_func(self._ros_node.update_resource, True, **{"resources": wells})
submit_time = time.time()
while not task.done():
if time.time() - submit_time > 10:
self._ros_node.lab_logger().info(f"set_liquid_from_plate {plate} 超时")
break
time.sleep(0.01)
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 --------------------------------------------------
# ---------------------------------------------------------------
@@ -676,7 +761,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
target_rack = child
target_rack = cast(TipRack, target_rack)
available_tips = {}
for (idx, tipSpot) in enumerate(target_rack.get_all_items()):
for idx, tipSpot in enumerate(target_rack.get_all_items()):
if tipSpot.has_tip():
available_tips[idx] = tipSpot
continue
@@ -684,8 +769,8 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
print("channel_num", self.channel_num)
if self.channel_num == 8:
tip_prefix = list(available_tips.values())[0].name.split('_')[0]
colnum_list = [int(tip.name.split('_')[-1][1:]) for tip in available_tips.values()]
tip_prefix = list(available_tips.values())[0].name.split("_")[0]
colnum_list = [int(tip.name.split("_")[-1][1:]) for tip in available_tips.values()]
available_cols = [colnum for colnum, count in dict(Counter(colnum_list)).items() if count == 8]
available_cols.sort()
available_tips_dict = {tip.name: tip for tip in available_tips.values()}
@@ -729,7 +814,6 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
"""Create a new protocol with the given metadata."""
pass
async def remove_liquid(
self,
vols: List[float],
@@ -788,10 +872,11 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
elif len(use_channels) == 8 and self.backend.num_channels == 8:
# 对于8个的情况需要判断此时任务是不是能被8通道移液站来成功处理
if len(sources) % 8 != 0:
raise ValueError(f"Length of `sources` {len(sources)} must be a multiple of 8 for 8-channel mode.")
raise ValueError(
f"Length of `sources` {len(sources)} must be a multiple of 8 for 8-channel mode."
)
# 8个8个来取任务序列
@@ -805,13 +890,23 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
current_asp_vols = vols[i : i + 8]
current_dis_vols = vols[i : i + 8]
current_asp_flow_rates = flow_rates[i : i + 8] if flow_rates else [None] * 8
current_dis_flow_rates = flow_rates[-i*8-8:len(flow_rates)-i*8] if flow_rates else [None] * 8
current_dis_flow_rates = (
flow_rates[-i * 8 - 8 : len(flow_rates) - i * 8] if flow_rates else [None] * 8
)
current_asp_offset = offsets[i : i + 8] if offsets else [None] * 8
current_dis_offset = offsets[-i * 8 - 8 : len(offsets) - i * 8] if offsets else [None] * 8
current_asp_liquid_height = liquid_height[i : i + 8] if liquid_height else [None] * 8
current_dis_liquid_height = liquid_height[-i*8-8:len(liquid_height)-i*8] if liquid_height else [None] * 8
current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
current_dis_blow_out_air_volume = blow_out_air_volume[-i*8-8:len(blow_out_air_volume)-i*8] if blow_out_air_volume else [None] * 8
current_dis_liquid_height = (
liquid_height[-i * 8 - 8 : len(liquid_height) - i * 8] if liquid_height else [None] * 8
)
current_asp_blow_out_air_volume = (
blow_out_air_volume[i : i + 8] if blow_out_air_volume else [None] * 8
)
current_dis_blow_out_air_volume = (
blow_out_air_volume[-i * 8 - 8 : len(blow_out_air_volume) - i * 8]
if blow_out_air_volume
else [None] * 8
)
await self.aspirate(
resources=current_reagent_sources,
@@ -941,13 +1036,23 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
current_asp_vols = asp_vols[i : i + 8]
current_dis_vols = dis_vols[i : i + 8]
current_asp_flow_rates = flow_rates[i : i + 8] if flow_rates else [None] * 8
current_dis_flow_rates = flow_rates[-i*8-8:len(flow_rates)-i*8] if flow_rates else [None] * 8
current_dis_flow_rates = (
flow_rates[-i * 8 - 8 : len(flow_rates) - i * 8] if flow_rates else [None] * 8
)
current_asp_offset = offsets[i : i + 8] if offsets else [None] * 8
current_dis_offset = offsets[-i * 8 - 8 : len(offsets) - i * 8] if offsets else [None] * 8
current_asp_liquid_height = liquid_height[i : i + 8] if liquid_height else [None] * 8
current_dis_liquid_height = liquid_height[-i*8-8:len(liquid_height)-i*8] if liquid_height else [None] * 8
current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
current_dis_blow_out_air_volume = blow_out_air_volume[-i*8-8:len(blow_out_air_volume)-i*8] if blow_out_air_volume else [None] * 8
current_dis_liquid_height = (
liquid_height[-i * 8 - 8 : len(liquid_height) - i * 8] if liquid_height else [None] * 8
)
current_asp_blow_out_air_volume = (
blow_out_air_volume[i : i + 8] if blow_out_air_volume else [None] * 8
)
current_dis_blow_out_air_volume = (
blow_out_air_volume[-i * 8 - 8 : len(blow_out_air_volume) - i * 8]
if blow_out_air_volume
else [None] * 8
)
await self.aspirate(
resources=current_reagent_sources,
@@ -989,7 +1094,6 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
await self.touch_tip(current_targets)
await self.discard_tips()
# except Exception as e:
# traceback.print_exc()
# raise RuntimeError(f"Liquid addition failed: {e}") from e
@@ -1021,7 +1125,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
mix_liquid_height: Optional[float] = None,
delays: Optional[List[int]] = None,
none_keys: List[str] = [],
):
) -> TransferLiquidReturn:
"""Transfer liquid with automatic mode detection.
Supports three transfer modes:
@@ -1089,29 +1193,71 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
if num_sources == 1 and num_targets > 1:
# 模式1: 一对多 (1 source -> N targets)
await self._transfer_one_to_many(
sources[0], targets, tip_racks, use_channels,
asp_vols, dis_vols, asp_flow_rates, dis_flow_rates,
offsets, touch_tip, liquid_height, blow_out_air_volume,
spread, mix_stage, mix_times, mix_vol, mix_rate,
mix_liquid_height, delays
sources[0],
targets,
tip_racks,
use_channels,
asp_vols,
dis_vols,
asp_flow_rates,
dis_flow_rates,
offsets,
touch_tip,
liquid_height,
blow_out_air_volume,
spread,
mix_stage,
mix_times,
mix_vol,
mix_rate,
mix_liquid_height,
delays,
)
elif num_sources > 1 and num_targets == 1:
# 模式2: 多对一 (N sources -> 1 target)
await self._transfer_many_to_one(
sources, targets[0], tip_racks, use_channels,
asp_vols, dis_vols, asp_flow_rates, dis_flow_rates,
offsets, touch_tip, liquid_height, blow_out_air_volume,
spread, mix_stage, mix_times, mix_vol, mix_rate,
mix_liquid_height, delays
sources,
targets[0],
tip_racks,
use_channels,
asp_vols,
dis_vols,
asp_flow_rates,
dis_flow_rates,
offsets,
touch_tip,
liquid_height,
blow_out_air_volume,
spread,
mix_stage,
mix_times,
mix_vol,
mix_rate,
mix_liquid_height,
delays,
)
elif num_sources == num_targets:
# 模式3: 一对一 (N sources -> N targets)
await self._transfer_one_to_one(
sources, targets, tip_racks, use_channels,
asp_vols, dis_vols, asp_flow_rates, dis_flow_rates,
offsets, touch_tip, liquid_height, blow_out_air_volume,
spread, mix_stage, mix_times, mix_vol, mix_rate,
mix_liquid_height, delays
sources,
targets,
tip_racks,
use_channels,
asp_vols,
dis_vols,
asp_flow_rates,
dis_flow_rates,
offsets,
touch_tip,
liquid_height,
blow_out_air_volume,
spread,
mix_stage,
mix_times,
mix_vol,
mix_rate,
mix_liquid_height,
delays,
)
else:
raise ValueError(
@@ -1119,6 +1265,11 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
"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(
self,
sources: Sequence[Container],
@@ -1174,7 +1325,9 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
flow_rates=[asp_flow_rates[_]] if asp_flow_rates and len(asp_flow_rates) > _ else None,
offsets=[offsets[_]] if offsets and len(offsets) > _ else None,
liquid_height=[liquid_height[_]] if liquid_height and len(liquid_height) > _ else None,
blow_out_air_volume=[blow_out_air_volume[_]] if blow_out_air_volume and len(blow_out_air_volume) > _ else None,
blow_out_air_volume=(
[blow_out_air_volume[_]] if blow_out_air_volume and len(blow_out_air_volume) > _ else None
),
spread=spread,
)
if delays is not None:
@@ -1185,7 +1338,9 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
use_channels=use_channels,
flow_rates=[dis_flow_rates[_]] if dis_flow_rates and len(dis_flow_rates) > _ else None,
offsets=[offsets[_]] if offsets and len(offsets) > _ else None,
blow_out_air_volume=[blow_out_air_volume[_]] if blow_out_air_volume and len(blow_out_air_volume) > _ else None,
blow_out_air_volume=(
[blow_out_air_volume[_]] if blow_out_air_volume and len(blow_out_air_volume) > _ else None
),
liquid_height=[liquid_height[_]] if liquid_height and len(liquid_height) > _ else None,
spread=spread,
)
@@ -1337,7 +1492,9 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
flow_rates=[asp_flow_rates[0]] if asp_flow_rates and len(asp_flow_rates) > 0 else None,
offsets=[offsets[0]] if offsets and len(offsets) > 0 else None,
liquid_height=[liquid_height[0]] if liquid_height and len(liquid_height) > 0 else None,
blow_out_air_volume=[blow_out_air_volume[0]] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None,
blow_out_air_volume=(
[blow_out_air_volume[0]] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None
),
spread=spread,
)
@@ -1352,7 +1509,9 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
use_channels=use_channels,
flow_rates=[dis_flow_rates[idx]] if dis_flow_rates and len(dis_flow_rates) > idx else None,
offsets=[offsets[idx]] if offsets and len(offsets) > idx else None,
blow_out_air_volume=[blow_out_air_volume[idx]] if blow_out_air_volume and len(blow_out_air_volume) > idx else None,
blow_out_air_volume=(
[blow_out_air_volume[idx]] if blow_out_air_volume and len(blow_out_air_volume) > idx else None
),
liquid_height=[liquid_height[idx]] if liquid_height and len(liquid_height) > idx else None,
spread=spread,
)
@@ -1388,10 +1547,18 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
current_dis_vols = dis_vols[i : i + 8]
# 8个通道都从同一个源容器吸液每个通道的吸液体积等于对应的分液体积
current_asp_flow_rates = asp_flow_rates[0:1] * 8 if asp_flow_rates and len(asp_flow_rates) > 0 else None
current_asp_flow_rates = (
asp_flow_rates[0:1] * 8 if asp_flow_rates and len(asp_flow_rates) > 0 else None
)
current_asp_offset = offsets[0:1] * 8 if offsets and len(offsets) > 0 else [None] * 8
current_asp_liquid_height = liquid_height[0:1] * 8 if liquid_height and len(liquid_height) > 0 else [None] * 8
current_asp_blow_out_air_volume = blow_out_air_volume[0:1] * 8 if blow_out_air_volume and len(blow_out_air_volume) > 0 else [None] * 8
current_asp_liquid_height = (
liquid_height[0:1] * 8 if liquid_height and len(liquid_height) > 0 else [None] * 8
)
current_asp_blow_out_air_volume = (
blow_out_air_volume[0:1] * 8
if blow_out_air_volume and len(blow_out_air_volume) > 0
else [None] * 8
)
if mix_stage in ["before", "both"] and mix_times is not None and mix_times > 0:
await self.mix(
@@ -1524,7 +1691,9 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
flow_rates=[asp_flow_rates[idx]] if asp_flow_rates and len(asp_flow_rates) > idx else None,
offsets=[offsets[idx]] if offsets and len(offsets) > idx else None,
liquid_height=[liquid_height[idx]] if liquid_height and len(liquid_height) > idx else None,
blow_out_air_volume=[blow_out_air_volume[idx]] if blow_out_air_volume and len(blow_out_air_volume) > idx else None,
blow_out_air_volume=(
[blow_out_air_volume[idx]] if blow_out_air_volume and len(blow_out_air_volume) > idx else None
),
spread=spread,
)
@@ -1538,14 +1707,18 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
dis_flow_rate = dis_flow_rates[idx] if dis_flow_rates and len(dis_flow_rates) > idx else None
dis_offset = offsets[idx] if offsets and len(offsets) > idx else None
dis_liquid_height = liquid_height[idx] if liquid_height and len(liquid_height) > idx else None
dis_blow_out = blow_out_air_volume[idx] if blow_out_air_volume and len(blow_out_air_volume) > idx else None
dis_blow_out = (
blow_out_air_volume[idx] if blow_out_air_volume and len(blow_out_air_volume) > idx else None
)
else:
# 标准模式:分液体积等于吸液体积
dis_vol = asp_vols[idx]
dis_flow_rate = dis_flow_rates[0] if dis_flow_rates and len(dis_flow_rates) > 0 else None
dis_offset = offsets[0] if offsets and len(offsets) > 0 else None
dis_liquid_height = liquid_height[0] if liquid_height and len(liquid_height) > 0 else None
dis_blow_out = blow_out_air_volume[0] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None
dis_blow_out = (
blow_out_air_volume[0] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None
)
await self.dispense(
resources=[target],
@@ -1628,14 +1801,18 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
current_dis_flow_rates = dis_flow_rates[i : i + 8] if dis_flow_rates else None
current_dis_offset = offsets[i : i + 8] if offsets else [None] * 8
current_dis_liquid_height = liquid_height[i : i + 8] if liquid_height else [None] * 8
current_dis_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
current_dis_blow_out_air_volume = (
blow_out_air_volume[i : i + 8] if blow_out_air_volume else [None] * 8
)
else:
# 标准模式:每个通道分液体积等于其吸液体积
current_dis_vols = current_asp_vols
current_dis_flow_rates = dis_flow_rates[0:1] * 8 if dis_flow_rates else None
current_dis_offset = offsets[0:1] * 8 if offsets else [None] * 8
current_dis_liquid_height = liquid_height[0:1] * 8 if liquid_height else [None] * 8
current_dis_blow_out_air_volume = blow_out_air_volume[0:1] * 8 if blow_out_air_volume else [None] * 8
current_dis_blow_out_air_volume = (
blow_out_air_volume[0:1] * 8 if blow_out_air_volume else [None] * 8
)
await self.dispense(
resources=[target] * 8, # 8个通道都分到同一个目标
@@ -1671,7 +1848,6 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
# traceback.print_exc()
# raise RuntimeError(f"Liquid addition failed: {e}") from e
# ---------------------------------------------------------------
# Helper utilities
# ---------------------------------------------------------------
@@ -1692,7 +1868,6 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
print(f"Current time: {time.strftime('%H:%M:%S')}")
async def touch_tip(self, targets: Sequence[Container]):
"""Touch the tip to the side of the well."""
if not self.support_touch_tip:

View File

@@ -30,9 +30,31 @@ from pylabrobot.liquid_handling.standard import (
ResourceMove,
ResourceDrop,
)
from pylabrobot.resources import ResourceHolder, ResourceStack, Tip, Deck, Plate, Well, TipRack, Resource, Container, Coordinate, TipSpot, Trash, PlateAdapter, TubeRack
from pylabrobot.resources import (
ResourceHolder,
ResourceStack,
Tip,
Deck,
Plate,
Well,
TipRack,
Resource,
Container,
Coordinate,
TipSpot,
Trash,
PlateAdapter,
TubeRack,
)
from unilabos.devices.liquid_handling.liquid_handler_abstract import LiquidHandlerAbstract, SimpleReturn
from unilabos.devices.liquid_handling.liquid_handler_abstract import (
LiquidHandlerAbstract,
SimpleReturn,
SetLiquidReturn,
SetLiquidFromPlateReturn,
TransferLiquidReturn,
)
from unilabos.registry.placeholder_type import ResourceSlot
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
@@ -80,6 +102,7 @@ class PRCXI9300Deck(Deck):
self.slots[slot - 1] = resource
super().assign_child_resource(resource, location=self.slot_locations[slot - 1])
class PRCXI9300Container(Plate):
"""PRCXI 9300 的专用 Container 类,继承自 Plate用于槽位定位和未知模块。
@@ -109,73 +132,80 @@ class PRCXI9300Container(Plate):
data = super().serialize_state()
data.update(self._unilabos_state)
return data
class PRCXI9300Plate(Plate):
"""
专用孔板类:
1. 继承自 PLR 原生 Plate保留所有物理特性。
2. 增加 material_info 参数,用于在初始化时直接绑定 Unilab UUID。
"""
def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str = "plate",
ordered_items: collections.OrderedDict = None,
ordering: Optional[collections.OrderedDict] = None,
model: Optional[str] = None,
material_info: Optional[Dict[str, Any]] = None,
**kwargs):
**kwargs,
):
# 如果 ordered_items 不为 None直接使用
items = None
ordering_param = None
if ordered_items is not None:
items = ordered_items
elif ordering is not None:
# 检查 ordering 中的值是否是字符串(从 JSON 反序列化时的情况)
# 如果是字符串,说明这是位置名称,需要让 Plate 自己创建 Well 对象
# 我们只传递位置信息(键),不传递值,使用 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 参数而不是 ordered_items让 Plate 自己创建 Well 对象
items = None
# 使用 ordering 参数,只包含位置信息(键)
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
elif value is None:
ordering_param = ordering
else:
# ordering 的值已经是对象,可以直接使用
items = ordering
ordering_param = None
else:
items = None
ordering_param = None
# 根据情况传递不同的参数
if items is not None:
super().__init__(name, size_x, size_y, size_z,
ordered_items=items,
category=category,
model=model, **kwargs)
super().__init__(
name, size_x, size_y, size_z, ordered_items=items, category=category, model=model, **kwargs
)
elif ordering_param is not None:
# 传递 ordering 参数,让 Plate 自己创建 Well 对象
super().__init__(name, size_x, size_y, size_z,
ordering=ordering_param,
category=category,
model=model, **kwargs)
super().__init__(
name, size_x, size_y, size_z, ordering=ordering_param, category=category, model=model, **kwargs
)
else:
super().__init__(name, size_x, size_y, size_z,
category=category,
model=model, **kwargs)
super().__init__(name, size_x, size_y, size_z, category=category, model=model, **kwargs)
self._unilabos_state = {}
if material_info:
self._unilabos_state["Material"] = material_info
def load_state(self, state: Dict[str, Any]) -> None:
super().load_state(state)
self._unilabos_state = state
def serialize_state(self) -> Dict[str, Dict[str, Any]]:
try:
data = super().serialize_state()
except AttributeError:
data = {}
if hasattr(self, '_unilabos_state') and self._unilabos_state:
if hasattr(self, "_unilabos_state") and self._unilabos_state:
safe_state = {}
for k, v in self._unilabos_state.items():
# 如果是 Material 字典,深入检查
@@ -196,15 +226,24 @@ class PRCXI9300Plate(Plate):
data.update(safe_state)
return data # 其他顶层属性也进行类型检查
class PRCXI9300TipRack(TipRack):
"""专用吸头盒类"""
def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str = "tip_rack",
ordered_items: collections.OrderedDict = None,
ordering: Optional[collections.OrderedDict] = None,
model: Optional[str] = None,
material_info: Optional[Dict[str, Any]] = None,
**kwargs):
**kwargs,
):
# 如果 ordered_items 不为 None直接使用
if ordered_items is not None:
items = ordered_items
@@ -228,20 +267,16 @@ class PRCXI9300TipRack(TipRack):
# 根据情况传递不同的参数
if items is not None:
super().__init__(name, size_x, size_y, size_z,
ordered_items=items,
category=category,
model=model, **kwargs)
super().__init__(
name, size_x, size_y, size_z, ordered_items=items, category=category, model=model, **kwargs
)
elif ordering_param is not None:
# 传递 ordering 参数,让 TipRack 自己创建 Tip 对象
super().__init__(name, size_x, size_y, size_z,
ordering=ordering_param,
category=category,
model=model, **kwargs)
super().__init__(
name, size_x, size_y, size_z, ordering=ordering_param, category=category, model=model, **kwargs
)
else:
super().__init__(name, size_x, size_y, size_z,
category=category,
model=model, **kwargs)
super().__init__(name, size_x, size_y, size_z, category=category, model=model, **kwargs)
self._unilabos_state = {}
if material_info:
self._unilabos_state["Material"] = material_info
@@ -255,7 +290,7 @@ class PRCXI9300TipRack(TipRack):
data = super().serialize_state()
except AttributeError:
data = {}
if hasattr(self, '_unilabos_state') and self._unilabos_state:
if hasattr(self, "_unilabos_state") and self._unilabos_state:
safe_state = {}
for k, v in self._unilabos_state.items():
# 如果是 Material 字典,深入检查
@@ -277,16 +312,23 @@ class PRCXI9300TipRack(TipRack):
data.update(safe_state)
return data
class PRCXI9300Trash(Trash):
"""PRCXI 9300 的专用 Trash 类,继承自 Trash。
该类定义了 PRCXI 9300 的工作台布局和槽位信息。
"""
def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str = "trash",
material_info: Optional[Dict[str, Any]] = None,
**kwargs):
**kwargs,
):
if name != "trash":
print(f"Warning: PRCXI9300Trash usually expects name='trash' for backend logic, but got '{name}'.")
@@ -306,7 +348,7 @@ class PRCXI9300Trash(Trash):
data = super().serialize_state()
except AttributeError:
data = {}
if hasattr(self, '_unilabos_state') and self._unilabos_state:
if hasattr(self, "_unilabos_state") and self._unilabos_state:
safe_state = {}
for k, v in self._unilabos_state.items():
# 如果是 Material 字典,深入检查
@@ -328,19 +370,27 @@ class PRCXI9300Trash(Trash):
data.update(safe_state)
return data
class PRCXI9300TubeRack(TubeRack):
"""
专用管架类:用于 EP 管架、试管架等。
继承自 PLR 的 TubeRack并支持注入 material_info (UUID)。
"""
def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str = "tube_rack",
items: Optional[Dict[str, Any]] = None,
ordered_items: Optional[OrderedDict] = None,
ordering: Optional[OrderedDict] = None,
model: Optional[str] = None,
material_info: Optional[Dict[str, Any]] = None,
**kwargs):
**kwargs,
):
# 如果 ordered_items 不为 None直接使用
if ordered_items is not None:
@@ -370,20 +420,12 @@ class PRCXI9300TubeRack(TubeRack):
# 根据情况传递不同的参数
if items_to_pass is not None:
super().__init__(name, size_x, size_y, size_z,
ordered_items=items_to_pass,
model=model,
**kwargs)
super().__init__(name, size_x, size_y, size_z, ordered_items=items_to_pass, model=model, **kwargs)
elif ordering_param is not None:
# 传递 ordering 参数,让 TubeRack 自己创建 Tube 对象
super().__init__(name, size_x, size_y, size_z,
ordering=ordering_param,
model=model,
**kwargs)
super().__init__(name, size_x, size_y, size_z, ordering=ordering_param, model=model, **kwargs)
else:
super().__init__(name, size_x, size_y, size_z,
model=model,
**kwargs)
super().__init__(name, size_x, size_y, size_z, model=model, **kwargs)
self._unilabos_state = {}
if material_info:
@@ -394,7 +436,7 @@ class PRCXI9300TubeRack(TubeRack):
data = super().serialize_state()
except AttributeError:
data = {}
if hasattr(self, '_unilabos_state') and self._unilabos_state:
if hasattr(self, "_unilabos_state") and self._unilabos_state:
safe_state = {}
for k, v in self._unilabos_state.items():
# 如果是 Material 字典,深入检查
@@ -416,12 +458,19 @@ class PRCXI9300TubeRack(TubeRack):
data.update(safe_state)
return data
class PRCXI9300PlateAdapter(PlateAdapter):
"""
专用板式适配器类:用于承载 Plate 的底座(如 PCR 适配器、磁吸架等)。
支持注入 material_info (UUID)。
"""
def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str = "plate_adapter",
model: Optional[str] = None,
material_info: Optional[Dict[str, Any]] = None,
@@ -432,7 +481,8 @@ class PRCXI9300PlateAdapter(PlateAdapter):
dx: Optional[float] = None,
dy: Optional[float] = None,
dz: float = 0.0, # 默认Z轴偏移
**kwargs):
**kwargs,
):
# 自动居中计算:如果未指定 dx/dy则根据适配器尺寸和孔尺寸计算居中位置
if dx is None:
@@ -452,7 +502,7 @@ class PRCXI9300PlateAdapter(PlateAdapter):
adapter_hole_size_y=adapter_hole_size_y,
adapter_hole_size_z=adapter_hole_size_z,
model=model,
**kwargs
**kwargs,
)
self._unilabos_state = {}
@@ -464,7 +514,7 @@ class PRCXI9300PlateAdapter(PlateAdapter):
data = super().serialize_state()
except AttributeError:
data = {}
if hasattr(self, '_unilabos_state') and self._unilabos_state:
if hasattr(self, "_unilabos_state") and self._unilabos_state:
safe_state = {}
for k, v in self._unilabos_state.items():
# 如果是 Material 字典,深入检查
@@ -486,6 +536,7 @@ class PRCXI9300PlateAdapter(PlateAdapter):
data.update(safe_state)
return data
class PRCXI9300Handler(LiquidHandlerAbstract):
support_touch_tip = False
@@ -518,7 +569,9 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
if "Material" in child.children[0]._unilabos_state:
number = int(child.name.replace("T", ""))
tablets_info.append(
WorkTablets(Number=number, Code=f"T{number}", Material=child.children[0]._unilabos_state["Material"])
WorkTablets(
Number=number, Code=f"T{number}", Material=child.children[0]._unilabos_state["Material"]
)
)
if is_9320:
print("当前设备是9320")
@@ -538,9 +591,14 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
super().post_init(ros_node)
self._unilabos_backend.post_init(ros_node)
def set_liquid(self, wells: list[Well], liquid_names: list[str], volumes: list[float]) -> SimpleReturn:
def set_liquid(self, wells: list[Well], liquid_names: list[str], volumes: list[float]) -> SetLiquidReturn:
return super().set_liquid(wells, liquid_names, volumes)
def set_liquid_from_plate(
self, plate: List[ResourceSlot], well_names: list[str], liquid_names: list[str], volumes: list[float]
) -> SetLiquidFromPlateReturn:
return super().set_liquid_from_plate(plate, well_names, liquid_names, volumes)
def set_group(self, group_name: str, wells: List[Well], volumes: List[float]):
return super().set_group(group_name, wells, volumes)
@@ -660,7 +718,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
mix_liquid_height: Optional[float] = None,
delays: Optional[List[int]] = None,
none_keys: List[str] = [],
):
) -> TransferLiquidReturn:
return await super().transfer_liquid(
sources,
targets,
@@ -800,6 +858,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
async def heater_action(self, temperature: float, time: int):
return await self._unilabos_backend.heater_action(temperature, time)
async def move_plate(
self,
plate: Plate,
@@ -826,6 +885,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
**backend_kwargs,
)
class PRCXI9300Backend(LiquidHandlerBackend):
"""PRCXI 9300 的后端实现,继承自 LiquidHandlerBackend。
@@ -878,7 +938,6 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self.steps_todo_list.append(step)
return step
async def pick_up_resource(self, pickup: ResourcePickup, **backend_kwargs):
resource = pickup.resource
@@ -896,13 +955,11 @@ class PRCXI9300Backend(LiquidHandlerBackend):
async def drop_resource(self, drop: ResourceDrop, **backend_kwargs):
plate_number = None
target_plate_number = backend_kwargs.get("target_plate_number", None)
if target_plate_number is not None:
plate_number = int(target_plate_number.name.replace("T", ""))
is_whole_plate = True
balance_height = 0
if plate_number is None:
@@ -911,7 +968,6 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self.steps_todo_list.append(step)
return step
async def heater_action(self, temperature: float, time: int):
print(f"\n\nHeater action: temperature={temperature}, time={time}\n\n")
# return await self.api_client.heater_action(temperature, time)
@@ -980,7 +1036,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
# 检查重置状态并等待完成
while not self.is_reset_ok:
print("Waiting for PRCXI9300 to reset...")
if hasattr(self, '_ros_node') and self._ros_node is not None:
if hasattr(self, "_ros_node") and self._ros_node is not None:
await self._ros_node.sleep(1)
else:
await asyncio.sleep(1)
@@ -998,7 +1054,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
"""Pick up tips from the specified resource."""
# INSERT_YOUR_CODE
# Ensure use_channels is converted to a list of ints if it's an array
if hasattr(use_channels, 'tolist'):
if hasattr(use_channels, "tolist"):
_use_channels = use_channels.tolist()
else:
_use_channels = list(use_channels) if use_channels is not None else None
@@ -1052,7 +1108,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
async def drop_tips(self, ops: List[Drop], use_channels: List[int] = None):
"""Pick up tips from the specified resource."""
if hasattr(use_channels, 'tolist'):
if hasattr(use_channels, "tolist"):
_use_channels = use_channels.tolist()
else:
_use_channels = list(use_channels) if use_channels is not None else None
@@ -1178,7 +1234,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int] = None):
"""Aspirate liquid from the specified resources."""
if hasattr(use_channels, 'tolist'):
if hasattr(use_channels, "tolist"):
_use_channels = use_channels.tolist()
else:
_use_channels = list(use_channels) if use_channels is not None else None
@@ -1235,7 +1291,7 @@ class PRCXI9300Backend(LiquidHandlerBackend):
async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int] = None):
"""Dispense liquid into the specified resources."""
if hasattr(use_channels, 'tolist'):
if hasattr(use_channels, "tolist"):
_use_channels = use_channels.tolist()
else:
_use_channels = list(use_channels) if use_channels is not None else None
@@ -1416,7 +1472,6 @@ class PRCXI9300Api:
time.sleep(1)
return success
def call(self, service: str, method: str, params: Optional[list] = None) -> Any:
payload = json.dumps(
{"ServiceName": service, "MethodName": method, "Paramters": params or []}, separators=(",", ":")
@@ -1681,11 +1736,11 @@ class PRCXI9300Api:
"LiquidDispensingMethod": liquid_method,
}
def clamp_jaw_pick_up(self,
def clamp_jaw_pick_up(
self,
plate_no: int,
is_whole_plate: bool,
balance_height: int,
) -> Dict[str, Any]:
return {
"StepAxis": "ClampingJaw",
@@ -1695,7 +1750,7 @@ class PRCXI9300Api:
"HoleRow": 1,
"HoleCol": 1,
"BalanceHeight": balance_height,
"PlateOrHoleNum": f"T{plate_no}"
"PlateOrHoleNum": f"T{plate_no}",
}
def clamp_jaw_drop(
@@ -1703,7 +1758,6 @@ class PRCXI9300Api:
plate_no: int,
is_whole_plate: bool,
balance_height: int,
) -> Dict[str, Any]:
return {
"StepAxis": "ClampingJaw",
@@ -1713,7 +1767,7 @@ class PRCXI9300Api:
"HoleRow": 1,
"HoleCol": 1,
"BalanceHeight": balance_height,
"PlateOrHoleNum": f"T{plate_no}"
"PlateOrHoleNum": f"T{plate_no}",
}
def shaker_action(self, time: int, module_no: int, amplitude: int, is_wait: bool):
@@ -1726,6 +1780,7 @@ class PRCXI9300Api:
"AssistFun4": is_wait,
}
class DefaultLayout:
def __init__(self, product_name: str = "PRCXI9300"):
@@ -2104,7 +2159,9 @@ if __name__ == "__main__":
size_y=50,
size_z=10,
category="tip_rack",
ordered_items=collections.OrderedDict({k: f"{child_prefix}_{k}" for k, v in tip_racks["ordering"].items()}),
ordered_items=collections.OrderedDict(
{k: f"{child_prefix}_{k}" for k, v in tip_racks["ordering"].items()}
),
)
tip_rack_serialized = tip_rack.serialize()
tip_rack_serialized["parent_name"] = deck.name
@@ -2299,21 +2356,19 @@ if __name__ == "__main__":
A = tree_to_list([resource_plr_to_ulab(deck)])
with open("deck.json", "w", encoding="utf-8") as f:
A.insert(0, {
A.insert(
0,
{
"id": "PRCXI",
"name": "PRCXI",
"parent": None,
"type": "device",
"class": "liquid_handler.prcxi",
"position": {
"x": 0,
"y": 0,
"z": 0
},
"position": {"x": 0, "y": 0, "z": 0},
"config": {
"deck": {
"_resource_child_name": "PRCXI_Deck",
"_resource_type": "unilabos.devices.liquid_handling.prcxi.prcxi:PRCXI9300Deck"
"_resource_type": "unilabos.devices.liquid_handling.prcxi.prcxi:PRCXI9300Deck",
},
"host": "192.168.0.121",
"port": 9999,
@@ -2324,18 +2379,14 @@ if __name__ == "__main__":
"debug": True,
"simulator": True,
"matrix_id": "5de524d0-3f95-406c-86dd-f83626ebc7cb",
"is_9320": True
"is_9320": True,
},
"data": {},
"children": [
"PRCXI_Deck"
]
})
"children": ["PRCXI_Deck"],
},
)
A[1]["parent"] = "PRCXI"
json.dump({
"nodes": A,
"links": []
}, f, indent=4, ensure_ascii=False)
json.dump({"nodes": A, "links": []}, f, indent=4, ensure_ascii=False)
handler = PRCXI9300Handler(
deck=deck,
@@ -2377,7 +2428,6 @@ if __name__ == "__main__":
time.sleep(5)
os._exit(0)
prcxi_api = PRCXI9300Api(host="192.168.0.121", port=9999)
prcxi_api.list_matrices()
prcxi_api.get_all_materials()

View File

@@ -638,7 +638,7 @@ liquid_handler:
placeholder_keys: {}
result: {}
schema:
description: 吸头迭代函数。用于自动管理和切换吸头架中的吸头,实现批量实验中的吸头自动分配和追踪。该函数监控吸头使用状态,自动切换到下一个可用吸头位置,确保实验流程的连续性。适用于高通量实验、批量处理、自动化流水线等需要大量吸头管理的应用场景。
description: 吸头迭代函数。用于自动管理和切换枪头盒中的吸头,实现批量实验中的吸头自动分配和追踪。该函数监控吸头使用状态,自动切换到下一个可用吸头位置,确保实验流程的连续性。适用于高通量实验、批量处理、自动化流水线等需要大量吸头管理的应用场景。
properties:
feedback: {}
goal:
@@ -712,6 +712,43 @@ liquid_handler:
title: set_group参数
type: object
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:
feedback: {}
goal: {}
@@ -721,7 +758,7 @@ liquid_handler:
placeholder_keys: {}
result: {}
schema:
description: 吸头架设置函数。用于配置和初始化液体处理系统的吸头架信息,包括吸头架位置、类型、容量等参数。该函数建立吸头资源管理系统,为后续的吸头选择和使用提供基础配置。适用于系统初始化、吸头架更换、实验配置等需要吸头资源管理的操作场景。
description: 枪头盒设置函数。用于配置和初始化液体处理系统的枪头盒信息,包括枪头盒位置、类型、容量等参数。该函数建立吸头资源管理系统,为后续的吸头选择和使用提供基础配置。适用于系统初始化、枪头盒更换、实验配置等需要吸头资源管理的操作场景。
properties:
feedback: {}
goal:
@@ -4093,32 +4130,32 @@ liquid_handler:
- 0
handles:
input:
- data_key: liquid
- data_key: sources
data_source: handle
data_type: resource
handler_key: sources
label: sources
- data_key: liquid
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
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:
- data_key: sources.@flatten
data_source: executor
data_type: resource
handler_key: sources_out
label: sources
- data_key: liquid
label: 移液后源孔
- data_key: targets.@flatten
data_source: executor
data_type: resource
handler_key: targets_out
label: targets
label: 移液后目标孔
placeholder_keys:
sources: unilabos_resources
targets: unilabos_resources
@@ -5114,19 +5151,34 @@ liquid_handler.biomek:
- 0
handles:
input:
- data_key: liquid
- data_key: sources
data_source: handle
data_type: resource
handler_key: liquid-input
handler_key: sources
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:
- data_key: liquid
- data_key: sources.@flatten
data_source: executor
data_type: resource
handler_key: liquid-output
handler_key: sources_out
io_type: source
label: Liquid Output
label: 移液后源孔
- data_key: targets.@flatten
data_source: executor
data_type: resource
handler_key: targets_out
label: 移液后目标孔
placeholder_keys:
sources: unilabos_resources
targets: unilabos_resources
@@ -9284,7 +9336,13 @@ liquid_handler.prcxi:
data_source: handle
data_type: resource
handler_key: input_wells
label: InputWells
label: 待设定液体孔
output:
- data_key: wells.@flatten
data_source: executor
data_type: resource
handler_key: output_wells
label: 已设定液体孔
placeholder_keys:
wells: unilabos_resources
result: {}
@@ -9400,6 +9458,165 @@ liquid_handler.prcxi:
title: LiquidHandlerSetLiquid
type: object
type: LiquidHandlerSetLiquid
set_liquid_from_plate:
feedback: {}
goal: {}
goal_default:
liquid_names: null
plate: null
volumes: null
well_names: null
handles:
input:
- data_key: plate
data_source: handle
data_type: resource
handler_key: input_plate
label: 待设定液体板
output:
- data_key: plate.@flatten
data_source: executor
data_type: resource
handler_key: output_plate
label: 已设定液体板
- data_key: wells.@flatten
data_source: executor
data_type: resource
handler_key: output_wells
label: 已设定液体孔
- data_key: volumes
data_source: executor
data_type: number_array
handler_key: output_volumes
label: 各孔设定体积
placeholder_keys:
plate: unilabos_resources
result: {}
schema:
description: ''
properties:
feedback: {}
goal:
properties:
liquid_names:
items:
type: string
type: array
plate:
items:
properties:
category:
type: string
children:
items:
type: string
type: array
config:
type: string
data:
type: string
id:
type: string
name:
type: string
parent:
type: string
pose:
properties:
orientation:
properties:
w:
type: number
x:
type: number
y:
type: number
z:
type: number
required:
- x
- y
- z
- w
title: orientation
type: object
position:
properties:
x:
type: number
y:
type: number
z:
type: number
required:
- x
- y
- z
title: position
type: object
required:
- position
- orientation
title: pose
type: object
sample_id:
type: string
type:
type: string
required:
- id
- name
- sample_id
- children
- parent
- type
- category
- pose
- config
- data
title: plate
type: object
title: plate
type: array
volumes:
items:
type: number
type: array
well_names:
items:
type: string
type: array
required:
- plate
- well_names
- liquid_names
- volumes
type: object
result:
properties:
plate:
items: {}
title: Plate
type: array
volumes:
items: {}
title: Volumes
type: array
wells:
items: {}
title: Wells
type: array
required:
- plate
- wells
- volumes
title: SetLiquidFromPlateReturn
type: object
required:
- goal
title: set_liquid_from_plate参数
type: object
type: UniLabJsonCommand
set_tiprack:
feedback: {}
goal:
@@ -9759,18 +9976,18 @@ liquid_handler.prcxi:
data_source: handle
data_type: resource
handler_key: tip_rack_identifier
label: 头盒
label: 头盒
output:
- data_key: liquid
data_source: handle
- data_key: sources.@flatten
data_source: executor
data_type: resource
handler_key: sources_out
label: sources
- data_key: liquid
label: 移液后源孔
- data_key: targets.@flatten
data_source: executor
data_type: resource
handler_key: targets_out
label: targets
label: 移液后目标孔
placeholder_keys:
sources: unilabos_resources
targets: unilabos_resources

View File

@@ -4,6 +4,8 @@ import os
import sys
import inspect
import importlib
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Dict, List, Union, Tuple
@@ -60,6 +62,7 @@ class Registry:
self.device_module_to_registry = {}
self.resource_type_registry = {}
self._setup_called = False # 跟踪setup是否已调用
self._registry_lock = threading.Lock() # 多线程加载时的锁
# 其他状态变量
# self.is_host_mode = False # 移至BasicConfig中
@@ -163,6 +166,7 @@ class Registry:
"res_id": "unilabos_resources", # 将当前实验室的全部物料id作为下拉框可选择
"device_id": "unilabos_devices", # 将当前实验室的全部设备id作为下拉框可选择
"parent": "unilabos_nodes", # 将当前实验室的设备/物料作为下拉框可选择
"class_name": "unilabos_class",
},
},
"test_latency": {
@@ -176,8 +180,7 @@ class Registry:
"result": {},
"schema": test_latency_schema,
"goal_default": {
arg["name"]: arg["default"]
for arg in test_latency_method_info.get("args", [])
arg["name"]: arg["default"] for arg in test_latency_method_info.get("args", [])
},
"handles": {},
},
@@ -261,18 +264,26 @@ class Registry:
# 标记setup已被调用
self._setup_called = True
def load_resource_types(self, path: os.PathLike, complete_registry: bool, upload_registry: bool):
abs_path = Path(path).absolute()
resource_path = abs_path / "resources"
files = list(resource_path.glob("*/*.yaml"))
logger.trace(f"[UniLab Registry] load resources? {resource_path.exists()}, total: {len(files)}")
current_resource_number = len(self.resource_type_registry) + 1
for i, file in enumerate(files):
def _load_single_resource_file(
self, file: Path, complete_registry: bool, upload_registry: bool
) -> Tuple[Dict[str, Any], Dict[str, Any], bool]:
"""
加载单个资源文件 (线程安全)
Returns:
(data, complete_data, is_valid): 资源数据, 完整数据, 是否有效
"""
try:
with open(file, encoding="utf-8", mode="r") as f:
data = yaml.safe_load(io.StringIO(f.read()))
except Exception as e:
logger.warning(f"[UniLab Registry] 读取资源文件失败: {file}, 错误: {e}")
return {}, {}, False
if not data:
return {}, {}, False
complete_data = {}
if data:
# 为每个资源添加文件路径信息
for resource_id, resource_info in data.items():
if "version" not in resource_info:
resource_info["version"] = "1.0.0"
@@ -300,28 +311,68 @@ class Registry:
if len(class_info) and "module" in class_info:
if class_info.get("type") == "pylabrobot":
res_class = get_class(class_info["module"])
if callable(res_class) and not isinstance(
res_class, type
): # 有的是类,有的是函数,这里暂时只登记函数类的
if callable(res_class) and not isinstance(res_class, type):
res_instance = res_class(res_class.__name__)
res_ulr = tree_to_list([resource_plr_to_ulab(res_instance)])
resource_info["config_info"] = res_ulr
resource_info["registry_type"] = "resource"
resource_info["file_path"] = str(file.absolute()).replace("\\", "/")
complete_data = dict(sorted(complete_data.items()))
complete_data = copy.deepcopy(complete_data)
if complete_registry:
try:
with open(file, "w", encoding="utf-8") as f:
yaml.dump(complete_data, f, allow_unicode=True, default_flow_style=False, Dumper=NoAliasDumper)
except Exception as e:
logger.warning(f"[UniLab Registry] 写入资源文件失败: {file}, 错误: {e}")
return data, complete_data, True
def load_resource_types(self, path: os.PathLike, complete_registry: bool, upload_registry: bool):
abs_path = Path(path).absolute()
resource_path = abs_path / "resources"
files = list(resource_path.glob("*/*.yaml"))
logger.debug(f"[UniLab Registry] resources: {resource_path.exists()}, total: {len(files)}")
if not files:
return
# 使用线程池并行加载
max_workers = min(8, len(files))
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_file = {
executor.submit(self._load_single_resource_file, file, complete_registry, upload_registry): file
for file in files
}
for future in as_completed(future_to_file):
file = future_to_file[future]
try:
data, complete_data, is_valid = future.result()
if is_valid:
results.append((file, data))
except Exception as e:
logger.warning(f"[UniLab Registry] 处理资源文件异常: {file}, 错误: {e}")
# 线程安全地更新注册表
current_resource_number = len(self.resource_type_registry) + 1
with self._registry_lock:
for i, (file, data) in enumerate(results):
self.resource_type_registry.update(data)
logger.trace( # type: ignore
f"[UniLab Registry] Resource-{current_resource_number} File-{i+1}/{len(files)} "
logger.trace(
f"[UniLab Registry] Resource-{current_resource_number} File-{i+1}/{len(results)} "
+ f"Add {list(data.keys())}"
)
current_resource_number += 1
else:
logger.debug(f"[UniLab Registry] Res File-{i+1}/{len(files)} Not Valid YAML File: {file.absolute()}")
# 记录无效文件
valid_files = {r[0] for r in results}
for file in files:
if file not in valid_files:
logger.debug(f"[UniLab Registry] Res File Not Valid YAML File: {file.absolute()}")
def _extract_class_docstrings(self, module_string: str) -> Dict[str, str]:
"""
@@ -673,32 +724,34 @@ class Registry:
"handles": {},
}
def load_device_types(self, path: os.PathLike, complete_registry: bool):
# return
abs_path = Path(path).absolute()
devices_path = abs_path / "devices"
device_comms_path = abs_path / "device_comms"
files = list(devices_path.glob("*.yaml")) + list(device_comms_path.glob("*.yaml"))
logger.trace( # type: ignore
f"[UniLab Registry] devices: {devices_path.exists()}, device_comms: {device_comms_path.exists()}, "
+ f"total: {len(files)}"
)
current_device_number = len(self.device_type_registry) + 1
from unilabos.app.web.utils.action_utils import get_yaml_from_goal_type
def _load_single_device_file(
self, file: Path, complete_registry: bool, get_yaml_from_goal_type
) -> Tuple[Dict[str, Any], Dict[str, Any], bool, List[str]]:
"""
加载单个设备文件 (线程安全)
for i, file in enumerate(files):
Returns:
(data, complete_data, is_valid, device_ids): 设备数据, 完整数据, 是否有效, 设备ID列表
"""
try:
with open(file, encoding="utf-8", mode="r") as f:
data = yaml.safe_load(io.StringIO(f.read()))
except Exception as e:
logger.warning(f"[UniLab Registry] 读取设备文件失败: {file}, 错误: {e}")
return {}, {}, False, []
if not data:
return {}, {}, False, []
complete_data = {}
action_str_type_mapping = {
"UniLabJsonCommand": "UniLabJsonCommand",
"UniLabJsonCommandAsync": "UniLabJsonCommandAsync",
}
status_str_type_mapping = {}
if data:
# 在添加到注册表前处理类型替换
device_ids = []
for device_id, device_config in data.items():
# 添加文件路径信息 - 使用规范化的完整文件路径
if "version" not in device_config:
device_config["version"] = "1.0.0"
if "category" not in device_config:
@@ -716,10 +769,7 @@ class Registry:
if "init_param_schema" not in device_config:
device_config["init_param_schema"] = {}
if "class" in device_config:
if (
"status_types" not in device_config["class"]
or device_config["class"]["status_types"] is None
):
if "status_types" not in device_config["class"] or device_config["class"]["status_types"] is None:
device_config["class"]["status_types"] = {}
if (
"action_value_mappings" not in device_config["class"]
@@ -737,25 +787,17 @@ class Registry:
)
for status_name, status_type in device_config["class"]["status_types"].items():
if isinstance(status_type, tuple) or status_type in ["Any", "None", "Unknown"]:
status_type = "String" # 替换成ROS的String便于显示
status_type = "String"
device_config["class"]["status_types"][status_name] = status_type
try:
target_type = self._replace_type_with_class(
status_type, device_id, f"状态 {status_name}"
)
target_type = self._replace_type_with_class(status_type, device_id, f"状态 {status_name}")
except ROSMsgNotFound:
continue
if target_type in [
dict,
list,
]: # 对于嵌套类型返回的对象,暂时处理成字符串,无法直接进行转换
if target_type in [dict, list]:
target_type = String
status_str_type_mapping[status_type] = target_type
device_config["class"]["status_types"] = dict(
sorted(device_config["class"]["status_types"].items())
)
device_config["class"]["status_types"] = dict(sorted(device_config["class"]["status_types"].items()))
if complete_registry:
# 保存原有的 action 配置(用于保留 schema 的 description 和 handles 等)
old_action_configs = {}
for action_name, action_config in device_config["class"]["action_value_mappings"].items():
old_action_configs[action_name] = action_config
@@ -765,7 +807,6 @@ class Registry:
for k, v in device_config["class"]["action_value_mappings"].items()
if not k.startswith("auto-")
}
# 处理动作值映射
device_config["class"]["action_value_mappings"].update(
{
f"auto-{k}": {
@@ -777,18 +818,15 @@ class Registry:
v["args"],
k,
v.get("return_annotation"),
# 传入旧的 schema 以保留字段 description
old_action_configs.get(f"auto-{k}", {}).get("schema"),
),
"goal_default": {i["name"]: i["default"] for i in v["args"]},
# 保留原有的 handles 配置
"handles": old_action_configs.get(f"auto-{k}", {}).get("handles", []),
"placeholder_keys": {
i["name"]: (
"unilabos_resources"
if i["type"] == "unilabos.registry.placeholder_type:ResourceSlot"
or i["type"]
== ("list", "unilabos.registry.placeholder_type:ResourceSlot")
or i["type"] == ("list", "unilabos.registry.placeholder_type:ResourceSlot")
else "unilabos_devices"
)
for i in v["args"]
@@ -801,14 +839,12 @@ class Registry:
]
},
}
# 不生成已配置action的动作
for k, v in enhanced_info["action_methods"].items()
if k not in device_config["class"]["action_value_mappings"]
}
)
# 恢复原有的 description 信息(非 auto- 开头的动作)
for action_name, old_config in old_action_configs.items():
if action_name in device_config["class"]["action_value_mappings"]: # 有一些会被删除
if action_name in device_config["class"]["action_value_mappings"]:
old_schema = old_config.get("schema", {})
if "description" in old_schema and old_schema["description"]:
device_config["class"]["action_value_mappings"][action_name]["schema"][
@@ -837,7 +873,6 @@ class Registry:
action_config["handles"] = {}
if "type" in action_config:
action_type_str: str = action_config["type"]
# 通过Json发放指令而不是通过特殊的ros action进行处理
if not action_type_str.startswith("UniLabJsonCommand"):
try:
target_type = self._replace_type_with_class(
@@ -855,31 +890,78 @@ class Registry:
logger.warning(
f"[UniLab Registry] 设备 {device_id} 的动作 {action_name} 类型为空,跳过替换"
)
complete_data[device_id] = copy.deepcopy(dict(sorted(device_config.items()))) # 稍后dump到文件
complete_data[device_id] = copy.deepcopy(dict(sorted(device_config.items())))
for status_name, status_type in device_config["class"]["status_types"].items():
device_config["class"]["status_types"][status_name] = status_str_type_mapping[status_type]
for action_name, action_config in device_config["class"]["action_value_mappings"].items():
if action_config["type"] not in action_str_type_mapping:
continue
action_config["type"] = action_str_type_mapping[action_config["type"]]
# 添加内置的驱动命令动作
self._add_builtin_actions(device_config, device_id)
device_config["file_path"] = str(file.absolute()).replace("\\", "/")
device_config["registry_type"] = "device"
logger.trace( # type: ignore
f"[UniLab Registry] Device-{current_device_number} File-{i+1}/{len(files)} Add {device_id} "
device_ids.append(device_id)
complete_data = dict(sorted(complete_data.items()))
complete_data = copy.deepcopy(complete_data)
try:
with open(file, "w", encoding="utf-8") as f:
yaml.dump(complete_data, f, allow_unicode=True, default_flow_style=False, Dumper=NoAliasDumper)
except Exception as e:
logger.warning(f"[UniLab Registry] 写入设备文件失败: {file}, 错误: {e}")
return data, complete_data, True, device_ids
def load_device_types(self, path: os.PathLike, complete_registry: bool):
abs_path = Path(path).absolute()
devices_path = abs_path / "devices"
device_comms_path = abs_path / "device_comms"
files = list(devices_path.glob("*.yaml")) + list(device_comms_path.glob("*.yaml"))
logger.trace(
f"[UniLab Registry] devices: {devices_path.exists()}, device_comms: {device_comms_path.exists()}, "
+ f"total: {len(files)}"
)
if not files:
return
from unilabos.app.web.utils.action_utils import get_yaml_from_goal_type
# 使用线程池并行加载
max_workers = min(8, len(files))
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_file = {
executor.submit(self._load_single_device_file, file, complete_registry, get_yaml_from_goal_type): file
for file in files
}
for future in as_completed(future_to_file):
file = future_to_file[future]
try:
data, complete_data, is_valid, device_ids = future.result()
if is_valid:
results.append((file, data, device_ids))
except Exception as e:
logger.warning(f"[UniLab Registry] 处理设备文件异常: {file}, 错误: {e}")
# 线程安全地更新注册表
current_device_number = len(self.device_type_registry) + 1
with self._registry_lock:
for file, data, device_ids in results:
self.device_type_registry.update(data)
for device_id in device_ids:
logger.trace(
f"[UniLab Registry] Device-{current_device_number} Add {device_id} "
+ f"[{data[device_id].get('name', '未命名设备')}]"
)
current_device_number += 1
complete_data = dict(sorted(complete_data.items()))
complete_data = copy.deepcopy(complete_data)
with open(file, "w", encoding="utf-8") as f:
yaml.dump(complete_data, f, allow_unicode=True, default_flow_style=False, Dumper=NoAliasDumper)
self.device_type_registry.update(data)
else:
logger.debug(
f"[UniLab Registry] Device File-{i+1}/{len(files)} Not Valid YAML File: {file.absolute()}"
)
# 记录无效文件
valid_files = {r[0] for r in results}
for file in files:
if file not in valid_files:
logger.debug(f"[UniLab Registry] Device File Not Valid YAML File: {file.absolute()}")
def obtain_registry_device_info(self):
devices = []

View File

@@ -151,12 +151,40 @@ def canonicalize_links_ports(links: List[Dict[str, Any]], resource_tree_set: Res
"""
# 构建 id 到 uuid 的映射
id_to_uuid: Dict[str, str] = {}
uuid_to_id: Dict[str, str] = {}
for node in resource_tree_set.all_nodes:
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转换为字典格式
for link in links:
port = link.get("port")
if port is None:
continue
if link.get("type", "physical") == "physical":
link["type"] = "fluid"
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}
# 构建边字典,键为(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信息
delete_reverses = []
for i, link in enumerate(links):
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:
reverse_key = (t, s)
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]
# 删除已被使用反向端口信息的反向边
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
@@ -260,7 +276,7 @@ def read_node_link_json(
resource_tree_set = canonicalize_nodes_data(nodes)
# 标准化边数据
links = data.get("links", [])
links = data.get("links", data.get("edges", []))
standardized_links = canonicalize_links_ports(links, resource_tree_set)
# 构建 NetworkX 图(需要转换回 dict 格式)
@@ -284,6 +300,8 @@ def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]
edge["sourceHandle"] = port[source]
elif "source_port" in edge:
edge["sourceHandle"] = edge.pop("source_port")
elif "source_handle" in edge:
edge["sourceHandle"] = edge.pop("source_handle")
else:
typ = edge.get("type")
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]
elif "target_port" in edge:
edge["targetHandle"] = edge.pop("target_port")
elif "target_handle" in edge:
edge["targetHandle"] = edge.pop("target_handle")
else:
typ = edge.get("type")
if typ == "communication":

View File

@@ -13,6 +13,9 @@ if TYPE_CHECKING:
from pylabrobot.resources import Resource as PLRResource
EXTRA_CLASS = "unilabos_resource_class"
class ResourceDictPositionSize(BaseModel):
depth: float = Field(description="Depth", default=0.0) # z
width: float = Field(description="Width", default=0.0) # x
@@ -393,7 +396,7 @@ class ResourceTreeSet(object):
"parent": parent_resource, # 直接传入 ResourceDict 对象
"parent_uuid": parent_uuid, # 使用 parent_uuid 而不是 parent 对象
"type": replace_plr_type(d.get("category", "")),
"class": d.get("class", ""),
"class": extra.get(EXTRA_CLASS, ""),
"position": pos,
"pose": pos,
"config": {
@@ -443,7 +446,7 @@ class ResourceTreeSet(object):
trees.append(tree_instance)
return cls(trees)
def to_plr_resources(self) -> List["PLRResource"]:
def to_plr_resources(self, skip_devices=True) -> List["PLRResource"]:
"""
将 ResourceTreeSet 转换为 PLR 资源列表
@@ -468,6 +471,7 @@ class ResourceTreeSet(object):
name_to_uuid[node.res_content.name] = node.res_content.uuid
all_states[node.res_content.name] = node.res_content.data
name_to_extra[node.res_content.name] = node.res_content.extra
name_to_extra[node.res_content.name][EXTRA_CLASS] = node.res_content.klass
for child in node.children:
collect_node_data(child, name_to_uuid, all_states, name_to_extra)
@@ -512,7 +516,10 @@ class ResourceTreeSet(object):
plr_dict = node_to_plr_dict(tree.root_node, has_model)
try:
sub_cls = find_subclass(plr_dict["type"], PLRResource)
if sub_cls is None:
if skip_devices and plr_dict["type"] == "device":
logger.info(f"跳过更新 {plr_dict['name']} 设备是class")
continue
elif sub_cls is None:
raise ValueError(
f"无法找到类型 {plr_dict['type']} 对应的 PLR 资源类。原始信息:{tree.root_node.res_content}"
)
@@ -520,6 +527,10 @@ class ResourceTreeSet(object):
if "category" not in spec.parameters:
plr_dict.pop("category", None)
plr_resource = sub_cls.deserialize(plr_dict, allow_marshal=True)
from pylabrobot.resources import Coordinate
from pylabrobot.serializer import deserialize
location = cast(Coordinate, deserialize(plr_dict["location"]))
plr_resource.location = location
plr_resource.load_all_state(all_states)
# 使用 DeviceNodeResourceTracker 设置 UUID 和 Extra
tracker.loop_set_uuid(plr_resource, name_to_uuid)
@@ -986,7 +997,7 @@ class DeviceNodeResourceTracker(object):
extra = name_to_extra_map[resource_name]
self.set_resource_extra(res, extra)
if len(extra):
logger.debug(f"设置资源Extra: {resource_name} -> {extra}")
logger.trace(f"设置资源Extra: {resource_name} -> {extra}")
return 1
return 0

View File

@@ -884,6 +884,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
parent_appended = True
# 加载状态
original_instance.location = plr_resource.location
original_instance.rotation = plr_resource.rotation
original_instance.barcode = plr_resource.barcode
original_instance.load_all_state(states)
child_count = len(original_instance.get_all_children())
self.lab_logger().info(
@@ -1578,7 +1581,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
f"转换ResourceSlot列表参数 {arg_name} 失败: {e}\n{traceback.format_exc()}"
)
raise JsonCommandInitError(f"ResourceSlot列表参数转换失败: {arg_name}")
# todo: 默认反报送
return function(**function_args)
except KeyError as ex:
raise JsonCommandInitError(
@@ -1611,8 +1614,8 @@ class BaseROS2DeviceNode(Node, Generic[T]):
timeout = 30.0
elapsed = 0.0
while not future.done() and elapsed < timeout:
time.sleep(0.05)
elapsed += 0.05
time.sleep(0.02)
elapsed += 0.02
if not future.done():
raise Exception(f"资源查询超时: {uuids_list}")

View File

@@ -807,7 +807,7 @@ class HostNode(BaseROS2DeviceNode):
assign_sample_id(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}: {goal_msg}")
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:
from unilabos.app.communication import get_communication_client
from unilabos.app.web.client import HTTPClient, http_client

View File

@@ -6,8 +6,6 @@ from typing import List, Dict, Any, Optional, TYPE_CHECKING
import rclpy
from rosidl_runtime_py import message_to_ordereddict
from unilabos_msgs.msg import Resource
from unilabos_msgs.srv import ResourceUpdate
from unilabos.messages import * # type: ignore # protocol names
from rclpy.action import ActionServer, ActionClient
@@ -15,7 +13,6 @@ from rclpy.action.server import ServerGoalHandle
from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response
from unilabos.compile import action_protocol_generators
from unilabos.resources.graphio import nested_dict_to_list
from unilabos.ros.initialize_device import initialize_device_from_dict
from unilabos.ros.msgs.message_converter import (
get_action_type,
@@ -231,15 +228,15 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
try:
# 统一处理单个或多个资源
resource_id = (
protocol_kwargs[k]["id"] if v == "unilabos_msgs/Resource" else protocol_kwargs[k][0]["id"]
protocol_kwargs[k]["id"]
if v == "unilabos_msgs/Resource"
else protocol_kwargs[k][0]["id"]
)
resource_uuid = protocol_kwargs[k].get("uuid", None)
r = SerialCommand_Request()
r.command = json.dumps({"id": resource_id, "uuid": resource_uuid, "with_children": True})
# 发送请求并等待响应
response: SerialCommand_Response = await self._resource_clients[
"resource_get"
].call_async(
response: SerialCommand_Response = await self._resource_clients["resource_get"].call_async(
r
) # type: ignore
raw_data = json.loads(response.response)
@@ -307,12 +304,52 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
# 向Host更新物料当前状态
for k, v in goal.get_fields_and_field_types().items():
if v in ["unilabos_msgs/Resource", "sequence<unilabos_msgs/Resource>"]:
r = ResourceUpdate.Request()
r.resources = [
convert_to_ros_msg(Resource, rs) for rs in nested_dict_to_list(protocol_kwargs[k])
]
response = await self._resource_clients["resource_update"].call_async(r)
if v not in ["unilabos_msgs/Resource", "sequence<unilabos_msgs/Resource>"]:
continue
self.lab_logger().info(f"更新资源状态: {k}")
try:
# 去重:使用 seen 集合获取唯一的资源对象
seen = set()
unique_resources = []
# 获取资源数据,统一转换为列表
resource_data = protocol_kwargs[k]
is_sequence = v != "unilabos_msgs/Resource"
if not is_sequence:
resource_list = [resource_data] if isinstance(resource_data, dict) else resource_data
else:
# 处理序列类型,可能是嵌套列表
resource_list = []
if isinstance(resource_data, list):
for item in resource_data:
if isinstance(item, list):
resource_list.extend(item)
else:
resource_list.append(item)
else:
resource_list = [resource_data]
for res_data in resource_list:
if not isinstance(res_data, dict):
continue
res_name = res_data.get("id") or res_data.get("name")
if not res_name:
continue
# 使用 resource_tracker 获取本地 PLR 实例
plr = self.resource_tracker.figure_resource({"name": res_name}, try_mode=False)
# 获取父资源
res = self.resource_tracker.parent_resource(plr)
if id(res) not in seen:
seen.add(id(res))
unique_resources.append(res)
# 使用新的资源树接口更新
if unique_resources:
await self.update_resource(unique_resources)
except Exception as e:
self.lab_logger().error(f"资源更新失败: {e}")
self.lab_logger().error(traceback.format_exc())
# 设置成功状态和返回值
execution_success = True

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

@@ -1,3 +1,100 @@
"""
工作流转换模块 - JSON 到 WorkflowGraph 的转换流程
==================== 输入格式 (JSON) ====================
{
"workflow": [
{"action": "transfer_liquid", "action_args": {"sources": "cell_lines", "targets": "Liquid_1", "asp_vol": 100.0, "dis_vol": 74.75, ...}},
...
],
"reagent": {
"cell_lines": {"slot": 4, "well": ["A1", "A3", "A5"], "labware": "DRUG + YOYO-MEDIA"},
"Liquid_1": {"slot": 1, "well": ["A4", "A7", "A10"], "labware": "rep 1"},
...
}
}
==================== 转换步骤 ====================
第一步: 按 slot 去重创建 create_resource 节点(创建板子)
--------------------------------------------------------------------------------
- 首先创建一个 Group 节点type="Group", minimized=true用于包含所有 create_resource 节点
- 遍历所有 reagent按 slot 去重,为每个唯一的 slot 创建一个板子
- 所有 create_resource 节点的 parent_uuid 指向 Group 节点minimized=true
- 生成参数:
res_id: plate_slot_{slot}
device_id: /PRCXI
class_name: PRCXI_BioER_96_wellplate
parent: /PRCXI/PRCXI_Deck/T{slot}
slot_on_deck: "{slot}"
- 输出端口: labware用于连接 set_liquid_from_plate
- 控制流: create_resource 之间通过 ready 端口串联
示例: slot=1, slot=4 -> 创建 1 个 Group + 2 个 create_resource 节点
第二步: 为每个 reagent 创建 set_liquid_from_plate 节点(设置液体)
--------------------------------------------------------------------------------
- 首先创建一个 Group 节点type="Group", minimized=true用于包含所有 set_liquid_from_plate 节点
- 遍历所有 reagent为每个试剂创建 set_liquid_from_plate 节点
- 所有 set_liquid_from_plate 节点的 parent_uuid 指向 Group 节点minimized=true
- 生成参数:
plate: [](通过连接传递,来自 create_resource 的 labware
well_names: ["A1", "A3", "A5"](来自 reagent 的 well 数组)
liquid_names: ["cell_lines", "cell_lines", "cell_lines"](与 well 数量一致)
volumes: [1e5, 1e5, 1e5](与 well 数量一致,默认体积)
- 输入连接: create_resource (labware) -> set_liquid_from_plate (input_plate)
- 输出端口: output_wells用于连接 transfer_liquid
- 控制流: set_liquid_from_plate 连接在所有 create_resource 之后,通过 ready 端口串联
第三步: 解析 workflow创建 transfer_liquid 等动作节点
--------------------------------------------------------------------------------
- 遍历 workflow 数组,为每个动作创建步骤节点
- 参数重命名: asp_vol -> asp_vols, dis_vol -> dis_vols, asp_flow_rate -> asp_flow_rates, dis_flow_rate -> dis_flow_rates
- 参数扩展: 根据 targets 的 wells 数量,将单值扩展为数组
例: asp_vol=100.0, targets 有 3 个 wells -> asp_vols=[100.0, 100.0, 100.0]
- 连接处理: 如果 sources/targets 已通过 set_liquid_from_plate 连接,参数值改为 []
- 输入连接: set_liquid_from_plate (output_wells) -> transfer_liquid (sources_identifier / targets_identifier)
- 输出端口: sources_out, targets_out用于连接下一个 transfer_liquid
==================== 连接关系图 ====================
控制流 (ready 端口串联):
create_resource_1 -> create_resource_2 -> ... -> set_liquid_1 -> set_liquid_2 -> ... -> transfer_liquid_1 -> transfer_liquid_2 -> ...
物料流:
[create_resource] --labware--> [set_liquid_from_plate] --output_wells--> [transfer_liquid] --sources_out/targets_out--> [下一个 transfer_liquid]
(slot=1) (cell_lines) (input_plate) (sources_identifier) (sources_identifier)
(slot=4) (Liquid_1) (targets_identifier) (targets_identifier)
==================== 端口映射 ====================
create_resource:
输出: labware
set_liquid_from_plate:
输入: input_plate
输出: output_plate, output_wells
transfer_liquid:
输入: sources -> sources_identifier, targets -> targets_identifier
输出: 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 的 wells 数量是否匹配
- 检查参数数组长度是否与 wells 数量一致
- 如有问题,在 footer 中添加 [WARN: ...] 标记
"""
import re
import uuid
@@ -8,6 +105,35 @@ from typing import Dict, List, Any, Tuple, Optional
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_DEFAULTS = {
"device_id": "/PRCXI",
"parent_template": "/PRCXI/PRCXI_Deck/T{slot}", # {slot} 会被替换为实际的 slot 值
"class_name": "PRCXI_BioER_96_wellplate",
}
# 默认液体体积 (uL)
DEFAULT_LIQUID_VOLUME = 1e5
# 参数重命名映射:单数 -> 复数(用于 transfer_liquid 等动作)
PARAM_RENAME_MAPPING = {
"asp_vol": "asp_vols",
"dis_vol": "dis_vols",
"asp_flow_rate": "asp_flow_rates",
"dis_flow_rate": "dis_flow_rates",
}
# ---------------- Graph ----------------
@@ -228,7 +354,7 @@ def refactor_data(
def build_protocol_graph(
labware_info: List[Dict[str, Any]],
labware_info: Dict[str, Dict[str, Any]],
protocol_steps: List[Dict[str, Any]],
workstation_name: str,
action_resource_mapping: Optional[Dict[str, str]] = None,
@@ -236,112 +362,267 @@ def build_protocol_graph(
"""统一的协议图构建函数,根据设备类型自动选择构建逻辑
Args:
labware_info: labware 信息字典
labware_info: labware 信息字典,格式为 {name: {slot, well, labware, ...}, ...}
protocol_steps: 协议步骤列表
workstation_name: 工作站名称
action_resource_mapping: action 到 resource_name 的映射字典,可选
"""
G = WorkflowGraph()
resource_last_writer = {}
resource_last_writer = {} # reagent_name -> "node_id:port"
slot_to_create_resource = {} # slot -> create_resource node_id
protocol_steps = refactor_data(protocol_steps, action_resource_mapping)
# 有机化学&移液站协议图构建
WORKSTATION_ID = workstation_name
# 为所有labware创建资源节点
res_index = 0
# ==================== 第一步:按 slot 去重创建 create_resource 节点 ====================
# 收集所有唯一的 slot
slots_info = {} # slot -> {labware, res_id}
for labware_id, item in labware_info.items():
# item_id = item.get("id") or item.get("name", f"item_{uuid.uuid4()}")
node_id = str(uuid.uuid4())
slot = str(item.get("slot", ""))
if slot and slot not in slots_info:
res_id = f"plate_slot_{slot}"
slots_info[slot] = {
"labware": item.get("labware", ""),
"res_id": res_id,
}
# 判断节点类型
if "Rack" in str(labware_id) or "Tip" in str(labware_id):
lab_node_type = "Labware"
description = f"Prepare Labware: {labware_id}"
liquid_type = []
liquid_volume = []
elif item.get("type") == "hardware" or "reactor" in str(labware_id).lower():
if "reactor" not in str(labware_id).lower():
continue
lab_node_type = "Sample"
description = f"Prepare Reactor: {labware_id}"
liquid_type = []
liquid_volume = []
else:
lab_node_type = "Reagent"
description = f"Add Reagent to Flask: {labware_id}"
liquid_type = [labware_id]
liquid_volume = [1e5]
# 创建 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 节点
res_index = 0
last_create_resource_id = None
for slot, info in slots_info.items():
node_id = str(uuid.uuid4())
res_id = info["res_id"]
res_index += 1
G.add_node(
node_id,
template_name="create_resource",
resource_name="host_node",
name=f"Res {res_index}",
description=description,
lab_node_type=lab_node_type,
name=f"Plate {res_index}",
description=f"Create plate on slot {slot}",
lab_node_type="Labware",
footer="create_resource-host_node",
device_name=DEVICE_NAME_HOST,
type=NODE_TYPE_DEFAULT,
parent_uuid=group_node_id, # 指向 Group 节点
minimized=True, # 折叠显示
param={
"res_id": labware_id,
"device_id": WORKSTATION_ID,
"class_name": "container",
"parent": WORKSTATION_ID,
"res_id": res_id,
"device_id": CREATE_RESOURCE_DEFAULTS["device_id"],
"class_name": CREATE_RESOURCE_DEFAULTS["class_name"],
"parent": CREATE_RESOURCE_DEFAULTS["parent_template"].format(slot=slot),
"bind_locations": {"x": 0.0, "y": 0.0, "z": 0.0},
"liquid_input_slot": [-1],
"liquid_type": liquid_type,
"liquid_volume": liquid_volume,
"slot_on_deck": "",
"slot_on_deck": slot,
},
)
resource_last_writer[labware_id] = f"{node_id}:labware"
slot_to_create_resource[slot] = node_id
last_control_node_id = None
# create_resource 之间通过 ready 串联
if last_create_resource_id is not None:
G.add_edge(last_create_resource_id, node_id, source_port="ready", target_port="ready")
last_create_resource_id = node_id
# ==================== 第二步:为每个 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
last_set_liquid_id = last_create_resource_id # set_liquid_from_plate 连接在 create_resource 之后
for labware_id, item in labware_info.items():
# 跳过 Tip/Rack 类型
if "Rack" in str(labware_id) or "Tip" in str(labware_id):
continue
if item.get("type") == "hardware":
continue
slot = str(item.get("slot", ""))
wells = item.get("well", [])
if not wells or not slot:
continue
# res_id 不能有空格
res_id = str(labware_id).replace(" ", "_")
well_count = len(wells)
node_id = str(uuid.uuid4())
set_liquid_index += 1
G.add_node(
node_id,
template_name="set_liquid_from_plate",
resource_name="liquid_handler.prcxi",
name=f"SetLiquid {set_liquid_index}",
description=f"Set liquid: {labware_id}",
lab_node_type="Reagent",
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={
"plate": [], # 通过连接传递
"well_names": wells, # 孔位名数组,如 ["A1", "A3", "A5"]
"liquid_names": [res_id] * well_count,
"volumes": [DEFAULT_LIQUID_VOLUME] * well_count,
},
)
# ready 连接:上一个节点 -> set_liquid_from_plate
if last_set_liquid_id is not None:
G.add_edge(last_set_liquid_id, node_id, source_port="ready", target_port="ready")
last_set_liquid_id = node_id
# 物料流create_resource 的 labware -> set_liquid_from_plate 的 input_plate
create_res_node_id = slot_to_create_resource.get(slot)
if create_res_node_id:
G.add_edge(create_res_node_id, node_id, source_port="labware", target_port="input_plate")
# set_liquid_from_plate 的输出 output_wells 用于连接 transfer_liquid
resource_last_writer[labware_id] = f"{node_id}:output_wells"
last_control_node_id = last_set_liquid_id
# 端口名称映射JSON 字段名 -> 实际 handle key
INPUT_PORT_MAPPING = {
"sources": "sources_identifier",
"targets": "targets_identifier",
"vessel": "vessel",
"to_vessel": "to_vessel",
"from_vessel": "from_vessel",
"reagent": "reagent",
"solvent": "solvent",
"compound": "compound",
}
OUTPUT_PORT_MAPPING = {
"sources": "sources_out", # 输出端口是 xxx_out
"targets": "targets_out", # 输出端口是 xxx_out
"vessel": "vessel_out",
"to_vessel": "to_vessel_out",
"from_vessel": "from_vessel_out",
"filtrate_vessel": "filtrate_out",
"reagent": "reagent",
"solvent": "solvent",
"compound": "compound",
}
# 需要根据 wells 数量扩展的参数列表(复数形式)
EXPAND_BY_WELLS_PARAMS = ["asp_vols", "dis_vols", "asp_flow_rates", "dis_flow_rates"]
# 处理协议步骤
for step in protocol_steps:
node_id = str(uuid.uuid4())
G.add_node(node_id, **step)
params = step.get("param", {}).copy() # 复制一份,避免修改原数据
connected_params = set() # 记录被连接的参数
warnings = [] # 收集警告信息
# 参数重命名:单数 -> 复数
for old_name, new_name in PARAM_RENAME_MAPPING.items():
if old_name in params:
params[new_name] = params.pop(old_name)
# 处理输入连接
for param_key, target_port in INPUT_PORT_MAPPING.items():
resource_name = params.get(param_key)
if resource_name and resource_name in resource_last_writer:
source_node, source_port = resource_last_writer[resource_name].split(":")
G.add_edge(source_node, node_id, source_port=source_port, target_port=target_port)
connected_params.add(param_key)
elif resource_name and resource_name not in resource_last_writer:
# 资源名在 labware_info 中不存在
warnings.append(f"{param_key}={resource_name} 未找到")
# 获取 targets 对应的 wells 数量,用于扩展参数
targets_name = params.get("targets")
sources_name = params.get("sources")
targets_wells_count = 1
sources_wells_count = 1
if targets_name and targets_name in labware_info:
target_wells = labware_info[targets_name].get("well", [])
targets_wells_count = len(target_wells) if target_wells else 1
elif targets_name:
warnings.append(f"targets={targets_name} 未在 reagent 中定义")
if sources_name and sources_name in labware_info:
source_wells = labware_info[sources_name].get("well", [])
sources_wells_count = len(source_wells) if source_wells else 1
elif sources_name:
warnings.append(f"sources={sources_name} 未在 reagent 中定义")
# 检查 sources 和 targets 的 wells 数量是否匹配
if targets_wells_count != sources_wells_count and targets_name and sources_name:
warnings.append(f"wells 数量不匹配: sources={sources_wells_count}, targets={targets_wells_count}")
# 使用 targets 的 wells 数量来扩展参数
wells_count = targets_wells_count
# 扩展单值参数为数组(根据 targets 的 wells 数量)
for expand_param in EXPAND_BY_WELLS_PARAMS:
if expand_param in params:
value = params[expand_param]
# 如果是单个值,扩展为数组
if not isinstance(value, list):
params[expand_param] = [value] * wells_count
# 如果已经是数组但长度不对,记录警告
elif len(value) != wells_count:
warnings.append(f"{expand_param} 数量({len(value)})与 wells({wells_count})不匹配")
# 如果 sources/targets 已通过连接传递,将参数值改为空数组
for param_key in connected_params:
if param_key in params:
params[param_key] = []
# 更新 step 的 param、footer、device_name 和 type
step_copy = step.copy()
step_copy["param"] = params
step_copy["device_name"] = DEVICE_NAME_DEFAULT # 动作节点使用默认设备名
step_copy["type"] = NODE_TYPE_DEFAULT # 节点类型
# 如果有警告,修改 footer 添加警告标记(警告放前面)
if warnings:
original_footer = step.get("footer", "")
step_copy["footer"] = f"[WARN: {'; '.join(warnings)}] {original_footer}"
G.add_node(node_id, **step_copy)
# 控制流
if last_control_node_id is not None:
G.add_edge(last_control_node_id, node_id, source_port="ready", target_port="ready")
last_control_node_id = node_id
# 物料流
params = step.get("param", {})
input_resources_possible_names = [
"vessel",
"to_vessel",
"from_vessel",
"reagent",
"solvent",
"compound",
"sources",
"targets",
]
for target_port in input_resources_possible_names:
resource_name = params.get(target_port)
if resource_name and resource_name in resource_last_writer:
source_node, source_port = resource_last_writer[resource_name].split(":")
G.add_edge(source_node, node_id, source_port=source_port, target_port=target_port)
output_resources = {
"vessel_out": params.get("vessel"),
"from_vessel_out": params.get("from_vessel"),
"to_vessel_out": params.get("to_vessel"),
"filtrate_out": params.get("filtrate_vessel"),
"reagent": params.get("reagent"),
"solvent": params.get("solvent"),
"compound": params.get("compound"),
"sources_out": params.get("sources"),
"targets_out": params.get("targets"),
}
for source_port, resource_name in output_resources.items():
# 处理输出:更新 resource_last_writer
for param_key, output_port in OUTPUT_PORT_MAPPING.items():
resource_name = step.get("param", {}).get(param_key) # 使用原始参数值
if resource_name:
resource_last_writer[resource_name] = f"{node_id}:{source_port}"
resource_last_writer[resource_name] = f"{node_id}:{output_port}"
return G

View File

@@ -1,21 +1,68 @@
"""
JSON 工作流转换模块
提供从多种 JSON 格式转换为统一工作流格式的功能
支持的格式:
1. workflow/reagent 格式
2. steps_info/labware_info 格式
将 workflow/reagent 格式的 JSON 转换为统一工作流格式。
输入格式:
{
"workflow": [
{"action": "...", "action_args": {...}},
...
],
"reagent": {
"reagent_name": {"slot": int, "well": [...], "labware": "..."},
...
}
}
"""
import json
from os import PathLike
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from typing import Any, Dict, List, Optional, Tuple, Union
from unilabos.workflow.common import WorkflowGraph, build_protocol_graph
from unilabos.registry.registry import lab_registry
# ==================== 字段映射配置 ====================
# action 到 resource_name 的映射
ACTION_RESOURCE_MAPPING: Dict[str, str] = {
# 生物实验操作
"transfer_liquid": "liquid_handler.prcxi",
"transfer": "liquid_handler.prcxi",
"incubation": "incubator.prcxi",
"move_labware": "labware_mover.prcxi",
"oscillation": "shaker.prcxi",
# 有机化学操作
"HeatChillToTemp": "heatchill.chemputer",
"StopHeatChill": "heatchill.chemputer",
"StartHeatChill": "heatchill.chemputer",
"HeatChill": "heatchill.chemputer",
"Dissolve": "stirrer.chemputer",
"Transfer": "liquid_handler.chemputer",
"Evaporate": "rotavap.chemputer",
"Recrystallize": "reactor.chemputer",
"Filter": "filter.chemputer",
"Dry": "dryer.chemputer",
"Add": "liquid_handler.chemputer",
}
# action_args 字段到 parameters 字段的映射
# 格式: {"old_key": "new_key"}, 仅映射需要重命名的字段
ARGS_FIELD_MAPPING: Dict[str, str] = {
# 如果需要字段重命名,在这里配置
# "old_field_name": "new_field_name",
}
# 默认工作站名称
DEFAULT_WORKSTATION = "PRCXI"
# ==================== 核心转换函数 ====================
def get_action_handles(resource_name: str, template_name: str) -> Dict[str, List[str]]:
"""
从 registry 获取指定设备和动作的 handles 配置
@@ -39,12 +86,10 @@ def get_action_handles(resource_name: str, template_name: str) -> Dict[str, List
handles = action_config.get("handles", {})
if isinstance(handles, dict):
# 处理 input handles (作为 target)
for handle in handles.get("input", []):
handler_key = handle.get("handler_key", "")
if handler_key:
result["source"].append(handler_key)
# 处理 output handles (作为 source)
for handle in handles.get("output", []):
handler_key = handle.get("handler_key", "")
if handler_key:
@@ -69,12 +114,9 @@ def validate_workflow_handles(graph: WorkflowGraph) -> Tuple[bool, List[str]]:
for edge in graph.edges:
left_uuid = edge.get("source")
right_uuid = edge.get("target")
# target_handle_key是target, right的输入节点入节点
# source_handle_key是source, left的输出节点出节点
right_source_conn_key = edge.get("target_handle_key", "")
left_target_conn_key = edge.get("source_handle_key", "")
# 获取源节点和目标节点信息
left_node = nodes.get(left_uuid, {})
right_node = nodes.get(right_uuid, {})
@@ -83,164 +125,93 @@ def validate_workflow_handles(graph: WorkflowGraph) -> Tuple[bool, List[str]]:
right_res_name = right_node.get("resource_name", "")
right_template_name = right_node.get("template_name", "")
# 获取源节点的 output handles
left_node_handles = get_action_handles(left_res_name, left_template_name)
target_valid_keys = left_node_handles.get("target", [])
target_valid_keys.append("ready")
# 获取目标节点的 input handles
right_node_handles = get_action_handles(right_res_name, right_template_name)
source_valid_keys = right_node_handles.get("source", [])
source_valid_keys.append("ready")
# 如果节点配置了 output handles则 source_port 必须有效
# 验证目标节点right的输入端口
if not right_source_conn_key:
node_name = left_node.get("name", left_uuid[:8])
errors.append(f"节点 '{node_name}' source_handle_key 为空," f"应设置为: {source_valid_keys}")
node_name = right_node.get("name", right_uuid[:8])
errors.append(f"目标节点 '{node_name}'输入端口 (target_handle_key) 为空,应设置为: {source_valid_keys}")
elif right_source_conn_key not in source_valid_keys:
node_name = left_node.get("name", left_uuid[:8])
node_name = right_node.get("name", right_uuid[:8])
errors.append(
f"节点 '{node_name}' source 端点 '{right_source_conn_key}' 不存在," f"支持的端点: {source_valid_keys}"
f"目标节点 '{node_name}'输入端口 '{right_source_conn_key}' 不存在,支持的输入端口: {source_valid_keys}"
)
# 如果节点配置了 input handles则 target_port 必须有效
# 验证源节点left的输出端口
if not left_target_conn_key:
node_name = right_node.get("name", right_uuid[:8])
errors.append(f"目标节点 '{node_name}' target_handle_key 为空," f"应设置为: {target_valid_keys}")
node_name = left_node.get("name", left_uuid[:8])
errors.append(f"节点 '{node_name}'输出端口 (source_handle_key) 为空,应设置为: {target_valid_keys}")
elif left_target_conn_key not in target_valid_keys:
node_name = right_node.get("name", right_uuid[:8])
node_name = left_node.get("name", left_uuid[:8])
errors.append(
f"目标节点 '{node_name}' target 端点 '{left_target_conn_key}' 不存在,"
f"支持的端点: {target_valid_keys}"
f"节点 '{node_name}'输出端口 '{left_target_conn_key}' 不存在,支持的输出端口: {target_valid_keys}"
)
return len(errors) == 0, errors
# action 到 resource_name 的映射
ACTION_RESOURCE_MAPPING: Dict[str, str] = {
# 生物实验操作
"transfer_liquid": "liquid_handler.prcxi",
"transfer": "liquid_handler.prcxi",
"incubation": "incubator.prcxi",
"move_labware": "labware_mover.prcxi",
"oscillation": "shaker.prcxi",
# 有机化学操作
"HeatChillToTemp": "heatchill.chemputer",
"StopHeatChill": "heatchill.chemputer",
"StartHeatChill": "heatchill.chemputer",
"HeatChill": "heatchill.chemputer",
"Dissolve": "stirrer.chemputer",
"Transfer": "liquid_handler.chemputer",
"Evaporate": "rotavap.chemputer",
"Recrystallize": "reactor.chemputer",
"Filter": "filter.chemputer",
"Dry": "dryer.chemputer",
"Add": "liquid_handler.chemputer",
}
def normalize_steps(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
def normalize_workflow_steps(workflow: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
不同格式的步骤数据规范化为统一格式
workflow 格式的步骤数据规范化
支持的输入格式
- action + parameters
- action + action_args
- operation + parameters
输入格式:
[{"action": "...", "action_args": {...}}, ...]
输出格式:
[{"action": "...", "parameters": {...}, "step_number": int}, ...]
Args:
data: 原始步骤数据列表
workflow: workflow 数组
Returns:
规范化后的步骤列表,格式为 [{"action": str, "parameters": dict, "description": str?, "step_number": int?}, ...]
规范化后的步骤列表
"""
normalized = []
for idx, step in enumerate(data):
# 获取动作名称(支持 action 或 operation 字段)
action = step.get("action") or step.get("operation")
for idx, step in enumerate(workflow):
action = step.get("action")
if not action:
continue
# 获取参数(支持 parameters 或 action_args 字段)
raw_params = step.get("parameters") or step.get("action_args") or {}
params = dict(raw_params)
# 获取参数: action_args
raw_params = step.get("action_args", {})
params = {}
# 规范化 source/target -> sources/targets
if "source" in raw_params and "sources" not in raw_params:
params["sources"] = raw_params["source"]
if "target" in raw_params and "targets" not in raw_params:
params["targets"] = raw_params["target"]
# 应用字段映射
for key, value in raw_params.items():
mapped_key = ARGS_FIELD_MAPPING.get(key, key)
params[mapped_key] = value
# 获取描述(支持 description 或 purpose 字段)
description = step.get("description") or step.get("purpose")
step_dict = {
"action": action,
"parameters": params,
"step_number": idx + 1,
}
# 获取步骤编号(优先使用原始数据中的 step_number否则使用索引+1
step_number = step.get("step_number", idx + 1)
step_dict = {"action": action, "parameters": params, "step_number": step_number}
if description:
step_dict["description"] = description
# 保留描述字段
if "description" in step:
step_dict["description"] = step["description"]
normalized.append(step_dict)
return normalized
def normalize_labware(data: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
"""
将不同格式的 labware 数据规范化为统一的字典格式
支持的输入格式:
- reagent_name + material_name + positions
- name + labware + slot
Args:
data: 原始 labware 数据列表
Returns:
规范化后的 labware 字典,格式为 {name: {"slot": int, "labware": str, "well": list, "type": str, "role": str, "name": str}, ...}
"""
labware = {}
for item in data:
# 获取 key 名称(优先使用 reagent_name其次是 material_name 或 name
reagent_name = item.get("reagent_name")
key = reagent_name or item.get("material_name") or item.get("name")
if not key:
continue
key = str(key)
# 处理重复 key自动添加后缀
idx = 1
original_key = key
while key in labware:
idx += 1
key = f"{original_key}_{idx}"
labware[key] = {
"slot": item.get("positions") or item.get("slot"),
"labware": item.get("material_name") or item.get("labware"),
"well": item.get("well", []),
"type": item.get("type", "reagent"),
"role": item.get("role", ""),
"name": key,
}
return labware
def convert_from_json(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
workstation_name: str = DEFAULT_WORKSTATION,
validate: bool = True,
) -> WorkflowGraph:
"""
从 JSON 数据或文件转换为 WorkflowGraph
支持的 JSON 格式
1. {"workflow": [...], "reagent": {...}} - 直接格式
2. {"steps_info": [...], "labware_info": [...]} - 需要规范化的格式
JSON 格式:
{"workflow": [...], "reagent": {...}}
Args:
data: JSON 文件路径、字典数据、或 JSON 字符串
@@ -251,7 +222,7 @@ def convert_from_json(
WorkflowGraph: 构建好的工作流图
Raises:
ValueError: 不支持的 JSON 格式 或 句柄校验失败
ValueError: 不支持的 JSON 格式
FileNotFoundError: 文件不存在
json.JSONDecodeError: JSON 解析失败
"""
@@ -262,7 +233,6 @@ def convert_from_json(
with path.open("r", encoding="utf-8") as fp:
json_data = json.load(fp)
elif isinstance(data, str):
# 尝试作为 JSON 字符串解析
json_data = json.loads(data)
else:
raise FileNotFoundError(f"文件不存在: {data}")
@@ -271,30 +241,24 @@ def convert_from_json(
else:
raise TypeError(f"不支持的数据类型: {type(data)}")
# 根据格式解析数据
if "workflow" in json_data and "reagent" in json_data:
# 格式1: workflow/reagent已经是规范格式
protocol_steps = json_data["workflow"]
labware_info = json_data["reagent"]
elif "steps_info" in json_data and "labware_info" in json_data:
# 格式2: steps_info/labware_info需要规范化
protocol_steps = normalize_steps(json_data["steps_info"])
labware_info = normalize_labware(json_data["labware_info"])
elif "steps" in json_data and "labware" in json_data:
# 格式3: steps/labware另一种常见格式
protocol_steps = normalize_steps(json_data["steps"])
if isinstance(json_data["labware"], list):
labware_info = normalize_labware(json_data["labware"])
else:
labware_info = json_data["labware"]
else:
# 校验格式
if "workflow" not in json_data or "reagent" not in json_data:
raise ValueError(
"不支持的 JSON 格式。支持的格式\n"
"1. {'workflow': [...], 'reagent': {...}}\n"
"2. {'steps_info': [...], 'labware_info': [...]}\n"
"3. {'steps': [...], 'labware': [...]}"
"不支持的 JSON 格式。请使用标准格式:\n"
'{"workflow": [{"action": "...", "action_args": {...}}, ...], '
'"reagent": {"name": {"slot": int, "well": [...], "labware": "..."}, ...}}'
)
# 提取数据
workflow = json_data["workflow"]
reagent = json_data["reagent"]
# 规范化步骤数据
protocol_steps = normalize_workflow_steps(workflow)
# reagent 已经是字典格式,直接使用
labware_info = reagent
# 构建工作流图
graph = build_protocol_graph(
labware_info=labware_info,
@@ -317,7 +281,7 @@ def convert_from_json(
def convert_json_to_node_link(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
workstation_name: str = DEFAULT_WORKSTATION,
) -> Dict[str, Any]:
"""
将 JSON 数据转换为 node-link 格式的字典
@@ -335,7 +299,7 @@ def convert_json_to_node_link(
def convert_json_to_workflow_list(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
workstation_name: str = DEFAULT_WORKSTATION,
) -> List[Dict[str, Any]]:
"""
将 JSON 数据转换为工作流列表格式
@@ -349,8 +313,3 @@ def convert_json_to_workflow_list(
"""
graph = convert_from_json(data, workstation_name)
return graph.to_dict()
# 为了向后兼容,保留下划线前缀的别名
_normalize_steps = normalize_steps
_normalize_labware = normalize_labware

View File

@@ -0,0 +1,356 @@
"""
JSON 工作流转换模块
提供从多种 JSON 格式转换为统一工作流格式的功能。
支持的格式:
1. workflow/reagent 格式
2. steps_info/labware_info 格式
"""
import json
from os import PathLike
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from unilabos.workflow.common import WorkflowGraph, build_protocol_graph
from unilabos.registry.registry import lab_registry
def get_action_handles(resource_name: str, template_name: str) -> Dict[str, List[str]]:
"""
从 registry 获取指定设备和动作的 handles 配置
Args:
resource_name: 设备资源名称,如 "liquid_handler.prcxi"
template_name: 动作模板名称,如 "transfer_liquid"
Returns:
包含 source 和 target handler_keys 的字典:
{"source": ["sources_out", "targets_out", ...], "target": ["sources", "targets", ...]}
"""
result = {"source": [], "target": []}
device_info = lab_registry.device_type_registry.get(resource_name, {})
if not device_info:
return result
action_mappings = device_info.get("class", {}).get("action_value_mappings", {})
action_config = action_mappings.get(template_name, {})
handles = action_config.get("handles", {})
if isinstance(handles, dict):
# 处理 input handles (作为 target)
for handle in handles.get("input", []):
handler_key = handle.get("handler_key", "")
if handler_key:
result["source"].append(handler_key)
# 处理 output handles (作为 source)
for handle in handles.get("output", []):
handler_key = handle.get("handler_key", "")
if handler_key:
result["target"].append(handler_key)
return result
def validate_workflow_handles(graph: WorkflowGraph) -> Tuple[bool, List[str]]:
"""
校验工作流图中所有边的句柄配置是否正确
Args:
graph: 工作流图对象
Returns:
(is_valid, errors): 是否有效,错误信息列表
"""
errors = []
nodes = graph.nodes
for edge in graph.edges:
left_uuid = edge.get("source")
right_uuid = edge.get("target")
# target_handle_key是target, right的输入节点入节点
# source_handle_key是source, left的输出节点出节点
right_source_conn_key = edge.get("target_handle_key", "")
left_target_conn_key = edge.get("source_handle_key", "")
# 获取源节点和目标节点信息
left_node = nodes.get(left_uuid, {})
right_node = nodes.get(right_uuid, {})
left_res_name = left_node.get("resource_name", "")
left_template_name = left_node.get("template_name", "")
right_res_name = right_node.get("resource_name", "")
right_template_name = right_node.get("template_name", "")
# 获取源节点的 output handles
left_node_handles = get_action_handles(left_res_name, left_template_name)
target_valid_keys = left_node_handles.get("target", [])
target_valid_keys.append("ready")
# 获取目标节点的 input handles
right_node_handles = get_action_handles(right_res_name, right_template_name)
source_valid_keys = right_node_handles.get("source", [])
source_valid_keys.append("ready")
# 如果节点配置了 output handles则 source_port 必须有效
if not right_source_conn_key:
node_name = left_node.get("name", left_uuid[:8])
errors.append(f"源节点 '{node_name}' 的 source_handle_key 为空," f"应设置为: {source_valid_keys}")
elif right_source_conn_key not in source_valid_keys:
node_name = left_node.get("name", left_uuid[:8])
errors.append(
f"源节点 '{node_name}' 的 source 端点 '{right_source_conn_key}' 不存在," f"支持的端点: {source_valid_keys}"
)
# 如果节点配置了 input handles则 target_port 必须有效
if not left_target_conn_key:
node_name = right_node.get("name", right_uuid[:8])
errors.append(f"目标节点 '{node_name}' 的 target_handle_key 为空," f"应设置为: {target_valid_keys}")
elif left_target_conn_key not in target_valid_keys:
node_name = right_node.get("name", right_uuid[:8])
errors.append(
f"目标节点 '{node_name}' 的 target 端点 '{left_target_conn_key}' 不存在,"
f"支持的端点: {target_valid_keys}"
)
return len(errors) == 0, errors
# action 到 resource_name 的映射
ACTION_RESOURCE_MAPPING: Dict[str, str] = {
# 生物实验操作
"transfer_liquid": "liquid_handler.prcxi",
"transfer": "liquid_handler.prcxi",
"incubation": "incubator.prcxi",
"move_labware": "labware_mover.prcxi",
"oscillation": "shaker.prcxi",
# 有机化学操作
"HeatChillToTemp": "heatchill.chemputer",
"StopHeatChill": "heatchill.chemputer",
"StartHeatChill": "heatchill.chemputer",
"HeatChill": "heatchill.chemputer",
"Dissolve": "stirrer.chemputer",
"Transfer": "liquid_handler.chemputer",
"Evaporate": "rotavap.chemputer",
"Recrystallize": "reactor.chemputer",
"Filter": "filter.chemputer",
"Dry": "dryer.chemputer",
"Add": "liquid_handler.chemputer",
}
def normalize_steps(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
将不同格式的步骤数据规范化为统一格式
支持的输入格式:
- action + parameters
- action + action_args
- operation + parameters
Args:
data: 原始步骤数据列表
Returns:
规范化后的步骤列表,格式为 [{"action": str, "parameters": dict, "description": str?, "step_number": int?}, ...]
"""
normalized = []
for idx, step in enumerate(data):
# 获取动作名称(支持 action 或 operation 字段)
action = step.get("action") or step.get("operation")
if not action:
continue
# 获取参数(支持 parameters 或 action_args 字段)
raw_params = step.get("parameters") or step.get("action_args") or {}
params = dict(raw_params)
# 规范化 source/target -> sources/targets
if "source" in raw_params and "sources" not in raw_params:
params["sources"] = raw_params["source"]
if "target" in raw_params and "targets" not in raw_params:
params["targets"] = raw_params["target"]
# 获取描述(支持 description 或 purpose 字段)
description = step.get("description") or step.get("purpose")
# 获取步骤编号(优先使用原始数据中的 step_number否则使用索引+1
step_number = step.get("step_number", idx + 1)
step_dict = {"action": action, "parameters": params, "step_number": step_number}
if description:
step_dict["description"] = description
normalized.append(step_dict)
return normalized
def normalize_labware(data: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
"""
将不同格式的 labware 数据规范化为统一的字典格式
支持的输入格式:
- reagent_name + material_name + positions
- name + labware + slot
Args:
data: 原始 labware 数据列表
Returns:
规范化后的 labware 字典,格式为 {name: {"slot": int, "labware": str, "well": list, "type": str, "role": str, "name": str}, ...}
"""
labware = {}
for item in data:
# 获取 key 名称(优先使用 reagent_name其次是 material_name 或 name
reagent_name = item.get("reagent_name")
key = reagent_name or item.get("material_name") or item.get("name")
if not key:
continue
key = str(key)
# 处理重复 key自动添加后缀
idx = 1
original_key = key
while key in labware:
idx += 1
key = f"{original_key}_{idx}"
labware[key] = {
"slot": item.get("positions") or item.get("slot"),
"labware": item.get("material_name") or item.get("labware"),
"well": item.get("well", []),
"type": item.get("type", "reagent"),
"role": item.get("role", ""),
"name": key,
}
return labware
def convert_from_json(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
validate: bool = True,
) -> WorkflowGraph:
"""
从 JSON 数据或文件转换为 WorkflowGraph
支持的 JSON 格式:
1. {"workflow": [...], "reagent": {...}} - 直接格式
2. {"steps_info": [...], "labware_info": [...]} - 需要规范化的格式
Args:
data: JSON 文件路径、字典数据、或 JSON 字符串
workstation_name: 工作站名称,默认 "PRCXi"
validate: 是否校验句柄配置,默认 True
Returns:
WorkflowGraph: 构建好的工作流图
Raises:
ValueError: 不支持的 JSON 格式 或 句柄校验失败
FileNotFoundError: 文件不存在
json.JSONDecodeError: JSON 解析失败
"""
# 处理输入数据
if isinstance(data, (str, PathLike)):
path = Path(data)
if path.exists():
with path.open("r", encoding="utf-8") as fp:
json_data = json.load(fp)
elif isinstance(data, str):
# 尝试作为 JSON 字符串解析
json_data = json.loads(data)
else:
raise FileNotFoundError(f"文件不存在: {data}")
elif isinstance(data, dict):
json_data = data
else:
raise TypeError(f"不支持的数据类型: {type(data)}")
# 根据格式解析数据
if "workflow" in json_data and "reagent" in json_data:
# 格式1: workflow/reagent已经是规范格式
protocol_steps = json_data["workflow"]
labware_info = json_data["reagent"]
elif "steps_info" in json_data and "labware_info" in json_data:
# 格式2: steps_info/labware_info需要规范化
protocol_steps = normalize_steps(json_data["steps_info"])
labware_info = normalize_labware(json_data["labware_info"])
elif "steps" in json_data and "labware" in json_data:
# 格式3: steps/labware另一种常见格式
protocol_steps = normalize_steps(json_data["steps"])
if isinstance(json_data["labware"], list):
labware_info = normalize_labware(json_data["labware"])
else:
labware_info = json_data["labware"]
else:
raise ValueError(
"不支持的 JSON 格式。支持的格式:\n"
"1. {'workflow': [...], 'reagent': {...}}\n"
"2. {'steps_info': [...], 'labware_info': [...]}\n"
"3. {'steps': [...], 'labware': [...]}"
)
# 构建工作流图
graph = build_protocol_graph(
labware_info=labware_info,
protocol_steps=protocol_steps,
workstation_name=workstation_name,
action_resource_mapping=ACTION_RESOURCE_MAPPING,
)
# 校验句柄配置
if validate:
is_valid, errors = validate_workflow_handles(graph)
if not is_valid:
import warnings
for error in errors:
warnings.warn(f"句柄校验警告: {error}")
return graph
def convert_json_to_node_link(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
) -> Dict[str, Any]:
"""
将 JSON 数据转换为 node-link 格式的字典
Args:
data: JSON 文件路径、字典数据、或 JSON 字符串
workstation_name: 工作站名称,默认 "PRCXi"
Returns:
Dict: node-link 格式的工作流数据
"""
graph = convert_from_json(data, workstation_name)
return graph.to_node_link_dict()
def convert_json_to_workflow_list(
data: Union[str, PathLike, Dict[str, Any]],
workstation_name: str = "PRCXi",
) -> List[Dict[str, Any]]:
"""
将 JSON 数据转换为工作流列表格式
Args:
data: JSON 文件路径、字典数据、或 JSON 字符串
workstation_name: 工作站名称,默认 "PRCXi"
Returns:
List: 工作流节点列表
"""
graph = convert_from_json(data, workstation_name)
return graph.to_dict()
# 为了向后兼容,保留下划线前缀的别名
_normalize_steps = normalize_steps
_normalize_labware = normalize_labware