From ab599486efc794e4ec74f505e78e238a9663a812 Mon Sep 17 00:00:00 2001 From: wuzhuorong <973204353@qq.com> Date: Tue, 16 Jun 2026 10:16:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(event):=20=E6=96=B0=E5=A2=9E=E5=A4=9A?= =?UTF-8?q?=E5=B8=A7=E7=B4=AF=E7=A7=AF=E5=88=86=E6=9E=90=E5=99=A8=20MultiF?= =?UTF-8?q?rameAccumulator?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/event/frame_accumulator.py | 295 ++++++++++++++++++ 1 file changed, 295 insertions(+) create mode 100644 apps/server/services/event/frame_accumulator.py diff --git a/apps/server/services/event/frame_accumulator.py b/apps/server/services/event/frame_accumulator.py new file mode 100644 index 0000000..e298a40 --- /dev/null +++ b/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"]