From d124a7c8969552a1daa5d4e2db8f9adbf62ba8f7 Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Thu, 21 May 2026 15:42:14 +0200 Subject: [PATCH] refactor(config): add set_config_value for non-merge replacement --- src/presa_mano/api/pose_3d_routes.py | 22 ++------------ src/presa_mano/config.py | 30 +++++++++++++++++++ tests/test_config_set_value.py | 45 ++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 19 deletions(-) create mode 100644 tests/test_config_set_value.py diff --git a/src/presa_mano/api/pose_3d_routes.py b/src/presa_mano/api/pose_3d_routes.py index 5f41c66..9d7612f 100644 --- a/src/presa_mano/api/pose_3d_routes.py +++ b/src/presa_mano/api/pose_3d_routes.py @@ -8,9 +8,8 @@ from typing import Any, Literal, Optional from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field -from .. import config as _config_mod 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 ..timing import Stopwatch 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): position: 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 {} poses = dict(cfg.get("named_poses") or {}) poses[name] = pose - _replace_named_poses(poses) + set_config_value(["pose_3d", "named_poses"], poses) return {"name": name, "pose": pose} @@ -153,5 +137,5 @@ def delete_pose(name: str) -> dict[str, Any]: if name not in poses: raise HTTPException(status_code=404, detail=f"pose '{name}' non trovata") del poses[name] - _replace_named_poses(poses) + set_config_value(["pose_3d", "named_poses"], poses) return {"deleted": name} diff --git a/src/presa_mano/config.py b/src/presa_mano/config.py index 3b416f8..dbd7af7 100644 --- a/src/presa_mano/config.py +++ b/src/presa_mano/config.py @@ -63,6 +63,36 @@ def update_config(patch: dict[str, Any]) -> dict[str, Any]: 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(): diff --git a/tests/test_config_set_value.py b/tests/test_config_set_value.py new file mode 100644 index 0000000..49175c2 --- /dev/null +++ b/tests/test_config_set_value.py @@ -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}}}