Venket / src /config /loader.py
venki101's picture
Add files using upload-large-folder tool
1a12965 verified
Raw History Blame Contribute Delete
2.42 kB
"""OmegaConf-based configuration loading, merging, and Pydantic validation.
Supports:
* A base YAML file (typically ``configs/default.yaml``).
* An optional user YAML file merged on top of the base.
* Dot-list CLI overrides (``key.subkey=value``), e.g. as produced by
``OmegaConf.from_dotlist(["placement.rotation_deg.min=-5"])``.
The merged structure is converted to a plain Python container and validated
against :class:`~src.config.schema.PipelineConfig`, so any typo or
out-of-range value fails fast with a clear error instead of silently being
ignored.
"""
from __future__ import annotations
from pathlib import Path
from omegaconf import OmegaConf
from src.config.schema import PipelineConfig
def _default_config_path() -> Path:
return Path(__file__).resolve().parents[2] / "configs" / "default.yaml"
def load_config(
config_path: "str | Path | None" = None,
overrides: "list[str] | None" = None,
) -> PipelineConfig:
"""Load, merge, and validate the pipeline configuration.
Args:
config_path: Path to a user YAML config. If ``None``, only the
built-in ``configs/default.yaml`` is used.
overrides: Optional list of ``"dotted.key=value"`` CLI-style
overrides, applied last (highest precedence).
Returns:
A fully validated :class:`~src.config.schema.PipelineConfig`.
Raises:
pydantic.ValidationError: If the merged configuration is invalid.
FileNotFoundError: If ``config_path`` is given but does not exist.
"""
base = OmegaConf.load(_default_config_path())
merged = base
if config_path is not None:
config_path = Path(config_path)
if not config_path.exists():
raise FileNotFoundError(f"Config file not found: {config_path}")
user_conf = OmegaConf.load(config_path)
merged = OmegaConf.merge(base, user_conf)
if overrides:
cli_conf = OmegaConf.from_dotlist(overrides)
merged = OmegaConf.merge(merged, cli_conf)
resolved = OmegaConf.to_container(merged, resolve=True)
return PipelineConfig.model_validate(resolved)
def save_resolved_config(config: PipelineConfig, path: "str | Path") -> None:
"""Write a fully-resolved configuration (as actually used) to YAML for provenance."""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
OmegaConf.save(OmegaConf.create(config.model_dump()), path)