From 2a3ac28dc8a19f8a827f436f42294167ecb18548 Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 27 Jul 2026 12:00:20 +0200 Subject: [PATCH 1/2] feat(config): add core component config engine Introduce physicalai.config with export_config, to_config, instantiate, normalization, envelope validation, and consolidated dotted-path importing. Refactor inference and robot transport importers to delegate to the shared module. Co-authored-by: Cursor --- docs/reference/config-schema.md | 105 +-- src/physicalai/config/__init__.py | 60 ++ src/physicalai/config/_envelope.py | 126 ++++ src/physicalai/config/_errors.py | 18 + src/physicalai/config/_export.py | 492 +++++++++++++ src/physicalai/config/_instantiate.py | 276 +++++++ src/physicalai/config/_normalize.py | 396 ++++++++++ src/physicalai/config/_path.py | 11 + src/physicalai/config/_types.py | 45 ++ src/physicalai/config/_yaml.py | 79 ++ src/physicalai/config/importing.py | 38 + src/physicalai/inference/_importing.py | 39 +- src/physicalai/robot/transport/_importing.py | 39 +- tests/unit/config/__init__.py | 3 + tests/unit/config/test_component_config.py | 723 +++++++++++++++++++ tests/unit/config/test_yaml.py | 86 +++ 16 files changed, 2409 insertions(+), 127 deletions(-) create mode 100644 src/physicalai/config/__init__.py create mode 100644 src/physicalai/config/_envelope.py create mode 100644 src/physicalai/config/_errors.py create mode 100644 src/physicalai/config/_export.py create mode 100644 src/physicalai/config/_instantiate.py create mode 100644 src/physicalai/config/_normalize.py create mode 100644 src/physicalai/config/_path.py create mode 100644 src/physicalai/config/_types.py create mode 100644 src/physicalai/config/_yaml.py create mode 100644 src/physicalai/config/importing.py create mode 100644 tests/unit/config/__init__.py create mode 100644 tests/unit/config/test_component_config.py create mode 100644 tests/unit/config/test_yaml.py diff --git a/docs/reference/config-schema.md b/docs/reference/config-schema.md index bfc3f4a0..4fb99389 100644 --- a/docs/reference/config-schema.md +++ b/docs/reference/config-schema.md @@ -1,12 +1,9 @@ # Config Schema Reference -> **Preview:** The config system is a planned API. The schemas below document the target config design. +## ComponentConfig -Config files use `class_path` and `init_args` to describe explicit component construction. - -## ComponentSpec - -Direct class mode: +`physicalai.config.ComponentConfig` uses the jsonargparse-compatible +`class_path` + `init_args` shape. ```yaml class_path: package.module.ClassName @@ -14,83 +11,61 @@ init_args: key: value ``` -Registry mode: - -```yaml -type: registered_name -key: value -``` +| Field | Required | Type | Description | +| ------------ | -------- | ------ | ------------------------------------------ | +| `class_path` | yes | string | Non-empty import path resolving to a class | +| `init_args` | no | object | String-keyed constructor arguments | -The `ComponentSpec` fields are listed below. +Omitted `init_args` means an empty mapping. No other top-level keys are +accepted. -| Field | Type | Description | -| ------------ | ------ | --------------------------------------- | -| `class_path` | string | Fully qualified import path | -| `init_args` | object | Constructor keyword arguments | -| `type` | string | Registered short name | -| extra fields | any | Flat constructor args for registry mode | +Values may be `null`, booleans, integers, finite floats, strings, lists, or +string-keyed mappings. Nested components use another `ComponentConfig`. +Every mapping containing `class_path` is reserved as a nested component and +must contain only `class_path` and optional `init_args`. Nesting through +components, lists, and mappings is limited to 10 levels. -The core rules are straightforward. +Export converts `Path` to its unchanged string form, `Enum` to its JSON-safe +value, and tuples to lists. `instantiate()` accepts the JSON representation, +not those original Python objects. -- A component spec must include either `class_path` or `type`. -- If both fields are present, `class_path` takes precedence. -- Nested component specs are instantiated recursively. +## Runtime Documents -## RuntimeConfig +A CLI runtime document places constructor arguments under `runtime:` and may +include method arguments under `run:`. ```yaml runtime: robot: - class_path: physicalai.robot.so101.SO101 + class_path: physicalai.robot.SO101 init_args: port: /dev/ttyACM0 + calibration: ./calibration.json action_source: - class_path: physicalai.runtime.PolicySource + class_path: physicalai.runtime.TeleopSource init_args: - model: - class_path: physicalai.inference.InferenceModel + leader: + class_path: physicalai.robot.SharedRobot init_args: - export_dir: ./exports/act_policy - execution: - class_path: physicalai.runtime.SyncExecution + name: leader-arm fps: 30 +run: + duration_s: 60 ``` -The most common runtime fields are listed below. - -| Field | Type | Description | -| ----------------------- | --------------- | -------------------------------------------------------------------- | -| `runtime.robot` | `ComponentSpec` | Robot implementation | -| `runtime.action_source` | `ComponentSpec` | Action source — always explicit, e.g. `PolicySource`, `TeleopSource` | -| `runtime.fps` | number | Control loop frequency | -| `runtime.cameras` | mapping | Optional camera components | -| `runtime.callbacks` | list | Optional runtime callbacks | - -`action_source.init_args` depends on the chosen class — `PolicySource` takes `model` and `execution` (and optionally `task`, `action_queue`); `TeleopSource` takes `leader`. - -## InferenceConfig - -```yaml -model: - class_path: physicalai.inference.InferenceModel - init_args: - export_dir: ./exports/act_policy - backend: openvino - device: CPU -``` +`physicalai run --config` and `RobotRuntime.from_config()` also accept a bare +exported `RobotRuntime` ComponentConfig whose top-level `class_path` is +`physicalai.runtime.RobotRuntime`. -The most common inference fields are listed below. +## Manifest ComponentSpec -| Field | Type | Description | -| ---------------------------- | --------------- | -------------------------- | -| `model` | `ComponentSpec` | Inference model component | -| `model.init_args.export_dir` | string | Exported package directory | -| `model.init_args.backend` | string | Backend name or `auto` | -| `model.init_args.device` | string | Backend device or `auto` | +Inference manifests retain their separate `ComponentSpec` compatibility +model. It supports registry `type` aliases, artifact resolution, and existing +extra-field behavior. It is not the strict captured `ComponentConfig` schema; +manifest unification is a separate design decision. -## Config vs Manifest +## Security -| Schema | Use | -| --------------- | -------------------------------------------------------- | -| Workflow config | A workflow config describes a workflow before execution. | -| Manifest | A manifest describes an exported package after export. | +`class_path` is executable local configuration. `instantiate()` is for +trusted application and user-authored configs only, never metadata or control +messages received from peers. diff --git a/src/physicalai/config/__init__.py b/src/physicalai/config/__init__.py new file mode 100644 index 00000000..db130070 --- /dev/null +++ b/src/physicalai/config/__init__.py @@ -0,0 +1,60 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Captured component configuration: export live components and instantiate configs. + +Trusted local application and parent→child startup configs only. Never pass +network metadata or untrusted peer payloads to :func:`instantiate`. + +Opt-in path: + +- ``@export_config`` / ``@export_config(class_path=...)`` — remember + caller-supplied constructor args for :func:`to_config`. +- ``@export_config(..., scalar_var_kwargs=True)`` — seal flattened + ``**kwargs`` to JSON scalars (non-scalars fail at :func:`to_config`). +- Domain ctor args may implement :meth:`~ConfigValue.to_config_value` to + return a JSON-compatible fragment (re-normalized). + +Transport and other callers import public names from here (no private +``physicalai.config._*`` imports). Studio needs only +:func:`is_config_exportable` and :func:`to_config`, then domain +``from_config`` helpers (for example ``SharedRobot.from_config``). +""" + +from ._envelope import ( + normalize_component_config, + validate_envelope, +) +from ._errors import ComponentConfigError, ComponentImportError +from ._export import ( + export_config, + is_config_exportable, + resolve_public_class_path, + to_config, +) +from ._instantiate import instantiate +from ._normalize import validate_component_config +from ._types import ComponentConfig, ConfigValue, JsonScalar, JsonValue +from ._yaml import load_yaml, save_yaml, to_yaml +from .importing import import_dotted_path + +__all__ = [ + "ComponentConfig", + "ComponentConfigError", + "ComponentImportError", + "ConfigValue", + "JsonScalar", + "JsonValue", + "export_config", + "import_dotted_path", + "instantiate", + "is_config_exportable", + "load_yaml", + "normalize_component_config", + "resolve_public_class_path", + "save_yaml", + "to_config", + "to_yaml", + "validate_component_config", + "validate_envelope", +] diff --git a/src/physicalai/config/_envelope.py b/src/physicalai/config/_envelope.py new file mode 100644 index 00000000..8d9dc0a4 --- /dev/null +++ b/src/physicalai/config/_envelope.py @@ -0,0 +1,126 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Shared validation for transport construction envelopes. + +Robot-owner and camera-publisher stdin envelopes carry one nested +:class:`ComponentConfig` plus transport-only keys. These helpers keep the +schema-positive validation in one place; transports supply their key names +and allowlists. + +Nothing here imports a ``class_path``. Envelopes are built in the subscriber +process, which must stay free of the driver package — the import happens in +the process that calls :func:`~physicalai.config.instantiate`. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any + +from ._errors import ComponentConfigError +from ._normalize import validate_component_config + +if TYPE_CHECKING: + from ._types import ComponentConfig + + +def validate_envelope( + data: Mapping[str, Any], + *, + component_key: str, + allowed_keys: frozenset[str], + envelope_name: str, +) -> Mapping[str, object]: + """Validate a transport stdin envelope schema-positively. + + Requires *component_key* with a valid ComponentConfig shape and allows + only *allowed_keys*. Unknown keys (including legacy flat shapes) raise a + clear schema error before any import or hardware access. + + Args: + data: Full stdin envelope dict. + component_key: Envelope key holding the nested ComponentConfig + (for example ``"robot"`` or ``"camera"``). + allowed_keys: Complete allowlist of envelope keys. + envelope_name: Short envelope label for error messages + (for example ``"owner"`` or ``"publisher"``). + + Returns: + The validated ComponentConfig mapping (see + :func:`normalize_component_config` for the JSON-serializability check). + + Raises: + TypeError: If *data* or the component value is not a mapping. + ValueError: If the component key is missing or unknown keys are present. + """ + if not isinstance(data, Mapping): + msg = f"{envelope_name} config must be a mapping, got {type(data).__name__}" + raise TypeError(msg) + + unknown = sorted(set(data) - allowed_keys) + if unknown: + msg = ( + f"unknown {envelope_name} config keys {unknown}; " + f"require {component_key!r} with class_path + init_args " + f"(allowed envelope keys: {sorted(allowed_keys)})" + ) + raise ValueError(msg) + + if component_key not in data: + msg = f"{envelope_name} config missing required {component_key!r} ComponentConfig" + raise ValueError(msg) + + component = data[component_key] + if not isinstance(component, Mapping): + msg = f"{envelope_name} {component_key!r} must be a mapping, got {type(component).__name__}" + raise TypeError(msg) + + return validate_component_config(dict(component), path=component_key) + + +def normalize_component_config( + config: Mapping[str, object], + *, + component_key: str, + class_label: str, + json_hint: str = "", +) -> ComponentConfig: + """Validate a ComponentConfig without importing its ``class_path``. + + The ``class_path`` is trusted and kept exactly as written: envelopes are + built in the subscriber process, which must not load the driver package + (and often cannot — the vendor SDK is only installed where the hardware + lives). Import errors surface later, in the process that calls + :func:`~physicalai.config.instantiate`. + + Args: + config: Candidate ``class_path`` + ``init_args`` mapping. + component_key: Path prefix for validation errors (``"robot"`` / ``"camera"``). + class_label: Argument label for ``class_path`` errors. + json_hint: Optional suffix appended to the JSON-serializability error. + + Returns: + A validated config whose ``class_path`` is a dotted import path. + + Raises: + ComponentConfigError: If *config* is not a mapping. + ValueError: If ``class_path`` is not a dotted path or ``init_args`` is + not JSON-serializable. + """ + if not isinstance(config, Mapping): + msg = f"{component_key} must be a ComponentConfig mapping, got {type(config).__name__}" + raise ComponentConfigError(msg) + validated = validate_component_config(dict(config), path=component_key) + class_path = validated["class_path"] + if not class_path.strip() or "." not in class_path: + msg = f"{class_label} must be a nonempty dotted path, got {class_path!r}" + raise ValueError(msg) + init_args = validated["init_args"] + try: + json.dumps({"class_path": class_path, "init_args": init_args}) + except TypeError as exc: + msg = f"{component_key}.init_args must be JSON-serializable{json_hint}: {exc}" + raise ValueError(msg) from exc + return {"class_path": class_path, "init_args": dict(init_args)} diff --git a/src/physicalai/config/_errors.py b/src/physicalai/config/_errors.py new file mode 100644 index 00000000..5b8515f2 --- /dev/null +++ b/src/physicalai/config/_errors.py @@ -0,0 +1,18 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Errors for component configuration export and instantiation.""" + +from __future__ import annotations + + +class ComponentConfigError(Exception): + """Raised when a component config cannot be exported, validated, or built. + + Callers typically recover by correcting configuration. Subclasses mark + distinct failure phases when callers need to distinguish them. + """ + + +class ComponentImportError(ComponentConfigError): + """Raised when a ``class_path`` cannot be resolved to an importable class.""" diff --git a/src/physicalai/config/_export.py b/src/physicalai/config/_export.py new file mode 100644 index 00000000..5a6ac0ba --- /dev/null +++ b/src/physicalai/config/_export.py @@ -0,0 +1,492 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Opt-in constructor-config export for live components.""" + +from __future__ import annotations + +import functools +import inspect +from typing import TYPE_CHECKING, TypeVar, overload + +from ._errors import ComponentConfigError +from ._normalize import ( + normalize_value, + snapshot_captured_value, +) +from ._path import format_path +from ._types import ( + _CAPTURED_INIT_ARGS_ATTR, + _CONFIG_ARGS_ATTR, + _CONFIG_CLASS_PATH_ATTR, + _EXPORT_DEPTH_ATTR, + _EXPORT_MARKER_ATTR, + _MAX_CONFIG_DEPTH, + _NORMALIZE_CAPTURED_INIT_ARGS_ATTR, + ComponentConfig, + JsonValue, +) +from .importing import import_dotted_path + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence + +_T = TypeVar("_T", bound=type) + + +class _NonScalarVarKwarg: + """Poison value so ``to_config`` rejects non-scalar ``**kwargs`` entries. + + Used when ``@export_config(scalar_var_kwargs=True)`` seals flattened + var-keyword arguments to JSON scalars only. + """ + + __slots__ = ("_name",) + + def __init__(self, name: str) -> None: + self._name = name + + def __repr__(self) -> str: + return f"" + + +def _is_json_scalar(value: object) -> bool: + """Return whether *value* is a :data:`~physicalai.config.JsonScalar`.""" + if value is None or isinstance(value, (bool, str)): + return True + if isinstance(value, int) and not isinstance(value, bool): + return True + return isinstance(value, float) + + +def _has_export_marker(obj: object) -> bool: + return bool(getattr(obj, _EXPORT_MARKER_ATTR, False)) + + +def _encode_domain_value(value: object) -> tuple[JsonValue] | None: + """Encode a domain value via ``to_config_value()`` if present. + + Returns: + ``(payload,)`` when a codec applies — the 1-tuple keeps a real JSON + ``null`` payload distinct from "no codec" — or ``None`` when the value + has no domain encoder. The payload is re-normalized by the caller. + """ + encode = getattr(value, "to_config_value", None) + if not callable(encode): + return None + return (encode(),) # type: ignore[return-value] + + +def is_config_exportable(value: object) -> bool: + """Return whether *value* can export a :class:`ComponentConfig`. + + A value is exportable if and only if its most-derived class's effective + ``__init__`` carries the ``@export_config`` marker. + """ + init = type(value).__init__ + return _has_export_marker(init) + + +def declared_config_args(cls: type) -> frozenset[str]: + """Return init-arg names *cls* consumes as ComponentConfig data. + + Declared via ``@export_config(config_args=...)``. :func:`instantiate` + passes these arguments through as plain mappings instead of building the + nested component, so the component receives the recipe it needs to hand to + another process rather than a live object it would immediately discard. + + Returns: + The declared argument names, or an empty set when none are declared. + """ + return getattr(cls.__init__, _CONFIG_ARGS_ATTR, frozenset()) + + +def _class_path_override(cls: type) -> str | None: + """Return the decorator ``class_path=`` only when *cls* owns the decorated ``__init__``. + + Returns: + The override path, or ``None`` for inherited constructors (subclasses + export ``__module__.__qualname__`` unless they re-decorate). + """ + owned_init = cls.__dict__.get("__init__") + if owned_init is None: + return None + explicit = getattr(owned_init, _CONFIG_CLASS_PATH_ATTR, None) + if isinstance(explicit, str) and explicit: + return explicit + return None + + +def resolve_public_class_path(cls: type) -> str: + """Resolve the stable public ``class_path`` for an opted-in component class. + + Uses the decorator ``class_path=`` override when present, otherwise + ``cls.__module__ + "." + cls.__qualname__``. Verifies that importing the + path yields exactly *cls*. + + Args: + cls: The concrete class to resolve. + + Returns: + The importable public dotted path. + + Raises: + ComponentConfigError: If *cls* is local, the path is not importable, or + the path resolves to a different object. + """ + path = _class_path_override(cls) or f"{cls.__module__}.{cls.__qualname__}" + + if "" in cls.__qualname__: + msg = f"local class {path!r} cannot export a stable class_path" + raise ComponentConfigError(msg) + + try: + resolved = import_dotted_path(path) + except (ValueError, ImportError, AttributeError) as exc: + msg = f"class_path {path!r} for {cls.__qualname__} is not importable: {exc}" + raise ComponentConfigError(msg) from exc + if resolved is not cls: + msg = f"class_path {path!r} resolves to {resolved!r}, expected exactly {cls!r}" + raise ComponentConfigError(msg) + return path + + +def _component_path_prefix(value: object) -> str: + cls = type(value) + return _class_path_override(cls) or f"{cls.__module__}.{cls.__qualname__}" + + +def _arg_path(prefix: str, class_path: str, key: str) -> str: + if prefix: + return f"{prefix}.init_args.{key}" + return f"{class_path}.init_args.{key}" + + +def to_config( + value: object, + *, + _path: str = "", + _depth: int = 0, + _seen: set[int] | None = None, +) -> ComponentConfig: + """Export an opted-in live component as JSON-safe ``class_path`` + ``init_args``. + + Nested constructor values may be components (``@export_config``) or domain + values with :meth:`~ConfigValue.to_config_value` (re-normalized; absence of + the method means no codec; a returned ``None`` is JSON null). + + Args: + value: An instance whose class uses ``@export_config``. + + Returns: + A :class:`ComponentConfig` describing the object as constructed. + + Raises: + ComponentConfigError: If *value* is not exportable or captured values + cannot be normalized. + """ + if _depth > _MAX_CONFIG_DEPTH: + msg = f"{format_path(_path)}: nesting depth exceeds {_MAX_CONFIG_DEPTH}" + raise ComponentConfigError(msg) + + seen = _seen if _seen is not None else set() + + if not _has_export_marker(type(value).__init__): + msg = ( + f"{_path or _component_path_prefix(value)}: not config-exportable; " + "decorate the concrete class with @export_config" + ) + raise ComponentConfigError(msg) + + captured = getattr(value, _CAPTURED_INIT_ARGS_ATTR, None) + if captured is None: + msg = ( + f"{_path or _component_path_prefix(value)}: no captured constructor " + "arguments; the object was not constructed through the decorated __init__" + ) + raise ComponentConfigError(msg) + + class_path = resolve_public_class_path(type(value)) + init_args: dict[str, JsonValue] = {} + for key, item in captured.items(): + init_args[key] = normalize_value( + item, + path=_arg_path(_path, class_path, key), + depth=_depth + 1, + seen=seen, + to_config=to_config, + is_exportable=is_config_exportable, + domain_codec=_encode_domain_value, + ) + return {"class_path": class_path, "init_args": init_args} + + +def _instance_to_config(self: object) -> ComponentConfig: + """Instance-method sugar for :func:`to_config`; prefer the module function. + + Returns: + The component config for *self*. + """ + return to_config(self) + + +def _validate_replayable_signature(cls: type, signature: inspect.Signature) -> None: + for param in signature.parameters.values(): + if param.name == "self": + continue + if param.kind is inspect.Parameter.POSITIONAL_ONLY: + msg = ( + f"@export_config on {cls.__qualname__}: positional-only parameter " + f"{param.name!r} is not replayable through keyword init_args" + ) + raise TypeError(msg) + if param.kind is inspect.Parameter.VAR_POSITIONAL: + msg = f"@export_config on {cls.__qualname__}: *args is not replayable through keyword init_args" + raise TypeError(msg) + + +def _flatten_var_kwargs( + supplied: dict[str, object], + *, + cls_name: str, + var_kw_name: str, + scalar_var_kwargs: bool, +) -> None: + """Move flattened ``**kwargs`` entries into *supplied*; optionally seal scalars. + + Raises: + TypeError: If the var-keyword value is not a string-keyed mapping. + """ + extra = supplied.pop(var_kw_name) + if not isinstance(extra, dict): + msg = f"{cls_name}: **{var_kw_name} must be a mapping" + raise TypeError(msg) + for key, value in extra.items(): + if not isinstance(key, str): + msg = f"{cls_name}: **{var_kw_name} keys must be strings" + raise TypeError(msg) + if scalar_var_kwargs and not _is_json_scalar(value): + # Seal so normalize fails at to_config (no silent JSON nest). + supplied[key] = _NonScalarVarKwarg(key) + else: + supplied[key] = snapshot_captured_value(value) + + +def _validate_config_args( + cls: type, + signature: inspect.Signature, + config_args: Sequence[str], + var_kw_name: str | None, +) -> frozenset[str]: + """Check that every declared config arg is a real keyword parameter. + + Returns: + The validated names. + + Raises: + TypeError: If a name is not a string, is not declared by ``__init__``, + or names the ``**kwargs`` parameter. + """ + names: set[str] = set() + for name in config_args: + if not isinstance(name, str) or not name: + msg = f"@export_config on {cls.__qualname__}: config_args entries must be non-empty strings" + raise TypeError(msg) + if name == var_kw_name: + msg = ( + f"@export_config on {cls.__qualname__}: config_args cannot name the " + f"**{name} parameter; declare the individual arguments instead" + ) + raise TypeError(msg) + if name not in signature.parameters or name == "self": + msg = f"@export_config on {cls.__qualname__}: config_args {name!r} is not an __init__ parameter" + raise TypeError(msg) + names.add(name) + return frozenset(names) + + +def _decorate_export_config( + cls: _T, + *, + class_path: str | None, + scalar_var_kwargs: bool, + config_args: Sequence[str] | None, +) -> _T: + if not isinstance(cls, type): + msg = f"@export_config expects a class, got {type(cls).__name__}" + raise TypeError(msg) + + if "__init__" not in cls.__dict__: + msg = ( + f"@export_config on {cls.__qualname__}: decorate only classes that " + "define their own __init__; inherited decorated constructors remain " + "exportable without re-decorating" + ) + raise TypeError(msg) + + original_init = cls.__dict__["__init__"] + if original_init is object.__init__: + msg = f"@export_config on {cls.__qualname__}: class has no custom __init__" + raise TypeError(msg) + + # Re-decorating the same class body is a no-op (keeps prior class_path if any). + if _has_export_marker(original_init): + return cls + + if class_path is not None and (not isinstance(class_path, str) or not class_path): + msg = f"@export_config on {cls.__qualname__}: class_path must be a non-empty string" + raise TypeError(msg) + + signature = inspect.signature(original_init) + _validate_replayable_signature(cls, signature) + + var_kw_name = next( + (name for name, param in signature.parameters.items() if param.kind is inspect.Parameter.VAR_KEYWORD), + None, + ) + if scalar_var_kwargs and var_kw_name is None: + msg = f"@export_config on {cls.__qualname__}: scalar_var_kwargs=True requires a **kwargs parameter" + raise TypeError(msg) + + declared = _validate_config_args(cls, signature, config_args or (), var_kw_name) + + @functools.wraps(original_init) + def wrapped_init(self: object, *args: object, **kwargs: object) -> None: + bound = signature.bind(self, *args, **kwargs) + # Do not apply_defaults — omit unsupplied arguments so reconstruction + # uses current constructor defaults. + supplied = {name: snapshot_captured_value(value) for name, value in bound.arguments.items() if name != "self"} + if var_kw_name is not None and var_kw_name in supplied: + _flatten_var_kwargs( + supplied, + cls_name=cls.__qualname__, + var_kw_name=var_kw_name, + scalar_var_kwargs=scalar_var_kwargs, + ) + + depth = getattr(self, _EXPORT_DEPTH_ATTR, 0) + setattr(self, _EXPORT_DEPTH_ATTR, depth + 1) + try: + original_init(self, *args, **kwargs) + # Only the outermost successful decorated constructor commits. + if getattr(self, _EXPORT_DEPTH_ATTR, 0) == 1: + normalize_captured = getattr(self, _NORMALIZE_CAPTURED_INIT_ARGS_ATTR, None) + if normalize_captured is not None: + if not callable(normalize_captured): + msg = f"{cls.__qualname__}: {_NORMALIZE_CAPTURED_INIT_ARGS_ATTR} must be callable" + raise TypeError(msg) + normalize_captured(supplied) + setattr(self, _CAPTURED_INIT_ARGS_ATTR, supplied) + finally: + setattr(self, _EXPORT_DEPTH_ATTR, depth) + if depth == 0 and hasattr(self, _EXPORT_DEPTH_ATTR): + delattr(self, _EXPORT_DEPTH_ATTR) + + setattr(wrapped_init, _EXPORT_MARKER_ATTR, True) + if class_path is not None: + setattr(wrapped_init, _CONFIG_CLASS_PATH_ATTR, class_path) + if declared: + setattr(wrapped_init, _CONFIG_ARGS_ATTR, declared) + cls.__init__ = wrapped_init # type: ignore[method-assign] + # Convenience method; type checkers do not see the injection — prefer module to_config. + cls.to_config = _instance_to_config # type: ignore[attr-defined] + return cls + + +@overload +def export_config(cls: _T, /) -> _T: ... + + +@overload +def export_config( + *, + class_path: str | None = None, + scalar_var_kwargs: bool = False, + config_args: Sequence[str] | None = None, +) -> Callable[[_T], _T]: ... + + +def export_config( + cls: _T | None = None, + /, + *, + class_path: str | None = None, + scalar_var_kwargs: bool = False, + config_args: Sequence[str] | None = None, +) -> _T | Callable[[_T], _T]: + """Opt a concrete class into constructor-config export via :func:`to_config`. + + Remembers caller-supplied ``__init__`` arguments (not defaults). Rejects + constructors that declare positional-only parameters or ``*args``. Injects + an instance ``to_config()`` convenience method; library code should still + call the module-level :func:`to_config`. + + Usage:: + + @export_config + class MyRobot: ... + + @export_config(class_path="physicalai.robot.SO101") + class SO101: ... + + @export_config(class_path="physicalai.inference.InferenceModel", scalar_var_kwargs=True) + class InferenceModel: ... + + When ``class_path`` is omitted, export uses + ``type(self).__module__ + "." + type(self).__qualname__``. Pass + ``class_path=`` when the public import path differs from the defining + module (for example a package re-export). + + Pass ``scalar_var_kwargs=True`` when flattened ``**kwargs`` must export as + JSON scalars only (``None`` / ``bool`` / ``int`` / ``float`` / ``str``). + Non-scalar var-keyword values then fail at :func:`to_config` instead of + being normalized as nested JSON. Requires a ``**kwargs`` parameter. + + Nested non-component domain values (for example calibration objects) may + implement :meth:`~ConfigValue.to_config_value` so they normalize to + constructor-compatible JSON; that method's output is re-normalized. + Absence of the method means no codec; a returned ``None`` is JSON null. + + Inheritance: + + - Decorate every concrete class that **overrides** ``__init__``. + - An undecorated overriding ``__init__`` fails at :func:`to_config` (partial + base recipes are not emitted). + - A subclass that inherits a decorated constructor unchanged remains valid + without re-decorating. + - Do not apply ``@export_config`` to a subclass that does not define its + own ``__init__`` (that would double-wrap the inherited wrapper). + + Args: + cls: Class to decorate when used as ``@export_config``. Must define + ``__init__`` in its own class body. + class_path: Optional stable public import path for export. Verified on + export to resolve exactly to the decorated class. + scalar_var_kwargs: When ``True``, seal flattened ``**kwargs`` to JSON + scalars so non-scalars fail at :func:`to_config`. + config_args: Init-arg names the class consumes as ComponentConfig + *data*. :func:`instantiate` passes these through as plain mappings + instead of constructing the nested component — use it for spawn + recipes that must cross a process boundary (for example + ``SharedCamera(camera=...)``). + + Returns: + The decorated class, or a decorator when keyword options are passed. + """ + if cls is not None: + return _decorate_export_config( + cls, + class_path=class_path, + scalar_var_kwargs=scalar_var_kwargs, + config_args=config_args, + ) + + def decorator(target: _T) -> _T: + return _decorate_export_config( + target, + class_path=class_path, + scalar_var_kwargs=scalar_var_kwargs, + config_args=config_args, + ) + + return decorator diff --git a/src/physicalai/config/_instantiate.py b/src/physicalai/config/_instantiate.py new file mode 100644 index 00000000..5f665c86 --- /dev/null +++ b/src/physicalai/config/_instantiate.py @@ -0,0 +1,276 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Instantiate trusted ComponentConfig trees.""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from enum import Enum +from typing import cast + +from ._errors import ComponentConfigError, ComponentImportError +from ._export import declared_config_args +from ._normalize import validate_component_config +from ._path import format_path +from ._types import _MAX_CONFIG_DEPTH, ComponentConfig, JsonValue +from .importing import import_dotted_path + + +def _is_nested_config(value: object) -> bool: + return isinstance(value, Mapping) and "class_path" in value + + +def _check_preflight_depth(path: str, depth: int) -> None: + if depth > _MAX_CONFIG_DEPTH: + msg = f"{format_path(path)}: nesting depth exceeds {_MAX_CONFIG_DEPTH}" + raise ComponentConfigError(msg) + + +def _preflight_config( + config: object, + *, + path: str, + depth: int, + seen: set[int], +) -> None: + """Validate a complete component subtree before any class import. + + Raises: + ComponentConfigError: If the config tree is malformed. + """ + _check_preflight_depth(path, depth) + + validated = validate_component_config(config, path=path) + config_id = id(config) + if config_id in seen: + msg = f"{format_path(path)}: cyclic component config is not instantiable" + raise ComponentConfigError(msg) + seen.add(config_id) + try: + for key, item in validated["init_args"].items(): + child_path = f"{path}.init_args.{key}" if path else f"{validated['class_path']}.init_args.{key}" + _preflight_value(item, path=child_path, depth=depth + 1, seen=seen) + finally: + seen.discard(config_id) + + +def _preflight_mapping( + value: dict[object, object], + *, + path: str, + depth: int, + seen: set[int], +) -> None: + if "class_path" in value: + _preflight_config(value, path=path, depth=depth, seen=seen) + return + + value_id = id(value) + if value_id in seen: + msg = f"{format_path(path)}: cyclic mapping is not instantiable" + raise ComponentConfigError(msg) + seen.add(value_id) + try: + for key, item in value.items(): + if not isinstance(key, str): + msg = f"{format_path(path)}: mapping keys must be strings, got {type(key).__name__}" + raise ComponentConfigError(msg) + child_path = f"{path}.{key}" if path else key + _preflight_value(item, path=child_path, depth=depth + 1, seen=seen) + finally: + seen.discard(value_id) + + +def _preflight_list( + value: list[object], + *, + path: str, + depth: int, + seen: set[int], +) -> None: + value_id = id(value) + if value_id in seen: + msg = f"{format_path(path)}: cyclic sequence is not instantiable" + raise ComponentConfigError(msg) + seen.add(value_id) + try: + for index, item in enumerate(value): + child_path = f"{path}[{index}]" if path else f"[{index}]" + _preflight_value(item, path=child_path, depth=depth + 1, seen=seen) + finally: + seen.discard(value_id) + + +def _preflight_value( + value: object, + *, + path: str, + depth: int, + seen: set[int], +) -> None: + """Validate one constructor value against the recursive JSON model. + + Raises: + ComponentConfigError: If the value is outside the JSON model. + """ + _check_preflight_depth(path, depth) + + if isinstance(value, Enum): + msg = f"{format_path(path)}: Enum is not a JSON-compatible component config value; pass its value instead" + raise ComponentConfigError(msg) + if value is None or isinstance(value, (bool, str)): + return + if isinstance(value, int): + return + if isinstance(value, float): + if not math.isfinite(value): + msg = f"{format_path(path)}: non-finite float {value!r} is not JSON-portable" + raise ComponentConfigError(msg) + return + + if isinstance(value, dict): + _preflight_mapping(value, path=path, depth=depth, seen=seen) + return + + if isinstance(value, list): + _preflight_list(value, path=path, depth=depth, seen=seen) + return + + msg = f"{format_path(path)}: {type(value).__name__} is not a JSON-compatible component config value" + raise ComponentConfigError(msg) + + +def _decode_value( + value: object, + *, + path: str, + depth: int, + seen: set[int], +) -> object: + if depth > _MAX_CONFIG_DEPTH: + msg = f"{format_path(path)}: nesting depth exceeds {_MAX_CONFIG_DEPTH}" + raise ComponentConfigError(msg) + + if _is_nested_config(value): + return _instantiate_impl( + cast("ComponentConfig | Mapping[str, JsonValue]", value), + path=path, + depth=depth, + seen=seen, + ) + + if isinstance(value, Mapping): + obj_id = id(value) + if obj_id in seen: + msg = f"{format_path(path)}: cyclic mapping is not instantiable" + raise ComponentConfigError(msg) + seen.add(obj_id) + try: + result: dict[str, object] = {} + for key, item in value.items(): + if not isinstance(key, str): + msg = f"{format_path(path)}: mapping keys must be strings, got {type(key).__name__}" + raise ComponentConfigError(msg) + child_path = f"{path}.{key}" if path else key + result[key] = _decode_value(item, path=child_path, depth=depth + 1, seen=seen) + return result + finally: + seen.discard(obj_id) + + if isinstance(value, list): + obj_id = id(value) + if obj_id in seen: + msg = f"{format_path(path)}: cyclic sequence is not instantiable" + raise ComponentConfigError(msg) + seen.add(obj_id) + try: + items: list[object] = [] + for index, item in enumerate(value): + child_path = f"{path}[{index}]" if path else f"[{index}]" + items.append(_decode_value(item, path=child_path, depth=depth + 1, seen=seen)) + return items + finally: + seen.discard(obj_id) + + return value + + +def _resolve_class(class_path: str, *, path: str) -> type: + if "" in class_path: + msg = f"{format_path(path)}: local class {class_path!r} cannot be imported" + raise ComponentConfigError(msg) + try: + obj = import_dotted_path(class_path) + except (ValueError, ImportError, AttributeError) as exc: + msg = f"{format_path(path)}: cannot import class_path {class_path!r}: {exc}" + raise ComponentImportError(msg) from exc + if not isinstance(obj, type): + msg = f"{format_path(path)}: {class_path!r} does not resolve to a class (got {type(obj).__name__})" + raise ComponentImportError(msg) + if "" in obj.__qualname__: + msg = f"{format_path(path)}: local class {class_path!r} cannot be instantiated" + raise ComponentConfigError(msg) + return obj + + +def _instantiate_impl( + config: ComponentConfig | Mapping[str, JsonValue], + *, + path: str, + depth: int, + seen: set[int], +) -> object: + if depth > _MAX_CONFIG_DEPTH: + msg = f"{format_path(path)}: nesting depth exceeds {_MAX_CONFIG_DEPTH}" + raise ComponentConfigError(msg) + + validated = validate_component_config(config, path=path) + cls = _resolve_class(validated["class_path"], path=path) + config_args = declared_config_args(cls) + + decoded_args: dict[str, object] = {} + for key, item in validated["init_args"].items(): + if key in config_args: + # Declared ComponentConfig data: hand the recipe over untouched so + # nothing nested is constructed in this process. + decoded_args[key] = item + continue + child_path = f"{path}.init_args.{key}" if path else f"{validated['class_path']}.init_args.{key}" + decoded_args[key] = _decode_value(item, path=child_path, depth=depth + 1, seen=seen) + + try: + return cls(**decoded_args) + except Exception as exc: + loc = path or validated["class_path"] + exc.add_note(f"{format_path(loc)}: constructor failed while instantiating component config") + raise + + +def instantiate(config: ComponentConfig | Mapping[str, JsonValue]) -> object: + """Build a fresh component from a trusted :class:`ComponentConfig`. + + Validates *config* before importing. Recursively instantiates nested + component configs in ``init_args``, then calls the class with keyword + arguments. Init args a class declares via + ``@export_config(config_args=...)`` are passed through as plain mappings + instead of being constructed. Does not invoke lifecycle methods beyond the + constructor. + + Trusted local application and parent→child startup configs only. Never + pass network metadata or untrusted peer payloads. + + Malformed configs raise :class:`ComponentConfigError` (including + :class:`ComponentImportError` for unresolved ``class_path``). Constructor + failures propagate as their original exception type with path context + attached via :meth:`BaseException.add_note`. + + Args: + config: Trusted ``class_path`` + ``init_args`` mapping. + + Returns: + A new instance of the configured class. + """ + _preflight_config(config, path="", depth=0, seen=set()) + return _instantiate_impl(config, path="", depth=0, seen=set()) diff --git a/src/physicalai/config/_normalize.py b/src/physicalai/config/_normalize.py new file mode 100644 index 00000000..f4ee4a0d --- /dev/null +++ b/src/physicalai/config/_normalize.py @@ -0,0 +1,396 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Normalize live values into JSON-safe component configuration fragments.""" + +from __future__ import annotations + +import math +from collections.abc import Callable, Mapping +from enum import Enum +from pathlib import Path +from typing import cast + +from ._errors import ComponentConfigError +from ._path import format_path +from ._types import ( + _MAX_CONFIG_DEPTH, + _REPR_LIMIT, + ComponentConfig, + JsonValue, +) + +ToConfigFn = Callable[..., ComponentConfig] +IsExportableFn = Callable[[object], bool] +# Present codec result is a 1-tuple so JSON ``null`` (``None``) is distinct from +# “no codec” (plain ``None`` return from the codec function). +DomainCodecFn = Callable[[object], tuple[JsonValue] | None] + + +def _is_component_config_mapping(value: Mapping[object, object]) -> bool: + return "class_path" in value + + +def validate_component_config(config: object, *, path: str = "") -> ComponentConfig: + """Validate a mapping as a :class:`ComponentConfig` without importing. + + Args: + config: Candidate config mapping. + path: Argument path prefix for error messages. + + Returns: + The validated config (``init_args`` defaulted to ``{}`` when omitted). + + Raises: + ComponentConfigError: If the mapping is malformed. + """ + loc = format_path(path) + if not isinstance(config, dict): + msg = f"{loc}: component config must be a mapping, got {type(config).__name__}" + raise ComponentConfigError(msg) + + keys = set(config) + allowed = {"class_path", "init_args"} + extra = keys - allowed + if extra: + extras = ", ".join(sorted(repr(k) for k in extra)) + msg = ( + f"{loc}: nested component config may only contain 'class_path' and " + f"'init_args'; unexpected keys: {extras}. For an ordinary mapping that " + "needs a 'class_path' data key, encode it via to_config_value() " + "(or another domain wrapper) instead of nesting it as a component config" + ) + raise ComponentConfigError(msg) + if "class_path" not in config: + msg = f"{loc}: component config missing required 'class_path'" + raise ComponentConfigError(msg) + + class_path = config["class_path"] + if not isinstance(class_path, str) or not class_path: + msg = f"{loc}: 'class_path' must be a non-empty string" + raise ComponentConfigError(msg) + + if "init_args" not in config: + init_args: object = {} + else: + init_args = config["init_args"] + if init_args is None: + msg = f"{loc}: 'init_args' must be a mapping, got None" + raise ComponentConfigError(msg) + if not isinstance(init_args, dict): + msg = f"{loc}: 'init_args' must be a mapping, got {type(init_args).__name__}" + raise ComponentConfigError(msg) + for key in init_args: + if not isinstance(key, str): + msg = f"{loc}.init_args: keys must be strings, got {type(key).__name__}" + raise ComponentConfigError(msg) + + return {"class_path": class_path, "init_args": dict(init_args)} + + +def _check_depth(path: str, depth: int) -> None: + if depth > _MAX_CONFIG_DEPTH: + msg = f"{format_path(path)}: nesting depth exceeds {_MAX_CONFIG_DEPTH}" + raise ComponentConfigError(msg) + + +def _try_normalize_scalar(value: object, *, path: str) -> tuple[bool, JsonValue]: + """Return ``(True, json_value)`` for scalars/Path, else ``(False, None)``. + + Raises: + ComponentConfigError: If *value* is a non-finite float. + """ + if value is None or isinstance(value, (bool, str)): + return True, value + if isinstance(value, int) and not isinstance(value, bool): + return True, value + if isinstance(value, float): + if not math.isfinite(value): + msg = f"{format_path(path)}: non-finite float {value!r} is not JSON-portable" + raise ComponentConfigError(msg) + return True, value + if isinstance(value, Path): + return True, str(value) + return False, None + + +def _normalize_enum(value: Enum, *, path: str) -> JsonValue: + enum_value = value.value + if isinstance(enum_value, (bool, str)) or ( + isinstance(enum_value, (int, float)) and not isinstance(enum_value, bool) + ): + if isinstance(enum_value, float) and not math.isfinite(enum_value): + msg = f"{format_path(path)}: non-finite enum value {enum_value!r}" + raise ComponentConfigError(msg) + return enum_value # type: ignore[return-value] + msg = f"{format_path(path)}: enum {type(value).__name__} value {type(enum_value).__name__} is not JSON-safe" + raise ComponentConfigError(msg) + + +def _normalize_exportable( + value: object, + *, + path: str, + depth: int, + seen: set[int], + to_config: ToConfigFn, +) -> JsonValue: + # Match instantiate: a nested component config found at *depth* is itself at + # *depth* (its init_args then advance to depth + 1 inside to_config). + nested = to_config(value, _path=path, _depth=depth, _seen=seen) + return cast("JsonValue", dict(nested)) + + +def _normalize_mapping( + value: Mapping[object, object], + *, + path: str, + depth: int, + seen: set[int], + codec_seen: set[int], + to_config: ToConfigFn | None, + is_exportable: IsExportableFn | None, + domain_codec: DomainCodecFn | None, +) -> JsonValue: + obj_id = id(value) + if obj_id in seen: + msg = f"{format_path(path)}: cyclic mapping is not serializable" + raise ComponentConfigError(msg) + + if _is_component_config_mapping(value): + validated = validate_component_config(value, path=path) + nested_args: dict[str, JsonValue] = {} + seen.add(obj_id) + try: + for key, item in validated["init_args"].items(): + child_path = f"{path}.init_args.{key}" if path else f"init_args.{key}" + nested_args[key] = normalize_value( + item, + path=child_path, + depth=depth + 1, + seen=seen, + codec_seen=codec_seen, + to_config=to_config, + is_exportable=is_exportable, + domain_codec=domain_codec, + ) + finally: + seen.discard(obj_id) + return {"class_path": validated["class_path"], "init_args": nested_args} + + seen.add(obj_id) + try: + result: dict[str, JsonValue] = {} + for key, item in value.items(): + if not isinstance(key, str): + msg = f"{format_path(path)}: mapping keys must be strings, got {type(key).__name__}" + raise ComponentConfigError(msg) + child_path = f"{path}.{key}" if path else key + result[key] = normalize_value( + item, + path=child_path, + depth=depth + 1, + seen=seen, + codec_seen=codec_seen, + to_config=to_config, + is_exportable=is_exportable, + domain_codec=domain_codec, + ) + return result + finally: + seen.discard(obj_id) + + +def _normalize_sequence( + value: list[object] | tuple[object, ...], + *, + path: str, + depth: int, + seen: set[int], + codec_seen: set[int], + to_config: ToConfigFn | None, + is_exportable: IsExportableFn | None, + domain_codec: DomainCodecFn | None, +) -> JsonValue: + obj_id = id(value) + if obj_id in seen: + msg = f"{format_path(path)}: cyclic sequence is not serializable" + raise ComponentConfigError(msg) + seen.add(obj_id) + try: + items: list[JsonValue] = [] + for index, item in enumerate(value): + child_path = f"{path}[{index}]" if path else f"[{index}]" + items.append( + normalize_value( + item, + path=child_path, + depth=depth + 1, + seen=seen, + codec_seen=codec_seen, + to_config=to_config, + is_exportable=is_exportable, + domain_codec=domain_codec, + ) + ) + return items + finally: + seen.discard(obj_id) + + +def _unsupported_message(value: object, *, path: str) -> str: + type_name = type(value).__name__ + repr_value = repr(value) + if len(repr_value) > _REPR_LIMIT: + repr_value = repr_value[: _REPR_LIMIT - 3] + "..." + return f"{format_path(path)}: cannot encode {type_name} {repr_value}; omit it or use a supported component value" + + +def normalize_value( + value: object, + *, + path: str = "", + depth: int = 0, + seen: set[int] | None = None, + codec_seen: set[int] | None = None, + to_config: ToConfigFn | None = None, + is_exportable: IsExportableFn | None = None, + domain_codec: DomainCodecFn | None = None, +) -> JsonValue: + """Recursively normalize *value* to a JSON-safe representation. + + Args: + value: Captured constructor argument or nested structure. + path: Argument path for error messages. + depth: Current nesting depth (configs, lists, mappings, and codec hops). + seen: Container identities already visited (cycle detection). + codec_seen: Domain values currently mid-encode via ``to_config_value``. + to_config: Callback for nested exportable components. + is_exportable: Predicate for nested exportable components. + domain_codec: Optional codec. Returns ``None`` when there is no codec for + *value*; returns a 1-tuple ``(payload,)`` when a codec applies + (``payload`` may be JSON ``null`` / ``None``). Tuple payloads are + re-normalized (fail closed). + + Returns: + A JSON-safe value. + + Raises: + ComponentConfigError: On unsupported values, cycles, depth overflow, + or a domain codec that returns the same object identity. + """ + _check_depth(path, depth) + if seen is None: + seen = set() + if codec_seen is None: + codec_seen = set() + + handled, scalar = _try_normalize_scalar(value, path=path) + if handled: + return scalar + + if isinstance(value, Enum): + return _normalize_enum(value, path=path) + + if is_exportable is not None and to_config is not None and is_exportable(value): + return _normalize_exportable(value, path=path, depth=depth, seen=seen, to_config=to_config) + + if domain_codec is not None: + encoded_result = domain_codec(value) + if encoded_result is not None: + obj_id = id(value) + if obj_id in codec_seen: + msg = f"{format_path(path)}: cyclic to_config_value() encoding is not serializable" + raise ComponentConfigError(msg) + (encoded,) = encoded_result + # Identity guard: a method that returns self would recurse forever. + if encoded is value: + msg = ( + f"{format_path(path)}: to_config_value() must return a new " + "JSON-compatible value, not the same object " + "(absence of the method means no codec)" + ) + raise ComponentConfigError(msg) + codec_seen.add(obj_id) + try: + return normalize_value( + encoded, + path=path, + depth=depth + 1, + seen=seen, + codec_seen=codec_seen, + to_config=to_config, + is_exportable=is_exportable, + domain_codec=domain_codec, + ) + finally: + codec_seen.discard(obj_id) + + if isinstance(value, Mapping): + return _normalize_mapping( + value, + path=path, + depth=depth, + seen=seen, + codec_seen=codec_seen, + to_config=to_config, + is_exportable=is_exportable, + domain_codec=domain_codec, + ) + + if isinstance(value, (list, tuple)): + return _normalize_sequence( + value, + path=path, + depth=depth, + seen=seen, + codec_seen=codec_seen, + to_config=to_config, + is_exportable=is_exportable, + domain_codec=domain_codec, + ) + + msg = _unsupported_message(value, path=path) + raise ComponentConfigError(msg) + + +def snapshot_captured_value( + value: object, + *, + memo: dict[int, object] | None = None, +) -> object: + """Deep-snapshot built-in mutable containers; keep other values by reference. + + Non-container values — including nested exportable components and domain + values — are retained by identity so their own conversion remains + authoritative. Cyclic containers are preserved via *memo* so + :func:`~physicalai.config.to_config` can reject them during normalization. + + Returns: + A snapshot of *value* suitable for later normalization. + """ + if memo is None: + memo = {} + + if isinstance(value, (dict, list, tuple)): + obj_id = id(value) + if obj_id in memo: + return memo[obj_id] + if isinstance(value, dict): + result: dict[object, object] = {} + memo[obj_id] = result + for key, item in value.items(): + result[key] = snapshot_captured_value(item, memo=memo) + snap: object = result + elif isinstance(value, list): + result_list: list[object] = [] + memo[obj_id] = result_list + result_list.extend(snapshot_captured_value(item, memo=memo) for item in value) + snap = result_list + else: + memo[obj_id] = value + snap = tuple(snapshot_captured_value(item, memo=memo) for item in value) + memo[obj_id] = snap + return snap + + return value diff --git a/src/physicalai/config/_path.py b/src/physicalai/config/_path.py new file mode 100644 index 00000000..921336d1 --- /dev/null +++ b/src/physicalai/config/_path.py @@ -0,0 +1,11 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Shared argument-path formatting for component-config errors.""" + +from __future__ import annotations + + +def format_path(path: str) -> str: + """Return *path* for error messages, or ```` when empty.""" + return path or "" diff --git a/src/physicalai/config/_types.py b/src/physicalai/config/_types.py new file mode 100644 index 00000000..f727b335 --- /dev/null +++ b/src/physicalai/config/_types.py @@ -0,0 +1,45 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Types for JSON-safe component configuration.""" + +from __future__ import annotations + +from typing import Protocol, TypedDict, runtime_checkable + +JsonScalar = None | bool | int | float | str +JsonValue = JsonScalar | list["JsonValue"] | dict[str, "JsonValue"] + + +class ComponentConfig(TypedDict): + """Wire shape shared with jsonargparse: ``class_path`` + ``init_args``.""" + + class_path: str + init_args: dict[str, JsonValue] + + +@runtime_checkable +class ConfigValue(Protocol): + """Domain value that encodes to constructor-compatible JSON for export.""" + + def to_config_value(self) -> JsonValue: + """Return ctor-compatible JSON for component-config export.""" + ... + + +# Maximum nesting depth for recursive normalization and instantiation. +# Counts traversal through lists and mappings as well as nested component configs. +_MAX_CONFIG_DEPTH = 10 + +# Private attribute names used by @export_config. +_CAPTURED_INIT_ARGS_ATTR = "_physicalai_captured_init_args" +_EXPORT_DEPTH_ATTR = "_physicalai_export_config_depth" +_EXPORT_MARKER_ATTR = "_physicalai_export_config" +_NORMALIZE_CAPTURED_INIT_ARGS_ATTR = "_physicalai_normalize_captured_init_args" +# Optional public class_path override stored on the decorated __init__ wrapper. +_CONFIG_CLASS_PATH_ATTR = "_physicalai_config_class_path" +# Init-arg names the component consumes as ComponentConfig *data* rather than as +# a constructed object. instantiate() passes these through undecoded. +_CONFIG_ARGS_ATTR = "_physicalai_config_args" + +_REPR_LIMIT = 80 diff --git a/src/physicalai/config/_yaml.py b/src/physicalai/config/_yaml.py new file mode 100644 index 00000000..80120af6 --- /dev/null +++ b/src/physicalai/config/_yaml.py @@ -0,0 +1,79 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""YAML round-trip helpers for component configs. + +``to_yaml`` / ``save_yaml`` serialize a live ``@export_config`` component (or +an existing ``class_path`` + ``init_args`` mapping) to a YAML document that +``load_yaml`` + :func:`~physicalai.config.instantiate` can rebuild. The same +document is accepted by ``physicalai run --config`` when the top-level +component is a ``RobotRuntime``. + +Trusted local configs only — never feed network-received YAML to +:func:`~physicalai.config.instantiate`. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import TYPE_CHECKING + +import yaml + +from ._errors import ComponentConfigError +from ._export import to_config +from ._normalize import validate_component_config + +if TYPE_CHECKING: + from ._types import ComponentConfig, JsonValue + + +def to_yaml(component: object) -> str: + """Serialize a component to a YAML ``class_path`` + ``init_args`` document. + + Args: + component: A live ``@export_config`` instance, or an existing + :class:`ComponentConfig` mapping (validated, not re-exported). + + Returns: + YAML text of the validated :class:`ComponentConfig`. + """ + config: ComponentConfig + config = validate_component_config(dict(component)) if isinstance(component, Mapping) else to_config(component) + return yaml.safe_dump(dict(config), sort_keys=False, default_flow_style=False) + + +def save_yaml(component: object, path: str | Path) -> None: + """Write :func:`to_yaml` output to *path* (parent directories must exist). + + Args: + component: A live ``@export_config`` instance or a + :class:`ComponentConfig` mapping. + path: Destination file path. + """ + Path(path).write_text(to_yaml(component), encoding="utf-8") + + +def load_yaml(path: str | Path) -> dict[str, JsonValue]: + """Load a YAML document as a mapping, ready for :func:`~physicalai.config.instantiate`. + + The document is parsed with ``yaml.safe_load`` and only shape-checked to + be a mapping — full component validation happens in + :func:`~physicalai.config.instantiate`. + + Args: + path: YAML file previously written by :func:`save_yaml` (or + hand-authored in the same shape). + + Returns: + The loaded top-level mapping. + + Raises: + ComponentConfigError: If the document is not a mapping. + """ + loaded = yaml.safe_load(Path(path).read_text(encoding="utf-8")) + if not isinstance(loaded, dict): + msg = f"{str(path)!r}: YAML config must be a mapping, got {type(loaded).__name__}" + raise ComponentConfigError(msg) + return loaded diff --git a/src/physicalai/config/importing.py b/src/physicalai/config/importing.py new file mode 100644 index 00000000..bb49b720 --- /dev/null +++ b/src/physicalai/config/importing.py @@ -0,0 +1,38 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 + +"""Dotted-path imports for trusted component configuration.""" + +from __future__ import annotations + +import importlib + + +def import_dotted_path(path: str) -> object: + """Resolve a fully-qualified path, including nested attributes. + + Args: + path: Dotted path such as ``"pkg.mod.Outer.Inner"``. + + Returns: + The resolved object. + + Raises: + ValueError: If no module prefix can be imported, or *path* has no ``.``. + """ + if "." not in path: + msg = f"dotted path must contain at least one '.': {path!r}" + raise ValueError(msg) + + segments = path.split(".") + for split_index in range(len(segments), 0, -1): + try: + obj: object = importlib.import_module(".".join(segments[:split_index])) + except ImportError: + continue + for attr in segments[split_index:]: + obj = getattr(obj, attr) + return obj + + msg = f"could not import any module prefix of {path!r}" + raise ValueError(msg) diff --git a/src/physicalai/inference/_importing.py b/src/physicalai/inference/_importing.py index 8965d973..8d277b20 100644 --- a/src/physicalai/inference/_importing.py +++ b/src/physicalai/inference/_importing.py @@ -1,38 +1,15 @@ # Copyright (C) 2026 Intel Corporation # SPDX-License-Identifier: Apache-2.0 -"""Dotted-path imports for inference components.""" +"""Dotted-path imports for inference components. -from __future__ import annotations - -import importlib - - -def import_dotted_path(path: str) -> object: - """Resolve a fully-qualified path, including nested attributes. +Canonical implementation lives in :mod:`physicalai.config.importing`. This +module re-exports for backward-compatible imports without loading export or +instantiate. +""" - Args: - path: Dotted path such as ``"pkg.mod.Outer.Inner"``. - - Returns: - The resolved object. - - Raises: - ValueError: If no module prefix can be imported. - """ - if "." not in path: - msg = f"dotted path must contain at least one '.': {path!r}" - raise ValueError(msg) +from __future__ import annotations - segments = path.split(".") - for split_index in range(len(segments), 0, -1): - try: - obj: object = importlib.import_module(".".join(segments[:split_index])) - except ImportError: - continue - for attr in segments[split_index:]: - obj = getattr(obj, attr) - return obj +from physicalai.config.importing import import_dotted_path - msg = f"could not import any module prefix of {path!r}" - raise ValueError(msg) +__all__ = ["import_dotted_path"] diff --git a/src/physicalai/robot/transport/_importing.py b/src/physicalai/robot/transport/_importing.py index 0f3dc58c..021561f7 100644 --- a/src/physicalai/robot/transport/_importing.py +++ b/src/physicalai/robot/transport/_importing.py @@ -1,38 +1,15 @@ # Copyright (C) 2026 Intel Corporation # SPDX-License-Identifier: Apache-2.0 -"""Dotted-path imports for trusted robot owner configuration.""" +"""Dotted-path imports for trusted robot owner configuration. -from __future__ import annotations - -import importlib - - -def import_dotted_path(path: str) -> object: - """Resolve a fully-qualified path, including nested attributes. +Canonical implementation lives in :mod:`physicalai.config.importing`. This +module re-exports for backward-compatible imports without loading export or +instantiate. +""" - Args: - path: Dotted path such as ``"pkg.mod.Outer.Inner"``. - - Returns: - The resolved object. - - Raises: - ValueError: If no module prefix can be imported. - """ - if "." not in path: - msg = f"dotted path must contain at least one '.': {path!r}" - raise ValueError(msg) +from __future__ import annotations - segments = path.split(".") - for split_index in range(len(segments), 0, -1): - try: - obj: object = importlib.import_module(".".join(segments[:split_index])) - except ImportError: - continue - for attr in segments[split_index:]: - obj = getattr(obj, attr) - return obj +from physicalai.config.importing import import_dotted_path - msg = f"could not import any module prefix of {path!r}" - raise ValueError(msg) +__all__ = ["import_dotted_path"] diff --git a/tests/unit/config/__init__.py b/tests/unit/config/__init__.py new file mode 100644 index 00000000..888f2e02 --- /dev/null +++ b/tests/unit/config/__init__.py @@ -0,0 +1,3 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# ruff:file-ignore[undocumented-public-package] diff --git a/tests/unit/config/test_component_config.py b/tests/unit/config/test_component_config.py new file mode 100644 index 00000000..2ceb6cef --- /dev/null +++ b/tests/unit/config/test_component_config.py @@ -0,0 +1,723 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# ruff:file-ignore[undocumented-public-module, undocumented-public-class, undocumented-public-method, undocumented-public-init, undocumented-magic-method, bad-dunder-method-name, magic-value-comparison, no-self-use, assert, unused-method-argument, too-many-public-methods] + +from __future__ import annotations + +import inspect +import json +import math +from collections.abc import Mapping +from enum import Enum +from pathlib import Path +from typing import Protocol, cast, runtime_checkable +from unittest.mock import MagicMock, patch + +import pytest + +from physicalai.config import ( + ComponentConfig, + ComponentConfigError, + ComponentImportError, + JsonValue, + export_config, + import_dotted_path, + instantiate, + is_config_exportable, + to_config, +) + +# Matches physicalai.config._types._MAX_CONFIG_DEPTH +_MAX_CONFIG_DEPTH = 10 + + +def _as_mapping(value: object) -> dict[str, object]: + assert isinstance(value, dict) + return value + + +def _as_list(value: object) -> list[object]: + assert isinstance(value, list) + return value + + +class Color(Enum): + RED = "red" + BLUE = "blue" + + +@runtime_checkable +class Named(Protocol): + @property + def name(self) -> str: ... + + +@export_config +class Point: + def __init__(self, x: int, y: int = 0) -> None: + self.x = x + self.y = y + + +@export_config +class Box: + def __init__(self, origin: Point, label: str = "box") -> None: + self.origin = origin + self.label = label + + +@export_config +class Nest: + def __init__(self, child: Nest | None = None) -> None: + self.child = child + + +@export_config +class BadInner: + def __init__(self, fn: object) -> None: + self.fn = fn + + +@export_config +class Outer: + def __init__(self, child: BadInner) -> None: + self.child = child + + +@export_config +class BaseWidget: + def __init__(self, name: str) -> None: + self.name = name + + +@export_config +class DerivedWidget(BaseWidget): + def __init__(self, name: str, size: int) -> None: + super().__init__(name) + self.size = size + + +class UndecoratedOverride(BaseWidget): + def __init__(self, name: str, extra: int) -> None: + super().__init__(name) + self.extra = extra + + +class InheritsDecorated(BaseWidget): + """Subclass that inherits the decorated constructor unchanged.""" + + +@export_config +class WithExtras: + def __init__(self, base: int, **kwargs: object) -> None: + self.base = base + self.kwargs = kwargs + + +@export_config(scalar_var_kwargs=True) +class ScalarVarKwargs: + def __init__(self, base: int, **kwargs: object) -> None: + self.base = base + self.kwargs = kwargs + + +@export_config +class Leaf: + def __init__(self, value: int) -> None: + self.value = value + + +@export_config(config_args=("recipe",)) +class Holder: + def __init__(self, recipe: ComponentConfig, eager: object) -> None: + self.recipe = recipe + self.eager = eager + + +@export_config +class PathHolder: + def __init__(self, path: str | Path) -> None: + self.path = path + + +@export_config +class EnumHolder: + def __init__(self, color: Color | str) -> None: + self.color = Color(color) if not isinstance(color, Color) else color + + +@export_config +class MappingHolder: + def __init__(self, data: Mapping[str, object]) -> None: + self.data = dict(data) + + +@export_config +class ListHolder: + def __init__(self, items: list[object]) -> None: + self.items = list(items) + + +@export_config +class OptionalName: + def __init__(self, name: str | None = "default") -> None: + self.name = name + + +@export_config +class CanonicalName: + normalize_calls = 0 + + def __init__(self, name: str = "default") -> None: + self.name = name + + @classmethod + def _physicalai_normalize_captured_init_args(cls, supplied: dict[str, object]) -> None: + cls.normalize_calls += 1 + if "name" in supplied: + supplied["name"] = str(supplied["name"]).lower() + + +@export_config +class Boom: + def __init__(self, x: int) -> None: + msg = "nope" + raise RuntimeError(msg) + + +@export_config +class CtorBoom: + def __init__(self, x: int) -> None: + msg = "boom" + raise ValueError(msg) + + +class DomainPayload: + """Domain value that encodes via ``to_config_value()`` (not a component).""" + + def __init__(self, amount: int) -> None: + self.amount = amount + + def to_config_value(self) -> dict[str, int]: + return {"amount": self.amount} + + +class BadNanDomain: + def to_config_value(self) -> dict[str, float]: + return {"x": math.nan} + + +class BadReservedDomain: + def to_config_value(self) -> dict[str, object]: + return {"class_path": "not.a.component", "other": 1} + + +class NullDomain: + def to_config_value(self) -> None: + return None + + +class CodecPeer: + """Domain value that can point at another domain value for cycle tests.""" + + def __init__(self) -> None: + self.other: object | None = None + + def to_config_value(self) -> object: + return self.other + + +@export_config +class DomainHolder: + def __init__(self, payload: object) -> None: + self.payload = payload + + +@export_config(class_path="tests.unit.config.test_component_config.ExportAlias") +class _HiddenExport: + """Defining-module class with a public re-export alias (see ExportAlias).""" + + def __init__(self, x: int) -> None: + self.x = x + + +ExportAlias = _HiddenExport + + +class InheritsAliasedExport(_HiddenExport): + """Inherits decorated constructor; must not inherit ``class_path=`` override.""" + + +class TestImportDottedPath: + def test_resolves_nested_class(self) -> None: + assert import_dotted_path("tests.unit.config.test_component_config.Point") is Point + + +class TestNormalizeAndInstantiate: + def test_primitives_round_trip(self) -> None: + point = Point(1, y=2) + config = to_config(point) + wire = json.loads(json.dumps(config)) + restored = instantiate(wire) + assert isinstance(restored, Point) + assert restored.x == 1 + assert restored.y == 2 + assert to_config(restored) == wire + + def test_omitted_defaults_stay_omitted(self) -> None: + point = Point(3) + config = to_config(point) + assert config["init_args"] == {"x": 3} + restored = cast(Point, instantiate(config)) + assert restored.y == 0 + + def test_explicit_none_is_preserved(self) -> None: + obj = OptionalName(None) + config = to_config(obj) + assert config["init_args"] == {"name": None} + restored = cast(OptionalName, instantiate(config)) + assert restored.name is None + + def test_path_as_given(self) -> None: + relative = PathHolder(Path("calib.json")) + absolute = PathHolder(Path("/var/calib.json")) + assert to_config(relative)["init_args"]["path"] == "calib.json" + assert to_config(absolute)["init_args"]["path"] == "/var/calib.json" + str_relative = PathHolder("./relative.json") + assert to_config(str_relative)["init_args"]["path"] == "./relative.json" + + def test_enum_value(self) -> None: + holder = EnumHolder(Color.RED) + config = to_config(holder) + assert config["init_args"]["color"] == "red" + restored = cast(EnumHolder, instantiate(config)) + assert restored.color is Color.RED + + def test_non_finite_float_rejected(self) -> None: + holder = MappingHolder({"x": math.nan}) + with pytest.raises(ComponentConfigError, match="non-finite"): + to_config(holder) + + def test_nested_component(self) -> None: + box = Box(Point(1, 2), label="b") + config = to_config(box) + origin = _as_mapping(config["init_args"]["origin"]) + assert cast(str, origin["class_path"]).endswith(".Point") + assert origin["init_args"] == {"x": 1, "y": 2} + restored = cast(Box, instantiate(json.loads(json.dumps(config)))) + assert restored.origin.x == 1 + assert to_config(restored) == json.loads(json.dumps(config)) + + def test_list_and_mapping(self) -> None: + holder = ListHolder([Point(1), {"a": 1}]) + config = to_config(holder) + items = _as_list(config["init_args"]["items"]) + first = _as_mapping(items[0]) + assert _as_mapping(first["init_args"])["x"] == 1 + assert items[1] == {"a": 1} + restored = cast(ListHolder, instantiate(config)) + assert isinstance(restored.items[0], Point) + + def test_mutable_container_snapshot(self) -> None: + data: dict[str, object] = {"a": 1} + holder = MappingHolder(data) + data["a"] = 99 + assert _as_mapping(to_config(holder)["init_args"]["data"])["a"] == 1 + + def test_cyclic_mapping_rejected(self) -> None: + data: dict[str, object] = {} + data["self"] = data + holder = MappingHolder(data) + with pytest.raises(ComponentConfigError, match="cyclic"): + to_config(holder) + + def test_depth_limit_on_mappings(self) -> None: + nested: dict[str, object] = {"leaf": 1} + for _ in range(_MAX_CONFIG_DEPTH + 2): + nested = {"child": nested} + holder = MappingHolder(nested) + with pytest.raises(ComponentConfigError, match="nesting depth"): + to_config(holder) + + def test_nested_component_depth_symmetric(self) -> None: + """Export and instantiate share the same nested-component depth limit. + + A Nest chain of length ``_MAX_CONFIG_DEPTH`` (depths 0..MAX-1 for + components, leaf ``None`` at depth MAX) must round-trip. Length + ``_MAX_CONFIG_DEPTH + 1`` must fail on both sides. + """ + + def nest_chain(length: int) -> Nest: + node: Nest | None = None + for _ in range(length): + node = Nest(node) + assert node is not None + return node + + allowed = nest_chain(_MAX_CONFIG_DEPTH) + config = to_config(allowed) + restored = instantiate(config) + assert isinstance(restored, Nest) + assert to_config(restored) == config + + too_deep = nest_chain(_MAX_CONFIG_DEPTH + 1) + with pytest.raises(ComponentConfigError, match="nesting depth"): + to_config(too_deep) + + deeper: ComponentConfig = { + "class_path": "tests.unit.config.test_component_config.Nest", + "init_args": cast("dict[str, JsonValue]", {"child": config}), + } + with pytest.raises(ComponentConfigError, match="nesting depth"): + instantiate(deeper) + + def test_nested_error_path_includes_parent(self) -> None: + outer = Outer(BadInner(lambda: None)) + with pytest.raises(ComponentConfigError, match=r"Outer\.init_args\.child\.init_args\.fn"): + to_config(outer) + + def test_malformed_nested_config_before_import(self) -> None: + with pytest.raises(ComponentConfigError, match="unexpected keys"): + instantiate({ + "class_path": "tests.unit.config.test_component_config.Point", + "init_args": {}, + "extra": 1, + }) + + def test_null_init_args_rejected(self) -> None: + with pytest.raises(ComponentConfigError, match="init_args"): + instantiate({ + "class_path": "tests.unit.config.test_component_config.Point", + "init_args": None, + }) + + def test_missing_class_path_rejected(self) -> None: + with pytest.raises(ComponentConfigError, match="missing required"): + instantiate({"init_args": {}}) # type: ignore[arg-type] + + def test_non_class_import_target(self) -> None: + with pytest.raises(ComponentImportError, match="does not resolve to a class"): + instantiate({"class_path": "os.path.join", "init_args": {}}) + + def test_unimportable_class_path(self) -> None: + with pytest.raises(ComponentImportError, match="cannot import"): + instantiate({"class_path": "totally.unknown.module.Cls", "init_args": {}}) + + def test_dict_with_class_path_is_reserved(self) -> None: + holder = MappingHolder({"class_path": "not.a.component", "other": 1}) + with pytest.raises(ComponentConfigError, match="to_config_value"): + to_config(holder) + + def test_unsupported_object_reports_path(self) -> None: + holder = MappingHolder({"fn": lambda: None}) + with pytest.raises(ComponentConfigError, match=r"init_args\.data\.fn"): + to_config(holder) + + def test_local_class_export_fails(self) -> None: + @export_config + class LocalPoint: + def __init__(self, x: int) -> None: + self.x = x + + obj = LocalPoint(1) + with pytest.raises(ComponentConfigError, match=""): + to_config(obj) + + def test_local_class_instantiate_fails(self) -> None: + with pytest.raises(ComponentConfigError, match=""): + instantiate({ + "class_path": "tests.unit.config.test_component_config.Local..X", + "init_args": {}, + }) + + def test_constructor_failure_propagates_with_note(self) -> None: + with pytest.raises(ValueError, match="boom") as info: + instantiate({ + "class_path": "tests.unit.config.test_component_config.CtorBoom", + "init_args": {"x": 1}, + }) + notes = getattr(info.value, "__notes__", []) + assert any("constructor failed" in note for note in notes) + + def test_malformed_nested_config_rejected_before_any_import(self) -> None: + config = { + "class_path": "tests.unit.config.test_component_config.Point", + "init_args": { + "x": { + "class_path": "tests.unit.config.test_component_config.Point", + "init_args": None, + }, + }, + } + with ( + patch("physicalai.config._instantiate.import_dotted_path") as import_path, + pytest.raises(ComponentConfigError, match="init_args"), + ): + instantiate(config) # type: ignore[arg-type] + import_path.assert_not_called() + + @pytest.mark.parametrize("value", [math.nan, math.inf, (1, 2), Path("config.json"), Color.RED, object()]) + def test_non_json_value_rejected_before_any_import(self, value: object) -> None: + config = { + "class_path": "tests.unit.config.test_component_config.DomainHolder", + "init_args": {"payload": value}, + } + with ( + patch("physicalai.config._instantiate.import_dotted_path") as import_path, + pytest.raises(ComponentConfigError), + ): + instantiate(config) # type: ignore[arg-type] + import_path.assert_not_called() + + def test_non_string_plain_mapping_key_rejected_before_any_import(self) -> None: + config = { + "class_path": "tests.unit.config.test_component_config.MappingHolder", + "init_args": {"data": {1: "value"}}, + } + with ( + patch("physicalai.config._instantiate.import_dotted_path") as import_path, + pytest.raises(ComponentConfigError, match="mapping keys must be strings"), + ): + instantiate(config) # type: ignore[arg-type] + import_path.assert_not_called() + + def test_cyclic_component_config_rejected_before_any_import(self) -> None: + config: dict[str, object] = { + "class_path": "tests.unit.config.test_component_config.Nest", + "init_args": {}, + } + cast("dict[str, object]", config["init_args"])["child"] = config + with ( + patch("physicalai.config._instantiate.import_dotted_path") as import_path, + pytest.raises(ComponentConfigError, match="cyclic component config"), + ): + instantiate(config) # type: ignore[arg-type] + import_path.assert_not_called() + + def test_plain_nested_mapping_remains_valid(self) -> None: + config = cast("ComponentConfig", { + "class_path": "tests.unit.config.test_component_config.MappingHolder", + "init_args": {"data": {"nested": {"value": 1}}}, + }) + restored = cast(MappingHolder, instantiate(config)) + assert restored.data == {"nested": {"value": 1}} + + +class TestExportConfig: + def test_positional_binds_to_names(self) -> None: + point = Point(4, 5) + assert to_config(point)["init_args"] == {"x": 4, "y": 5} + + def test_kwargs_flatten(self) -> None: + obj = WithExtras(1, color="red", count=2) + assert to_config(obj)["init_args"] == {"base": 1, "color": "red", "count": 2} + + def test_rejects_var_positional(self) -> None: + with pytest.raises(TypeError, match=r"\*args"): + + @export_config + class Bad: + def __init__(self, *args: object) -> None: + self.args = args + + def test_rejects_positional_only(self) -> None: + with pytest.raises(TypeError, match="positional-only"): + + @export_config + class BadPosOnly: + def __init__(self, x: int, /) -> None: + self.x = x + + def test_rejects_decorating_subclass_without_own_init(self) -> None: + with pytest.raises(TypeError, match="define their own __init__"): + + @export_config + class Redundant(BaseWidget): + pass + + def test_outermost_super_wins(self) -> None: + widget = DerivedWidget("w", 10) + config = to_config(widget) + assert config["class_path"].endswith(".DerivedWidget") + assert config["init_args"] == {"name": "w", "size": 10} + + def test_undecorated_override_fails(self) -> None: + obj = UndecoratedOverride("n", 1) + assert not is_config_exportable(obj) + with pytest.raises(ComponentConfigError, match="not config-exportable"): + to_config(obj) + + def test_inherited_decorated_constructor(self) -> None: + obj = InheritsDecorated("ok") + assert is_config_exportable(obj) + config = to_config(obj) + assert config["class_path"].endswith(".InheritsDecorated") + assert config["init_args"] == {"name": "ok"} + + def test_failed_constructor_does_not_capture(self) -> None: + with pytest.raises(RuntimeError, match="nope"): + Boom(1) + + def test_explicit_class_path_override(self) -> None: + obj = _HiddenExport(3) + assert is_config_exportable(obj) + config = to_config(obj) + assert config["class_path"] == "tests.unit.config.test_component_config.ExportAlias" + assert config["init_args"] == {"x": 3} + restored = instantiate(config) + assert type(restored) is _HiddenExport + assert restored.x == 3 # type: ignore[union-attr] + + def test_inherited_class_path_override_does_not_leak(self) -> None: + obj = InheritsAliasedExport(9) + assert is_config_exportable(obj) + config = to_config(obj) + assert config["class_path"] == ( + "tests.unit.config.test_component_config.InheritsAliasedExport" + ) + assert config["init_args"] == {"x": 9} + restored = instantiate(config) + assert type(restored) is InheritsAliasedExport + + def test_domain_value_hook_encodes_to_json(self) -> None: + holder = DomainHolder(DomainPayload(42)) + config = to_config(holder) + assert config["init_args"]["payload"] == {"amount": 42} + wire = json.loads(json.dumps(config)) + restored = cast(DomainHolder, instantiate(wire)) + assert restored.payload == {"amount": 42} + assert to_config(restored) == wire + + def test_domain_value_none_is_json_null(self) -> None: + holder = DomainHolder(NullDomain()) + config = to_config(holder) + assert config["init_args"]["payload"] is None + wire = json.loads(json.dumps(config)) + restored = cast(DomainHolder, instantiate(wire)) + assert restored.payload is None + + def test_domain_value_codec_cycle_raises(self) -> None: + left = CodecPeer() + right = CodecPeer() + left.other = right + right.other = left + holder = DomainHolder(left) + with pytest.raises(ComponentConfigError, match="cyclic to_config_value"): + to_config(holder) + + def test_domain_value_hook_output_is_renormalized(self) -> None: + holder = DomainHolder(BadNanDomain()) + with pytest.raises(ComponentConfigError, match="non-finite"): + to_config(holder) + + def test_domain_value_hook_reserved_class_path_validated(self) -> None: + holder = DomainHolder(BadReservedDomain()) + with pytest.raises(ComponentConfigError, match="to_config_value"): + to_config(holder) + + def test_instance_to_config_sugar(self) -> None: + point = Point(1, 2) + assert point.to_config() == to_config(point) # type: ignore[attr-defined] + + def test_injected_to_config_does_not_break_protocol(self) -> None: + widget = BaseWidget("ok") + assert isinstance(widget, Named) + assert widget.to_config()["init_args"] == {"name": "ok"} # type: ignore[attr-defined] + + def test_signature_preserved(self) -> None: + sig = inspect.signature(Point.__init__) + assert list(sig.parameters) == ["self", "x", "y"] + + def test_depth_attr_cleaned_after_construction(self) -> None: + point = Point(1) + assert is_config_exportable(point) + assert "_physicalai_export_config_depth" not in vars(point) + + def test_private_capture_normalizer_canonicalizes_supplied_args_only(self) -> None: + CanonicalName.normalize_calls = 0 + explicit = CanonicalName("LOUD") + omitted = CanonicalName() + assert to_config(explicit)["init_args"] == {"name": "loud"} + assert to_config(omitted)["init_args"] == {} + assert CanonicalName.normalize_calls == 2 + + +class TestScalarVarKwargs: + def test_scalar_var_kwargs_round_trip(self) -> None: + obj = ScalarVarKwargs(1, count=2, flag=True, label="x", missing=None) + config = to_config(obj) + assert config["init_args"] == { + "base": 1, + "count": 2, + "flag": True, + "label": "x", + "missing": None, + } + restored = cast(ScalarVarKwargs, instantiate(json.loads(json.dumps(config)))) + assert restored.base == 1 + assert restored.kwargs == {"count": 2, "flag": True, "label": "x", "missing": None} + + def test_non_scalar_dict_var_kwarg_fails(self) -> None: + obj = ScalarVarKwargs(1, config_blob={"a": 1}) + with pytest.raises(ComponentConfigError, match=r"init_args\.config_blob"): + to_config(obj) + + def test_non_scalar_list_var_kwarg_fails(self) -> None: + obj = ScalarVarKwargs(1, tags=["x", "y"]) + with pytest.raises(ComponentConfigError, match=r"init_args\.tags"): + to_config(obj) + + def test_named_mapping_still_exports_without_scalar_flag(self) -> None: + # Default **kwargs flattening still accepts nested JSON. + obj = WithExtras(1, nested={"a": 1}) + assert to_config(obj)["init_args"]["nested"] == {"a": 1} + + def test_scalar_var_kwargs_requires_var_keyword(self) -> None: + with pytest.raises(TypeError, match="scalar_var_kwargs=True requires"): + + @export_config(scalar_var_kwargs=True) + class NoVarKwargs: + def __init__(self, x: int) -> None: + self.x = x + + +class TestConfigArgs: + def test_declared_config_arg_is_not_instantiated(self) -> None: + config: ComponentConfig = { + "class_path": f"{__name__}.Holder", + "init_args": { + "recipe": {"class_path": f"{__name__}.Leaf", "init_args": {"value": 3}}, + "eager": {"class_path": f"{__name__}.Leaf", "init_args": {"value": 3}}, + }, + } + holder = cast(Holder, instantiate(config)) + assert holder.recipe == config["init_args"]["recipe"] + assert isinstance(holder.eager, Leaf) + + def test_declared_config_arg_round_trips(self) -> None: + recipe: ComponentConfig = { + "class_path": f"{__name__}.Leaf", + "init_args": {"value": 3}, + } + holder = Holder(recipe=recipe, eager=Leaf(value=3)) + config = to_config(holder) + assert config["init_args"]["recipe"] == recipe + wire = json.loads(json.dumps(config)) + restored = cast(Holder, instantiate(wire)) + assert to_config(restored) == wire + + def test_unknown_config_arg_name_rejected(self) -> None: + with pytest.raises(TypeError, match="config_args 'missing' is not an __init__ parameter"): + + @export_config(config_args=("missing",)) + class Bad: + def __init__(self, x: int) -> None: + self.x = x + + def test_config_args_cannot_name_var_keyword(self) -> None: + with pytest.raises(TypeError, match=r"config_args cannot name the \*\*kwargs parameter"): + + @export_config(config_args=("kwargs",)) + class BadVarKw: + def __init__(self, **kwargs: object) -> None: + self.kwargs = kwargs diff --git a/tests/unit/config/test_yaml.py b/tests/unit/config/test_yaml.py new file mode 100644 index 00000000..cea53dc8 --- /dev/null +++ b/tests/unit/config/test_yaml.py @@ -0,0 +1,86 @@ +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# ruff:file-ignore[undocumented-public-module, undocumented-public-class, undocumented-public-method, undocumented-public-init, undocumented-magic-method, magic-value-comparison, no-self-use, assert] + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +import yaml + +from physicalai.config import ( + ComponentConfigError, + export_config, + instantiate, + load_yaml, + save_yaml, + to_config, + to_yaml, +) + +if TYPE_CHECKING: + from pathlib import Path + + +@export_config +class Gadget: + def __init__(self, size: int, label: str = "gadget") -> None: + self.size = size + self.label = label + + +@export_config +class Holder: + def __init__(self, gadget: Gadget, note: str | None = None) -> None: + self.gadget = gadget + self.note = note + + +class TestToYaml: + def test_live_component_round_trips_through_yaml(self) -> None: + holder = Holder(Gadget(3, label="inner"), note="hi") + + text = to_yaml(holder) + rebuilt = instantiate(yaml.safe_load(text)) + + assert isinstance(rebuilt, Holder) + assert rebuilt.note == "hi" + assert rebuilt.gadget.size == 3 + assert rebuilt.gadget.label == "inner" + + def test_accepts_existing_component_config_mapping(self) -> None: + config = to_config(Gadget(7)) + + text = to_yaml(config) + loaded = yaml.safe_load(text) + + assert loaded == {"class_path": f"{__name__}.Gadget", "init_args": {"size": 7}} + + def test_rejects_malformed_mapping(self) -> None: + with pytest.raises(ComponentConfigError, match="class_path"): + to_yaml({"init_args": {"size": 1}}) + + def test_rejects_non_exportable_object(self) -> None: + with pytest.raises(ComponentConfigError): + to_yaml(object()) + + +class TestSaveLoadYaml: + def test_save_then_load_then_instantiate(self, tmp_path: Path) -> None: + target = tmp_path / "gadget.yaml" + save_yaml(Gadget(5), target) + + loaded = load_yaml(target) + rebuilt = instantiate(loaded) + + assert isinstance(rebuilt, Gadget) + assert rebuilt.size == 5 + assert rebuilt.label == "gadget" + + def test_load_rejects_non_mapping_document(self, tmp_path: Path) -> None: + target = tmp_path / "list.yaml" + target.write_text("- 1\n- 2\n", encoding="utf-8") + + with pytest.raises(ComponentConfigError, match="must be a mapping"): + load_yaml(target) From 2efb4ee27ce4064174b08d3f58d509d9cbdd5347 Mon Sep 17 00:00:00 2001 From: Max Xiang Date: Mon, 27 Jul 2026 16:55:26 +0200 Subject: [PATCH 2/2] address copilot comments --- src/physicalai/config/_envelope.py | 4 +- src/physicalai/config/importing.py | 15 ++++-- tests/unit/config/test_component_config.py | 56 ++++++++++++++++++++++ 3 files changed, 70 insertions(+), 5 deletions(-) diff --git a/src/physicalai/config/_envelope.py b/src/physicalai/config/_envelope.py index 8d9dc0a4..3e93fd33 100644 --- a/src/physicalai/config/_envelope.py +++ b/src/physicalai/config/_envelope.py @@ -119,8 +119,8 @@ def normalize_component_config( raise ValueError(msg) init_args = validated["init_args"] try: - json.dumps({"class_path": class_path, "init_args": init_args}) - except TypeError as exc: + json.dumps({"class_path": class_path, "init_args": init_args}, allow_nan=False) + except (TypeError, ValueError) as exc: msg = f"{component_key}.init_args must be JSON-serializable{json_hint}: {exc}" raise ValueError(msg) from exc return {"class_path": class_path, "init_args": dict(init_args)} diff --git a/src/physicalai/config/importing.py b/src/physicalai/config/importing.py index bb49b720..4fe3b92b 100644 --- a/src/physicalai/config/importing.py +++ b/src/physicalai/config/importing.py @@ -19,6 +19,9 @@ def import_dotted_path(path: str) -> object: Raises: ValueError: If no module prefix can be imported, or *path* has no ``.``. + ModuleNotFoundError: If an existing module prefix fails because a + dependency is missing (not merely because a longer prefix is not a + module). """ if "." not in path: msg = f"dotted path must contain at least one '.': {path!r}" @@ -26,10 +29,16 @@ def import_dotted_path(path: str) -> object: segments = path.split(".") for split_index in range(len(segments), 0, -1): + module_name = ".".join(segments[:split_index]) try: - obj: object = importlib.import_module(".".join(segments[:split_index])) - except ImportError: - continue + obj: object = importlib.import_module(module_name) + except ModuleNotFoundError as exc: + # Continue only when this prefix itself is missing — not when a real + # module fails because an inner dependency is missing. + missing = exc.name + if missing is not None and (module_name == missing or module_name.startswith(f"{missing}.")): + continue + raise for attr in segments[split_index:]: obj = getattr(obj, attr) return obj diff --git a/tests/unit/config/test_component_config.py b/tests/unit/config/test_component_config.py index 2ceb6cef..169e4193 100644 --- a/tests/unit/config/test_component_config.py +++ b/tests/unit/config/test_component_config.py @@ -7,6 +7,7 @@ import inspect import json import math +import sys from collections.abc import Mapping from enum import Enum from pathlib import Path @@ -24,6 +25,7 @@ import_dotted_path, instantiate, is_config_exportable, + normalize_component_config, to_config, ) @@ -252,6 +254,60 @@ class TestImportDottedPath: def test_resolves_nested_class(self) -> None: assert import_dotted_path("tests.unit.config.test_component_config.Point") is Point + def test_real_import_failures_are_not_masked(self, tmp_path: Path) -> None: + pkg = tmp_path / "mask_pkg" + pkg.mkdir() + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "broken.py").write_text( + "import totally_missing_dep_xyz\n\nclass Robot:\n pass\n", + encoding="utf-8", + ) + (pkg / "raises_ie.py").write_text( + "raise ImportError('boom from module body')\n", + encoding="utf-8", + ) + sys.path.insert(0, str(tmp_path)) + try: + with pytest.raises(ModuleNotFoundError, match="totally_missing_dep_xyz"): + import_dotted_path("mask_pkg.broken.Robot") + with pytest.raises(ImportError, match="boom from module body"): + import_dotted_path("mask_pkg.raises_ie.Robot") + finally: + sys.path.remove(str(tmp_path)) + sys.modules.pop("mask_pkg.broken", None) + sys.modules.pop("mask_pkg.raises_ie", None) + sys.modules.pop("mask_pkg", None) + + def test_unimportable_prefix_raises(self) -> None: + with pytest.raises(ValueError, match="could not import"): + import_dotted_path("totally.unknown.module.Cls") + + +class TestNormalizeComponentConfig: + def test_rejects_nan(self) -> None: + with pytest.raises(ValueError, match="JSON-serializable"): + normalize_component_config( + {"class_path": f"{__name__}.Point", "init_args": {"x": math.nan}}, + component_key="robot", + class_label="robot_class", + ) + + def test_rejects_infinity(self) -> None: + with pytest.raises(ValueError, match="JSON-serializable"): + normalize_component_config( + {"class_path": f"{__name__}.Point", "init_args": {"x": float("inf")}}, + component_key="camera", + class_label="camera_class", + ) + + def test_rejects_non_serializable_object(self) -> None: + with pytest.raises(ValueError, match="JSON-serializable"): + normalize_component_config( + {"class_path": f"{__name__}.Point", "init_args": {"x": object()}}, + component_key="robot", + class_label="robot_class", + ) + class TestNormalizeAndInstantiate: def test_primitives_round_trip(self) -> None: