Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions src/physicalai/config/__init__.py
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"]
95 changes: 95 additions & 0 deletions src/physicalai/config/base.py
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"]
Comment thread
samet-akcay marked this conversation as resolved.
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)
101 changes: 101 additions & 0 deletions src/physicalai/config/component.py
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"]
146 changes: 146 additions & 0 deletions src/physicalai/config/instantiate.py
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)

Comment thread
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)
Loading
Loading