dsh/dual_camera_system/core/group_state.py

"""各摄像头组运行时共享状态."""
import copy
import logging
import threading
from typing import Any, Dict


class GroupState:
    def __init__(self):
        self._data: Dict[str, Dict[str, Any]] = {}
        self._lock = threading.Lock()

    def reset(self):
        """重置所有状态（主要用于测试）."""
        with self._lock:
            self._data.clear()

    def init_group(self, group_id: str, panorama_rtsp: str, ptz_rtsp: str, ptz_config: Dict):
        with self._lock:
            if group_id in self._data:
                raise ValueError(f"Group {group_id} already initialized")
            self._data[group_id] = {
                "panorama_rtsp": panorama_rtsp,
                "ptz_rtsp": ptz_rtsp,
                "ptz_position": {"pan": 0.0, "tilt": 0.0, "zoom": 1},
                "polling_state": "idle",  # idle|scanning|polling|paused
                "scan_progress": {"total": 0, "current": 0, "state": "idle"},
                "last_detection": None,
                "logs": [],
            }

    def get(self, group_id: str) -> Dict[str, Any]:
        with self._lock:
            return copy.deepcopy(self._data.get(group_id, {}))

    def list_groups(self) -> list:
        with self._lock:
            return list(self._data.keys())

    def update(self, group_id: str, key: str, value: Any):
        with self._lock:
            if group_id not in self._data:
                logging.warning("Cannot update unknown group: %s", group_id)
                return
            self._data[group_id][key] = value

    def compare_and_update(
        self, group_id: str, key: str, expected: Any, new_value: Any
    ) -> bool:
        with self._lock:
            if group_id not in self._data:
                return False
            if self._data[group_id].get(key) != expected:
                return False
            self._data[group_id][key] = new_value
            return True

    def update_nested(self, group_id: str, key: str, sub_key: str, value: Any):
        with self._lock:
            if group_id not in self._data:
                raise KeyError(f"Group {group_id} not initialized")
            if key not in self._data[group_id] or not isinstance(self._data[group_id][key], dict):
                raise KeyError(f"Key {key} is not a dict")
            self._data[group_id][key][sub_key] = value

    def append_log(self, group_id: str, message: str):
        with self._lock:
            if group_id not in self._data:
                logging.warning("Cannot append log to unknown group: %s", group_id)
                return
            logs = self._data[group_id]["logs"]
            logs.append(message)
            self._data[group_id]["logs"] = logs[-200:]


# 全局单例（保持 import 兼容性；测试可调用 reset() 重置）
group_state = GroupState()