feat(bioyond_studio): 添加项目API接口支持及优化物料管理功能

添加通用项目API接口方法(_post_project_api, _delete_project_api)用于与LIMS系统交互
实现compute_experiment_design方法用于实验设计计算
新增brief_step_parameters等订单相关接口方法
优化物料转移逻辑,增加异步任务处理
扩展BioyondV1RPC类,添加批量物料操作、订单状态管理等功能
This commit is contained in:
ZiWei
2025-11-18 18:57:47 +08:00
committed by Xuwznln
parent d39662f65f
commit 931614fe64
3 changed files with 1020 additions and 111 deletions

View File

@@ -192,6 +192,23 @@ class BioyondV1RPC(BaseRequest):
return [] return []
return str(response.get("data", {})) return str(response.get("data", {}))
def material_type_list(self) -> list:
"""查询物料类型列表
返回值:
list: 物料类型数组,失败返回空列表
"""
response = self.post(
url=f'{self.host}/api/lims/storage/material-type-list',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": {},
})
if not response or response['code'] != 1:
return []
return response.get("data", [])
def material_inbound(self, material_id: str, location_id: str) -> dict: def material_inbound(self, material_id: str, location_id: str) -> dict:
""" """
描述:指定库位入库一个物料 描述:指定库位入库一个物料
@@ -221,6 +238,26 @@ class BioyondV1RPC(BaseRequest):
# 入库成功时,即使没有 data 字段,也返回成功标识 # 入库成功时,即使没有 data 字段,也返回成功标识
return response.get("data") or {"success": True} return response.get("data") or {"success": True}
def batch_inbound(self, inbound_items: List[Dict[str, Any]]) -> int:
"""批量入库物料
参数:
inbound_items: 入库条目列表,每项包含 materialId/locationId/quantity 等
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/storage/batch-inbound',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": inbound_items,
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
def delete_material(self, material_id: str) -> dict: def delete_material(self, material_id: str) -> dict:
""" """
描述:删除尚未入库的物料 描述:删除尚未入库的物料
@@ -289,6 +326,66 @@ class BioyondV1RPC(BaseRequest):
return None return None
return response return response
def batch_outbound(self, outbound_items: List[Dict[str, Any]]) -> int:
"""批量出库物料
参数:
outbound_items: 出库条目列表,每项包含 materialId/locationId/quantity 等
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/storage/batch-outbound',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": outbound_items,
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
def material_info(self, material_id: str) -> dict:
"""查询物料详情
参数:
material_id: 物料ID
返回值:
dict: 物料信息字典,失败返回空字典
"""
response = self.post(
url=f'{self.host}/api/lims/storage/material-info',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": material_id,
})
if not response or response['code'] != 1:
return {}
return response.get("data", {})
def reset_location(self, location_id: str) -> int:
"""复位库位
参数:
location_id: 库位ID
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/storage/reset-location',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": location_id,
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
# ==================== 工作流查询相关接口 ==================== # ==================== 工作流查询相关接口 ====================
def query_workflow(self, json_str: str) -> dict: def query_workflow(self, json_str: str) -> dict:
@@ -332,6 +429,66 @@ class BioyondV1RPC(BaseRequest):
return {} return {}
return response.get("data", {}) return response.get("data", {})
def split_workflow_list(self, params: Dict[str, Any]) -> dict:
"""查询可拆分工作流列表
参数:
params: 查询条件参数
返回值:
dict: 返回数据字典,失败返回空字典
"""
response = self.post(
url=f'{self.host}/api/lims/workflow/split-workflow-list',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": params,
})
if not response or response['code'] != 1:
return {}
return response.get("data", {})
def merge_workflow(self, data: Dict[str, Any]) -> dict:
"""合并工作流(无参数版)
参数:
data: 合并请求体,包含待合并的子工作流信息
返回值:
dict: 合并结果,失败返回空字典
"""
response = self.post(
url=f'{self.host}/api/lims/workflow/merge-workflow',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": data,
})
if not response or response['code'] != 1:
return {}
return response.get("data", {})
def merge_workflow_with_parameters(self, data: Dict[str, Any]) -> dict:
"""合并工作流(携带参数)
参数:
data: 合并请求体,包含 name、workflows 以及 stepParameters 等
返回值:
dict: 合并结果,失败返回空字典
"""
response = self.post(
url=f'{self.host}/api/lims/workflow/merge-workflow-with-parameters',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": data,
})
if not response or response['code'] != 1:
return {}
return response.get("data", {})
def validate_workflow_parameters(self, workflows: List[Dict[str, Any]]) -> Dict[str, Any]: def validate_workflow_parameters(self, workflows: List[Dict[str, Any]]) -> Dict[str, Any]:
"""验证工作流参数格式""" """验证工作流参数格式"""
try: try:
@@ -494,35 +651,34 @@ class BioyondV1RPC(BaseRequest):
return {} return {}
return response.get("data", {}) return response.get("data", {})
def order_report(self, json_str: str) -> dict: def order_report(self, order_id: str) -> dict:
""" """查询订单报告
描述:查询某个任务明细
json_str 格式为JSON字符串:
'{"order_id": "order123"}'
"""
try:
data = json.loads(json_str)
order_id = data.get("order_id", "")
except json.JSONDecodeError:
return {}
参数:
order_id: 订单ID
返回值:
dict: 报告数据,失败返回空字典
"""
response = self.post( response = self.post(
url=f'{self.host}/api/lims/order/project-order-report', url=f'{self.host}/api/lims/order/order-report',
params={ params={
"apiKey": self.api_key, "apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(), "requestTime": self.get_current_time_iso8601(),
"data": order_id, "data": order_id,
}) })
if not response or response['code'] != 1: if not response or response['code'] != 1:
return {} return {}
return response.get("data", {}) return response.get("data", {})
def order_takeout(self, json_str: str) -> int: def order_takeout(self, json_str: str) -> int:
""" """取出任务产物
描述:取出任务产物
json_str 格式为JSON字符串: 参数:
'{"order_id": "order123", "preintake_id": "preintake123"}' json_str: JSON字符串包含 order_id 与 preintake_id
返回值:
int: 成功返回1失败返回0
""" """
try: try:
data = json.loads(json_str) data = json.loads(json_str)
@@ -545,14 +701,15 @@ class BioyondV1RPC(BaseRequest):
return 0 return 0
return response.get("code", 0) return response.get("code", 0)
def sample_waste_removal(self, order_id: str) -> dict: def sample_waste_removal(self, order_id: str) -> dict:
""" """样品/废料取出
样品/废料取出接口
参数: 参数:
- order_id: 订单ID order_id: 订单ID
返回: 取出结果 返回:
dict: 取出结果,失败返回空字典
""" """
params = {"orderId": order_id} params = {"orderId": order_id}
@@ -574,10 +731,13 @@ class BioyondV1RPC(BaseRequest):
return response.get("data", {}) return response.get("data", {})
def cancel_order(self, json_str: str) -> bool: def cancel_order(self, json_str: str) -> bool:
""" """取消指定任务
描述:取消指定任务
json_str 格式为JSON字符串: 参数:
'{"order_id": "order123"}' json_str: JSON字符串包含 order_id
返回值:
bool: 成功返回 True失败返回 False
""" """
try: try:
data = json.loads(json_str) data = json.loads(json_str)
@@ -597,6 +757,126 @@ class BioyondV1RPC(BaseRequest):
return False return False
return True return True
def cancel_experiment(self, order_id: str) -> int:
"""取消指定实验
参数:
order_id: 订单ID
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/order/cancel-experiment',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": order_id,
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
def batch_cancel_experiment(self, order_ids: List[str]) -> int:
"""批量取消实验
参数:
order_ids: 订单ID列表
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/order/batch-cancel-experiment',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": order_ids,
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
def gantts_by_order_id(self, order_id: str) -> dict:
"""查询订单甘特图数据
参数:
order_id: 订单ID
返回值:
dict: 甘特数据,失败返回空字典
"""
response = self.post(
url=f'{self.host}/api/lims/order/gantts-by-order-id',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": order_id,
})
if not response or response['code'] != 1:
return {}
return response.get("data", {})
def simulation_gantt_by_order_id(self, order_id: str) -> dict:
"""查询订单模拟甘特图数据
参数:
order_id: 订单ID
返回值:
dict: 模拟甘特数据,失败返回空字典
"""
response = self.post(
url=f'{self.host}/api/lims/order/simulation-gantt-by-order-id',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": order_id,
})
if not response or response['code'] != 1:
return {}
return response.get("data", {})
def reset_order_status(self, order_id: str) -> int:
"""复位订单状态
参数:
order_id: 订单ID
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/order/reset-order-status',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": order_id,
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
def gantt_with_simulation_by_order_id(self, order_id: str) -> dict:
"""查询订单甘特与模拟联合数据
参数:
order_id: 订单ID
返回值:
dict: 联合数据,失败返回空字典
"""
response = self.post(
url=f'{self.host}/api/lims/order/gantt-with-simulation-by-order-id',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": order_id,
})
if not response or response['code'] != 1:
return {}
return response.get("data", {})
# ==================== 设备管理相关接口 ==================== # ==================== 设备管理相关接口 ====================
def device_list(self, json_str: str = "") -> list: def device_list(self, json_str: str = "") -> list:
@@ -628,9 +908,13 @@ class BioyondV1RPC(BaseRequest):
return response.get("data", []) return response.get("data", [])
def device_operation(self, json_str: str) -> int: def device_operation(self, json_str: str) -> int:
""" """设备操作
描述:操作设备
json_str 格式为JSON字符串 参数:
json_str: JSON字符串包含 device_no/operationType/operationParams
返回值:
int: 成功返回1失败返回0
""" """
try: try:
data = json.loads(json_str) data = json.loads(json_str)
@@ -643,7 +927,7 @@ class BioyondV1RPC(BaseRequest):
return 0 return 0
response = self.post( response = self.post(
url=f'{self.host}/api/lims/device/device-operation', url=f'{self.host}/api/lims/device/execute-operation',
params={ params={
"apiKey": self.api_key, "apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(), "requestTime": self.get_current_time_iso8601(),
@@ -654,9 +938,30 @@ class BioyondV1RPC(BaseRequest):
return 0 return 0
return response.get("code", 0) return response.get("code", 0)
def reset_devices(self) -> int:
"""复位设备集合
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/device/reset-devices',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
# ==================== 调度器相关接口 ==================== # ==================== 调度器相关接口 ====================
def scheduler_status(self) -> dict: def scheduler_status(self) -> dict:
"""查询调度器状态
返回值:
dict: 包含 schedulerStatus/hasTask/creationTime 等
"""
response = self.post( response = self.post(
url=f'{self.host}/api/lims/scheduler/scheduler-status', url=f'{self.host}/api/lims/scheduler/scheduler-status',
params={ params={
@@ -669,7 +974,7 @@ class BioyondV1RPC(BaseRequest):
return response.get("data", {}) return response.get("data", {})
def scheduler_start(self) -> int: def scheduler_start(self) -> int:
"""描述:启动调度器""" """启动调度器"""
response = self.post( response = self.post(
url=f'{self.host}/api/lims/scheduler/start', url=f'{self.host}/api/lims/scheduler/start',
params={ params={
@@ -682,7 +987,7 @@ class BioyondV1RPC(BaseRequest):
return response.get("code", 0) return response.get("code", 0)
def scheduler_pause(self) -> int: def scheduler_pause(self) -> int:
"""描述:暂停调度器""" """暂停调度器"""
response = self.post( response = self.post(
url=f'{self.host}/api/lims/scheduler/pause', url=f'{self.host}/api/lims/scheduler/pause',
params={ params={
@@ -694,8 +999,21 @@ class BioyondV1RPC(BaseRequest):
return 0 return 0
return response.get("code", 0) return response.get("code", 0)
def scheduler_smart_pause(self) -> int:
"""智能暂停调度器"""
response = self.post(
url=f'{self.host}/api/lims/scheduler/smart-pause',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
def scheduler_continue(self) -> int: def scheduler_continue(self) -> int:
"""描述:继续调度器""" """继续调度器"""
response = self.post( response = self.post(
url=f'{self.host}/api/lims/scheduler/continue', url=f'{self.host}/api/lims/scheduler/continue',
params={ params={
@@ -708,7 +1026,7 @@ class BioyondV1RPC(BaseRequest):
return response.get("code", 0) return response.get("code", 0)
def scheduler_stop(self) -> int: def scheduler_stop(self) -> int:
"""描述:停止调度器""" """停止调度器"""
response = self.post( response = self.post(
url=f'{self.host}/api/lims/scheduler/stop', url=f'{self.host}/api/lims/scheduler/stop',
params={ params={
@@ -721,7 +1039,7 @@ class BioyondV1RPC(BaseRequest):
return response.get("code", 0) return response.get("code", 0)
def scheduler_reset(self) -> int: def scheduler_reset(self) -> int:
"""描述:复位调度器""" """复位调度器"""
response = self.post( response = self.post(
url=f'{self.host}/api/lims/scheduler/reset', url=f'{self.host}/api/lims/scheduler/reset',
params={ params={
@@ -733,6 +1051,26 @@ class BioyondV1RPC(BaseRequest):
return 0 return 0
return response.get("code", 0) return response.get("code", 0)
def scheduler_reply_error_handling(self, data: Dict[str, Any]) -> int:
"""调度错误处理回复
参数:
data: 错误处理参数
返回值:
int: 成功返回1失败返回0
"""
response = self.post(
url=f'{self.host}/api/lims/scheduler/reply-error-handling',
params={
"apiKey": self.api_key,
"requestTime": self.get_current_time_iso8601(),
"data": data,
})
if not response or response['code'] != 1:
return 0
return response.get("code", 0)
# ==================== 辅助方法 ==================== # ==================== 辅助方法 ====================
def _load_material_cache(self): def _load_material_cache(self):
@@ -796,3 +1134,23 @@ class BioyondV1RPC(BaseRequest):
def get_available_materials(self): def get_available_materials(self):
"""获取所有可用的材料名称列表""" """获取所有可用的材料名称列表"""
return list(self.material_cache.keys()) return list(self.material_cache.keys())
def get_scheduler_state(self) -> Optional[MachineState]:
"""将调度状态字符串映射为枚举值
返回值:
Optional[MachineState]: 映射后的枚举,失败返回 None
"""
data = self.scheduler_status()
if not isinstance(data, dict):
return None
status = data.get("schedulerStatus")
mapping = {
"Init": MachineState.INITIAL,
"Stop": MachineState.STOPPED,
"Running": MachineState.RUNNING,
"Pause": MachineState.PAUSED,
"ErrorPause": MachineState.ERROR_PAUSED,
"ErrorStop": MachineState.ERROR_STOPPED,
}
return mapping.get(status)

View File

@@ -1,11 +1,17 @@
from datetime import datetime from datetime import datetime
import json import json
import time import time
from typing import Optional, Dict, Any from typing import Optional, Dict, Any, List
import requests
from unilabos.devices.workstation.bioyond_studio.config import API_CONFIG
from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import BioyondException from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import BioyondException
from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation
from unilabos.ros.nodes.base_device_node import ROS2DeviceNode, BaseROS2DeviceNode
import json
import sys
from pathlib import Path
import importlib
class BioyondDispensingStation(BioyondWorkstation): class BioyondDispensingStation(BioyondWorkstation):
def __init__( def __init__(
@@ -28,6 +34,108 @@ class BioyondDispensingStation(BioyondWorkstation):
# 用于跟踪任务完成状态的字典: {orderCode: {status, order_id, timestamp}} # 用于跟踪任务完成状态的字典: {orderCode: {status, order_id, timestamp}}
self.order_completion_status = {} self.order_completion_status = {}
def _post_project_api(self, endpoint: str, data: Any) -> Dict[str, Any]:
"""项目接口通用POST调用
参数:
endpoint: 接口路径(例如 /api/lims/order/brief-step-paramerers
data: 请求体中的 data 字段内容
返回:
dict: 服务端响应,失败时返回 {code:0,message,...}
"""
request_data = {
"apiKey": API_CONFIG["api_key"],
"requestTime": self.hardware_interface.get_current_time_iso8601(),
"data": data
}
try:
response = requests.post(
f"{self.hardware_interface.host}{endpoint}",
json=request_data,
headers={"Content-Type": "application/json"},
timeout=30
)
result = response.json()
return result if isinstance(result, dict) else {"code": 0, "message": "非JSON响应"}
except json.JSONDecodeError:
return {"code": 0, "message": "非JSON响应"}
except requests.exceptions.Timeout:
return {"code": 0, "message": "请求超时"}
except requests.exceptions.RequestException as e:
return {"code": 0, "message": str(e)}
def _delete_project_api(self, endpoint: str, data: Any) -> Dict[str, Any]:
"""项目接口通用DELETE调用
参数:
endpoint: 接口路径(例如 /api/lims/order/workflows
data: 请求体中的 data 字段内容
返回:
dict: 服务端响应,失败时返回 {code:0,message,...}
"""
request_data = {
"apiKey": API_CONFIG["api_key"],
"requestTime": self.hardware_interface.get_current_time_iso8601(),
"data": data
}
try:
response = requests.delete(
f"{self.hardware_interface.host}{endpoint}",
json=request_data,
headers={"Content-Type": "application/json"},
timeout=30
)
result = response.json()
return result if isinstance(result, dict) else {"code": 0, "message": "非JSON响应"}
except json.JSONDecodeError:
return {"code": 0, "message": "非JSON响应"}
except requests.exceptions.Timeout:
return {"code": 0, "message": "请求超时"}
except requests.exceptions.RequestException as e:
return {"code": 0, "message": str(e)}
def compute_experiment_design(
self,
ratio: dict,
wt_percent: str = "0.25",
m_tot: str = "70",
titration_percent: str = "0.03",
) -> dict:
try:
if isinstance(ratio, str):
try:
ratio = json.loads(ratio)
except Exception:
ratio = {}
root = str(Path(__file__).resolve().parents[3])
if root not in sys.path:
sys.path.append(root)
try:
mod = importlib.import_module("tem.compute")
except Exception as e:
raise BioyondException(f"无法导入计算模块: {e}")
try:
wp = float(wt_percent) if isinstance(wt_percent, str) else wt_percent
mt = float(m_tot) if isinstance(m_tot, str) else m_tot
tp = float(titration_percent) if isinstance(titration_percent, str) else titration_percent
except Exception as e:
raise BioyondException(f"参数解析失败: {e}")
res = mod.generate_experiment_design(ratio=ratio, wt_percent=wp, m_tot=mt, titration_percent=tp)
out = {
"solutions": res.get("solutions", []),
"titration": res.get("titration", {}),
"solvents": res.get("solvents", {}),
"feeding_order": res.get("feeding_order", []),
"return_info": json.dumps(res, ensure_ascii=False)
}
return out
except BioyondException:
raise
except Exception as e:
raise BioyondException(str(e))
# 90%10%小瓶投料任务创建方法 # 90%10%小瓶投料任务创建方法
def create_90_10_vial_feeding_task(self, def create_90_10_vial_feeding_task(self,
order_name: str = None, order_name: str = None,
@@ -649,6 +757,40 @@ class BioyondDispensingStation(BioyondWorkstation):
self.hardware_interface._logger.error(error_msg) self.hardware_interface._logger.error(error_msg)
raise BioyondException(error_msg) raise BioyondException(error_msg)
def brief_step_parameters(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""获取简要步骤参数(站点项目接口)
参数:
data: 查询参数字典
返回值:
dict: 接口返回数据
"""
return self._post_project_api("/api/lims/order/brief-step-paramerers", data)
def project_order_report(self, order_id: str) -> Dict[str, Any]:
"""查询项目端订单报告(兼容旧路径)
参数:
order_id: 订单ID
返回值:
dict: 报告数据
"""
return self._post_project_api("/api/lims/order/project-order-report", order_id)
def workflow_sample_locations(self, workflow_id: str) -> Dict[str, Any]:
"""查询工作流样品库位(站点项目接口)
参数:
workflow_id: 工作流ID
返回值:
dict: 位置信息数据
"""
return self._post_project_api("/api/lims/storage/workflow-sample-locations", workflow_id)
# 批量创建90%10%小瓶投料任务 # 批量创建90%10%小瓶投料任务
def batch_create_90_10_vial_feeding_tasks(self, def batch_create_90_10_vial_feeding_tasks(self,
titration, titration,
@@ -779,8 +921,38 @@ class BioyondDispensingStation(BioyondWorkstation):
self.hardware_interface._logger.error(error_msg) self.hardware_interface._logger.error(error_msg)
raise BioyondException(error_msg) raise BioyondException(error_msg)
def _extract_actuals_from_report(self, report) -> Dict[str, Any]:
data = report.get('data') if isinstance(report, dict) else None
actual_target_weigh = None
actual_volume = None
if data:
extra = data.get('extraProperties') or {}
if isinstance(extra, dict):
for v in extra.values():
obj = None
try:
obj = json.loads(v) if isinstance(v, str) else v
except Exception:
obj = None
if isinstance(obj, dict):
tw = obj.get('targetWeigh')
vol = obj.get('volume')
if tw is not None:
try:
actual_target_weigh = float(tw)
except Exception:
pass
if vol is not None:
try:
actual_volume = float(vol)
except Exception:
pass
return {
'actualTargetWeigh': actual_target_weigh,
'actualVolume': actual_volume
}
# 等待多个任务完成并获取实验报告
def wait_for_multiple_orders_and_get_reports(self, def wait_for_multiple_orders_and_get_reports(self,
batch_create_result: str = None, batch_create_result: str = None,
timeout: int = 7200, timeout: int = 7200,
@@ -902,6 +1074,7 @@ class BioyondDispensingStation(BioyondWorkstation):
"status": "timeout", "status": "timeout",
"completion_status": None, "completion_status": None,
"report": None, "report": None,
"extracted": None,
"elapsed_time": elapsed_time "elapsed_time": elapsed_time
}) })
@@ -921,8 +1094,7 @@ class BioyondDispensingStation(BioyondWorkstation):
# 获取实验报告 # 获取实验报告
try: try:
report_query = json.dumps({"order_id": order_id}) report = self.project_order_report(order_id)
report = self.hardware_interface.order_report(report_query)
if not report: if not report:
self.hardware_interface._logger.warning( self.hardware_interface._logger.warning(
@@ -940,6 +1112,7 @@ class BioyondDispensingStation(BioyondWorkstation):
"status": "completed", "status": "completed",
"completion_status": completion_info.get('status'), "completion_status": completion_info.get('status'),
"report": report, "report": report,
"extracted": self._extract_actuals_from_report(report),
"elapsed_time": elapsed_time "elapsed_time": elapsed_time
}) })
@@ -959,6 +1132,7 @@ class BioyondDispensingStation(BioyondWorkstation):
"status": "error", "status": "error",
"completion_status": completion_info.get('status'), "completion_status": completion_info.get('status'),
"report": None, "report": None,
"extracted": None,
"error": str(e), "error": str(e),
"elapsed_time": elapsed_time "elapsed_time": elapsed_time
}) })
@@ -1052,6 +1226,266 @@ class BioyondDispensingStation(BioyondWorkstation):
self.hardware_interface._logger.error(f"处理任务完成报送失败: {e}") self.hardware_interface._logger.error(f"处理任务完成报送失败: {e}")
return {"processed": False, "error": str(e)} return {"processed": False, "error": str(e)}
def transfer_materials_to_reaction_station(
self,
target_device_id: str,
transfer_groups: list
) -> dict:
"""
将配液站完成的物料转移到指定反应站的堆栈库位
支持多组转移任务,每组包含物料名称、目标堆栈和目标库位
Args:
target_device_id: 目标反应站设备ID(所有转移组使用同一个设备)
transfer_groups: 转移任务组列表,每组包含:
- materials: 物料名称(字符串,将通过RPC查询)
- target_stack: 目标堆栈名称(如"堆栈1左")
- target_sites: 目标库位(如"A01")
Returns:
dict: 转移结果
{
"success": bool,
"total_groups": int,
"successful_groups": int,
"failed_groups": int,
"target_device_id": str,
"details": [...]
}
"""
try:
# 验证参数
if not target_device_id:
raise ValueError("目标设备ID不能为空")
if not transfer_groups:
raise ValueError("转移任务组列表不能为空")
if not isinstance(transfer_groups, list):
raise ValueError("transfer_groups必须是列表类型")
# 标准化设备ID格式: 确保以 /devices/ 开头
if not target_device_id.startswith("/devices/"):
if target_device_id.startswith("/"):
target_device_id = f"/devices{target_device_id}"
else:
target_device_id = f"/devices/{target_device_id}"
self.hardware_interface._logger.info(
f"目标设备ID标准化为: {target_device_id}"
)
self.hardware_interface._logger.info(
f"开始执行批量物料转移: {len(transfer_groups)}组任务 -> {target_device_id}"
)
from .config import WAREHOUSE_MAPPING
results = []
successful_count = 0
failed_count = 0
for idx, group in enumerate(transfer_groups, 1):
try:
# 提取参数
material_name = group.get("materials", "")
target_stack = group.get("target_stack", "")
target_sites = group.get("target_sites", "")
# 验证必填参数
if not material_name:
raise ValueError(f"{idx}组: 物料名称不能为空")
if not target_stack:
raise ValueError(f"{idx}组: 目标堆栈不能为空")
if not target_sites:
raise ValueError(f"{idx}组: 目标库位不能为空")
self.hardware_interface._logger.info(
f"处理第{idx}组转移: {material_name} -> "
f"{target_device_id}/{target_stack}/{target_sites}"
)
# 通过物料名称从deck获取ResourcePLR对象
try:
material_resource = self.deck.get_resource(material_name)
if not material_resource:
raise ValueError(f"在deck中未找到物料: {material_name}")
self.hardware_interface._logger.info(
f"从deck获取到物料 {material_name}: {material_resource}"
)
except Exception as e:
raise ValueError(
f"获取物料 {material_name} 失败: {str(e)}请确认物料已正确加载到deck中"
)
# 验证目标堆栈是否存在
if target_stack not in WAREHOUSE_MAPPING:
raise ValueError(
f"未知的堆栈名称: {target_stack}"
f"可选值: {list(WAREHOUSE_MAPPING.keys())}"
)
# 验证库位是否有效
stack_sites = WAREHOUSE_MAPPING[target_stack].get("site_uuids", {})
if target_sites not in stack_sites:
raise ValueError(
f"库位 {target_sites} 不存在于堆栈 {target_stack} 中,"
f"可选库位: {list(stack_sites.keys())}"
)
# 获取目标库位的UUID
target_site_uuid = stack_sites[target_sites]
if not target_site_uuid:
raise ValueError(
f"库位 {target_sites} 的 UUID 未配置,请在 WAREHOUSE_MAPPING 中完善"
)
# 目标位点包含UUID
future = ROS2DeviceNode.run_async_func(
self._ros_node.get_resource_with_dir,
True,
**{
"resource_id": f"/reaction_station_bioyond/Bioyond_Deck/{target_stack}",
"with_children": True,
},
)
# 等待异步完成后再获取结果
if not future:
raise ValueError(f"获取目标堆栈资源future无效: {target_stack}")
while not future.done():
time.sleep(0.1)
target_site_resource = future.result()
# 调用父类的 transfer_resource_to_another 方法
# 传入ResourcePLR对象和目标位点资源
future = self.transfer_resource_to_another(
resource=[material_resource],
mount_resource=[target_site_resource],
sites=[target_sites],
mount_device_id=target_device_id
)
# 等待异步任务完成(轮询直到完成,再取结果)
if future:
try:
while not future.done():
time.sleep(0.1)
future.result()
self.hardware_interface._logger.info(
f"异步转移任务已完成: {material_name}"
)
except Exception as e:
raise ValueError(f"转移任务执行失败: {str(e)}")
self.hardware_interface._logger.info(
f"{idx}组转移成功: {material_name} -> "
f"{target_device_id}/{target_stack}/{target_sites}"
)
successful_count += 1
results.append({
"group_index": idx,
"success": True,
"material_name": material_name,
"target_stack": target_stack,
"target_site": target_sites,
"message": "转移成功"
})
except Exception as e:
error_msg = f"{idx}组转移失败: {str(e)}"
self.hardware_interface._logger.error(error_msg)
failed_count += 1
results.append({
"group_index": idx,
"success": False,
"material_name": group.get("materials", ""),
"error": str(e)
})
# 返回汇总结果
return {
"success": failed_count == 0,
"total_groups": len(transfer_groups),
"successful_groups": successful_count,
"failed_groups": failed_count,
"target_device_id": target_device_id,
"details": results,
"message": f"完成 {len(transfer_groups)} 组转移任务到 {target_device_id}: "
f"{successful_count} 成功, {failed_count} 失败"
}
except Exception as e:
error_msg = f"批量转移物料失败: {str(e)}"
self.hardware_interface._logger.error(error_msg)
return {
"success": False,
"total_groups": len(transfer_groups) if transfer_groups else 0,
"successful_groups": 0,
"failed_groups": len(transfer_groups) if transfer_groups else 0,
"target_device_id": target_device_id if target_device_id else "",
"error": error_msg
}
def query_resource_by_name(self, material_name: str):
"""
通过物料名称查询资源对象(适用于Bioyond系统)
Args:
material_name: 物料名称
Returns:
物料ID或None
"""
try:
# Bioyond系统使用material_cache存储物料信息
if not hasattr(self.hardware_interface, 'material_cache'):
self.hardware_interface._logger.error(
"hardware_interface没有material_cache属性"
)
return None
material_cache = self.hardware_interface.material_cache
self.hardware_interface._logger.info(
f"查询物料 '{material_name}', 缓存中共有 {len(material_cache)} 个物料"
)
# 调试: 打印前几个物料信息
if material_cache:
cache_items = list(material_cache.items())[:5]
for name, material_id in cache_items:
self.hardware_interface._logger.debug(
f"缓存物料: name={name}, id={material_id}"
)
# 直接从缓存中查找
if material_name in material_cache:
material_id = material_cache[material_name]
self.hardware_interface._logger.info(
f"找到物料: {material_name} -> ID: {material_id}"
)
return material_id
self.hardware_interface._logger.warning(
f"未找到物料: {material_name} (缓存中无此物料)"
)
# 打印所有可用物料名称供参考
available_materials = list(material_cache.keys())
if available_materials:
self.hardware_interface._logger.info(
f"可用物料列表(前10个): {available_materials[:10]}"
)
return None
except Exception as e:
self.hardware_interface._logger.error(
f"查询物料失败 {material_name}: {str(e)}"
)
return None
if __name__ == "__main__": if __name__ == "__main__":
bioyond = BioyondDispensingStation(config={ bioyond = BioyondDispensingStation(config={

View File

@@ -5,6 +5,7 @@ from typing import List, Dict, Any
from pathlib import Path from pathlib import Path
from datetime import datetime from datetime import datetime
from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation from unilabos.devices.workstation.bioyond_studio.station import BioyondWorkstation
from unilabos.devices.workstation.bioyond_studio.bioyond_rpc import MachineState
from unilabos.ros.msgs.message_converter import convert_to_ros_msg, Float64, String from unilabos.ros.msgs.message_converter import convert_to_ros_msg, Float64, String
from unilabos.devices.workstation.bioyond_studio.config import ( from unilabos.devices.workstation.bioyond_studio.config import (
WORKFLOW_STEP_IDS, WORKFLOW_STEP_IDS,
@@ -717,8 +718,7 @@ class BioyondReactionStation(BioyondWorkstation):
if oc in self.order_completion_status: if oc in self.order_completion_status:
info = self.order_completion_status[oc] info = self.order_completion_status[oc]
try: try:
rq = json.dumps({"order_id": oid}) rep = self.hardware_interface.order_report(oid)
rep = self.hardware_interface.order_report(rq)
if not rep: if not rep:
rep = {"error": "无法获取报告"} rep = {"error": "无法获取报告"}
reports.append({ reports.append({
@@ -912,6 +912,106 @@ class BioyondReactionStation(BioyondWorkstation):
""" """
return self.hardware_interface.create_order(json_str) return self.hardware_interface.create_order(json_str)
def hard_delete_merged_workflows(self, workflow_ids: List[str]) -> Dict[str, Any]:
"""
调用新接口:硬删除合并后的工作流
Args:
workflow_ids: 要删除的工作流ID数组
Returns:
删除结果
"""
try:
if not isinstance(workflow_ids, list):
raise ValueError("workflow_ids必须是字符串数组")
return self._delete_project_api("/api/lims/order/workflows", workflow_ids)
except Exception as e:
print(f"❌ 硬删除异常: {str(e)}")
return {"code": 0, "message": str(e), "timestamp": int(time.time())}
# ==================== 项目接口通用方法 ====================
def _post_project_api(self, endpoint: str, data: Any) -> Dict[str, Any]:
"""项目接口通用POST调用
参数:
endpoint: 接口路径(例如 /api/lims/order/skip-titration-steps
data: 请求体中的 data 字段内容
返回:
dict: 服务端响应,失败时返回 {code:0,message,...}
"""
request_data = {
"apiKey": API_CONFIG["api_key"],
"requestTime": self.hardware_interface.get_current_time_iso8601(),
"data": data
}
print(f"\n📤 项目POST请求: {self.hardware_interface.host}{endpoint}")
print(json.dumps(request_data, indent=4, ensure_ascii=False))
try:
response = requests.post(
f"{self.hardware_interface.host}{endpoint}",
json=request_data,
headers={"Content-Type": "application/json"},
timeout=30
)
result = response.json()
if result.get("code") == 1:
print("✅ 请求成功")
else:
print(f"❌ 请求失败: {result.get('message','未知错误')}")
return result
except json.JSONDecodeError:
print("❌ 非JSON响应")
return {"code": 0, "message": "非JSON响应", "timestamp": int(time.time())}
except requests.exceptions.Timeout:
print("❌ 请求超时")
return {"code": 0, "message": "请求超时", "timestamp": int(time.time())}
except requests.exceptions.RequestException as e:
print(f"❌ 网络异常: {str(e)}")
return {"code": 0, "message": str(e), "timestamp": int(time.time())}
def _delete_project_api(self, endpoint: str, data: Any) -> Dict[str, Any]:
"""项目接口通用DELETE调用
参数:
endpoint: 接口路径(例如 /api/lims/order/workflows
data: 请求体中的 data 字段内容
返回:
dict: 服务端响应,失败时返回 {code:0,message,...}
"""
request_data = {
"apiKey": API_CONFIG["api_key"],
"requestTime": self.hardware_interface.get_current_time_iso8601(),
"data": data
}
print(f"\n📤 项目DELETE请求: {self.hardware_interface.host}{endpoint}")
print(json.dumps(request_data, indent=4, ensure_ascii=False))
try:
response = requests.delete(
f"{self.hardware_interface.host}{endpoint}",
json=request_data,
headers={"Content-Type": "application/json"},
timeout=30
)
result = response.json()
if result.get("code") == 1:
print("✅ 请求成功")
else:
print(f"❌ 请求失败: {result.get('message','未知错误')}")
return result
except json.JSONDecodeError:
print("❌ 非JSON响应")
return {"code": 0, "message": "非JSON响应", "timestamp": int(time.time())}
except requests.exceptions.Timeout:
print("❌ 请求超时")
return {"code": 0, "message": "请求超时", "timestamp": int(time.time())}
except requests.exceptions.RequestException as e:
print(f"❌ 网络异常: {str(e)}")
return {"code": 0, "message": str(e), "timestamp": int(time.time())}
# ==================== 工作流执行核心方法 ==================== # ==================== 工作流执行核心方法 ====================
def process_web_workflows(self, web_workflow_json: str) -> List[Dict[str, str]]: def process_web_workflows(self, web_workflow_json: str) -> List[Dict[str, str]]:
@@ -942,76 +1042,6 @@ class BioyondReactionStation(BioyondWorkstation):
print(f"错误:处理工作流失败: {e}") print(f"错误:处理工作流失败: {e}")
return [] return []
def process_and_execute_workflow(self, workflow_name: str, task_name: str) -> dict:
"""
一站式处理工作流程:解析网页工作流列表,合并工作流(带参数),然后发布任务
Args:
workflow_name: 合并后的工作流名称
task_name: 任务名称
Returns:
任务创建结果
"""
web_workflow_list = self.get_workflow_sequence()
print(f"\n{'='*60}")
print(f"📋 处理网页工作流列表: {web_workflow_list}")
print(f"{'='*60}")
web_workflow_json = json.dumps({"web_workflow_list": web_workflow_list})
workflows_result = self.process_web_workflows(web_workflow_json)
if not workflows_result:
return self._create_error_result("处理网页工作流列表失败", "process_web_workflows")
print(f"workflows_result 类型: {type(workflows_result)}")
print(f"workflows_result 内容: {workflows_result}")
workflows_with_params = self._build_workflows_with_parameters(workflows_result)
merge_data = {
"name": workflow_name,
"workflows": workflows_with_params
}
# print(f"\n🔄 合并工作流(带参数),名称: {workflow_name}")
merged_workflow = self.merge_workflow_with_parameters(json.dumps(merge_data))
if not merged_workflow:
return self._create_error_result("合并工作流失败", "merge_workflow_with_parameters")
workflow_id = merged_workflow.get("subWorkflows", [{}])[0].get("id", "")
# print(f"\n📤 使用工作流创建任务: {workflow_name} (ID: {workflow_id})")
order_params = [{
"orderCode": f"task_{self.hardware_interface.get_current_time_iso8601()}",
"orderName": task_name,
"workFlowId": workflow_id,
"borderNumber": 1,
"paramValues": {}
}]
result = self.create_order(json.dumps(order_params))
if not result:
return self._create_error_result("创建任务失败", "create_order")
# 清空工作流序列和参数,防止下次执行时累积重复
self.pending_task_params = []
self.clear_workflows() # 清空工作流序列,避免重复累积
# print(f"\n✅ 任务创建成功: {result}")
# print(f"\n✅ 任务创建成功")
print(f"{'='*60}\n")
# 返回结果,包含合并后的工作流数据和订单参数
return json.dumps({
"success": True,
"result": result,
"merged_workflow": merged_workflow,
"order_params": order_params
})
def _build_workflows_with_parameters(self, workflows_result: list) -> list: def _build_workflows_with_parameters(self, workflows_result: list) -> list:
""" """
构建带参数的工作流列表 构建带参数的工作流列表
@@ -1211,3 +1241,90 @@ class BioyondReactionStation(BioyondWorkstation):
print(f" ❌ 工作流ID验证失败: {e}") print(f" ❌ 工作流ID验证失败: {e}")
print(f" 💡 将重新合并工作流") print(f" 💡 将重新合并工作流")
return False return False
def process_and_execute_workflow(self, workflow_name: str, task_name: str) -> dict:
"""
一站式处理工作流程:解析网页工作流列表,合并工作流(带参数),然后发布任务
Args:
workflow_name: 合并后的工作流名称
task_name: 任务名称
Returns:
任务创建结果
"""
web_workflow_list = self.get_workflow_sequence()
print(f"\n{'='*60}")
print(f"📋 处理网页工作流列表: {web_workflow_list}")
print(f"{'='*60}")
web_workflow_json = json.dumps({"web_workflow_list": web_workflow_list})
workflows_result = self.process_web_workflows(web_workflow_json)
if not workflows_result:
return self._create_error_result("处理网页工作流列表失败", "process_web_workflows")
print(f"workflows_result 类型: {type(workflows_result)}")
print(f"workflows_result 内容: {workflows_result}")
workflows_with_params = self._build_workflows_with_parameters(workflows_result)
merge_data = {
"name": workflow_name,
"workflows": workflows_with_params
}
# print(f"\n🔄 合并工作流(带参数),名称: {workflow_name}")
merged_workflow = self.merge_workflow_with_parameters(json.dumps(merge_data))
if not merged_workflow:
return self._create_error_result("合并工作流失败", "merge_workflow_with_parameters")
workflow_id = merged_workflow.get("subWorkflows", [{}])[0].get("id", "")
# print(f"\n📤 使用工作流创建任务: {workflow_name} (ID: {workflow_id})")
order_params = [{
"orderCode": f"task_{self.hardware_interface.get_current_time_iso8601()}",
"orderName": task_name,
"workFlowId": workflow_id,
"borderNumber": 1,
"paramValues": {}
}]
result = self.create_order(json.dumps(order_params))
if not result:
return self._create_error_result("创建任务失败", "create_order")
# 清空工作流序列和参数,防止下次执行时累积重复
self.pending_task_params = []
self.clear_workflows() # 清空工作流序列,避免重复累积
# print(f"\n✅ 任务创建成功: {result}")
# print(f"\n✅ 任务创建成功")
print(f"{'='*60}\n")
# 返回结果,包含合并后的工作流数据和订单参数
return json.dumps({
"success": True,
"result": result,
"merged_workflow": merged_workflow,
"order_params": order_params
})
# ==================== 反应器操作接口 ====================
def skip_titration_steps(self, preintake_id: str) -> Dict[str, Any]:
"""跳过当前正在进行的滴定步骤
Args:
preintake_id: 通量ID
Returns:
Dict[str, Any]: 服务器响应,包含状态码、消息和时间戳
"""
try:
return self._post_project_api("/api/lims/order/skip-titration-steps", preintake_id)
except Exception as e:
print(f"❌ 跳过滴定异常: {str(e)}")
return {"code": 0, "message": str(e), "timestamp": int(time.time())}