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

liquid states

Browse Source
This commit is contained in:
Xuwznln
2025-06-09 00:52:07 +08:00
parent d2782943b4
commit f2e1d8a6c8
8 changed files with 17248 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8235
test/experiments/plr_test_converted.json
View File

File diff suppressed because it is too large Load Diff

4
unilabos/app/web/client.py
View File

@@ -60,7 +60,7 @@ class HTTPClient:
Dict: 返回的资源数据
"""
response = requests.get(
f"{self.remote_addr}/lab/resource/",
f"{self.remote_addr}/lab/resource/?edge_format=1",
params={"id": id, "with_children": with_children},
headers={"Authorization": f"lab {self.auth}"},
timeout=5,
@@ -96,7 +96,7 @@ class HTTPClient:
Response: API响应对象
"""
response = requests.patch(
f"{self.remote_addr}/lab/resource/batch_update/",
f"{self.remote_addr}/lab/resource/batch_update/?edge_format=1",
json=resources,
headers={"Authorization": f"lab {self.auth}"},
timeout=5,

9409
unilabos/devices/liquid_handling/converted protocol/sci-lucif-assay4_plr_background_tested.ipynb
View File

File diff suppressed because it is too large Load Diff

2
unilabos/resources/graphio.py
View File

@@ -326,8 +326,6 @@ def convert_resources_to_type(
elif all(issubclass(t, ResourcePLR) for t in resource_type):
resources_tree = dict_to_tree({r["id"]: r for r in resources_list})
return [resource_ulab_to_plr(r, plr_model) for r in resources_tree]
else:
return None

9
unilabos/ros/main_slave_run.py
View File

@@ -6,7 +6,6 @@ import time
from typing import Optional, Dict, Any, List
import rclpy
from unilabos.ros.nodes.presets.joint_republisher import JointRepublisher
from unilabos.ros.nodes.presets.resource_mesh_manager import ResourceMeshManager
from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker
from unilabos.devices.ros_dev.liquid_handler_joint_publisher import LiquidHandlerJointPublisher
@@ -70,6 +69,8 @@ def main(
)
if visual != "disable":
from unilabos.ros.nodes.presets.joint_republisher import JointRepublisher
resource_mesh_manager = ResourceMeshManager(
resources_mesh_config,
resources_config,
@@ -80,11 +81,11 @@ def main(
'joint_republisher',
host_node.resource_tracker
)
lh_joint_pub = LiquidHandlerJointPublisher(resources_config=resources_config, resource_tracker=host_node.resource_tracker)
executor.add_node(lh_joint_pub)
lh_joint_pub = LiquidHandlerJointPublisher(resources_config=resources_config,
resource_tracker=host_node.resource_tracker)
executor.add_node(resource_mesh_manager)
executor.add_node(joint_republisher)
executor.add_node(lh_joint_pub)
thread = threading.Thread(target=executor.spin, daemon=True, name="host_executor_thread")
thread.start()

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

@@ -740,16 +740,23 @@ class BaseROS2DeviceNode(Node, Generic[T]):
self.lab_logger().info(f"更新资源状态: {k}")
r = ResourceUpdate.Request()
# 仅当action_kwargs[k]不为None时尝试转换
akv = action_kwargs[k]
akv = action_kwargs[k] # 已经是完成转换的物料了，只需要转换成ros msg Resource了
apv = action_paramtypes[k]
final_type = get_type_class(apv)
if final_type is None:
continue
try:
r.resources = [
convert_to_ros_msg(Resource, self.resource_tracker.root_resource(rs))
for rs in convert_resources_from_type(akv, final_type) # type: ignore # FIXME # 考虑反查到最大的
]
seen = set()
unique_resources = []
for rs in akv:
res = self.resource_tracker.parent_resource(rs) # 获取 resource 对象
if id(res) not in seen:
seen.add(id(res))
converted_list = convert_resources_from_type([res], final_type)
unique_resources.extend([convert_to_ros_msg(Resource, converted) for converted in converted_list])
r.resources = unique_resources
response = await self._resource_clients["resource_update"].call_async(r)
self.lab_logger().debug(f"资源更新结果: {response}")
except Exception as e:

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

@@ -1,3 +1,5 @@
from typing import List, Tuple, Any
from unilabos.utils.log import logger
@@ -5,12 +7,12 @@ class DeviceNodeResourceTracker(object):
def __init__(self):
self.resources = []
self.root_resource2resource = {}
self.resource2parent_resource = {}
pass
def root_resource(self, resource):
if id(resource) in self.root_resource2resource:
return self.root_resource2resource[id(resource)]
def parent_resource(self, resource):
if id(resource) in self.resource2parent_resource:
return self.resource2parent_resource[id(resource)]
else:
return resource
@@ -44,20 +46,21 @@ class DeviceNodeResourceTracker(object):
self.loop_find_resource(r, resource_cls_type, identifier_key, getattr(query_resource, identifier_key))
)
assert len(res_list) == 1, f"{query_resource} 找到多个资源，请检查资源是否唯一: {res_list}"
self.root_resource2resource[id(query_resource)] = res_list[0]
self.resource2parent_resource[id(query_resource)] = res_list[0][0]
self.resource2parent_resource[id(res_list[0][1])] = res_list[0][0]
# 后续加入其他对比方式
return res_list[0]
return res_list[0][1]
def loop_find_resource(self, resource, target_resource_cls_type, identifier_key, compare_value):
def loop_find_resource(self, resource, target_resource_cls_type, identifier_key, compare_value, parent_res=None) -> List[Tuple[Any, Any]]:
res_list = []
# 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, target_resource_cls_type, identifier_key, compare_value))
res_list.extend(self.loop_find_resource(child, target_resource_cls_type, identifier_key, compare_value, resource))
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)
res_list.append((parent_res, resource))
return res_list
def filter_find_list(self, res_list, compare_std_dict):

48
unilabos/ros/utils/driver_creator.py
View File

@@ -105,40 +105,41 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
return nested_dict_to_list(resource), Resource
return resource, source_type
def _process_resource_references(self, data: Any, to_dict=False) -> Any:
def _process_resource_references(self, data: Any, to_dict=False, states=None, prefix_path="") -> Any:
"""
递归处理资源引用，替换_resource_child_name对应的资源
Args:
data: 需要处理的数据，可能是字典、列表或其他类型
to_dict: 转换成对应的实例，还是转换成对应的字典
to_dict: 是否返回字典形式的资源
states: 用于保存所有资源状态
prefix_path: 当前递归路径
Returns:
处理后的数据
"""
from pylabrobot.resources import Deck, Resource
if states is None:
states = {}
if isinstance(data, dict):
# 检查是否包含资源引用
if "_resource_child_name" in data:
child_name = data["_resource_child_name"]
if child_name in self.children:
# 找到了对应的资源
resource = self.children[child_name]
# 检查是否需要转换资源类型
if "_resource_type" in data:
type_path = data["_resource_type"]
try:
# 尝试导入指定的类型
target_type = import_manager.get_class(type_path)
contain_model = not issubclass(target_type, Deck)
resource, target_type = self._process_resource_mapping(resource, target_type)
# 在截图中格式，是deserialize，所以这里要转成plr resource可deserialize的字典
# 这样后面执行deserialize的时候能够正确反序列化对应的物料
resource_instance: Resource = resource_ulab_to_plr(resource, contain_model)
# 使用 prefix_path 作为 key 存储资源状态
if to_dict:
return resource_instance.serialize()
serialized = resource_instance.serialize()
states[prefix_path] = resource_instance.serialize_all_state()
return serialized
else:
self.resource_tracker.add_resource(resource_instance)
return resource_instance
@@ -151,17 +152,20 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
else:
logger.warning(f"找不到资源引用 '{child_name}'，保持原值不变")
# 递归处理字典的每个值
# 递归处理每个键值
result = {}
for key, value in data.items():
result[key] = self._process_resource_references(value, to_dict)
new_prefix = f"{prefix_path}.{key}" if prefix_path else key
result[key] = self._process_resource_references(value, to_dict, states, new_prefix)
return result
# 处理列表类型
elif isinstance(data, list):
return [self._process_resource_references(item, to_dict) for item in data]
return [
self._process_resource_references(item, to_dict, states, f"{prefix_path}[{i}]")
for i, item in enumerate(data)
]
# 其他类型直接返回
else:
return data
def create_instance(self, data: Dict[str, Any]) -> Optional[T]:
@@ -187,10 +191,18 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
logger.debug(f"自动补充 _resource_type: {data[param_name]['_resource_type']}")
# 首先处理资源引用
processed_data = self._process_resource_references(data, to_dict=True)
states = {}
processed_data = self._process_resource_references(data, to_dict=True, states=states)
try:
self.device_instance = deserialize_method(**processed_data)
all_states = self.device_instance.serialize_all_state()
for k, v in states.items():
logger.debug(f"PyLabRobot反序列化设置状态：{k}")
for kk, vv in all_states.items():
if kk not in v:
v[kk] = vv
self.device_instance.deck.load_all_state(v)
self.resource_tracker.add_resource(self.device_instance)
self.post_create()
return self.device_instance # type: ignore
@@ -225,6 +237,10 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
if hasattr(self.device_instance, "setup") and asyncio.iscoroutinefunction(getattr(self.device_instance, "setup")):
from unilabos.ros.nodes.base_device_node import ROS2DeviceNode
def done_cb(*args):
from pylabrobot.resources import set_volume_tracking
# from pylabrobot.resources import set_tip_tracking
set_volume_tracking(enabled=True)
# set_tip_tracking(enabled=True) # 序列化tip_spot has为False
logger.debug(f"PyLabRobot设备实例 {self.device_instance} 设置完成")
from unilabos.config.config import BasicConfig
if BasicConfig.vis_2d_enable: