"""多帧累积分析器 (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"]