Resource update & asyncio fix

correct bioyond config

prcxi example

fix append_resource

fix regularcontainer

fix cancel error

fix resource_get param

fix json dumps

support name change during materials change

enable slave mode

change uuid logger to trace level

correct remove_resource stats

disable slave connect websocket

adjust with_children param

modify devices to use correct executor (sleep, create_task)

support sleep and create_task in node

fix run async execution error
This commit is contained in:
Xuwznln
2025-10-31 21:43:25 +08:00
parent 8807865649
commit b6dfe2b944
32 changed files with 1456 additions and 943 deletions

View File

@@ -18,7 +18,8 @@ from unilabos_msgs.srv import (
ResourceDelete,
ResourceUpdate,
ResourceList,
SerialCommand, ResourceGet,
SerialCommand,
ResourceGet,
) # type: ignore
from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response
from unique_identifier_msgs.msg import UUID
@@ -164,29 +165,16 @@ class HostNode(BaseROS2DeviceNode):
# resources_config 的 root node 是
# # 创建反向映射new_uuid -> old_uuid
# reverse_uuid_mapping = {new_uuid: old_uuid for old_uuid, new_uuid in uuid_mapping.items()}
# for tree in resources_config.trees:
# node = tree.root_node
# if node.res_content.type == "device":
# if node.res_content.id == "host_node":
# continue
# # slave节点走c2s更新接口拿到add自行update uuid
# device_tracker = self.devices_instances[node.res_content.id].resource_tracker
# old_uuid = reverse_uuid_mapping.get(node.res_content.uuid)
# if old_uuid:
# # 找到旧UUID使用UUID查找
# resource_instance = device_tracker.uuid_to_resources.get(old_uuid)
# else:
# # 未找到旧UUID使用name查找
# resource_instance = device_tracker.figure_resource(
# {"name": node.res_content.name}
# )
# device_tracker.loop_update_uuid(resource_instance, uuid_mapping)
# else:
# try:
# for plr_resource in ResourceTreeSet([tree]).to_plr_resources():
# self.resource_tracker.add_resource(plr_resource)
# except Exception as ex:
# self.lab_logger().warning("[Host Node-Resource] 根节点物料序列化失败!")
for tree in resources_config.trees:
node = tree.root_node
if node.res_content.type == "device":
continue
else:
try:
for plr_resource in ResourceTreeSet([tree]).to_plr_resources():
self._resource_tracker.add_resource(plr_resource)
except Exception as ex:
self.lab_logger().warning(f"[Host Node-Resource] 根节点物料{tree}序列化失败!")
except Exception as ex:
logger.error(f"[Host Node-Resource] 添加物料出错!\n{traceback.format_exc()}")
# 初始化Node基类传递空参数覆盖列表
@@ -746,46 +734,116 @@ class HostNode(BaseROS2DeviceNode):
def get_result_callback(self, item: "QueueItem", action_id: str, future) -> None:
"""获取结果回调"""
job_id = item.job_id
result_msg = future.result().result
result_data = convert_from_ros_msg(result_msg)
status = "success"
return_info_str = result_data.get("return_info")
if return_info_str is not None:
try:
return_info = json.loads(return_info_str)
suc = return_info.get("suc", False)
if not suc:
status = "failed"
except json.JSONDecodeError:
try:
result = future.result()
result_msg = result.result
goal_status = result.status
# 检查是否是被取消的任务
if goal_status == GoalStatus.STATUS_CANCELED:
self.lab_logger().info(f"[Host Node] Goal {action_id} ({job_id[:8]}) was cancelled")
status = "failed"
return_info = serialize_result_info("", False, result_data)
self.lab_logger().critical("错误的return_info类型请断点修复")
else:
# 无 return_info 字段时,回退到 success 字段(若存在)
suc_field = result_data.get("success")
if isinstance(suc_field, bool):
status = "success" if suc_field else "failed"
return_info = serialize_result_info("", suc_field, result_data)
return_info = serialize_result_info("Job was cancelled", False, {})
else:
# 最保守的回退标记失败并返回空JSON
status = "failed"
return_info = serialize_result_info("缺少return_info", False, result_data)
result_data = convert_from_ros_msg(result_msg)
status = "success"
return_info_str = result_data.get("return_info")
if return_info_str is not None:
try:
return_info = json.loads(return_info_str)
suc = return_info.get("suc", False)
if not suc:
status = "failed"
except json.JSONDecodeError:
status = "failed"
return_info = serialize_result_info("", False, result_data)
self.lab_logger().critical("错误的return_info类型请断点修复")
else:
# 无 return_info 字段时,回退到 success 字段(若存在)
suc_field = result_data.get("success")
if isinstance(suc_field, bool):
status = "success" if suc_field else "failed"
return_info = serialize_result_info("", suc_field, result_data)
else:
# 最保守的回退标记失败并返回空JSON
status = "failed"
return_info = serialize_result_info("缺少return_info", False, result_data)
self.lab_logger().info(f"[Host Node] Result for {action_id} ({job_id}): {status}")
self.lab_logger().debug(f"[Host Node] Result data: {result_data}")
self.lab_logger().info(f"[Host Node] Result for {action_id} ({job_id[:8]}): {status}")
if goal_status != GoalStatus.STATUS_CANCELED:
self.lab_logger().debug(f"[Host Node] Result data: {result_data}")
if job_id:
# 清理 _goals 中的记录
if job_id in self._goals:
del self._goals[job_id]
self.lab_logger().debug(f"[Host Node] Removed goal {job_id[:8]} from _goals")
# 发布状态到桥接器
if job_id:
for bridge in self.bridges:
if hasattr(bridge, "publish_job_status"):
if goal_status == GoalStatus.STATUS_CANCELED:
bridge.publish_job_status({}, item, status, return_info)
else:
bridge.publish_job_status(result_data, item, status, return_info)
except Exception as e:
self.lab_logger().error(
f"[Host Node] Error in get_result_callback for {action_id} ({job_id[:8]}): {str(e)}"
)
import traceback
self.lab_logger().error(traceback.format_exc())
# 清理 _goals 中的记录
if job_id in self._goals:
del self._goals[job_id]
# 发布失败状态
for bridge in self.bridges:
if hasattr(bridge, "publish_job_status"):
bridge.publish_job_status(result_data, item, status, return_info)
bridge.publish_job_status(
{}, item, "failed", serialize_result_info(f"Callback error: {str(e)}", False, {})
)
def cancel_goal(self, goal_uuid: str) -> None:
"""取消目标"""
def cancel_goal(self, goal_uuid: str) -> bool:
"""
取消目标
Args:
goal_uuid: 目标UUIDjob_id
Returns:
bool: 如果找到目标并发起取消请求返回True否则返回False
"""
if goal_uuid in self._goals:
self.lab_logger().info(f"[Host Node] Cancelling goal {goal_uuid}")
self._goals[goal_uuid].cancel_goal_async()
self.lab_logger().info(f"[Host Node] Cancelling goal {goal_uuid[:8]}")
goal_handle = self._goals[goal_uuid]
# 发起异步取消请求
cancel_future = goal_handle.cancel_goal_async()
# 添加取消完成的回调
cancel_future.add_done_callback(lambda future: self._cancel_goal_callback(goal_uuid, future))
return True
else:
self.lab_logger().warning(f"[Host Node] Goal {goal_uuid} not found, cannot cancel")
self.lab_logger().warning(f"[Host Node] Goal {goal_uuid[:8]} not found in _goals, cannot cancel")
return False
def _cancel_goal_callback(self, goal_uuid: str, future) -> None:
"""取消目标的回调"""
try:
cancel_response = future.result()
if cancel_response.goals_canceling:
self.lab_logger().info(f"[Host Node] Goal {goal_uuid[:8]} cancel request accepted")
else:
self.lab_logger().warning(f"[Host Node] Goal {goal_uuid[:8]} cancel request rejected")
except Exception as e:
self.lab_logger().error(f"[Host Node] Error cancelling goal {goal_uuid[:8]}: {str(e)}")
import traceback
self.lab_logger().error(traceback.format_exc())
def get_goal_status(self, job_id: str) -> int:
"""获取目标状态"""
@@ -877,11 +935,10 @@ class HostNode(BaseROS2DeviceNode):
success = False
uuid_mapping = {}
if len(self.bridges) > 0:
from unilabos.app.web.client import HTTPClient
from unilabos.app.web.client import HTTPClient, http_client
client: HTTPClient = self.bridges[-1]
resource_start_time = time.time()
uuid_mapping = client.resource_tree_add(resource_tree_set, mount_uuid, first_add)
uuid_mapping = http_client.resource_tree_add(resource_tree_set, mount_uuid, first_add)
success = True
resource_end_time = time.time()
self.lab_logger().info(
@@ -989,9 +1046,10 @@ class HostNode(BaseROS2DeviceNode):
"""
更新节点信息回调
"""
self.lab_logger().info(f"[Host Node] Node info update request received: {request}")
# self.lab_logger().info(f"[Host Node] Node info update request received: {request}")
try:
from unilabos.app.communication import get_communication_client
from unilabos.app.web.client import HTTPClient, http_client
info = json.loads(request.command)
if "SYNC_SLAVE_NODE_INFO" in info:
@@ -1000,10 +1058,10 @@ class HostNode(BaseROS2DeviceNode):
edge_device_id = info["edge_device_id"]
self.device_machine_names[edge_device_id] = machine_name
else:
comm_client = get_communication_client()
registry_config = info["registry_config"]
for device_config in registry_config:
comm_client.publish_registry(device_config["id"], device_config)
devices_config = info.pop("devices_config")
registry_config = info.pop("registry_config")
if registry_config:
http_client.resource_registry({"resources": registry_config})
self.lab_logger().debug(f"[Host Node] Node info update: {info}")
response.response = "OK"
except Exception as e:
@@ -1029,10 +1087,9 @@ class HostNode(BaseROS2DeviceNode):
success = False
if len(self.bridges) > 0: # 边的提交待定
from unilabos.app.web.client import HTTPClient
from unilabos.app.web.client import HTTPClient, http_client
client: HTTPClient = self.bridges[-1]
r = client.resource_add(add_schema(resources))
r = http_client.resource_add(add_schema(resources))
success = bool(r)
response.success = success