-
Notifications
You must be signed in to change notification settings - Fork 1
chore(config): move shared config primitives into runtime package #123
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
samet-akcay
wants to merge
3
commits into
main
Choose a base branch
from
feat/shared-config-runtime
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
+714
−109
Open
Changes from all commits
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| # Copyright (C) 2025-2026 Intel Corporation | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| """Configuration primitives shared by runtime and training packages.""" | ||
|
|
||
| from physicalai.config.base import Config | ||
| from physicalai.config.component import ComponentSpec | ||
| from physicalai.config.instantiate import instantiate_obj | ||
| from physicalai.config.mixin import FromConfig, from_config | ||
|
|
||
| __all__ = ["ComponentSpec", "Config", "FromConfig", "from_config", "instantiate_obj"] |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| # Copyright (C) 2025 Intel Corporation | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| # ruff: noqa: DOC201, DOC501 | ||
|
|
||
| """Base configuration class for typed constructor configs.""" | ||
|
|
||
| import dataclasses | ||
| from collections.abc import Mapping | ||
| from pathlib import Path | ||
| from typing import Any, Literal, Self | ||
|
|
||
| from physicalai.config.serializable import dataclass_to_dict, dict_to_dataclass | ||
|
|
||
| __all__ = ["Config"] | ||
|
|
||
|
|
||
| class Config: | ||
| """Base class for dataclass-backed configuration objects.""" | ||
|
|
||
| def to_dict(self) -> dict[str, Any]: | ||
| """Convert this config to a plain dict for serialization.""" | ||
| if not dataclasses.is_dataclass(self): | ||
| msg = f"{self.__class__.__name__} must be a dataclass to use Config" | ||
| raise TypeError(msg) | ||
|
|
||
| result = dataclass_to_dict(self) | ||
| if not isinstance(result, dict): | ||
| msg = f"Expected dict from dataclass_to_dict, got {type(result)}" | ||
| raise TypeError(msg) | ||
| return result | ||
|
|
||
| @classmethod | ||
| def from_dict(cls, data: Mapping[str, Any]) -> Self: | ||
| """Reconstruct this config from a dict.""" | ||
| if not dataclasses.is_dataclass(cls): | ||
| msg = f"{cls.__name__} must be a dataclass to use Config" | ||
| raise TypeError(msg) | ||
| return dict_to_dataclass(cls, data) | ||
|
|
||
| def to_jsonargparse(self) -> dict[str, Any]: | ||
| """Convert config to ``class_path``/``init_args`` format.""" | ||
| return { | ||
| "class_path": f"{self.__class__.__module__}.{self.__class__.__qualname__}", | ||
| "init_args": self.to_dict(), | ||
| } | ||
|
|
||
| def save( | ||
| self, | ||
| path: str | Path, | ||
| *, | ||
| format: Literal["jsonargparse", "dict"] = "jsonargparse", # noqa: A002 | ||
| ) -> None: | ||
| """Save config to a YAML file.""" | ||
| path = Path(path) | ||
| data = self.to_dict() if format == "dict" else self.to_jsonargparse() | ||
|
|
||
| if path.suffix not in {".yaml", ".yml"}: | ||
| msg = f"Unsupported file extension: {path.suffix}. Use .yaml or .yml" | ||
| raise ValueError(msg) | ||
|
|
||
| import yaml # noqa: PLC0415 | ||
|
|
||
| with path.open("w") as f: | ||
| yaml.safe_dump(data, f, default_flow_style=False, sort_keys=False) | ||
|
|
||
| @classmethod | ||
| def load(cls, path: str | Path) -> Self: | ||
| """Load config from a YAML file.""" | ||
| path = Path(path) | ||
|
|
||
| if path.suffix not in {".yaml", ".yml"}: | ||
| msg = f"Unsupported file extension: {path.suffix}. Use .yaml or .yml" | ||
| raise ValueError(msg) | ||
|
|
||
| import yaml # noqa: PLC0415 | ||
|
|
||
| with path.open() as f: | ||
| data = yaml.safe_load(f) | ||
|
|
||
| if data is None: | ||
| data = {} | ||
| if not isinstance(data, Mapping): | ||
| msg = f"Expected YAML root to be a mapping, got {type(data).__name__}" | ||
| raise TypeError(msg) | ||
|
|
||
| if "init_args" in data: | ||
| data = data["init_args"] | ||
| if data is None: | ||
| data = {} | ||
| if not isinstance(data, Mapping): | ||
| msg = f"Expected 'init_args' to be a mapping, got {type(data).__name__}" | ||
| raise TypeError(msg) | ||
|
|
||
| return cls.from_dict(data) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| # Copyright (C) 2026 Intel Corporation | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| # ruff: noqa: DOC201, DOC501 | ||
|
|
||
| """Generic component specifications for dynamic instantiation.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import inspect | ||
| from typing import Any | ||
|
|
||
| from pydantic import BaseModel, ConfigDict, Field, model_validator | ||
|
|
||
| # Alias builtin ``type`` so it remains accessible inside classes that define a | ||
| # Pydantic field with the same name (e.g. ``ComponentSpec.type``). | ||
| _type = type | ||
|
|
||
|
|
||
| class ComponentSpec(BaseModel): | ||
| """Dual-resolution component descriptor for dynamic instantiation. | ||
|
|
||
| Supports two resolution modes: | ||
|
|
||
| 1. **type + flat params** (LeRobot-compatible):: | ||
|
|
||
| {"type": "single_pass"} | ||
|
|
||
| 2. **class_path + init_args** (full-power PhysicalAI):: | ||
|
|
||
| {"class_path": "physicalai.inference.runners.SinglePass", | ||
| "init_args": {}} | ||
|
|
||
| When ``class_path`` is present it takes precedence. When only ``type`` is | ||
| present, a component registry can resolve it. | ||
|
|
||
| Attributes: | ||
| type: Registered short name (e.g. ``"single_pass"``). | ||
| class_path: Fully-qualified class path for direct import. | ||
| init_args: Keyword arguments forwarded to the constructor | ||
| (used with ``class_path`` mode). | ||
| """ | ||
|
|
||
| model_config = ConfigDict(frozen=True, extra="allow") | ||
| type: str = "" | ||
| class_path: str = "" | ||
| init_args: dict[str, Any] = Field(default_factory=dict) | ||
|
|
||
| @model_validator(mode="after") | ||
| def _must_have_type_or_class_path(self) -> ComponentSpec: | ||
| if not self.type and not self.class_path: | ||
| msg = "ComponentSpec requires either 'type' or 'class_path'" | ||
| raise ValueError(msg) | ||
| return self | ||
|
|
||
| @property | ||
| def flat_params(self) -> dict[str, Any]: | ||
| """Return extra fields as flat params for type-based resolution.""" | ||
| return dict(self.model_extra) if self.model_extra else {} | ||
|
|
||
| @classmethod | ||
| def from_class(cls, target: _type, **overrides: Any) -> ComponentSpec: # noqa: ANN401 | ||
| """Build a spec by introspecting a class constructor. | ||
|
|
||
| Parameters not present in *overrides* use their default values. Required | ||
| parameters without defaults must be provided in *overrides* or a | ||
| TypeError is raised. | ||
| """ | ||
| sig = inspect.signature(target) | ||
| init_args: dict[str, Any] = {} | ||
| missing: list[str] = [] | ||
|
|
||
| for name, param in sig.parameters.items(): | ||
| if name == "self": | ||
| continue | ||
| if name in overrides: | ||
| value = overrides[name] | ||
| elif param.default is not param.empty: | ||
| value = param.default | ||
| else: | ||
| missing.append(name) | ||
| continue | ||
|
|
||
| if isinstance(value, ComponentSpec): | ||
| value = value.model_dump() | ||
| init_args[name] = value | ||
|
|
||
| if missing: | ||
| msg = ( | ||
| f"Missing required parameters for {target.__qualname__}: " | ||
| f"{', '.join(missing)}. Pass them as keyword arguments." | ||
| ) | ||
| raise TypeError(msg) | ||
|
|
||
| return cls( | ||
| class_path=f"{target.__module__}.{target.__qualname__}", | ||
| init_args=init_args, | ||
| ) | ||
|
|
||
|
|
||
| __all__ = ["ComponentSpec"] |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| # Copyright (C) 2025 Intel Corporation | ||
| # SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| # ruff: noqa: DOC201, DOC501 | ||
|
|
||
| """Configuration instantiation helpers.""" | ||
|
|
||
| import dataclasses | ||
| import importlib | ||
| from collections.abc import Mapping | ||
| from pathlib import Path | ||
| from typing import TYPE_CHECKING | ||
|
|
||
| import yaml | ||
| from pydantic import BaseModel | ||
|
|
||
| if TYPE_CHECKING: | ||
| from typing import Any | ||
|
|
||
|
|
||
| ConfigMapping = Mapping[str, "Any"] | ||
|
|
||
|
|
||
| def _import_class(class_path: str) -> type: | ||
| """Import a class from a module path.""" | ||
| try: | ||
| module_path, class_name = class_path.rsplit(".", 1) | ||
| module = importlib.import_module(module_path) # nosemgrep | ||
| return getattr(module, class_name) | ||
| except (ValueError, ImportError, AttributeError) as e: | ||
| msg = f"Cannot import '{class_path}': {e}" | ||
| raise ImportError(msg) from e | ||
|
|
||
|
|
||
| def _instantiate_recursive(value: "Any") -> "Any": # noqa: ANN401 | ||
| """Walk a value and instantiate nested ``{class_path, init_args}`` dicts.""" | ||
| if isinstance(value, dict): | ||
| if "class_path" in value: | ||
| return instantiate_obj_from_dict(value) | ||
| return {k: _instantiate_recursive(v) for k, v in value.items()} | ||
| if isinstance(value, list): | ||
| return [_instantiate_recursive(item) for item in value] | ||
| if isinstance(value, tuple): | ||
| return tuple(_instantiate_recursive(item) for item in value) | ||
| return value | ||
|
|
||
|
|
||
| def instantiate_obj_from_dict( | ||
| config: ConfigMapping, | ||
| *, | ||
| key: str | None = None, | ||
| target_cls: type | None = None, | ||
| ) -> object: | ||
| """Instantiate an object from a configuration dictionary.""" | ||
| if key is not None: | ||
| if key not in config: | ||
| msg = f"Configuration must contain '{key}' key. Got keys: {list(config.keys())}" | ||
| raise ValueError(msg) | ||
| config = config[key] | ||
| if not isinstance(config, Mapping): | ||
| msg = f"Configuration at key '{key}' must be a mapping, got {type(config).__name__}" | ||
| raise TypeError(msg) | ||
|
|
||
| if "class_path" in config: | ||
| cls = _import_class(config["class_path"]) | ||
| init_args = config.get("init_args", {}) | ||
| elif target_cls is not None: | ||
| cls = target_cls | ||
| init_args = config | ||
| else: | ||
| msg = ( | ||
| "Configuration must contain 'class_path' for instantiation, " | ||
| f"or pass target_cls explicitly. Got keys: {list(config.keys())}" | ||
| ) | ||
| raise ValueError(msg) | ||
|
|
||
| if not isinstance(init_args, dict): | ||
| return cls(init_args) | ||
|
|
||
| instantiated_args = {k: _instantiate_recursive(v) for k, v in init_args.items()} | ||
|
|
||
| if "args" in instantiated_args: | ||
| args = instantiated_args.pop("args") | ||
| return cls(*args, **instantiated_args) | ||
| return cls(**instantiated_args) | ||
|
|
||
|
samet-akcay marked this conversation as resolved.
|
||
|
|
||
| def instantiate_obj_from_pydantic( | ||
| config: BaseModel, | ||
| *, | ||
| key: str | None = None, | ||
| target_cls: type | None = None, | ||
| ) -> object: | ||
| """Instantiate an object from a Pydantic model.""" | ||
| return instantiate_obj_from_dict(config.model_dump(), key=key, target_cls=target_cls) | ||
|
|
||
|
|
||
| def instantiate_obj_from_dataclass( | ||
| config: object, | ||
| *, | ||
| key: str | None = None, | ||
| target_cls: type | None = None, | ||
| ) -> object: | ||
| """Instantiate an object from a dataclass instance.""" | ||
| if not dataclasses.is_dataclass(config) or isinstance(config, type): | ||
| msg = f"Expected dataclass instance, got {type(config)}" | ||
| raise TypeError(msg) | ||
|
|
||
| return instantiate_obj_from_dict(dataclasses.asdict(config), key=key, target_cls=target_cls) | ||
|
|
||
|
|
||
| def instantiate_obj_from_file( | ||
| file_path: str | Path, | ||
| *, | ||
| key: str | None = None, | ||
| target_cls: type | None = None, | ||
| ) -> object: | ||
| """Instantiate an object from a YAML/JSON configuration file.""" | ||
| with Path(file_path).open("r", encoding="utf-8") as f: | ||
| config = yaml.safe_load(f) | ||
| if config is None: | ||
| config = {} | ||
| if not isinstance(config, Mapping): | ||
| msg = f"Expected YAML root to be a mapping, got {type(config).__name__}" | ||
| raise TypeError(msg) | ||
| return instantiate_obj_from_dict(config, key=key, target_cls=target_cls) | ||
|
|
||
|
|
||
| def instantiate_obj( | ||
| config: ConfigMapping | BaseModel | object | str | Path, | ||
| *, | ||
| key: str | None = None, | ||
| target_cls: type | None = None, | ||
| ) -> object: | ||
| """Instantiate an object from dict, Pydantic, dataclass, or file config.""" | ||
| if isinstance(config, (str, Path)): | ||
| return instantiate_obj_from_file(config, key=key, target_cls=target_cls) | ||
| if isinstance(config, BaseModel): | ||
| return instantiate_obj_from_pydantic(config, key=key, target_cls=target_cls) | ||
| if dataclasses.is_dataclass(config) and not isinstance(config, type): | ||
| return instantiate_obj_from_dataclass(config, key=key, target_cls=target_cls) | ||
| if isinstance(config, dict): | ||
| return instantiate_obj_from_dict(config, key=key, target_cls=target_cls) | ||
|
|
||
| msg = f"Unsupported configuration type: {type(config)}. Expected dict, file path, Pydantic model, or dataclass." | ||
| raise TypeError(msg) | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.