104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
"""Configuration loader with runtime override support."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import threading
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
|
|
|
|
_LOCK = threading.RLock()
|
|
_STATE: dict[str, Any] = {"config": None, "path": None}
|
|
|
|
|
|
def _default_config_path() -> Path:
|
|
env = os.environ.get("PRESA_MANO_CONFIG")
|
|
if env:
|
|
return Path(env)
|
|
return Path(__file__).resolve().parents[2] / "config.yaml"
|
|
|
|
|
|
def data_dir() -> Path:
|
|
env = os.environ.get("PRESA_MANO_DATA")
|
|
p = Path(env) if env else _default_config_path().parent / "data"
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
return p
|
|
|
|
|
|
def load_config(path: Path | None = None) -> dict[str, Any]:
|
|
with _LOCK:
|
|
if _STATE["config"] is not None and path is None:
|
|
return deepcopy(_STATE["config"])
|
|
cfg_path = path or _default_config_path()
|
|
with open(cfg_path, "r") as f:
|
|
cfg = yaml.safe_load(f) or {}
|
|
_STATE["config"] = cfg
|
|
_STATE["path"] = cfg_path
|
|
return deepcopy(cfg)
|
|
|
|
|
|
def get_config() -> dict[str, Any]:
|
|
with _LOCK:
|
|
if _STATE["config"] is None:
|
|
load_config()
|
|
return deepcopy(_STATE["config"])
|
|
|
|
|
|
def update_config(patch: dict[str, Any]) -> dict[str, Any]:
|
|
"""Merge a patch into the in-memory config and persist to disk."""
|
|
with _LOCK:
|
|
if _STATE["config"] is None:
|
|
load_config()
|
|
merged = _deep_merge(_STATE["config"], patch)
|
|
_STATE["config"] = merged
|
|
if _STATE["path"]:
|
|
with open(_STATE["path"], "w") as f:
|
|
yaml.safe_dump(merged, f, sort_keys=False)
|
|
return deepcopy(merged)
|
|
|
|
|
|
def set_config_value(path: list[str], value: Any) -> dict[str, Any]:
|
|
"""Set a value at a nested path in config, REPLACING the existing value
|
|
(unlike update_config which deep-merges). Useful when you need to wholesale
|
|
replace a dict (e.g., to remove keys not present in the new value).
|
|
|
|
Example:
|
|
set_config_value(["pose_3d", "named_poses"], {"a": {...}, "b": {...}})
|
|
|
|
Walks the path creating intermediate dicts if missing. Persists to disk.
|
|
Returns the new full config (deep-copied)."""
|
|
if not path:
|
|
raise ValueError("path must be a non-empty list of keys")
|
|
with _LOCK:
|
|
if _STATE["config"] is None:
|
|
load_config()
|
|
cfg = _STATE["config"]
|
|
node = cfg
|
|
for key in path[:-1]:
|
|
child = node.get(key)
|
|
if not isinstance(child, dict):
|
|
child = {}
|
|
node[key] = child
|
|
node = child
|
|
node[path[-1]] = value
|
|
if _STATE["path"]:
|
|
with open(_STATE["path"], "w") as f:
|
|
yaml.safe_dump(cfg, f, sort_keys=False)
|
|
return deepcopy(cfg)
|
|
|
|
|
|
def _deep_merge(base: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]:
|
|
out = deepcopy(base)
|
|
for k, v in patch.items():
|
|
if isinstance(v, dict) and isinstance(out.get(k), dict):
|
|
out[k] = _deep_merge(out[k], v)
|
|
else:
|
|
out[k] = v
|
|
return out
|