Uni-Lab-OS/unilabos/registry/registry.py

import copy
import io
import os
import sys
import inspect
import importlib
from pathlib import Path
from typing import Any, Dict, List, Union, Tuple

import yaml
from unilabos_msgs.msg import Resource

from unilabos.config.config import BasicConfig
from unilabos.resources.graphio import resource_plr_to_ulab, tree_to_list
from unilabos.ros.msgs.message_converter import (
    msg_converter_manager,
    ros_action_to_json_schema,
    String,
    ros_message_to_json_schema,
)
from unilabos.utils import logger
from unilabos.utils.decorator import singleton
from unilabos.utils.import_manager import get_enhanced_class_info, get_class
from unilabos.utils.type_check import NoAliasDumper

DEFAULT_PATHS = [Path(__file__).absolute().parent]


class ROSMsgNotFound(Exception):
    pass


@singleton
class Registry:
    def __init__(self, registry_paths=None):
        import ctypes

        try:
            import unilabos_msgs
        except ImportError:
            logger.error("[UniLab Registry] unilabos_msgs模块未找到，请确保已根据官方文档安装unilabos_msgs包。")
            sys.exit(1)
        try:
            ctypes.CDLL(str(Path(unilabos_msgs.__file__).parent / "unilabos_msgs_s__rosidl_typesupport_c.pyd"))
        except OSError as e:
            pass

        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.device_module_to_registry = {}
        self.resource_type_registry = {}
        self._setup_called = False  # 跟踪setup是否已调用
        # 其他状态变量
        # self.is_host_mode = False  # 移至BasicConfig中

    def setup(self, complete_registry=False, upload_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",
                                        }
                                    ]
                                },
                                # todo: support nested keys, switch to non ros message schema
                                "placeholder_keys": {
                                    "res_id": "unilabos_resources",  # 将当前实验室的全部物料id作为下拉框可选择
                                    "device_id": "unilabos_devices",  # 将当前实验室的全部设备id作为下拉框可选择
                                    "parent": "unilabos_nodes",  # 将当前实验室的设备/物料作为下拉框可选择
                                },
                            },
                            "test_latency": {
                                "type": self.EmptyIn,
                                "goal": {},
                                "feedback": {},
                                "result": {},
                                "schema": ros_action_to_json_schema(
                                    self.EmptyIn, "用于测试延迟的动作，返回延迟时间和时间差。"
                                ),
                                "goal_default": {},
                                "handles": {},
                            },
                            "auto-test_resource": {
                                "type": "UniLabJsonCommand",
                                "goal": {},
                                "feedback": {},
                                "result": {},
                                "schema": {
                                    "description": "",
                                    "properties": {
                                        "feedback": {},
                                        "goal": {
                                            "properties": {
                                                "resource": ros_message_to_json_schema(Resource, "resource"),
                                                "resources": {
                                                    "items": {
                                                        "properties": ros_message_to_json_schema(
                                                            Resource, "resources"
                                                        ),
                                                        "type": "object",
                                                    },
                                                    "type": "array",
                                                },
                                                "device": {"type": "string"},
                                                "devices": {"items": {"type": "string"}, "type": "array"},
                                            },
                                            "type": "object",
                                        },
                                        "result": {},
                                    },
                                    "title": "test_resource",
                                    "type": "object",
                                },
                                "placeholder_keys": {
                                    "device": "unilabos_devices",
                                    "devices": "unilabos_devices",
                                    "resource": "unilabos_resources",
                                    "resources": "unilabos_resources",
                                },
                                "goal_default": {},
                                "handles": {},
                            },
                        },
                    },
                    "version": "1.0.0",
                    "category": [],
                    "config_info": [],
                    "icon": "icon_device.webp",
                    "registry_type": "device",
                    "handles": [],  # virtue采用了不同的handle
                    "init_param_schema": {},
                    "file_path": "/",
                }
            }
        )
        # 为host_node添加内置的驱动命令动作
        self._add_builtin_actions(self.device_type_registry["host_node"], "host_node")
        logger.trace(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.trace(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)
            if BasicConfig.enable_resource_load:
                self.load_resource_types(path, complete_registry, upload_registry)
            else:
                logger.warning("跳过了资源注册表加载！")
        logger.info("[UniLab Registry] 注册表设置完成")
        # 标记setup已被调用
        self._setup_called = True

    def load_resource_types(self, path: os.PathLike, complete_registry: bool, upload_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):
            with open(file, encoding="utf-8", mode="r") as f:
                data = yaml.safe_load(io.StringIO(f.read()))
            complete_data = {}
            if data:
                # 为每个资源添加文件路径信息
                for resource_id, resource_info in data.items():
                    if "version" not in resource_info:
                        resource_info["version"] = "1.0.0"
                    if "category" not in resource_info:
                        resource_info["category"] = [file.stem]
                    elif file.stem not in resource_info["category"]:
                        resource_info["category"].append(file.stem)
                    if "config_info" not in resource_info:
                        resource_info["config_info"] = []
                    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"] = {}
                    if "config_info" in resource_info:
                        del resource_info["config_info"]
                    if "file_path" in resource_info:
                        del resource_info["file_path"]
                    complete_data[resource_id] = copy.deepcopy(dict(sorted(resource_info.items())))
                    if upload_registry:
                        class_info = resource_info.get("class", {})
                        if len(class_info) and "module" in class_info:
                            if class_info.get("type") == "pylabrobot":
                                res_class = get_class(class_info["module"])
                                if callable(res_class) and not isinstance(
                                    res_class, type
                                ):  # 有的是类，有的是函数，这里暂时只登记函数类的
                                    res_instance = res_class(res_class.__name__)
                                    res_ulr = tree_to_list([resource_plr_to_ulab(res_instance)])
                                    resource_info["config_info"] = res_ulr
                    resource_info["registry_type"] = "resource"
                    resource_info["file_path"] = str(file.absolute()).replace("\\", "/")
                complete_data = dict(sorted(complete_data.items()))
                complete_data = copy.deepcopy(complete_data)
                if complete_registry:
                    with open(file, "w", encoding="utf-8") as f:
                        yaml.dump(complete_data, f, allow_unicode=True, default_flow_style=False, Dumper=NoAliasDumper)

                self.resource_type_registry.update(data)
                logger.trace(  # type: ignore
                    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 _extract_class_docstrings(self, module_string: str) -> Dict[str, str]:
        """
        从模块字符串中提取类和方法的docstring信息

        Args:
            module_string: 模块字符串，格式为 "module.path:ClassName"

        Returns:
            包含类和方法docstring信息的字典
        """
        docstrings = {"class_docstring": "", "methods": {}}

        if not module_string or ":" not in module_string:
            return docstrings

        try:
            module_path, class_name = module_string.split(":", 1)

            # 动态导入模块
            module = importlib.import_module(module_path)

            # 获取类
            if hasattr(module, class_name):
                cls = getattr(module, class_name)

                # 获取类的docstring
                class_doc = inspect.getdoc(cls)
                if class_doc:
                    docstrings["class_docstring"] = class_doc.strip()

                # 获取所有方法的docstring
                for method_name, method in inspect.getmembers(cls, predicate=inspect.isfunction):
                    method_doc = inspect.getdoc(method)
                    if method_doc:
                        docstrings["methods"][method_name] = method_doc.strip()

                # 也获取属性方法的docstring
                for method_name, method in inspect.getmembers(cls, predicate=lambda x: isinstance(x, property)):
                    if hasattr(method, "fget") and method.fget:
                        method_doc = inspect.getdoc(method.fget)
                        if method_doc:
                            docstrings["methods"][method_name] = method_doc.strip()

        except Exception as e:
            logger.warning(f"[UniLab Registry] 无法提取docstring信息，模块: {module_string}, 错误: {str(e)}")

        return docstrings

    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
        convert_manager = {  # 将python基本对象转为ros2基本对象
            "str": "String",
            "bool": "Bool",
            "int": "Int64",
            "float": "Float64",
        }
        type_name = convert_manager.get(type_name, type_name)  # 替换为ROS2类型
        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}")
            raise ROSMsgNotFound(f"类型 '{type_name}' 未找到，用于设备 {device_id} 的 {field_name}")

    def _get_json_schema_type(self, type_str: str) -> str:
        """
        根据类型字符串返回对应的JSON Schema类型

        Args:
            type_str: 类型字符串

        Returns:
            JSON Schema类型字符串
        """
        type_lower = type_str.lower()
        type_mapping = {
            ("str", "string"): "string",
            ("int", "integer"): "integer",
            ("float", "number"): "number",
            ("bool", "boolean"): "boolean",
            ("list", "array"): "array",
            ("dict", "object"): "object",
        }

        # 遍历映射找到匹配的类型
        for type_variants, json_type in type_mapping.items():
            if type_lower in type_variants:
                return json_type

        # 特殊处理包含冒号的类型（如ROS消息类型）
        if ":" in type_lower:
            return "object"

        # 默认返回字符串类型
        return "string"

    def _generate_schema_from_info(
        self,
        param_name: str,
        param_type: Union[str, Tuple[str]],
        param_default: Any,
    ) -> Dict[str, Any]:
        """
        根据参数信息生成JSON Schema
        """
        prop_schema = {}

        # 处理嵌套类型（Tuple[str]）
        if isinstance(param_type, tuple):
            if len(param_type) == 2:
                outer_type, inner_type = param_type
                outer_json_type = self._get_json_schema_type(outer_type)
                inner_json_type = self._get_json_schema_type(inner_type)

                prop_schema["type"] = outer_json_type

                # 根据外层类型设置内层类型信息
                if outer_json_type == "array":
                    prop_schema["items"] = {"type": inner_json_type}
                elif outer_json_type == "object":
                    prop_schema["additionalProperties"] = {"type": inner_json_type}
            else:
                # 不是标准的嵌套类型，默认为字符串
                prop_schema["type"] = "string"
        else:
            # 处理非嵌套类型
            if param_type:
                prop_schema["type"] = self._get_json_schema_type(param_type)
            else:
                # 如果没有类型信息，默认为字符串
                prop_schema["type"] = "string"

        # 设置默认值
        if param_default is not None:
            prop_schema["default"] = param_default

        return prop_schema

    def _generate_status_types_schema(self, status_types: Dict[str, Any]) -> Dict[str, Any]:
        """
        根据状态类型生成JSON Schema
        """
        status_schema = {
            "type": "object",
            "properties": {},
            "required": [],
        }
        for status_name, status_type in status_types.items():
            status_schema["properties"][status_name] = self._generate_schema_from_info(
                status_name, status_type["return_type"], None
            )
            status_schema["required"].append(status_name)
        return status_schema

    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 = {
            "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)
            if param_type == "unilabos.registry.placeholder_type:ResourceSlot":
                schema["properties"][param_name] = ros_message_to_json_schema(Resource, param_name)
            elif param_type == ("list", "unilabos.registry.placeholder_type:ResourceSlot"):
                schema["properties"][param_name] = {
                    "items": ros_message_to_json_schema(Resource, param_name),
                    "type": "array",
                }
            else:
                schema["properties"][param_name] = self._generate_schema_from_info(
                    param_name, param_type, param_default
                )
            if param_required:
                schema["required"].append(param_name)

        return {
            "title": f"{method_name}参数",
            "description": f"",
            "type": "object",
            "properties": {"goal": schema, "feedback": {}, "result": {}},
            "required": ["goal"],
        }

    def _add_builtin_actions(self, device_config: Dict[str, Any], device_id: str):
        """
        为设备配置添加内置的执行驱动命令动作

        Args:
            device_config: 设备配置字典
            device_id: 设备ID
        """
        from unilabos.app.web.utils.action_utils import get_yaml_from_goal_type

        if "class" not in device_config:
            return

        if "action_value_mappings" not in device_config["class"]:
            device_config["class"]["action_value_mappings"] = {}

        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": {},
            }

    def load_device_types(self, path: os.PathLike, complete_registry: bool):
        # return
        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.trace(  # type: ignore
            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 "version" not in device_config:
                        device_config["version"] = "1.0.0"
                    if "category" not in device_config:
                        device_config["category"] = [file.stem]
                    elif file.stem not in device_config["category"]:
                        device_config["category"].append(file.stem)
                    if "config_info" not in device_config:
                        device_config["config_info"] = []
                    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:
                            device_config["class"]["status_types"].clear()
                            enhanced_info = get_enhanced_class_info(device_config["class"]["module"], use_dynamic=True)
                            if not enhanced_info.get("dynamic_import_success", False):
                                continue
                            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 isinstance(status_type, tuple) or status_type in ["Any", "None", "Unknown"]:
                                status_type = "String"  # 替换成ROS的String，便于显示
                                device_config["class"]["status_types"][status_name] = status_type
                            try:
                                target_type = self._replace_type_with_class(
                                    status_type, device_id, f"状态 {status_name}"
                                )
                            except ROSMsgNotFound:
                                continue
                            if target_type in [
                                dict,
                                list,
                            ]:  # 对于嵌套类型返回的对象，暂时处理成字符串，无法直接进行转换
                                target_type = String
                            status_str_type_mapping[status_type] = target_type
                        device_config["class"]["status_types"] = dict(
                            sorted(device_config["class"]["status_types"].items())
                        )
                        if complete_registry:
                            # 保存原有的description信息
                            old_descriptions = {}
                            for action_name, action_config in device_config["class"]["action_value_mappings"].items():
                                if "description" in action_config.get("schema", {}):
                                    description = action_config["schema"]["description"]
                                    if len(description):
                                        old_descriptions[action_name] = action_config["schema"]["description"]

                            device_config["class"]["action_value_mappings"] = {
                                k: v
                                for k, v in device_config["class"]["action_value_mappings"].items()
                                if not k.startswith("auto-")
                            }
                            # 处理动作值映射
                            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": [],
                                        "placeholder_keys": {
                                            i["name"]: (
                                                "unilabos_resources"
                                                if i["type"] == "unilabos.registry.placeholder_type:ResourceSlot"
                                                or i["type"]
                                                == ("list", "unilabos.registry.placeholder_type:ResourceSlot")
                                                else "unilabos_devices"
                                            )
                                            for i in v["args"]
                                            if i.get("type", "")
                                            in [
                                                "unilabos.registry.placeholder_type:ResourceSlot",
                                                "unilabos.registry.placeholder_type:DeviceSlot",
                                                ("list", "unilabos.registry.placeholder_type:ResourceSlot"),
                                                ("list", "unilabos.registry.placeholder_type:DeviceSlot"),
                                            ]
                                        },
                                    }
                                    # 不生成已配置action的动作
                                    for k, v in enhanced_info["action_methods"].items()
                                    if k not in device_config["class"]["action_value_mappings"]
                                }
                            )
                            # 恢复原有的description信息（auto开头的不修改）
                            for action_name, description in old_descriptions.items():
                                if action_name in device_config["class"]["action_value_mappings"]:  # 有一些会被删除
                                    device_config["class"]["action_value_mappings"][action_name]["schema"][
                                        "description"
                                    ] = description
                            device_config["init_param_schema"] = {}
                            device_config["init_param_schema"]["config"] = self._generate_unilab_json_command_schema(
                                enhanced_info["init_params"], "__init__"
                            )["properties"]["goal"]
                            device_config["init_param_schema"]["data"] = self._generate_status_types_schema(
                                enhanced_info["status_methods"]
                            )

                        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"] = {}
                            elif isinstance(action_config["handles"], list):
                                if len(action_config["handles"]):
                                    logger.error(f"设备{device_id} {action_name} 的handles配置错误，应该是字典类型")
                                    continue
                                else:
                                    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"):
                                    try:
                                        target_type = self._replace_type_with_class(
                                            action_type_str, device_id, f"动作 {action_name}"
                                        )
                                    except ROSMsgNotFound:
                                        continue
                                    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"]]
                        # 添加内置的驱动命令动作
                        self._add_builtin_actions(device_config, device_id)
                    device_config["file_path"] = str(file.absolute()).replace("\\", "/")
                    device_config["registry_type"] = "device"
                    logger.trace(  # type: ignore
                        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"],
                            }
                    # 将 placeholder_keys 信息添加到 schema 中
                    if "placeholder_keys" in action_config and action_config.get("schema", {}).get(
                        "properties", {}
                    ).get("goal", {}):
                        action_config["schema"]["properties"]["goal"]["_unilabos_placeholder_info"] = action_config[
                            "placeholder_keys"
                        ]

            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, upload_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, upload_registry)

    return lab_registry