open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2025-12-17 21:11:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

xinyu1024修改

Browse Source
This commit is contained in:
lixinyu1011
2025-10-24 11:37:36 +08:00
parent fd73731130
commit 7206e42bf1
7 changed files with 282 additions and 556 deletions
Download Patch File Download Diff File Expand all files Collapse all files

582
unilabos/devices/workstation/bioyond_studio/bioyond_cell/bioyond_cell_workstation.py
View File

@@ -9,14 +9,12 @@ import time
from datetime import datetime, timedelta from datetime import datetime, timedelta
import re import re
import threading import threading
import os import json
import socket
from urllib3 import response from urllib3 import response
from unilabos.devices.workstation.workstation_base import WorkstationBase from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation, BioyondResourceSynchronizer
from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation
from unilabos.devices.workstation.bioyond_studio.config import ( from unilabos.devices.workstation.bioyond_studio.config import (
BIOYOND_FULL_CONFIG, WORKFLOW_MAPPINGS, MATERIAL_TYPE_MAPPINGS, WAREHOUSE_MAPPING API_CONFIG, MATERIAL_TYPE_MAPPINGS, WAREHOUSE_MAPPING, SOLID_LIQUID_MAPPINGS
) )
from unilabos.devices.workstation.workstation_http_service import WorkstationHTTPService from unilabos.devices.workstation.workstation_http_service import WorkstationHTTPService
from unilabos.utils.log import logger from unilabos.utils.log import logger
@@ -43,32 +41,52 @@ class BioyondCellWorkstation(BioyondWorkstation):
*args, **kwargs, *args, **kwargs,
): ):
# 使用统一配置，支持自定义覆盖 # 使用统一配置，支持自定义覆盖, 从 config.py 加载完整配置
self.bioyond_config = bioyond_config or { self.bioyond_config = bioyond_config or {
**BIOYOND_FULL_CONFIG, # 从 config.py 加载完整配置 **API_CONFIG,
"workflow_mappings": WORKFLOW_MAPPINGS,
"material_type_mappings": MATERIAL_TYPE_MAPPINGS, "material_type_mappings": MATERIAL_TYPE_MAPPINGS,
"warehouse_mapping": WAREHOUSE_MAPPING "warehouse_mapping": WAREHOUSE_MAPPING
} }
# "material_type_mappings": MATERIAL_TYPE_MAPPINGS
# "warehouse_mapping": WAREHOUSE_MAPPING
print(self.bioyond_config)
self.debug_mode = self.bioyond_config["debug_mode"] self.debug_mode = self.bioyond_config["debug_mode"]
self.http_service_started = False self.http_service_started = self.debug_mode
deck = kwargs.pop("deck", None) deck = kwargs.pop("deck", None)
self.device_id = kwargs.pop("device_id", "bioyond_cell_workstation") self.device_id = kwargs.pop("device_id", "bioyond_cell_workstation")
super().__init__(bioyond_config=self.bioyond_config, deck=deck, station_resource=station_resource, *args, **kwargs) super().__init__(bioyond_config=self.bioyond_config, deck=deck, station_resource=station_resource, *args, **kwargs)
# 步骤通量任务通知铃 self.update_push_ip() #直接修改奔耀端的报送ip地址
self._pending_events: dict[str, threading.Event] = {} logger.info("已更新奔耀端推送 IP 地址")
# 启动 HTTP 服务线程
t = threading.Thread(target=self._start_http_service, daemon=True, name="unilab_http")
t.start()
logger.info("HTTP 服务线程已启动")
# 等到任务报送成功
self.order_finish_event = threading.Event()
self.last_order_status = None
self.last_order_code = None
logger.info(f"Bioyond工作站初始化完成 (debug_mode={self.debug_mode})") logger.info(f"Bioyond工作站初始化完成 (debug_mode={self.debug_mode})")
# 实例化并在后台线程启动 HTTP 报送服务 def _start_http_service(self):
self.order_status = {} # 记录任务完成情况，用于接受bioyond post信息和反馈信息，尤其用于硬件查询和物料信息变化 """启动 HTTP 服务"""
host = self.bioyond_config.get("HTTP_host", "")
port = self.bioyond_config.get("HTTP_port", None)
try: try:
logger.info("准备开始unilab_HTTP后台线程") self.service = WorkstationHTTPService(self, host=host, port=port)
t = threading.Thread(target=self._start_http_service, daemon=True, name="unilab_http") self.service.start()
t.start() self.http_service_started = True
logger.info(f"WorkstationHTTPService 成功启动: {host}:{port}")
while True:
time.sleep(1) #一直挂着，直到进程退出
except Exception as e: except Exception as e:
logger.error(f"unilab-HTTP后台线程启动失败: {e}") self.http_service_started = False
logger.error(f"启动 WorkstationHTTPService 失败: {e}", exc_info=True)
# http报送服务
# http报送服务，返回数据部分
def process_step_finish_report(self, report_request): def process_step_finish_report(self, report_request):
stepId = report_request.data.get("stepId") stepId = report_request.data.get("stepId")
logger.info(f"步骤完成: stepId: {stepId}, stepName:{report_request.data.get('stepName')}") logger.info(f"步骤完成: stepId: {stepId}, stepName:{report_request.data.get('stepName')}")
@@ -82,147 +100,65 @@ class BioyondCellWorkstation(BioyondWorkstation):
order_code = report_request.data.get("orderCode") order_code = report_request.data.get("orderCode")
status = report_request.data.get("status") status = report_request.data.get("status")
logger.info(f"report_request: {report_request}") logger.info(f"report_request: {report_request}")
logger.info(f"任务完成: {order_code}, status={status}") logger.info(f"任务完成: {order_code}, status={status}")
# 记录订单状态码 # 保存完整报文
if order_code: self.last_order_report = report_request.data
self.order_status[order_code] = status # 如果是当前等待的订单，触发事件
if self.last_order_code == order_code:
self.order_finish_event.set()
self._set_pending_event(order_code)
return {"status": "received"} return {"status": "received"}
def _set_pending_event(self, taskname: Optional[str]) -> None: def wait_for_order_finish(self, order_code: str, timeout: int = 1800) -> Dict[str, Any]:
if not taskname: """
return 等待指定 orderCode 的 /report/order_finish 报送。
event = self._pending_events.get(taskname) Args:
if event is None: order_code: 任务编号
event = threading.Event() timeout: 超时时间（秒）
self._pending_events[taskname] = event Returns:
event.set() 完整的报送数据 + 状态判断结果
"""
def _wait_for_order_completion(self, order_code: Optional[str], timeout: int = 600) -> bool:
if not order_code: if not order_code:
logger.warning("无法等待任务完成：order_code 为空") logger.error("wait_for_order_finish() 被调用，但 order_code 为空！")
return False return {"status": "error", "message": "empty order_code"}
event = self._pending_events.get(order_code)
if event is None:
event = threading.Event()
self._pending_events[order_code] = event
elif event.is_set():
logger.info(f"任务 {order_code} 在等待之前已完成")
self._pending_events.pop(order_code, None)
return True
logger.info(f"等待任务 {order_code} 完成 (timeout={timeout}s)")
finished = event.wait(timeout)
if not finished:
logger.warning(f"等待任务 {order_code} 完成超时（{timeout}s）")
self._pending_events.pop(order_code, None)
return finished
def _wait_for_response_orders(self, response: Dict[str, Any], context: str, timeout: int = 600) -> None: self.last_order_code = order_code
order_codes = self._extract_order_codes(response) self.last_order_report = None
if not order_codes: self.order_finish_event.clear()
logger.warning(f"{context} 响应中未找到 orderCode，无法跟踪任务完成")
return
for code in order_codes:
finished = self._wait_for_order_completion(code, timeout=timeout)
if finished:
# 检查订单返回码是否为30（正常完成）
status = self.order_status.get(code)
if status == 30 or status == "30":
logger.info(f"订单 {code} 成功完成，状态码: {status}")
else:
logger.warning(f"订单 {code} 完成但状态码异常: {status} (期望: 30, -11=异常停止, -12=人工停止)")
# 清理状态记录
self.order_status.pop(code, None)
else:
logger.error(f"订单 {code} 等待超时，未收到完成通知")
@staticmethod logger.info(f"等待任务完成报送: orderCode={order_code} (timeout={timeout}s)")
def _extract_order_codes(response: Dict[str, Any]) -> List[str]:
order_codes: List[str] = []
if not isinstance(response, dict):
return order_codes
data = response.get("data")
keys = ["orderCode", "order_code", "orderId", "order_id"]
if isinstance(data, dict):
for key in keys:
if key in data and data[key]:
order_codes.append(str(data[key]))
if not order_codes and "orders" in data and isinstance(data["orders"], list):
for order in data["orders"]:
if isinstance(order, dict):
for key in keys:
if key in order and order[key]:
order_codes.append(str(order[key]))
elif isinstance(data, list):
for item in data:
if isinstance(item, dict):
for key in keys:
if key in item and item[key]:
order_codes.append(str(item[key]))
elif isinstance(data, str):
if data:
order_codes.append(data)
meta = response.get("orderCode")
if meta:
order_codes.append(str(meta))
# 去重
seen = set()
unique_codes: List[str] = []
for code in order_codes:
if code not in seen:
seen.add(code)
unique_codes.append(code)
return unique_codes
if not self.order_finish_event.wait(timeout=timeout):
logger.error(f"等待任务超时: orderCode={order_code}")
return {"status": "timeout", "orderCode": order_code}
def _start_http_service(self, host: Optional[str] = None, port: Optional[int] = None) -> None: # 报送数据匹配验证
host = host or self.bioyond_config.get("HTTP_host", "") report = self.last_order_report or {}
port = port or self.bioyond_config.get("HTTP_port", ) report_code = report.get("orderCode")
status = str(report.get("status", ""))
logger.info("准备开始unilab_HTTP服务") if report_code != order_code:
try: logger.warning(f"收到的报送 orderCode 不匹配: {report_code} ≠ {order_code}")
self.service = WorkstationHTTPService(self, host=host, port=port) return {"status": "mismatch", "report": report}
logger.info("WorkstationHTTPService 实例化完成")
self.service.start()
self.http_service_started = True
logger.info(f"WorkstationHTTPService成功启动: {host}:{port}")
# 启动成功后，上报本机推送地址（3.36） if status == "30":
try: logger.info(f"任务成功完成 (orderCode={order_code})")
# 优先使用配置中的 report_ip return {"status": "success", "report": report}
report_ip = self.bioyond_config.get("report_ip", "").strip() elif status == "-11":
logger.error(f"任务异常停止 (orderCode={order_code})")
return {"status": "abnormal_stop", "report": report}
elif status == "-12":
logger.warning(f"任务人工停止 (orderCode={order_code})")
return {"status": "manual_stop", "report": report}
else:
logger.warning(f"任务未知状态 ({status}) (orderCode={order_code})")
return {"status": f"unknown_{status}", "report": report}
# 如果配置中没有指定 report_ip，且监听地址是 0.0.0.0，则自动检测
if not report_ip and host in ("0.0.0.0", ""):
# 从 Bioyond 配置中提取服务器地址
bioyond_server = self.bioyond_config.get("base_url", "")
if bioyond_server:
import urllib.parse
parsed = urllib.parse.urlparse(bioyond_server)
elif not report_ip:
# 如果没有配置 report_ip，使用监听地址
report_ip = host
r = self.update_push_ip(report_ip, port)
logger.info(f"向 Bioyond 报送推送地址: {report_ip}:{port}, 结果: {r}")
except Exception as e:
logger.warning(f"调用更新推送IP接口失败: {e}")
#一直挂着，直到进程退出
while True:
time.sleep(1)
except Exception as e:
self.http_service_started = False # 调试用
logger.error(f"启动WorkstationHTTPService失败: {e}", exc_info=True)
# -------------------- 基础HTTP封装 -------------------- # -------------------- 基础HTTP封装 --------------------
def _url(self, path: str) -> str: def _url(self, path: str) -> str:
return f"{self.bioyond_config['api_host'].rstrip('/')}/{path.lstrip('/')}"
return f"{self.bioyond_config['base_url'].rstrip('/')}/{path.lstrip('/')}"
def _post_lims(self, path: str, data: Optional[Any] = None) -> Dict[str, Any]: def _post_lims(self, path: str, data: Optional[Any] = None) -> Dict[str, Any]:
"""LIMS API：大多数接口用 {apiKey/requestTime,data} 包装""" """LIMS API：大多数接口用 {apiKey/requestTime,data} 包装"""
@@ -236,9 +172,11 @@ class BioyondCellWorkstation(BioyondWorkstation):
if self.debug_mode: if self.debug_mode:
# 模拟返回，不发真实请求 # 模拟返回，不发真实请求
logger.info(f"[DEBUG] POST {path} with payload={payload}") logger.info(f"[DEBUG] POST {path} with payload={payload}")
return {"debug": True, "url": self._url(path), "payload": payload, "status": "ok"} return {"debug": True, "url": self._url(path), "payload": payload, "status": "ok"}
try: try:
logger.info(json.dumps(payload, ensure_ascii=False))
response = requests.post( response = requests.post(
self._url(path), self._url(path),
json=payload, json=payload,
@@ -248,7 +186,7 @@ class BioyondCellWorkstation(BioyondWorkstation):
response.raise_for_status() response.raise_for_status()
return response.json() return response.json()
except Exception as e: except Exception as e:
logger.info(f"{self.bioyond_config['base_url'].rstrip('/')}/{path.lstrip('/')}") logger.info(f"{self.bioyond_config['api_host'].rstrip('/')}/{path.lstrip('/')}")
logger.error(f"POST {path} 失败: {e}") logger.error(f"POST {path} 失败: {e}")
return {"error": str(e)} return {"error": str(e)}
@@ -263,7 +201,7 @@ class BioyondCellWorkstation(BioyondWorkstation):
if self.debug_mode: if self.debug_mode:
logger.info(f"[DEBUG] PUT {path} with payload={payload}") logger.info(f"[DEBUG] PUT {path} with payload={payload}")
return {"debug": True, "url": self._url(path), "payload": payload, "status": "ok"} return {"debug_mode": True, "url": self._url(path), "payload": payload, "status": "ok"}
try: try:
response = requests.put( response = requests.put(
@@ -275,7 +213,7 @@ class BioyondCellWorkstation(BioyondWorkstation):
response.raise_for_status() response.raise_for_status()
return response.json() return response.json()
except Exception as e: except Exception as e:
logger.info(f"{self.bioyond_config['base_url'].rstrip('/')}/{path.lstrip('/')}") logger.info(f"{self.bioyond_config['api_host'].rstrip('/')}/{path.lstrip('/')}")
logger.error(f"PUT {path} 失败: {e}") logger.error(f"PUT {path} 失败: {e}")
return {"error": str(e)} return {"error": str(e)}
@@ -310,11 +248,10 @@ class BioyondCellWorkstation(BioyondWorkstation):
return self._post_lims("/api/lims/storage/batch-inbound", items) return self._post_lims("/api/lims/storage/batch-inbound", items)
def auto_feeding4to3( def auto_feeding4to3(
self, self,
# ★ 修改点：默认模板路径 # ★ 修改点：默认模板路径
xlsx_path: Optional[str] = "/Users/calvincao/Desktop/work/uni-lab-all/Uni-Lab-OS/unilabos/devices/workstation/bioyond_studio/bioyond_cell/样品导入模板.xlsx", xlsx_path: Optional[str] = "unilabos\\devices\\workstation\\bioyond_studio\\bioyond_cell\\样品导入模板.xlsx",
# ---------------- WH4 - 加样头面 (Z=1, 12个点位) ---------------- # ---------------- WH4 - 加样头面 (Z=1, 12个点位) ----------------
WH4_x1_y1_z1_1_materialName: str = "", WH4_x1_y1_z1_1_quantity: float = 0.0, WH4_x1_y1_z1_1_materialName: str = "", WH4_x1_y1_z1_1_quantity: float = 0.0,
WH4_x2_y1_z1_2_materialName: str = "", WH4_x2_y1_z1_2_quantity: float = 0.0, WH4_x2_y1_z1_2_materialName: str = "", WH4_x2_y1_z1_2_quantity: float = 0.0,
@@ -443,9 +380,15 @@ class BioyondCellWorkstation(BioyondWorkstation):
return {"code": 0, "message": "no valid items", "data": []} return {"code": 0, "message": "no valid items", "data": []}
logger.info(items) logger.info(items)
response = self._post_lims("/api/lims/order/auto-feeding4to3", items) response = self._post_lims("/api/lims/order/auto-feeding4to3", items)
self._wait_for_response_orders(response, "auto_feeding4to3")
return response
# 等待任务报送成功
order_code = response.get("data", {}).get("orderCode")
if not order_code:
logger.error("上料任务未返回有效 orderCode！")
return response
# 等待完成报送
result = self.wait_for_order_finish(order_code)
return result
@@ -681,31 +624,17 @@ class BioyondCellWorkstation(BioyondWorkstation):
} }
orders.append(order_data) orders.append(order_data)
# print(orders)
while True:
time.sleep(5)
response = self._post_lims("/api/lims/order/orders", orders)
if response.get("data", []):
break
logger.info(f"等待配液实验创建完成")
response = self._post_lims("/api/lims/order/orders", orders)
print(response)
# self.order_status[response["data"]["orderCode"]] = "running" # 等待任务报送成功
order_code = response.get("data", {}).get("orderCode")
# while True: if not order_code:
# time.sleep(5) logger.error("上料任务未返回有效 orderCode！")
# if self.order_status.get(response["data"]["orderCode"], None) == "finished": return response
# logger.info(f"配液实验已完成 ，即将执行 3-2-1 转运") # 等待完成报送
# break result = self.wait_for_order_finish(order_code)
# logger.info(f"等待配液实验完成") return result
# self.transfer_3_to_2_to_1()
# self.wait_for_transfer_task()
# logger.info(f"3-2-1 转运完成，返回结果")
# return r321
self._wait_for_response_orders(response, "create_orders", timeout=1800)
return response
# 2.7 启动调度 # 2.7 启动调度
def scheduler_start(self) -> Dict[str, Any]: def scheduler_start(self) -> Dict[str, Any]:
@@ -726,6 +655,13 @@ class BioyondCellWorkstation(BioyondWorkstation):
请求体只包含 apiKey 和 requestTime 请求体只包含 apiKey 和 requestTime
""" """
return self._post_lims("/api/lims/scheduler/continue") return self._post_lims("/api/lims/scheduler/continue")
def scheduler_reset(self) -> Dict[str, Any]:
"""
复位调度 (2.11)
请求体只包含 apiKey 和 requestTime
"""
return self._post_lims("/api/lims/scheduler/reset")
# 2.24 物料变更推送 # 2.24 物料变更推送
def report_material_change(self, material_obj: Dict[str, Any]) -> Dict[str, Any]: def report_material_change(self, material_obj: Dict[str, Any]) -> Dict[str, Any]:
@@ -744,7 +680,16 @@ class BioyondCellWorkstation(BioyondWorkstation):
} }
if source_wh_id: if source_wh_id:
payload["sourceWHID"] = source_wh_id payload["sourceWHID"] = source_wh_id
return self._post_lims("/api/lims/order/transfer-task3To2To1", payload)
response = self._post_lims("/api/lims/order/transfer-task3To2To1", payload)
# 等待任务报送成功
order_code = response.get("data", {}).get("orderCode")
if not order_code:
logger.error("上料任务未返回有效 orderCode！")
return response
# 等待完成报送
result = self.wait_for_order_finish(order_code)
return result
# 3.35 1→2 物料转运 # 3.35 1→2 物料转运
def transfer_1_to_2(self) -> Dict[str, Any]: def transfer_1_to_2(self) -> Dict[str, Any]:
@@ -753,7 +698,15 @@ class BioyondCellWorkstation(BioyondWorkstation):
URL: /api/lims/order/transfer-task1To2 URL: /api/lims/order/transfer-task1To2
只需要 apiKey 和 requestTime 只需要 apiKey 和 requestTime
""" """
return self._post_lims("/api/lims/order/transfer-task1To2") response = self._post_lims("/api/lims/order/transfer-task1To2")
# 等待任务报送成功
order_code = response.get("data", {}).get("orderCode")
if not order_code:
logger.error("上料任务未返回有效 orderCode！")
return response
# 等待完成报送
result = self.wait_for_order_finish(order_code)
return result
# 2.5 批量查询实验报告(post过滤关键字查询) # 2.5 批量查询实验报告(post过滤关键字查询)
def order_list_v2(self, def order_list_v2(self,
@@ -825,241 +778,38 @@ class BioyondCellWorkstation(BioyondWorkstation):
logger.warning("超时未找到成功的物料转移任务") logger.warning("超时未找到成功的物料转移任务")
return False return False
def create_solid_materials(self, material_names: List[str], type_id: str = "3a190ca0-b2f6-9aeb-8067-547e72c11469") -> List[Dict[str, Any]]: def create_materials(self, mappings: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
""" """
批量创建固体物料 将 SOLID_LIQUID_MAPPINGS 中的所有物料逐个 POST 到 /api/lims/storage/material
Args:
material_names: 物料名称列表
type_id: 物料类型ID（默认为固体物料类型）
Returns:
创建的物料列表，每个元素包含物料信息和ID
""" """
created_materials = [] results = []
total = len(material_names)
for i, name in enumerate(material_names, 1): for name, data in mappings.items():
# 根据接口文档构建完整的请求体 data = {
material_data = { "typeId": data["typeId"],
"typeId": type_id, "code": data.get("code", ""),
"name": name, "barCode": data.get("barCode", ""),
"unit": "g", # 添加单位 "name": data["name"],
"quantity": 1, # 添加数量（默认1） "unit": data.get("unit", "g"),
"parameters": "" # 参数字段（空字符串表示无参数） "parameters": data.get("parameters", ""),
"quantity": data.get("quantity", ""),
"warningQuantity": data.get("warningQuantity", ""),
"details": data.get("details", [])
} }
logger.info(f"正在创建第 {i}/{total} 个固体物料: {name}") logger.info(f"正在 POST 创建物料: {name}")
result = self._post_lims("/api/lims/storage/material", material_data) try:
# ✅ 真正执行 POST
result = self._post_lims("/api/lims/storage/material", data)
logger.info(f"响应: {result}")
except Exception as e:
logger.error(f"✗ 创建物料失败: {name}, 错误: {e}")
results.append({name: {"error": str(e)}})
time.sleep(0.3) # 避免请求过快
return results
if result and result.get("code") == 1:
# data 字段可能是字符串（物料ID）或字典（包含id字段）
data = result.get("data")
if isinstance(data, str):
# data 直接是物料ID字符串
material_id = data
elif isinstance(data, dict):
# data 是字典，包含id字段
material_id = data.get("id")
else:
material_id = None
if material_id:
created_materials.append({
"name": name,
"materialId": material_id,
"typeId": type_id
})
logger.info(f"✓ 成功创建物料: {name}, ID: {material_id}")
else:
logger.error(f"✗ 创建物料失败: {name}, 未返回ID")
logger.error(f" 响应数据: {result}")
else:
error_msg = result.get("error") or result.get("message", "未知错误")
logger.error(f"✗ 创建物料失败: {name}")
logger.error(f" 错误信息: {error_msg}")
logger.error(f" 完整响应: {result}")
# 避免请求过快
time.sleep(0.3)
logger.info(f"物料创建完成，成功创建 {len(created_materials)}/{total} 个固体物料")
return created_materials
def create_and_inbound_materials_from_csv(
self,
csv_path: str = "solid_materials.csv",
type_id: str = "3a190ca0-b2f6-9aeb-8067-547e72c11469",
warehouse_name: str = "粉末加样头堆栈"
) -> Dict[str, Any]:
"""
从CSV文件读取物料列表，创建物料并批量入库到指定堆栈
Args:
csv_path: CSV文件路径
type_id: 物料类型ID（默认为固体物料类型）
warehouse_name: 仓库名称（默认为"粉末加样头堆栈"）
Returns:
包含执行结果的字典
"""
logger.info("=" * 60)
logger.info(f"开始执行：从CSV读取物料列表并批量创建入库到 {warehouse_name}")
logger.info("=" * 60)
# 从配置中获取位置ID列表
warehouse_mapping = self.bioyond_config.get("warehouse_mapping", WAREHOUSE_MAPPING)
if warehouse_name not in warehouse_mapping:
error_msg = f"配置中未找到仓库: {warehouse_name}"
logger.error(error_msg)
logger.info(f"可用的仓库: {list(warehouse_mapping.keys())}")
return {"success": False, "error": error_msg}
warehouse_config = warehouse_mapping[warehouse_name]
site_uuids = warehouse_config.get("site_uuids", {})
if not site_uuids:
error_msg = f"仓库 {warehouse_name} 没有配置位置"
logger.error(error_msg)
return {"success": False, "error": error_msg}
# 按顺序获取位置ID（A01, B01, C01...）
all_location_ids = []
position_names = []
for key in sorted(site_uuids.keys()):
all_location_ids.append(site_uuids[key])
position_names.append(key)
logger.info(f"✓ 从配置文件加载 {len(all_location_ids)} 个位置")
logger.info(f" 仓库: {warehouse_name}")
logger.info(f" 位置范围: {position_names[0]} ~ {position_names[-1]}")
# 读取CSV文件
csv_file_path = Path(csv_path)
material_names = []
try:
df_materials = pd.read_csv(csv_file_path)
if 'material_name' in df_materials.columns:
material_names = df_materials['material_name'].dropna().astype(str).str.strip().tolist()
logger.info(f"✓ 成功从CSV文件读取 {len(material_names)} 个物料名称")
logger.info(f" 文件路径: {csv_file_path}")
else:
logger.error(f"✗ CSV文件缺少 'material_name' 列")
return {"success": False, "error": "CSV文件缺少 'material_name' 列"}
except FileNotFoundError:
logger.error(f"✗ 未找到CSV文件: {csv_file_path}")
logger.info("请创建CSV文件，格式：")
logger.info(" material_name")
logger.info(" LiPF6")
logger.info(" LiDFOB")
logger.info(" ...")
return {"success": False, "error": f"未找到CSV文件: {csv_file_path}"}
except Exception as e:
logger.error(f"✗ 读取CSV文件失败: {e}")
return {"success": False, "error": f"读取CSV文件失败: {e}"}
if not material_names:
logger.error("CSV文件中没有有效的物料名称")
return {"success": False, "error": "CSV文件中没有有效的物料名称"}
# 检查物料数量
if len(material_names) > len(all_location_ids):
logger.warning(f"物料数量({len(material_names)})超过可用位置数量({len(all_location_ids)})！")
logger.warning(f"将仅创建前 {len(all_location_ids)} 个物料")
material_names = material_names[:len(all_location_ids)]
# 准备位置信息
location_ids = all_location_ids[:len(material_names)]
selected_positions = position_names[:len(material_names)]
# 步骤1: 创建固体物料
logger.info(f"\n【步骤1/2】创建 {len(material_names)} 个固体物料...")
logger.info(f"物料类型ID: {type_id}")
logger.info(f"物料列表: {', '.join(material_names)}")
created_materials = self.create_solid_materials(
material_names=material_names,
type_id=type_id
)
if len(created_materials) != len(material_names):
logger.warning(f"创建的物料数量({len(created_materials)})与计划数量({len(material_names)})不匹配！")
logger.warning("将仅对成功创建的物料进行入库操作")
if not created_materials:
logger.error("没有成功创建任何物料")
return {"success": False, "error": "没有成功创建任何物料"}
# 步骤2: 批量入库到指定位置
logger.info(f"\n【步骤2/2】批量入库物料到 {warehouse_name}...")
inbound_items = []
for idx, material in enumerate(created_materials):
if idx < len(location_ids):
inbound_items.append({
"materialId": material["materialId"],
"locationId": location_ids[idx]
})
logger.info(f" - {material['name']} (ID: {material['materialId'][:8]}...) → 位置 {selected_positions[idx]}")
logger.info(f"\n正在执行批量入库，共 {len(inbound_items)} 条记录...")
result = self.storage_batch_inbound(inbound_items)
inbound_success = result.get("code") == 1
if inbound_success:
logger.info(f"✓ 批量入库成功！")
logger.info(f" 响应数据: {result.get('data', {})}")
# 步骤3: 同步物料数据
logger.info(f"\n【步骤3/3】同步物料数据到系统...")
if hasattr(self, 'resource_synchronizer') and self.resource_synchronizer:
try:
# 尝试同步不同类型的物料
# typeMode: 0=耗材, 1=样品, 2=试剂
sync_success = False
for type_mode in [0, 1, 2]:
try:
logger.info(f" 尝试同步 typeMode={type_mode} 的物料...")
bioyond_data = self.hardware_interface.stock_material(
f'{{"typeMode": {type_mode}, "includeDetail": true}}'
)
if bioyond_data:
logger.info(f" ✓ 获取到 {len(bioyond_data) if isinstance(bioyond_data, list) else 1} 条物料数据")
sync_success = True
except Exception as e:
logger.debug(f" typeMode={type_mode} 同步失败: {e}")
continue
if sync_success:
logger.info(f"✓ 物料数据同步完成")
else:
logger.warning(f"⚠ 物料数据同步未获取到数据（这是正常的，新创建的物料可能需要时间才能查询到）")
except Exception as e:
logger.warning(f"⚠ 物料数据同步出错: {e}")
logger.info(f" 提示：新创建的物料已成功入库，同步失败不影响使用")
else:
logger.warning("⚠ 资源同步器未初始化，跳过同步")
else:
logger.error(f"✗ 批量入库失败！")
logger.error(f" 响应: {result}")
logger.info("\n" + "=" * 60)
logger.info("固体物料创建和入库流程完成")
logger.info("=" * 60 + "\n")
return {
"success": inbound_success,
"created_materials": created_materials,
"inbound_result": result,
"total_created": len(created_materials),
"total_inbound": len(inbound_items),
"warehouse": warehouse_name,
"positions": selected_positions
}
# -------------------------------- # --------------------------------
@@ -1067,20 +817,24 @@ class BioyondCellWorkstation(BioyondWorkstation):
if __name__ == "__main__": if __name__ == "__main__":
ws = BioyondCellWorkstation() ws = BioyondCellWorkstation()
logger.info(ws.scheduler_start()) logger.info(ws.scheduler_stop())
results = ws.create_materials(SOLID_LIQUID_MAPPINGS)
for r in results:
logger.info(r)
# 从CSV文件读取物料列表并批量创建入库 # 从CSV文件读取物料列表并批量创建入库
result = ws.create_and_inbound_materials_from_csv() # logger.info(ws.create_and_inbound_materials_from_csv())
# 继续后续流程 # 继续后续流程
logger.info(ws.auto_feeding4to3()) #搬运物料到3号箱 # logger.info(ws.auto_feeding4to3()) #搬运物料到3号箱
# 使用正斜杠或 Path 对象来指定文件路径 # # 使用正斜杠或 Path 对象来指定文件路径
excel_path = Path("unilabos/devices/workstation/bioyond_studio/bioyond_cell/2025092701.xlsx") # excel_path = Path("unilabos\\devices\\workstation\\bioyond_studio\\bioyond_cell\\2025092701.xlsx")
logger.info(ws.create_orders(excel_path)) # logger.info(ws.create_orders(excel_path))
logger.info(ws.transfer_3_to_2_to_1()) # logger.info(ws.transfer_3_to_2_to_1())
logger.info(ws.transfer_1_to_2())
# logger.info(ws.transfer_1_to_2())
# logger.info(ws.scheduler_start())
while True: while True:

222
unilabos/devices/workstation/bioyond_studio/config.py
View File

@@ -5,17 +5,13 @@
import os import os
# ==================== API 基础配置 ==================== # ==================== API 基础配置 ====================
# 支持通过环境变量覆盖默认值
API_CONFIG = {
"api_key": os.getenv("BIOYOND_API_KEY", "8A819E5C"),
"api_host": os.getenv("BIOYOND_API_HOST", "http://172.16.11.219:44388"),
}
# ==================== 完整的 Bioyond 配置 ==================== # ==================== 完整的 Bioyond 配置 ====================
# BioyondCellWorkstation 默认配置（包含所有必需参数） # BioyondCellWorkstation 默认配置（包含所有必需参数）
BIOYOND_FULL_CONFIG = { API_CONFIG = {
# API 连接配置 # API 连接配置
"base_url": os.getenv("BIOYOND_API_HOST", "http://172.16.11.219:44388"), "api_host": os.getenv("BIOYOND_API_HOST", "http://172.16.11.219:44388"),
"api_key": os.getenv("BIOYOND_API_KEY", "8A819E5C"), "api_key": os.getenv("BIOYOND_API_KEY", "8A819E5C"),
"timeout": int(os.getenv("BIOYOND_TIMEOUT", "30")), "timeout": int(os.getenv("BIOYOND_TIMEOUT", "30")),
@@ -23,96 +19,15 @@ BIOYOND_FULL_CONFIG = {
"report_token": os.getenv("BIOYOND_REPORT_TOKEN", "CHANGE_ME_TOKEN"), "report_token": os.getenv("BIOYOND_REPORT_TOKEN", "CHANGE_ME_TOKEN"),
# HTTP 服务配置 # HTTP 服务配置
"HTTP_host": os.getenv("BIOYOND_HTTP_HOST", "0.0.0.0"), # HTTP服务监听地址（0.0.0.0 表示监听所有网络接口） "HTTP_host": os.getenv("unilab_HTTP_HOST", "172.21.32.164"), # HTTP服务监听地址（0.0.0.0 表示监听所有网络接口）
"HTTP_port": int(os.getenv("BIOYOND_HTTP_PORT", "8080")), "HTTP_port": int(os.getenv("unilab_HTTP_PORT", "8080")),
"report_ip": os.getenv("BIOYOND_REPORT_IP", "172.21.32.172"), # 报送给 Bioyond 的本机IP地址（留空则自动检测）
# 调试模式 # 调试模式
"debug_mode": os.getenv("BIOYOND_DEBUG_MODE", "False").lower() == "true", "debug_mode": False,
}
# 工作流映射配置
WORKFLOW_MAPPINGS = {
"reactor_taken_out": "",
"reactor_taken_in": "",
"Solid_feeding_vials": "",
"Liquid_feeding_vials(non-titration)": "",
"Liquid_feeding_solvents": "",
"Liquid_feeding(titration)": "",
"liquid_feeding_beaker": "",
"Drip_back": "",
}
# 工作流名称到DisplaySectionName的映射
WORKFLOW_TO_SECTION_MAP = {
'reactor_taken_in': '反应器放入',
'liquid_feeding_beaker': '液体投料-烧杯',
'Liquid_feeding_vials(non-titration)': '液体投料-小瓶（非滴定）',
'Liquid_feeding_solvents': '液体投料-溶剂',
'Solid_feeding_vials': '固体投料-小瓶',
'Liquid_feeding(titration)': '液体投料-滴定',
'reactor_taken_out': '反应器取出'
} }
# 库位映射配置 # 库位映射配置
WAREHOUSE_MAPPING = { WAREHOUSE_MAPPING = {
"粉末堆栈": {
"uuid": "",
"site_uuids": {
# 样品板
"A1": "3a14198e-6929-31f0-8a22-0f98f72260df",
"A2": "3a14198e-6929-4379-affa-9a2935c17f99",
"A3": "3a14198e-6929-56da-9a1c-7f5fbd4ae8af",
"A4": "3a14198e-6929-5e99-2b79-80720f7cfb54",
"B1": "3a14198e-6929-f525-9a1b-1857552b28ee",
"B2": "3a14198e-6929-bf98-0fd5-26e1d68bf62d",
"B3": "3a14198e-6929-2d86-a468-602175a2b5aa",
"B4": "3a14198e-6929-1a98-ae57-e97660c489ad",
# 分装板
"C1": "3a14198e-6929-46fe-841e-03dd753f1e4a",
"C2": "3a14198e-6929-1bc9-a9bd-3b7ca66e7f95",
"C3": "3a14198e-6929-72ac-32ce-9b50245682b8",
"C4": "3a14198e-6929-3bd8-e6c7-4a9fd93be118",
"D1": "3a14198e-6929-8a0b-b686-6f4a2955c4e2",
"D2": "3a14198e-6929-dde1-fc78-34a84b71afdf",
"D3": "3a14198e-6929-a0ec-5f15-c0f9f339f963",
"D4": "3a14198e-6929-7ac8-915a-fea51cb2e884"
}
},
"溶液堆栈": {
"uuid": "",
"site_uuids": {
"A1": "3a14198e-d724-e036-afdc-2ae39a7f3383",
"A2": "3a14198e-d724-afa4-fc82-0ac8a9016791",
"A3": "3a14198e-d724-ca48-bb9e-7e85751e55b6",
"A4": "3a14198e-d724-df6d-5e32-5483b3cab583",
"B1": "3a14198e-d724-d818-6d4f-5725191a24b5",
"B2": "3a14198e-d724-be8a-5e0b-012675e195c6",
"B3": "3a14198e-d724-cc1e-5c2c-228a130f40a8",
"B4": "3a14198e-d724-1e28-c885-574c3df468d0",
"C1": "3a14198e-d724-b5bb-adf3-4c5a0da6fb31",
"C2": "3a14198e-d724-ab4e-48cb-817c3c146707",
"C3": "3a14198e-d724-7f18-1853-39d0c62e1d33",
"C4": "3a14198e-d724-28a2-a760-baa896f46b66",
"D1": "3a14198e-d724-d378-d266-2508a224a19f",
"D2": "3a14198e-d724-f56e-468b-0110a8feb36a",
"D3": "3a14198e-d724-0cf1-dea9-a1f40fe7e13c",
"D4": "3a14198e-d724-0ddd-9654-f9352a421de9"
}
},
"试剂堆栈": {
"uuid": "",
"site_uuids": {
"A1": "3a14198c-c2cf-8b40-af28-b467808f1c36",
"A2": "3a14198c-c2d0-f3e7-871a-e470d144296f",
"A3": "3a14198c-c2d0-dc7d-b8d0-e1d88cee3094",
"A4": "3a14198c-c2d0-2070-efc8-44e245f10c6f",
"B1": "3a14198c-c2d0-354f-39ad-642e1a72fcb8",
"B2": "3a14198c-c2d0-1559-105d-0ea30682cab4",
"B3": "3a14198c-c2d0-725e-523d-34c037ac2440",
"B4": "3a14198c-c2d0-efce-0939-69ca5a7dfd39"
}
},
"粉末加样头堆栈": { "粉末加样头堆栈": {
"uuid": "", "uuid": "",
"site_uuids": { "site_uuids": {
@@ -142,44 +57,99 @@ WAREHOUSE_MAPPING = {
# 物料类型配置 # 物料类型配置
MATERIAL_TYPE_MAPPINGS = { MATERIAL_TYPE_MAPPINGS = {
"烧杯": ("BIOYOND_PolymerStation_1FlaskCarrier", "3a14196b-24f2-ca49-9081-0cab8021bf1a"), "20ml分液瓶": ("YB_6x20ml_DispensingVialCarrier", "3a14196e-5dfe-6e21-0c79-fe2036d052c4"),
"试剂瓶": ("BIOYOND_PolymerStation_1BottleCarrier", ""), "100ml液体": ("BIOYOND_PolymerStation_100ml_Liquid_Bottle", "d37166b3-ecaa-481e-bd84-3032b795ba07"),
"样品板": ("BIOYOND_PolymerStation_6StockCarrier", "3a14196e-b7a0-a5da-1931-35f3000281e9"), "液": ("BIOYOND_PolymerStation_Liquid_Bottle", "3a190ca1-2add-2b23-f8e1-bbd348b7f790"),
"分装板": ("BIOYOND_PolymerStation_6VialCarrier", "3a14196e-5dfe-6e21-0c79-fe2036d052c4"), "高粘液": ("BIOYOND_PolymerStation_High_Viscosity_Liquid_Bottle", "abe8df30-563d-43d2-85e0-cabec59ddc16"),
"样品瓶": ("BIOYOND_PolymerStation_Solid_Stock", "3a14196a-cf7d-8aea-48d8-b9662c7dba94"), "加样头(大)": ("BIOYOND_PolymerStation_Large_Dispense_Head", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
"90%分装小瓶": ("BIOYOND_PolymerStation_Solid_Vial", "3a14196c-cdcf-088d-dc7d-5cf38f0ad9ea"), "5ml分液瓶板": ("BIOYOND_PolymerStation_6x5ml_DispensingVialCarrier", "3a192fa4-007d-ec7b-456e-2a8be7a13f23"),
"10%分装小瓶": ("BIOYOND_PolymerStation_Liquid_Vial", "3a14196c-76be-2279-4e22-7310d69aed68"), "5ml分液瓶": ("BIOYOND_PolymerStation_5ml_Dispensing_Vial", "3a192c2a-ebb7-58a1-480d-8b3863bf74f4"),
"20ml分液瓶板": ("BIOYOND_PolymerStation_6x20ml_DispensingVialCarrier", "3a192fa4-47db-3449-162a-eaf8aba57e27"),
"配液瓶(小)板": ("BIOYOND_PolymerStation_6x_SmallSolutionBottleCarrier", "3a190c8b-3284-af78-d29f-9a69463ad047"),
"配液瓶(小)": ("BIOYOND_PolymerStation_Small_Solution_Bottle", "3a190c8c-fe8f-bf48-0dc3-97afc7f508eb"),
"配液瓶(大)板": ("BIOYOND_PolymerStation_4x_LargeSolutionBottleCarrier", "53e50377-32dc-4781-b3c0-5ce45bc7dc27"),
"配液瓶(大)": ("BIOYOND_PolymerStation_Large_Solution_Bottle", "19c52ad1-51c5-494f-8854-576f4ca9c6ca"),
"加样头(大)板": ("BIOYOND_PolymerStation_6x_LargeDispenseHeadCarrier", "a8e714ae-2a4e-4eb9-9614-e4c140ec3f16"),
"适配器块": ("BIOYOND_PolymerStation_AdapterBlock", "efc3bb32-d504-4890-91c0-b64ed3ac80cf"),
"枪头盒": ("BIOYOND_PolymerStation_TipBox", "3a192c2e-20f3-a44a-0334-c8301839d0b3"),
"枪头": ("BIOYOND_PolymerStation_Pipette_Tip", "b6196971-1050-46da-9927-333e8dea062d"),
# YB信息
} }
# 步骤参数配置（各工作流的步骤UUID） SOLID_LIQUID_MAPPINGS = {
WORKFLOW_STEP_IDS = { # 固体
"reactor_taken_in": { "LiDFOB": {
"config": "" "typeId": "3a190ca0-b2f6-9aeb-8067-547e72c11469",
"code": "",
"barCode": "",
"name": "LiDFOB",
"unit": "g",
"parameters": "",
"quantity": "2",
"warningQuantity": "1",
"details": []
}, },
"liquid_feeding_beaker": { # "LiPF6": {
"liquid": "", # "typeId": "3a190ca0-b2f6-9aeb-8067-547e72c11469",
"observe": "" # "code": "",
}, # "barCode": "",
"liquid_feeding_vials_non_titration": { # "name": "LiPF6",
"liquid": "", # "unit": "g",
"observe": "" # "parameters": "",
}, # "quantity": 2,
"liquid_feeding_solvents": { # "warningQuantity": 1,
"liquid": "", # "details": []
"observe": "" # },
}, # "LiFSI": {
"solid_feeding_vials": { # "typeId": "3a190ca0-b2f6-9aeb-8067-547e72c11469",
"feeding": "", # "code": "",
"observe": "" # "barCode": "",
}, # "name": "LiFSI",
"liquid_feeding_titration": { # "unit": "g",
"liquid": "", # "parameters": {"Density": "1.533"},
"observe": "" # "quantity": 2,
}, # "warningQuantity": 1,
"drip_back": { # "details": [{}]
"liquid": "", # },
"observe": "" # "DTC": {
} # "typeId": "3a190ca0-b2f6-9aeb-8067-547e72c11469",
# "code": "",
# "barCode": "",
# "name": "DTC",
# "unit": "g",
# "parameters": {"Density": "1.533"},
# "quantity": 2,
# "warningQuantity": 1,
# "details": [{}]
# },
# "LiPO2F2": {
# "typeId": "3a190ca0-b2f6-9aeb-8067-547e72c11469",
# "code": "",
# "barCode": "",
# "name": "LiPO2F2",
# "unit": "g",
# "parameters": {"Density": "1.533"},
# "quantity": 2,
# "warningQuantity": 1,
# "details": [{}]
# },
# 液体
# "SA": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "EC": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "VC": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "AND": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "HTCN": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "DENE": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "TMSP": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "TMSB": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "EP": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "DEC": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "EMC": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "SN": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "DMC": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
# "FEC": ("BIOYOND_PolymerStation_Solid_Stock", "3a190ca0-b2f6-9aeb-8067-547e72c11469"),
} }
WORKFLOW_MAPPINGS = {}
LOCATION_MAPPING = {} LOCATION_MAPPING = {}

8
unilabos/devices/workstation/workstation_http_service.py
View File

@@ -71,11 +71,11 @@ class WorkstationHTTPHandler(BaseHTTPRequestHandler):
if content_length > 0: if content_length > 0:
post_data = self.rfile.read(content_length) post_data = self.rfile.read(content_length)
request_data = json.loads(post_data.decode('utf-8')) request_data = json.loads(post_data.decode('utf-8'))
else: else:
request_data = {} request_data = {}
logger.info(f"收到工作站报送: {endpoint} 收到接受数据：{request_data}")
# logger.info(f"收到的json数据: {request_data}") logger.info(f"收到工作站报送: {endpoint} - {request_data.get('token', 'unknown')}")
# 统一的报送端点路由（基于LIMS协议规范） # 统一的报送端点路由（基于LIMS协议规范）
if endpoint == '/report/step_finish': if endpoint == '/report/step_finish':
response = self._handle_step_finish_report(request_data) response = self._handle_step_finish_report(request_data)
@@ -668,7 +668,7 @@ __all__ = [
if __name__ == "__main__": if __name__ == "__main__":
# 简单测试HTTP服务 # 简单测试HTTP服务
class BioyondWorkstation: class DummyWorkstation:
device_id = "WS-001" device_id = "WS-001"
def process_step_finish_report(self, report_request): def process_step_finish_report(self, report_request):

2
unilabos/registry/resources/bioyond/deck.yaml
View File

@@ -22,7 +22,7 @@ BIOYOND_PolymerReactionStation_Deck:
init_param_schema: {} init_param_schema: {}
registry_type: resource registry_type: resource
version: 1.0.0 version: 1.0.0
YB_Deck15: YB_Deck16:
category: category:
- deck - deck
class: class:

2
unilabos/resources/bioyond/bottles.py
View File

@@ -12,7 +12,7 @@ def BIOYOND_PolymerStation_Solid_Stock(
"""创建粉末瓶""" """创建粉末瓶"""
return Bottle( return Bottle(
name=name, name=name,
diameter=diameter, diameter=diameter,# 未知
height=height, height=height,
max_volume=max_volume, max_volume=max_volume,
barcode=barcode, barcode=barcode,

2
unilabos/resources/bioyond/warehouses.py
View File

@@ -48,7 +48,7 @@ def bioyond_warehouse_1x2x2(name: str) -> WareHouse:
item_dx=137.0, item_dx=137.0,
item_dy=96.0, item_dy=96.0,
item_dz=120.0, item_dz=120.0,
category="warehouse", category="YB_warehouse",
) )
def bioyond_warehouse_10x1x1(name: str) -> WareHouse: def bioyond_warehouse_10x1x1(name: str) -> WareHouse:
"""创建BioYond 4x1x4仓库""" """创建BioYond 4x1x4仓库"""

2
unilabos/resources/graphio.py
View File

@@ -624,6 +624,8 @@ def resource_bioyond_to_plr(bioyond_materials: list[dict], type_mapping: Dict[st
Returns: Returns:
pylabrobot 格式的物料列表 pylabrobot 格式的物料列表
""" """
print("1:bioyond_materials:",bioyond_materials)
# print("2:type_mapping:",type_mapping)
plr_materials = [] plr_materials = []
for material in bioyond_materials: for material in bioyond_materials: