From 98c27cde40b6b26ca4a2d094a503466c481c9e32 Mon Sep 17 00:00:00 2001 From: yxz321 Date: Fri, 8 May 2026 11:24:56 +0800 Subject: [PATCH] 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. --- .../sirna_station/sirna_station.py | 376 +++++++++++++++++- unilabos/resources/bioyond/sirna_materials.py | 114 ++---- unilabos/resources/bioyond/warehouses.py | 25 +- unilabos/resources/graphio.py | 6 + 4 files changed, 425 insertions(+), 96 deletions(-) diff --git a/unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py b/unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py index 52a36e17..0c40674a 100644 --- a/unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py +++ b/unilabos/devices/workstation/bioyond_studio/sirna_station/sirna_station.py @@ -12,7 +12,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Annotated, Any, Dict, Iterable, List, Literal, Optional, Tuple from urllib import error, request -from uuid import UUID +from uuid import NAMESPACE_URL, UUID, uuid5 try: from typing_extensions import TypedDict @@ -38,7 +38,15 @@ except Exception: # pragma: no cover - 允许无 pylabrobot 依赖时导入轻 _SIRNA_DECK_CLASS = None try: - from unilabos.registry.decorators import NodeType, action, device, not_action + from unilabos.registry.decorators import ( + ActionInputHandle, + ActionOutputHandle, + DataSource, + NodeType, + action, + device, + not_action, + ) _REGISTRY_IMPORT_ERROR: Optional[Exception] = None except Exception as exc: # pragma: no cover - 允许无完整依赖时导入轻量 helper _REGISTRY_IMPORT_ERROR = exc @@ -46,6 +54,23 @@ except Exception as exc: # pragma: no cover - 允许无完整依赖时导入轻 class NodeType: # type: ignore[no-redef] MANUAL_CONFIRM = "manual_confirm" + class DataSource: # type: ignore[no-redef] + HANDLE = "handle" + EXECUTOR = "executor" + + class _FallbackActionHandle: # pragma: no cover - lightweight import-only fallback + def __init__(self, **kwargs: Any) -> None: + self.__dict__.update(kwargs) + + def to_registry_dict(self) -> Dict[str, Any]: + return dict(self.__dict__) + + class ActionInputHandle(_FallbackActionHandle): # type: ignore[no-redef] + pass + + class ActionOutputHandle(_FallbackActionHandle): # type: ignore[no-redef] + pass + def device(*args: Any, **kwargs: Any): def decorator(cls): return cls @@ -64,6 +89,15 @@ except Exception as exc: # pragma: no cover - 允许无完整依赖时导入轻 def not_action(func): return func +try: + from unilabos.registry.placeholder_type import DeviceSlot, ResourceSlot +except Exception: # pragma: no cover - 允许无 pylabrobot 依赖时导入轻量 helper + class ResourceSlot: # type: ignore[no-redef] + pass + + class DeviceSlot(str): # type: ignore[no-redef] + pass + try: from unilabos.devices.workstation.workstation_base import WorkstationBase from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation @@ -244,7 +278,18 @@ class BioyondSirnaStation(BioyondWorkstation): logger.info(f" - API Host: {self.bioyond_config.get('api_host', '')}") logger.info(f" - Workflow 映射数量: {len(self.bioyond_config.get('workflow_mappings', {}))}") - self._lazy_frontend_init = deck is None or not self._has_required_api_config(self.bioyond_config) + missing_api_keys = self._missing_api_config_keys(self.bioyond_config) + if missing_api_keys: + logger.warning( + "BioyondSirnaStation 缺少 Bioyond API 配置 %s，进入延迟初始化模式。" + "动作可在调用前通过 config 重新配置或在 goal 中传入 api_host/api_key 后再使用。" + "缺失项可来自 graph 节点 config，或环境变量 " + "BIOYOND_SIRNA_API_HOST / BIOYOND_SIRNA_EXP1_API_HOST 与 " + "BIOYOND_SIRNA_API_KEY / BIOYOND_SIRNA_EXP1_API_KEY。", + missing_api_keys, + ) + + self._lazy_frontend_init = deck is None or bool(missing_api_keys) if self._lazy_frontend_init: WorkstationBase.__init__(self, deck=deck) self.is_running = False @@ -256,10 +301,11 @@ class BioyondSirnaStation(BioyondWorkstation): self._http_service_config = self.bioyond_config.get("http_service_config", {}) if "workflow_mappings" in self.bioyond_config: self.workflow_mappings = dict(self.bioyond_config.get("workflow_mappings") or {}) - logger.warning( - "BioyondSirnaStation 以延迟模式初始化：缺少 deck 或 api_host/api_key，" - "设备会先注册动作，首次调用动作时再检查 Bioyond API 配置。" - ) + if deck is None: + logger.warning( + "BioyondSirnaStation deck 未提供（graph 节点缺少 config.deck），" + "Bioyond 后台同步与 HTTP 服务暂不启动；动作将在补齐 deck 后才能完成 RPC 调用。" + ) else: super().__init__(bioyond_config=self.bioyond_config, deck=deck) @@ -269,10 +315,30 @@ class BioyondSirnaStation(BioyondWorkstation): def post_init(self, ros_node: Any) -> None: if getattr(self, "_lazy_frontend_init", False): WorkstationBase.post_init(self, ros_node) - logger.warning("BioyondSirnaStation 延迟模式跳过 Bioyond 后台同步和 HTTP 服务启动") + logger.warning( + "BioyondSirnaStation 处于延迟模式：跳过 Bioyond 后台同步和 HTTP 服务启动。" + "首次调用动作时会重新检查 api_host/api_key 与环境变量。" + ) return super().post_init(ros_node) + @staticmethod + def _missing_api_config_keys(config: Dict[str, Any]) -> List[str]: + missing: List[str] = [] + if BioyondSirnaStation._is_blank_value(config.get("api_host")): + missing.append("api_host") + if BioyondSirnaStation._is_blank_value(config.get("api_key")): + missing.append("api_key") + return missing + + @staticmethod + def _is_blank_value(value: Any) -> bool: + if value is None: + return True + if isinstance(value, str): + return value.strip() == "" + return False + def fetch_workflow_list( self, workflow_type: int = 0, @@ -339,6 +405,58 @@ class BioyondSirnaStation(BioyondWorkstation): }, feedback_interval=300, description="提交小核酸实验1（报告基因检测）", + handles=[ + ActionOutputHandle( + key="order_ids", + data_type="bioyond_order_ids", + label="实验ID", + data_key="order_ids", + data_source=DataSource.EXECUTOR, + ), + ActionOutputHandle( + key="order_code", + data_type="bioyond_order_code", + label="订单编号", + data_key="order_code", + data_source=DataSource.EXECUTOR, + ), + ActionOutputHandle( + key="order_name", + data_type="bioyond_order_name", + label="订单名称", + data_key="order_name", + data_source=DataSource.EXECUTOR, + ), + ActionOutputHandle( + key="target_device", + data_type="device_id", + label="目标设备", + data_key="target_device", + data_source=DataSource.EXECUTOR, + ), + ActionOutputHandle( + # TEMP frontend-compat key: material/resource name. + key="resource", + data_type="resource", + label="待装载物料", + data_key="resource", + data_source=DataSource.EXECUTOR, + ), + ActionOutputHandle( + # TEMP frontend-compat key: display as material/resource name. + key="coin_cell_code", data_type="array", + label="物料名称", + data_key="coin_cell_code", + data_source=DataSource.EXECUTOR, + ), + ActionOutputHandle( + # TEMP frontend-compat key: display as mount/position. + key="mount_resource", data_type="resource", + label="库位", + data_key="mount_resource", + data_source=DataSource.EXECUTOR, + ), + ], ) def submit_experiment_1( self, @@ -385,6 +503,11 @@ class BioyondSirnaStation(BioyondWorkstation): order_name = optional_params.get("order_name", "") parameter_overrides = optional_params.get("parameter_overrides", "") auto_register_materials = optional_params.get("auto_register_materials", True) + target_device = ( + self._kwarg_text(kwargs, "unilabos_device_id") + or self._kwarg_text(kwargs, "device_id") + or "bioyond_sirna_station" + ) del timeout_seconds, assignee_user_ids, kwargs rpc = self._require_hardware_interface("create_order") @@ -454,6 +577,7 @@ class BioyondSirnaStation(BioyondWorkstation): "order_code": resolved_order_code, "order_name": resolved_order_name, "order_ids": order_ids, + "target_device": target_device, "workflow": workflow, "sample_throughput": int(sample_throughput), "payload": order_payload, @@ -461,11 +585,18 @@ class BioyondSirnaStation(BioyondWorkstation): "create_order_result": parsed_result, "materials": material_records, "materials_by_type": confirmation_data.get("materials_by_type", {}), + "manual_load_tables": self._build_manual_load_tables( + confirmation_data.get("materials_by_type", {}) + ), + "manual_load_probe": self._build_manual_load_probe( + confirmation_data.get("materials_by_type", {}) + ), "suggested_locations": suggested_locations, "start_experiment": start_experiment_info, "confirmation_message": confirmation_data.get("confirmation_message", ""), "registration_result": registration_result, } + result.update(result["manual_load_probe"]) return result def _parse_parameter_overrides_text(self, text: str) -> Dict[str, Any]: @@ -508,16 +639,73 @@ class BioyondSirnaStation(BioyondWorkstation): @action( always_free=True, node_type=NodeType.MANUAL_CONFIRM, - placeholder_keys={"assignee_user_ids": "unilabos_manual_confirm"}, - goal_default={"timeout_seconds": 3600, "assignee_user_ids": []}, + placeholder_keys={ + "resource": "unilabos_resources", + "target_device": "unilabos_devices", + "mount_resource": "unilabos_resources", + "assignee_user_ids": "unilabos_manual_confirm", + }, + goal_default={ + "materials_loaded": False, + "timeout_seconds": 3600, + "assignee_user_ids": [], + }, feedback_interval=300, - description="启动小核酸实验调度", + description="请核对并装载实验物料；勾选装载确认后方可启动调度", + handles=[ + ActionInputHandle( + key="target_device", + data_type="device_id", + label="目标设备", + data_key="target_device", + data_source=DataSource.HANDLE, + io_type="source", + ), + ActionInputHandle( + key="resource", + data_type="resource", + label="待装载物料", + data_key="resource", + data_source=DataSource.HANDLE, + io_type="source", + ), + ActionInputHandle( + # TEMP frontend-compat key: material/resource name. + key="coin_cell_code", + data_type="array", + label="物料名称", + data_key="coin_cell_code", + data_source=DataSource.HANDLE, + io_type="source", + ), + ActionInputHandle( + # TEMP frontend-compat key: mount/position. + key="mount_resource", + data_type="resource", + label="库位", + data_key="mount_resource", + data_source=DataSource.HANDLE, + io_type="source", + ), + # Order metadata for scheduler start. + ActionInputHandle( + key="order_ids", data_type="bioyond_order_ids", + label="实验ID", data_key="order_ids", + data_source=DataSource.HANDLE, + io_type="source", + ), + ], ) def start_experiment( self, + target_device: DeviceSlot = "", + resource: Optional[List[ResourceSlot]] = None, + coin_cell_code: Optional[List[str]] = None, + mount_resource: Optional[List[ResourceSlot]] = None, submit_experiment_result: Optional[Dict[str, Any]] = None, order_id: str = "", order_ids: Optional[List[str]] = None, + materials_loaded: bool = False, api_host: str = "", api_key: str = "", ready_signal: str = "READY", @@ -525,11 +713,33 @@ class BioyondSirnaStation(BioyondWorkstation): assignee_user_ids: Optional[List[str]] = None, **kwargs: Any, ) -> Dict[str, Any]: - """启动 Bioyond 调度器。""" - del timeout_seconds, assignee_user_ids, kwargs + """Guided manual-load checkpoint that gates ``rpc.scheduler_start()``.""" + del target_device, timeout_seconds, assignee_user_ids, kwargs self._update_runtime_api_config(api_host=api_host, api_key=api_key) self._require_ready_signal(ready_signal) - start_info = self._resolve_start_experiment_info(submit_experiment_result, order_id, order_ids) + + category_arrays = { + "materials_loaded": ( + "物料", + self._as_manual_gate(materials_loaded), + [resource, coin_cell_code, mount_resource], + ), + } + gates: Dict[str, Dict[str, Any]] = {} + missing_labels: List[str] = [] + for gate_key, (label, ticked, arrays) in category_arrays.items(): + required = any(bool(arr) for arr in arrays) + gates[gate_key] = {"label": label, "required": required, "ticked": bool(ticked)} + if required and not ticked: + missing_labels.append(label) + if missing_labels: + raise RuntimeError( + f"以下分类装载尚未确认，无法启动调度: {', '.join(missing_labels)}" + ) + + start_info = self._resolve_start_experiment_info( + submit_experiment_result, order_id, order_ids + ) rpc = self._require_hardware_interface("scheduler_start") logger.info("正在启动小核酸调度器") result = rpc.scheduler_start() @@ -538,6 +748,7 @@ class BioyondSirnaStation(BioyondWorkstation): "return_info": result, "scheduler_start_result": result, "start_experiment": start_info, + "gates": gates, "confirmation_message": "调度器启动成功" if result == 1 else "调度器启动失败，请检查 LIMS 状态", }) @@ -557,18 +768,29 @@ class BioyondSirnaStation(BioyondWorkstation): def _initialize_hardware_interface_from_config(self) -> Any: self._apply_env_api_config(self.bioyond_config) - if not self._has_required_api_config(self.bioyond_config): - raise RuntimeError( - "Bioyond RPC 客户端未初始化，且缺少 api_host/api_key。" - "请在前端节点 config 中配置 api_host、api_key，或在动作参数中传入 api_host、api_key。" - ) + missing = self._missing_api_config_keys(self.bioyond_config) + if missing: + api_host_present = "api_host" not in missing + api_key_present = "api_key" not in missing + lines = [ + f"无法调用 Bioyond RPC：缺少 {missing}（站点处于延迟初始化模式，构造时未提供完整 API 配置）。", + f" - 当前 api_host: {'已配置' if api_host_present else '<缺失>'}", + f" - 当前 api_key: {'已配置' if api_key_present else '<缺失>'}", + "请按以下任一方式补齐后重试：", + " 1) 在前端节点 config 中填入 api_host / api_key 并重新下发 graph；", + " 2) 在动作 goal 参数中直接传入 api_host / api_key（reset / start_experiment 已支持）；", + " 3) 设置环境变量后重启 edge：", + " BIOYOND_SIRNA_API_HOST 或 BIOYOND_SIRNA_EXP1_API_HOST", + " BIOYOND_SIRNA_API_KEY 或 BIOYOND_SIRNA_EXP1_API_KEY", + ] + raise RuntimeError("\n".join(lines)) from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import BioyondV1RPC self.hardware_interface = BioyondV1RPC(self.bioyond_config) return self.hardware_interface def _has_required_api_config(self, config: Dict[str, Any]) -> bool: - return not self._is_blank(config.get("api_host")) and not self._is_blank(config.get("api_key")) + return not self._missing_api_config_keys(config) def _apply_env_api_config(self, config: Dict[str, Any]) -> None: env_pairs = { @@ -1051,6 +1273,18 @@ class BioyondSirnaStation(BioyondWorkstation): payload["received_ready_signal"] = DEFAULT_READY_SIGNAL return payload + @staticmethod + def _as_manual_gate(value: Any) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"true", "1", "yes", "y", "on", "checked"}: + return True + if normalized in {"false", "0", "no", "n", "off", "unchecked", ""}: + return False + return bool(value) + def _resolve_start_experiment_info( self, submit_experiment_result: Optional[Dict[str, Any]], @@ -1327,6 +1561,108 @@ class BioyondSirnaStation(BioyondWorkstation): grouped[mode].append(mat) return grouped + def _build_manual_load_tables( + self, + materials_by_type: Dict[str, List[Dict[str, Any]]], + ) -> Dict[str, Dict[str, List[str]]]: + """Derive sibling-array tables per Bioyond material category. + + Output shape matches the production sibling-array handles on + ``submit_experiment_1``: ``manual_load_tables..`` where + ```` is ``material_name | material_code | location | quantity`` + and each value is a row-aligned ``List[str]``. + """ + tables: Dict[str, Dict[str, List[str]]] = {} + for mode in ("Sample", "Consumables", "Reagent"): + rows = (materials_by_type or {}).get(mode, []) or [] + tables[mode] = { + "material_name": [str(row.get("materialName") or "") for row in rows], + "material_code": [str(row.get("materialCode") or "") for row in rows], + "location": [ + str(row.get("locationShowName") or row.get("locationCode") or "") + for row in rows + ], + "quantity": [str(row.get("quantity") or "") for row in rows], + } + return tables + + @staticmethod + def _manual_load_resource_stub( + row: Dict[str, Any], + *, + name: str, + kind: str, + index: int, + ) -> Dict[str, Any]: + location_label = str(row.get("locationShowName") or row.get("locationCode") or "") + display_name = name or location_label or f"{kind}_{index + 1}" + stable_key = f"bioyond-sirna-manual-load:{kind}:{index}:{display_name}" + return { + "id": display_name, + "uuid": str(uuid5(NAMESPACE_URL, stable_key)), + "name": display_name, + "description": "Bioyond 手动装载临时资源", + "schema": {}, + "model": {}, + "icon": "", + "parent_uuid": None, + "type": "resource", + "class": "", + "config": { + "materialId": str(row.get("materialId") or ""), + "materialCode": str(row.get("materialCode") or ""), + "materialName": str(row.get("materialName") or ""), + "locationId": str(row.get("locationId") or ""), + "locationCode": str(row.get("locationCode") or ""), + "locationShowName": str(row.get("locationShowName") or ""), + }, + "data": { + "display_name": display_name, + "locationShowName": str(row.get("locationShowName") or ""), + "locationCode": str(row.get("locationCode") or ""), + }, + "extra": { + "bioyond_manual_load_kind": kind, + "bioyond_location_display": location_label, + }, + "machine_name": "", + } + + def _build_manual_load_probe( + self, + materials_by_type: Dict[str, List[Dict[str, Any]]], + ) -> Dict[str, List[Any]]: + """构造临时前端兼容的手动装载表格值。 + + 当前临时使用 ``coin_cell_code`` 表示物料名称，``mount_resource`` 表示库位。 + 因为 ``mount_resource`` 声明为 ``data_type="resource"``，这里要把 + ``locationShowName/locationCode`` 放到资源形状的 ``name`` 字段上，而不是 + 直接返回字符串列表。前端支持通用 action-input 表格后应替换为 Sirna 自有 key。 + """ + rows: List[Dict[str, Any]] = [] + for mode in ("Sample", "Consumables", "Reagent"): + rows.extend((materials_by_type or {}).get(mode, []) or []) + material_names = [ + str(row.get("materialName") or row.get("materialCode") or "") for row in rows + ] + location_names = [ + str(row.get("locationShowName") or row.get("locationCode") or "") for row in rows + ] + return { + "resource": material_names, + "coin_cell_code": material_names, + "mount_resource": [ + self._manual_load_resource_stub( + row, + name=location_names[index], + kind="location", + index=index, + ) + for index, row in enumerate(rows) + ], + "mount_resource_label": location_names, + } + def _register_materials_to_tree(self, material_records: List[Dict[str, Any]]) -> Dict[str, Any]: """Register Bioyond materials to UniLabOS resource tree after user confirmation.""" from unilabos.resources.bioyond.sirna_materials import get_material_class_by_type_code diff --git a/unilabos/resources/bioyond/sirna_materials.py b/unilabos/resources/bioyond/sirna_materials.py index 9ec44b06..0a32f172 100644 --- a/unilabos/resources/bioyond/sirna_materials.py +++ b/unilabos/resources/bioyond/sirna_materials.py @@ -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 diff --git a/unilabos/resources/bioyond/warehouses.py b/unilabos/resources/bioyond/warehouses.py index 98b405f4..5e9f975c 100644 --- a/unilabos/resources/bioyond/warehouses.py +++ b/unilabos/resources/bioyond/warehouses.py @@ -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") # ================ 反应站相关堆栈 ================ diff --git a/unilabos/resources/graphio.py b/unilabos/resources/graphio.py index b3ad7368..decfefb7 100644 --- a/unilabos/resources/graphio.py +++ b/unilabos/resources/graphio.py @@ -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