From 6b94bdd2da4c6566f2bccb68c346c9f83e7cbdc7 Mon Sep 17 00:00:00 2001 From: yxz321 Date: Sat, 9 May 2026 13:07:48 +0800 Subject: [PATCH] fix: align Bioyond deck warehouse axes - Preserve Sirna col-row labels while flipping visual stack dimensions. - Rebuild Peptide deck warehouses from live API slot geometry and avoid initial graph overlap. - Add Peptide deck layout tests and keep Sirna resource tests passing. --- unilabos/resources/bioyond/decks.py | 160 ++++++++++++++++++++++- unilabos/resources/bioyond/warehouses.py | 111 ++++++++++++++-- unilabos/resources/graphio.py | 8 +- 3 files changed, 258 insertions(+), 21 deletions(-) diff --git a/unilabos/resources/bioyond/decks.py b/unilabos/resources/bioyond/decks.py index 8e2f7543..67b48fec 100644 --- a/unilabos/resources/bioyond/decks.py +++ b/unilabos/resources/bioyond/decks.py @@ -28,6 +28,8 @@ from unilabos.resources.bioyond.warehouses import ( 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, ) @@ -118,6 +120,11 @@ class BIOYOND_SirnaStation_Deck(Deck): "自动化堆栈": "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 覆盖。 @@ -148,14 +155,18 @@ class BIOYOND_SirnaStation_Deck(Deck): data = data.copy() data["setup"] = False result = super().deserialize(data, allow_marshal=allow_marshal) - result._ensure_sirna_warehouse_axis() + result._ensure_sirna_warehouse_metadata() return result - def _ensure_sirna_warehouse_axis(self) -> None: + def _ensure_sirna_warehouse_metadata(self) -> None: for child in getattr(self, "children", []): - axis = self.WAREHOUSE_BIOYOND_AXIS.get(getattr(child, "name", "")) + 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 返回完整固定堆栈清单。 @@ -167,8 +178,8 @@ class BIOYOND_SirnaStation_Deck(Deck): } self.warehouse_locations = { "G3移液站": Coordinate(0.0, 0.0, 0.0), - "自动化堆栈": Coordinate(0.0, 180.0, 0.0), - "离心机配平板堆栈": Coordinate(0.0, 1300.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(): @@ -223,6 +234,145 @@ 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, "row_col") + + def __init__( + self, + name: str = "PeptideStation_Deck", + size_x: float = 3500.0, + size_y: float = 1800.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: + # 已有序列化子资源，跳过 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 setup(self) -> None: + # 多肽工作站仓库配置 + # 基于 2026-05-09 live API probe 发现的实际仓库拓扑 (12个仓库) + # 数据来源: temp_benyao/peptide/_logs/warehouse_discovery_raw_live_2026-05-09.json + self.warehouses = { + # 主自动化堆栈 - live API: code 10-17 -> x=17, y=10，显示为 10 行×17 列 + "自动化堆栈": bioyond_warehouse_numeric_stack( + "自动化堆栈", rows=10, columns=17, bioyond_axis="xy_col_row", bioyond_key_axis="row_col" + ), + + # 低温存储 + "低温冰箱仓库": bioyond_warehouse_live_grid( + "低温冰箱仓库", rows=2, columns=3, slot_keys=["1", "2", "3", "4", "5", "6"] + ), + + # 移液站库位 + "Tecan移液站库": bioyond_warehouse_live_grid( + "Tecan移液站库", rows=1, columns=18, slot_keys=[str(index) for index in range(1, 19)] + ), + "G3移液站库": bioyond_warehouse_live_grid( + "G3移液站库", + rows=1, + columns=18, + slot_keys=["1", "2", "3", "4", "垃圾桶", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18"], + ), + "IDOT移液站库": bioyond_warehouse_live_grid( + "IDOT移液站库", + rows=1, + columns=12, + slot_keys=[f"0009-{index:04d}" for index in range(1, 13)], + ), + + # 缓冲库位 + "G3缓冲库": bioyond_warehouse_live_grid( + "G3缓冲库", rows=1, columns=5, slot_keys=[str(index) for index in range(1, 6)] + ), + "盖板缓冲库": bioyond_warehouse_live_grid( + "盖板缓冲库", rows=1, columns=7, slot_keys=[str(index) for index in range(1, 8)] + ), + "配平板缓冲库": bioyond_warehouse_live_grid( + "配平板缓冲库", rows=1, columns=3, slot_keys=[str(index) for index in range(1, 4)] + ), + "IDOT缓冲库": bioyond_warehouse_live_grid( + "IDOT缓冲库", rows=1, columns=2, slot_keys=["1", "1"] + ), + "固相合成板底座缓冲位": bioyond_warehouse_live_grid( + "固相合成板底座缓冲位", + rows=1, + columns=4, + slot_keys=[f"0015-{index:04d}" for index in range(1, 5)], + ), + + # 设备库位 + "离心机库位": bioyond_warehouse_live_grid( + "离心机库位", rows=1, columns=4, slot_keys=[f"0017-{index:04d}" for index in range(1, 5)] + ), + "热封膜机位": bioyond_warehouse_live_grid( + "热封膜机位", rows=1, columns=2, slot_keys=[f"0016-{index:04d}" for index in range(1, 3)] + ), + } + + # 仓库位置布局 (需根据实际硬件布局调整) + self.warehouse_locations = { + "自动化堆栈": Coordinate(0.0, 0.0, 0.0), + "Tecan移液站库": Coordinate(0.0, 1150.0, 0.0), + "G3移液站库": Coordinate(0.0, 1300.0, 0.0), + "IDOT移液站库": Coordinate(0.0, 1450.0, 0.0), + "G3缓冲库": Coordinate(0.0, 1600.0, 0.0), + "盖板缓冲库": Coordinate(850.0, 1600.0, 0.0), + "低温冰箱仓库": Coordinate(2700.0, 0.0, 0.0), + "配平板缓冲库": Coordinate(2700.0, 300.0, 0.0), + "IDOT缓冲库": Coordinate(2700.0, 450.0, 0.0), + "固相合成板底座缓冲位": Coordinate(2700.0, 600.0, 0.0), + "离心机库位": Coordinate(2700.0, 750.0, 0.0), + "热封膜机位": Coordinate(2700.0, 900.0, 0.0), + } + + 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() diff --git a/unilabos/resources/bioyond/warehouses.py b/unilabos/resources/bioyond/warehouses.py index 7b4efa47..0c512607 100644 --- a/unilabos/resources/bioyond/warehouses.py +++ b/unilabos/resources/bioyond/warehouses.py @@ -5,15 +5,17 @@ from unilabos.resources.warehouse import WareHouse, warehouse_factory class BioyondWareHouse(WareHouse): - """Bioyond 仓库，额外保存服务端 x/y 坐标语义。""" + """Bioyond 仓库，额外保存服务端 x/y 坐标和库位标签语义。""" - def __init__(self, *args, bioyond_axis: str = "xy_row_col", **kwargs): + 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 @@ -22,12 +24,17 @@ def bioyond_warehouse_numeric_stack( rows: int = 10, columns: int = 17, bioyond_axis: str = "xy_row_col", + bioyond_key_axis: str = "row_col", ) -> 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 @@ -52,11 +59,20 @@ def bioyond_warehouse_numeric_stack( resource_size_z=25.0, name_prefix=name, ) - keys = [ - f"{row + 1}-{col + 1}" - for row in range(num_items_y) - for col in range(num_items_x) - ] + 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, @@ -69,25 +85,94 @@ def bioyond_warehouse_numeric_stack( 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", +) -> 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 + locations = [ + Coordinate(dx + col * item_dx, dy + row * item_dy, dz) + for row in range(num_items_y) + for col in range(num_items_x) + ] + holders = create_homogeneous_resources( + klass=ResourceHolder, + locations=locations, + resource_size_x=127.0, + resource_size_y=86.0, + resource_size_z=25.0, + 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=dy + item_dy * num_items_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 移液站库位堆栈：1 行 x 14 列。""" - return bioyond_warehouse_numeric_stack(name, rows=1, columns=14, bioyond_axis="xy_col_row") + """创建小核酸 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: - """创建小核酸自动化堆栈：10 行 x 17 列。""" - return bioyond_warehouse_numeric_stack(name, rows=10, columns=17, bioyond_axis="xy_col_row") + """创建小核酸自动化堆栈：显示为 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: - """创建小核酸离心机配平板堆栈：2 行 x 1 列。""" - return bioyond_warehouse_numeric_stack(name, rows=2, columns=1, bioyond_axis="xy_col_row") + """创建小核酸离心机配平板堆栈：显示为 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" + ) # ================ 反应站相关堆栈 ================ diff --git a/unilabos/resources/graphio.py b/unilabos/resources/graphio.py index decfefb7..7a52c2a2 100644 --- a/unilabos/resources/graphio.py +++ b/unilabos/resources/graphio.py @@ -869,10 +869,12 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st y = loc.get("y", 1) # 列号 (1-based: 1=01, 2=02, 3=03...) z = loc.get("z", 1) # 层号 (1-based, 通常为1) - # 仓库级别的轴约定覆盖：部分工作站 (Sirna 实测) 的 Bioyond 返回 x=列/y=行， - # 与上面的默认 "xy_row_col" 相反。warehouse.bioyond_axis="xy_col_row" 时交换 x/y。 + # 仓库级别的轴约定覆盖。 + # 对旧的 row-col 视觉标签，bioyond_axis="xy_col_row" 需要交换 x/y。 + # 对 Sirna 的 col-row 视觉标签，原始 x 已是视觉行、y 已是视觉列，不再交换。 bioyond_axis = getattr(warehouse, "bioyond_axis", "xy_row_col") - if bioyond_axis == "xy_col_row": + 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)