refactor the data structures

Author: TheCodingLand

app/config.py

@@ -1,9 +1,31 @@
 import json
 from pathlib import Path
-from typing import Literal
+from typing import Literal, Optional, Union
+
+from pydantic import ValidationError
 from pydantic_settings import BaseSettings
 
-APP_PATH = Path(__file__).parent    
+APP_PATH = Path(__file__).parent
+DEFAULT_CONFIG_PATH = APP_PATH.parent / "config.json"
+
+_config_path = DEFAULT_CONFIG_PATH
+
+_config_cache: Optional["Config"] = None
+
+
+def _ensure_config_dir() -> None:
+    get_config_path().parent.mkdir(parents=True, exist_ok=True)
+
+
+def set_config_path(new_path: Union[str, Path]) -> None:
+    """Update the configuration file path and clear the in-memory cache."""
+    global _config_path, _config_cache
+    _config_path = Path(new_path)
+    _config_cache = None
+
+
+def get_config_path() -> Path:
+    return _config_path
 
 
 class Config(BaseSettings):
@@ -19,15 +41,40 @@ class Config(BaseSettings):
     joystick_side_identifier_right: str = "R"
     modifier_key: str = "rctrl"
 
-    def save(self):
-        with open(APP_PATH.parent / "config.json", "w") as f:
-            f.write(self.model_dump_json(indent=4))
+    def save(self) -> None:
+        _ensure_config_dir()
+        get_config_path().write_text(self.model_dump_json(indent=4))
+        self._cache_self()
+
+    def _cache_self(self) -> None:
+        global _config_cache
+        _config_cache = self
 
     @classmethod
-    def get_config(cls) -> "Config":
-        if not (APP_PATH.parent / "config.json").exists():
-            with open(APP_PATH.parent / "config.json", "w") as f:
-                f.write(cls().model_dump_json(indent=4))
-        with open(APP_PATH.parent / "config.json", "r") as f: # should be next to the executable
-            config_data = json.load(f)
-        return cls(**config_data)
+    def get_config(cls, force_reload: bool = False) -> "Config":
+        global _config_cache
+        if not force_reload and _config_cache is not None:
+            return _config_cache
+
+        config_path = get_config_path()
+
+        if not config_path.exists():
+            config = cls()
+            _ensure_config_dir()
+            config_path.write_text(config.model_dump_json(indent=4))
+            config._cache_self()
+            return config
+
+        try:
+            data = config_path.read_text()
+            config_data = json.loads(data)
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"Invalid configuration file: {config_path}") from exc
+
+        try:
+            config = cls(**config_data)
+        except ValidationError as exc:
+            raise ValueError(f"Configuration file contains invalid values: {config_path}") from exc
+
+        config._cache_self()
+        return config

app/domain/models.py

@@ -141,14 +141,14 @@ class BindingPlan:
     """Represents a staged set of binding mutations to be applied."""
 
     to_add: list[Binding] = field(default_factory=list)
-    to_remove: list[str] = field(default_factory=list)
+    to_remove: list[Binding] = field(default_factory=list)
     validation: ValidationReport = field(default_factory=ValidationReport)
 
     def record_add(self, binding: Binding) -> None:
         self.to_add.append(binding)
 
-    def record_remove(self, binding_key: str) -> None:
-        self.to_remove.append(binding_key)
+    def record_remove(self, binding: Binding) -> None:
+        self.to_remove.append(binding)
 
     def merge(self, other: "BindingPlan") -> None:
         self.to_add.extend(other.to_add)

app/services/binding_planner.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
+from collections import defaultdict
 from dataclasses import dataclass
-from typing import Iterable, Optional
+from typing import Dict, Iterable, List, Optional, Tuple
 
 from app.domain import Binding, BindingPlan, ControlProfile, ValidationIssue, ValidationReport
 
@@ -41,7 +42,7 @@ def plan_diff(
         desired_keys = {binding.key: binding for binding in desired_bindings}
 
         for key in current_keys.keys() - desired_keys.keys():
-            plan.record_remove(key)
+            plan.record_remove(current_keys[key])
 
         for key, binding in desired_keys.items():
             if key not in current_keys:
@@ -60,4 +61,82 @@ def validate_plan(self, plan: BindingPlan) -> ValidationReport:
                     message="No binding changes detected.",
                 )
             )
+            return report
+
+        slot_key = self._make_slot_key
+
+        occupancy: Dict[Tuple[str, str, str], List[Binding]] = defaultdict(list)
+        if self.context.default_profile is not None:
+            for binding in self.context.default_profile.iter_bindings():
+                occupancy[slot_key(binding)].append(binding)
+
+        for binding in plan.to_remove:
+            key = slot_key(binding)
+            if key not in occupancy:
+                continue
+            remaining = [existing for existing in occupancy[key] if existing.key != binding.key]
+            if remaining:
+                occupancy[key] = remaining
+            else:
+                occupancy.pop(key)
+
+        additions_by_slot: Dict[Tuple[str, str, str], List[Binding]] = defaultdict(list)
+        for binding in plan.to_add:
+            additions_by_slot[slot_key(binding)].append(binding)
+
+        for key, bindings in additions_by_slot.items():
+            slot_desc = f"{key[0]}:{key[2]}"
+            actions_list = ", ".join(sorted({binding.action.name for binding in bindings}))
+
+            if len(bindings) > 1:
+                modifier_values = {binding.modifier for binding in bindings}
+                if len(modifier_values) > 1:
+                    report.add(
+                        ValidationIssue(
+                            level="error",
+                            message=(
+                                f"Modifier conflict: slot {slot_desc} receives both modifier and "
+                                f"non-modifier bindings ({actions_list})."
+                            ),
+                            slot=bindings[0].slot,
+                        )
+                    )
+                else:
+                    report.add(
+                        ValidationIssue(
+                            level="error",
+                            message=(
+                                f"Duplicate slot assignment: slot {slot_desc} receives multiple "
+                                f"bindings ({actions_list})."
+                            ),
+                            slot=bindings[0].slot,
+                        )
+                    )
+
+            existing_bindings = occupancy.get(key, [])
+            if existing_bindings:
+                existing_actions = ", ".join(
+                    sorted({binding.action.name for binding in existing_bindings})
+                )
+                existing_modifiers = {binding.modifier for binding in existing_bindings}
+                addition_modifiers = {binding.modifier for binding in bindings}
+                if existing_modifiers ^ addition_modifiers:
+                    reason = "modifier conflict"
+                else:
+                    reason = "slot already mapped"
+                report.add(
+                    ValidationIssue(
+                        level="error",
+                        message=(
+                            f"{reason.capitalize()}: slot {slot_desc} currently mapped to {existing_actions}; "
+                            f"cannot add {actions_list}."
+                        ),
+                        slot=bindings[0].slot,
+                    )
+                )
+
         return report
+
+    @staticmethod
+    def _make_slot_key(binding: Binding) -> Tuple[str, str, str]:
+        return binding.slot.device_uid, binding.slot.side, binding.slot.slot_id