重构:完善配置加载机制与初始化逻辑

新增环境变量覆盖机制,增强配置灵活性

优化 bioyond_rpc.py 与 bioyond_cell_workstation.py 的初始化流程与结构

修正 station.py 工作流映射逻辑,确保正确性

提高代码可读性与模块间解耦程度
This commit is contained in:
calvincao
2025-10-22 16:13:36 +08:00
parent e70c545ec8
commit de7c80c3c2
4 changed files with 44 additions and 18 deletions

View File

@@ -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_token0928未启用
**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())

View File

@@ -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()

View File

@@ -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",
}
# 工作流映射配置

View File

@@ -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):