mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2025-12-14 13:14:39 +00:00
7c440d10ab
* Fix ResourceTreeSet load error * Raise error when using unsupported type to create ResourceTreeSet * Fix children key error * Fix children key error * Fix workstation resource not tracking * Fix workstation deck & children resource dupe * Fix workstation deck & children resource dupe * Fix multiple resource error * Fix resource tree update * Fix resource tree update * Force confirm uuid * Tip more error log * Refactor Bioyond workstation and experiment workflow (#105) Refactored the Bioyond workstation classes to improve parameter handling and workflow management. Updated experiment.py to use BioyondReactionStation with deck and material mappings, and enhanced workflow step parameter mapping and execution logic. Adjusted JSON experiment configs, improved workflow sequence handling, and added UUID assignment to PLR materials. Removed unused station_config and material cache logic, and added detailed docstrings and debug output for workflow methods. * Fix resource get. Fix resource parent not found. Mapping uuid for all resources. * mount parent uuid * Add logging configuration based on BasicConfig in main function * fix workstation node error * fix workstation node error * Update boot example * temp fix for resource get * temp fix for resource get * provide error info when cant find plr type * pack repo info * fix to plr type error * fix to plr type error * Update regular container method * support no size init * fix comprehensive_station.json * fix comprehensive_station.json * fix type conversion * fix state loading for regular container * Update deploy-docs.yml * Update deploy-docs.yml --------- Co-authored-by: ZiWei <131428629+ZiWei09@users.noreply.github.com>
1173 lines
47 KiB
Python
1173 lines
47 KiB
Python
import traceback
|
||
import uuid
|
||
from pydantic import BaseModel, field_serializer, field_validator
|
||
from pydantic import Field
|
||
from typing import List, Tuple, Any, Dict, Literal, Optional, cast, TYPE_CHECKING
|
||
|
||
from unilabos.utils.log import logger
|
||
|
||
if TYPE_CHECKING:
|
||
from unilabos.devices.workstation.workstation_base import WorkstationBase
|
||
from pylabrobot.resources import Resource as PLRResource
|
||
|
||
|
||
class ResourceDictPositionSize(BaseModel):
|
||
depth: float = Field(description="Depth", default=0.0)
|
||
width: float = Field(description="Width", default=0.0)
|
||
height: float = Field(description="Height", default=0.0)
|
||
|
||
|
||
class ResourceDictPositionScale(BaseModel):
|
||
x: float = Field(description="x scale", default=0.0)
|
||
y: float = Field(description="y scale", default=0.0)
|
||
z: float = Field(description="z scale", default=0.0)
|
||
|
||
|
||
class ResourceDictPositionObject(BaseModel):
|
||
x: float = Field(description="X coordinate", default=0.0)
|
||
y: float = Field(description="Y coordinate", default=0.0)
|
||
z: float = Field(description="Z coordinate", default=0.0)
|
||
|
||
|
||
class ResourceDictPosition(BaseModel):
|
||
size: ResourceDictPositionSize = Field(description="Resource size", default_factory=ResourceDictPositionSize)
|
||
scale: ResourceDictPositionScale = Field(description="Resource scale", default_factory=ResourceDictPositionScale)
|
||
layout: Literal["2d"] = Field(description="Resource layout", default="2d")
|
||
position: ResourceDictPositionObject = Field(
|
||
description="Resource position", default_factory=ResourceDictPositionObject
|
||
)
|
||
position3d: ResourceDictPositionObject = Field(
|
||
description="Resource position in 3D space", default_factory=ResourceDictPositionObject
|
||
)
|
||
rotation: ResourceDictPositionObject = Field(
|
||
description="Resource rotation", default_factory=ResourceDictPositionObject
|
||
)
|
||
|
||
|
||
# 统一的资源字典模型，parent 自动序列化为 parent_uuid，children 不序列化
|
||
class ResourceDict(BaseModel):
|
||
id: str = Field(description="Resource ID")
|
||
uuid: str = Field(description="Resource UUID")
|
||
name: str = Field(description="Resource name")
|
||
description: str = Field(description="Resource description", default="")
|
||
resource_schema: Dict[str, Any] = Field(description="Resource schema", default_factory=dict, serialization_alias="schema", validation_alias="schema")
|
||
model: Dict[str, Any] = Field(description="Resource model", default_factory=dict)
|
||
icon: str = Field(description="Resource icon", default="")
|
||
parent_uuid: Optional["str"] = Field(description="Parent resource uuid", default=None) # 先设定parent_uuid
|
||
parent: Optional["ResourceDict"] = Field(description="Parent resource object", default=None, exclude=True)
|
||
type: Literal["device"] | str = Field(description="Resource type")
|
||
klass: str = Field(alias="class", description="Resource class name")
|
||
position: ResourceDictPosition = Field(description="Resource position", default_factory=ResourceDictPosition)
|
||
config: Dict[str, Any] = Field(description="Resource configuration")
|
||
data: Dict[str, Any] = Field(description="Resource data")
|
||
|
||
@field_serializer("parent_uuid")
|
||
def _serialize_parent(self, parent_uuid: Optional["ResourceDict"]):
|
||
return self.uuid_parent
|
||
|
||
@field_validator("parent", mode="before")
|
||
@classmethod
|
||
def _deserialize_parent(cls, parent: Optional["ResourceDict"]):
|
||
if isinstance(parent, ResourceDict):
|
||
return parent
|
||
else:
|
||
return None
|
||
|
||
@property
|
||
def uuid_parent(self) -> str:
|
||
"""获取父节点的UUID"""
|
||
parent_instance_uuid = self.parent_instance_uuid
|
||
if parent_instance_uuid is not None and self.parent_uuid and parent_instance_uuid != self.parent_uuid:
|
||
logger.warning(f"{self.name}[{self.uuid}]的parent uuid未同步！") # 现在强制要求设置
|
||
if parent_instance_uuid is not None:
|
||
return parent_instance_uuid
|
||
return self.parent_uuid
|
||
|
||
@property
|
||
def parent_instance_uuid(self) -> Optional[str]:
|
||
"""获取父节点的UUID"""
|
||
return self.parent.uuid if self.parent is not None else None
|
||
|
||
@property
|
||
def parent_instance_name(self) -> Optional[str]:
|
||
"""获取父节点的名字"""
|
||
return self.parent.name if self.parent is not None else None
|
||
|
||
@property
|
||
def is_root_node(self) -> bool:
|
||
"""判断资源是否为根节点"""
|
||
return self.parent is None
|
||
|
||
|
||
class GraphData(BaseModel):
|
||
"""图数据结构，包含节点和边"""
|
||
|
||
nodes: List["ResourceTreeInstance"] = Field(description="Resource nodes list", default_factory=list)
|
||
links: List[Dict[str, Any]] = Field(description="Resource links/edges list", default_factory=list)
|
||
|
||
|
||
class ResourceDictInstance(object):
|
||
"""ResourceDict的实例，同时提供一些方法"""
|
||
|
||
def __init__(self, res_content: "ResourceDict"):
|
||
self.res_content = res_content
|
||
self.children: List[ResourceDictInstance] = []
|
||
self.typ = "dict"
|
||
|
||
@classmethod
|
||
def get_resource_instance_from_dict(cls, content: Dict[str, Any]) -> "ResourceDictInstance":
|
||
"""从字典创建资源实例"""
|
||
if "id" not in content:
|
||
content["id"] = content["name"]
|
||
if "uuid" not in content:
|
||
content["uuid"] = str(uuid.uuid4())
|
||
if "description" in content and content["description"] is None:
|
||
del content["description"]
|
||
if "model" in content and content["model"] is None:
|
||
del content["model"]
|
||
if "schema" in content and content["schema"] is None:
|
||
del content["schema"]
|
||
if "x" in content.get("position", {}):
|
||
# 说明是老版本的position格式，转换成新的
|
||
content["position"] = {"position": content["position"]}
|
||
if not content.get("class"):
|
||
content["class"] = ""
|
||
if not content.get("config"): # todo: 后续从后端保证字段非空
|
||
content["config"] = {}
|
||
if not content.get("data"):
|
||
content["data"] = {}
|
||
return ResourceDictInstance(ResourceDict.model_validate(content))
|
||
|
||
def get_nested_dict(self) -> Dict[str, Any]:
|
||
"""获取资源实例的嵌套字典表示"""
|
||
res_dict = self.res_content.model_dump(by_alias=True)
|
||
res_dict["children"] = {child.res_content.id: child.get_nested_dict() for child in self.children}
|
||
res_dict["parent"] = self.res_content.parent_instance_name
|
||
res_dict["position"] = self.res_content.position.position.model_dump()
|
||
return res_dict
|
||
|
||
|
||
class ResourceTreeInstance(object):
|
||
"""
|
||
资源树，表示一个根节点及其所有子节点的层次结构，继承ResourceDictInstance表示自己是根节点
|
||
"""
|
||
|
||
@staticmethod
|
||
def _build_uuid_map(resource_list: List[ResourceDictInstance]) -> Dict[str, ResourceDictInstance]:
|
||
"""构建uuid到资源对象的映射，并检查重复"""
|
||
uuid_map: Dict[str, ResourceDictInstance] = {}
|
||
for res_instance in resource_list:
|
||
res = res_instance.res_content
|
||
if res.uuid in uuid_map:
|
||
raise ValueError(f"发现重复的uuid: {res.uuid}")
|
||
uuid_map[res.uuid] = res_instance
|
||
return uuid_map
|
||
|
||
@staticmethod
|
||
def _build_uuid_instance_map(
|
||
resource_list: List[ResourceDictInstance],
|
||
) -> Dict[str, ResourceDictInstance]:
|
||
"""构建uuid到资源实例的映射"""
|
||
return {res_instance.res_content.uuid: res_instance for res_instance in resource_list}
|
||
|
||
@staticmethod
|
||
def _collect_tree_nodes(
|
||
root_instance: ResourceDictInstance, uuid_map: Dict[str, ResourceDict]
|
||
) -> List[ResourceDictInstance]:
|
||
"""使用BFS收集属于某个根节点的所有节点"""
|
||
# BFS遍历，根据parent_uuid字段找到所有属于这棵树的节点
|
||
tree_nodes = [root_instance]
|
||
visited = {root_instance.res_content.uuid}
|
||
queue = [root_instance.res_content.uuid]
|
||
|
||
while queue:
|
||
current_uuid = queue.pop(0)
|
||
# 查找所有parent_uuid指向当前节点的子节点
|
||
for uuid_str, res in uuid_map.items():
|
||
if res.uuid_parent == current_uuid and uuid_str not in visited:
|
||
child_instance = ResourceDictInstance(res)
|
||
tree_nodes.append(child_instance)
|
||
visited.add(uuid_str)
|
||
queue.append(uuid_str)
|
||
|
||
return tree_nodes
|
||
|
||
def __init__(self, resource: ResourceDictInstance):
|
||
self.root_node = resource
|
||
self._validate_tree()
|
||
|
||
def _validate_tree(self):
|
||
"""
|
||
验证树结构的一致性
|
||
- 验证uuid唯一性
|
||
- 验证parent-children关系一致性
|
||
|
||
Raises:
|
||
ValueError: 当发现不一致时
|
||
"""
|
||
known_uuids: set = set()
|
||
|
||
def validate_node(node: ResourceDictInstance):
|
||
# 检查uuid唯一性
|
||
if node.res_content.uuid in known_uuids:
|
||
raise ValueError(f"发现重复的uuid: {node.res_content.uuid}")
|
||
if node.res_content.uuid:
|
||
known_uuids.add(node.res_content.uuid)
|
||
else:
|
||
logger.warning(f"警告: 资源 {node.res_content.id} 没有uuid")
|
||
|
||
# 验证并递归处理子节点
|
||
for child in node.children:
|
||
if child.res_content.parent != node.res_content:
|
||
parent_id = child.res_content.parent.id if child.res_content.parent else None
|
||
raise ValueError(
|
||
f"节点 {child.res_content.id} 的parent引用不正确，应该指向 {node.res_content.id}，但实际指向 {parent_id}"
|
||
)
|
||
validate_node(child)
|
||
|
||
validate_node(self.root_node)
|
||
|
||
def get_all_nodes(self) -> List[ResourceDictInstance]:
|
||
"""
|
||
获取树中的所有节点（深度优先遍历）
|
||
|
||
Returns:
|
||
所有节点的资源实例列表
|
||
"""
|
||
nodes = []
|
||
|
||
def collect_nodes(node: ResourceDictInstance):
|
||
nodes.append(node)
|
||
for child in node.children:
|
||
collect_nodes(child)
|
||
|
||
collect_nodes(self.root_node)
|
||
return nodes
|
||
|
||
def find_by_uuid(self, target_uuid: str) -> Optional[ResourceDictInstance]:
|
||
"""
|
||
通过uuid查找节点
|
||
|
||
Args:
|
||
target_uuid: 目标uuid
|
||
|
||
Returns:
|
||
找到的节点资源实例，如果没找到返回None
|
||
"""
|
||
|
||
def search(node: ResourceDictInstance) -> Optional[ResourceDictInstance]:
|
||
if node.res_content.uuid == target_uuid:
|
||
return node
|
||
for child in node.children:
|
||
res = search(child)
|
||
if res:
|
||
return res
|
||
return None
|
||
|
||
result = search(self.root_node)
|
||
return result
|
||
|
||
|
||
class ResourceTreeSet(object):
|
||
"""
|
||
多个根节点的resource集合，包含多个ResourceTree
|
||
"""
|
||
|
||
def __init__(self, resource_list: List[List[ResourceDictInstance]] | List[ResourceTreeInstance]):
|
||
"""
|
||
初始化资源树集合
|
||
|
||
Args:
|
||
resource_list: 可以是以下两种类型之一：
|
||
- List[ResourceTree]: 已经构建好的树列表
|
||
- List[List[ResourceInstanceDict]]: 嵌套列表，每个内部列表代表一棵树
|
||
|
||
Raises:
|
||
TypeError: 当传入不支持的类型时
|
||
"""
|
||
if not resource_list:
|
||
self.trees: List[ResourceTreeInstance] = []
|
||
elif isinstance(resource_list[0], ResourceTreeInstance):
|
||
# 已经是ResourceTree列表
|
||
self.trees = cast(List[ResourceTreeInstance], resource_list)
|
||
else:
|
||
raise TypeError(
|
||
f"不支持的类型: {type(resource_list[0])}。"
|
||
f"ResourceTreeSet 只接受 List[ResourceTree] 或 List[List[ResourceInstanceDict]]"
|
||
)
|
||
|
||
@classmethod
|
||
def from_plr_resources(cls, resources: List["PLRResource"]) -> "ResourceTreeSet":
|
||
"""
|
||
从plr资源创建ResourceTreeSet
|
||
"""
|
||
|
||
def replace_plr_type(source: str):
|
||
replace_info = {
|
||
"plate": "plate",
|
||
"well": "well",
|
||
"deck": "deck",
|
||
}
|
||
if source in replace_info:
|
||
return replace_info[source]
|
||
else:
|
||
print("转换pylabrobot的时候，出现未知类型", source)
|
||
return "container"
|
||
|
||
def build_uuid_mapping(res: "PLRResource", uuid_list: list):
|
||
"""递归构建uuid映射字典"""
|
||
uid = getattr(res, "unilabos_uuid", "")
|
||
if not uid:
|
||
uid = str(uuid.uuid4())
|
||
res.unilabos_uuid = uid
|
||
logger.warning(f"{res}没有uuid，请设置后再传入，默认填充{uid}！\n{traceback.format_exc()}")
|
||
uuid_list.append(uid)
|
||
for child in res.children:
|
||
build_uuid_mapping(child, uuid_list)
|
||
|
||
def resource_plr_inner(
|
||
d: dict, parent_resource: Optional[ResourceDict], states: dict, uuids: list
|
||
) -> ResourceDictInstance:
|
||
current_uuid = uuids.pop(0)
|
||
|
||
# 先构建当前节点的字典（不包含children）
|
||
r_dict = {
|
||
"id": d["name"],
|
||
"uuid": current_uuid,
|
||
"name": d["name"],
|
||
"parent": parent_resource, # 直接传入 ResourceDict 对象
|
||
"type": replace_plr_type(d.get("category", "")),
|
||
"class": d.get("class", ""),
|
||
"position": (
|
||
{"x": d["location"]["x"], "y": d["location"]["y"], "z": d["location"]["z"]}
|
||
if d["location"]
|
||
else {"x": 0, "y": 0, "z": 0}
|
||
),
|
||
"config": {k: v for k, v in d.items() if k not in ["name", "children", "parent_name", "location"]},
|
||
"data": states[d["name"]],
|
||
}
|
||
|
||
# 先转换为 ResourceDictInstance，获取其中的 ResourceDict
|
||
current_instance = ResourceDictInstance.get_resource_instance_from_dict(r_dict)
|
||
current_resource = current_instance.res_content
|
||
|
||
# 递归处理子节点，传入当前节点的 ResourceDict 作为 parent
|
||
current_instance.children = [
|
||
resource_plr_inner(child, current_resource, states, uuids) for child in d["children"]
|
||
]
|
||
|
||
return current_instance
|
||
|
||
trees = []
|
||
for resource in resources:
|
||
# 构建uuid列表
|
||
uuid_list = []
|
||
build_uuid_mapping(resource, uuid_list)
|
||
|
||
serialized_data = resource.serialize()
|
||
all_states = resource.serialize_all_state()
|
||
|
||
# 根节点没有父节点，传入 None
|
||
root_instance = resource_plr_inner(serialized_data, None, all_states, uuid_list)
|
||
tree_instance = ResourceTreeInstance(root_instance)
|
||
trees.append(tree_instance)
|
||
return cls(trees)
|
||
|
||
def to_plr_resources(self) -> List["PLRResource"]:
|
||
"""
|
||
将 ResourceTreeSet 转换为 PLR 资源列表
|
||
|
||
Returns:
|
||
List[PLRResource]: PLR 资源实例列表
|
||
"""
|
||
from pylabrobot.resources import Resource as PLRResource
|
||
from pylabrobot.utils.object_parsing import find_subclass
|
||
import inspect
|
||
|
||
# 类型映射
|
||
TYPE_MAP = {"plate": "Plate", "well": "Well", "deck": "Deck"}
|
||
|
||
def collect_node_data(node: ResourceDictInstance, name_to_uuid: dict, all_states: dict):
|
||
"""一次遍历收集 name_to_uuid 和 all_states"""
|
||
name_to_uuid[node.res_content.name] = node.res_content.uuid
|
||
all_states[node.res_content.name] = node.res_content.data
|
||
for child in node.children:
|
||
collect_node_data(child, name_to_uuid, all_states)
|
||
|
||
def node_to_plr_dict(node: ResourceDictInstance, has_model: bool):
|
||
"""转换节点为 PLR 字典格式"""
|
||
res = node.res_content
|
||
plr_type = TYPE_MAP.get(res.type, res.type)
|
||
if res.type not in TYPE_MAP:
|
||
logger.warning(f"未知类型 {res.type}")
|
||
|
||
d = {
|
||
"name": res.name,
|
||
"type": res.config.get("type", plr_type),
|
||
"size_x": res.config.get("size_x", 0),
|
||
"size_y": res.config.get("size_y", 0),
|
||
"size_z": res.config.get("size_z", 0),
|
||
"location": {
|
||
"x": res.position.position.x,
|
||
"y": res.position.position.y,
|
||
"z": res.position.position.z,
|
||
"type": "Coordinate",
|
||
},
|
||
"rotation": {"x": 0, "y": 0, "z": 0, "type": "Rotation"},
|
||
"category": res.config.get("category", plr_type),
|
||
"children": [node_to_plr_dict(child, has_model) for child in node.children],
|
||
"parent_name": res.parent_instance_name,
|
||
**res.config,
|
||
}
|
||
if has_model:
|
||
d["model"] = res.config.get("model", None)
|
||
return d
|
||
|
||
plr_resources = []
|
||
trees = []
|
||
tracker = DeviceNodeResourceTracker()
|
||
|
||
for tree in self.trees:
|
||
name_to_uuid: Dict[str, str] = {}
|
||
all_states: Dict[str, Any] = {}
|
||
collect_node_data(tree.root_node, name_to_uuid, all_states)
|
||
has_model = tree.root_node.res_content.type != "deck"
|
||
plr_dict = node_to_plr_dict(tree.root_node, has_model)
|
||
try:
|
||
sub_cls = find_subclass(plr_dict["type"], PLRResource)
|
||
if sub_cls is None:
|
||
raise ValueError(f"无法找到类型 {plr_dict['type']} 对应的 PLR 资源类。原始信息：{tree.root_node.res_content}")
|
||
spec = inspect.signature(sub_cls)
|
||
if "category" not in spec.parameters:
|
||
plr_dict.pop("category", None)
|
||
plr_resource = sub_cls.deserialize(plr_dict, allow_marshal=True)
|
||
plr_resource.load_all_state(all_states)
|
||
# 使用 DeviceNodeResourceTracker 设置 UUID
|
||
tracker.loop_set_uuid(plr_resource, name_to_uuid)
|
||
plr_resources.append(plr_resource)
|
||
|
||
except Exception as e:
|
||
logger.error(f"转换 PLR 资源失败: {e}")
|
||
import traceback
|
||
|
||
logger.error(f"堆栈: {traceback.format_exc()}")
|
||
raise
|
||
|
||
return plr_resources
|
||
|
||
@classmethod
|
||
def from_raw_list(cls, raw_list: List[Dict[str, Any]]) -> "ResourceTreeSet":
|
||
"""
|
||
从原始字典列表创建 ResourceTreeSet，自动建立 parent-children 关系
|
||
|
||
Args:
|
||
raw_list: 原始字典列表，每个字典代表一个资源节点
|
||
|
||
Returns:
|
||
ResourceTreeSet 实例
|
||
|
||
Raises:
|
||
ValueError: 当建立关系时发现不一致
|
||
"""
|
||
# 第一步：将字典列表转换为 ResourceDictInstance 列表
|
||
instances = [ResourceDictInstance.get_resource_instance_from_dict(node_dict) for node_dict in raw_list]
|
||
|
||
# 第二步：建立映射关系
|
||
uuid_to_instance: Dict[str, ResourceDictInstance] = {}
|
||
id_to_instance: Dict[str, ResourceDictInstance] = {}
|
||
|
||
for raw_node, instance in zip(raw_list, instances):
|
||
# 建立 uuid 映射
|
||
if instance.res_content.uuid:
|
||
uuid_to_instance[instance.res_content.uuid] = instance
|
||
# 建立 id 映射
|
||
if instance.res_content.id:
|
||
id_to_instance[instance.res_content.id] = instance
|
||
|
||
# 第三步：建立 parent-children 关系
|
||
for raw_node, instance in zip(raw_list, instances):
|
||
# 优先使用 parent_uuid 进行匹配，如果不存在则使用 parent (id)
|
||
parent_uuid = raw_node.get("parent_uuid")
|
||
parent_id = raw_node.get("parent")
|
||
parent_instance = None
|
||
|
||
# 优先用 parent_uuid 匹配
|
||
if parent_uuid and parent_uuid in uuid_to_instance:
|
||
parent_instance = uuid_to_instance[parent_uuid]
|
||
# 否则用 parent (id) 匹配
|
||
elif parent_id and parent_id in id_to_instance:
|
||
parent_instance = id_to_instance[parent_id]
|
||
|
||
# 设置 parent 引用并建立 children 关系
|
||
if parent_instance:
|
||
instance.res_content.parent = parent_instance.res_content
|
||
# 将当前节点添加到父节点的 children 列表（避免重复添加）
|
||
if instance not in parent_instance.children:
|
||
parent_instance.children.append(instance)
|
||
|
||
# 第四步：使用 from_nested_list 创建 ResourceTreeSet
|
||
return cls.from_nested_list(instances)
|
||
|
||
@classmethod
|
||
def from_nested_list(cls, nested_list: List[ResourceDictInstance]) -> "ResourceTreeSet":
|
||
"""
|
||
从扁平化的资源列表创建ResourceTreeSet，自动按根节点分组
|
||
|
||
Args:
|
||
nested_list: 扁平化的资源实例列表，可能包含多个根节点
|
||
|
||
Returns:
|
||
ResourceTreeSet实例
|
||
|
||
Raises:
|
||
ValueError: 当没有找到任何根节点时
|
||
"""
|
||
# 找到所有根节点
|
||
known_uuids = {res_instance.res_content.uuid for res_instance in nested_list}
|
||
root_instances = [
|
||
ResourceTreeInstance(res_instance)
|
||
for res_instance in nested_list
|
||
if res_instance.res_content.is_root_node or res_instance.res_content.uuid_parent not in known_uuids
|
||
]
|
||
return cls(root_instances)
|
||
|
||
@property
|
||
def root_nodes(self) -> List[ResourceDictInstance]:
|
||
"""
|
||
获取所有树的根节点
|
||
|
||
Returns:
|
||
所有根节点的资源实例列表
|
||
"""
|
||
return [tree.root_node for tree in self.trees]
|
||
|
||
@property
|
||
def all_nodes(self) -> List[ResourceDictInstance]:
|
||
"""
|
||
获取所有树中的所有节点
|
||
|
||
Returns:
|
||
所有节点的资源实例列表
|
||
"""
|
||
return [node for tree in self.trees for node in tree.get_all_nodes()]
|
||
|
||
@property
|
||
def all_nodes_uuid(self) -> List[str]:
|
||
"""
|
||
获取所有树中的所有节点
|
||
|
||
Returns:
|
||
所有节点的资源实例列表
|
||
"""
|
||
return [node.res_content.uuid for tree in self.trees for node in tree.get_all_nodes()]
|
||
|
||
def find_by_uuid(self, target_uuid: str) -> Optional[ResourceDictInstance]:
|
||
"""
|
||
在所有树中通过uuid查找节点
|
||
|
||
Args:
|
||
target_uuid: 目标uuid
|
||
|
||
Returns:
|
||
找到的节点资源实例，如果没找到返回None
|
||
"""
|
||
for tree in self.trees:
|
||
result = tree.find_by_uuid(target_uuid)
|
||
if result:
|
||
return result
|
||
return None
|
||
|
||
def merge_remote_resources(self, remote_tree_set: "ResourceTreeSet") -> "ResourceTreeSet":
|
||
"""
|
||
将远端物料同步到本地物料中（以子树为单位）
|
||
|
||
同步规则：
|
||
1. 一级节点（根节点）：如果不存在的物料，引入整个子树
|
||
2. 一级设备下的二级物料：如果不存在，引入整个子树
|
||
3. 二级设备下的三级物料：如果不存在，引入整个子树
|
||
如果存在则跳过并提示
|
||
|
||
Args:
|
||
remote_tree_set: 远端的资源树集合
|
||
|
||
Returns:
|
||
合并后的资源树集合（self）
|
||
"""
|
||
# 构建本地映射：一级 device id -> 根节点实例
|
||
local_device_map: Dict[str, ResourceDictInstance] = {}
|
||
for root_node in self.root_nodes:
|
||
if root_node.res_content.type == "device":
|
||
local_device_map[root_node.res_content.id] = root_node
|
||
|
||
# 记录需要添加的新根节点（不属于任何 device 的物料）
|
||
new_root_nodes: List[ResourceDictInstance] = []
|
||
|
||
# 遍历远端根节点
|
||
for remote_root in remote_tree_set.root_nodes:
|
||
remote_root_id = remote_root.res_content.id
|
||
remote_root_type = remote_root.res_content.type
|
||
|
||
if remote_root_type == "device":
|
||
# 情况1: 一级是 device
|
||
if remote_root_id not in local_device_map:
|
||
logger.warning(f"Device '{remote_root_id}' 在本地不存在，跳过该 device 下的物料同步")
|
||
continue
|
||
|
||
local_device = local_device_map[remote_root_id]
|
||
|
||
# 构建本地一级 device 下的子节点映射
|
||
local_children_map = {child.res_content.name: child for child in local_device.children}
|
||
|
||
# 遍历远端一级 device 的子节点
|
||
for remote_child in remote_root.children:
|
||
remote_child_name = remote_child.res_content.name
|
||
remote_child_type = remote_child.res_content.type
|
||
|
||
if remote_child_type == "device":
|
||
# 情况2: 二级是 device
|
||
if remote_child_name not in local_children_map:
|
||
logger.warning(f"Device '{remote_root_id}/{remote_child_name}' 在本地不存在，跳过")
|
||
continue
|
||
|
||
local_sub_device = local_children_map[remote_child_name]
|
||
|
||
# 构建本地二级 device 下的子节点映射
|
||
local_sub_children_map = {child.res_content.name: child for child in local_sub_device.children}
|
||
|
||
# 遍历远端二级 device 的子节点（三级物料）
|
||
added_count = 0
|
||
for remote_material in remote_child.children:
|
||
remote_material_name = remote_material.res_content.name
|
||
|
||
# 情况3: 三级物料
|
||
if remote_material_name not in local_sub_children_map:
|
||
# 引入整个子树
|
||
remote_material.res_content.parent = local_sub_device.res_content
|
||
local_sub_device.children.append(remote_material)
|
||
added_count += 1
|
||
else:
|
||
logger.info(
|
||
f"物料 '{remote_root_id}/{remote_child_name}/{remote_material_name}' "
|
||
f"已存在，跳过"
|
||
)
|
||
|
||
if added_count > 0:
|
||
logger.info(
|
||
f"Device '{remote_root_id}/{remote_child_name}': "
|
||
f"从远端同步了 {added_count} 个物料子树"
|
||
)
|
||
else:
|
||
# 情况2: 二级是物料（不是 device）
|
||
if remote_child_name not in local_children_map:
|
||
# 引入整个子树
|
||
remote_child.res_content.parent = local_device.res_content
|
||
local_device.children.append(remote_child)
|
||
logger.info(f"Device '{remote_root_id}': 从远端同步物料子树 '{remote_child_name}'")
|
||
else:
|
||
logger.info(f"物料 '{remote_root_id}/{remote_child_name}' 已存在，跳过")
|
||
else:
|
||
# 情况1: 一级节点是物料（不是 device）
|
||
# 检查是否已存在
|
||
existing = False
|
||
for local_root in self.root_nodes:
|
||
if local_root.res_content.name == remote_root.res_content.name:
|
||
existing = True
|
||
logger.info(f"根节点物料 '{remote_root.res_content.name}' 已存在，跳过")
|
||
break
|
||
|
||
if not existing:
|
||
# 引入整个子树
|
||
new_root_nodes.append(remote_root)
|
||
logger.info(f"添加远端独立物料根节点子树: '{remote_root_id}'")
|
||
|
||
# 将新的根节点添加到本地树集合
|
||
if new_root_nodes:
|
||
for new_root in new_root_nodes:
|
||
self.trees.append(ResourceTreeInstance(new_root))
|
||
|
||
return self
|
||
|
||
def dump(self) -> List[List[Dict[str, Any]]]:
|
||
"""
|
||
将 ResourceTreeSet 序列化为嵌套列表格式
|
||
|
||
序列化时：
|
||
- parent 自动转换为 parent_uuid（在 ResourceDict.model_dump 中处理）
|
||
- children 不会被序列化（exclude=True）
|
||
|
||
Returns:
|
||
List[List[Dict]]: 每个内层列表代表一棵树的扁平化资源字典列表
|
||
"""
|
||
result = []
|
||
for tree in self.trees:
|
||
# 获取树的所有节点并序列化
|
||
tree_nodes = [node.res_content.model_dump(by_alias=True) for node in tree.get_all_nodes()]
|
||
result.append(tree_nodes)
|
||
return result
|
||
|
||
@classmethod
|
||
def load(cls, data: List[List[Dict[str, Any]]]) -> "ResourceTreeSet":
|
||
"""
|
||
从序列化的嵌套列表格式反序列化为 ResourceTreeSet
|
||
|
||
Args:
|
||
data: List[List[Dict]]: 序列化的数据，每个内层列表代表一棵树
|
||
|
||
Returns:
|
||
ResourceTreeSet: 反序列化后的资源树集合
|
||
"""
|
||
nested_lists = []
|
||
for tree_data in data:
|
||
nested_lists.extend(ResourceTreeSet.from_raw_list(tree_data).trees)
|
||
return cls(nested_lists)
|
||
|
||
|
||
class DeviceNodeResourceTracker(object):
|
||
|
||
def __init__(self):
|
||
self.resources = []
|
||
self.resource2parent_resource = {}
|
||
self.uuid_to_resources = {}
|
||
pass
|
||
|
||
def prefix_path(self, resource):
|
||
resource_prefix_path = "/"
|
||
resource_parent = getattr(resource, "parent", None)
|
||
while resource_parent is not None:
|
||
resource_prefix_path = f"/{resource_parent.name}" + resource_prefix_path
|
||
resource_parent = resource_parent.parent
|
||
|
||
return resource_prefix_path
|
||
|
||
def map_uuid_to_resource(self, resource, uuid_map: Dict[str, str]):
|
||
for old_uuid, new_uuid in uuid_map.items():
|
||
if old_uuid != new_uuid:
|
||
if old_uuid in self.uuid_to_resources:
|
||
instance = self.uuid_to_resources.pop(old_uuid)
|
||
if isinstance(resource, dict):
|
||
resource["uuid"] = new_uuid
|
||
else: # 实例的
|
||
setattr(instance, "unilabos_uuid", new_uuid)
|
||
self.uuid_to_resources[new_uuid] = instance
|
||
print(f"更新uuid映射: {old_uuid} -> {new_uuid} | {instance}")
|
||
|
||
def _get_resource_attr(self, resource, attr_name: str, uuid_attr: Optional[str] = None):
|
||
"""
|
||
获取资源的属性值，统一处理 dict 和 instance 两种类型
|
||
|
||
Args:
|
||
resource: 资源对象（dict或实例）
|
||
attr_name: dict类型使用的属性名
|
||
uuid_attr: instance类型使用的属性名（用于uuid字段），默认与attr_name相同
|
||
|
||
Returns:
|
||
属性值，不存在则返回None
|
||
"""
|
||
if uuid_attr is None:
|
||
uuid_attr = attr_name
|
||
|
||
if isinstance(resource, dict):
|
||
return resource.get(attr_name)
|
||
else:
|
||
return getattr(resource, uuid_attr, None)
|
||
|
||
@classmethod
|
||
def set_resource_uuid(cls, resource, new_uuid: str):
|
||
"""
|
||
设置资源的 uuid，统一处理 dict 和 instance 两种类型
|
||
|
||
Args:
|
||
resource: 资源对象（dict或实例）
|
||
new_uuid: 新的uuid值
|
||
"""
|
||
if isinstance(resource, dict):
|
||
resource["uuid"] = new_uuid
|
||
else:
|
||
setattr(resource, "unilabos_uuid", new_uuid)
|
||
|
||
def _traverse_and_process(self, resource, process_func) -> int:
|
||
"""
|
||
递归遍历资源树，对每个节点执行处理函数
|
||
|
||
Args:
|
||
resource: 资源对象（可以是list、dict或实例）
|
||
process_func: 处理函数，接收resource参数，返回处理的节点数量
|
||
|
||
Returns:
|
||
处理的节点总数量
|
||
"""
|
||
if isinstance(resource, list):
|
||
return sum(self._traverse_and_process(r, process_func) for r in resource)
|
||
|
||
# 先递归处理所有子节点
|
||
count = 0
|
||
children = getattr(resource, "children", [])
|
||
for child in children:
|
||
count += self._traverse_and_process(child, process_func)
|
||
|
||
# 处理当前节点
|
||
count += process_func(resource)
|
||
return count
|
||
|
||
def loop_set_uuid(self, resource, name_to_uuid_map: Dict[str, str]) -> int:
|
||
"""
|
||
递归遍历资源树，根据 name 设置所有节点的 uuid
|
||
|
||
Args:
|
||
resource: 资源对象（可以是dict或实例）
|
||
name_to_uuid_map: name到uuid的映射字典，{name: uuid}
|
||
|
||
Returns:
|
||
更新的资源数量
|
||
"""
|
||
|
||
def process(res):
|
||
resource_name = self._get_resource_attr(res, "name")
|
||
if resource_name and resource_name in name_to_uuid_map:
|
||
new_uuid = name_to_uuid_map[resource_name]
|
||
self.set_resource_uuid(res, new_uuid)
|
||
self.uuid_to_resources[new_uuid] = res
|
||
logger.debug(f"设置资源UUID: {resource_name} -> {new_uuid}")
|
||
return 1
|
||
return 0
|
||
|
||
return self._traverse_and_process(resource, process)
|
||
|
||
def loop_update_uuid(self, resource, uuid_map: Dict[str, str]) -> int:
|
||
"""
|
||
递归遍历资源树，更新所有节点的uuid
|
||
|
||
Args:0
|
||
resource: 资源对象（可以是dict或实例）
|
||
uuid_map: uuid映射字典，{old_uuid: new_uuid}
|
||
|
||
Returns:
|
||
更新的资源数量
|
||
"""
|
||
|
||
def process(res):
|
||
current_uuid = self._get_resource_attr(res, "uuid", "unilabos_uuid")
|
||
replaced = 0
|
||
if current_uuid and current_uuid in uuid_map:
|
||
new_uuid = uuid_map[current_uuid]
|
||
if current_uuid != new_uuid:
|
||
self.set_resource_uuid(res, new_uuid)
|
||
# 更新uuid_to_resources映射
|
||
if current_uuid in self.uuid_to_resources:
|
||
self.uuid_to_resources.pop(current_uuid)
|
||
self.uuid_to_resources[new_uuid] = res
|
||
logger.debug(f"更新uuid: {current_uuid} -> {new_uuid}")
|
||
replaced = 1
|
||
return replaced
|
||
|
||
return self._traverse_and_process(resource, process)
|
||
|
||
def _collect_uuid_mapping(self, resource):
|
||
"""
|
||
递归收集资源的 uuid 映射到 uuid_to_resources
|
||
|
||
Args:
|
||
resource: 资源对象（可以是dict或实例）
|
||
"""
|
||
|
||
def process(res):
|
||
current_uuid = self._get_resource_attr(res, "uuid", "unilabos_uuid")
|
||
if current_uuid:
|
||
old = self.uuid_to_resources.get(current_uuid)
|
||
self.uuid_to_resources[current_uuid] = res
|
||
logger.debug(f"收集资源UUID映射: {current_uuid} -> {res} {'' if old is None else f'(覆盖旧值: {old})'}")
|
||
return 0
|
||
|
||
self._traverse_and_process(resource, process)
|
||
|
||
def _remove_uuid_mapping(self, resource):
|
||
"""
|
||
递归清除资源的 uuid 映射
|
||
|
||
Args:
|
||
resource: 资源对象（可以是dict或实例）
|
||
"""
|
||
|
||
def process(res):
|
||
current_uuid = self._get_resource_attr(res, "uuid", "unilabos_uuid")
|
||
if current_uuid and current_uuid in self.uuid_to_resources:
|
||
self.uuid_to_resources.pop(current_uuid)
|
||
logger.debug(f"移除资源UUID映射: {current_uuid} -> {res}")
|
||
return 0
|
||
|
||
self._traverse_and_process(resource, process)
|
||
|
||
def parent_resource(self, resource):
|
||
if id(resource) in self.resource2parent_resource:
|
||
return self.resource2parent_resource[id(resource)]
|
||
else:
|
||
return resource
|
||
|
||
def add_resource(self, resource):
|
||
"""
|
||
添加资源到追踪器
|
||
|
||
Args:
|
||
resource: 资源对象（可以是dict或实例）
|
||
"""
|
||
root_uuids = {}
|
||
for r in self.resources:
|
||
res_uuid = r.get("uuid") if isinstance(r, dict) else getattr(r, "unilabos_uuid", None)
|
||
if res_uuid:
|
||
root_uuids[res_uuid] = r
|
||
if id(r) == id(resource):
|
||
return
|
||
|
||
# 这里只做uuid的根节点比较
|
||
if isinstance(resource, dict):
|
||
res_uuid = resource.get("uuid")
|
||
else:
|
||
res_uuid = getattr(resource, "unilabos_uuid", None)
|
||
if res_uuid in root_uuids:
|
||
old_res = root_uuids[res_uuid]
|
||
# self.remove_resource(old_res)
|
||
logger.warning(f"资源{resource}已存在，旧资源: {old_res}")
|
||
self.resources.append(resource)
|
||
# 递归收集uuid映射
|
||
self._collect_uuid_mapping(resource)
|
||
|
||
def remove_resource(self, resource) -> bool:
|
||
"""
|
||
从追踪器中移除资源
|
||
|
||
Args:
|
||
resource: 资源对象（可以是dict或实例）
|
||
|
||
Returns:
|
||
bool: 如果成功移除返回True，资源不存在返回False
|
||
"""
|
||
# 从 resources 列表中移除
|
||
resource_id = id(resource)
|
||
removed = False
|
||
for i, r in enumerate(self.resources):
|
||
if id(r) == resource_id:
|
||
self.resources.pop(i)
|
||
removed = True
|
||
break
|
||
|
||
if not removed:
|
||
logger.warning(f"尝试移除不存在的资源: {resource}")
|
||
return False
|
||
|
||
# 递归清除uuid映射
|
||
self._remove_uuid_mapping(resource)
|
||
|
||
# 清除 resource2parent_resource 中与该资源相关的映射
|
||
# 需要清除：1) 该资源作为 key 的映射 2) 该资源作为 value 的映射
|
||
keys_to_remove = []
|
||
for key, value in self.resource2parent_resource.items():
|
||
if id(value) == resource_id:
|
||
keys_to_remove.append(key)
|
||
|
||
if resource_id in self.resource2parent_resource:
|
||
keys_to_remove.append(resource_id)
|
||
|
||
for key in keys_to_remove:
|
||
self.resource2parent_resource.pop(key, None)
|
||
|
||
logger.debug(f"成功移除资源: {resource}")
|
||
return True
|
||
|
||
def clear_resource(self):
|
||
"""清空所有资源"""
|
||
self.resources = []
|
||
self.uuid_to_resources.clear()
|
||
self.resource2parent_resource.clear()
|
||
|
||
def figure_resource(self, query_resource, try_mode=False):
|
||
if isinstance(query_resource, list):
|
||
return [self.figure_resource(r, try_mode) for r in query_resource]
|
||
elif (
|
||
isinstance(query_resource, dict)
|
||
and "id" not in query_resource
|
||
and "name" not in query_resource
|
||
and "uuid" not in query_resource
|
||
): # 临时处理，要删除的，driver有太多类型错误标注
|
||
return [self.figure_resource(r, try_mode) for r in query_resource.values()]
|
||
|
||
# 优先尝试通过 uuid 查找
|
||
res_uuid = None
|
||
if isinstance(query_resource, dict):
|
||
res_uuid = query_resource.get("uuid")
|
||
else:
|
||
res_uuid = getattr(query_resource, "unilabos_uuid", None)
|
||
|
||
# 如果有 uuid，优先使用 uuid 查找
|
||
if res_uuid:
|
||
res_list = []
|
||
for r in self.resources:
|
||
if isinstance(query_resource, dict):
|
||
res_list.extend(self.loop_find_resource(r, object, "uuid", res_uuid))
|
||
else:
|
||
res_list.extend(self.loop_find_resource(r, type(query_resource), "unilabos_uuid", res_uuid))
|
||
|
||
if not try_mode:
|
||
assert len(res_list) > 0, f"没有找到资源 (uuid={res_uuid})，请检查资源是否存在"
|
||
assert len(res_list) == 1, f"通过uuid={res_uuid} 找到多个资源，请检查资源是否唯一: {res_list}"
|
||
else:
|
||
return [i[1] for i in res_list]
|
||
|
||
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][1]
|
||
|
||
# 回退到 id/name 查找
|
||
res_id = (
|
||
query_resource.id # type: ignore
|
||
if hasattr(query_resource, "id")
|
||
else (query_resource.get("id") if isinstance(query_resource, dict) else None)
|
||
)
|
||
res_name = (
|
||
query_resource.name # type: ignore
|
||
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(query_resource)
|
||
if res_identifier is None:
|
||
logger.warning(f"resource {query_resource} 没有id、name或uuid，暂不能对应figure")
|
||
res_list = []
|
||
for r in self.resources:
|
||
if isinstance(query_resource, dict):
|
||
res_list.extend(self.loop_find_resource(r, object, 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)
|
||
)
|
||
)
|
||
if not try_mode:
|
||
assert len(res_list) > 0, f"没有找到资源 {query_resource}，请检查资源是否存在"
|
||
assert len(res_list) == 1, f"{query_resource} 找到多个资源，请检查资源是否唯一: {res_list}"
|
||
else:
|
||
return [i[1] for i in res_list]
|
||
# 后续加入其他对比方式
|
||
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][1]
|
||
|
||
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 = []
|
||
if not isinstance(resource, dict):
|
||
children = getattr(resource, "children", [])
|
||
else:
|
||
children = resource.get("children")
|
||
if children is not None:
|
||
children = list(children.values()) if isinstance(children, dict) else children
|
||
for child in children:
|
||
res_list.extend(
|
||
self.loop_find_resource(child, target_resource_cls_type, identifier_key, compare_value, resource)
|
||
)
|
||
if issubclass(type(resource), target_resource_cls_type):
|
||
if type(resource) == dict:
|
||
# 对于字典类型，直接检查 identifier_key
|
||
if identifier_key in resource:
|
||
if resource[identifier_key] == compare_value:
|
||
res_list.append((parent_res, resource))
|
||
else:
|
||
# 对于实例类型，需要特殊处理 uuid 字段
|
||
# 如果查找的是 unilabos_uuid，使用 getattr
|
||
if identifier_key == "uuid":
|
||
identifier_key = "unilabos_uuid"
|
||
if hasattr(resource, identifier_key):
|
||
if getattr(resource, identifier_key) == compare_value:
|
||
res_list.append((parent_res, resource))
|
||
return res_list
|
||
|
||
def filter_find_list(self, res_list, compare_std_dict):
|
||
new_list = []
|
||
for res in res_list:
|
||
for k, v in compare_std_dict.items():
|
||
if hasattr(res, k):
|
||
if getattr(res, k) == v:
|
||
new_list.append(res)
|
||
return new_list
|
||
|
||
|
||
if __name__ == "__main__":
|
||
from pylabrobot.resources import corning_6_wellplate_16point8ml_flat
|
||
|
||
# 测试 from_plr_resources 和 to_plr_resources 的往返转换
|
||
print("=" * 60)
|
||
print("测试 PLR 资源转换往返")
|
||
print("=" * 60)
|
||
|
||
# 1. 创建一个 PLR 资源并设置 UUID
|
||
original_plate = corning_6_wellplate_16point8ml_flat("test_plate")
|
||
|
||
# 使用 DeviceNodeResourceTracker 设置 UUID
|
||
tracker = DeviceNodeResourceTracker()
|
||
name_to_uuid = {}
|
||
|
||
# 递归生成 name_to_uuid 映射
|
||
def build_uuid_map(resource):
|
||
name_to_uuid[resource.name] = str(uuid.uuid4())
|
||
for child in resource.children:
|
||
build_uuid_map(child)
|
||
|
||
build_uuid_map(original_plate)
|
||
|
||
# 使用 tracker 的 loop_set_uuid 方法设置 UUID
|
||
tracker.loop_set_uuid(original_plate, name_to_uuid)
|
||
|
||
print(f"\n1. 原始 PLR 资源: {original_plate.name}")
|
||
print(f" - UUID: {getattr(original_plate, 'unilabos_uuid', 'N/A')}")
|
||
print(f" - 子节点数量: {len(original_plate.children)}")
|
||
if original_plate.children:
|
||
print(f" - 第一个子节点: {original_plate.children[0].name}")
|
||
print(f" - 第一个子节点 UUID: {getattr(original_plate.children[0], 'unilabos_uuid', 'N/A')}")
|
||
|
||
# 2. 将 PLR 资源转换为 ResourceTreeSet
|
||
resource_tree_set = ResourceTreeSet.from_plr_resources([original_plate])
|
||
print(f"\n2. 转换为 ResourceTreeSet:")
|
||
print(f" - 树的数量: {len(resource_tree_set.trees)}")
|
||
print(f" - 根节点: {resource_tree_set.root_nodes[0].res_content.name}")
|
||
print(f" - 所有节点数量: {len(resource_tree_set.all_nodes)}")
|
||
|
||
# 3. 将 ResourceTreeSet 转换回 PLR 资源
|
||
plr_resources = resource_tree_set.to_plr_resources()
|
||
converted_plate = plr_resources[0]
|
||
print(f"\n3. 转换回 PLR 资源: {converted_plate.name}")
|
||
print(f" - 子节点数量: {len(converted_plate.children)}")
|
||
if converted_plate.children:
|
||
print(f" - 第一个子节点: {converted_plate.children[0].name}")
|
||
|
||
# 4. 验证 unilabos_uuid 属性
|
||
print(f"\n4. 验证 unilabos_uuid 设置:")
|
||
if hasattr(converted_plate, "unilabos_uuid"):
|
||
print(f" - 根节点 UUID: {getattr(converted_plate, 'unilabos_uuid')}")
|
||
if converted_plate.children and hasattr(converted_plate.children[0], "unilabos_uuid"):
|
||
print(f" - 第一个子节点 UUID: {getattr(converted_plate.children[0], 'unilabos_uuid')}")
|
||
else:
|
||
print(" - 警告: unilabos_uuid 未设置")
|
||
|
||
# 5. 验证 UUID 保持不变
|
||
print(f"\n5. 验证 UUID 在往返过程中保持不变:")
|
||
original_uuid = getattr(original_plate, "unilabos_uuid")
|
||
converted_uuid = getattr(converted_plate, "unilabos_uuid")
|
||
print(f" - 原始 UUID: {original_uuid}")
|
||
print(f" - 转换后 UUID: {converted_uuid}")
|
||
print(f" - UUID 保持不变: {original_uuid == converted_uuid}")
|
||
|
||
# 6. 再次往返转换，验证稳定性
|
||
resource_tree_set_2 = ResourceTreeSet.from_plr_resources([converted_plate])
|
||
plr_resources_2 = resource_tree_set_2.to_plr_resources()
|
||
print(f"\n6. 第二次往返转换:")
|
||
print(f" - 资源名称: {plr_resources_2[0].name}")
|
||
print(f" - 子节点数量: {len(plr_resources_2[0].children)}")
|
||
print(f" - UUID 依然保持: {getattr(plr_resources_2[0], 'unilabos_uuid') == original_uuid}")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("✅ 测试完成! 所有转换正常工作")
|
||
print("=" * 60)
|