open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/deepmodeling/Uni-Lab-OS synced 2026-05-25 03:49:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: PEP add Bioyond peptide station runtime

Browse Source 
- Add the Bioyond peptide station package with the station-facing Day2 submission flow inlined into BioyondPeptideStation.
- Add LIMS sample Excel upload, Day2/Day3 order creation helpers, scheduler/reset controls, and manual-confirm start/reset actions.
- Register peptide material PLR resource classes and default peptide material type mappings for runtime resource synchronization.
- Add the Bioyond peptide deck definition and warehouse axis/key-axis metadata needed for peptide layout conversion.
- Update shared Bioyond warehouse/resource conversion helpers so peptide deck coordinates round-trip correctly.
- Include shared Bioyond raw-call debug logging support used by station actions, with a generic local debug output default.
- Register the peptide deck in PLR additional resources for deserialization/import visibility.
- Exclude private temp_benyao docs, HAR/API inputs, live diagnostics, and siRNA-only station/material files from this handoff commit.
This commit is contained in:
yxz321
2026-05-13 19:43:57 +08:00
parent 927c7e95f5
commit 26155b8343
10 changed files with 4453 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
unilabos/resources/bioyond/__init__.py
View File

@@ -0,0 +1,9 @@
try:
from . import peptide_materials # noqa: F401 ensure @resource classes are importable for PLR deserialize
except Exception: # pragma: no cover - 允许轻量环境导入非资源辅助函数
peptide_materials = None # type: ignore[assignment]
try:
from . import sirna_materials # noqa: F401 ensure @resource classes are importable for PLR deserialize
except Exception: # pragma: no cover - 允许轻量环境导入非资源辅助函数
sirna_materials = None # type: ignore[assignment]

289
unilabos/resources/bioyond/decks.py
View File

@@ -1,6 +1,8 @@
from os import name
from pylabrobot.resources import Deck, Coordinate, Rotation
from unilabos.registry.decorators import resource
from unilabos.resources.bioyond.YB_warehouses import (
bioyond_warehouse_1x4x4,
bioyond_warehouse_1x4x4_right, # 新增：右侧仓库 (A05～D08)
@@ -23,6 +25,11 @@ from unilabos.resources.bioyond.YB_warehouses import (
from unilabos.resources.bioyond.warehouses import (
bioyond_warehouse_tipbox_storage_left, # 新增：Tip盒堆栈(左)
bioyond_warehouse_tipbox_storage_right, # 新增：Tip盒堆栈(右)
bioyond_warehouse_sirna_automation_stack,
bioyond_warehouse_sirna_centrifuge_balance_plate_stack,
bioyond_warehouse_sirna_g3_liquid_handler,
bioyond_warehouse_numeric_stack, # 新增：数字编码堆栈 (用于多肽站)
bioyond_warehouse_live_grid,
)
@@ -101,6 +108,83 @@ class BIOYOND_PolymerPreparationStation_Deck(Deck):
for warehouse_name, warehouse in self.warehouses.items():
self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
@resource(
id="BIOYOND_SirnaStation_Deck",
category=["deck"],
description="BIOYOND 小核酸工作站 Deck",
icon="配液站.webp",
)
class BIOYOND_SirnaStation_Deck(Deck):
WAREHOUSE_BIOYOND_AXIS = {
"G3移液站": "xy_col_row",
"自动化堆栈": "xy_col_row",
"离心机配平板堆栈": "xy_col_row",
}
WAREHOUSE_BIOYOND_KEY_AXIS = {
"G3移液站": "col_row",
"自动化堆栈": "col_row",
"离心机配平板堆栈": "col_row",
}
# Bioyond warehouse UUID -> 本地仓库名称 映射。
# 留空时由配置（station config 的 ``warehouse_bioyond_ids``）注入。
# graph 节点也可在 deck.config.warehouse_bioyond_ids 覆盖。
WAREHOUSE_BIOYOND_IDS: dict = {}
def __init__(
self,
name: str = "SirnaStation_Deck",
size_x: float = 2700.0,
size_y: float = 1080.0,
size_z: float = 1500.0,
category: str = "deck",
setup: bool = False,
warehouse_bioyond_ids: dict | None = None,
**kwargs,
) -> None:
super().__init__(name=name, size_x=size_x, size_y=size_y, size_z=size_z)
# 按需写入实例级覆盖；保留默认空 mapping，避免改动模型常量。
self.warehouse_bioyond_ids: dict = dict(self.WAREHOUSE_BIOYOND_IDS)
if warehouse_bioyond_ids:
self.warehouse_bioyond_ids.update(warehouse_bioyond_ids)
if setup:
self.setup()
@classmethod
def deserialize(cls, data: dict, allow_marshal: bool = False):
if data.get("children") and data.get("setup") is True:
data = data.copy()
data["setup"] = False
result = super().deserialize(data, allow_marshal=allow_marshal)
result._ensure_sirna_warehouse_metadata()
return result
def _ensure_sirna_warehouse_metadata(self) -> None:
for child in getattr(self, "children", []):
name = getattr(child, "name", "")
axis = self.WAREHOUSE_BIOYOND_AXIS.get(name)
if axis and not hasattr(child, "bioyond_axis"):
child.bioyond_axis = axis
key_axis = self.WAREHOUSE_BIOYOND_KEY_AXIS.get(name)
if key_axis and not hasattr(child, "bioyond_key_axis"):
child.bioyond_key_axis = key_axis
def setup(self) -> None:
# Sirna 读接口 /api/storage/location/locations-by-type 返回完整固定堆栈清单。
# LIMS 在库物料接口仍使用相同的 自动化堆栈 名称和数字库位编码。
self.warehouses = {
"G3移液站": bioyond_warehouse_sirna_g3_liquid_handler(),
"自动化堆栈": bioyond_warehouse_sirna_automation_stack(),
"离心机配平板堆栈": bioyond_warehouse_sirna_centrifuge_balance_plate_stack(),
}
self.warehouse_locations = {
"G3移液站": Coordinate(0.0, 0.0, 0.0),
"自动化堆栈": Coordinate(220.0, 0.0, 0.0),
"离心机配平板堆栈": Coordinate(1740.0, 0.0, 0.0),
}
for warehouse_name, warehouse in self.warehouses.items():
self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
class BIOYOND_YB_Deck(Deck):
def __init__(
self,
@@ -150,12 +234,207 @@ class BIOYOND_YB_Deck(Deck):
for warehouse_name, warehouse in self.warehouses.items():
self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
@resource(
id="BIOYOND_PeptideStation_Deck",
category=["deck"],
description="BIOYOND 多肽工作站 Deck",
icon="preparation_station.webp",
)
class BIOYOND_PeptideStation_Deck(Deck):
WAREHOUSE_BIOYOND_AXIS = dict.fromkeys(
[
"自动化堆栈",
"低温冰箱仓库",
"Tecan移液站库",
"G3移液站库",
"IDOT移液站库",
"G3缓冲库",
"盖板缓冲库",
"配平板缓冲库",
"IDOT缓冲库",
"固相合成板底座缓冲位",
"离心机库位",
"热封膜机位",
],
"xy_col_row",
)
WAREHOUSE_BIOYOND_KEY_AXIS = dict.fromkeys(WAREHOUSE_BIOYOND_AXIS, "col_row")
def __init__(
self,
name: str = "PeptideStation_Deck",
size_x: float = 2700.0,
size_y: float = 2000.0,
size_z: float = 1500.0,
category: str = "deck",
setup: bool = False
) -> None:
super().__init__(name=name, size_x=size_x, size_y=size_y, size_z=size_z)
if setup:
self.setup()
@classmethod
def deserialize(cls, data: dict, allow_marshal: bool = False):
if data.get("children") and data.get("setup") is True:
data = data.copy()
data["setup"] = False
# 已有序列化子资源，跳过 setup 避免重复创建
result = super(BIOYOND_PeptideStation_Deck, cls).deserialize(data, allow_marshal=allow_marshal)
else:
result = super(BIOYOND_PeptideStation_Deck, cls).deserialize(data, allow_marshal=allow_marshal)
result._ensure_peptide_warehouse_metadata()
return result
def _ensure_peptide_warehouse_metadata(self) -> None:
for child in getattr(self, "children", []):
name = getattr(child, "name", "")
axis = self.WAREHOUSE_BIOYOND_AXIS.get(name)
if axis and not hasattr(child, "bioyond_axis"):
child.bioyond_axis = axis
key_axis = self.WAREHOUSE_BIOYOND_KEY_AXIS.get(name)
if key_axis and not hasattr(child, "bioyond_key_axis"):
child.bioyond_key_axis = key_axis
def _frontend_y_flipped_coordinate(self, display_x: float, display_y: float, child) -> Coordinate:
"""把期望显示坐标转换为兼容前端 y 轴翻转的存储坐标。"""
return Coordinate(display_x, self.get_size_y() - display_y - child.get_size_y(), 0.0)
def setup(self) -> None:
# 多肽工作站仓库配置
# 基于 2026-05-09 live API probe 发现的实际仓库拓扑 (12个仓库)
# 数据来源: Bioyond 现场仓库发现结果。
self.warehouses = {
# 主自动化堆栈 - live API: code 10-17 -> x=17, y=10，显示为 17 行×10 列
"自动化堆栈": bioyond_warehouse_numeric_stack(
"自动化堆栈",
rows=17,
columns=10,
bioyond_axis="xy_col_row",
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 低温存储
"低温冰箱仓库": bioyond_warehouse_live_grid(
"低温冰箱仓库",
rows=3,
columns=2,
slot_keys=["1", "2", "3", "4", "5", "6"],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 移液站库位
"Tecan移液站库": bioyond_warehouse_live_grid(
"Tecan移液站库",
rows=18,
columns=1,
slot_keys=[str(index) for index in range(1, 19)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"G3移液站库": bioyond_warehouse_live_grid(
"G3移液站库",
rows=18,
columns=1,
slot_keys=["1", "2", "3", "4", "垃圾桶", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18"],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"IDOT移液站库": bioyond_warehouse_live_grid(
"IDOT移液站库",
rows=12,
columns=1,
slot_keys=[f"0009-{index:04d}" for index in range(1, 13)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 缓冲库位
"G3缓冲库": bioyond_warehouse_live_grid(
"G3缓冲库",
rows=5,
columns=1,
slot_keys=[str(index) for index in range(1, 6)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"盖板缓冲库": bioyond_warehouse_live_grid(
"盖板缓冲库",
rows=7,
columns=1,
slot_keys=[str(index) for index in range(1, 8)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"配平板缓冲库": bioyond_warehouse_live_grid(
"配平板缓冲库",
rows=3,
columns=1,
slot_keys=[str(index) for index in range(1, 4)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"IDOT缓冲库": bioyond_warehouse_live_grid(
"IDOT缓冲库",
rows=2,
columns=1,
slot_keys=["1", "1"],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"固相合成板底座缓冲位": bioyond_warehouse_live_grid(
"固相合成板底座缓冲位",
rows=4,
columns=1,
slot_keys=[f"0015-{index:04d}" for index in range(1, 5)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
# 设备库位
"离心机库位": bioyond_warehouse_live_grid(
"离心机库位",
rows=4,
columns=1,
slot_keys=[f"0017-{index:04d}" for index in range(1, 5)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
"热封膜机位": bioyond_warehouse_live_grid(
"热封膜机位",
rows=2,
columns=1,
slot_keys=[f"0016-{index:04d}" for index in range(1, 3)],
bioyond_key_axis="col_row",
frontend_y_flip=True,
),
}
# 仓库显示布局：紧凑排列；存储 y 坐标按前端兼容翻转预先反向。
display_layout = {
"自动化堆栈": (0.0, 0.0),
"Tecan移液站库": (1520.0, 0.0),
"G3移液站库": (1710.0, 0.0),
"IDOT移液站库": (1900.0, 0.0),
"G3缓冲库": (2090.0, 0.0),
"盖板缓冲库": (2090.0, 580.0),
"低温冰箱仓库": (2280.0, 0.0),
"配平板缓冲库": (2280.0, 370.0),
"IDOT缓冲库": (2470.0, 370.0),
"固相合成板底座缓冲位": (2280.0, 740.0),
"离心机库位": (2470.0, 740.0),
"热封膜机位": (2280.0, 1210.0),
}
self.warehouse_locations = {
name: self._frontend_y_flipped_coordinate(x, y, self.warehouses[name])
for name, (x, y) in display_layout.items()
}
for warehouse_name, warehouse in self.warehouses.items():
self.assign_child_resource(warehouse, location=self.warehouse_locations[warehouse_name])
def YB_Deck(name: str) -> Deck:
by=BIOYOND_YB_Deck(name=name)
by.setup()
return by

247
unilabos/resources/bioyond/peptide_materials.py Normal file
View File

@@ -0,0 +1,247 @@
"""Peptide Station Material Resource Definitions."""
from __future__ import annotations
from collections import OrderedDict
try:
from pylabrobot.resources import Container, Plate, TipRack
except Exception: # pragma: no cover - 允许无 pylabrobot 的轻量动作测试导入
class _FallbackResource:
def __init__(self, *args, **kwargs):
self.args = args
self.kwargs = kwargs
class Container(_FallbackResource): # type: ignore[no-redef]
pass
class Plate(_FallbackResource): # type: ignore[no-redef]
pass
class TipRack(_FallbackResource): # type: ignore[no-redef]
pass
try:
from unilabos.registry.decorators import resource
except Exception: # pragma: no cover - 允许无完整 registry 依赖时导入常量
def resource(*args, **kwargs):
def decorator(cls):
return cls
return decorator
def _ensure_itemized_ordering(kwargs: dict) -> None:
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
class _PeptideTipRack(TipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 64.0)
kwargs.setdefault("with_tips", True)
_ensure_itemized_ordering(kwargs)
super().__init__(*args, **kwargs)
class _PeptidePlate(Plate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 14.35)
kwargs.setdefault("plate_type", "skirted")
_ensure_itemized_ordering(kwargs)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_1000ul_tip_rack",
category=["labware", "tip_rack"],
description="1000uL tip rack for Bioyond peptide station",
)
class BioyondPeptide_1000ul_TipRack(_PeptideTipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_1000ul_tip_rack")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_200ul_tip_rack",
category=["labware", "tip_rack"],
description="200uL tip rack for Bioyond peptide station",
)
class BioyondPeptide_200ul_TipRack(_PeptideTipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_200ul_tip_rack")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_50ul_tip_rack",
category=["labware", "tip_rack"],
description="50uL tip rack for Bioyond peptide station",
)
class BioyondPeptide_50ul_TipRack(_PeptideTipRack):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_50ul_tip_rack")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_deep_well_plate",
category=["labware", "plate"],
description="96 well deep well plate for Bioyond peptide station",
)
class BioyondPeptide_96WellDeepWellPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_deep_well_plate")
kwargs.setdefault("size_z", 44.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_synthesis_plate",
category=["labware", "plate"],
description="96 well solid-phase synthesis plate for Bioyond peptide station",
)
class BioyondPeptide_96WellSynthesisPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_synthesis_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_synthesis_plate_base",
category=["labware", "adapter"],
description="96 well solid-phase synthesis plate base for Bioyond peptide station",
)
class BioyondPeptide_96WellSynthesisPlateBase(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_synthesis_plate_base")
kwargs.setdefault("size_z", 20.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_96_well_balance_plate",
category=["labware", "plate"],
description="96 well balance plate for Bioyond peptide station",
)
class BioyondPeptide_96WellBalancePlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_96_well_balance_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_384_well_plate",
category=["labware", "plate"],
description="384 well plate for Bioyond peptide station",
)
class BioyondPeptide_384WellPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_384_well_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_384_lcms_plate",
category=["labware", "plate"],
description="384 well LCMS plate for Bioyond peptide station",
)
class BioyondPeptide_384LCMSPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_384_lcms_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_384_balance_plate",
category=["labware", "plate"],
description="384 well balance plate for Bioyond peptide station",
)
class BioyondPeptide_384BalancePlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_384_balance_plate")
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_cover_plate",
category=["labware", "cover"],
description="Cover plate for Bioyond peptide station",
)
class BioyondPeptide_CoverPlate(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_cover_plate")
kwargs.setdefault("size_z", 8.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_sealing_base",
category=["labware", "adapter"],
description="Sealing base for Bioyond peptide station",
)
class BioyondPeptide_SealingBase(_PeptidePlate):
def __init__(self, *args, **kwargs):
kwargs.setdefault("model", "bioyond_peptide_sealing_base")
kwargs.setdefault("size_z", 20.0)
super().__init__(*args, **kwargs)
@resource(
id="bioyond_peptide_reagent_trough",
category=["labware", "trough"],
description="Reagent trough for Bioyond peptide station",
)
class BioyondPeptide_ReagentTrough(Container):
def __init__(self, *args, **kwargs):
kwargs.setdefault("size_x", 127.76)
kwargs.setdefault("size_y", 85.48)
kwargs.setdefault("size_z", 44.0)
kwargs.setdefault("max_volume", 300000.0)
kwargs.setdefault("model", "bioyond_peptide_reagent_trough")
super().__init__(*args, **kwargs)
DEFAULT_PEPTIDE_MATERIAL_TYPE_MAPPINGS = {
"bioyond_peptide_1000ul_tip_rack": ["1000μL枪头盒", "3a1890bb-736e-cfdd-3213-eb314e8a60f9"],
"bioyond_peptide_200ul_tip_rack": ["200μL枪头盒", "3a1890bb-36d1-964a-18bd-0bf0f2877a7b"],
"bioyond_peptide_50ul_tip_rack": ["50μL枪头盒", "3a1890bc-5fae-361c-cc09-e6f2f6dcd71d"],
"bioyond_peptide_96_well_deep_well_plate": ["96孔深孔板", "3a1890bc-1fa8-fe39-9faa-12279ed4569b"],
"bioyond_peptide_96_well_synthesis_plate": ["96孔固相合成板", "3a1871cb-99f3-f01d-23e2-08dbbd0045b5"],
"bioyond_peptide_96_well_synthesis_plate_base": ["96孔固相合成板底座", "3a1b997e-241b-64f0-80d1-47bca08799d1"],
"bioyond_peptide_96_well_balance_plate": ["96孔配平板", "3a187661-2378-1e20-fa5c-a27d49fdc15d"],
"bioyond_peptide_384_well_plate": ["384孔酶标板", "3a1890bf-2148-ed20-92bd-d85869947d9a"],
"bioyond_peptide_384_lcms_plate": ["384孔LCMS板", "3a1e6a8b-cb61-74da-a089-8e6f197f80f0"],
"bioyond_peptide_384_balance_plate": ["384孔配平板", "3a18be7e-47cc-888c-fc68-055753286826"],
"bioyond_peptide_cover_plate": ["防挥发盖板", "3a19d5a6-b0e2-b486-e5eb-bcabc632f4de"],
"bioyond_peptide_sealing_base": ["封膜底座", "3a1d1d7b-e33b-6975-165d-c56cba5ed345"],
"bioyond_peptide_reagent_trough": ["12道试剂槽", "3a18b431-ac58-ca2e-9680-2a4f5880ea45"],
}
MATERIAL_TYPE_CODE_TO_CLASS = {
"0001": BioyondPeptide_96WellSynthesisPlate,
"0002": BioyondPeptide_96WellBalancePlate,
"0008": BioyondPeptide_200ul_TipRack,
"0009": BioyondPeptide_1000ul_TipRack,
"0011": BioyondPeptide_96WellDeepWellPlate,
"0012": BioyondPeptide_50ul_TipRack,
"0016": BioyondPeptide_384WellPlate,
"0018": BioyondPeptide_384WellPlate,
"0024": BioyondPeptide_ReagentTrough,
"0026": BioyondPeptide_384BalancePlate,
"0035": BioyondPeptide_CoverPlate,
"0039": BioyondPeptide_96WellSynthesisPlateBase,
"0041": BioyondPeptide_SealingBase,
"0049": BioyondPeptide_384LCMSPlate,
}
def get_material_class_by_type_code(type_code: str):
"""Return a peptide material class by Bioyond material type code."""
return MATERIAL_TYPE_CODE_TO_CLASS.get(type_code)

187
unilabos/resources/bioyond/warehouses.py
View File

@@ -1,5 +1,192 @@
from pylabrobot.resources import Coordinate
from pylabrobot.resources.carrier import ResourceHolder, create_homogeneous_resources
from unilabos.resources.warehouse import WareHouse, warehouse_factory
class BioyondWareHouse(WareHouse):
"""Bioyond 仓库，额外保存服务端 x/y 坐标和库位标签语义。"""
def __init__(self, *args, bioyond_axis: str = "xy_row_col", bioyond_key_axis: str = "row_col", **kwargs):
super().__init__(*args, **kwargs)
self.bioyond_axis = bioyond_axis
self.bioyond_key_axis = bioyond_key_axis
def serialize(self) -> dict:
data = super().serialize()
data["bioyond_axis"] = self.bioyond_axis
data["bioyond_key_axis"] = self.bioyond_key_axis
return data
def bioyond_warehouse_numeric_stack(
name: str,
rows: int = 10,
columns: int = 17,
bioyond_axis: str = "xy_row_col",
bioyond_key_axis: str = "row_col",
frontend_y_flip: bool = False,
) -> WareHouse:
"""创建 Bioyond 数字库位堆栈，库位名使用服务端返回的 行-列 格式。
bioyond_axis: 仓库级别的 Bioyond 坐标轴约定，供 graphio 的坐标映射使用。
- "xy_row_col" (default): Bioyond x→row, y→col (reaction/peptide 历史约定).
- "xy_col_row": Bioyond x→col, y→row (Sirna live API 实测约定).
bioyond_key_axis: 库位标签生成约定。
- "row_col" (default): 视觉行列和标签行列一致，例如 10 行 x 17 列 → 1-1..10-17。
- "col_row": 视觉行列转置，但标签仍保持 Bioyond row-col，例如
17 行 x 10 列 → 1-1..10-17。
未设置时 graphio 回退到默认 "xy_row_col"，其他调用方保持原行为。
"""
num_items_x = columns
num_items_y = rows
num_items_z = 1
dx = 10.0
dy = 10.0
dz = 10.0
item_dx = 147.0
item_dy = 106.0
item_dz = 130.0
resource_size_x = 127.0
resource_size_y = 86.0
resource_size_z = 25.0
size_y = dy + item_dy * num_items_y
locations = []
for row in range(num_items_y):
display_y = dy + row * item_dy
y = size_y - display_y - resource_size_y if frontend_y_flip else display_y
for col in range(num_items_x):
locations.append(Coordinate(dx + col * item_dx, y, dz))
holders = create_homogeneous_resources(
klass=ResourceHolder,
locations=locations,
resource_size_x=resource_size_x,
resource_size_y=resource_size_y,
resource_size_z=resource_size_z,
name_prefix=name,
)
if bioyond_key_axis == "row_col":
keys = [
f"{row + 1}-{col + 1}"
for row in range(num_items_y)
for col in range(num_items_x)
]
elif bioyond_key_axis == "col_row":
keys = [
f"{col + 1}-{row + 1}"
for row in range(num_items_y)
for col in range(num_items_x)
]
else:
raise ValueError(f"未知 Bioyond 库位标签约定: {bioyond_key_axis!r}")
warehouse = BioyondWareHouse(
name=name,
size_x=dx + item_dx * num_items_x,
size_y=size_y,
size_z=dz + item_dz * num_items_z,
num_items_x=num_items_x,
num_items_y=num_items_y,
num_items_z=num_items_z,
ordering_layout="row-major",
sites={key: holder for key, holder in zip(keys, holders.values())},
category="warehouse",
bioyond_axis=bioyond_axis,
bioyond_key_axis=bioyond_key_axis,
)
return warehouse
def bioyond_warehouse_live_grid(
name: str,
rows: int,
columns: int,
slot_keys: list[str] | None = None,
bioyond_axis: str = "xy_col_row",
bioyond_key_axis: str = "row_col",
frontend_y_flip: bool = False,
) -> WareHouse:
"""创建 Bioyond 实测库位网格，按服务端 code 保存位点标签。
默认用于 Peptide live API 返回的坐标：x 是视觉列，y 是视觉行。
当服务端 code 重复时，为保持 PLR ordering 唯一性，会给后续重复项追加 ``#N``。
"""
num_items_x = columns
num_items_y = rows
num_items_z = 1
dx = 10.0
dy = 10.0
dz = 10.0
item_dx = 147.0
item_dy = 106.0
item_dz = 130.0
resource_size_x = 127.0
resource_size_y = 86.0
resource_size_z = 25.0
size_y = dy + item_dy * num_items_y
locations = []
for row in range(num_items_y):
display_y = dy + row * item_dy
y = size_y - display_y - resource_size_y if frontend_y_flip else display_y
for col in range(num_items_x):
locations.append(Coordinate(dx + col * item_dx, y, dz))
holders = create_homogeneous_resources(
klass=ResourceHolder,
locations=locations,
resource_size_x=resource_size_x,
resource_size_y=resource_size_y,
resource_size_z=resource_size_z,
name_prefix=name,
)
keys = slot_keys or [str(index + 1) for index in range(num_items_x * num_items_y)]
if len(keys) != len(holders):
raise ValueError(f"{name} 库位数量不匹配: keys={len(keys)}, holders={len(holders)}")
seen: dict[str, int] = {}
unique_keys: list[str] = []
for key in keys:
count = seen.get(key, 0) + 1
seen[key] = count
unique_keys.append(key if count == 1 else f"{key}#{count}")
return BioyondWareHouse(
name=name,
size_x=dx + item_dx * num_items_x,
size_y=size_y,
size_z=dz + item_dz * num_items_z,
num_items_x=num_items_x,
num_items_y=num_items_y,
num_items_z=num_items_z,
ordering_layout="row-major",
sites={key: holder for key, holder in zip(unique_keys, holders.values())},
category="warehouse",
bioyond_axis=bioyond_axis,
bioyond_key_axis=bioyond_key_axis,
)
# ================ 小核酸工作站相关堆栈 ================
def bioyond_warehouse_sirna_g3_liquid_handler(name: str = "G3移液站") -> WareHouse:
"""创建小核酸 G3 移液站库位堆栈：显示为 14 行 x 1 列，标签保持 1-1..1-14。"""
return bioyond_warehouse_numeric_stack(
name, rows=14, columns=1, bioyond_axis="xy_col_row", bioyond_key_axis="col_row"
)
def bioyond_warehouse_sirna_automation_stack(name: str = "自动化堆栈") -> WareHouse:
"""创建小核酸自动化堆栈：显示为 17 行 x 10 列，标签保持 1-1..10-17。"""
return bioyond_warehouse_numeric_stack(
name, rows=17, columns=10, bioyond_axis="xy_col_row", bioyond_key_axis="col_row"
)
def bioyond_warehouse_sirna_centrifuge_balance_plate_stack(name: str = "离心机配平板堆栈") -> WareHouse:
"""创建小核酸离心机配平板堆栈：显示为 1 行 x 2 列，标签保持 1-1、2-1。"""
return bioyond_warehouse_numeric_stack(
name, rows=1, columns=2, bioyond_axis="xy_col_row", bioyond_key_axis="col_row"
)
# ================ 反应站相关堆栈 ================
def bioyond_warehouse_1x4x4(name: str) -> WareHouse:

56
unilabos/resources/graphio.py
View File

@@ -736,7 +736,7 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
logger.warning(f"物料 {unique_name} 不是有效的 ResourcePLR 实例，类型: {type(plr_material)}")
continue
plr_material.code = material.get("code", "") and material.get("barCode", "") or ""
plr_material.code = material.get("barCode") or material.get("code") or ""
plr_material.unilabos_uuid = str(uuid.uuid4())
# ⭐ 保存 Bioyond 原始信息到 unilabos_extra（用于出库时查询）
@@ -864,11 +864,22 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
warehouse = deck.warehouses[wh_name]
logger.debug(f"[Warehouse匹配] 找到warehouse: {wh_name} (容量: {warehouse.capacity}, 行×列: {warehouse.num_items_x}×{warehouse.num_items_y})")
# Bioyond坐标映射 (重要！): x→行(1=A,2=B...), y→列(1=01,2=02...), z→层(通常=1)
x = loc.get("x", 1) # 行号 (1-based: 1=A, 2=B, 3=C, 4=D)
y = loc.get("y", 1) # 列号 (1-based: 1=01, 2=02, 3=03...)
# Bioyond坐标映射:
# - 历史 row_col 仓库中 x/y 直接按行/列参与索引。
# - Sirna 的库位标签为 col-row，stock-material 返回 x=标签第二段、y=标签第一段。
# 因此 x=13,y=4 应落到 key=4-13，而不是交换后落到 3-5。
x = loc.get("x", 1)
y = loc.get("y", 1)
z = loc.get("z", 1) # 层号 (1-based, 通常为1)
# 仓库级别的轴约定覆盖。
# 对旧的 row-col 视觉标签，bioyond_axis="xy_col_row" 需要交换 x/y。
# 对 Sirna 的 col-row 库位标签，原始 x/y 已能直接索引到 code 对应位置，不再交换。
bioyond_axis = getattr(warehouse, "bioyond_axis", "xy_row_col")
bioyond_key_axis = getattr(warehouse, "bioyond_key_axis", "row_col")
if bioyond_axis == "xy_col_row" and bioyond_key_axis != "col_row":
x, y = y, x
# 如果是右侧堆栈，需要调整列号 (5→1, 6→2, 7→3, 8→4)
if wh_name == "堆栈1右":
y = y - 4 # 将5-8映射到1-4
@@ -912,10 +923,43 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
logger.debug(f"列优先warehouse {wh_name}: x={x}(行),y={y}(列) → row={row_idx},col={col_idx} → idx={idx}")
if 0 <= idx < warehouse.capacity:
if warehouse[idx] is None or isinstance(warehouse[idx], ResourceHolder):
slot_key = None
ordering = getattr(warehouse, "_ordering", {})
sites = getattr(warehouse, "sites", [])
if isinstance(ordering, dict) and idx < len(sites):
site_at_idx = sites[idx]
slot_key = next(
(key for key, site in ordering.items() if site is site_at_idx),
None,
)
current_resource = warehouse[idx]
if current_resource is None or isinstance(current_resource, (ResourceHolder, str)):
if isinstance(current_resource, str):
logger.warning(
f"⚠️ 物料 {unique_name} 覆盖 {wh_name}[{idx}]"
f"{f'({slot_key})' if slot_key else ''} 的旧占位 occupied_by={current_resource!r}"
)
# 物料尺寸已在放入warehouse前根据需要进行了交换
warehouse[idx] = plr_material
logger.debug(f"✅ 物料 {unique_name} 放置到 {wh_name}[{idx}] (Bioyond坐标: x={loc.get('x')}, y={loc.get('y')})")
logger.debug(
f"✅ 物料 {unique_name} 放置到 {wh_name}[{idx}]"
f"{f'({slot_key})' if slot_key else ''} "
f"(Bioyond坐标: x={loc.get('x')}, y={loc.get('y')})"
)
else:
parent = getattr(current_resource, "parent", None)
current_repr = repr(current_resource)
current_len = len(current_resource) if isinstance(current_resource, str) else None
logger.warning(
f"⚠️ 物料 {unique_name} 跳过放置到 {wh_name}[{idx}]"
f"{f'({slot_key})' if slot_key else ''}：目标库位已有 "
f"{type(current_resource).__name__}"
f"(value={current_repr}, len={current_len})"
f"(name={getattr(current_resource, 'name', None)}, "
f"parent={getattr(parent, 'name', None)}, "
f"uuid={getattr(current_resource, 'unilabos_uuid', None)})"
)
else:
logger.warning(f"❌ 物料 {unique_name} 的索引 {idx} 超出仓库 {wh_name} 容量 {warehouse.capacity}")
else:

4
unilabos/resources/plr_additional_res_reg.py
View File

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