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

feat: resource tracker support dict

Browse Source
This commit is contained in:
wznln
2025-05-06 17:28:06 +08:00
parent 49a9f05c51
commit b6a3f17e9b
4 changed files with 44 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
unilabos/ros/nodes/resource_tracker.py
View File

@@ -1,7 +1,7 @@
from unilabos.utils.log import logger
class DeviceNodeResourceTracker:
class DeviceNodeResourceTracker(object):
def __init__(self):
self.resources = []
@@ -15,44 +15,46 @@ class DeviceNodeResourceTracker:
return resource
def add_resource(self, resource):
# 使用内存地址跟踪是否为同一个resource
for r in self.resources:
if id(r) == id(resource):
return
# 添加资源到跟踪器
self.resources.append(resource)
def clear_resource(self):
self.resources = []
def figure_resource(self, resource):
# 使用内存地址跟踪是否为同一个resource
if isinstance(resource, list):
return [self.figure_resource(r) for r in resource]
res_id = resource.id if hasattr(resource, "id") else None
res_name = resource.name if hasattr(resource, "name") else None
def figure_resource(self, query_resource):
if isinstance(query_resource, list):
return [self.figure_resource(r) for r in query_resource]
res_id = query_resource.id if hasattr(query_resource, "id") else (query_resource.get("id") if isinstance(query_resource, dict) else None)
res_name = query_resource.name if hasattr(query_resource, "name") else (query_resource.get("name") if isinstance(query_resource, dict) else None)
res_identifier = res_id if res_id else res_name
identifier_key = "id" if res_id else "name"
resource_cls_type = type(resource)
resource_cls_type = type(query_resource)
if res_identifier is None:
logger.warning(f"resource {resource} 没有id或name，暂不能对应figure")
logger.warning(f"resource {query_resource} 没有id或name，暂不能对应figure")
res_list = []
for r in self.resources:
res_list.extend(
self.loop_find_resource(r, resource_cls_type, identifier_key, getattr(resource, identifier_key))
)
if isinstance(query_resource, dict):
res_list.extend(
self.loop_find_resource(r, resource_cls_type, identifier_key, query_resource[identifier_key])
)
else:
res_list.extend(
self.loop_find_resource(r, resource_cls_type, identifier_key, getattr(query_resource, identifier_key))
)
assert len(res_list) == 1, f"找到多个资源，请检查资源是否唯一: {res_list}"
self.root_resource2resource[id(resource)] = res_list[0]
self.root_resource2resource[id(query_resource)] = res_list[0]
# 后续加入其他对比方式
return res_list[0]
def loop_find_resource(self, resource, resource_cls_type, identifier_key, compare_value):
def loop_find_resource(self, resource, target_resource_cls_type, identifier_key, compare_value):
res_list = []
print(resource, resource_cls_type, identifier_key, compare_value)
# print(resource, target_resource_cls_type, identifier_key, compare_value)
children = getattr(resource, "children", [])
for child in children:
res_list.extend(self.loop_find_resource(child, resource_cls_type, identifier_key, compare_value))
if resource_cls_type == type(resource):
res_list.extend(self.loop_find_resource(child, target_resource_cls_type, identifier_key, compare_value))
if target_resource_cls_type == type(resource) or target_resource_cls_type == dict:
if hasattr(resource, identifier_key):
if getattr(resource, identifier_key) == compare_value:
res_list.append(resource)