mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2026-02-04 13:25:13 +00:00
471 lines
23 KiB
Python
471 lines
23 KiB
Python
import copy
|
||
import io
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List
|
||
|
||
import yaml
|
||
|
||
from unilabos.ros.msgs.message_converter import msg_converter_manager, ros_action_to_json_schema
|
||
from unilabos.utils import logger
|
||
from unilabos.utils.decorator import singleton
|
||
from unilabos.utils.import_manager import get_enhanced_class_info
|
||
from unilabos.utils.type_check import NoAliasDumper
|
||
|
||
DEFAULT_PATHS = [Path(__file__).absolute().parent]
|
||
|
||
|
||
@singleton
|
||
class Registry:
|
||
def __init__(self, registry_paths=None):
|
||
self.registry_paths = DEFAULT_PATHS.copy() # 使用copy避免修改默认值
|
||
if registry_paths:
|
||
self.registry_paths.extend(registry_paths)
|
||
self.ResourceCreateFromOuter = self._replace_type_with_class(
|
||
"ResourceCreateFromOuter", "host_node", f"动作 create_resource_detailed"
|
||
)
|
||
self.ResourceCreateFromOuterEasy = self._replace_type_with_class(
|
||
"ResourceCreateFromOuterEasy", "host_node", f"动作 create_resource"
|
||
)
|
||
self.EmptyIn = self._replace_type_with_class("EmptyIn", "host_node", f"")
|
||
self.device_type_registry = {}
|
||
self.resource_type_registry = {}
|
||
self._setup_called = False # 跟踪setup是否已调用
|
||
# 其他状态变量
|
||
# self.is_host_mode = False # 移至BasicConfig中
|
||
|
||
def setup(self, complete_registry=False):
|
||
# 检查是否已调用过setup
|
||
if self._setup_called:
|
||
logger.critical("[UniLab Registry] setup方法已被调用过，不允许多次调用")
|
||
return
|
||
|
||
from unilabos.app.web.utils.action_utils import get_yaml_from_goal_type
|
||
|
||
self.device_type_registry.update(
|
||
{
|
||
"host_node": {
|
||
"description": "UniLabOS主机节点",
|
||
"class": {
|
||
"module": "unilabos.ros.nodes.presets.host_node",
|
||
"type": "python",
|
||
"status_types": {},
|
||
"action_value_mappings": {
|
||
"create_resource_detailed": {
|
||
"type": self.ResourceCreateFromOuter,
|
||
"goal": {
|
||
"resources": "resources",
|
||
"device_ids": "device_ids",
|
||
"bind_parent_ids": "bind_parent_ids",
|
||
"bind_locations": "bind_locations",
|
||
"other_calling_params": "other_calling_params",
|
||
},
|
||
"feedback": {},
|
||
"result": {"success": "success"},
|
||
"schema": ros_action_to_json_schema(self.ResourceCreateFromOuter),
|
||
"goal_default": yaml.safe_load(
|
||
io.StringIO(get_yaml_from_goal_type(self.ResourceCreateFromOuter.Goal))
|
||
),
|
||
"handles": {},
|
||
},
|
||
"create_resource": {
|
||
"type": self.ResourceCreateFromOuterEasy,
|
||
"goal": {
|
||
"res_id": "res_id",
|
||
"class_name": "class_name",
|
||
"parent": "parent",
|
||
"device_id": "device_id",
|
||
"bind_locations": "bind_locations",
|
||
"liquid_input_slot": "liquid_input_slot[]",
|
||
"liquid_type": "liquid_type[]",
|
||
"liquid_volume": "liquid_volume[]",
|
||
"slot_on_deck": "slot_on_deck",
|
||
},
|
||
"feedback": {},
|
||
"result": {"success": "success"},
|
||
"schema": ros_action_to_json_schema(self.ResourceCreateFromOuterEasy),
|
||
"goal_default": yaml.safe_load(
|
||
io.StringIO(get_yaml_from_goal_type(self.ResourceCreateFromOuterEasy.Goal))
|
||
),
|
||
"handles": {
|
||
"output": [
|
||
{
|
||
"handler_key": "labware",
|
||
"label": "Labware",
|
||
"data_type": "resource",
|
||
"data_source": "handle",
|
||
"data_key": "liquid",
|
||
}
|
||
]
|
||
},
|
||
},
|
||
"test_latency": {
|
||
"type": self.EmptyIn,
|
||
"goal": {},
|
||
"feedback": {},
|
||
"result": {"latency_ms": "latency_ms", "time_diff_ms": "time_diff_ms"},
|
||
"schema": ros_action_to_json_schema(self.EmptyIn),
|
||
"goal_default": {},
|
||
"handles": {},
|
||
},
|
||
},
|
||
},
|
||
"icon": "icon_device.webp",
|
||
"registry_type": "device",
|
||
"handles": [],
|
||
"init_param_schema": {},
|
||
"file_path": "/",
|
||
}
|
||
}
|
||
)
|
||
logger.debug(f"[UniLab Registry] ----------Setup----------")
|
||
self.registry_paths = [Path(path).absolute() for path in self.registry_paths]
|
||
for i, path in enumerate(self.registry_paths):
|
||
sys_path = path.parent
|
||
logger.debug(f"[UniLab Registry] Path {i+1}/{len(self.registry_paths)}: {sys_path}")
|
||
sys.path.append(str(sys_path))
|
||
self.load_device_types(path, complete_registry)
|
||
self.load_resource_types(path, complete_registry)
|
||
logger.info("[UniLab Registry] 注册表设置完成")
|
||
# 标记setup已被调用
|
||
self._setup_called = True
|
||
|
||
def load_resource_types(self, path: os.PathLike, complete_registry: bool):
|
||
abs_path = Path(path).absolute()
|
||
resource_path = abs_path / "resources"
|
||
files = list(resource_path.glob("*/*.yaml"))
|
||
logger.debug(f"[UniLab Registry] resources: {resource_path.exists()}, total: {len(files)}")
|
||
current_resource_number = len(self.resource_type_registry) + 1
|
||
for i, file in enumerate(files):
|
||
data = yaml.safe_load(open(file, encoding="utf-8"))
|
||
if data:
|
||
# 为每个资源添加文件路径信息
|
||
for resource_id, resource_info in data.items():
|
||
resource_info["file_path"] = str(file.absolute()).replace("\\", "/")
|
||
if "description" not in resource_info:
|
||
resource_info["description"] = ""
|
||
if "icon" not in resource_info:
|
||
resource_info["icon"] = ""
|
||
if "handles" not in resource_info:
|
||
resource_info["handles"] = []
|
||
if "init_param_schema" not in resource_info:
|
||
resource_info["init_param_schema"] = {}
|
||
resource_info["registry_type"] = "resource"
|
||
self.resource_type_registry.update(data)
|
||
logger.debug(
|
||
f"[UniLab Registry] Resource-{current_resource_number} File-{i+1}/{len(files)} "
|
||
+ f"Add {list(data.keys())}"
|
||
)
|
||
current_resource_number += 1
|
||
else:
|
||
logger.debug(f"[UniLab Registry] Res File-{i+1}/{len(files)} Not Valid YAML File: {file.absolute()}")
|
||
|
||
def _replace_type_with_class(self, type_name: str, device_id: str, field_name: str) -> Any:
|
||
"""
|
||
将类型名称替换为实际的类对象
|
||
|
||
Args:
|
||
type_name: 类型名称
|
||
device_id: 设备ID，用于错误信息
|
||
field_name: 字段名称，用于错误信息
|
||
|
||
Returns:
|
||
找到的类对象或原始字符串
|
||
|
||
Raises:
|
||
SystemExit: 如果找不到类型则终止程序
|
||
"""
|
||
# 如果类型名为空，跳过替换
|
||
if not type_name or type_name == "":
|
||
logger.warning(f"[UniLab Registry] 设备 {device_id} 的 {field_name} 类型为空，跳过替换")
|
||
return type_name
|
||
if ":" in type_name:
|
||
type_class = msg_converter_manager.get_class(type_name)
|
||
else:
|
||
type_class = msg_converter_manager.search_class(type_name)
|
||
if type_class:
|
||
return type_class
|
||
else:
|
||
logger.error(f"[UniLab Registry] 无法找到类型 '{type_name}' 用于设备 {device_id} 的 {field_name}")
|
||
sys.exit(1)
|
||
|
||
def _generate_unilab_json_command_schema(
|
||
self, method_args: List[Dict[str, Any]], method_name: str
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
根据UniLabJsonCommand方法信息生成JSON Schema，暂不支持嵌套类型
|
||
|
||
Args:
|
||
method_args: 方法信息字典，包含args等
|
||
method_name: 方法名称
|
||
|
||
Returns:
|
||
JSON Schema格式的参数schema
|
||
"""
|
||
schema = {
|
||
"description": f"UniLabJsonCommand {method_name} 的参数schema",
|
||
"type": "object",
|
||
"properties": {},
|
||
"required": [],
|
||
}
|
||
|
||
for arg_info in method_args:
|
||
param_name = arg_info.get("name", "")
|
||
param_type = arg_info.get("type")
|
||
param_default = arg_info.get("default")
|
||
param_required = arg_info.get("required", True)
|
||
|
||
prop_schema = {"description": f"参数: {param_name}"}
|
||
|
||
# 根据类型设置schema FIXME 不完整
|
||
if param_type:
|
||
param_type_lower = param_type.lower()
|
||
if param_type_lower in ["str", "string"]:
|
||
prop_schema["type"] = "string"
|
||
elif param_type_lower in ["int", "integer"]:
|
||
prop_schema["type"] = "integer"
|
||
elif param_type_lower in ["float", "number"]:
|
||
prop_schema["type"] = "number"
|
||
elif param_type_lower in ["bool", "boolean"]:
|
||
prop_schema["type"] = "boolean"
|
||
elif param_type_lower in ["list", "array"]:
|
||
prop_schema["type"] = "array"
|
||
elif param_type_lower in ["dict", "object"]:
|
||
prop_schema["type"] = "object"
|
||
else:
|
||
# 默认为字符串类型
|
||
prop_schema["type"] = "string"
|
||
else:
|
||
# 如果没有类型信息，默认为字符串
|
||
prop_schema["type"] = "string"
|
||
|
||
# 设置默认值
|
||
if param_default is not None:
|
||
prop_schema["default"] = param_default
|
||
|
||
schema["properties"][param_name] = prop_schema
|
||
|
||
# 如果是必需参数，添加到required列表
|
||
if param_required:
|
||
schema["required"].append(param_name)
|
||
return {
|
||
"title": f"{method_name} 命令参数",
|
||
"description": f"UniLabJsonCommand {method_name} 的参数schema",
|
||
"type": "object",
|
||
"properties": {"goal": schema, "feedback": {}, "result": {}},
|
||
"required": ["goal"],
|
||
}
|
||
|
||
def load_device_types(self, path: os.PathLike, complete_registry: bool):
|
||
abs_path = Path(path).absolute()
|
||
devices_path = abs_path / "devices"
|
||
device_comms_path = abs_path / "device_comms"
|
||
files = list(devices_path.glob("*.yaml")) + list(device_comms_path.glob("*.yaml"))
|
||
logger.debug(
|
||
f"[UniLab Registry] devices: {devices_path.exists()}, device_comms: {device_comms_path.exists()}, "
|
||
+ f"total: {len(files)}"
|
||
)
|
||
current_device_number = len(self.device_type_registry) + 1
|
||
from unilabos.app.web.utils.action_utils import get_yaml_from_goal_type
|
||
|
||
for i, file in enumerate(files):
|
||
with open(file, encoding="utf-8", mode="r") as f:
|
||
data = yaml.safe_load(io.StringIO(f.read()))
|
||
complete_data = {}
|
||
action_str_type_mapping = {
|
||
"UniLabJsonCommand": "UniLabJsonCommand",
|
||
"UniLabJsonCommandAsync": "UniLabJsonCommandAsync",
|
||
}
|
||
status_str_type_mapping = {}
|
||
if data:
|
||
# 在添加到注册表前处理类型替换
|
||
for device_id, device_config in data.items():
|
||
# 添加文件路径信息 - 使用规范化的完整文件路径
|
||
if "description" not in device_config:
|
||
device_config["description"] = ""
|
||
if "icon" not in device_config:
|
||
device_config["icon"] = ""
|
||
if "handles" not in device_config:
|
||
device_config["handles"] = []
|
||
if "init_param_schema" not in device_config:
|
||
device_config["init_param_schema"] = {}
|
||
if "class" in device_config:
|
||
if "status_types" not in device_config["class"]:
|
||
device_config["class"]["status_types"] = {}
|
||
if "action_value_mappings" not in device_config["class"]:
|
||
device_config["class"]["action_value_mappings"] = {}
|
||
enhanced_info = {}
|
||
if complete_registry:
|
||
enhanced_info = get_enhanced_class_info(device_config["class"]["module"], use_dynamic=True)
|
||
device_config["class"]["status_types"].update(
|
||
{k: v["return_type"] for k, v in enhanced_info["status_methods"].items()}
|
||
)
|
||
for status_name, status_type in device_config["class"]["status_types"].items():
|
||
if status_type in ["Any", "None"]:
|
||
status_type = "String" # 替换成ROS的String，便于显示
|
||
device_config["class"]["status_types"][status_name] = status_type
|
||
target_type = self._replace_type_with_class(status_type, device_id, f"状态 {status_name}")
|
||
status_str_type_mapping[status_type] = target_type
|
||
device_config["class"]["status_types"] = dict(
|
||
sorted(device_config["class"]["status_types"].items())
|
||
)
|
||
if complete_registry:
|
||
# 处理动作值映射
|
||
device_config["class"]["action_value_mappings"].update(
|
||
{
|
||
f"auto-{k}": {
|
||
"type": "UniLabJsonCommandAsync" if v["is_async"] else "UniLabJsonCommand",
|
||
"goal": {},
|
||
"feedback": {},
|
||
"result": {},
|
||
"schema": self._generate_unilab_json_command_schema(v["args"], k),
|
||
"goal_default": {i["name"]: i["default"] for i in v["args"]},
|
||
"handles": [],
|
||
}
|
||
for k, v in enhanced_info["action_methods"].items()
|
||
}
|
||
)
|
||
device_config["init_param_schema"] = self._generate_unilab_json_command_schema(
|
||
enhanced_info["init_params"], "__init__"
|
||
)
|
||
device_config.pop("schema", None)
|
||
device_config["class"]["action_value_mappings"] = dict(
|
||
sorted(device_config["class"]["action_value_mappings"].items())
|
||
)
|
||
for action_name, action_config in device_config["class"]["action_value_mappings"].items():
|
||
if "handles" not in action_config:
|
||
action_config["handles"] = []
|
||
if "type" in action_config:
|
||
action_type_str: str = action_config["type"]
|
||
# 通过Json发放指令，而不是通过特殊的ros action进行处理
|
||
if not action_type_str.startswith("UniLabJsonCommand"):
|
||
target_type = self._replace_type_with_class(
|
||
action_type_str, device_id, f"动作 {action_name}"
|
||
)
|
||
action_str_type_mapping[action_type_str] = target_type
|
||
if target_type is not None:
|
||
action_config["goal_default"] = yaml.safe_load(
|
||
io.StringIO(get_yaml_from_goal_type(target_type.Goal))
|
||
)
|
||
action_config["schema"] = ros_action_to_json_schema(target_type)
|
||
else:
|
||
logger.warning(
|
||
f"[UniLab Registry] 设备 {device_id} 的动作 {action_name} 类型为空，跳过替换"
|
||
)
|
||
complete_data[device_id] = copy.deepcopy(dict(sorted(device_config.items()))) # 稍后dump到文件
|
||
for status_name, status_type in device_config["class"]["status_types"].items():
|
||
device_config["class"]["status_types"][status_name] = status_str_type_mapping[status_type]
|
||
for action_name, action_config in device_config["class"]["action_value_mappings"].items():
|
||
action_config["type"] = action_str_type_mapping[action_config["type"]]
|
||
for additional_action in ["_execute_driver_command", "_execute_driver_command_async"]:
|
||
device_config["class"]["action_value_mappings"][additional_action] = {
|
||
"type": self._replace_type_with_class(
|
||
"StrSingleInput", device_id, f"动作 {additional_action}"
|
||
),
|
||
"goal": {"string": "string"},
|
||
"feedback": {},
|
||
"result": {},
|
||
"schema": ros_action_to_json_schema(
|
||
self._replace_type_with_class(
|
||
"StrSingleInput", device_id, f"动作 {additional_action}"
|
||
)
|
||
),
|
||
"goal_default": yaml.safe_load(
|
||
io.StringIO(
|
||
get_yaml_from_goal_type(
|
||
self._replace_type_with_class(
|
||
"StrSingleInput", device_id, f"动作 {additional_action}"
|
||
).Goal
|
||
)
|
||
)
|
||
),
|
||
"handles": [],
|
||
}
|
||
if "registry_type" not in device_config:
|
||
device_config["registry_type"] = "device"
|
||
device_config["file_path"] = str(file.absolute()).replace("\\", "/")
|
||
device_config["registry_type"] = "device"
|
||
logger.debug(
|
||
f"[UniLab Registry] Device-{current_device_number} File-{i+1}/{len(files)} Add {device_id} "
|
||
+ f"[{data[device_id].get('name', '未命名设备')}]"
|
||
)
|
||
current_device_number += 1
|
||
complete_data = dict(sorted(complete_data.items()))
|
||
complete_data = copy.deepcopy(complete_data)
|
||
with open(file, "w", encoding="utf-8") as f:
|
||
yaml.dump(complete_data, f, allow_unicode=True, default_flow_style=False, Dumper=NoAliasDumper)
|
||
self.device_type_registry.update(data)
|
||
else:
|
||
logger.debug(
|
||
f"[UniLab Registry] Device File-{i+1}/{len(files)} Not Valid YAML File: {file.absolute()}"
|
||
)
|
||
|
||
def obtain_registry_device_info(self):
|
||
devices = []
|
||
for device_id, device_info in self.device_type_registry.items():
|
||
device_info_copy = copy.deepcopy(device_info)
|
||
if "class" in device_info_copy and "action_value_mappings" in device_info_copy["class"]:
|
||
action_mappings = device_info_copy["class"]["action_value_mappings"]
|
||
for action_name, action_config in action_mappings.items():
|
||
if "schema" in action_config and action_config["schema"]:
|
||
schema = action_config["schema"]
|
||
# 确保schema结构存在
|
||
if (
|
||
"properties" in schema
|
||
and "goal" in schema["properties"]
|
||
and "properties" in schema["properties"]["goal"]
|
||
):
|
||
schema["properties"]["goal"]["properties"] = {
|
||
"unilabos_device_id": {
|
||
"type": "string",
|
||
"default": "",
|
||
"description": "UniLabOS设备ID，用于指定执行动作的具体设备实例",
|
||
},
|
||
**schema["properties"]["goal"]["properties"],
|
||
}
|
||
|
||
msg = {"id": device_id, **device_info_copy}
|
||
devices.append(msg)
|
||
return devices
|
||
|
||
def obtain_registry_resource_info(self):
|
||
resources = []
|
||
for resource_id, resource_info in self.resource_type_registry.items():
|
||
msg = {"id": resource_id, **resource_info}
|
||
resources.append(msg)
|
||
return resources
|
||
|
||
|
||
# 全局单例实例
|
||
lab_registry = Registry()
|
||
|
||
|
||
def build_registry(registry_paths=None, complete_registry=False):
|
||
"""
|
||
构建或获取Registry单例实例
|
||
|
||
Args:
|
||
registry_paths: 额外的注册表路径列表
|
||
|
||
Returns:
|
||
Registry实例
|
||
"""
|
||
logger.info("[UniLab Registry] 构建注册表实例")
|
||
|
||
# 由于使用了单例，这里不需要重新创建实例
|
||
global lab_registry
|
||
|
||
# 如果有额外路径，添加到registry_paths
|
||
if registry_paths:
|
||
current_paths = lab_registry.registry_paths.copy()
|
||
# 检查是否有新路径需要添加
|
||
for path in registry_paths:
|
||
if path not in current_paths:
|
||
lab_registry.registry_paths.append(path)
|
||
|
||
# 初始化注册表
|
||
lab_registry.setup(complete_registry)
|
||
|
||
return lab_registry
|