From de7c80c3c2d8eb7b262b8e597173edef8b9691df Mon Sep 17 00:00:00 2001 From: calvincao Date: Wed, 22 Oct 2025 16:13:36 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=EF=BC=9A=E5=AE=8C=E5=96=84?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=8A=A0=E8=BD=BD=E6=9C=BA=E5=88=B6=E4=B8=8E?= =?UTF-8?q?=E5=88=9D=E5=A7=8B=E5=8C=96=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增环境变量覆盖机制，增强配置灵活性 优化 bioyond_rpc.py 与 bioyond_cell_workstation.py 的初始化流程与结构 修正 station.py 工作流映射逻辑，确保正确性 提高代码可读性与模块间解耦程度 --- .../bioyond_cell/bioyond_cell_workstation.py | 27 +++++++++++-------- .../workstation/bioyond_studio/bioyond_rpc.py | 4 +-- .../workstation/bioyond_studio/config.py | 27 ++++++++++++++++--- .../workstation/bioyond_studio/station.py | 4 +-- 4 files changed, 44 insertions(+), 18 deletions(-) diff --git a/unilabos/devices/workstation/bioyond_studio/bioyond_cell/bioyond_cell_workstation.py b/unilabos/devices/workstation/bioyond_studio/bioyond_cell/bioyond_cell_workstation.py index 7ed0e9f6..78b635ce 100644 --- a/unilabos/devices/workstation/bioyond_studio/bioyond_cell/bioyond_cell_workstation.py +++ b/unilabos/devices/workstation/bioyond_studio/bioyond_cell/bioyond_cell_workstation.py @@ -9,9 +9,14 @@ import time from datetime import datetime, timedelta import re import threading +import os from urllib3 import response from unilabos.devices.workstation.workstation_base import WorkstationBase +from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation +from unilabos.devices.workstation.bioyond_studio.config import ( + BIOYOND_FULL_CONFIG, WORKFLOW_MAPPINGS, MATERIAL_TYPE_MAPPINGS, WAREHOUSE_MAPPING +) from unilabos.devices.workstation.workstation_http_service import WorkstationHTTPService from unilabos.utils.log import logger @@ -22,7 +27,7 @@ def _iso_local_now_ms() -> str: return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(dt.microsecond/1000):03d}Z" -class BioyondCellWorkstation(WorkstationBase): +class BioyondCellWorkstation(BioyondWorkstation): """ 集成 Bioyond LIMS 的工作站示例， 覆盖：入库(2.17/2.18) → 新建实验(2.14) → 启动调度(2.7) → @@ -37,20 +42,18 @@ class BioyondCellWorkstation(WorkstationBase): *args, **kwargs, ): + # 使用统一配置，支持自定义覆盖 self.bioyond_config = bioyond_config or { - "base_url": "http://172.16.11.219:44388", - "api_key": "8A819E5C", - "timeout": 30, - "report_token": "CHANGE_ME_TOKEN", - "HTTP_host": "172.21.33.126", - "HTTP_port": 8080, - "debug_mode": False - } # report_token ：unilab自己的令牌report_token（0928未启用） + **BIOYOND_FULL_CONFIG, # 从 config.py 加载完整配置 + "workflow_mappings": WORKFLOW_MAPPINGS, + "material_type_mappings": MATERIAL_TYPE_MAPPINGS, + "warehouse_mapping": WAREHOUSE_MAPPING + } self.debug_mode = self.bioyond_config["debug_mode"] self.http_service_started = False deck = kwargs.pop("deck", None) self.device_id = kwargs.pop("device_id", "bioyond_cell_workstation") - super().__init__(deck=deck, station_resource=station_resource, *args, **kwargs) + super().__init__(bioyond_config=self.bioyond_config, deck=deck, station_resource=station_resource, *args, **kwargs) # 步骤通量任务通知铃 self._pending_events: dict[str, threading.Event] = {} logger.info(f"Bioyond工作站初始化完成 (debug_mode={self.debug_mode})") @@ -793,7 +796,9 @@ if __name__ == "__main__": logger.info(ws.scheduler_start()) logger.info(ws.auto_feeding4to3()) - logger.info(ws.create_orders(r"unilabos\devices\workstation\bioyond_studio\bioyond_cell\2025092701.xlsx")) + # 使用正斜杠或 Path 对象来指定文件路径 + excel_path = Path("unilabos/devices/workstation/bioyond_studio/bioyond_cell/2025092701.xlsx") + logger.info(ws.create_orders(excel_path)) logger.info(ws.transfer_3_to_2_to_1()) logger.info(ws.transfer_1_to_2()) diff --git a/unilabos/devices/workstation/bioyond_studio/bioyond_rpc.py b/unilabos/devices/workstation/bioyond_studio/bioyond_rpc.py index 5aef45ba..ef6ff77f 100644 --- a/unilabos/devices/workstation/bioyond_studio/bioyond_rpc.py +++ b/unilabos/devices/workstation/bioyond_studio/bioyond_rpc.py @@ -47,8 +47,8 @@ class BioyondV1RPC(BaseRequest): super().__init__() print("开始初始化 BioyondV1RPC") self.config = config - self.api_key = config["8A819E5C"] - self.host = config["http://172.16.11.219:44388"] + self.api_key = config.get("api_key", "") + self.host = config.get("api_host", "") or config.get("base_url", "") self._logger = SimpleLogger() self.material_cache = {} self._load_material_cache() diff --git a/unilabos/devices/workstation/bioyond_studio/config.py b/unilabos/devices/workstation/bioyond_studio/config.py index 52ae82e6..c73abd35 100644 --- a/unilabos/devices/workstation/bioyond_studio/config.py +++ b/unilabos/devices/workstation/bioyond_studio/config.py @@ -2,11 +2,32 @@ """ 配置文件 - 包含所有配置信息和映射关系 """ +import os -# API配置 +# ==================== API 基础配置 ==================== +# 支持通过环境变量覆盖默认值 API_CONFIG = { - "api_key": "", - "api_host": "" + "api_key": os.getenv("BIOYOND_API_KEY", "8A819E5C"), + "api_host": os.getenv("BIOYOND_API_HOST", "http://172.16.11.219:44388"), +} + +# ==================== 完整的 Bioyond 配置 ==================== +# BioyondCellWorkstation 默认配置（包含所有必需参数） +BIOYOND_FULL_CONFIG = { + # API 连接配置 + "base_url": os.getenv("BIOYOND_API_HOST", "http://172.16.11.219:44388"), + "api_key": os.getenv("BIOYOND_API_KEY", "8A819E5C"), + "timeout": int(os.getenv("BIOYOND_TIMEOUT", "30")), + + # 报送配置 + "report_token": os.getenv("BIOYOND_REPORT_TOKEN", "CHANGE_ME_TOKEN"), + + # HTTP 服务配置 + "HTTP_host": os.getenv("BIOYOND_HTTP_HOST", "0.0.0.0"), # 0.0.0.0 绑定所有网络接口 + "HTTP_port": int(os.getenv("BIOYOND_HTTP_PORT", "8080")), + + # 调试模式 + "debug_mode": os.getenv("BIOYOND_DEBUG_MODE", "False").lower() == "true", } # 工作流映射配置 diff --git a/unilabos/devices/workstation/bioyond_studio/station.py b/unilabos/devices/workstation/bioyond_studio/station.py index 3c596a2e..795b7408 100644 --- a/unilabos/devices/workstation/bioyond_studio/station.py +++ b/unilabos/devices/workstation/bioyond_studio/station.py @@ -165,8 +165,8 @@ class BioyondWorkstation(WorkstationBase): self.workflow_sequence = [] self.pending_task_params = [] - if "workflow_mappings" in bioyond_config: - self._set_workflow_mappings(bioyond_config["workflow_mappings"]) + if self.bioyond_config and "workflow_mappings" in self.bioyond_config: + self._set_workflow_mappings(self.bioyond_config["workflow_mappings"]) logger.info(f"Bioyond工作站初始化完成") def post_init(self, ros_node: ROS2WorkstationNode):