feat(event): 新增多帧累积分析器 MultiFrameAccumulator

apps/server/services/event/frame_accumulator.py

@@ -0,0 +1,295 @@
+"""多帧累积分析器 (MVP-3 / D26-D27)
+
+针对同一目标在时间窗口内的连续帧候选事件进行累积，
+为 ``LLMTrigger`` 提供"是否值得调用 LLM 二次判断"的决策依据。
+
+设计要点:
+
+1. 以 ``(source_id, event_type, target_identity)`` 作为唯一累积键，
+   ``target_identity`` 优先采用 ``track_id``，缺失时回退到 bbox 网格哈希
+   (与 ``EventAggregator`` 一致，确保两个模块对"同一目标"的认定口径相同)。
+2. 累积条目记录连续命中帧数、累积平均置信度、首末时间戳、最大严重性。
+3. 按时间窗口自动淘汰过期条目，并使用 OrderedDict 实现 LRU 容量保护，
+   避免多路摄像头长时间运行导致内存无限增长。
+4. 当某次新候选事件与已有键的 ``last_seen`` 间隔超出窗口时，
+   认为序列中断，自动重置 ``consecutive_hits`` 计数。
+
+Thread-safety: 当前实现为非线程安全，调用方应在单事件循环中按序使用。
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections import OrderedDict
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Tuple, TypeAlias
+
+from models.event_schemas import (
+    BBox,
+    CandidateEvent,
+    EventType,
+    SeverityLevel,
+    UnifiedDetection,
+)
+
+logger = logging.getLogger(__name__)
+
+
+_AccKey: TypeAlias = Tuple[Optional[str], str, str]
+
+_SEVERITY_ORDER: Tuple[str, ...] = (
+    SeverityLevel.INFO.value,
+    SeverityLevel.LOW.value,
+    SeverityLevel.MEDIUM.value,
+    SeverityLevel.HIGH.value,
+    SeverityLevel.CRITICAL.value,
+)
+
+
+# ---------------------------------------------------------------------------
+# 累积条目
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class AccumulationEntry:
+    """单个目标在窗口内的累积统计。
+
+    Attributes:
+        key: (source_id, event_type, target_identity)
+        event_type: 事件类型
+        source_id: 摄像头/视频流标识
+        first_seen: 首次出现时间戳
+        last_seen: 最近一次出现时间戳
+        total_hits: 窗口内累计命中次数 (含被时间窗口截断重置前的不计)
+        consecutive_hits: 当前连续命中帧数
+        confidence_sum: 置信度累积和 (用于计算平均值)
+        max_confidence: 窗口内最高置信度
+        max_severity: 窗口内最高严重性
+        last_event: 最近一次的候选事件 (用于 LLM 抽帧)
+        last_bbox: 最近一次目标 bbox
+    """
+
+    key: _AccKey
+    event_type: EventType
+    source_id: Optional[str]
+    first_seen: float
+    last_seen: float
+    total_hits: int = 1
+    consecutive_hits: int = 1
+    confidence_sum: float = 0.0
+    max_confidence: float = 0.0
+    max_severity: SeverityLevel = SeverityLevel.INFO
+    last_event: Optional[CandidateEvent] = None
+    last_bbox: Optional[BBox] = None
+    metadata: Dict[str, float] = field(default_factory=dict)
+
+    @property
+    def avg_confidence(self) -> float:
+        """窗口内平均置信度 (按累积命中数平均)。"""
+
+        if self.total_hits <= 0:
+            return 0.0
+        return self.confidence_sum / self.total_hits
+
+    @property
+    def duration(self) -> float:
+        """累积持续时间 (秒)。"""
+
+        return max(0.0, self.last_seen - self.first_seen)
+
+    def to_dict(self) -> Dict[str, object]:
+        return {
+            "source_id": self.source_id,
+            "event_type": self.event_type.value,
+            "target_identity": self.key[2],
+            "total_hits": self.total_hits,
+            "consecutive_hits": self.consecutive_hits,
+            "avg_confidence": round(self.avg_confidence, 4),
+            "max_confidence": round(self.max_confidence, 4),
+            "max_severity": self.max_severity.value,
+            "first_seen": self.first_seen,
+            "last_seen": self.last_seen,
+            "duration": round(self.duration, 3),
+        }
+
+
+# ---------------------------------------------------------------------------
+# MultiFrameAccumulator
+# ---------------------------------------------------------------------------
+
+
+class MultiFrameAccumulator:
+    """多帧候选事件累积器。
+
+    Args:
+        window_seconds: 累积时间窗口，超过此值的条目会被淘汰；
+            同一 key 两次命中间隔超出窗口时，连续帧计数会被重置
+        max_capacity: 最大跟踪条目数，超出时按 LRU 淘汰
+        grid_size: 缺失 track_id 时用于构造目标 hash 的网格尺寸 (像素)
+    """
+
+    def __init__(
+        self,
+        window_seconds: float = 3.0,
+        max_capacity: int = 2000,
+        grid_size: int = 50,
+    ) -> None:
+        if window_seconds <= 0:
+            raise ValueError("window_seconds 必须 > 0")
+        if max_capacity < 1:
+            raise ValueError("max_capacity 必须 >= 1")
+        if grid_size < 1:
+            raise ValueError("grid_size 必须 >= 1")
+
+        self.window_seconds = window_seconds
+        self.max_capacity = max_capacity
+        self.grid_size = grid_size
+
+        self._entries: "OrderedDict[_AccKey, AccumulationEntry]" = OrderedDict()
+
+    # ------------------------------------------------------------------
+    # 主入口
+    # ------------------------------------------------------------------
+
+    def accumulate(
+        self,
+        events: List[CandidateEvent],
+        now: Optional[float] = None,
+    ) -> List[AccumulationEntry]:
+        """累积一批候选事件，返回受影响的累积条目快照列表。
+
+        Args:
+            events: 当前帧的候选事件
+            now: 当前时间戳 (供测试注入)，默认 ``time.time()``
+        """
+
+        if now is None:
+            now = time.time()
+
+        self._evict_expired(now)
+
+        affected: List[AccumulationEntry] = []
+        for event in events:
+            entry = self._update_one(event, now)
+            if entry is not None:
+                affected.append(entry)
+
+        # LRU 容量保护
+        while len(self._entries) > self.max_capacity:
+            dropped_key, _ = self._entries.popitem(last=False)
+            logger.debug("MultiFrameAccumulator LRU 淘汰: %s", dropped_key)
+
+        return affected
+
+    # ------------------------------------------------------------------
+    # 单事件累积
+    # ------------------------------------------------------------------
+
+    def _update_one(
+        self,
+        event: CandidateEvent,
+        now: float,
+    ) -> Optional[AccumulationEntry]:
+        key = self._make_key(event)
+        existing = self._entries.get(key)
+
+        if existing is None:
+            entry = AccumulationEntry(
+                key=key,
+                event_type=event.event_type,
+                source_id=event.source_id,
+                first_seen=now,
+                last_seen=now,
+                total_hits=1,
+                consecutive_hits=1,
+                confidence_sum=event.confidence,
+                max_confidence=event.confidence,
+                max_severity=event.severity,
+                last_event=event,
+                last_bbox=event.detection.bbox,
+            )
+            self._entries[key] = entry
+            return entry
+
+        # 序列中断检测：超出窗口则重置连续计数与平均累积
+        if now - existing.last_seen > self.window_seconds:
+            existing.first_seen = now
+            existing.total_hits = 0
+            existing.consecutive_hits = 0
+            existing.confidence_sum = 0.0
+
+        existing.last_seen = now
+        existing.total_hits += 1
+        existing.consecutive_hits += 1
+        existing.confidence_sum += event.confidence
+        existing.max_confidence = max(existing.max_confidence, event.confidence)
+        existing.max_severity = self._max_severity(
+            existing.max_severity, event.severity
+        )
+        existing.last_event = event
+        existing.last_bbox = event.detection.bbox
+
+        # 移到队尾保持 LRU
+        self._entries.move_to_end(key)
+        return existing
+
+    # ------------------------------------------------------------------
+    # 工具
+    # ------------------------------------------------------------------
+
+    def _make_key(self, event: CandidateEvent) -> _AccKey:
+        return (
+            event.source_id,
+            event.event_type.value,
+            self._target_identity(event.detection),
+        )
+
+    def _target_identity(self, det: UnifiedDetection) -> str:
+        """构造目标稳定标识：优先 track_id，否则 bbox 网格哈希。"""
+
+        if det.track_id is not None:
+            return f"t{det.track_id}"
+        cx, cy = det.bbox.center
+        gx = int(cx) // self.grid_size
+        gy = int(cy) // self.grid_size
+        return f"g{gx}_{gy}_{det.class_name}"
+
+    @staticmethod
+    def _max_severity(a: SeverityLevel, b: SeverityLevel) -> SeverityLevel:
+        try:
+            ai = _SEVERITY_ORDER.index(a.value)
+            bi = _SEVERITY_ORDER.index(b.value)
+        except ValueError:
+            return a
+        return a if ai >= bi else b
+
+    # ------------------------------------------------------------------
+    # 淘汰 / 自省
+    # ------------------------------------------------------------------
+
+    def _evict_expired(self, now: float) -> None:
+        expired = [
+            key
+            for key, entry in self._entries.items()
+            if now - entry.last_seen > self.window_seconds
+        ]
+        for key in expired:
+            self._entries.pop(key, None)
+
+    def get(self, event: CandidateEvent) -> Optional[AccumulationEntry]:
+        return self._entries.get(self._make_key(event))
+
+    def clear(self) -> None:
+        self._entries.clear()
+
+    @property
+    def active_count(self) -> int:
+        return len(self._entries)
+
+    def snapshot(self) -> List[Dict[str, object]]:
+        return [entry.to_dict() for entry in self._entries.values()]
+
+
+__all__ = ["MultiFrameAccumulator", "AccumulationEntry"]