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

fix: multi channel

Browse Source
This commit is contained in:
wznln
2025-05-07 02:36:53 +08:00
parent f3637d4043
commit 83abf877b5
1 changed files with 16 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
unilabos/ros/nodes/base_device_node.py
View File

@@ -514,20 +514,30 @@ class BaseROS2DeviceNode(Node, Generic[T]):
for k, v in goal.get_fields_and_field_types().items():
if v in ["unilabos_msgs/Resource", "sequence<unilabos_msgs/Resource>"]:
self.lab_logger().info(f"查询资源状态: Key: {k} Type: {v}")
current_resources = []
try:
r = ResourceGet.Request()
r.id = action_kwargs[k]["id"] if v == "unilabos_msgs/Resource" else action_kwargs[k][0]["id"]
r.with_children = True
response = await self._resource_clients["resource_get"].call_async(r)
if len(action_kwargs[k]) > 1:
for i in action_kwargs[k]:
r = ResourceGet.Request()
r.id = i["id"]
r.with_children = True
response = await self._resource_clients["resource_get"].call_async(r)
current_resources.extend(response.resources)
else:
r = ResourceGet.Request()
r.id = action_kwargs[k]["id"] if v == "unilabos_msgs/Resource" else action_kwargs[k][0]["id"]
r.with_children = True
response = await self._resource_clients["resource_get"].call_async(r)
current_resources.extend(response.resources)
except Exception:
logger.error(f"资源查询失败，默认使用本地资源")
# 删除对response.resources的检查，因为它总是存在
resources_list = [convert_from_ros_msg(rs) for rs in response.resources] # type: ignore # FIXME
resources_list = [convert_from_ros_msg(rs) for rs in current_resources] # type: ignore # FIXME
self.lab_logger().debug(f"资源查询结果: {len(resources_list)} 个资源")
type_hint = action_paramtypes[k]
final_type = get_type_class(type_hint)
# 判断 ACTION 是否需要特殊的物料类型如 pylabrobot.resources.Resource，并做转换
final_resource = convert_resources_to_type(resources_list, final_type)
final_resource = [convert_resources_to_type([i], final_type)[0] for i in resources_list]
action_kwargs[k] = self.resource_tracker.figure_resource(final_resource)
self.lab_logger().info(f"准备执行: {action_kwargs}, 函数: {ACTION.__name__}")