re-design pr 291

Author: Xuwznln

unilabos/app/main.py

@@ -830,29 +830,6 @@ def main():
             # 社区包设备直接以 community.<ns>.<id> 注册（扫描期命名空间化），不做 alias 桥接
             args_dict["_community_namespaces"] = community_result.namespaces
 
-    # Plan 09 Task 5: 发现外源包的 unilabos_registry/ 目录,并入 build_registry。
-    # 来源:device dirs(社区包 + 显式 --devices)各自的包根 + `unilabos.registry` entry points。
-    try:
-        from pathlib import Path as _Path
-
-        from unilabos.registry.external_registry_discovery import (
-            discover_registry_paths_from_entry_points,
-            discover_registry_paths_from_project,
-        )
-
-        _ext: list = []
-        for _d in args_dict.get("devices") or []:
-            _dp = _Path(_d)
-            _ext.extend(discover_registry_paths_from_project(_dp))
-            _ext.extend(discover_registry_paths_from_project(_dp.parent))
-        _ext.extend(discover_registry_paths_from_entry_points())
-        _ext_str = list(dict.fromkeys(str(p) for p in _ext))
-        if _ext_str:
-            args_dict["_external_registry_paths"] = _ext_str
-            print_status(f"发现 {len(_ext_str)} 个外源 registry 目录", "info")
-    except Exception as _ext_exc:  # noqa: BLE001
-        logger.warning(f"[ext-registry] 外源 registry 发现跳过: {_ext_exc}")
-
     # Step 0: AST 分析优先 + YAML 注册表加载
     # check_mode 和 upload_registry 都会执行实际 import 验证
     devices_dirs = args_dict.get("devices", None)
@@ -866,7 +843,6 @@ def main():
         check_mode=check_mode,
         complete_registry=complete_registry,
         external_only=external_only,
-        external_registry_paths=args_dict.get("_external_registry_paths"),
     )
 
     # Check mode: 注册表验证完成后直接退出

unilabos/app/package_cli.py

@@ -13,6 +13,7 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
+from unilabos.registry.init_enforce import validate_init_param_enforce
 from unilabos.utils import logger
 from unilabos.utils.banner_print import print_status
 
@@ -53,6 +54,44 @@ def resolve_class_namespace(project_name: str, namespace: Optional[str]) -> str:
     return COMMUNITY_PREFIX + normalize_name(project_name)
 
 
+def discover_registry_paths_from_project(project_root: Path | str) -> List[Path]:
+    """从包根推导目录化注册表路径。
+
+    ``[tool.unilabos.registry].paths`` 相对包含 ``pyproject.toml`` 的包根解析；
+    未声明时回退到包根下的 ``unilabos_registry/``。
+    """
+    root = Path(project_root).resolve()
+    pyproject_paths = _read_pyproject_registry_paths(root)
+    if pyproject_paths:
+        return pyproject_paths
+
+    fallback = root / "unilabos_registry"
+    if fallback.is_dir():
+        return [fallback]
+    return []
+
+
+def _read_pyproject_registry_paths(project_root: Path) -> List[Path]:
+    pyproject = project_root / "pyproject.toml"
+    if not pyproject.is_file():
+        return []
+
+    data = _load_toml(pyproject)
+    registry_config = data.get("tool", {}).get("unilabos", {}).get("registry", {})
+    raw_paths = registry_config.get("paths", [])
+    if not isinstance(raw_paths, list):
+        return []
+
+    paths: List[Path] = []
+    for raw_path in raw_paths:
+        if not isinstance(raw_path, str):
+            continue
+        registry_path = (project_root / raw_path).resolve()
+        if registry_path.is_dir():
+            paths.append(registry_path)
+    return paths
+
+
 def read_pyproject(pkg_dir: Path) -> Dict[str, Any]:
     """读取 pyproject.toml 的 [project] 表，返回 name/version/summary/license/homepage/dependencies 等。"""
     pyproject_path = pkg_dir / "pyproject.toml"
@@ -163,7 +202,6 @@ def read_external_registry_devices(pkg_dir: Path) -> Dict[str, Dict[str, Any]]:
         logger.warning("[package] 未安装 pyyaml，跳过外部注册表读取")
         return {}
 
-    from unilabos.registry.external_registry_discovery import discover_registry_paths_from_project
     from unilabos.registry.yaml_ref import resolve_yaml_refs
 
     registry_roots = discover_registry_paths_from_project(pkg_dir)
@@ -333,6 +371,13 @@ def build_resources_from_registry(
     for device_id, entry in entries.items():
         cls = entry.get("class") if isinstance(entry.get("class"), dict) else {}
         init_schema = entry.get("init_param_schema") if isinstance(entry.get("init_param_schema"), dict) else None
+        init_enforce = entry.get("init_param_enforce")
+        validate_init_param_enforce(
+            device_id,
+            init_schema,
+            init_enforce,
+            error_factory=PackageCLIError,
+        )
         category = entry.get("category") or entry.get("tags") or []
         if isinstance(category, str):
             category = [category]
@@ -360,6 +405,8 @@ def build_resources_from_registry(
         }
         if init_schema is not None:
             resource["init_param_schema"] = init_schema
+        if "init_param_enforce" in entry:
+            resource["init_param_enforce"] = init_enforce
         resources.append(resource)
     return resources
 

unilabos/registry/community_alias.py

@@ -0,0 +1,100 @@
+"""Validation and merge helpers for enforced registry init params."""
+
+import copy
+from typing import Any, Callable, Dict, Optional
+
+
+def merge_init_param_enforce(config: Any, init_enforce: Any) -> Dict[str, Any]:
+    """Merge runtime config with enforced registry params, letting registry win.
+
+    Runtime config may still provide fields not owned by the device model, but
+    every field present in ``init_param_enforce`` is applied last and therefore
+    cannot be overridden by instance-level config.
+    """
+    base = copy.deepcopy(config) if isinstance(config, dict) else {}
+    if not isinstance(init_enforce, dict):
+        return base
+    return _merge_dict_with_enforce(base, init_enforce)
+
+
+def _merge_dict_with_enforce(config: Dict[str, Any], init_enforce: Dict[str, Any]) -> Dict[str, Any]:
+    merged = copy.deepcopy(config)
+    for key, enforced_value in init_enforce.items():
+        current_value = merged.get(key)
+        if isinstance(current_value, dict) and isinstance(enforced_value, dict):
+            merged[key] = _merge_dict_with_enforce(current_value, enforced_value)
+        else:
+            merged[key] = copy.deepcopy(enforced_value)
+    return merged
+
+
+def validate_init_param_enforce(
+    device_id: str,
+    init_schema: Optional[Dict[str, Any]],
+    init_enforce: Any,
+    error_factory: Callable[[str], Exception] = ValueError,
+) -> None:
+    """Validate the JSON enforced config paired with ``init_param_schema``.
+
+    ``init_param_enforce`` is a plain JSON config object. It must not reintroduce
+    the old ``class.init`` object factory DSL; drivers should construct rich
+    objects from JSON-friendly type strings and params inside ``__init__``.
+    """
+    if not isinstance(init_schema, dict):
+        return
+
+    config_schema = init_schema.get("config")
+    if not isinstance(config_schema, dict):
+        return
+
+    required = config_schema.get("required")
+    required_fields = [str(field) for field in required] if isinstance(required, list) else []
+    properties = config_schema.get("properties")
+    has_config_contract = bool(required_fields) or (
+        isinstance(properties, dict) and bool(properties)
+    )
+    if not has_config_contract:
+        return
+
+    if not isinstance(init_enforce, dict):
+        raise error_factory(
+            f"{device_id}: init_param_schema.config 已声明参数，必须提供对象形式的 init_param_enforce"
+        )
+
+    missing = [field for field in required_fields if field not in init_enforce]
+    if missing:
+        raise error_factory(
+            f"{device_id}: init_param_enforce 缺少 required 字段: {', '.join(missing)}"
+        )
+
+    _reject_legacy_init_enforce(
+        init_enforce,
+        f"{device_id}.init_param_enforce",
+        error_factory,
+    )
+
+
+def _reject_legacy_init_enforce(
+    value: Any,
+    path: str,
+    error_factory: Callable[[str], Exception],
+) -> None:
+    if isinstance(value, str):
+        if "${" in value and "}" in value:
+            raise error_factory(f"{path}: init_param_enforce 不支持 ${{...}} 模板，动态值应来自实例 config")
+        return
+
+    if isinstance(value, list):
+        for index, item in enumerate(value):
+            _reject_legacy_init_enforce(item, f"{path}[{index}]", error_factory)
+        return
+
+    if not isinstance(value, dict):
+        return
+
+    keys = set(value)
+    if "factory" in keys or "args" in keys or "kwargs" in keys or keys == {"value"}:
+        raise error_factory(f"{path}: init_param_enforce 不支持 class.init 的 factory/args/kwargs/value DSL")
+
+    for key, item in value.items():
+        _reject_legacy_init_enforce(item, f"{path}.{key}", error_factory)