Spaces:
Running on Zero
Running on Zero
File size: 4,940 Bytes
87608ea | 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 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """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
|