refactor(config): add set_config_value for non-merge replacement

This commit is contained in:
Adriano Dal Pastro committed 2026-05-21 15:42:14 +02:00
1 parent fb05409403
commit d124a7c896
3 files changed
+78 -19

No files matched your search

+3 -19
View File
@@ -8,9 +8,8 @@ from typing import Any, Literal, Optional
from fastapi import APIRouter, HTTPException from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from .. import config as _config_mod
from ..clients import G1Unavailable from ..clients import G1Unavailable
from ..config import get_config, update_config from ..config import get_config, set_config_value
from ..geometry import normalize_quaternion from ..geometry import normalize_quaternion
from ..timing import Stopwatch from ..timing import Stopwatch
from . import _deps from . import _deps
@@ -29,21 +28,6 @@ def _validate_name(name: str) -> None:
) )
def _replace_named_poses(poses: dict[str, Any]) -> None:
"""Wholesale replace pose_3d.named_poses.
update_config does a recursive deep-merge so a dict patch can't remove
keys from a nested dict. We bypass it by mutating the in-memory state
directly, then call update_config with an empty patch to persist."""
with _config_mod._LOCK:
if _config_mod._STATE["config"] is None:
_config_mod.load_config()
cfg = _config_mod._STATE["config"]
if "pose_3d" not in cfg or not isinstance(cfg["pose_3d"], dict):
cfg["pose_3d"] = {}
cfg["pose_3d"]["named_poses"] = poses
update_config({}) # persist to disk
class WristPose(BaseModel): class WristPose(BaseModel):
position: list[float] position: list[float]
quaternion: list[float] quaternion: list[float]
@@ -141,7 +125,7 @@ def put_pose(name: str, req: NamedPose) -> dict[str, Any]:
cfg = get_config().get("pose_3d", {}) or {} cfg = get_config().get("pose_3d", {}) or {}
poses = dict(cfg.get("named_poses") or {}) poses = dict(cfg.get("named_poses") or {})
poses[name] = pose poses[name] = pose
_replace_named_poses(poses) set_config_value(["pose_3d", "named_poses"], poses)
return {"name": name, "pose": pose} return {"name": name, "pose": pose}
@@ -153,5 +137,5 @@ def delete_pose(name: str) -> dict[str, Any]:
if name not in poses: if name not in poses:
raise HTTPException(status_code=404, detail=f"pose '{name}' non trovata") raise HTTPException(status_code=404, detail=f"pose '{name}' non trovata")
del poses[name] del poses[name]
_replace_named_poses(poses) set_config_value(["pose_3d", "named_poses"], poses)
return {"deleted": name} return {"deleted": name}
+30
View File
@@ -63,6 +63,36 @@ def update_config(patch: dict[str, Any]) -> dict[str, Any]:
return deepcopy(merged) 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]: def _deep_merge(base: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]:
out = deepcopy(base) out = deepcopy(base)
for k, v in patch.items(): for k, v in patch.items():
+45
View File
@@ -0,0 +1,45 @@
"""Unit test for config.set_config_value (replaces, doesn't merge)."""
from __future__ import annotations
import yaml
from presa_mano.config import (
set_config_value,
get_config,
load_config,
_STATE,
_default_config_path,
)
def test_replace_nested_dict_removes_old_keys(tmp_path, monkeypatch):
cfg_file = tmp_path / "cfg.yaml"
cfg_file.write_text(yaml.safe_dump({
"section": {"items": {"a": 1, "b": 2}},
}))
monkeypatch.setenv("PRESA_MANO_CONFIG", str(cfg_file))
# Reset in-memory singleton
_STATE["config"] = None
_STATE["path"] = None
load_config()
set_config_value(["section", "items"], {"c": 3})
out = get_config()
assert out["section"]["items"] == {"c": 3}
# File on disk also updated
on_disk = yaml.safe_load(cfg_file.read_text())
assert on_disk["section"]["items"] == {"c": 3}
def test_create_intermediate_dicts(tmp_path, monkeypatch):
cfg_file = tmp_path / "cfg.yaml"
cfg_file.write_text(yaml.safe_dump({}))
monkeypatch.setenv("PRESA_MANO_CONFIG", str(cfg_file))
_STATE["config"] = None
_STATE["path"] = None
load_config()
set_config_value(["a", "b", "c"], 42)
assert get_config() == {"a": {"b": {"c": 42}}}