pxdepth/config.py

"""Load, compose, expand, and validate PXDepth JSON configurations.

The loader deliberately stays JSON based.  It adds only three conveniences that
are useful to external users: optional ``_base_`` composition, environment
variable expansion in strings, and optional module imports for custom registry
entries.  There is no framework-specific config object or runtime magic.
"""

import json
import os
from copy import deepcopy
from importlib import import_module
from pathlib import Path
from typing import Any, Dict, Iterable


def _merge(base: Any, update: Any) -> Any:
    """Recursively merge one configuration value into another.

    Args:
        base: Existing value inherited from a base config.
        update: Value from the child config. Dictionaries merge recursively,
            while lists and scalar values replace ``base`` completely.

    Returns:
        A deep-copied merged value. Neither input object is mutated.
    """
    if not isinstance(base, dict) or not isinstance(update, dict):
        return deepcopy(update)
    result = deepcopy(base)
    for key, value in update.items():
        result[key] = _merge(result[key], value) if key in result else deepcopy(value)
    return result


def _expand(value: Any) -> Any:
    """Expand filesystem shorthand throughout a nested config structure.

    Args:
        value: Arbitrarily nested dictionaries, lists, strings, and scalar
            values loaded from JSON.

    Returns:
        A matching nested structure where every string has environment
        variables and a leading ``~`` expanded. Non-string values are retained.
    """
    if isinstance(value, dict):
        return {key: _expand(item) for key, item in value.items()}
    if isinstance(value, list):
        return [_expand(item) for item in value]
    if isinstance(value, str):
        return os.path.expanduser(os.path.expandvars(value))
    return value


def _read(path: Path, stack: tuple[Path, ...] = ()) -> Dict[str, Any]:
    """Read one JSON file and recursively compose optional base files.

    Args:
        path: JSON file to load. Relative ``_base_`` entries resolve beside it.
        stack: Internal chain of resolved paths used to detect inheritance
            cycles. Callers normally leave this empty.

    Returns:
        Merged plain dictionary before string expansion and validation.

    Raises:
        ValueError: If configs form a circular ``_base_`` dependency.
    """
    path = path.resolve()
    if path in stack:
        chain = " -> ".join(str(item) for item in (*stack, path))
        raise ValueError(f"Circular config inheritance: {chain}")
    config = json.loads(path.read_text())
    bases = config.pop("_base_", [])
    if isinstance(bases, str):
        bases = [bases]
    merged: Dict[str, Any] = {}
    for base in bases:
        base_path = Path(os.path.expanduser(os.path.expandvars(str(base))))
        if not base_path.is_absolute():
            base_path = path.parent / base_path
        merged = _merge(merged, _read(base_path, (*stack, path)))
    return _merge(merged, config)


def import_modules(names: Iterable[str]) -> None:
    """Import extension modules so their registry decorators execute.

    Args:
        names: Iterable of importable Python module names, such as
            ``my_project.components``.

    Returns:
        ``None``. Imports are performed for their registration side effects.
    """
    for name in names:
        import_module(str(name))


def validate_config(config: Dict[str, Any], kind: str | None = None) -> None:
    """Fail early for missing or structurally invalid public configuration fields.

    Args:
        config: Fully composed configuration dictionary.
        kind: Optional ``'eval'`` validation profile.

    Returns:
        ``None``. A descriptive ``ValueError`` is raised for invalid structure.
    """
    if not isinstance(config, dict):
        raise ValueError("The config root must be a JSON object.")
    if kind not in {None, "eval"}:
        raise ValueError(f"Unsupported config kind: {kind!r}")
    if kind == "eval" and not config:
        raise ValueError("Evaluation config must contain at least one benchmark.")


def load_config(path: str | Path, kind: str | None = None) -> Dict[str, Any]:
    """Load a resolved config and import optional external extension modules.

    Args:
        path: JSON config path. Relative ``_base_`` paths resolve beside this file.
        kind: Optional validation profile passed to :func:`validate_config`.

    Returns:
        Plain nested dictionaries/lists suitable for JSON serialization. The
        optional top-level ``imports`` list is retained for external registry
        extensions.
    """
    config = _expand(_read(Path(path)))
    imports = config.get("imports", [])
    if isinstance(imports, str):
        imports = [imports]
    import_modules(imports)
    validate_config(config, kind=kind)
    return config