File size: 2,423 Bytes
1a12965 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | """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)
|