Compare commits

..

6 Commits

Author SHA1 Message Date
Xuwznln
9a7d5c7c82 add registry name & add always free 2026-02-07 02:11:43 +08:00
Xuwznln
4f7d431c0b correct config organic synthesis 2026-02-06 12:04:19 +08:00
Xuwznln
341a1b537c Adapt to new scheduler, sampels, and edge upload format (#230)
* add sample_material

* adapt to new samples sys

* fix pump transfer. fix resource update when protocol & ros callback

* Adapt to new scheduler.
2026-02-06 00:49:53 +08:00
Xuwznln
957fb41a6f Feat/samples (#229)
* add sample_material

* adapt to new samples sys
2026-02-05 00:42:12 +08:00
Xuwznln
26271bcab8 adapt to new samples sys 2026-02-04 18:49:08 +08:00
Xuwznln
84a8223173 adapt to new edge format 2026-02-03 23:22:38 +08:00
30 changed files with 1122 additions and 401 deletions

View File

@@ -452,8 +452,9 @@ unilab --ak your_ak --sk your_sk -g test/experiments/mock_devices/mock_all.json
**操作步骤:** **操作步骤:**
1. 将两个 `container` 拖拽到 `workstation` 1. 将两个 `container` 拖拽到 `workstation`
2.`virtual_transfer_pump` 拖拽到 `workstation` 2.`virtual_multiway_valve` 拖拽到 `workstation`
3. 在画布上连接它们(建立父子关系) 3. `virtual_transfer_pump` 拖拽到 `workstation`
4. 在画布上连接它们(建立父子关系)
![设备连接](image/links.png) ![设备连接](image/links.png)

Binary file not shown.

Before

Width:  |  Height:  |  Size: 275 KiB

After

Width:  |  Height:  |  Size: 415 KiB

View File

@@ -54,6 +54,7 @@ class JobAddReq(BaseModel):
action_type: str = Field( action_type: str = Field(
examples=["unilabos_msgs.action._str_single_input.StrSingleInput"], description="action type", default="" examples=["unilabos_msgs.action._str_single_input.StrSingleInput"], description="action type", default=""
) )
sample_material: dict = Field(examples=[{"string": "string"}], description="sample uuid to material uuid")
action_args: dict = Field(examples=[{"string": "string"}], description="action arguments", default_factory=dict) action_args: dict = Field(examples=[{"string": "string"}], description="action arguments", default_factory=dict)
task_id: str = Field(examples=["task_id"], description="task uuid (auto-generated if empty)", default="") task_id: str = Field(examples=["task_id"], description="task uuid (auto-generated if empty)", default="")
job_id: str = Field(examples=["job_id"], description="goal uuid (auto-generated if empty)", default="") job_id: str = Field(examples=["job_id"], description="goal uuid (auto-generated if empty)", default="")

View File

@@ -327,6 +327,7 @@ def job_add(req: JobAddReq) -> JobData:
queue_item, queue_item,
action_type=action_type, action_type=action_type,
action_kwargs=action_args, action_kwargs=action_args,
sample_material=req.sample_material,
server_info=server_info, server_info=server_info,
) )

View File

@@ -76,6 +76,7 @@ class JobInfo:
start_time: float start_time: float
last_update_time: float = field(default_factory=time.time) last_update_time: float = field(default_factory=time.time)
ready_timeout: Optional[float] = None # READY状态的超时时间 ready_timeout: Optional[float] = None # READY状态的超时时间
always_free: bool = False # 是否为永久闲置动作(不受排队限制)
def update_timestamp(self): def update_timestamp(self):
"""更新最后更新时间""" """更新最后更新时间"""
@@ -127,6 +128,15 @@ class DeviceActionManager:
# 总是将job添加到all_jobs中 # 总是将job添加到all_jobs中
self.all_jobs[job_info.job_id] = job_info self.all_jobs[job_info.job_id] = job_info
# always_free的动作不受排队限制直接设为READY
if job_info.always_free:
job_info.status = JobStatus.READY
job_info.update_timestamp()
job_info.set_ready_timeout(10)
job_log = format_job_log(job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name)
logger.trace(f"[DeviceActionManager] Job {job_log} always_free, start immediately")
return True
# 检查是否有正在执行或准备执行的任务 # 检查是否有正在执行或准备执行的任务
if device_key in self.active_jobs: if device_key in self.active_jobs:
# 有正在执行或准备执行的任务,加入队列 # 有正在执行或准备执行的任务,加入队列
@@ -176,9 +186,13 @@ class DeviceActionManager:
logger.error(f"[DeviceActionManager] Job {job_log} is not in READY status, current: {job_info.status}") logger.error(f"[DeviceActionManager] Job {job_log} is not in READY status, current: {job_info.status}")
return False return False
# always_free的job不需要检查active_jobs
if not job_info.always_free:
# 检查设备上是否是这个job # 检查设备上是否是这个job
if device_key not in self.active_jobs or self.active_jobs[device_key].job_id != job_id: if device_key not in self.active_jobs or self.active_jobs[device_key].job_id != job_id:
job_log = format_job_log(job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name) job_log = format_job_log(
job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name
)
logger.error(f"[DeviceActionManager] Job {job_log} is not the active job for {device_key}") logger.error(f"[DeviceActionManager] Job {job_log} is not the active job for {device_key}")
return False return False
@@ -203,6 +217,13 @@ class DeviceActionManager:
job_info = self.all_jobs[job_id] job_info = self.all_jobs[job_id]
device_key = job_info.device_action_key device_key = job_info.device_action_key
# always_free的job直接清理不影响队列
if job_info.always_free:
job_info.status = JobStatus.ENDED
job_info.update_timestamp()
del self.all_jobs[job_id]
return None
# 移除活跃任务 # 移除活跃任务
if device_key in self.active_jobs and self.active_jobs[device_key].job_id == job_id: if device_key in self.active_jobs and self.active_jobs[device_key].job_id == job_id:
del self.active_jobs[device_key] del self.active_jobs[device_key]
@@ -234,9 +255,14 @@ class DeviceActionManager:
return None return None
def get_active_jobs(self) -> List[JobInfo]: def get_active_jobs(self) -> List[JobInfo]:
"""获取所有正在执行的任务""" """获取所有正在执行的任务(含active_jobs和always_free的STARTED job)"""
with self.lock: with self.lock:
return list(self.active_jobs.values()) jobs = list(self.active_jobs.values())
# 补充 always_free 的 STARTED job(它们不在 active_jobs 中)
for job in self.all_jobs.values():
if job.always_free and job.status == JobStatus.STARTED and job not in jobs:
jobs.append(job)
return jobs
def get_queued_jobs(self) -> List[JobInfo]: def get_queued_jobs(self) -> List[JobInfo]:
"""获取所有排队中的任务""" """获取所有排队中的任务"""
@@ -261,6 +287,14 @@ class DeviceActionManager:
job_info = self.all_jobs[job_id] job_info = self.all_jobs[job_id]
device_key = job_info.device_action_key device_key = job_info.device_action_key
# always_free的job直接清理
if job_info.always_free:
job_info.status = JobStatus.ENDED
del self.all_jobs[job_id]
job_log = format_job_log(job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name)
logger.trace(f"[DeviceActionManager] Always-free job {job_log} cancelled")
return True
# 如果是正在执行的任务 # 如果是正在执行的任务
if device_key in self.active_jobs and self.active_jobs[device_key].job_id == job_id: if device_key in self.active_jobs and self.active_jobs[device_key].job_id == job_id:
# 清理active job状态 # 清理active job状态
@@ -334,13 +368,18 @@ class DeviceActionManager:
timeout_jobs = [] timeout_jobs = []
with self.lock: with self.lock:
# 统计READY状态的任务数量 # 收集所有需要检查的 READY 任务(active_jobs + always_free READY jobs)
ready_jobs_count = sum(1 for job in self.active_jobs.values() if job.status == JobStatus.READY) ready_candidates = list(self.active_jobs.values())
for job in self.all_jobs.values():
if job.always_free and job.status == JobStatus.READY and job not in ready_candidates:
ready_candidates.append(job)
ready_jobs_count = sum(1 for job in ready_candidates if job.status == JobStatus.READY)
if ready_jobs_count > 0: if ready_jobs_count > 0:
logger.trace(f"[DeviceActionManager] Checking {ready_jobs_count} READY jobs for timeout") # type: ignore # noqa: E501 logger.trace(f"[DeviceActionManager] Checking {ready_jobs_count} READY jobs for timeout") # type: ignore # noqa: E501
# 找到所有超时的READY任务只检测不处理 # 找到所有超时的READY任务只检测不处理
for job_info in self.active_jobs.values(): for job_info in ready_candidates:
if job_info.is_ready_timeout(): if job_info.is_ready_timeout():
timeout_jobs.append(job_info) timeout_jobs.append(job_info)
job_log = format_job_log( job_log = format_job_log(
@@ -545,7 +584,7 @@ class MessageProcessor:
try: try:
message_str = json.dumps(msg, ensure_ascii=False) message_str = json.dumps(msg, ensure_ascii=False)
await self.websocket.send(message_str) await self.websocket.send(message_str)
logger.trace(f"[MessageProcessor] Message sent: {msg.get('action', 'unknown')}") # type: ignore # noqa: E501 # logger.trace(f"[MessageProcessor] Message sent: {msg.get('action', 'unknown')}") # type: ignore # noqa: E501
except Exception as e: except Exception as e:
logger.error(f"[MessageProcessor] Failed to send message: {str(e)}") logger.error(f"[MessageProcessor] Failed to send message: {str(e)}")
logger.error(traceback.format_exc()) logger.error(traceback.format_exc())
@@ -608,6 +647,24 @@ class MessageProcessor:
if host_node: if host_node:
host_node.handle_pong_response(pong_data) host_node.handle_pong_response(pong_data)
def _check_action_always_free(self, device_id: str, action_name: str) -> bool:
"""检查该action是否标记为always_free通过HostNode统一的_action_value_mappings查找"""
try:
host_node = HostNode.get_instance(0)
if not host_node:
return False
# noinspection PyProtectedMember
action_mappings = host_node._action_value_mappings.get(device_id)
if not action_mappings:
return False
# 尝试直接匹配或 auto- 前缀匹配
for key in [action_name, f"auto-{action_name}"]:
if key in action_mappings:
return action_mappings[key].get("always_free", False)
return False
except Exception:
return False
async def _handle_query_action_state(self, data: Dict[str, Any]): async def _handle_query_action_state(self, data: Dict[str, Any]):
"""处理query_action_state消息""" """处理query_action_state消息"""
device_id = data.get("device_id", "") device_id = data.get("device_id", "")
@@ -622,6 +679,9 @@ class MessageProcessor:
device_action_key = f"/devices/{device_id}/{action_name}" device_action_key = f"/devices/{device_id}/{action_name}"
# 检查action是否为always_free
action_always_free = self._check_action_always_free(device_id, action_name)
# 创建任务信息 # 创建任务信息
job_info = JobInfo( job_info = JobInfo(
job_id=job_id, job_id=job_id,
@@ -631,6 +691,7 @@ class MessageProcessor:
device_action_key=device_action_key, device_action_key=device_action_key,
status=JobStatus.QUEUE, status=JobStatus.QUEUE,
start_time=time.time(), start_time=time.time(),
always_free=action_always_free,
) )
# 添加到设备管理器 # 添加到设备管理器
@@ -657,6 +718,8 @@ class MessageProcessor:
async def _handle_job_start(self, data: Dict[str, Any]): async def _handle_job_start(self, data: Dict[str, Any]):
"""处理job_start消息""" """处理job_start消息"""
try: try:
if not data.get("sample_material"):
data["sample_material"] = {}
req = JobAddReq(**data) req = JobAddReq(**data)
job_log = format_job_log(req.job_id, req.task_id, req.device_id, req.action) job_log = format_job_log(req.job_id, req.task_id, req.device_id, req.action)
@@ -688,6 +751,7 @@ class MessageProcessor:
queue_item, queue_item,
action_type=req.action_type, action_type=req.action_type,
action_kwargs=req.action_args, action_kwargs=req.action_args,
sample_material=req.sample_material,
server_info=req.server_info, server_info=req.server_info,
) )
@@ -1301,7 +1365,7 @@ class WebSocketClient(BaseCommunicationClient):
}, },
} }
self.message_processor.send_message(message) self.message_processor.send_message(message)
logger.trace(f"[WebSocketClient] Device status published: {device_id}.{property_name}") # logger.trace(f"[WebSocketClient] Device status published: {device_id}.{property_name}")
def publish_job_status( def publish_job_status(
self, feedback_data: dict, item: QueueItem, status: str, return_info: Optional[dict] = None self, feedback_data: dict, item: QueueItem, status: str, return_info: Optional[dict] = None

View File

@@ -95,8 +95,29 @@ def get_vessel_liquid_volume(G: nx.DiGraph, vessel: str) -> float:
return total_volume return total_volume
def is_integrated_pump(node_name): def is_integrated_pump(node_class: str, node_name: str = "") -> bool:
return "pump" in node_name and "valve" in node_name """
判断是否为泵阀一体设备
"""
class_lower = (node_class or "").lower()
name_lower = (node_name or "").lower()
if "pump" not in class_lower and "pump" not in name_lower:
return False
integrated_markers = [
"valve",
"pump_valve",
"pumpvalve",
"integrated",
"transfer_pump",
]
for marker in integrated_markers:
if marker in class_lower or marker in name_lower:
return True
return False
def find_connected_pump(G, valve_node): def find_connected_pump(G, valve_node):
@@ -186,7 +207,9 @@ def build_pump_valve_maps(G, pump_backbone):
debug_print(f"🔧 过滤后的骨架: {filtered_backbone}") debug_print(f"🔧 过滤后的骨架: {filtered_backbone}")
for node in filtered_backbone: for node in filtered_backbone:
if is_integrated_pump(G.nodes[node]["class"]): node_data = G.nodes.get(node, {})
node_class = node_data.get("class", "") or ""
if is_integrated_pump(node_class, node):
pumps_from_node[node] = node pumps_from_node[node] = node
valve_from_node[node] = node valve_from_node[node] = node
debug_print(f" - 集成泵-阀: {node}") debug_print(f" - 集成泵-阀: {node}")

View File

@@ -27,7 +27,12 @@ from typing_extensions import TypedDict
from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
from unilabos.registry.placeholder_type import ResourceSlot from unilabos.registry.placeholder_type import ResourceSlot
from unilabos.resources.resource_tracker import ResourceTreeSet, ResourceDict from unilabos.resources.resource_tracker import (
ResourceTreeSet,
ResourceDict,
EXTRA_SAMPLE_UUID,
EXTRA_UNILABOS_SAMPLE_UUID,
)
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode
@@ -231,12 +236,11 @@ class LiquidHandlerMiddleware(LiquidHandler):
res_samples = [] res_samples = []
res_volumes = [] res_volumes = []
for resource, volume, channel in zip(resources, vols, use_channels): for resource, volume, channel in zip(resources, vols, use_channels):
res_samples.append( sample_uuid_value = resource.unilabos_extra.get(EXTRA_SAMPLE_UUID, None)
{"name": resource.name, "sample_uuid": resource.unilabos_extra.get("sample_uuid", None)} res_samples.append({"name": resource.name, EXTRA_SAMPLE_UUID: sample_uuid_value})
)
res_volumes.append(volume) res_volumes.append(volume)
self.pending_liquids_dict[channel] = { self.pending_liquids_dict[channel] = {
"sample_uuid": resource.unilabos_extra.get("sample_uuid", None), EXTRA_SAMPLE_UUID: sample_uuid_value,
"volume": volume, "volume": volume,
} }
return SimpleReturn(samples=res_samples, volumes=res_volumes) return SimpleReturn(samples=res_samples, volumes=res_volumes)
@@ -278,10 +282,10 @@ class LiquidHandlerMiddleware(LiquidHandler):
res_samples = [] res_samples = []
res_volumes = [] res_volumes = []
for resource, volume, channel in zip(resources, vols, use_channels): for resource, volume, channel in zip(resources, vols, use_channels):
res_uuid = self.pending_liquids_dict[channel]["sample_uuid"] res_uuid = self.pending_liquids_dict[channel][EXTRA_SAMPLE_UUID]
self.pending_liquids_dict[channel]["volume"] -= volume self.pending_liquids_dict[channel]["volume"] -= volume
resource.unilabos_extra["sample_uuid"] = res_uuid resource.unilabos_extra[EXTRA_SAMPLE_UUID] = res_uuid
res_samples.append({"name": resource.name, "sample_uuid": res_uuid}) res_samples.append({"name": resource.name, EXTRA_SAMPLE_UUID: res_uuid})
res_volumes.append(volume) res_volumes.append(volume)
return SimpleReturn(samples=res_samples, volumes=res_volumes) return SimpleReturn(samples=res_samples, volumes=res_volumes)
@@ -686,16 +690,14 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
) )
def set_liquid_from_plate( def set_liquid_from_plate(
self, plate: List[ResourceSlot], well_names: list[str], liquid_names: list[str], volumes: list[float] self, plate: ResourceSlot, well_names: list[str], liquid_names: list[str], volumes: list[float]
) -> SetLiquidFromPlateReturn: ) -> SetLiquidFromPlateReturn:
"""Set the liquid in wells of a plate by well names (e.g., A1, A2, B3). """Set the liquid in wells of a plate by well names (e.g., A1, A2, B3).
如果 liquid_names 和 volumes 为空,但 plate 和 well_names 不为空,直接返回 plate 和 wells。 如果 liquid_names 和 volumes 为空,但 plate 和 well_names 不为空,直接返回 plate 和 wells。
""" """
if isinstance(plate, list): # 未来移除
plate = plate[0]
assert issubclass(plate.__class__, Plate), "plate must be a Plate" assert issubclass(plate.__class__, Plate), "plate must be a Plate"
plate: Plate = cast(Plate, plate) plate: Plate = cast(Plate, cast(Resource, plate))
# 根据 well_names 获取对应的 Well 对象 # 根据 well_names 获取对应的 Well 对象
wells = [plate.get_well(name) for name in well_names] wells = [plate.get_well(name) for name in well_names]
res_volumes = [] res_volumes = []

View File

@@ -595,7 +595,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
return super().set_liquid(wells, liquid_names, volumes) return super().set_liquid(wells, liquid_names, volumes)
def set_liquid_from_plate( def set_liquid_from_plate(
self, plate: List[ResourceSlot], well_names: list[str], liquid_names: list[str], volumes: list[float] self, plate: ResourceSlot, well_names: list[str], liquid_names: list[str], volumes: list[float]
) -> SetLiquidFromPlateReturn: ) -> SetLiquidFromPlateReturn:
return super().set_liquid_from_plate(plate, well_names, liquid_names, volumes) return super().set_liquid_from_plate(plate, well_names, liquid_names, volumes)

View File

@@ -19,10 +19,11 @@ from rclpy.node import Node
import re import re
class LiquidHandlerJointPublisher(BaseROS2DeviceNode): class LiquidHandlerJointPublisher(BaseROS2DeviceNode):
def __init__(self,resources_config:list, resource_tracker, rate=50, device_id:str = "lh_joint_publisher", **kwargs): def __init__(self,resources_config:list, resource_tracker, rate=50, device_id:str = "lh_joint_publisher", registry_name: str = "lh_joint_publisher", **kwargs):
super().__init__( super().__init__(
driver_instance=self, driver_instance=self,
device_id=device_id, device_id=device_id,
registry_name=registry_name,
status_types={}, status_types={},
action_value_mappings={}, action_value_mappings={},
hardware_interface={}, hardware_interface={},

View File

@@ -31,14 +31,14 @@ class VirtualTransferPump:
# 从config或kwargs中获取参数确保类型正确 # 从config或kwargs中获取参数确保类型正确
if config: if config:
self.max_volume = float(config.get('max_volume', 25.0)) self.max_volume = float(config.get("max_volume", 25.0))
self.port = config.get('port', 'VIRTUAL') self.port = config.get("port", "VIRTUAL")
else: else:
self.max_volume = float(kwargs.get('max_volume', 25.0)) self.max_volume = float(kwargs.get("max_volume", 25.0))
self.port = kwargs.get('port', 'VIRTUAL') self.port = kwargs.get("port", "VIRTUAL")
self._transfer_rate = float(kwargs.get('transfer_rate', 0)) self._transfer_rate = float(kwargs.get("transfer_rate", 0))
self.mode = kwargs.get('mode', VirtualPumpMode.Normal) self.mode = kwargs.get("mode", VirtualPumpMode.Normal)
# 状态变量 - 确保都是正确类型 # 状态变量 - 确保都是正确类型
self._status = "Idle" self._status = "Idle"
@@ -54,7 +54,9 @@ class VirtualTransferPump:
self.logger = logging.getLogger(f"VirtualTransferPump.{self.device_id}") self.logger = logging.getLogger(f"VirtualTransferPump.{self.device_id}")
print(f"🚰 === 虚拟转移泵 {self.device_id} 已创建 === ✨") print(f"🚰 === 虚拟转移泵 {self.device_id} 已创建 === ✨")
print(f"💨 快速模式: {'启用' if self._fast_mode else '禁用'} | 移动时间: {self._fast_move_time}s | 喷射时间: {self._fast_dispense_time}s") print(
f"💨 快速模式: {'启用' if self._fast_mode else '禁用'} | 移动时间: {self._fast_move_time}s | 喷射时间: {self._fast_dispense_time}s"
)
print(f"📊 最大容量: {self.max_volume}mL | 端口: {self.port}") print(f"📊 最大容量: {self.max_volume}mL | 端口: {self.port}")
def post_init(self, ros_node: BaseROS2DeviceNode): def post_init(self, ros_node: BaseROS2DeviceNode):
@@ -189,7 +191,9 @@ class VirtualTransferPump:
operation_emoji = "📍" operation_emoji = "📍"
self.logger.info(f"🎯 SET_POSITION: {operation_type} {operation_emoji}") self.logger.info(f"🎯 SET_POSITION: {operation_type} {operation_emoji}")
self.logger.info(f" 📍 位置: {self._position:.2f}mL → {target_position:.2f}mL (移动 {volume_to_move:.2f}mL)") self.logger.info(
f" 📍 位置: {self._position:.2f}mL → {target_position:.2f}mL (移动 {volume_to_move:.2f}mL)"
)
self.logger.info(f" 🌊 速度: {velocity:.2f} mL/s") self.logger.info(f" 🌊 速度: {velocity:.2f} mL/s")
self.logger.info(f" ⏰ 预计时间: {display_duration:.2f}s") self.logger.info(f" ⏰ 预计时间: {display_duration:.2f}s")
@@ -207,7 +211,11 @@ class VirtualTransferPump:
for i in range(steps + 1): for i in range(steps + 1):
# 计算当前位置和进度 # 计算当前位置和进度
progress = (i / steps) * 100 if steps > 0 else 100 progress = (i / steps) * 100 if steps > 0 else 100
current_pos = start_position + (target_position - start_position) * (i / steps) if steps > 0 else target_position current_pos = (
start_position + (target_position - start_position) * (i / steps)
if steps > 0
else target_position
)
# 更新状态 # 更新状态
if i < steps: if i < steps:
@@ -244,7 +252,9 @@ class VirtualTransferPump:
# 📊 最终状态日志 # 📊 最终状态日志
if volume_to_move > 0.01: if volume_to_move > 0.01:
self.logger.info(f"🎉 SET_POSITION 完成! 📍 最终位置: {self._position:.2f}mL | 💧 当前体积: {self._current_volume:.2f}mL") self.logger.info(
f"🎉 SET_POSITION 完成! 📍 最终位置: {self._position:.2f}mL | 💧 当前体积: {self._current_volume:.2f}mL"
)
# 返回符合action定义的结果 # 返回符合action定义的结果
return { return {
@@ -252,7 +262,7 @@ class VirtualTransferPump:
"message": f"✅ 成功移动到位置 {self._position:.2f}mL ({operation_type})", "message": f"✅ 成功移动到位置 {self._position:.2f}mL ({operation_type})",
"final_position": self._position, "final_position": self._position,
"final_volume": self._current_volume, "final_volume": self._current_volume,
"operation_type": operation_type "operation_type": operation_type,
} }
except Exception as e: except Exception as e:
@@ -262,7 +272,7 @@ class VirtualTransferPump:
"success": False, "success": False,
"message": error_msg, "message": error_msg,
"final_position": self._position, "final_position": self._position,
"final_volume": self._current_volume "final_volume": self._current_volume,
} }
# 其他泵操作方法 # 其他泵操作方法
@@ -388,7 +398,9 @@ class VirtualTransferPump:
return self._current_volume >= (self.max_volume - 0.01) # 允许小量误差 return self._current_volume >= (self.max_volume - 0.01) # 允许小量误差
def __str__(self): def __str__(self):
return f"VirtualTransferPump({self.device_id}: {self._current_volume:.2f}/{self.max_volume} ml, {self._status})" return (
f"VirtualTransferPump({self.device_id}: {self._current_volume:.2f}/{self.max_volume} ml, {self._status})"
)
def __repr__(self): def __repr__(self):
return self.__str__() return self.__str__()

View File

@@ -11,9 +11,10 @@ Virtual Workbench Device - 模拟工作台设备
注意:调用来自线程池,使用 threading.Lock 进行同步 注意:调用来自线程池,使用 threading.Lock 进行同步
""" """
import logging import logging
import time import time
from typing import Dict, Any, Optional from typing import Dict, Any, Optional, List
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum from enum import Enum
from threading import Lock, RLock from threading import Lock, RLock
@@ -21,38 +22,47 @@ from threading import Lock, RLock
from typing_extensions import TypedDict from typing_extensions import TypedDict
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
from unilabos.utils.decorator import not_action from unilabos.utils.decorator import not_action, always_free
from unilabos.resources.resource_tracker import SampleUUIDsType, LabSample, RETURN_UNILABOS_SAMPLES
# ============ TypedDict 返回类型定义 ============ # ============ TypedDict 返回类型定义 ============
class MoveToHeatingStationResult(TypedDict): class MoveToHeatingStationResult(TypedDict):
"""move_to_heating_station 返回类型""" """move_to_heating_station 返回类型"""
success: bool success: bool
station_id: int station_id: int
material_id: str material_id: str
material_number: int material_number: int
message: str message: str
unilabos_samples: List[LabSample]
class StartHeatingResult(TypedDict): class StartHeatingResult(TypedDict):
"""start_heating 返回类型""" """start_heating 返回类型"""
success: bool success: bool
station_id: int station_id: int
material_id: str material_id: str
material_number: int material_number: int
message: str message: str
unilabos_samples: List[LabSample]
class MoveToOutputResult(TypedDict): class MoveToOutputResult(TypedDict):
"""move_to_output 返回类型""" """move_to_output 返回类型"""
success: bool success: bool
station_id: int station_id: int
material_id: str material_id: str
unilabos_samples: List[LabSample]
class PrepareMaterialsResult(TypedDict): class PrepareMaterialsResult(TypedDict):
"""prepare_materials 返回类型 - 批量准备物料""" """prepare_materials 返回类型 - 批量准备物料"""
success: bool success: bool
count: int count: int
material_1: int # 物料编号1 material_1: int # 物料编号1
@@ -61,12 +71,15 @@ class PrepareMaterialsResult(TypedDict):
material_4: int # 物料编号4 material_4: int # 物料编号4
material_5: int # 物料编号5 material_5: int # 物料编号5
message: str message: str
unilabos_samples: List[LabSample]
# ============ 状态枚举 ============ # ============ 状态枚举 ============
class HeatingStationState(Enum): class HeatingStationState(Enum):
"""加热台状态枚举""" """加热台状态枚举"""
IDLE = "idle" # 空闲 IDLE = "idle" # 空闲
OCCUPIED = "occupied" # 已放置物料,等待加热 OCCUPIED = "occupied" # 已放置物料,等待加热
HEATING = "heating" # 加热中 HEATING = "heating" # 加热中
@@ -75,6 +88,7 @@ class HeatingStationState(Enum):
class ArmState(Enum): class ArmState(Enum):
"""机械臂状态枚举""" """机械臂状态枚举"""
IDLE = "idle" # 空闲 IDLE = "idle" # 空闲
BUSY = "busy" # 工作中 BUSY = "busy" # 工作中
@@ -82,6 +96,7 @@ class ArmState(Enum):
@dataclass @dataclass
class HeatingStation: class HeatingStation:
"""加热台数据结构""" """加热台数据结构"""
station_id: int station_id: int
state: HeatingStationState = HeatingStationState.IDLE state: HeatingStationState = HeatingStationState.IDLE
current_material: Optional[str] = None # 当前物料 (如 "A1", "A2") current_material: Optional[str] = None # 当前物料 (如 "A1", "A2")
@@ -108,8 +123,8 @@ class VirtualWorkbench:
_ros_node: BaseROS2DeviceNode _ros_node: BaseROS2DeviceNode
# 配置常量 # 配置常量
ARM_OPERATION_TIME: float = 3.0 # 机械臂操作时间(秒) ARM_OPERATION_TIME: float = 2 # 机械臂操作时间(秒)
HEATING_TIME: float = 10.0 # 加热时间(秒) HEATING_TIME: float = 60.0 # 加热时间(秒)
NUM_HEATING_STATIONS: int = 3 # 加热台数量 NUM_HEATING_STATIONS: int = 3 # 加热台数量
def __init__(self, device_id: Optional[str] = None, config: Optional[Dict[str, Any]] = None, **kwargs): def __init__(self, device_id: Optional[str] = None, config: Optional[Dict[str, Any]] = None, **kwargs):
@@ -126,9 +141,9 @@ class VirtualWorkbench:
self.data: Dict[str, Any] = {} self.data: Dict[str, Any] = {}
# 从config中获取可配置参数 # 从config中获取可配置参数
self.ARM_OPERATION_TIME = float(self.config.get("arm_operation_time", 3.0)) self.ARM_OPERATION_TIME = float(self.config.get("arm_operation_time", self.ARM_OPERATION_TIME))
self.HEATING_TIME = float(self.config.get("heating_time", 10.0)) self.HEATING_TIME = float(self.config.get("heating_time", self.HEATING_TIME))
self.NUM_HEATING_STATIONS = int(self.config.get("num_heating_stations", 3)) self.NUM_HEATING_STATIONS = int(self.config.get("num_heating_stations", self.NUM_HEATING_STATIONS))
# 机械臂状态和锁 (使用threading.Lock) # 机械臂状态和锁 (使用threading.Lock)
self._arm_lock = Lock() self._arm_lock = Lock()
@@ -137,8 +152,7 @@ class VirtualWorkbench:
# 加热台状态 (station_id -> HeatingStation) - 立即初始化不依赖initialize() # 加热台状态 (station_id -> HeatingStation) - 立即初始化不依赖initialize()
self._heating_stations: Dict[int, HeatingStation] = { self._heating_stations: Dict[int, HeatingStation] = {
i: HeatingStation(station_id=i) i: HeatingStation(station_id=i) for i in range(1, self.NUM_HEATING_STATIONS + 1)
for i in range(1, self.NUM_HEATING_STATIONS + 1)
} }
self._stations_lock = RLock() # 可重入锁,保护加热台状态 self._stations_lock = RLock() # 可重入锁,保护加热台状态
@@ -178,14 +192,16 @@ class VirtualWorkbench:
station.heating_progress = 0.0 station.heating_progress = 0.0
# 初始化状态 # 初始化状态
self.data.update({ self.data.update(
{
"status": "Ready", "status": "Ready",
"arm_state": ArmState.IDLE.value, "arm_state": ArmState.IDLE.value,
"arm_current_task": None, "arm_current_task": None,
"heating_stations": self._get_stations_status(), "heating_stations": self._get_stations_status(),
"active_tasks_count": 0, "active_tasks_count": 0,
"message": "工作台就绪", "message": "工作台就绪",
}) }
)
self.logger.info(f"工作台初始化完成: {self.NUM_HEATING_STATIONS}个加热台就绪") self.logger.info(f"工作台初始化完成: {self.NUM_HEATING_STATIONS}个加热台就绪")
return True return True
@@ -204,12 +220,14 @@ class VirtualWorkbench:
with self._tasks_lock: with self._tasks_lock:
self._active_tasks.clear() self._active_tasks.clear()
self.data.update({ self.data.update(
{
"status": "Offline", "status": "Offline",
"arm_state": ArmState.IDLE.value, "arm_state": ArmState.IDLE.value,
"heating_stations": {}, "heating_stations": {},
"message": "工作台已关闭", "message": "工作台已关闭",
}) }
)
return True return True
def _get_stations_status(self) -> Dict[int, Dict[str, Any]]: def _get_stations_status(self) -> Dict[int, Dict[str, Any]]:
@@ -227,12 +245,14 @@ class VirtualWorkbench:
def _update_data_status(self, message: Optional[str] = None): def _update_data_status(self, message: Optional[str] = None):
"""更新状态数据""" """更新状态数据"""
self.data.update({ self.data.update(
{
"arm_state": self._arm_state.value, "arm_state": self._arm_state.value,
"arm_current_task": self._arm_current_task, "arm_current_task": self._arm_current_task,
"heating_stations": self._get_stations_status(), "heating_stations": self._get_stations_status(),
"active_tasks_count": len(self._active_tasks), "active_tasks_count": len(self._active_tasks),
}) }
)
if message: if message:
self.data["message"] = message self.data["message"] = message
@@ -280,6 +300,7 @@ class VirtualWorkbench:
def prepare_materials( def prepare_materials(
self, self,
sample_uuids: SampleUUIDsType,
count: int = 5, count: int = 5,
) -> PrepareMaterialsResult: ) -> PrepareMaterialsResult:
""" """
@@ -297,10 +318,7 @@ class VirtualWorkbench:
# 生成物料列表 A1 - A{count} # 生成物料列表 A1 - A{count}
materials = [i for i in range(1, count + 1)] materials = [i for i in range(1, count + 1)]
self.logger.info( self.logger.info(f"[准备物料] 生成 {count} 个物料: " f"A1-A{count} -> material_1~material_{count}")
f"[准备物料] 生成 {count} 个物料: "
f"A1-A{count} -> material_1~material_{count}"
)
return { return {
"success": True, "success": True,
@@ -311,10 +329,12 @@ class VirtualWorkbench:
"material_4": materials[3] if len(materials) > 3 else 0, "material_4": materials[3] if len(materials) > 3 else 0,
"material_5": materials[4] if len(materials) > 4 else 0, "material_5": materials[4] if len(materials) > 4 else 0,
"message": f"已准备 {count} 个物料: A1-A{count}", "message": f"已准备 {count} 个物料: A1-A{count}",
"unilabos_samples": [LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for sample_uuid, content in sample_uuids.items()]
} }
def move_to_heating_station( def move_to_heating_station(
self, self,
sample_uuids: SampleUUIDsType,
material_number: int, material_number: int,
) -> MoveToHeatingStationResult: ) -> MoveToHeatingStationResult:
""" """
@@ -391,6 +411,9 @@ class VirtualWorkbench:
"material_id": material_id, "material_id": material_id,
"material_number": material_number, "material_number": material_number,
"message": f"{material_id}已成功移动到加热台{station_id}", "message": f"{material_id}已成功移动到加热台{station_id}",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
except Exception as e: except Exception as e:
@@ -403,10 +426,15 @@ class VirtualWorkbench:
"material_id": material_id, "material_id": material_id,
"material_number": material_number, "material_number": material_number,
"message": f"移动失败: {str(e)}", "message": f"移动失败: {str(e)}",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
@always_free
def start_heating( def start_heating(
self, self,
sample_uuids: SampleUUIDsType,
station_id: int, station_id: int,
material_number: int, material_number: int,
) -> StartHeatingResult: ) -> StartHeatingResult:
@@ -429,6 +457,9 @@ class VirtualWorkbench:
"material_id": "", "material_id": "",
"material_number": material_number, "material_number": material_number,
"message": f"无效的加热台ID: {station_id}", "message": f"无效的加热台ID: {station_id}",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
with self._stations_lock: with self._stations_lock:
@@ -441,6 +472,9 @@ class VirtualWorkbench:
"material_id": "", "material_id": "",
"material_number": material_number, "material_number": material_number,
"message": f"加热台{station_id}上没有物料", "message": f"加热台{station_id}上没有物料",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
if station.state == HeatingStationState.HEATING: if station.state == HeatingStationState.HEATING:
@@ -450,6 +484,9 @@ class VirtualWorkbench:
"material_id": station.current_material, "material_id": station.current_material,
"material_number": material_number, "material_number": material_number,
"message": f"加热台{station_id}已经在加热中", "message": f"加热台{station_id}已经在加热中",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
material_id = station.current_material material_id = station.current_material
@@ -465,10 +502,21 @@ class VirtualWorkbench:
self._update_data_status(f"加热台{station_id}开始加热{material_id}") self._update_data_status(f"加热台{station_id}开始加热{material_id}")
# 模拟加热过程 (10秒) # 打印当前所有正在加热的台位
with self._stations_lock:
heating_list = [
f"加热台{sid}:{s.current_material}"
for sid, s in self._heating_stations.items()
if s.state == HeatingStationState.HEATING and s.current_material
]
self.logger.info(f"[并行加热] 当前同时加热中: {', '.join(heating_list)}")
# 模拟加热过程
start_time = time.time() start_time = time.time()
last_countdown_log = start_time
while True: while True:
elapsed = time.time() - start_time elapsed = time.time() - start_time
remaining = max(0.0, self.HEATING_TIME - elapsed)
progress = min(100.0, (elapsed / self.HEATING_TIME) * 100) progress = min(100.0, (elapsed / self.HEATING_TIME) * 100)
with self._stations_lock: with self._stations_lock:
@@ -476,6 +524,11 @@ class VirtualWorkbench:
self._update_data_status(f"加热台{station_id}加热中: {progress:.1f}%") self._update_data_status(f"加热台{station_id}加热中: {progress:.1f}%")
# 每5秒打印一次倒计时
if time.time() - last_countdown_log >= 5.0:
self.logger.info(f"[加热台{station_id}] {material_id} 剩余 {remaining:.1f}s")
last_countdown_log = time.time()
if elapsed >= self.HEATING_TIME: if elapsed >= self.HEATING_TIME:
break break
@@ -499,10 +552,14 @@ class VirtualWorkbench:
"material_id": material_id, "material_id": material_id,
"material_number": material_number, "material_number": material_number,
"message": f"加热台{station_id}加热完成", "message": f"加热台{station_id}加热完成",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
def move_to_output( def move_to_output(
self, self,
sample_uuids: SampleUUIDsType,
station_id: int, station_id: int,
material_number: int, material_number: int,
) -> MoveToOutputResult: ) -> MoveToOutputResult:
@@ -525,6 +582,9 @@ class VirtualWorkbench:
"material_id": "", "material_id": "",
"output_position": f"C{output_number}", "output_position": f"C{output_number}",
"message": f"无效的加热台ID: {station_id}", "message": f"无效的加热台ID: {station_id}",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
with self._stations_lock: with self._stations_lock:
@@ -538,6 +598,9 @@ class VirtualWorkbench:
"material_id": "", "material_id": "",
"output_position": f"C{output_number}", "output_position": f"C{output_number}",
"message": f"加热台{station_id}上没有物料", "message": f"加热台{station_id}上没有物料",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
if station.state != HeatingStationState.COMPLETED: if station.state != HeatingStationState.COMPLETED:
@@ -547,6 +610,9 @@ class VirtualWorkbench:
"material_id": material_id, "material_id": material_id,
"output_position": f"C{output_number}", "output_position": f"C{output_number}",
"message": f"加热台{station_id}尚未完成加热 (当前状态: {station.state.value})", "message": f"加热台{station_id}尚未完成加热 (当前状态: {station.state.value})",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
output_position = f"C{output_number}" output_position = f"C{output_number}"
@@ -595,6 +661,9 @@ class VirtualWorkbench:
"material_id": material_id, "material_id": material_id,
"output_position": output_position, "output_position": output_position,
"message": f"{material_id}已成功移动到{output_position}", "message": f"{material_id}已成功移动到{output_position}",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
except Exception as e: except Exception as e:
@@ -607,6 +676,9 @@ class VirtualWorkbench:
"material_id": "", "material_id": "",
"output_position": output_position, "output_position": output_position,
"message": f"移动失败: {str(e)}", "message": f"移动失败: {str(e)}",
"unilabos_samples": [
LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
sample_uuid, content in sample_uuids.items()]
} }
# ============ 状态属性 ============ # ============ 状态属性 ============

View File

@@ -9468,7 +9468,7 @@ liquid_handler.prcxi:
well_names: null well_names: null
handles: handles:
input: input:
- data_key: plate - data_key: '@this.0@@@plate'
data_source: handle data_source: handle
data_type: resource data_type: resource
handler_key: input_plate handler_key: input_plate
@@ -9503,7 +9503,6 @@ liquid_handler.prcxi:
type: string type: string
type: array type: array
plate: plate:
items:
properties: properties:
category: category:
type: string type: string
@@ -9576,8 +9575,6 @@ liquid_handler.prcxi:
- data - data
title: plate title: plate
type: object type: object
title: plate
type: array
volumes: volumes:
items: items:
type: number type: number
@@ -9593,17 +9590,207 @@ liquid_handler.prcxi:
- volumes - volumes
type: object type: object
result: result:
$defs:
ResourceDict:
properties:
class:
description: Resource class name
title: Class
type: string
config:
additionalProperties: true
description: Resource configuration
title: Config
type: object
data:
additionalProperties: true
description: 'Resource data, eg: container liquid data'
title: Data
type: object
description:
default: ''
description: Resource description
title: Description
type: string
extra:
additionalProperties: true
description: 'Extra data, eg: slot index'
title: Extra
type: object
icon:
default: ''
description: Resource icon
title: Icon
type: string
id:
description: Resource ID
title: Id
type: string
model:
additionalProperties: true
description: Resource model
title: Model
type: object
name:
description: Resource name
title: Name
type: string
parent:
anyOf:
- $ref: '#/$defs/ResourceDict'
- type: 'null'
default: null
description: Parent resource object
parent_uuid:
anyOf:
- type: string
- type: 'null'
default: null
description: Parent resource uuid
title: Parent Uuid
pose:
$ref: '#/$defs/ResourceDictPosition'
description: Resource position
schema:
additionalProperties: true
description: Resource schema
title: Schema
type: object
type:
anyOf:
- const: device
type: string
- type: string
description: Resource type
title: Type
uuid:
description: Resource UUID
title: Uuid
type: string
required:
- id
- uuid
- name
- type
- class
- config
- data
- extra
title: ResourceDict
type: object
ResourceDictPosition:
properties:
cross_section_type:
default: rectangle
description: Cross section type
enum:
- rectangle
- circle
- rounded_rectangle
title: Cross Section Type
type: string
layout:
default: x-y
description: Resource layout
enum:
- 2d
- x-y
- z-y
- x-z
title: Layout
type: string
position:
$ref: '#/$defs/ResourceDictPositionObject'
description: Resource position
position3d:
$ref: '#/$defs/ResourceDictPositionObject'
description: Resource position in 3D space
rotation:
$ref: '#/$defs/ResourceDictPositionObject'
description: Resource rotation
scale:
$ref: '#/$defs/ResourceDictPositionScale'
description: Resource scale
size:
$ref: '#/$defs/ResourceDictPositionSize'
description: Resource size
title: ResourceDictPosition
type: object
ResourceDictPositionObject:
properties:
x:
default: 0.0
description: X coordinate
title: X
type: number
y:
default: 0.0
description: Y coordinate
title: Y
type: number
z:
default: 0.0
description: Z coordinate
title: Z
type: number
title: ResourceDictPositionObject
type: object
ResourceDictPositionScale:
properties:
x:
default: 0.0
description: x scale
title: X
type: number
y:
default: 0.0
description: y scale
title: Y
type: number
z:
default: 0.0
description: z scale
title: Z
type: number
title: ResourceDictPositionScale
type: object
ResourceDictPositionSize:
properties:
depth:
default: 0.0
description: Depth
title: Depth
type: number
height:
default: 0.0
description: Height
title: Height
type: number
width:
default: 0.0
description: Width
title: Width
type: number
title: ResourceDictPositionSize
type: object
properties: properties:
plate: plate:
items: {} items:
items:
$ref: '#/$defs/ResourceDict'
type: array
title: Plate title: Plate
type: array type: array
volumes: volumes:
items: {} items:
type: number
title: Volumes title: Volumes
type: array type: array
wells: wells:
items: {} items:
items:
$ref: '#/$defs/ResourceDict'
type: array
title: Wells title: Wells
type: array type: array
required: required:

View File

@@ -5835,6 +5835,25 @@ virtual_workbench:
- material_number - material_number
type: object type: object
result: result:
$defs:
LabSample:
properties:
extra:
additionalProperties: true
title: Extra
type: object
oss_path:
title: Oss Path
type: string
sample_uuid:
title: Sample Uuid
type: string
required:
- sample_uuid
- oss_path
- extra
title: LabSample
type: object
description: move_to_heating_station 返回类型 description: move_to_heating_station 返回类型
properties: properties:
material_id: material_id:
@@ -5853,12 +5872,18 @@ virtual_workbench:
success: success:
title: Success title: Success
type: boolean type: boolean
unilabos_samples:
items:
$ref: '#/$defs/LabSample'
title: Unilabos Samples
type: array
required: required:
- success - success
- station_id - station_id
- material_id - material_id
- material_number - material_number
- message - message
- unilabos_samples
title: MoveToHeatingStationResult title: MoveToHeatingStationResult
type: object type: object
required: required:
@@ -5903,6 +5928,25 @@ virtual_workbench:
- material_number - material_number
type: object type: object
result: result:
$defs:
LabSample:
properties:
extra:
additionalProperties: true
title: Extra
type: object
oss_path:
title: Oss Path
type: string
sample_uuid:
title: Sample Uuid
type: string
required:
- sample_uuid
- oss_path
- extra
title: LabSample
type: object
description: move_to_output 返回类型 description: move_to_output 返回类型
properties: properties:
material_id: material_id:
@@ -5914,10 +5958,16 @@ virtual_workbench:
success: success:
title: Success title: Success
type: boolean type: boolean
unilabos_samples:
items:
$ref: '#/$defs/LabSample'
title: Unilabos Samples
type: array
required: required:
- success - success
- station_id - station_id
- material_id - material_id
- unilabos_samples
title: MoveToOutputResult title: MoveToOutputResult
type: object type: object
required: required:
@@ -5972,6 +6022,25 @@ virtual_workbench:
required: [] required: []
type: object type: object
result: result:
$defs:
LabSample:
properties:
extra:
additionalProperties: true
title: Extra
type: object
oss_path:
title: Oss Path
type: string
sample_uuid:
title: Sample Uuid
type: string
required:
- sample_uuid
- oss_path
- extra
title: LabSample
type: object
description: prepare_materials 返回类型 - 批量准备物料 description: prepare_materials 返回类型 - 批量准备物料
properties: properties:
count: count:
@@ -5998,6 +6067,11 @@ virtual_workbench:
success: success:
title: Success title: Success
type: boolean type: boolean
unilabos_samples:
items:
$ref: '#/$defs/LabSample'
title: Unilabos Samples
type: array
required: required:
- success - success
- count - count
@@ -6007,6 +6081,7 @@ virtual_workbench:
- material_4 - material_4
- material_5 - material_5
- message - message
- unilabos_samples
title: PrepareMaterialsResult title: PrepareMaterialsResult
type: object type: object
required: required:
@@ -6015,6 +6090,7 @@ virtual_workbench:
type: object type: object
type: UniLabJsonCommand type: UniLabJsonCommand
auto-start_heating: auto-start_heating:
always_free: true
feedback: {} feedback: {}
goal: {} goal: {}
goal_default: goal_default:
@@ -6062,6 +6138,25 @@ virtual_workbench:
- material_number - material_number
type: object type: object
result: result:
$defs:
LabSample:
properties:
extra:
additionalProperties: true
title: Extra
type: object
oss_path:
title: Oss Path
type: string
sample_uuid:
title: Sample Uuid
type: string
required:
- sample_uuid
- oss_path
- extra
title: LabSample
type: object
description: start_heating 返回类型 description: start_heating 返回类型
properties: properties:
material_id: material_id:
@@ -6079,12 +6174,18 @@ virtual_workbench:
success: success:
title: Success title: Success
type: boolean type: boolean
unilabos_samples:
items:
$ref: '#/$defs/LabSample'
title: Unilabos Samples
type: array
required: required:
- success - success
- station_id - station_id
- material_id - material_id
- material_number - material_number
- message - message
- unilabos_samples
title: StartHeatingResult title: StartHeatingResult
type: object type: object
required: required:

View File

@@ -838,6 +838,7 @@ class Registry:
("list", "unilabos.registry.placeholder_type:DeviceSlot"), ("list", "unilabos.registry.placeholder_type:DeviceSlot"),
] ]
}, },
**({"always_free": True} if v.get("always_free") else {}),
} }
for k, v in enhanced_info["action_methods"].items() for k, v in enhanced_info["action_methods"].items()
if k not in device_config["class"]["action_value_mappings"] if k not in device_config["class"]["action_value_mappings"]

View File

@@ -151,12 +151,40 @@ def canonicalize_links_ports(links: List[Dict[str, Any]], resource_tree_set: Res
""" """
# 构建 id 到 uuid 的映射 # 构建 id 到 uuid 的映射
id_to_uuid: Dict[str, str] = {} id_to_uuid: Dict[str, str] = {}
uuid_to_id: Dict[str, str] = {}
for node in resource_tree_set.all_nodes: for node in resource_tree_set.all_nodes:
id_to_uuid[node.res_content.id] = node.res_content.uuid id_to_uuid[node.res_content.id] = node.res_content.uuid
uuid_to_id[node.res_content.uuid] = node.res_content.id
# 第三遍处理:为每个 link 添加 source_uuid 和 target_uuid
for link in links:
source_id = link.get("source")
target_id = link.get("target")
# 添加 source_uuid
if source_id and source_id in id_to_uuid:
link["source_uuid"] = id_to_uuid[source_id]
# 添加 target_uuid
if target_id and target_id in id_to_uuid:
link["target_uuid"] = id_to_uuid[target_id]
source_uuid = link.get("source_uuid")
target_uuid = link.get("target_uuid")
# 添加 source_uuid
if source_uuid and source_uuid in uuid_to_id:
link["source"] = uuid_to_id[source_uuid]
# 添加 target_uuid
if target_uuid and target_uuid in uuid_to_id:
link["target"] = uuid_to_id[target_uuid]
# 第一遍处理将字符串类型的port转换为字典格式 # 第一遍处理将字符串类型的port转换为字典格式
for link in links: for link in links:
port = link.get("port") port = link.get("port")
if port is None:
continue
if link.get("type", "physical") == "physical": if link.get("type", "physical") == "physical":
link["type"] = "fluid" link["type"] = "fluid"
if isinstance(port, int): if isinstance(port, int):
@@ -179,13 +207,15 @@ def canonicalize_links_ports(links: List[Dict[str, Any]], resource_tree_set: Res
link["port"] = {link["source"]: None, link["target"]: None} link["port"] = {link["source"]: None, link["target"]: None}
# 构建边字典,键为(source节点, target节点)值为对应的port信息 # 构建边字典,键为(source节点, target节点)值为对应的port信息
edges = {(link["source"], link["target"]): link["port"] for link in links} edges = {(link["source"], link["target"]): link["port"] for link in links if link.get("port")}
# 第二遍处理填充反向边的dest信息 # 第二遍处理填充反向边的dest信息
delete_reverses = [] delete_reverses = []
for i, link in enumerate(links): for i, link in enumerate(links):
s, t = link["source"], link["target"] s, t = link["source"], link["target"]
current_port = link["port"] current_port = link.get("port")
if current_port is None:
continue
if current_port.get(t) is None: if current_port.get(t) is None:
reverse_key = (t, s) reverse_key = (t, s)
reverse_port = edges.get(reverse_key) reverse_port = edges.get(reverse_key)
@@ -200,20 +230,6 @@ def canonicalize_links_ports(links: List[Dict[str, Any]], resource_tree_set: Res
current_port[t] = current_port[s] current_port[t] = current_port[s]
# 删除已被使用反向端口信息的反向边 # 删除已被使用反向端口信息的反向边
standardized_links = [link for i, link in enumerate(links) if i not in delete_reverses] standardized_links = [link for i, link in enumerate(links) if i not in delete_reverses]
# 第三遍处理:为每个 link 添加 source_uuid 和 target_uuid
for link in standardized_links:
source_id = link.get("source")
target_id = link.get("target")
# 添加 source_uuid
if source_id and source_id in id_to_uuid:
link["source_uuid"] = id_to_uuid[source_id]
# 添加 target_uuid
if target_id and target_id in id_to_uuid:
link["target_uuid"] = id_to_uuid[target_id]
return standardized_links return standardized_links
@@ -284,6 +300,8 @@ def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]
edge["sourceHandle"] = port[source] edge["sourceHandle"] = port[source]
elif "source_port" in edge: elif "source_port" in edge:
edge["sourceHandle"] = edge.pop("source_port") edge["sourceHandle"] = edge.pop("source_port")
elif "source_handle" in edge:
edge["sourceHandle"] = edge.pop("source_handle")
else: else:
typ = edge.get("type") typ = edge.get("type")
if typ == "communication": if typ == "communication":
@@ -292,6 +310,8 @@ def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]
edge["targetHandle"] = port[target] edge["targetHandle"] = port[target]
elif "target_port" in edge: elif "target_port" in edge:
edge["targetHandle"] = edge.pop("target_port") edge["targetHandle"] = edge.pop("target_port")
elif "target_handle" in edge:
edge["targetHandle"] = edge.pop("target_handle")
else: else:
typ = edge.get("type") typ = edge.get("type")
if typ == "communication": if typ == "communication":

View File

@@ -5,6 +5,8 @@ from pydantic import BaseModel, field_serializer, field_validator, ValidationErr
from pydantic import Field from pydantic import Field
from typing import List, Tuple, Any, Dict, Literal, Optional, cast, TYPE_CHECKING, Union from typing import List, Tuple, Any, Dict, Literal, Optional, cast, TYPE_CHECKING, Union
from typing_extensions import TypedDict
from unilabos.resources.plr_additional_res_reg import register from unilabos.resources.plr_additional_res_reg import register
from unilabos.utils.log import logger from unilabos.utils.log import logger
@@ -14,6 +16,26 @@ if TYPE_CHECKING:
EXTRA_CLASS = "unilabos_resource_class" EXTRA_CLASS = "unilabos_resource_class"
EXTRA_SAMPLE_UUID = "sample_uuid"
EXTRA_UNILABOS_SAMPLE_UUID = "unilabos_sample_uuid"
# 函数参数名常量 - 用于自动注入 sample_uuids 列表
PARAM_SAMPLE_UUIDS = "sample_uuids"
# JSON Command 中的系统参数字段名
JSON_UNILABOS_PARAM = "unilabos_param"
# 返回值中的 samples 字段名
RETURN_UNILABOS_SAMPLES = "unilabos_samples"
# sample_uuids 参数类型 (用于 virtual bench 等设备添加 sample_uuids 参数)
SampleUUIDsType = Dict[str, Optional["PLRResource"]]
class LabSample(TypedDict):
sample_uuid: str
oss_path: str
extra: Dict[str, Any]
class ResourceDictPositionSize(BaseModel): class ResourceDictPositionSize(BaseModel):
@@ -529,6 +551,7 @@ class ResourceTreeSet(object):
plr_resource = sub_cls.deserialize(plr_dict, allow_marshal=True) plr_resource = sub_cls.deserialize(plr_dict, allow_marshal=True)
from pylabrobot.resources import Coordinate from pylabrobot.resources import Coordinate
from pylabrobot.serializer import deserialize from pylabrobot.serializer import deserialize
location = cast(Coordinate, deserialize(plr_dict["location"])) location = cast(Coordinate, deserialize(plr_dict["location"]))
plr_resource.location = location plr_resource.location = location
plr_resource.load_all_state(all_states) plr_resource.load_all_state(all_states)

View File

@@ -44,8 +44,7 @@ def ros2_device_node(
# 从属性中自动发现可发布状态 # 从属性中自动发现可发布状态
if status_types is None: if status_types is None:
status_types = {} status_types = {}
if device_config is None: assert device_config is not None, "device_config cannot be None"
raise ValueError("device_config cannot be None")
if action_value_mappings is None: if action_value_mappings is None:
action_value_mappings = {} action_value_mappings = {}
if hardware_interface is None: if hardware_interface is None:

View File

@@ -4,8 +4,20 @@ import json
import threading import threading
import time import time
import traceback import traceback
from typing import get_type_hints, TypeVar, Generic, Dict, Any, Type, TypedDict, Optional, List, TYPE_CHECKING, Union, \ from typing import (
Tuple get_type_hints,
TypeVar,
Generic,
Dict,
Any,
Type,
TypedDict,
Optional,
List,
TYPE_CHECKING,
Union,
Tuple,
)
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
import asyncio import asyncio
@@ -48,6 +60,9 @@ from unilabos.resources.resource_tracker import (
ResourceTreeSet, ResourceTreeSet,
ResourceTreeInstance, ResourceTreeInstance,
ResourceDictInstance, ResourceDictInstance,
EXTRA_SAMPLE_UUID,
PARAM_SAMPLE_UUIDS,
JSON_UNILABOS_PARAM,
) )
from unilabos.ros.utils.driver_creator import WorkstationNodeCreator, PyLabRobotCreator, DeviceClassCreator from unilabos.ros.utils.driver_creator import WorkstationNodeCreator, PyLabRobotCreator, DeviceClassCreator
from rclpy.task import Task, Future from rclpy.task import Task, Future
@@ -131,7 +146,7 @@ def init_wrapper(
device_id: str, device_id: str,
device_uuid: str, device_uuid: str,
driver_class: type[T], driver_class: type[T],
device_config: ResourceTreeInstance, device_config: ResourceDictInstance,
status_types: Dict[str, Any], status_types: Dict[str, Any],
action_value_mappings: Dict[str, Any], action_value_mappings: Dict[str, Any],
hardware_interface: Dict[str, Any], hardware_interface: Dict[str, Any],
@@ -216,14 +231,15 @@ class PropertyPublisher:
def publish_property(self): def publish_property(self):
try: try:
self.node.lab_logger().trace(f"【.publish_property】开始发布属性: {self.name}") # self.node.lab_logger().trace(f"【.publish_property】开始发布属性: {self.name}")
value = self.get_property() value = self.get_property()
if self.print_publish: if self.print_publish:
self.node.lab_logger().trace(f"【.publish_property】发布 {self.msg_type}: {value}") pass
# self.node.lab_logger().trace(f"【.publish_property】发布 {self.msg_type}: {value}")
if value is not None: if value is not None:
msg = convert_to_ros_msg(self.msg_type, value) msg = convert_to_ros_msg(self.msg_type, value)
self.publisher_.publish(msg) self.publisher_.publish(msg)
self.node.lab_logger().trace(f"【.publish_property】属性 {self.name} 发布成功") # self.node.lab_logger().trace(f"【.publish_property】属性 {self.name} 发布成功")
except Exception as e: except Exception as e:
self.node.lab_logger().error( self.node.lab_logger().error(
f"【.publish_property】发布属性 {self.publisher_.topic} 出错: {str(e)}\n{traceback.format_exc()}" f"【.publish_property】发布属性 {self.publisher_.topic} 出错: {str(e)}\n{traceback.format_exc()}"
@@ -263,6 +279,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
self, self,
driver_instance: T, driver_instance: T,
device_id: str, device_id: str,
registry_name: str,
device_uuid: str, device_uuid: str,
status_types: Dict[str, Any], status_types: Dict[str, Any],
action_value_mappings: Dict[str, Any], action_value_mappings: Dict[str, Any],
@@ -284,6 +301,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
""" """
self.driver_instance = driver_instance self.driver_instance = driver_instance
self.device_id = device_id self.device_id = device_id
self.registry_name = registry_name
self.uuid = device_uuid self.uuid = device_uuid
self.publish_high_frequency = False self.publish_high_frequency = False
self.callback_group = ReentrantCallbackGroup() self.callback_group = ReentrantCallbackGroup()
@@ -361,6 +379,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
from pylabrobot.resources.deck import Deck from pylabrobot.resources.deck import Deck
from pylabrobot.resources import Coordinate from pylabrobot.resources import Coordinate
from pylabrobot.resources import Plate from pylabrobot.resources import Plate
# 物料传输到对应的node节点 # 物料传输到对应的node节点
client = self._resource_clients["c2s_update_resource_tree"] client = self._resource_clients["c2s_update_resource_tree"]
request = SerialCommand.Request() request = SerialCommand.Request()
@@ -388,33 +407,29 @@ class BaseROS2DeviceNode(Node, Generic[T]):
rts: ResourceTreeSet = ResourceTreeSet.from_raw_dict_list(input_resources) rts: ResourceTreeSet = ResourceTreeSet.from_raw_dict_list(input_resources)
parent_resource = None parent_resource = None
if bind_parent_id != self.node_name: if bind_parent_id != self.node_name:
parent_resource = self.resource_tracker.figure_resource( parent_resource = self.resource_tracker.figure_resource({"name": bind_parent_id})
{"name": bind_parent_id}
)
for r in rts.root_nodes: for r in rts.root_nodes:
# noinspection PyUnresolvedReferences # noinspection PyUnresolvedReferences
r.res_content.parent_uuid = parent_resource.unilabos_uuid r.res_content.parent_uuid = parent_resource.unilabos_uuid
else: else:
for r in rts.root_nodes: for r in rts.root_nodes:
r.res_content.parent_uuid = self.uuid r.res_content.parent_uuid = self.uuid
rts_plr_instances = rts.to_plr_resources()
if len(LIQUID_INPUT_SLOT) and LIQUID_INPUT_SLOT[0] == -1 and len(rts.root_nodes) == 1 and isinstance(rts.root_nodes[0], RegularContainer): if len(rts.root_nodes) == 1 and isinstance(rts_plr_instances[0], RegularContainer):
# noinspection PyTypeChecker # noinspection PyTypeChecker
container_instance: RegularContainer = rts.root_nodes[0] container_instance: RegularContainer = rts_plr_instances[0]
found_resources = self.resource_tracker.figure_resource( found_resources = self.resource_tracker.figure_resource(
{"id": container_instance.name}, try_mode=True {"name": container_instance.name}, try_mode=True
) )
if not len(found_resources): if not len(found_resources):
self.resource_tracker.add_resource(container_instance) self.resource_tracker.add_resource(container_instance)
logger.info(f"添加物料{container_instance.name}到资源跟踪器") logger.info(f"添加物料{container_instance.name}到资源跟踪器")
else: else:
assert ( assert len(found_resources) == 1, f"找到多个同名物料: {container_instance.name}, 请检查物料系统"
len(found_resources) == 1
), f"找到多个同名物料: {container_instance.name}, 请检查物料系统"
found_resource = found_resources[0] found_resource = found_resources[0]
if isinstance(found_resource, RegularContainer): if isinstance(found_resource, RegularContainer):
logger.info(f"更新物料{container_instance.name}的数据{found_resource.state}") logger.info(f"更新物料{container_instance.name}的数据{found_resource.state}")
found_resource.state.update(json.loads(container_instance.state)) found_resource.state.update(container_instance.state)
elif isinstance(found_resource, dict): elif isinstance(found_resource, dict):
raise ValueError("已不支持 字典 版本的RegularContainer") raise ValueError("已不支持 字典 版本的RegularContainer")
else: else:
@@ -422,14 +437,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
f"更新物料{container_instance.name}出现不支持的数据类型{type(found_resource)} {found_resource}" f"更新物料{container_instance.name}出现不支持的数据类型{type(found_resource)} {found_resource}"
) )
# noinspection PyUnresolvedReferences # noinspection PyUnresolvedReferences
request.command = json.dumps({ request.command = json.dumps(
{
"action": "add", "action": "add",
"data": { "data": {
"data": rts.dump(), "data": rts.dump(),
"mount_uuid": parent_resource.unilabos_uuid if parent_resource is not None else "", "mount_uuid": parent_resource.unilabos_uuid if parent_resource is not None else self.uuid,
"first_add": False, "first_add": False,
}, },
}) }
)
tree_response: SerialCommand.Response = await client.call_async(request) tree_response: SerialCommand.Response = await client.call_async(request)
uuid_maps = json.loads(tree_response.response) uuid_maps = json.loads(tree_response.response)
plr_instances = rts.to_plr_resources() plr_instances = rts.to_plr_resources()
@@ -471,7 +488,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
if len(ADD_LIQUID_TYPE) == 1 and len(LIQUID_VOLUME) == 1 and len(LIQUID_INPUT_SLOT) > 1: if len(ADD_LIQUID_TYPE) == 1 and len(LIQUID_VOLUME) == 1 and len(LIQUID_INPUT_SLOT) > 1:
ADD_LIQUID_TYPE = ADD_LIQUID_TYPE * len(LIQUID_INPUT_SLOT) ADD_LIQUID_TYPE = ADD_LIQUID_TYPE * len(LIQUID_INPUT_SLOT)
LIQUID_VOLUME = LIQUID_VOLUME * len(LIQUID_INPUT_SLOT) LIQUID_VOLUME = LIQUID_VOLUME * len(LIQUID_INPUT_SLOT)
self.lab_logger().warning(f"增加液体资源时数量为1自动补全为 {len(LIQUID_INPUT_SLOT)}") self.lab_logger().warning(
f"增加液体资源时数量为1自动补全为 {len(LIQUID_INPUT_SLOT)}"
)
for liquid_type, liquid_volume, liquid_input_slot in zip( for liquid_type, liquid_volume, liquid_input_slot in zip(
ADD_LIQUID_TYPE, LIQUID_VOLUME, LIQUID_INPUT_SLOT ADD_LIQUID_TYPE, LIQUID_VOLUME, LIQUID_INPUT_SLOT
): ):
@@ -490,9 +509,15 @@ class BaseROS2DeviceNode(Node, Generic[T]):
input_wells = [] input_wells = []
for r in LIQUID_INPUT_SLOT: for r in LIQUID_INPUT_SLOT:
input_wells.append(plr_instance.children[r]) input_wells.append(plr_instance.children[r])
final_response["liquid_input_resource_tree"] = ResourceTreeSet.from_plr_resources(input_wells).dump() final_response["liquid_input_resource_tree"] = ResourceTreeSet.from_plr_resources(
input_wells
).dump()
res.response = json.dumps(final_response) res.response = json.dumps(final_response)
if issubclass(parent_resource.__class__, Deck) and hasattr(parent_resource, "assign_child_at_slot") and "slot" in other_calling_param: if (
issubclass(parent_resource.__class__, Deck)
and hasattr(parent_resource, "assign_child_at_slot")
and "slot" in other_calling_param
):
other_calling_param["slot"] = int(other_calling_param["slot"]) other_calling_param["slot"] = int(other_calling_param["slot"])
parent_resource.assign_child_at_slot(plr_instance, **other_calling_param) parent_resource.assign_child_at_slot(plr_instance, **other_calling_param)
else: else:
@@ -507,14 +532,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
rts_with_parent = ResourceTreeSet.from_plr_resources([parent_resource]) rts_with_parent = ResourceTreeSet.from_plr_resources([parent_resource])
if rts_with_parent.root_nodes[0].res_content.uuid_parent is None: if rts_with_parent.root_nodes[0].res_content.uuid_parent is None:
rts_with_parent.root_nodes[0].res_content.parent_uuid = self.uuid rts_with_parent.root_nodes[0].res_content.parent_uuid = self.uuid
request.command = json.dumps({ request.command = json.dumps(
{
"action": "add", "action": "add",
"data": { "data": {
"data": rts_with_parent.dump(), "data": rts_with_parent.dump(),
"mount_uuid": rts_with_parent.root_nodes[0].res_content.uuid_parent, "mount_uuid": rts_with_parent.root_nodes[0].res_content.uuid_parent,
"first_add": False, "first_add": False,
}, },
}) }
)
tree_response: SerialCommand.Response = await client.call_async(request) tree_response: SerialCommand.Response = await client.call_async(request)
uuid_maps = json.loads(tree_response.response) uuid_maps = json.loads(tree_response.response)
self.resource_tracker.loop_update_uuid(input_resources, uuid_maps) self.resource_tracker.loop_update_uuid(input_resources, uuid_maps)
@@ -811,7 +838,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
} }
def _handle_update( def _handle_update(
plr_resources: List[Union[ResourcePLR, ResourceDictInstance]], tree_set: ResourceTreeSet, additional_add_params: Dict[str, Any] plr_resources: List[Union[ResourcePLR, ResourceDictInstance]],
tree_set: ResourceTreeSet,
additional_add_params: Dict[str, Any],
) -> Tuple[Dict[str, Any], List[ResourcePLR]]: ) -> Tuple[Dict[str, Any], List[ResourcePLR]]:
""" """
处理资源更新操作的内部函数 处理资源更新操作的内部函数
@@ -836,7 +865,10 @@ class BaseROS2DeviceNode(Node, Generic[T]):
original_parent_resource = original_instance.parent original_parent_resource = original_instance.parent
original_parent_resource_uuid = getattr(original_parent_resource, "unilabos_uuid", None) original_parent_resource_uuid = getattr(original_parent_resource, "unilabos_uuid", None)
target_parent_resource_uuid = tree.root_node.res_content.uuid_parent target_parent_resource_uuid = tree.root_node.res_content.uuid_parent
not_same_parent = original_parent_resource_uuid != target_parent_resource_uuid and original_parent_resource is not None not_same_parent = (
original_parent_resource_uuid != target_parent_resource_uuid
and original_parent_resource is not None
)
old_name = original_instance.name old_name = original_instance.name
new_name = plr_resource.name new_name = plr_resource.name
parent_appended = False parent_appended = False
@@ -872,8 +904,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
else: else:
# 判断是否变更了resource_site重新登记 # 判断是否变更了resource_site重新登记
target_site = original_instance.unilabos_extra.get("update_resource_site") target_site = original_instance.unilabos_extra.get("update_resource_site")
sites = original_instance.parent.sites if original_instance.parent is not None and hasattr(original_instance.parent, "sites") else None sites = (
site_names = list(original_instance.parent._ordering.keys()) if original_instance.parent is not None and hasattr(original_instance.parent, "sites") else [] original_instance.parent.sites
if original_instance.parent is not None and hasattr(original_instance.parent, "sites")
else None
)
site_names = (
list(original_instance.parent._ordering.keys())
if original_instance.parent is not None and hasattr(original_instance.parent, "sites")
else []
)
if target_site is not None and sites is not None and site_names is not None: if target_site is not None and sites is not None and site_names is not None:
site_index = sites.index(original_instance) site_index = sites.index(original_instance)
site_name = site_names[site_index] site_name = site_names[site_index]
@@ -910,9 +950,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
action = i.get("action") # remove, add, update action = i.get("action") # remove, add, update
resources_uuid: List[str] = i.get("data") # 资源数据 resources_uuid: List[str] = i.get("data") # 资源数据
additional_add_params = i.get("additional_add_params", {}) # 额外参数 additional_add_params = i.get("additional_add_params", {}) # 额外参数
self.lab_logger().trace( self.lab_logger().trace(f"[资源同步] 处理 {action}, " f"resources count: {len(resources_uuid)}")
f"[资源同步] 处理 {action}, " f"resources count: {len(resources_uuid)}"
)
tree_set = None tree_set = None
if action in ["add", "update"]: if action in ["add", "update"]:
tree_set = await self.get_resource( tree_set = await self.get_resource(
@@ -939,9 +977,13 @@ class BaseROS2DeviceNode(Node, Generic[T]):
tree.root_node.res_content.parent_uuid = self.uuid tree.root_node.res_content.parent_uuid = self.uuid
r = SerialCommand.Request() r = SerialCommand.Request()
r.command = json.dumps( r.command = json.dumps(
{"data": {"data": new_tree_set.dump()}, "action": "update"}) # 和Update Resource一致 {"data": {"data": new_tree_set.dump()}, "action": "update"}
) # 和Update Resource一致
response: SerialCommand_Response = await self._resource_clients[ response: SerialCommand_Response = await self._resource_clients[
"c2s_update_resource_tree"].call_async(r) # type: ignore "c2s_update_resource_tree"
].call_async(
r
) # type: ignore
self.lab_logger().info(f"确认资源云端 Add 结果: {response.response}") self.lab_logger().info(f"确认资源云端 Add 结果: {response.response}")
results.append(result) results.append(result)
elif action == "update": elif action == "update":
@@ -961,9 +1003,13 @@ class BaseROS2DeviceNode(Node, Generic[T]):
tree.root_node.res_content.parent_uuid = self.uuid tree.root_node.res_content.parent_uuid = self.uuid
r = SerialCommand.Request() r = SerialCommand.Request()
r.command = json.dumps( r.command = json.dumps(
{"data": {"data": new_tree_set.dump()}, "action": "update"}) # 和Update Resource一致 {"data": {"data": new_tree_set.dump()}, "action": "update"}
) # 和Update Resource一致
response: SerialCommand_Response = await self._resource_clients[ response: SerialCommand_Response = await self._resource_clients[
"c2s_update_resource_tree"].call_async(r) # type: ignore "c2s_update_resource_tree"
].call_async(
r
) # type: ignore
self.lab_logger().info(f"确认资源云端 Update 结果: {response.response}") self.lab_logger().info(f"确认资源云端 Update 结果: {response.response}")
results.append(result) results.append(result)
elif action == "remove": elif action == "remove":
@@ -1110,6 +1156,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
"machine_name": BasicConfig.machine_name, "machine_name": BasicConfig.machine_name,
"type": "slave", "type": "slave",
"edge_device_id": self.device_id, "edge_device_id": self.device_id,
"registry_name": self.registry_name,
} }
}, },
ensure_ascii=False, ensure_ascii=False,
@@ -1333,7 +1380,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
resource_id=resource_data["id"], with_children=True resource_id=resource_data["id"], with_children=True
) )
if "sample_id" in resource_data: if "sample_id" in resource_data:
plr_resource.unilabos_extra["sample_uuid"] = resource_data["sample_id"] plr_resource.unilabos_extra[EXTRA_SAMPLE_UUID] = resource_data["sample_id"]
queried_resources[idx] = plr_resource queried_resources[idx] = plr_resource
else: else:
uuid_indices.append((idx, unilabos_uuid, resource_data)) uuid_indices.append((idx, unilabos_uuid, resource_data))
@@ -1346,7 +1393,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
for i, (idx, _, resource_data) in enumerate(uuid_indices): for i, (idx, _, resource_data) in enumerate(uuid_indices):
plr_resource = plr_resources[i] plr_resource = plr_resources[i]
if "sample_id" in resource_data: if "sample_id" in resource_data:
plr_resource.unilabos_extra["sample_uuid"] = resource_data["sample_id"] plr_resource.unilabos_extra[EXTRA_SAMPLE_UUID] = resource_data["sample_id"]
queried_resources[idx] = plr_resource queried_resources[idx] = plr_resource
self.lab_logger().debug(f"资源查询结果: 共 {len(queried_resources)} 个资源") self.lab_logger().debug(f"资源查询结果: 共 {len(queried_resources)} 个资源")
@@ -1354,7 +1401,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
# 通过资源跟踪器获取本地实例 # 通过资源跟踪器获取本地实例
final_resources = queried_resources if is_sequence else queried_resources[0] final_resources = queried_resources if is_sequence else queried_resources[0]
if not is_sequence: if not is_sequence:
plr = self.resource_tracker.figure_resource({"name": final_resources.name}, try_mode=False) plr = self.resource_tracker.figure_resource(
{"name": final_resources.name}, try_mode=False
)
# 保留unilabos_extra # 保留unilabos_extra
if hasattr(final_resources, "unilabos_extra") and hasattr(plr, "unilabos_extra"): if hasattr(final_resources, "unilabos_extra") and hasattr(plr, "unilabos_extra"):
plr.unilabos_extra = getattr(final_resources, "unilabos_extra", {}).copy() plr.unilabos_extra = getattr(final_resources, "unilabos_extra", {}).copy()
@@ -1393,8 +1442,12 @@ class BaseROS2DeviceNode(Node, Generic[T]):
execution_success = True execution_success = True
except Exception as _: except Exception as _:
execution_error = traceback.format_exc() execution_error = traceback.format_exc()
error(f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{str(action_kwargs)[:1000]}") error(
trace(f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{action_kwargs}") f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{str(action_kwargs)[:1000]}"
)
trace(
f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{action_kwargs}"
)
future = ROS2DeviceNode.run_async_func(ACTION, trace_error=False, **action_kwargs) future = ROS2DeviceNode.run_async_func(ACTION, trace_error=False, **action_kwargs)
future.add_done_callback(_handle_future_exception) future.add_done_callback(_handle_future_exception)
@@ -1414,9 +1467,11 @@ class BaseROS2DeviceNode(Node, Generic[T]):
except Exception as _: except Exception as _:
execution_error = traceback.format_exc() execution_error = traceback.format_exc()
error( error(
f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{str(action_kwargs)[:1000]}") f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{str(action_kwargs)[:1000]}"
)
trace( trace(
f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{action_kwargs}") f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入:{action_kwargs}"
)
future.add_done_callback(_handle_future_exception) future.add_done_callback(_handle_future_exception)
@@ -1483,8 +1538,15 @@ class BaseROS2DeviceNode(Node, Generic[T]):
if isinstance(rs, list): if isinstance(rs, list):
for r in rs: for r in rs:
res = self.resource_tracker.parent_resource(r) # 获取 resource 对象 res = self.resource_tracker.parent_resource(r) # 获取 resource 对象
if res is None:
res = rs
if id(res) not in seen:
seen.add(id(res))
unique_resources.append(res)
else: else:
res = self.resource_tracker.parent_resource(rs) res = self.resource_tracker.parent_resource(rs)
if res is None:
res = rs
if id(res) not in seen: if id(res) not in seen:
seen.add(id(res)) seen.add(id(res))
unique_resources.append(res) unique_resources.append(res)
@@ -1539,20 +1601,37 @@ class BaseROS2DeviceNode(Node, Generic[T]):
try: try:
function_name = target["function_name"] function_name = target["function_name"]
function_args = target["function_args"] function_args = target["function_args"]
# 获取 unilabos 系统参数
unilabos_param: Dict[str, Any] = target[JSON_UNILABOS_PARAM]
assert isinstance(function_args, dict), "执行动作时JSON必须为dict类型\n原JSON: {string}" assert isinstance(function_args, dict), "执行动作时JSON必须为dict类型\n原JSON: {string}"
function = getattr(self.driver_instance, function_name) function = getattr(self.driver_instance, function_name)
assert callable( assert callable(
function function
), f"执行动作时JSON中的function_name对应的函数不可调用: {function_name}\n原JSON: {string}" ), f"执行动作时JSON中的function_name对应的函数不可调用: {function_name}\n原JSON: {string}"
# 处理 ResourceSlot 类型参数 # 处理参数(包含 unilabos 系统参数如 sample_uuids
args_list = default_manager._analyze_method_signature(function)["args"] args_list = default_manager._analyze_method_signature(function, skip_unilabos_params=False)["args"]
for arg in args_list: for arg in args_list:
arg_name = arg["name"] arg_name = arg["name"]
arg_type = arg["type"] arg_type = arg["type"]
# 跳过不在 function_args 中的参数 # 跳过不在 function_args 中的参数
if arg_name not in function_args: if arg_name not in function_args:
# 处理 sample_uuids 参数注入
if arg_name == PARAM_SAMPLE_UUIDS:
raw_sample_uuids = unilabos_param.get(PARAM_SAMPLE_UUIDS, {})
# 将 material uuid 转换为 resource 实例
# key: sample_uuid, value: material_uuid -> resource 实例
resolved_sample_uuids: Dict[str, Any] = {}
for sample_uuid, material_uuid in raw_sample_uuids.items():
if material_uuid and self.resource_tracker:
resource = self.resource_tracker.uuid_to_resources.get(material_uuid)
resolved_sample_uuids[sample_uuid] = resource if resource else material_uuid
else:
resolved_sample_uuids[sample_uuid] = material_uuid
function_args[PARAM_SAMPLE_UUIDS] = resolved_sample_uuids
self.lab_logger().debug(f"[JsonCommand] 注入 {PARAM_SAMPLE_UUIDS}: {resolved_sample_uuids}")
continue continue
# 处理单个 ResourceSlot # 处理单个 ResourceSlot
@@ -1581,6 +1660,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
f"转换ResourceSlot列表参数 {arg_name} 失败: {e}\n{traceback.format_exc()}" f"转换ResourceSlot列表参数 {arg_name} 失败: {e}\n{traceback.format_exc()}"
) )
raise JsonCommandInitError(f"ResourceSlot列表参数转换失败: {arg_name}") raise JsonCommandInitError(f"ResourceSlot列表参数转换失败: {arg_name}")
# todo: 默认反报送 # todo: 默认反报送
return function(**function_args) return function(**function_args)
except KeyError as ex: except KeyError as ex:
@@ -1601,14 +1681,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
raise ValueError("至少需要提供一个 UUID") raise ValueError("至少需要提供一个 UUID")
uuids_list = list(uuids) uuids_list = list(uuids)
future = self._resource_clients["c2s_update_resource_tree"].call_async(SerialCommand.Request( future = self._resource_clients["c2s_update_resource_tree"].call_async(
SerialCommand.Request(
command=json.dumps( command=json.dumps(
{ {
"data": {"data": uuids_list, "with_children": True}, "data": {"data": uuids_list, "with_children": True},
"action": "get", "action": "get",
} }
) )
)) )
)
# 等待结果使用while循环每次sleep 0.05秒最多等待30秒 # 等待结果使用while循环每次sleep 0.05秒最多等待30秒
timeout = 30.0 timeout = 30.0
@@ -1666,6 +1748,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
try: try:
function_name = target["function_name"] function_name = target["function_name"]
function_args = target["function_args"] function_args = target["function_args"]
# 获取 unilabos 系统参数
unilabos_param: Dict[str, Any] = target.get(JSON_UNILABOS_PARAM, {})
assert isinstance(function_args, dict), "执行动作时JSON必须为dict类型\n原JSON: {string}" assert isinstance(function_args, dict), "执行动作时JSON必须为dict类型\n原JSON: {string}"
function = getattr(self.driver_instance, function_name) function = getattr(self.driver_instance, function_name)
assert callable( assert callable(
@@ -1675,14 +1760,30 @@ class BaseROS2DeviceNode(Node, Generic[T]):
function function
), f"执行动作时JSON中的function并非异步: {function_name}\n原JSON: {string}" ), f"执行动作时JSON中的function并非异步: {function_name}\n原JSON: {string}"
# 处理 ResourceSlot 类型参数 # 处理参数(包含 unilabos 系统参数如 sample_uuids
args_list = default_manager._analyze_method_signature(function)["args"] args_list = default_manager._analyze_method_signature(function, skip_unilabos_params=False)["args"]
for arg in args_list: for arg in args_list:
arg_name = arg["name"] arg_name = arg["name"]
arg_type = arg["type"] arg_type = arg["type"]
# 跳过不在 function_args 中的参数 # 跳过不在 function_args 中的参数
if arg_name not in function_args: if arg_name not in function_args:
# 处理 sample_uuids 参数注入
if arg_name == PARAM_SAMPLE_UUIDS:
raw_sample_uuids = unilabos_param.get(PARAM_SAMPLE_UUIDS, {})
# 将 material uuid 转换为 resource 实例
# key: sample_uuid, value: material_uuid -> resource 实例
resolved_sample_uuids: Dict[str, Any] = {}
for sample_uuid, material_uuid in raw_sample_uuids.items():
if material_uuid and self.resource_tracker:
resource = self.resource_tracker.uuid_to_resources.get(material_uuid)
resolved_sample_uuids[sample_uuid] = resource if resource else material_uuid
else:
resolved_sample_uuids[sample_uuid] = material_uuid
function_args[PARAM_SAMPLE_UUIDS] = resolved_sample_uuids
self.lab_logger().debug(
f"[JsonCommandAsync] 注入 {PARAM_SAMPLE_UUIDS}: {resolved_sample_uuids}"
)
continue continue
# 处理单个 ResourceSlot # 处理单个 ResourceSlot
@@ -1907,6 +2008,7 @@ class ROS2DeviceNode:
if driver_is_ros: if driver_is_ros:
driver_params["device_id"] = device_id driver_params["device_id"] = device_id
driver_params["registry_name"] = device_config.res_content.klass
driver_params["resource_tracker"] = self.resource_tracker driver_params["resource_tracker"] = self.resource_tracker
self._driver_instance = self._driver_creator.create_instance(driver_params) self._driver_instance = self._driver_creator.create_instance(driver_params)
if self._driver_instance is None: if self._driver_instance is None:
@@ -1924,6 +2026,7 @@ class ROS2DeviceNode:
children=children, children=children,
driver_instance=self._driver_instance, # type: ignore driver_instance=self._driver_instance, # type: ignore
device_id=device_id, device_id=device_id,
registry_name=device_config.res_content.klass,
device_uuid=device_uuid, device_uuid=device_uuid,
status_types=status_types, status_types=status_types,
action_value_mappings=action_value_mappings, action_value_mappings=action_value_mappings,
@@ -1935,6 +2038,7 @@ class ROS2DeviceNode:
self._ros_node = BaseROS2DeviceNode( self._ros_node = BaseROS2DeviceNode(
driver_instance=self._driver_instance, driver_instance=self._driver_instance,
device_id=device_id, device_id=device_id,
registry_name=device_config.res_content.klass,
device_uuid=device_uuid, device_uuid=device_uuid,
status_types=status_types, status_types=status_types,
action_value_mappings=action_value_mappings, action_value_mappings=action_value_mappings,
@@ -1943,6 +2047,7 @@ class ROS2DeviceNode:
resource_tracker=self.resource_tracker, resource_tracker=self.resource_tracker,
) )
self._ros_node: BaseROS2DeviceNode self._ros_node: BaseROS2DeviceNode
# 将注册表类型名传递给BaseROS2DeviceNode,用于slave上报
self._ros_node.lab_logger().info(f"初始化完成 {self._ros_node.uuid} {self.driver_is_ros}") self._ros_node.lab_logger().info(f"初始化完成 {self._ros_node.uuid} {self.driver_is_ros}")
self.driver_instance._ros_node = self._ros_node # type: ignore self.driver_instance._ros_node = self._ros_node # type: ignore
self.driver_instance._execute_driver_command = self._ros_node._execute_driver_command # type: ignore self.driver_instance._execute_driver_command = self._ros_node._execute_driver_command # type: ignore
@@ -1960,7 +2065,9 @@ class ROS2DeviceNode:
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
loop.run_forever() loop.run_forever()
ROS2DeviceNode._asyncio_loop_thread = threading.Thread(target=run_event_loop, daemon=True, name="ROS2DeviceNode") ROS2DeviceNode._asyncio_loop_thread = threading.Thread(
target=run_event_loop, daemon=True, name="ROS2DeviceNode"
)
ROS2DeviceNode._asyncio_loop_thread.start() ROS2DeviceNode._asyncio_loop_thread.start()
logger.info(f"循环线程已启动") logger.info(f"循环线程已启动")

View File

@@ -6,12 +6,13 @@ from cv_bridge import CvBridge
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, DeviceNodeResourceTracker from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, DeviceNodeResourceTracker
class VideoPublisher(BaseROS2DeviceNode): class VideoPublisher(BaseROS2DeviceNode):
def __init__(self, device_id='video_publisher', device_uuid='', camera_index=0, period: float = 0.1, resource_tracker: DeviceNodeResourceTracker = None): def __init__(self, device_id='video_publisher', registry_name="", device_uuid='', camera_index=0, period: float = 0.1, resource_tracker: DeviceNodeResourceTracker = None):
# 初始化BaseROS2DeviceNode使用自身作为driver_instance # 初始化BaseROS2DeviceNode使用自身作为driver_instance
BaseROS2DeviceNode.__init__( BaseROS2DeviceNode.__init__(
self, self,
driver_instance=self, driver_instance=self,
device_id=device_id, device_id=device_id,
registry_name=registry_name,
device_uuid=device_uuid, device_uuid=device_uuid,
status_types={}, status_types={},
action_value_mappings={}, action_value_mappings={},

View File

@@ -10,6 +10,7 @@ class ControllerNode(BaseROS2DeviceNode):
def __init__( def __init__(
self, self,
device_id: str, device_id: str,
registry_name: str,
controller_func: Callable, controller_func: Callable,
update_rate: float, update_rate: float,
inputs: Dict[str, Dict[str, type | str]], inputs: Dict[str, Dict[str, type | str]],
@@ -51,6 +52,7 @@ class ControllerNode(BaseROS2DeviceNode):
self, self,
driver_instance=self, driver_instance=self,
device_id=device_id, device_id=device_id,
registry_name=registry_name,
status_types=status_types, status_types=status_types,
action_value_mappings=action_value_mappings, action_value_mappings=action_value_mappings,
hardware_interface=hardware_interface, hardware_interface=hardware_interface,

View File

@@ -1,17 +1,17 @@
import collections import collections
from dataclasses import dataclass, field
import json import json
import threading import threading
import time import time
import traceback import traceback
import uuid import uuid
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Optional, Dict, Any, List, ClassVar, Set, Union from typing import TYPE_CHECKING, Optional, Dict, Any, List, ClassVar, Set, Union
from typing_extensions import TypedDict
from action_msgs.msg import GoalStatus from action_msgs.msg import GoalStatus
from geometry_msgs.msg import Point from geometry_msgs.msg import Point
from rclpy.action import ActionClient, get_action_server_names_and_types_by_node from rclpy.action import ActionClient, get_action_server_names_and_types_by_node
from rclpy.service import Service from rclpy.service import Service
from typing_extensions import TypedDict
from unilabos_msgs.msg import Resource # type: ignore from unilabos_msgs.msg import Resource # type: ignore
from unilabos_msgs.srv import ( from unilabos_msgs.srv import (
ResourceAdd, ResourceAdd,
@@ -23,10 +23,20 @@ from unilabos_msgs.srv import (
from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response
from unique_identifier_msgs.msg import UUID from unique_identifier_msgs.msg import UUID
from unilabos.registry.placeholder_type import ResourceSlot, DeviceSlot
from unilabos.registry.registry import lab_registry from unilabos.registry.registry import lab_registry
from unilabos.resources.container import RegularContainer from unilabos.resources.container import RegularContainer
from unilabos.resources.graphio import initialize_resource from unilabos.resources.graphio import initialize_resource
from unilabos.resources.registry import add_schema from unilabos.resources.registry import add_schema
from unilabos.resources.resource_tracker import (
ResourceDict,
ResourceDictInstance,
ResourceTreeSet,
ResourceTreeInstance,
RETURN_UNILABOS_SAMPLES,
JSON_UNILABOS_PARAM,
PARAM_SAMPLE_UUIDS,
)
from unilabos.ros.initialize_device import initialize_device_from_dict from unilabos.ros.initialize_device import initialize_device_from_dict
from unilabos.ros.msgs.message_converter import ( from unilabos.ros.msgs.message_converter import (
get_msg_type, get_msg_type,
@@ -37,17 +47,10 @@ from unilabos.ros.msgs.message_converter import (
) )
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode, DeviceNodeResourceTracker from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode, DeviceNodeResourceTracker
from unilabos.ros.nodes.presets.controller_node import ControllerNode from unilabos.ros.nodes.presets.controller_node import ControllerNode
from unilabos.resources.resource_tracker import (
ResourceDict,
ResourceDictInstance,
ResourceTreeSet,
ResourceTreeInstance,
)
from unilabos.utils import logger from unilabos.utils import logger
from unilabos.utils.exception import DeviceClassInvalid from unilabos.utils.exception import DeviceClassInvalid
from unilabos.utils.log import warning from unilabos.utils.log import warning
from unilabos.utils.type_check import serialize_result_info from unilabos.utils.type_check import serialize_result_info
from unilabos.registry.placeholder_type import ResourceSlot, DeviceSlot
if TYPE_CHECKING: if TYPE_CHECKING:
from unilabos.app.ws_client import QueueItem from unilabos.app.ws_client import QueueItem
@@ -245,6 +248,7 @@ class HostNode(BaseROS2DeviceNode):
self, self,
driver_instance=self, driver_instance=self,
device_id=device_id, device_id=device_id,
registry_name="host_node",
device_uuid=host_node_dict["uuid"], device_uuid=host_node_dict["uuid"],
status_types={}, status_types={},
action_value_mappings=lab_registry.device_type_registry["host_node"]["class"]["action_value_mappings"], action_value_mappings=lab_registry.device_type_registry["host_node"]["class"]["action_value_mappings"],
@@ -299,7 +303,8 @@ class HostNode(BaseROS2DeviceNode):
} # 用来存储多个ActionClient实例 } # 用来存储多个ActionClient实例
self._action_value_mappings: Dict[str, Dict] = ( self._action_value_mappings: Dict[str, Dict] = (
{} {}
) # 用来存储多个ActionClient的type, goal, feedback, result的变量名映射关系 ) # device_id -> action_value_mappings(本地+远程设备统一存储)
self._slave_registry_configs: Dict[str, Dict] = {} # registry_name -> registry_config(含action_value_mappings)
self._goals: Dict[str, Any] = {} # 用来存储多个目标的状态 self._goals: Dict[str, Any] = {} # 用来存储多个目标的状态
self._online_devices: Set[str] = {f"{self.namespace}/{device_id}"} # 用于跟踪在线设备 self._online_devices: Set[str] = {f"{self.namespace}/{device_id}"} # 用于跟踪在线设备
self._last_discovery_time = 0.0 # 上次设备发现的时间 self._last_discovery_time = 0.0 # 上次设备发现的时间
@@ -633,6 +638,8 @@ class HostNode(BaseROS2DeviceNode):
self.device_machine_names[device_id] = "本地" self.device_machine_names[device_id] = "本地"
self.devices_instances[device_id] = d self.devices_instances[device_id] = d
# noinspection PyProtectedMember # noinspection PyProtectedMember
self._action_value_mappings[device_id] = d._ros_node._action_value_mappings
# noinspection PyProtectedMember
for action_name, action_value_mapping in d._ros_node._action_value_mappings.items(): for action_name, action_value_mapping in d._ros_node._action_value_mappings.items():
if action_name.startswith("auto-") or str(action_value_mapping.get("type", "")).startswith( if action_name.startswith("auto-") or str(action_value_mapping.get("type", "")).startswith(
"UniLabJsonCommand" "UniLabJsonCommand"
@@ -755,6 +762,7 @@ class HostNode(BaseROS2DeviceNode):
item: "QueueItem", item: "QueueItem",
action_type: str, action_type: str,
action_kwargs: Dict[str, Any], action_kwargs: Dict[str, Any],
sample_material: Dict[str, str],
server_info: Optional[Dict[str, Any]] = None, server_info: Optional[Dict[str, Any]] = None,
) -> None: ) -> None:
""" """
@@ -772,14 +780,14 @@ class HostNode(BaseROS2DeviceNode):
if action_name.startswith("auto-"): if action_name.startswith("auto-"):
action_name = action_name[5:] action_name = action_name[5:]
action_id = f"/devices/{device_id}/_execute_driver_command" action_id = f"/devices/{device_id}/_execute_driver_command"
action_kwargs = { json_command: Dict[str, Any] = {
"string": json.dumps(
{
"function_name": action_name, "function_name": action_name,
"function_args": action_kwargs, "function_args": action_kwargs,
JSON_UNILABOS_PARAM: {
PARAM_SAMPLE_UUIDS: sample_material,
},
} }
) action_kwargs = {"string": json.dumps(json_command)}
}
if action_type.startswith("UniLabJsonCommandAsync"): if action_type.startswith("UniLabJsonCommandAsync"):
action_id = f"/devices/{device_id}/_execute_driver_command_async" action_id = f"/devices/{device_id}/_execute_driver_command_async"
else: else:
@@ -790,21 +798,6 @@ class HostNode(BaseROS2DeviceNode):
raise ValueError(f"ActionClient {action_id} not found.") raise ValueError(f"ActionClient {action_id} not found.")
action_client: ActionClient = self._action_clients[action_id] action_client: ActionClient = self._action_clients[action_id]
# 遍历action_kwargs下的所有子dict将"sample_uuid"的值赋给"sample_id"
def assign_sample_id(obj):
if isinstance(obj, dict):
if "sample_uuid" in obj:
obj["sample_id"] = obj["sample_uuid"]
obj.pop("sample_uuid")
for k, v in obj.items():
if k != "unilabos_extra":
assign_sample_id(v)
elif isinstance(obj, list):
for item in obj:
assign_sample_id(item)
assign_sample_id(action_kwargs)
goal_msg = convert_to_ros_msg(action_client._action_type.Goal(), action_kwargs) goal_msg = convert_to_ros_msg(action_client._action_type.Goal(), action_kwargs)
# self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {str(goal_msg)[:1000]}") # self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {str(goal_msg)[:1000]}")
@@ -867,14 +860,14 @@ class HostNode(BaseROS2DeviceNode):
# 适配后端的一些额外处理 # 适配后端的一些额外处理
return_value = return_info.get("return_value") return_value = return_info.get("return_value")
if isinstance(return_value, dict): if isinstance(return_value, dict):
unilabos_samples = return_value.pop("unilabos_samples", None) unilabos_samples = return_value.pop(RETURN_UNILABOS_SAMPLES, None)
if isinstance(unilabos_samples, list) and unilabos_samples: if isinstance(unilabos_samples, list) and unilabos_samples:
self.lab_logger().info( self.lab_logger().info(
f"[Host Node] Job {job_id[:8]} returned {len(unilabos_samples)} sample(s): " f"[Host Node] Job {job_id[:8]} returned {len(unilabos_samples)} sample(s): "
f"{[s.get('name', s.get('id', 'unknown')) if isinstance(s, dict) else str(s)[:20] for s in unilabos_samples[:5]]}" f"{[s.get('name', s.get('id', 'unknown')) if isinstance(s, dict) else str(s)[:20] for s in unilabos_samples[:5]]}"
f"{'...' if len(unilabos_samples) > 5 else ''}" f"{'...' if len(unilabos_samples) > 5 else ''}"
) )
return_info["unilabos_samples"] = unilabos_samples return_info["samples"] = unilabos_samples
suc = return_info.get("suc", False) suc = return_info.get("suc", False)
if not suc: if not suc:
status = "failed" status = "failed"
@@ -1179,6 +1172,10 @@ class HostNode(BaseROS2DeviceNode):
def _node_info_update_callback(self, request, response): def _node_info_update_callback(self, request, response):
""" """
更新节点信息回调 更新节点信息回调
处理两种消息:
1. 首次上报(main_slave_run): 带 devices_config + registry_config,存储 action_value_mappings
2. 设备重注册(SYNC_SLAVE_NODE_INFO): 带 edge_device_id + registry_name,用 registry_name 索引已存储的 mappings
""" """
self.lab_logger().trace(f"[Host Node] Node info update request received: {request}") self.lab_logger().trace(f"[Host Node] Node info update request received: {request}")
try: try:
@@ -1190,12 +1187,48 @@ class HostNode(BaseROS2DeviceNode):
info = info["SYNC_SLAVE_NODE_INFO"] info = info["SYNC_SLAVE_NODE_INFO"]
machine_name = info["machine_name"] machine_name = info["machine_name"]
edge_device_id = info["edge_device_id"] edge_device_id = info["edge_device_id"]
registry_name = info.get("registry_name", "")
self.device_machine_names[edge_device_id] = machine_name self.device_machine_names[edge_device_id] = machine_name
# 用 registry_name 索引已存储的 registry_config,获取 action_value_mappings
if registry_name and registry_name in self._slave_registry_configs:
action_mappings = self._slave_registry_configs[registry_name].get(
"class", {}
).get("action_value_mappings", {})
if action_mappings:
self._action_value_mappings[edge_device_id] = action_mappings
self.lab_logger().info(
f"[Host Node] Loaded {len(action_mappings)} action mappings "
f"for remote device {edge_device_id} (registry: {registry_name})"
)
else: else:
devices_config = info.pop("devices_config") devices_config = info.pop("devices_config")
registry_config = info.pop("registry_config") registry_config = info.pop("registry_config")
if registry_config: if registry_config:
http_client.resource_registry({"resources": registry_config}) http_client.resource_registry({"resources": registry_config})
# 存储 slave 的 registry_config,用于后续 SYNC_SLAVE_NODE_INFO 索引
for reg_name, reg_data in registry_config.items():
if isinstance(reg_data, dict) and "class" in reg_data:
self._slave_registry_configs[reg_name] = reg_data
# 解析 devices_config,建立 device_id -> action_value_mappings 映射
if devices_config:
for device_tree in devices_config:
for device_dict in device_tree:
device_id = device_dict.get("id", "")
class_name = device_dict.get("class", "")
if device_id and class_name and class_name in self._slave_registry_configs:
action_mappings = self._slave_registry_configs[class_name].get(
"class", {}
).get("action_value_mappings", {})
if action_mappings:
self._action_value_mappings[device_id] = action_mappings
self.lab_logger().info(
f"[Host Node] Stored {len(action_mappings)} action mappings "
f"for remote device {device_id} (class: {class_name})"
)
self.lab_logger().debug(f"[Host Node] Node info update: {info}") self.lab_logger().debug(f"[Host Node] Node info update: {info}")
response.response = "OK" response.response = "OK"
except Exception as e: except Exception as e:

View File

@@ -7,10 +7,11 @@ from rclpy.callback_groups import ReentrantCallbackGroup
from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
class JointRepublisher(BaseROS2DeviceNode): class JointRepublisher(BaseROS2DeviceNode):
def __init__(self,device_id,resource_tracker, **kwargs): def __init__(self,device_id, registry_name, resource_tracker, **kwargs):
super().__init__( super().__init__(
driver_instance=self, driver_instance=self,
device_id=device_id, device_id=device_id,
registry_name=registry_name,
status_types={}, status_types={},
action_value_mappings={}, action_value_mappings={},
hardware_interface={}, hardware_interface={},

View File

@@ -26,7 +26,7 @@ from unilabos.resources.graphio import initialize_resources
from unilabos.registry.registry import lab_registry from unilabos.registry.registry import lab_registry
class ResourceMeshManager(BaseROS2DeviceNode): class ResourceMeshManager(BaseROS2DeviceNode):
def __init__(self, resource_model: dict, resource_config: list,resource_tracker, device_id: str = "resource_mesh_manager", rate=50, **kwargs): def __init__(self, resource_model: dict, resource_config: list,resource_tracker, device_id: str = "resource_mesh_manager", registry_name: str = "", rate=50, **kwargs):
"""初始化资源网格管理器节点 """初始化资源网格管理器节点
Args: Args:
@@ -37,6 +37,7 @@ class ResourceMeshManager(BaseROS2DeviceNode):
super().__init__( super().__init__(
driver_instance=self, driver_instance=self,
device_id=device_id, device_id=device_id,
registry_name=registry_name,
status_types={}, status_types={},
action_value_mappings={}, action_value_mappings={},
hardware_interface={}, hardware_interface={},

View File

@@ -7,7 +7,7 @@ from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, DeviceNodeRe
class ROS2SerialNode(BaseROS2DeviceNode): class ROS2SerialNode(BaseROS2DeviceNode):
def __init__(self, device_id, port: str, baudrate: int = 9600, resource_tracker: DeviceNodeResourceTracker=None): def __init__(self, device_id, registry_name, port: str, baudrate: int = 9600, resource_tracker: DeviceNodeResourceTracker=None):
# 保存属性,以便在调用父类初始化前使用 # 保存属性,以便在调用父类初始化前使用
self.port = port self.port = port
self.baudrate = baudrate self.baudrate = baudrate
@@ -28,6 +28,7 @@ class ROS2SerialNode(BaseROS2DeviceNode):
BaseROS2DeviceNode.__init__( BaseROS2DeviceNode.__init__(
self, self,
driver_instance=self, driver_instance=self,
registry_name=registry_name,
device_id=device_id, device_id=device_id,
status_types={}, status_types={},
action_value_mappings={}, action_value_mappings={},

View File

@@ -47,6 +47,7 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
*, *,
driver_instance: "WorkstationBase", driver_instance: "WorkstationBase",
device_id: str, device_id: str,
registry_name: str,
device_uuid: str, device_uuid: str,
status_types: Dict[str, Any], status_types: Dict[str, Any],
action_value_mappings: Dict[str, Any], action_value_mappings: Dict[str, Any],
@@ -62,6 +63,7 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
super().__init__( super().__init__(
driver_instance=driver_instance, driver_instance=driver_instance,
device_id=device_id, device_id=device_id,
registry_name=registry_name,
device_uuid=device_uuid, device_uuid=device_uuid,
status_types=status_types, status_types=status_types,
action_value_mappings={**action_value_mappings, **self.protocol_action_mappings}, action_value_mappings={**action_value_mappings, **self.protocol_action_mappings},
@@ -340,6 +342,8 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
plr = self.resource_tracker.figure_resource({"name": res_name}, try_mode=False) plr = self.resource_tracker.figure_resource({"name": res_name}, try_mode=False)
# 获取父资源 # 获取父资源
res = self.resource_tracker.parent_resource(plr) res = self.resource_tracker.parent_resource(plr)
if res is None:
res = plr
if id(res) not in seen: if id(res) not in seen:
seen.add(id(res)) seen.add(id(res))
unique_resources.append(res) unique_resources.append(res)

View File

@@ -52,7 +52,8 @@ class DeviceClassCreator(Generic[T]):
if self.device_instance is not None: if self.device_instance is not None:
for c in self.children: for c in self.children:
if c.res_content.type != "device": if c.res_content.type != "device":
self.resource_tracker.add_resource(c.get_plr_nested_dict()) res = ResourceTreeSet([ResourceTreeInstance(c)]).to_plr_resources()[0]
self.resource_tracker.add_resource(res)
def create_instance(self, data: Dict[str, Any]) -> T: def create_instance(self, data: Dict[str, Any]) -> T:
""" """
@@ -119,7 +120,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
# return resource, source_type # return resource, source_type
def _process_resource_references( def _process_resource_references(
self, data: Any, to_dict=False, states=None, prefix_path="", name_to_uuid=None self, data: Any, processed_child_names: Optional[Dict[str, Any]], to_dict=False, states=None, prefix_path="", name_to_uuid=None
) -> Any: ) -> Any:
""" """
递归处理资源引用替换_resource_child_name对应的资源 递归处理资源引用替换_resource_child_name对应的资源
@@ -164,6 +165,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
states[prefix_path] = resource_instance.serialize_all_state() states[prefix_path] = resource_instance.serialize_all_state()
return serialized return serialized
else: else:
processed_child_names[child_name] = resource_instance
self.resource_tracker.add_resource(resource_instance) self.resource_tracker.add_resource(resource_instance)
# 立即设置UUIDstate已经在resource_ulab_to_plr中处理过了 # 立即设置UUIDstate已经在resource_ulab_to_plr中处理过了
if name_to_uuid: if name_to_uuid:
@@ -182,12 +184,12 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
result = {} result = {}
for key, value in data.items(): for key, value in data.items():
new_prefix = f"{prefix_path}.{key}" if prefix_path else key new_prefix = f"{prefix_path}.{key}" if prefix_path else key
result[key] = self._process_resource_references(value, to_dict, states, new_prefix, name_to_uuid) result[key] = self._process_resource_references(value, processed_child_names, to_dict, states, new_prefix, name_to_uuid)
return result return result
elif isinstance(data, list): elif isinstance(data, list):
return [ return [
self._process_resource_references(item, to_dict, states, f"{prefix_path}[{i}]", name_to_uuid) self._process_resource_references(item, processed_child_names, to_dict, states, f"{prefix_path}[{i}]", name_to_uuid)
for i, item in enumerate(data) for i, item in enumerate(data)
] ]
@@ -234,7 +236,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
# 首先处理资源引用 # 首先处理资源引用
states = {} states = {}
processed_data = self._process_resource_references( processed_data = self._process_resource_references(
data, to_dict=True, states=states, name_to_uuid=name_to_uuid data, {}, to_dict=True, states=states, name_to_uuid=name_to_uuid
) )
try: try:
@@ -270,7 +272,12 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
arg_value = spec_args[param_name].annotation arg_value = spec_args[param_name].annotation
data[param_name]["_resource_type"] = self.device_cls.__module__ + ":" + arg_value data[param_name]["_resource_type"] = self.device_cls.__module__ + ":" + arg_value
logger.debug(f"自动补充 _resource_type: {data[param_name]['_resource_type']}") logger.debug(f"自动补充 _resource_type: {data[param_name]['_resource_type']}")
processed_data = self._process_resource_references(data, to_dict=False, name_to_uuid=name_to_uuid) processed_child_names = {}
processed_data = self._process_resource_references(data, processed_child_names, to_dict=False, name_to_uuid=name_to_uuid)
for child_name, resource_instance in processed_data.items():
for ind, name in enumerate([child.res_content.name for child in self.children]):
if name == child_name:
self.children.pop(ind)
self.device_instance = super(PyLabRobotCreator, self).create_instance(processed_data) # 补全变量后直接调用调用的自身的attach_resource self.device_instance = super(PyLabRobotCreator, self).create_instance(processed_data) # 补全变量后直接调用调用的自身的attach_resource
except Exception as e: except Exception as e:
logger.error(f"PyLabRobot创建实例失败: {e}") logger.error(f"PyLabRobot创建实例失败: {e}")
@@ -342,9 +349,10 @@ class WorkstationNodeCreator(DeviceClassCreator[T]):
try: try:
# 创建实例额外补充一个给protocol node的字段后面考虑取消 # 创建实例额外补充一个给protocol node的字段后面考虑取消
data["children"] = self.children data["children"] = self.children
for child in self.children: # super(WorkstationNodeCreator, self).create_instance(data)的时候会attach
if child.res_content.type != "device": # for child in self.children:
self.resource_tracker.add_resource(child.get_plr_nested_dict()) # if child.res_content.type != "device":
# self.resource_tracker.add_resource(child.get_plr_nested_dict())
deck_dict = data.get("deck") deck_dict = data.get("deck")
if deck_dict: if deck_dict:
from pylabrobot.resources import Deck, Resource from pylabrobot.resources import Deck, Resource

View File

@@ -339,13 +339,8 @@
"z": 0 "z": 0
}, },
"config": { "config": {
"max_volume": 500.0,
"type": "RegularContainer", "type": "RegularContainer",
"category": "container", "category": "container"
"max_temp": 200.0,
"min_temp": -20.0,
"has_stirrer": true,
"has_heater": true
}, },
"data": { "data": {
"liquids": [], "liquids": [],
@@ -769,9 +764,7 @@
"size_y": 250, "size_y": 250,
"size_z": 0, "size_z": 0,
"type": "RegularContainer", "type": "RegularContainer",
"category": "container", "category": "container"
"reagent": "sodium_chloride",
"physical_state": "solid"
}, },
"data": { "data": {
"current_mass": 500.0, "current_mass": 500.0,
@@ -792,14 +785,11 @@
"z": 0 "z": 0
}, },
"config": { "config": {
"volume": 500.0,
"size_x": 600, "size_x": 600,
"size_y": 250, "size_y": 250,
"size_z": 0, "size_z": 0,
"type": "RegularContainer", "type": "RegularContainer",
"category": "container", "category": "container"
"reagent": "sodium_carbonate",
"physical_state": "solid"
}, },
"data": { "data": {
"current_mass": 500.0, "current_mass": 500.0,
@@ -820,14 +810,11 @@
"z": 0 "z": 0
}, },
"config": { "config": {
"volume": 500.0,
"size_x": 650, "size_x": 650,
"size_y": 250, "size_y": 250,
"size_z": 0, "size_z": 0,
"type": "RegularContainer", "type": "RegularContainer",
"category": "container", "category": "container"
"reagent": "magnesium_chloride",
"physical_state": "solid"
}, },
"data": { "data": {
"current_mass": 500.0, "current_mass": 500.0,

View File

@@ -184,6 +184,51 @@ def get_all_subscriptions(instance) -> list:
return subscriptions return subscriptions
def always_free(func: F) -> F:
"""
标记动作为永久闲置(不受busy队列限制)的装饰器
被此装饰器标记的 action 方法,在执行时不会受到设备级别的排队限制,
任何时候请求都可以立即执行。适用于查询类、状态读取类等轻量级操作。
Example:
class MyDriver:
@always_free
def query_status(self, param: str):
# 这个动作可以随时执行,不需要排队
return self._status
def transfer(self, volume: float):
# 这个动作会按正常排队逻辑执行
pass
Note:
- 可以与其他装饰器组合使用,@always_free 应放在最外层
- 仅影响 WebSocket 调度层的 busy/free 判断,不影响 ROS2 层
"""
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
wrapper._is_always_free = True # type: ignore[attr-defined]
return wrapper # type: ignore[return-value]
def is_always_free(func) -> bool:
"""
检查函数是否被标记为永久闲置
Args:
func: 被检查的函数
Returns:
如果函数被 @always_free 装饰则返回 True否则返回 False
"""
return getattr(func, "_is_always_free", False)
def not_action(func: F) -> F: def not_action(func: F) -> F:
""" """
标记方法为非动作的装饰器 标记方法为非动作的装饰器

View File

@@ -27,8 +27,9 @@ __all__ = [
from ast import Constant from ast import Constant
from unilabos.resources.resource_tracker import PARAM_SAMPLE_UUIDS
from unilabos.utils import logger from unilabos.utils import logger
from unilabos.utils.decorator import is_not_action from unilabos.utils.decorator import is_not_action, is_always_free
class ImportManager: class ImportManager:
@@ -281,6 +282,9 @@ class ImportManager:
continue continue
# 其他非_开头的方法归类为action # 其他非_开头的方法归类为action
method_info = self._analyze_method_signature(method) method_info = self._analyze_method_signature(method)
# 检查是否被 @always_free 装饰器标记
if is_always_free(method):
method_info["always_free"] = True
result["action_methods"][name] = method_info result["action_methods"][name] = method_info
return result return result
@@ -338,16 +342,24 @@ class ImportManager:
if self._is_not_action_method(node): if self._is_not_action_method(node):
continue continue
# 其他非_开头的方法归类为action # 其他非_开头的方法归类为action
# 检查是否被 @always_free 装饰器标记
if self._is_always_free_method(node):
method_info["always_free"] = True
result["action_methods"][method_name] = method_info result["action_methods"][method_name] = method_info
return result return result
def _analyze_method_signature(self, method) -> Dict[str, Any]: def _analyze_method_signature(self, method, skip_unilabos_params: bool = True) -> Dict[str, Any]:
""" """
分析方法签名,提取具体的命名参数信息 分析方法签名,提取具体的命名参数信息
注意:此方法会跳过*args和**kwargs只提取具体的命名参数 注意:此方法会跳过*args和**kwargs只提取具体的命名参数
这样可以确保通过**dict方式传参时的准确性 这样可以确保通过**dict方式传参时的准确性
Args:
method: 要分析的方法
skip_unilabos_params: 是否跳过 unilabos 系统参数(如 sample_uuids
registry 补全时为 TrueJsonCommand 执行时为 False
示例用法: 示例用法:
method_info = self._analyze_method_signature(some_method) method_info = self._analyze_method_signature(some_method)
params = {"param1": "value1", "param2": "value2"} params = {"param1": "value1", "param2": "value2"}
@@ -368,6 +380,10 @@ class ImportManager:
if param.kind == param.VAR_KEYWORD: # **kwargs if param.kind == param.VAR_KEYWORD: # **kwargs
continue continue
# 跳过 sample_uuids 参数由系统自动注入registry 补全时跳过)
if skip_unilabos_params and param_name == PARAM_SAMPLE_UUIDS:
continue
is_required = param.default == inspect.Parameter.empty is_required = param.default == inspect.Parameter.empty
if is_required: if is_required:
num_required += 1 num_required += 1
@@ -464,6 +480,13 @@ class ImportManager:
return True return True
return False return False
def _is_always_free_method(self, node: ast.FunctionDef) -> bool:
"""检查是否是@always_free装饰的方法"""
for decorator in node.decorator_list:
if isinstance(decorator, ast.Name) and decorator.id == "always_free":
return True
return False
def _get_property_name_from_setter(self, node: ast.FunctionDef) -> str: def _get_property_name_from_setter(self, node: ast.FunctionDef) -> str:
"""从setter装饰器中获取属性名""" """从setter装饰器中获取属性名"""
for decorator in node.decorator_list: for decorator in node.decorator_list:
@@ -563,6 +586,9 @@ class ImportManager:
for i, arg in enumerate(node.args.args): for i, arg in enumerate(node.args.args):
if arg.arg == "self": if arg.arg == "self":
continue continue
# 跳过 sample_uuids 参数(由系统自动注入)
if arg.arg == PARAM_SAMPLE_UUIDS:
continue
arg_info = { arg_info = {
"name": arg.arg, "name": arg.arg,
"type": None, "type": None,

View File

@@ -60,7 +60,11 @@
==================== 连接关系图 ==================== ==================== 连接关系图 ====================
控制流 (ready 端口串联): 控制流 (ready 端口串联):
create_resource_1 -> create_resource_2 -> ... -> set_liquid_1 -> set_liquid_2 -> ... -> transfer_liquid_1 -> transfer_liquid_2 -> ... - create_resource 之间: 无 ready 连接
- set_liquid_from_plate 之间: 无 ready 连接
- create_resource 与 set_liquid_from_plate 之间: 无 ready 连接
- transfer_liquid 之间: 通过 ready 端口串联
transfer_liquid_1 -> transfer_liquid_2 -> transfer_liquid_3 -> ...
物料流: 物料流:
[create_resource] --labware--> [set_liquid_from_plate] --output_wells--> [transfer_liquid] --sources_out/targets_out--> [下一个 transfer_liquid] [create_resource] --labware--> [set_liquid_from_plate] --output_wells--> [transfer_liquid] --sources_out/targets_out--> [下一个 transfer_liquid]
@@ -402,7 +406,6 @@ def build_protocol_graph(
# 为每个唯一的 slot 创建 create_resource 节点 # 为每个唯一的 slot 创建 create_resource 节点
res_index = 0 res_index = 0
last_create_resource_id = None
for slot, info in slots_info.items(): for slot, info in slots_info.items():
node_id = str(uuid.uuid4()) node_id = str(uuid.uuid4())
res_id = info["res_id"] res_id = info["res_id"]
@@ -431,10 +434,7 @@ def build_protocol_graph(
) )
slot_to_create_resource[slot] = node_id slot_to_create_resource[slot] = node_id
# create_resource 之间通过 ready 串联 # create_resource 之间不需要 ready 连接
if last_create_resource_id is not None:
G.add_edge(last_create_resource_id, node_id, source_port="ready", target_port="ready")
last_create_resource_id = node_id
# ==================== 第二步:为每个 reagent 创建 set_liquid_from_plate 节点 ==================== # ==================== 第二步:为每个 reagent 创建 set_liquid_from_plate 节点 ====================
# 创建 Group 节点,包含所有 set_liquid_from_plate 节点 # 创建 Group 节点,包含所有 set_liquid_from_plate 节点
@@ -453,7 +453,6 @@ def build_protocol_graph(
) )
set_liquid_index = 0 set_liquid_index = 0
last_set_liquid_id = last_create_resource_id # set_liquid_from_plate 连接在 create_resource 之后
for labware_id, item in labware_info.items(): for labware_id, item in labware_info.items():
# 跳过 Tip/Rack 类型 # 跳过 Tip/Rack 类型
@@ -494,10 +493,7 @@ def build_protocol_graph(
}, },
) )
# ready 连接:上一个节点 -> set_liquid_from_plate # set_liquid_from_plate 之间不需要 ready 连接
if last_set_liquid_id is not None:
G.add_edge(last_set_liquid_id, node_id, source_port="ready", target_port="ready")
last_set_liquid_id = node_id
# 物料流create_resource 的 labware -> set_liquid_from_plate 的 input_plate # 物料流create_resource 的 labware -> set_liquid_from_plate 的 input_plate
create_res_node_id = slot_to_create_resource.get(slot) create_res_node_id = slot_to_create_resource.get(slot)
@@ -507,7 +503,8 @@ def build_protocol_graph(
# set_liquid_from_plate 的输出 output_wells 用于连接 transfer_liquid # set_liquid_from_plate 的输出 output_wells 用于连接 transfer_liquid
resource_last_writer[labware_id] = f"{node_id}:output_wells" resource_last_writer[labware_id] = f"{node_id}:output_wells"
last_control_node_id = last_set_liquid_id # transfer_liquid 之间通过 ready 串联,从 None 开始
last_control_node_id = None
# 端口名称映射JSON 字段名 -> 实际 handle key # 端口名称映射JSON 字段名 -> 实际 handle key
INPUT_PORT_MAPPING = { INPUT_PORT_MAPPING = {