feat: RNA add guided siRNA manual load gate

- Expose siRNA order and material handles for manual-confirm load workflows.
- Gate scheduler start on explicit material-load confirmation before calling Bioyond RPC.
- Improve lazy API config diagnostics and Sirna warehouse/material resource handling.
This commit is contained in:
yxz321
2026-05-08 11:24:56 +08:00
parent 18c3263e92
commit 98c27cde40
4 changed files with 425 additions and 96 deletions

View File

@@ -4,8 +4,8 @@ Defines PyLabRobot resource classes for Bioyond Sirna station materials.
Each class is decorated with @resource for AST-based registry discovery.
"""
from typing import Optional
from collections import OrderedDict
from pylabrobot.resources import Plate, TipRack, Container
from unilabos.registry.decorators import resource
@@ -19,20 +19,15 @@ from unilabos.registry.decorators import resource
class BioyondSirna_G3_200ul_TipRack(TipRack):
"""G3-200ul tip rack for Sirna liquid handling."""
def __init__(
self,
name: str,
with_tips: bool = True,
):
super().__init__(
name=name,
size_x=127.76,
size_y=85.48,
size_z=64.0,
model="bioyond_sirna_g3_200ul_tip_rack",
with_tips=with_tips,
ordering=OrderedDict(), # Empty ordering to satisfy PyLabRobot requirement
)
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("model", "bioyond_sirna_g3_200ul_tip_rack")
kwargs.setdefault("with_tips", True)
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
@@ -43,20 +38,15 @@ class BioyondSirna_G3_200ul_TipRack(TipRack):
class BioyondSirna_G3_50ul_TipRack(TipRack):
"""G3-50ul tip rack for Sirna liquid handling."""
def __init__(
self,
name: str,
with_tips: bool = True,
):
super().__init__(
name=name,
size_x=127.76,
size_y=85.48,
size_z=64.0,
model="bioyond_sirna_g3_50ul_tip_rack",
with_tips=with_tips,
ordering=OrderedDict(), # Empty ordering to satisfy PyLabRobot requirement
)
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("model", "bioyond_sirna_g3_50ul_tip_rack")
kwargs.setdefault("with_tips", True)
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
@@ -67,20 +57,15 @@ class BioyondSirna_G3_50ul_TipRack(TipRack):
class BioyondSirna_384WellPlate(Plate):
"""384-well plate for Sirna reporter gene detection."""
def __init__(
self,
name: str,
lid: Optional[object] = None,
):
super().__init__(
name=name,
size_x=127.76,
size_y=85.48,
size_z=14.35,
lid=lid,
model="bioyond_sirna_384_well_plate",
plate_type="skirted",
)
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("model", "bioyond_sirna_384_well_plate")
kwargs.setdefault("plate_type", "skirted")
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
@@ -91,20 +76,15 @@ class BioyondSirna_384WellPlate(Plate):
class BioyondSirna_CellCulturePlate(Plate):
"""Cell culture plate for Sirna experiments."""
def __init__(
self,
name: str,
lid: Optional[object] = None,
):
super().__init__(
name=name,
size_x=127.76,
size_y=85.48,
size_z=14.35,
lid=lid,
model="bioyond_sirna_cell_culture_plate",
plate_type="skirted",
)
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("model", "bioyond_sirna_cell_culture_plate")
kwargs.setdefault("plate_type", "skirted")
if kwargs.get("ordering") is None and kwargs.get("ordered_items") is None:
kwargs["ordering"] = OrderedDict()
super().__init__(*args, **kwargs)
@resource(
@@ -115,19 +95,13 @@ class BioyondSirna_CellCulturePlate(Plate):
class BioyondSirna_ReagentTrough(Container):
"""Reagent trough for Sirna station reagents (RiboGreen, etc.)."""
def __init__(
self,
name: str,
max_volume: float = 300000.0, # 300mL default
):
super().__init__(
name=name,
size_x=127.76,
size_y=85.48,
size_z=44.0,
max_volume=max_volume,
model="bioyond_sirna_reagent_trough",
)
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_sirna_reagent_trough")
super().__init__(*args, **kwargs)
# Material type code mapping for dynamic instantiation

View File

@@ -4,8 +4,19 @@ from pylabrobot.resources.carrier import ResourceHolder, create_homogeneous_reso
from unilabos.resources.warehouse import WareHouse, warehouse_factory
def bioyond_warehouse_numeric_stack(name: str, rows: int = 10, columns: int = 17) -> WareHouse:
"""创建 Bioyond 数字库位堆栈,库位名使用服务端返回的 行-列 格式。"""
def bioyond_warehouse_numeric_stack(
name: str,
rows: int = 10,
columns: int = 17,
bioyond_axis: str = "xy_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 实测约定).
未设置时 graphio 回退到默认 "xy_row_col",其他调用方保持原行为。
"""
num_items_x = columns
num_items_y = rows
num_items_z = 1
@@ -33,7 +44,7 @@ def bioyond_warehouse_numeric_stack(name: str, rows: int = 10, columns: int = 17
for row in range(num_items_y)
for col in range(num_items_x)
]
return WareHouse(
warehouse = WareHouse(
name=name,
size_x=dx + item_dx * num_items_x,
size_y=dy + item_dy * num_items_y,
@@ -45,23 +56,25 @@ def bioyond_warehouse_numeric_stack(name: str, rows: int = 10, columns: int = 17
sites={key: holder for key, holder in zip(keys, holders.values())},
category="warehouse",
)
warehouse.bioyond_axis = bioyond_axis
return warehouse
# ================ 小核酸工作站相关堆栈 ================
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)
return bioyond_warehouse_numeric_stack(name, rows=1, columns=14, bioyond_axis="xy_col_row")
def bioyond_warehouse_sirna_automation_stack(name: str = "自动化堆栈") -> WareHouse:
"""创建小核酸自动化堆栈10 行 x 17 列。"""
return bioyond_warehouse_numeric_stack(name, rows=10, columns=17)
return bioyond_warehouse_numeric_stack(name, rows=10, columns=17, bioyond_axis="xy_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)
return bioyond_warehouse_numeric_stack(name, rows=2, columns=1, bioyond_axis="xy_col_row")
# ================ 反应站相关堆栈 ================

View File

@@ -869,6 +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。
bioyond_axis = getattr(warehouse, "bioyond_axis", "xy_row_col")
if bioyond_axis == "xy_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