Compare commits

..

2 Commits

Author SHA1 Message Date
q434343
5c9c8a4ee9 Merge branch 'prcix9320' into sjs_middle_school 2026-03-31 18:48:20 +08:00
q434343
a48985720c 添加run_protocol参数 2026-03-31 16:11:11 +08:00
11 changed files with 1765 additions and 332 deletions

539
test/devices/test_prcxi.py Normal file
View File

@@ -0,0 +1,539 @@
import pytest
import json
import os
import asyncio
import collections
from typing import List, Dict, Any
from pylabrobot.resources import Coordinate
from pylabrobot.resources.opentrons.tip_racks import opentrons_96_tiprack_300ul, opentrons_96_tiprack_10ul
from pylabrobot.resources.opentrons.plates import corning_96_wellplate_360ul_flat, nest_96_wellplate_2ml_deep
from unilabos.devices.liquid_handling.prcxi.prcxi import (
PRCXI9300Deck,
PRCXI9300Container,
PRCXI9300Trash,
PRCXI9300Handler,
PRCXI9300Backend,
DefaultLayout,
Material,
WorkTablets,
MatrixInfo
)
@pytest.fixture
def prcxi_materials() -> Dict[str, Any]:
"""加载 PRCXI 物料数据"""
print("加载 PRCXI 物料数据...")
material_path = os.path.join(os.path.dirname(__file__), "..", "..", "unilabos", "devices", "liquid_handling", "prcxi", "prcxi_material.json")
with open(material_path, "r", encoding="utf-8") as f:
data = json.load(f)
print(f"加载了 {len(data)} 条物料数据")
return data
@pytest.fixture
def prcxi_9300_deck() -> PRCXI9300Deck:
"""创建 PRCXI 9300 工作台"""
return PRCXI9300Deck(name="PRCXI_Deck_9300", size_x=100, size_y=100, size_z=100, model="9300")
@pytest.fixture
def prcxi_9320_deck() -> PRCXI9300Deck:
"""创建 PRCXI 9320 工作台"""
return PRCXI9300Deck(name="PRCXI_Deck_9320", size_x=100, size_y=100, size_z=100, model="9320")
@pytest.fixture
def prcxi_9300_handler(prcxi_9300_deck) -> PRCXI9300Handler:
"""创建 PRCXI 9300 处理器(模拟模式)"""
return PRCXI9300Handler(
deck=prcxi_9300_deck,
host="192.168.1.201",
port=9999,
timeout=10.0,
channel_num=8,
axis="Left",
setup=False,
debug=True,
simulator=True,
matrix_id="test-matrix-9300"
)
@pytest.fixture
def prcxi_9320_handler(prcxi_9320_deck) -> PRCXI9300Handler:
"""创建 PRCXI 9320 处理器(模拟模式)"""
return PRCXI9300Handler(
deck=prcxi_9320_deck,
host="192.168.1.201",
port=9999,
timeout=10.0,
channel_num=1,
axis="Right",
setup=False,
debug=True,
simulator=True,
matrix_id="test-matrix-9320",
is_9320=True
)
@pytest.fixture
def tip_rack_300ul(prcxi_materials) -> PRCXI9300Container:
"""创建 300μL 枪头盒"""
tip_rack = PRCXI9300Container(
name="tip_rack_300ul",
size_x=50,
size_y=50,
size_z=10,
category="tip_rack",
ordering=collections.OrderedDict()
)
tip_rack.load_state({
"Material": {
"uuid": prcxi_materials["300μL Tip头"]["uuid"],
"Code": "ZX-001-300",
"Name": "300μL Tip头"
}
})
return tip_rack
@pytest.fixture
def tip_rack_10ul(prcxi_materials) -> PRCXI9300Container:
"""创建 10μL 枪头盒"""
tip_rack = PRCXI9300Container(
name="tip_rack_10ul",
size_x=50,
size_y=50,
size_z=10,
category="tip_rack",
ordering=collections.OrderedDict()
)
tip_rack.load_state({
"Material": {
"uuid": prcxi_materials["10μL加长 Tip头"]["uuid"],
"Code": "ZX-001-10+",
"Name": "10μL加长 Tip头"
}
})
return tip_rack
@pytest.fixture
def well_plate_96(prcxi_materials) -> PRCXI9300Container:
"""创建 96 孔板"""
plate = PRCXI9300Container(
name="well_plate_96",
size_x=50,
size_y=50,
size_z=10,
category="plate",
ordering=collections.OrderedDict()
)
plate.load_state({
"Material": {
"uuid": prcxi_materials["96深孔板"]["uuid"],
"Code": "ZX-019-2.2",
"Name": "96深孔板"
}
})
return plate
@pytest.fixture
def deep_well_plate(prcxi_materials) -> PRCXI9300Container:
"""创建深孔板"""
plate = PRCXI9300Container(
name="deep_well_plate",
size_x=50,
size_y=50,
size_z=10,
category="plate",
ordering=collections.OrderedDict()
)
plate.load_state({
"Material": {
"uuid": prcxi_materials["96深孔板"]["uuid"],
"Code": "ZX-019-2.2",
"Name": "96深孔板"
}
})
return plate
@pytest.fixture
def trash_container(prcxi_materials) -> PRCXI9300Trash:
"""创建垃圾桶"""
trash = PRCXI9300Trash(name="trash", size_x=50, size_y=50, size_z=10, category="trash")
trash.load_state({
"Material": {
"uuid": prcxi_materials["废弃槽"]["uuid"]
}
})
return trash
@pytest.fixture
def default_layout_9300() -> DefaultLayout:
"""创建 PRCXI 9300 默认布局"""
return DefaultLayout("PRCXI9300")
@pytest.fixture
def default_layout_9320() -> DefaultLayout:
"""创建 PRCXI 9320 默认布局"""
return DefaultLayout("PRCXI9320")
class TestPRCXIDeckSetup:
"""测试 PRCXI 工作台设置功能"""
def test_prcxi_9300_deck_creation(self, prcxi_9300_deck):
"""测试 PRCXI 9300 工作台创建"""
assert prcxi_9300_deck.name == "PRCXI_Deck_9300"
assert len(prcxi_9300_deck.sites) == 6
assert prcxi_9300_deck._size_x == 100
assert prcxi_9300_deck._size_y == 100
assert prcxi_9300_deck._size_z == 100
def test_prcxi_9320_deck_creation(self, prcxi_9320_deck):
"""测试 PRCXI 9320 工作台创建"""
assert prcxi_9320_deck.name == "PRCXI_Deck_9320"
assert len(prcxi_9320_deck.sites) == 16
assert prcxi_9320_deck._size_x == 100
assert prcxi_9320_deck._size_y == 100
assert prcxi_9320_deck._size_z == 100
def test_container_assignment(self, prcxi_9300_deck, tip_rack_300ul, well_plate_96, trash_container):
"""测试容器分配到工作台"""
# 分配枪头盒
prcxi_9300_deck.assign_child_resource(tip_rack_300ul, location=Coordinate(0, 0, 0))
assert tip_rack_300ul in prcxi_9300_deck.children
# 分配孔板
prcxi_9300_deck.assign_child_resource(well_plate_96, location=Coordinate(0, 0, 0))
assert well_plate_96 in prcxi_9300_deck.children
# 分配垃圾桶
prcxi_9300_deck.assign_child_resource(trash_container, location=Coordinate(0, 0, 0))
assert trash_container in prcxi_9300_deck.children
def test_container_material_loading(self, tip_rack_300ul, well_plate_96, prcxi_materials):
"""测试容器物料信息加载"""
# 测试枪头盒物料信息
tip_material = tip_rack_300ul._unilabos_state["Material"]
assert tip_material["uuid"] == prcxi_materials["300μL Tip头"]["uuid"]
assert tip_material["Name"] == "300μL Tip头"
# 测试孔板物料信息
plate_material = well_plate_96._unilabos_state["Material"]
assert plate_material["uuid"] == prcxi_materials["96深孔板"]["uuid"]
assert plate_material["Name"] == "96深孔板"
class TestPRCXISingleStepOperations:
"""测试 PRCXI 单步操作功能"""
@pytest.mark.asyncio
async def test_pick_up_tips_single_channel(self, prcxi_9320_handler, prcxi_9320_deck, tip_rack_10ul):
"""测试单通道拾取枪头"""
# 将枪头盒添加到工作台
prcxi_9320_deck.assign_child_resource(tip_rack_10ul, location=Coordinate(0, 0, 0))
# 初始化处理器
await prcxi_9320_handler.setup()
# 设置枪头盒
prcxi_9320_handler.set_tiprack([tip_rack_10ul])
# 创建模拟的枪头位置
from pylabrobot.resources import TipSpot, Tip
tip = Tip(has_filter=False, total_tip_length=10, maximal_volume=10, fitting_depth=5)
tip_spot = TipSpot("A1", size_x=1, size_y=1, size_z=1, make_tip=lambda: tip)
tip_rack_10ul.assign_child_resource(tip_spot, location=Coordinate(0, 0, 0))
# 直接测试后端方法
from pylabrobot.liquid_handling import Pickup
pickup = Pickup(resource=tip_spot, offset=None, tip=tip)
await prcxi_9320_handler._unilabos_backend.pick_up_tips([pickup], [0])
# 验证步骤已添加到待办列表
assert len(prcxi_9320_handler._unilabos_backend.steps_todo_list) == 1
step = prcxi_9320_handler._unilabos_backend.steps_todo_list[0]
assert step["Function"] == "Load"
@pytest.mark.asyncio
async def test_pick_up_tips_multi_channel(self, prcxi_9300_handler, tip_rack_300ul):
"""测试多通道拾取枪头"""
# 设置枪头盒
prcxi_9300_handler.set_tiprack([tip_rack_300ul])
# 拾取8个枪头
tip_spots = tip_rack_300ul.children[:8]
await prcxi_9300_handler.pick_up_tips(tip_spots, [0, 1, 2, 3, 4, 5, 6, 7])
# 验证步骤已添加到待办列表
assert len(prcxi_9300_handler._unilabos_backend.steps_todo_list) == 1
step = prcxi_9300_handler._unilabos_backend.steps_todo_list[0]
assert step["Function"] == "Load"
@pytest.mark.asyncio
async def test_aspirate_single_channel(self, prcxi_9320_handler, well_plate_96):
"""测试单通道吸取液体"""
# 设置液体
well = well_plate_96.get_item("A1")
prcxi_9320_handler.set_liquid([well], ["water"], [50])
# 吸取液体
await prcxi_9320_handler.aspirate([well], [50], [0])
# 验证步骤已添加到待办列表
assert len(prcxi_9320_handler._unilabos_backend.steps_todo_list) == 1
step = prcxi_9320_handler._unilabos_backend.steps_todo_list[0]
assert step["Function"] == "Imbibing"
assert step["DosageNum"] == 50
@pytest.mark.asyncio
async def test_dispense_single_channel(self, prcxi_9320_handler, well_plate_96):
"""测试单通道分配液体"""
# 分配液体
well = well_plate_96.get_item("A1")
await prcxi_9320_handler.dispense([well], [25], [0])
# 验证步骤已添加到待办列表
assert len(prcxi_9320_handler._unilabos_backend.steps_todo_list) == 1
step = prcxi_9320_handler._unilabos_backend.steps_todo_list[0]
assert step["Function"] == "Tapping"
assert step["DosageNum"] == 25
@pytest.mark.asyncio
async def test_mix_single_channel(self, prcxi_9320_handler, well_plate_96):
"""测试单通道混合液体"""
# 混合液体
well = well_plate_96.get_item("A1")
await prcxi_9320_handler.mix([well], mix_time=3, mix_vol=50)
# 验证步骤已添加到待办列表
assert len(prcxi_9320_handler._unilabos_backend.steps_todo_list) == 1
step = prcxi_9320_handler._unilabos_backend.steps_todo_list[0]
assert step["Function"] == "Blending"
assert step["BlendingTimes"] == 3
assert step["DosageNum"] == 50
@pytest.mark.asyncio
async def test_drop_tips_to_trash(self, prcxi_9320_handler, trash_container):
"""测试丢弃枪头到垃圾桶"""
# 丢弃枪头
await prcxi_9320_handler.drop_tips([trash_container], [0])
# 验证步骤已添加到待办列表
assert len(prcxi_9320_handler._unilabos_backend.steps_todo_list) == 1
step = prcxi_9320_handler._unilabos_backend.steps_todo_list[0]
assert step["Function"] == "UnLoad"
@pytest.mark.asyncio
async def test_discard_tips(self, prcxi_9320_handler):
"""测试丢弃枪头"""
# 丢弃枪头
await prcxi_9320_handler.discard_tips([0])
# 验证步骤已添加到待办列表
assert len(prcxi_9320_handler._unilabos_backend.steps_todo_list) == 1
step = prcxi_9320_handler._unilabos_backend.steps_todo_list[0]
assert step["Function"] == "UnLoad"
@pytest.mark.asyncio
async def test_liquid_transfer_workflow(self, prcxi_9320_handler, tip_rack_10ul, well_plate_96):
"""测试完整的液体转移工作流程"""
# 设置枪头盒和液体
prcxi_9320_handler.set_tiprack([tip_rack_10ul])
source_well = well_plate_96.get_item("A1")
target_well = well_plate_96.get_item("B1")
prcxi_9320_handler.set_liquid([source_well], ["water"], [100])
# 创建协议
await prcxi_9320_handler.create_protocol(protocol_name="Test Transfer Protocol")
# 执行转移流程
tip_spot = tip_rack_10ul.get_item("A1")
await prcxi_9320_handler.pick_up_tips([tip_spot], [0])
await prcxi_9320_handler.aspirate([source_well], [50], [0])
await prcxi_9320_handler.dispense([target_well], [50], [0])
await prcxi_9320_handler.discard_tips([0])
# 验证所有步骤都已添加
assert len(prcxi_9320_handler._unilabos_backend.steps_todo_list) == 4
functions = [step["Function"] for step in prcxi_9320_handler._unilabos_backend.steps_todo_list]
assert functions == ["Load", "Imbibing", "Tapping", "UnLoad"]
class TestPRCXILayoutRecommendation:
"""测试 PRCXI 板位推荐功能"""
def test_9300_layout_creation(self, default_layout_9300):
"""测试 PRCXI 9300 布局创建"""
layout_info = default_layout_9300.get_layout()
assert layout_info["rows"] == 2
assert layout_info["columns"] == 3
assert len(layout_info["layout"]) == 6
assert layout_info["trash_slot"] == 6
assert "waste_liquid_slot" not in layout_info
def test_9320_layout_creation(self, default_layout_9320):
"""测试 PRCXI 9320 布局创建"""
layout_info = default_layout_9320.get_layout()
assert layout_info["rows"] == 4
assert layout_info["columns"] == 4
assert len(layout_info["layout"]) == 16
assert layout_info["trash_slot"] == 16
assert layout_info["waste_liquid_slot"] == 12
def test_layout_recommendation_9320(self, default_layout_9320, prcxi_materials):
"""测试 PRCXI 9320 板位推荐功能"""
# 添加物料信息
default_layout_9320.add_lab_resource(prcxi_materials)
# 推荐布局
needs = [
("reagent_1", "96 细胞培养皿", 3),
("reagent_2", "12道储液槽", 1),
("reagent_3", "200μL Tip头", 7),
("reagent_4", "10μL加长 Tip头", 1),
]
matrix_layout, layout_list = default_layout_9320.recommend_layout(needs)
# 验证返回结果
assert "MatrixId" in matrix_layout
assert "MatrixName" in matrix_layout
assert "MatrixCount" in matrix_layout
assert "WorkTablets" in matrix_layout
assert len(layout_list) == 12 # 3+1+7+1 = 12个位置
# 验证推荐的位置不包含预留位置
reserved_positions = {12, 16}
recommended_positions = [item["positions"] for item in layout_list]
for pos in recommended_positions:
assert pos not in reserved_positions
def test_layout_recommendation_insufficient_space(self, default_layout_9320, prcxi_materials):
"""测试板位推荐空间不足的情况"""
# 添加物料信息
default_layout_9320.add_lab_resource(prcxi_materials)
# 尝试推荐超过可用空间的布局
needs = [
("reagent_1", "96 细胞培养皿", 15), # 需要15个位置但只有14个可用
]
with pytest.raises(ValueError, match="需要 .* 个位置,但只有 .* 个可用位置"):
default_layout_9320.recommend_layout(needs)
def test_layout_recommendation_material_not_found(self, default_layout_9320, prcxi_materials):
"""测试板位推荐物料不存在的情况"""
# 添加物料信息
default_layout_9320.add_lab_resource(prcxi_materials)
# 尝试推荐不存在的物料
needs = [
("reagent_1", "不存在的物料", 1),
]
with pytest.raises(ValueError, match="Material .* not found in lab resources"):
default_layout_9320.recommend_layout(needs)
class TestPRCXIBackendOperations:
"""测试 PRCXI 后端操作功能"""
def test_backend_initialization(self, prcxi_9300_handler):
"""测试后端初始化"""
backend = prcxi_9300_handler._unilabos_backend
assert isinstance(backend, PRCXI9300Backend)
assert backend._num_channels == 8
assert backend.debug is True
def test_protocol_creation(self, prcxi_9300_handler):
"""测试协议创建"""
backend = prcxi_9300_handler._unilabos_backend
backend.create_protocol("Test Protocol")
assert backend.protocol_name == "Test Protocol"
assert len(backend.steps_todo_list) == 0
def test_channel_validation(self):
"""测试通道验证"""
# 测试正确的8通道配置
valid_channels = [0, 1, 2, 3, 4, 5, 6, 7]
result = PRCXI9300Backend.check_channels(valid_channels)
assert result == valid_channels
# 测试错误的通道配置
invalid_channels = [0, 1, 2, 3]
result = PRCXI9300Backend.check_channels(invalid_channels)
assert result == [0, 1, 2, 3, 4, 5, 6, 7]
def test_matrix_info_creation(self, prcxi_9300_handler):
"""测试矩阵信息创建"""
backend = prcxi_9300_handler._unilabos_backend
backend.create_protocol("Test Protocol")
# 模拟运行协议时的矩阵信息创建
run_time = 1234567890
matrix_info = MatrixInfo(
MatrixId=f"{int(run_time)}",
MatrixName=f"protocol_{run_time}",
MatrixCount=len(backend.tablets_info),
WorkTablets=backend.tablets_info,
)
assert matrix_info["MatrixId"] == str(int(run_time))
assert matrix_info["MatrixName"] == f"protocol_{run_time}"
assert "WorkTablets" in matrix_info
class TestPRCXIContainerOperations:
"""测试 PRCXI 容器操作功能"""
def test_container_serialization(self, tip_rack_300ul):
"""测试容器序列化"""
serialized = tip_rack_300ul.serialize_state()
assert "Material" in serialized
assert serialized["Material"]["Name"] == "300μL Tip头"
def test_container_deserialization(self, tip_rack_300ul):
"""测试容器反序列化"""
# 序列化
serialized = tip_rack_300ul.serialize_state()
# 创建新容器并反序列化
new_tip_rack = PRCXI9300Container(
name="new_tip_rack",
size_x=50,
size_y=50,
size_z=10,
category="tip_rack",
ordering=collections.OrderedDict()
)
new_tip_rack.load_state(serialized)
assert new_tip_rack._unilabos_state["Material"]["Name"] == "300μL Tip头"
def test_trash_container_creation(self, prcxi_materials):
"""测试垃圾桶容器创建"""
trash = PRCXI9300Trash(name="trash", size_x=50, size_y=50, size_z=10, category="trash")
trash.load_state({
"Material": {
"uuid": prcxi_materials["废弃槽"]["uuid"]
}
})
assert trash.name == "trash"
assert trash._unilabos_state["Material"]["uuid"] == prcxi_materials["废弃槽"]["uuid"]
if __name__ == "__main__":
# 运行测试
pytest.main([__file__, "-v"])

View File

@@ -57,7 +57,8 @@ from unilabos.devices.liquid_handling.liquid_handler_abstract import (
) )
from unilabos.registry.placeholder_type import ResourceSlot from unilabos.registry.placeholder_type import ResourceSlot
from unilabos.resources.itemized_carrier import ItemizedCarrier from unilabos.resources.itemized_carrier import ItemizedCarrier
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode from unilabos.resources.resource_tracker import ResourceTreeSet
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
class PRCXIError(RuntimeError): class PRCXIError(RuntimeError):
@@ -109,11 +110,19 @@ class PRCXI9300Deck(Deck):
该类定义了 PRCXI 9300 的工作台布局和槽位信息。 该类定义了 PRCXI 9300 的工作台布局和槽位信息。
""" """
# T1-T16 默认位置 (4列×4行) _9320_SITE_POSITIONS = [((i%4)*137.5+5, (3-int(i/4))*96+13, 0) for i in range(0, 16)]
_DEFAULT_SITE_POSITIONS = [((i%4)*137.5+5, (3-int(i/4))*96+13, 0) for i in range(0, 16)]
# 9300: 3列×2行 = 6 slots间距与9320相同X: 138mm, Y: 96mm
_9300_SITE_POSITIONS = [
(0, 96, 0), (138, 96, 0), (276, 96, 0), # T1-T3 (第1行, 上)
(0, 0, 0), (138, 0, 0), (276, 0, 0), # T4-T6 (第2行, 下)
]
# 向后兼容别名
_DEFAULT_SITE_POSITIONS = _9320_SITE_POSITIONS
_DEFAULT_SITE_SIZE = {"width": 128.0, "height": 86, "depth": 0} _DEFAULT_SITE_SIZE = {"width": 128.0, "height": 86, "depth": 0}
_DEFAULT_CONTENT_TYPE = ["plate", "tip_rack", "plates", "tip_racks", "tube_rack", "adaptor"] _DEFAULT_CONTENT_TYPE = ["plate", "tip_rack", "plates", "tip_racks", "tube_rack", "adaptor", "plateadapter", "module", "trash"]
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,
sites: Optional[List[Dict[str, Any]]] = None, **kwargs): sites: Optional[List[Dict[str, Any]]] = None, **kwargs):
@@ -190,6 +199,7 @@ class PRCXI9300Deck(Deck):
def serialize(self) -> dict: def serialize(self) -> dict:
data = super().serialize() data = super().serialize()
data["model"] = self.model
sites_out = [] sites_out = []
for i, site in enumerate(self.sites): for i, site in enumerate(self.sites):
occupied = self._get_site_resource(i) occupied = self._get_site_resource(i)
@@ -276,30 +286,9 @@ class PRCXI9300Plate(Plate):
elif value is None: elif value is None:
ordering_param = ordering ordering_param = ordering
else: else:
# ordering 的值是对象(可能是 Well 对象),检查是否有有效的 location # ordering 的值已经是对象,可以直接使用
# 如果是反序列化过程Well 对象可能没有正确的 location需要让 Plate 重新创建
sample_value = next(iter(ordering.values()), None)
if sample_value is not None and hasattr(sample_value, 'location'):
# 如果是 Well 对象但 location 为 None说明是反序列化过程
# 让 Plate 自己创建 Well 对象
if sample_value.location is None:
items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else:
# Well 对象有有效的 location可以直接使用
items = ordering items = ordering
ordering_param = None ordering_param = None
elif sample_value is None:
# ordering 的值都是 None让 Plate 自己创建 Well 对象
items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else:
# 其他情况,直接使用
items = ordering
ordering_param = None
else:
items = None
ordering_param = collections.OrderedDict() # 提供空的 ordering
# 根据情况传递不同的参数 # 根据情况传递不同的参数
if items is not None: if items is not None:
@@ -381,16 +370,9 @@ class PRCXI9300TipRack(TipRack):
items = None items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys()) ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else: else:
# ordering 的值已经是对象,需要过滤掉 None 值 # ordering 的值已经是对象,可以直接使用
# 只保留有效的对象,用于 ordered_items 参数 items = ordering
valid_items = {k: v for k, v in ordering.items() if v is not None}
if valid_items:
items = valid_items
ordering_param = None ordering_param = None
else:
# 如果没有有效对象,使用 ordering 参数
items = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
else: else:
items = None items = None
ordering_param = None ordering_param = None
@@ -449,10 +431,16 @@ class PRCXI9300Trash(Trash):
该类定义了 PRCXI 9300 的工作台布局和槽位信息。 该类定义了 PRCXI 9300 的工作台布局和槽位信息。
""" """
def __init__(self, name: str, size_x: float, size_y: float, size_z: float, def __init__(
category: str = "plate", self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str = "trash",
material_info: Optional[Dict[str, Any]] = None, material_info: Optional[Dict[str, Any]] = None,
**kwargs): **kwargs,
):
if name != "trash": if name != "trash":
print(f"Warning: PRCXI9300Trash usually expects name='trash' for backend logic, but got '{name}'.") print(f"Warning: PRCXI9300Trash usually expects name='trash' for backend logic, but got '{name}'.")
@@ -521,24 +509,19 @@ class PRCXI9300TubeRack(TubeRack):
items_to_pass = ordered_items items_to_pass = ordered_items
ordering_param = None ordering_param = None
elif ordering is not None: elif ordering is not None:
# 检查 ordering 中的值是否是字符串(从 JSON 反序列化时的情况) # 检查 ordering 中的值类型来决定如何处理:
if ordering and isinstance(next(iter(ordering.values()), None), str): # - 字符串值(从 JSON 反序列化): 只用键创建 ordering_param
# ordering 的值是字符串,这种情况下我们让 TubeRack 使用默认行为 # - None 值(从第二次往返序列化): 同样只用键创建 ordering_param
# 不在初始化时创建 items而是在 deserialize 后处理 # - 对象值(已经是实际的 Resource 对象): 直接作为 ordered_items 使用
first_val = next(iter(ordering.values()), None) if ordering else None
if not ordering or first_val is None or isinstance(first_val, str):
# ordering 的值是字符串或 None只使用键位置信息创建新的 OrderedDict
# 传递 ordering 参数而不是 ordered_items让 TubeRack 自己创建 Tube 对象
items_to_pass = None items_to_pass = None
ordering_param = collections.OrderedDict((k, None) for k in ordering.keys()) # 提供空的 ordering 来满足要求 ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
# 保存 ordering 信息以便后续处理
self._temp_ordering = ordering
else: else:
# ordering 的值已经是对象,需要过滤掉 None 值 # ordering 的值已经是对象,可以直接使用
# 只保留有效的对象,用于 ordered_items 参数 items_to_pass = ordering
valid_items = {k: v for k, v in ordering.items() if v is not None}
if valid_items:
items_to_pass = valid_items
ordering_param = None
else:
# 如果没有有效对象,创建空的 ordered_items
items_to_pass = {}
ordering_param = None ordering_param = None
elif items is not None: elif items is not None:
# 兼容旧的 items 参数 # 兼容旧的 items 参数
@@ -561,29 +544,6 @@ class PRCXI9300TubeRack(TubeRack):
if material_info: if material_info:
self._unilabos_state["Material"] = material_info self._unilabos_state["Material"] = material_info
# 如果有临时 ordering 信息,在初始化完成后处理
if hasattr(self, '_temp_ordering') and self._temp_ordering:
self._process_temp_ordering()
def _process_temp_ordering(self):
"""处理临时的 ordering 信息,创建相应的 Tube 对象"""
from pylabrobot.resources import Tube, Coordinate
for location, item_type in self._temp_ordering.items():
if item_type == 'Tube' or item_type == 'tube':
# 为每个位置创建 Tube 对象
tube = Tube(name=f"{self.name}_{location}", size_x=10, size_y=10, size_z=50, max_volume=2000.0)
# 使用 assign_child_resource 添加到 rack 中
self.assign_child_resource(tube, location=Coordinate(0, 0, 0))
# 清理临时数据
del self._temp_ordering
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]]: def serialize_state(self) -> Dict[str, Dict[str, Any]]:
try: try:
data = super().serialize_state() data = super().serialize_state()
@@ -610,11 +570,20 @@ class PRCXI9300TubeRack(TubeRack):
data.update(safe_state) data.update(safe_state)
return data return data
class PRCXI9300PlateAdapterSite(ItemizedCarrier):
class PRCXI9300ModuleSite(ItemizedCarrier):
"""
PRCXI 功能模块的基础站点类(加热/冷却/震荡/磁吸等)。
- 继承 ItemizedCarrier可被拖放到 Deck 槽位上
- 顶面有一个 ResourceHolder 站点,可吸附板类资源(叠放)
- content_type 包含 "plateadapter" 以支持适配器叠放
- 支持 material_info 注入
"""
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,
material_info: Optional[Dict[str, Any]] = None, **kwargs): material_info: Optional[Dict[str, Any]] = None, **kwargs):
# 处理 sites 参数的不同格式
sites = create_homogeneous_resources( sites = create_homogeneous_resources(
klass=ResourceHolder, klass=ResourceHolder,
locations=[Coordinate(0, 0, 0)], locations=[Coordinate(0, 0, 0)],
@@ -624,48 +593,43 @@ class PRCXI9300PlateAdapterSite(ItemizedCarrier):
name_prefix=name, name_prefix=name,
)[0] )[0]
# 确保不传递重复的参数
kwargs.pop('layout', None) kwargs.pop('layout', None)
sites_in = kwargs.pop('sites', None) sites_in = kwargs.pop('sites', None)
# 创建默认的sites字典
sites_dict = {name: sites} sites_dict = {name: sites}
# 优先从 sites_in 读取 'content_type',否则使用默认值
content_type = [ content_type = [
"plate", "plate",
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter",
] ]
# 如果提供了sites参数则用sites_in中的值替换sites_dict中对应的元素
if sites_in is not None and isinstance(sites_in, dict): if sites_in is not None and isinstance(sites_in, dict):
for site_key, site_value in sites_in.items(): for site_key, site_value in sites_in.items():
if site_key in sites_dict: if site_key in sites_dict:
sites_dict[site_key] = site_value sites_dict[site_key] = site_value
super().__init__(name, size_x, size_y, size_z, super().__init__(
name, size_x, size_y, size_z,
sites=sites_dict, sites=sites_dict,
num_items_x=kwargs.pop('num_items_x', 1), num_items_x=kwargs.pop('num_items_x', 1),
num_items_y=kwargs.pop('num_items_y', 1), num_items_y=kwargs.pop('num_items_y', 1),
num_items_z=kwargs.pop('num_items_z', 1), num_items_z=kwargs.pop('num_items_z', 1),
content_type=content_type, content_type=content_type,
**kwargs) **kwargs,
)
self._unilabos_state = {} self._unilabos_state = {}
if material_info: if material_info:
self._unilabos_state["Material"] = material_info self._unilabos_state["Material"] = material_info
def assign_child_resource(self, resource, location=Coordinate(0, 0, 0), reassign=True, spot=None): def assign_child_resource(self, resource, location=Coordinate(0, 0, 0), reassign=True, spot=None):
"""重写 assign_child_resource 方法,对于适配器位置,不使用索引分配"""
# 直接调用 Resource 的 assign_child_resource避免 ItemizedCarrier 的索引逻辑
from pylabrobot.resources.resource import Resource from pylabrobot.resources.resource import Resource
Resource.assign_child_resource(self, resource, location=location, reassign=reassign) Resource.assign_child_resource(self, resource, location=location, reassign=reassign)
def unassign_child_resource(self, resource): def unassign_child_resource(self, resource):
"""重写 unassign_child_resource 方法,对于适配器位置,不使用 sites 列表"""
# 直接调用 Resource 的 unassign_child_resource避免 ItemizedCarrier 的 sites 逻辑
from pylabrobot.resources.resource import Resource from pylabrobot.resources.resource import Resource
Resource.unassign_child_resource(self, resource) Resource.unassign_child_resource(self, resource)
@@ -675,13 +639,10 @@ class PRCXI9300PlateAdapterSite(ItemizedCarrier):
except AttributeError: except AttributeError:
data = {} data = {}
# 包含 sites 配置信息,但避免序列化 ResourceHolder 对象
if hasattr(self, 'sites') and self.sites: if hasattr(self, 'sites') and self.sites:
# 只保存 sites 的基本信息,不保存 ResourceHolder 对象本身
sites_info = [] sites_info = []
for site in self.sites: for site in self.sites:
if hasattr(site, '__class__') and 'pylabrobot' in str(site.__class__.__module__): if hasattr(site, '__class__') and 'pylabrobot' in str(site.__class__.__module__):
# 对于 pylabrobot 对象,只保存基本信息
sites_info.append({ sites_info.append({
"__pylabrobot_object__": True, "__pylabrobot_object__": True,
"class": site.__class__.__name__, "class": site.__class__.__name__,
@@ -692,13 +653,23 @@ class PRCXI9300PlateAdapterSite(ItemizedCarrier):
sites_info.append(site) sites_info.append(site)
data['sites'] = sites_info data['sites'] = sites_info
if hasattr(self, "_unilabos_state") and self._unilabos_state:
safe_state: Dict[str, Any] = {}
for k, v in self._unilabos_state.items():
if k == "Material" and isinstance(v, dict):
safe_material: Dict[str, Any] = {}
for mk, mv in v.items():
if isinstance(mv, (str, int, float, bool, list, dict, type(None))):
safe_material[mk] = mv
safe_state[k] = safe_material
elif isinstance(v, (str, int, float, bool, list, dict, type(None))):
safe_state[k] = v
data.update(safe_state)
return data return data
def load_state(self, state: Dict[str, Any]) -> None: def load_state(self, state: Dict[str, Any]) -> None:
"""加载状态,包括 sites 配置信息"""
super().load_state(state) super().load_state(state)
# 从状态中恢复 sites 配置信息
if 'sites' in state: if 'sites' in state:
self.sites = [state['sites']] self.sites = [state['sites']]
@@ -793,7 +764,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
def __init__( def __init__(
self, self,
deck: Deck, deck: PRCXI9300Deck,
host: str, host: str,
port: int, port: int,
timeout: float, timeout: float,
@@ -828,12 +799,25 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
self.xy_coupling = xy_coupling self.xy_coupling = xy_coupling
self.left_2_claw = Coordinate(-130.2, 34, -134) self.left_2_claw = Coordinate(-130.2, 34, -134)
self.right_2_left = Coordinate(22,-1, 8) self.right_2_left = Coordinate(22,-1, 8)
tablets_info = {}
plate_positions = [] plate_positions = []
tablets_info = []
if is_9320 is None:
is_9320 = getattr(deck, 'model', '9300') == '9320'
if is_9320: if is_9320:
print("当前设备是9320") print("当前设备是9320")
else:
for site_id in range(len(deck.sites)):
child = deck._get_site_resource(site_id)
# 如果放其他类型的物料,是不可以的
if hasattr(child, "_unilabos_state") and "Material" in child._unilabos_state:
number = site_id + 1
tablets_info.append(
WorkTablets(
Number=number, Code=f"T{number}", Material=child._unilabos_state["Material"]
)
)
# 始终初始化 step_mode 属性 # 始终初始化 step_mode 属性
self.step_mode = False self.step_mode = False
if step_mode: if step_mode:
@@ -841,13 +825,9 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
self.step_mode = step_mode self.step_mode = step_mode
else: else:
print("9300设备不支持 单点动作模式") print("9300设备不支持 单点动作模式")
self._unilabos_backend = PRCXI9300Backend( self._unilabos_backend = PRCXI9300Backend(
tablets_info, plate_positions, host, port, timeout, channel_num, axis, setup, debug, matrix_id, is_9320, tablets_info, host, port, timeout, channel_num, axis, setup, debug, matrix_id, is_9320
x_increase, y_increase, x_offset, y_offset,
deck_z, deck_x=self.deck_x, deck_y=self.deck_y, xy_coupling=xy_coupling
) )
super().__init__(backend=self._unilabos_backend, deck=deck, simulator=simulator, channel_num=channel_num) super().__init__(backend=self._unilabos_backend, deck=deck, simulator=simulator, channel_num=channel_num)
self._first_transfer_done = False self._first_transfer_done = False
@@ -1064,8 +1044,8 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
): ):
self._unilabos_backend.create_protocol(protocol_name) self._unilabos_backend.create_protocol(protocol_name)
async def run_protocol(self): async def run_protocol(self, protocol_id: str = None):
return self._unilabos_backend.run_protocol() return self._unilabos_backend.run_protocol(protocol_id)
async def remove_liquid( async def remove_liquid(
self, self,
@@ -1245,7 +1225,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
spread: Literal["wide", "tight", "custom"] = "wide", spread: Literal["wide", "tight", "custom"] = "wide",
**backend_kwargs, **backend_kwargs,
): ):
try:
return await super().aspirate( return await super().aspirate(
resources, resources,
vols, vols,
@@ -1257,20 +1237,6 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
spread, spread,
**backend_kwargs, **backend_kwargs,
) )
except ValueError as e:
if "Resource is too small to space channels" in str(e) and spread != "custom":
return await super().aspirate(
resources,
vols,
use_channels,
flow_rates,
offsets,
liquid_height,
blow_out_air_volume,
spread="custom",
**backend_kwargs,
)
raise
async def drop_tips( async def drop_tips(
self, self,
@@ -1294,7 +1260,6 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
spread: Literal["wide", "tight", "custom"] = "wide", spread: Literal["wide", "tight", "custom"] = "wide",
**backend_kwargs, **backend_kwargs,
): ):
try:
return await super().dispense( return await super().dispense(
resources, resources,
vols, vols,
@@ -1306,21 +1271,6 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
spread, spread,
**backend_kwargs, **backend_kwargs,
) )
except ValueError as e:
if "Resource is too small to space channels" in str(e) and spread != "custom":
# 目标资源过小无法分布多通道时,退化为 custom所有通道对准中心
return await super().dispense(
resources,
vols,
use_channels,
flow_rates,
offsets,
liquid_height,
blow_out_air_volume,
"custom",
**backend_kwargs,
)
raise
async def discard_tips( async def discard_tips(
self, self,
@@ -1340,11 +1290,6 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
async def shaker_action(self, time: int, module_no: int, amplitude: int, is_wait: bool): async def shaker_action(self, time: int, module_no: int, amplitude: int, is_wait: bool):
return await self._unilabos_backend.shaker_action(time, module_no, amplitude, is_wait) return await self._unilabos_backend.shaker_action(time, module_no, amplitude, is_wait)
async def magnetic_action(self, time: int, module_no: int, height: int, is_wait: bool):
return await self._unilabos_backend.magnetic_action(time, module_no, height, is_wait)
async def shaking_incubation_action(self, time: int, module_no: int, amplitude: int, is_wait: bool, temperature: int):
return await self._unilabos_backend.shaking_incubation_action(time, module_no, amplitude, is_wait, temperature)
async def heater_action(self, temperature: float, time: int): async def heater_action(self, temperature: float, time: int):
return await self._unilabos_backend.heater_action(temperature, time) return await self._unilabos_backend.heater_action(temperature, time)
@@ -1361,7 +1306,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
**backend_kwargs, **backend_kwargs,
): ):
res = await super().move_plate( return await super().move_plate(
plate, plate,
to, to,
intermediate_locations, intermediate_locations,
@@ -1373,12 +1318,6 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
target_plate_number=to, target_plate_number=to,
**backend_kwargs, **backend_kwargs,
) )
plate.unassign()
to.assign_child_resource(plate, location=Coordinate(0, 0, 0))
ROS2DeviceNode.run_async_func(self._ros_node.update_resource, True, **{
"resources": [self.deck]
})
return res
class PRCXI9300Backend(LiquidHandlerBackend): class PRCXI9300Backend(LiquidHandlerBackend):
@@ -1403,7 +1342,6 @@ class PRCXI9300Backend(LiquidHandlerBackend):
def __init__( def __init__(
self, self,
tablets_info: list[WorkTablets], tablets_info: list[WorkTablets],
plate_positions: dict[int, Coordinate],
host: str = "127.0.0.1", host: str = "127.0.0.1",
port: int = 9999, port: int = 9999,
timeout: float = 10.0, timeout: float = 10.0,
@@ -1413,18 +1351,9 @@ class PRCXI9300Backend(LiquidHandlerBackend):
debug=False, debug=False,
matrix_id="", matrix_id="",
is_9320=False, is_9320=False,
x_increase = 0,
y_increase = 0,
x_offset = 0,
y_offset = 0,
deck_z = 300,
deck_x = 0,
deck_y = 0,
xy_coupling = 0.0,
) -> None: ) -> None:
super().__init__() super().__init__()
self.tablets_info = tablets_info self.tablets_info = tablets_info
self.plate_positions = plate_positions
self.matrix_id = matrix_id self.matrix_id = matrix_id
self.api_client = PRCXI9300Api(host, port, timeout, axis, debug, is_9320) self.api_client = PRCXI9300Api(host, port, timeout, axis, debug, is_9320)
self.host, self.port, self.timeout = host, port, timeout self.host, self.port, self.timeout = host, port, timeout
@@ -1432,15 +1361,6 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self._execute_setup = setup self._execute_setup = setup
self.debug = debug self.debug = debug
self.axis = "Left" self.axis = "Left"
self.x_increase = x_increase
self.y_increase = y_increase
self.xy_coupling = xy_coupling
self.x_offset = x_offset
self.y_offset = y_offset
self.deck_x = deck_x
self.deck_y = deck_y
self.deck_z = deck_z
self.tip_length = 0
@staticmethod @staticmethod
def _deck_plate_slot_no(plate, deck) -> int: def _deck_plate_slot_no(plate, deck) -> int:
@@ -1470,27 +1390,6 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self.steps_todo_list.append(step) self.steps_todo_list.append(step)
return step return step
async def shaking_incubation_action(self, time: int, module_no: int, amplitude: int, is_wait: bool, temperature: int):
step = self.api_client.shaking_incubation_action(
time=time,
module_no=module_no,
amplitude=amplitude,
is_wait=is_wait,
temperature=temperature,
)
self.steps_todo_list.append(step)
return step
async def magnetic_action(self, time: int, module_no: int, height: int, is_wait: bool):
step = self.api_client.magnetic_action(
time=time,
module_no=module_no,
height=height,
is_wait=is_wait,
)
self.steps_todo_list.append(step)
return step
async def pick_up_resource(self, pickup: ResourcePickup, **backend_kwargs): async def pick_up_resource(self, pickup: ResourcePickup, **backend_kwargs):
resource = pickup.resource resource = pickup.resource
@@ -1529,8 +1428,6 @@ class PRCXI9300Backend(LiquidHandlerBackend):
self._ros_node = ros_node self._ros_node = ros_node
def create_protocol(self, protocol_name): def create_protocol(self, protocol_name):
if protocol_name == "":
protocol_name = f"protocol_{time.time()}"
self.protocol_name = protocol_name self.protocol_name = protocol_name
self.steps_todo_list = [] self.steps_todo_list = []
@@ -1559,12 +1456,15 @@ class PRCXI9300Backend(LiquidHandlerBackend):
raise AssertionError(f"Failed to create matrix: {res.get('Message', 'Unknown error')}") raise AssertionError(f"Failed to create matrix: {res.get('Message', 'Unknown error')}")
print(f"PRCXI9300Backend created matrix with ID: {self.matrix_info['MatrixId']}, result: {res}") print(f"PRCXI9300Backend created matrix with ID: {self.matrix_info['MatrixId']}, result: {res}")
def run_protocol(self): def run_protocol(self, protocol_id: str = None):
assert self.is_reset_ok, "PRCXI9300Backend is not reset successfully. Please call setup() first." assert self.is_reset_ok, "PRCXI9300Backend is not reset successfully. Please call setup() first."
run_time = time.time() run_time = time.time()
if protocol_id == "" or protocol_id is None:
solution_id = self.api_client.add_solution( solution_id = self.api_client.add_solution(
f"protocol_{run_time}", self.matrix_id, self.steps_todo_list f"protocol_{run_time}", self.matrix_id, self.steps_todo_list
) )
else:
solution_id = protocol_id
print(f"PRCXI9300Backend created solution with ID: {solution_id}") print(f"PRCXI9300Backend created solution with ID: {solution_id}")
self.api_client.load_solution(solution_id) self.api_client.load_solution(solution_id)
print(json.dumps(self.steps_todo_list, indent=2)) print(json.dumps(self.steps_todo_list, indent=2))
@@ -2044,10 +1944,10 @@ class PRCXI9300Api:
start = False start = False
while not success: while not success:
status = self.step_state_list() status = self.step_state_list()
if status is None:
break
if len(status) == 1: if len(status) == 1:
start = True start = True
if status is None:
break
if len(status) == 0: if len(status) == 0:
break break
if status[-1]["State"] == 2 and start: if status[-1]["State"] == 2 and start:
@@ -2383,26 +2283,6 @@ class PRCXI9300Api:
"AssistFun4": is_wait, "AssistFun4": is_wait,
} }
def shaking_incubation_action(self, time: int, module_no: int, amplitude: int, is_wait: bool, temperature: int):
return {
"StepAxis": "Left",
"Function": "Shaking_Incubation",
"AssistFun1": time,
"AssistFun2": module_no,
"AssistFun3": amplitude,
"AssistFun4": is_wait,
"AssistFun5": temperature,
}
def magnetic_action(self, time: int, module_no: int, height: int, is_wait: bool):
return {
"StepAxis": "Left",
"Function": "Magnetic",
"AssistFun1": time,
"AssistFun2": module_no,
"AssistFun3": height,
"AssistFun4": is_wait,
}
class DefaultLayout: class DefaultLayout:
@@ -2417,8 +2297,20 @@ class DefaultLayout:
self.rows = 2 self.rows = 2
self.columns = 3 self.columns = 3
self.layout = [1, 2, 3, 4, 5, 6] self.layout = [1, 2, 3, 4, 5, 6]
self.trash_slot = 3 self.trash_slot = 6
self.waste_liquid_slot = 6 self.default_layout = {
"MatrixId": f"{time.time()}",
"MatrixName": f"{time.time()}",
"MatrixCount": 6,
"WorkTablets": [
{"Number": 1, "Code": "T1", "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0}},
{"Number": 2, "Code": "T2", "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0}},
{"Number": 3, "Code": "T3", "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0}},
{"Number": 4, "Code": "T4", "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0}},
{"Number": 5, "Code": "T5", "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0}},
{"Number": 6, "Code": "T6", "Material": {"uuid": "730067cf07ae43849ddf4034299030e9", "materialEnum": 0}}, # trash
],
}
elif product_name == "PRCXI9320": elif product_name == "PRCXI9320":
self.rows = 4 self.rows = 4
@@ -2434,94 +2326,96 @@ class DefaultLayout:
{ {
"Number": 1, "Number": 1,
"Code": "T1", "Code": "T1",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 2, "Number": 2,
"Code": "T2", "Code": "T2",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 3, "Number": 3,
"Code": "T3", "Code": "T3",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 4, "Number": 4,
"Code": "T4", "Code": "T4",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 5, "Number": 5,
"Code": "T5", "Code": "T5",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 6, "Number": 6,
"Code": "T6", "Code": "T6",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 7, "Number": 7,
"Code": "T7", "Code": "T7",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 8, "Number": 8,
"Code": "T8", "Code": "T8",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 9, "Number": 9,
"Code": "T9", "Code": "T9",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 10, "Number": 10,
"Code": "T10", "Code": "T10",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 11, "Number": 11,
"Code": "T11", "Code": "T11",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 12, "Number": 12,
"Code": "T12", "Code": "T12",
"Material": {"uuid": "730067cf07ae43849ddf4034299030e9"}, "Material": {"uuid": "730067cf07ae43849ddf4034299030e9", "materialEnum": 0},
}, # 这个设置成废液槽,用储液槽表示 }, # 这个设置成废液槽,用储液槽表示
{ {
"Number": 13, "Number": 13,
"Code": "T13", "Code": "T13",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 14, "Number": 14,
"Code": "T14", "Code": "T14",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 15, "Number": 15,
"Code": "T15", "Code": "T15",
"Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f"}, "Material": {"uuid": "57b1e4711e9e4a32b529f3132fc5931f", "materialEnum": 0},
}, },
{ {
"Number": 16, "Number": 16,
"Code": "T16", "Code": "T16",
"Material": {"uuid": "730067cf07ae43849ddf4034299030e9"}, "Material": {"uuid": "730067cf07ae43849ddf4034299030e9", "materialEnum": 0},
}, # 这个设置成垃圾桶,用储液槽表示 }, # 这个设置成垃圾桶,用储液槽表示
], ],
} }
def get_layout(self) -> Dict[str, Any]: def get_layout(self) -> Dict[str, Any]:
return { result = {
"rows": self.rows, "rows": self.rows,
"columns": self.columns, "columns": self.columns,
"layout": self.layout, "layout": self.layout,
"trash_slot": self.trash_slot, "trash_slot": self.trash_slot,
"waste_liquid_slot": self.waste_liquid_slot,
} }
if hasattr(self, 'waste_liquid_slot'):
result["waste_liquid_slot"] = self.waste_liquid_slot
return result
def get_trash_slot(self) -> int: def get_trash_slot(self) -> int:
return self.trash_slot return self.trash_slot
@@ -2539,15 +2433,18 @@ class DefaultLayout:
if material_name not in self.labresource: if material_name not in self.labresource:
raise ValueError(f"Material {reagent_name} not found in lab resources.") raise ValueError(f"Material {reagent_name} not found in lab resources.")
# 预留位置12和16不 # 预留位置动态计算
reserved_positions = {12, 16} reserved_positions = {self.trash_slot}
available_positions = [i for i in range(1, 17) if i not in reserved_positions] if hasattr(self, 'waste_liquid_slot'):
reserved_positions.add(self.waste_liquid_slot)
total_slots = self.rows * self.columns
available_positions = [i for i in range(1, total_slots + 1) if i not in reserved_positions]
# 计算总需求 # 计算总需求
total_needed = sum(count for _, _, count in needs) total_needed = sum(count for _, _, count in needs)
if total_needed > len(available_positions): if total_needed > len(available_positions):
raise ValueError( raise ValueError(
f"需要 {total_needed} 个位置,但只有 {len(available_positions)} 个可用位置(排除位置12和16" f"需要 {total_needed} 个位置,但只有 {len(available_positions)} 个可用位置(排除预留位置 {reserved_positions}"
) )
# 依次分配位置 # 依次分配位置

View File

@@ -0,0 +1,150 @@
from typing import Any, Dict, Optional
from .prcxi import PRCXI9300ModuleSite
class PRCXI9300FunctionalModule(PRCXI9300ModuleSite):
"""
PRCXI 9300 功能模块基类(加热/冷却/震荡/加热震荡/磁吸等)。
设计目标:
- 作为一个可以在工作台上拖拽摆放的实体资源(继承自 PRCXI9300ModuleSite -> ItemizedCarrier
- 顶面存在一个站点site可吸附标准板类资源plate / tip_rack / tube_rack 等)。
- 支持注入 `material_info` (UUID 等),并且在 serialize_state 时做安全过滤。
"""
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
module_type: Optional[str] = None,
category: str = "module",
model: Optional[str] = None,
material_info: Optional[Dict[str, Any]] = None,
**kwargs: Any,
):
super().__init__(
name=name,
size_x=size_x,
size_y=size_y,
size_z=size_z,
material_info=material_info,
model=model,
category=category,
**kwargs,
)
# 记录模块类型(加热 / 冷却 / 震荡 / 加热震荡 / 磁吸)
self.module_type = module_type or "generic"
# 与 PRCXI9300PlateAdapter 一致,使用 _unilabos_state 保存扩展信息
if not hasattr(self, "_unilabos_state") or self._unilabos_state is None:
self._unilabos_state = {}
# super().__init__ 已经在有 material_info 时写入 "Material",这里仅确保存在
if material_info is not None and "Material" not in self._unilabos_state:
self._unilabos_state["Material"] = material_info
# 额外标记 category 和模块类型,便于前端或上层逻辑区分
self._unilabos_state.setdefault("category", category)
self._unilabos_state["module_type"] = module_type
# ============================================================================
# 具体功能模块定义
# 这里的尺寸和 material_info 目前为占位参数,后续可根据实际测量/JSON 配置进行更新。
# 顶面站点尺寸与模块外形一致,保证可以吸附标准 96 板/储液槽等。
# ============================================================================
def PRCXI_Heating_Module(name: str) -> PRCXI9300FunctionalModule:
"""加热模块(顶面可吸附标准板)。"""
return PRCXI9300FunctionalModule(
name=name,
size_x=127.76,
size_y=85.48,
size_z=40.0,
module_type="heating",
model="PRCXI_Heating_Module",
material_info={
"uuid": "TODO-HEATING-MODULE-UUID",
"Code": "HEAT-MOD",
"Name": "PRCXI 加热模块",
"SupplyType": 3,
},
)
def PRCXI_MetalCooling_Module(name: str) -> PRCXI9300FunctionalModule:
"""金属冷却模块(顶面可吸附标准板)。"""
return PRCXI9300FunctionalModule(
name=name,
size_x=127.76,
size_y=85.48,
size_z=40.0,
module_type="metal_cooling",
model="PRCXI_MetalCooling_Module",
material_info={
"uuid": "TODO-METAL-COOLING-MODULE-UUID",
"Code": "METAL-COOL-MOD",
"Name": "PRCXI 金属冷却模块",
"SupplyType": 3,
},
)
def PRCXI_Shaking_Module(name: str) -> PRCXI9300FunctionalModule:
"""震荡模块(顶面可吸附标准板)。"""
return PRCXI9300FunctionalModule(
name=name,
size_x=127.76,
size_y=85.48,
size_z=50.0,
module_type="shaking",
model="PRCXI_Shaking_Module",
material_info={
"uuid": "TODO-SHAKING-MODULE-UUID",
"Code": "SHAKE-MOD",
"Name": "PRCXI 震荡模块",
"SupplyType": 3,
},
)
def PRCXI_Heating_Shaking_Module(name: str) -> PRCXI9300FunctionalModule:
"""加热震荡模块(顶面可吸附标准板)。"""
return PRCXI9300FunctionalModule(
name=name,
size_x=127.76,
size_y=85.48,
size_z=55.0,
module_type="heating_shaking",
model="PRCXI_Heating_Shaking_Module",
material_info={
"uuid": "TODO-HEATING-SHAKING-MODULE-UUID",
"Code": "HEAT-SHAKE-MOD",
"Name": "PRCXI 加热震荡模块",
"SupplyType": 3,
},
)
def PRCXI_Magnetic_Module(name: str) -> PRCXI9300FunctionalModule:
"""磁吸模块(顶面可吸附标准板)。"""
return PRCXI9300FunctionalModule(
name=name,
size_x=127.76,
size_y=85.48,
size_z=30.0,
module_type="magnetic",
model="PRCXI_Magnetic_Module",
material_info={
"uuid": "TODO-MAGNETIC-MODULE-UUID",
"Code": "MAG-MOD",
"Name": "PRCXI 磁吸模块",
"SupplyType": 3,
},
)

View File

@@ -7779,7 +7779,8 @@ liquid_handler.prcxi:
auto-run_protocol: auto-run_protocol:
feedback: {} feedback: {}
goal: {} goal: {}
goal_default: {} goal_default:
protocol_id: null
handles: {} handles: {}
placeholder_keys: {} placeholder_keys: {}
result: {} result: {}
@@ -7788,7 +7789,9 @@ liquid_handler.prcxi:
properties: properties:
feedback: {} feedback: {}
goal: goal:
properties: {} properties:
protocol_id:
type: string
required: [] required: []
type: object type: object
result: {} result: {}

View File

@@ -0,0 +1,70 @@
PRCXI_Heating_Module:
category:
- prcxi
- modules
class:
module: unilabos.devices.liquid_handling.prcxi.prcxi_modules:PRCXI_Heating_Module
type: pylabrobot
description: '加热模块 (Code: HEAT-MOD)'
handles: []
icon: ''
init_param_schema: {}
registry_type: resource
version: 1.0.0
PRCXI_MetalCooling_Module:
category:
- prcxi
- modules
class:
module: unilabos.devices.liquid_handling.prcxi.prcxi_modules:PRCXI_MetalCooling_Module
type: pylabrobot
description: '金属冷却模块 (Code: METAL-COOL-MOD)'
handles: []
icon: ''
init_param_schema: {}
registry_type: resource
version: 1.0.0
PRCXI_Shaking_Module:
category:
- prcxi
- modules
class:
module: unilabos.devices.liquid_handling.prcxi.prcxi_modules:PRCXI_Shaking_Module
type: pylabrobot
description: '震荡模块 (Code: SHAKE-MOD)'
handles: []
icon: ''
init_param_schema: {}
registry_type: resource
version: 1.0.0
PRCXI_Heating_Shaking_Module:
category:
- prcxi
- modules
class:
module: unilabos.devices.liquid_handling.prcxi.prcxi_modules:PRCXI_Heating_Shaking_Module
type: pylabrobot
description: '加热震荡模块 (Code: HEAT-SHAKE-MOD)'
handles: []
icon: ''
init_param_schema: {}
registry_type: resource
version: 1.0.0
PRCXI_Magnetic_Module:
category:
- prcxi
- modules
class:
module: unilabos.devices.liquid_handling.prcxi.prcxi_modules:PRCXI_Magnetic_Module
type: pylabrobot
description: '磁吸模块 (Code: MAG-MOD)'
handles: []
icon: ''
init_param_schema: {}
registry_type: resource
version: 1.0.0

View File

@@ -9,6 +9,9 @@ def register():
from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300TipRack from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300TipRack
from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300Trash from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300Trash
from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300TubeRack from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300TubeRack
from unilabos.devices.liquid_handling.prcxi.prcxi import PRCXI9300ModuleSite
# noinspection PyUnresolvedReferences
from unilabos.devices.liquid_handling.prcxi.prcxi_modules import PRCXI9300FunctionalModule
# noinspection PyUnresolvedReferences # noinspection PyUnresolvedReferences
from unilabos.devices.workstation.workstation_base import WorkStationContainer from unilabos.devices.workstation.workstation_base import WorkStationContainer

View File

@@ -123,6 +123,24 @@ class ResourceDictType(TypedDict):
machine_name: str machine_name: str
class ResourceDictType(TypedDict):
id: str
uuid: str
name: str
description: str
resource_schema: Dict[str, Any]
model: Dict[str, Any]
icon: str
parent_uuid: Optional[str]
parent: Optional["ResourceDictType"]
type: Union[Literal["device"], str]
klass: str
pose: ResourceDictPositionType
config: Dict[str, Any]
data: Dict[str, Any]
extra: Dict[str, Any]
# 统一的资源字典模型parent 自动序列化为 parent_uuidchildren 不序列化 # 统一的资源字典模型parent 自动序列化为 parent_uuidchildren 不序列化
class ResourceDict(BaseModel): class ResourceDict(BaseModel):
id: str = Field(description="Resource ID") id: str = Field(description="Resource ID")
@@ -441,6 +459,8 @@ class ResourceTreeSet(object):
"reagent_bottle": "reagent_bottle", "reagent_bottle": "reagent_bottle",
"flask": "flask", "flask": "flask",
"beaker": "beaker", "beaker": "beaker",
"module": "module",
"carrier": "carrier",
} }
if source in replace_info: if source in replace_info:
return replace_info[source] return replace_info[source]
@@ -553,10 +573,17 @@ class ResourceTreeSet(object):
trees.append(tree_instance) trees.append(tree_instance)
return cls(trees) return cls(trees)
def to_plr_resources(self, skip_devices=True) -> List["PLRResource"]: def to_plr_resources(
self, skip_devices: bool = True, requested_uuids: Optional[List[str]] = None
) -> List["PLRResource"]:
""" """
将 ResourceTreeSet 转换为 PLR 资源列表 将 ResourceTreeSet 转换为 PLR 资源列表
Args:
skip_devices: 是否跳过 device 类型节点
requested_uuids: 若指定,则按此 UUID 顺序返回对应资源(用于批量查询时一一对应),
否则返回各树的根节点列表
Returns: Returns:
List[PLRResource]: PLR 资源实例列表 List[PLRResource]: PLR 资源实例列表
""" """
@@ -571,6 +598,8 @@ class ResourceTreeSet(object):
"deck": "Deck", "deck": "Deck",
"container": "RegularContainer", "container": "RegularContainer",
"tip_spot": "TipSpot", "tip_spot": "TipSpot",
"module": "PRCXI9300ModuleSite",
"carrier": "ItemizedCarrier",
} }
def collect_node_data(node: ResourceDictInstance, name_to_uuid: dict, all_states: dict, name_to_extra: dict): def collect_node_data(node: ResourceDictInstance, name_to_uuid: dict, all_states: dict, name_to_extra: dict):
@@ -612,6 +641,71 @@ class ResourceTreeSet(object):
d["model"] = res.config.get("model", None) d["model"] = res.config.get("model", None)
return d return d
# deserialize 会单独处理的元数据 key不传给构造函数
_META_KEYS = {"type", "parent_name", "location", "children", "rotation", "barcode"}
# deserialize 自定义逻辑使用的 key如 TipSpot 用 prototype_tip 构建 make_tip需保留
_DESERIALIZE_PRESERVED_KEYS = {"prototype_tip"}
def remove_incompatible_params(plr_d: dict) -> None:
"""递归移除 PLR 类不接受的参数,避免 deserialize 报错。
- 移除构造函数不接受的参数(如 compute_height_from_volume、ordering、category
- 对 TubeRack将 ordering 转为 ordered_items
- 保留 deserialize 自定义逻辑需要的 key如 prototype_tip
"""
if "type" in plr_d:
sub_cls = find_subclass(plr_d["type"], PLRResource)
if sub_cls is not None:
spec = inspect.signature(sub_cls)
valid_params = set(spec.parameters.keys())
# TubeRack 特殊处理:先转换 ordering再参与后续过滤
if "ordering" not in valid_params and "ordering" in plr_d:
ordering = plr_d.pop("ordering", None)
if sub_cls.__name__ == "TubeRack":
plr_d["ordered_items"] = (
_ordering_to_ordered_items(plr_d, ordering)
if ordering
else {}
)
# 移除构造函数不接受的参数(保留 META 和 deserialize 自定义逻辑需要的 key
for key in list(plr_d.keys()):
if (
key not in _META_KEYS
and key not in _DESERIALIZE_PRESERVED_KEYS
and key not in valid_params
):
plr_d.pop(key, None)
for child in plr_d.get("children", []):
remove_incompatible_params(child)
def _ordering_to_ordered_items(plr_d: dict, ordering: dict) -> dict:
"""将 ordering 转为 ordered_items从 children 构建 Tube 对象"""
from pylabrobot.resources import Tube, Coordinate
from pylabrobot.serializer import deserialize as plr_deserialize
children = plr_d.get("children", [])
ordered_items = {}
for idx, (ident, child_name) in enumerate(ordering.items()):
child_data = children[idx] if idx < len(children) else None
if child_data is None:
continue
loc_data = child_data.get("location")
loc = (
plr_deserialize(loc_data)
if loc_data
else Coordinate(0, 0, 0)
)
tube = Tube(
name=child_data.get("name", child_name or ident),
size_x=child_data.get("size_x", 10),
size_y=child_data.get("size_y", 10),
size_z=child_data.get("size_z", 50),
max_volume=child_data.get("max_volume", 1000),
)
tube.location = loc
ordered_items[ident] = tube
plr_d["children"] = [] # 已并入 ordered_items避免重复反序列化
return ordered_items
plr_resources = [] plr_resources = []
tracker = DeviceNodeResourceTracker() tracker = DeviceNodeResourceTracker()
@@ -631,9 +725,7 @@ class ResourceTreeSet(object):
raise ValueError( raise ValueError(
f"无法找到类型 {plr_dict['type']} 对应的 PLR 资源类。原始信息:{tree.root_node.res_content}" f"无法找到类型 {plr_dict['type']} 对应的 PLR 资源类。原始信息:{tree.root_node.res_content}"
) )
spec = inspect.signature(sub_cls) remove_incompatible_params(plr_dict)
if "category" not in spec.parameters:
plr_dict.pop("category", None)
plr_resource = sub_cls.deserialize(plr_dict, allow_marshal=True) plr_resource = sub_cls.deserialize(plr_dict, allow_marshal=True)
from pylabrobot.resources import Coordinate from pylabrobot.resources import Coordinate
from pylabrobot.serializer import deserialize from pylabrobot.serializer import deserialize
@@ -653,6 +745,18 @@ class ResourceTreeSet(object):
logger.error(f"堆栈: {traceback.format_exc()}") logger.error(f"堆栈: {traceback.format_exc()}")
raise raise
if requested_uuids:
# 按请求的 UUID 顺序返回对应资源(从整棵树中按 uuid 提取)
result = []
for uid in requested_uuids:
if uid in tracker.uuid_to_resources:
result.append(tracker.uuid_to_resources[uid])
else:
raise ValueError(
f"请求的 UUID {uid} 在资源树中未找到。"
f"可用 UUID 数量: {len(tracker.uuid_to_resources)}"
)
return result
return plr_resources return plr_resources
@classmethod @classmethod
@@ -741,16 +845,6 @@ class ResourceTreeSet(object):
""" """
return [tree.root_node for tree in self.trees] return [tree.root_node for tree in self.trees]
@property
def root_nodes_uuid(self) -> List[ResourceDictInstance]:
"""
获取所有树的根节点
Returns:
所有根节点的资源实例列表
"""
return [tree.root_node.res_content.uuid for tree in self.trees]
@property @property
def all_nodes(self) -> List[ResourceDictInstance]: def all_nodes(self) -> List[ResourceDictInstance]:
""" """
@@ -868,6 +962,17 @@ class ResourceTreeSet(object):
f"从远端同步了 {added_count} 个物料子树" f"从远端同步了 {added_count} 个物料子树"
) )
else: else:
# 二级是物料
if remote_child_name not in local_children_map:
# 本地不存在该物料,直接引入
remote_child.res_content.parent = local_device.res_content
local_device.children.append(remote_child)
local_children_map[remote_child_name] = remote_child
logger.info(
f"物料 '{remote_root_id}/{remote_child_name}': "
f"从远端同步了整个子树"
)
continue
# 二级物料已存在,比较三级子节点是否缺失 # 二级物料已存在,比较三级子节点是否缺失
local_material = local_children_map[remote_child_name] local_material = local_children_map[remote_child_name]
local_material_children_map = {child.res_content.name: child for child in local_material_children_map = {child.res_content.name: child for child in

View File

@@ -320,9 +320,9 @@ class HostNode(BaseROS2DeviceNode):
"/devices/resource_mesh_manager/add_resource_mesh", "/devices/resource_mesh_manager/add_resource_mesh",
callback_group=self.callback_group, callback_group=self.callback_group,
) )
self._action_value_mappings: Dict[str, Dict] = ( self._action_value_mappings: Dict[str, Dict] = {
{} device_id: self._action_value_mappings
) # device_id -> action_value_mappings(本地+远程设备统一存储) } # device_id -> action_value_mappings(本地+远程设备统一存储)
self._slave_registry_configs: Dict[str, Dict] = {} # registry_name -> registry_config(含action_value_mappings) self._slave_registry_configs: Dict[str, Dict] = {} # registry_name -> registry_config(含action_value_mappings)
self._goals: Dict[str, Any] = {} # 用来存储多个目标的状态 self._goals: Dict[str, Any] = {} # 用来存储多个目标的状态
self._online_devices: Set[str] = {f"{self.namespace}/{device_id}"} # 用于跟踪在线设备 self._online_devices: Set[str] = {f"{self.namespace}/{device_id}"} # 用于跟踪在线设备

View File

@@ -0,0 +1,232 @@
{
"nodes": [
{
"id": "PRCXI",
"name": "PRCXI",
"type": "device",
"class": "liquid_handler.prcxi",
"parent": "",
"pose": {
"size": {
"width": 424,
"height": 202,
"depth": 0
}
},
"config": {
"axis": "Left",
"deck": {
"_resource_type": "unilabos.devices.liquid_handling.prcxi.prcxi:PRCXI9300Deck",
"_resource_child_name": "PRCXI_Deck"
},
"host": "10.20.31.167",
"port": 9999,
"debug": false,
"setup": true,
"timeout": 10,
"simulator": false,
"channel_num": 8
},
"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": 404,
"size_y": 182,
"size_z": 0,
"model": "9300",
"rotation": {
"x": 0,
"y": 0,
"z": 0,
"type": "Rotation"
},
"category": "deck",
"barcode": null,
"preferred_pickup_location": null,
"sites": [
{
"label": "T1",
"visible": true,
"occupied_by": null,
"position": {
"x": 0,
"y": 96,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T2",
"visible": true,
"occupied_by": null,
"position": {
"x": 138,
"y": 96,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T3",
"visible": true,
"occupied_by": null,
"position": {
"x": 276,
"y": 96,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T4",
"visible": true,
"occupied_by": null,
"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",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T5",
"visible": true,
"occupied_by": null,
"position": {
"x": 138,
"y": 0,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T6",
"visible": true,
"occupied_by": null,
"position": {
"x": 276,
"y": 0,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
}
]
},
"data": {}
}
],
"edges": []
}

View File

@@ -21,14 +21,13 @@
}, },
"host": "10.20.30.184", "host": "10.20.30.184",
"port": 9999, "port": 9999,
"debug": false, "debug": true,
"setup": false, "setup": true,
"is_9320": true, "is_9320": true,
"timeout": 10, "timeout": 10,
"matrix_id": "", "matrix_id": "5de524d0-3f95-406c-86dd-f83626ebc7cb",
"simulator": false, "simulator": true,
"channel_num": 2, "channel_num": 2
"step_mode": true
}, },
"data": { "data": {
"reset_ok": true "reset_ok": true
@@ -67,7 +66,426 @@
}, },
"category": "deck", "category": "deck",
"barcode": null, "barcode": null,
"preferred_pickup_location": null "preferred_pickup_location": null,
"sites": [
{
"label": "T1",
"visible": true,
"occupied_by": null,
"position": {
"x": 0,
"y": 288,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"container",
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T2",
"visible": true,
"occupied_by": null,
"position": {
"x": 138,
"y": 288,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T3",
"visible": true,
"occupied_by": null,
"position": {
"x": 276,
"y": 288,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T4",
"visible": true,
"occupied_by": null,
"position": {
"x": 414,
"y": 288,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T5",
"visible": true,
"occupied_by": null,
"position": {
"x": 0,
"y": 192,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T6",
"visible": true,
"occupied_by": null,
"position": {
"x": 138,
"y": 192,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T7",
"visible": true,
"occupied_by": null,
"position": {
"x": 276,
"y": 192,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T8",
"visible": true,
"occupied_by": null,
"position": {
"x": 414,
"y": 192,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T9",
"visible": true,
"occupied_by": null,
"position": {
"x": 0,
"y": 96,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T10",
"visible": true,
"occupied_by": null,
"position": {
"x": 138,
"y": 96,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T11",
"visible": true,
"occupied_by": null,
"position": {
"x": 276,
"y": 96,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T12",
"visible": true,
"occupied_by": null,
"position": {
"x": 414,
"y": 96,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T13",
"visible": true,
"occupied_by": null,
"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",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T14",
"visible": true,
"occupied_by": null,
"position": {
"x": 138,
"y": 0,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T15",
"visible": true,
"occupied_by": null,
"position": {
"x": 276,
"y": 0,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
},
{
"label": "T16",
"visible": true,
"occupied_by": null,
"position": {
"x": 414,
"y": 0,
"z": 0
},
"size": {
"width": 128.0,
"height": 86,
"depth": 0
},
"content_type": [
"plate",
"tip_rack",
"plates",
"tip_racks",
"tube_rack",
"adaptor",
"plateadapter",
"module",
"trash"
]
}
]
}, },
"data": {} "data": {}
} }

View File

@@ -108,7 +108,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -153,7 +154,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -198,7 +200,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -243,7 +246,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -288,7 +292,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -333,7 +338,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -378,7 +384,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -423,7 +430,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -468,7 +476,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -513,7 +522,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -558,7 +568,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -603,7 +614,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -648,7 +660,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -693,7 +706,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -738,7 +752,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]
@@ -783,7 +798,8 @@
"tip_rack", "tip_rack",
"plates", "plates",
"tip_racks", "tip_racks",
"tube_rack" "tube_rack",
"plateadapter"
] ]
} }
] ]