"""Contextra Event Gate — 이벤트 게이트 영속화. 문제: 모든 프레임을 무조건 PG에 저장하면 (~290만 행) 실제 이벤트(~255건)의 증거와 노이즈가 구분되지 않아 오탐/정답 판별이 불가능했다. 해법: persist_mode='event_gated' 스트림은 - 모든 병합 프레임을 메모리 링 버퍼에 유지 (PG 미기록) - 시나리오 엔진이 이벤트를 발생시킬 때 ±윈도우(전 10s / 후 5s) 프레임을 PG에 플러시 + frames.event_context={"event_id","event_type"} 태그 → 저장된 프레임 = 전부 이벤트 증거. 오탐/정답은 events.is_verified로 라벨링. 지연 플러시: 이벤트 직후(+post_window) 프레임은 발생 시점에 아직 버퍼에 없으므로, 플러시는 T + post_window_s 후에 실행된다 (스케줄러 스레드). 스레드 모델: - _persist_loop(5s 주기) → StorageWriter.persist() → gate.offer() (메모리만) - persist() 내 시나리오 엔진 → 이벤트 발생 시 gate.on_events() (pending 등록) - EventGate._flusher (1s 주기 daemon) → due된 이벤트 윈도우 PG 플러시 한계: 동일 프레임에 여러 이벤트가 겹치면 첫 이벤트 태그만 유지 (ON CONFLICT DO NOTHING). v1 허용 범위. """ import threading import time from collections import deque class EventGate: """이벤트 발생 시 ±윈도우 프레임을 PG에 플러시하는 게이트.""" def __init__(self, storage, stream_id: str = "CAM_01", site_id: str = "SITE01", pre_window_s: float = 10.0, post_window_s: float = 5.0, max_buffer_frames: int = 900): """ Args: storage: app.metadata_storage.storage 싱글턴 (insert_frame/insert_detections) stream_id/site_id: 프레임 메타에 기록할 식별자 pre_window_s: 이벤트 이전 윈도우 (초) — 증거 맥락 post_window_s: 이벤트 이후 윈도우 (초) — 직후 반응 max_buffer_frames: 링 버퍼 최대 프레임 수 (~90s @ 10fps 병합 기준) """ self.storage = storage self.stream_id = stream_id self.site_id = site_id # 프레임/탐지 메타 클래스 (flush 시 변환용) — lazy import, 실패 시 게이트 비활성 try: from app.metadata_storage import FrameMeta as _FM, DetectionMeta as _DM self.FrameMeta, self.DetectionMeta = _FM, _DM except Exception as e: print(f"[EVENT-GATE] meta classes unavailable — gate disabled: {e}", flush=True) self.FrameMeta = self.DetectionMeta = None self.pre_ms = int(pre_window_s * 1000) self.post_ms = int(post_window_s * 1000) # (fn_key, f_dict, w, h, ts_ms, pkey) — 메모리만 self.buffer: deque = deque(maxlen=max_buffer_frames) self._pending: list[dict] = [] # {ts, event_id, event_type, due} self._flushed_pkeys: set = set() # 이미 PG에 쓴 pkey (중복 방지) self._lock = threading.Lock() self._stop = False self.stats = {"offered": 0, "flushed_frames": 0, "events_flushed": 0} # -- offer / on_events (persist 경로에서 호출 — PG 미접촉) --------------- def offer(self, fn_key, f: dict, w: int, h: int, pkey: str): """병합 프레임을 링 버퍼에 유지 (메모리만).""" now_ms = int(time.time() * 1000) with self._lock: self.buffer.append((fn_key, f, w, h, now_ms, pkey)) self.stats["offered"] += 1 def on_events(self, events: list[dict]): """시나리오 엔진 이벤트 등록 → T+post_window 후 플러시 예약. events: [{"event_id": str, "event_type": str, "timestamp_utc_ms": int}] """ if not events: return with self._lock: for ev in events: ts = int(ev.get("timestamp_utc_ms") or time.time() * 1000) self._pending.append({ "ts": ts, "event_id": str(ev.get("event_id")), "event_type": ev.get("event_type", "unknown"), "due": ts + self.post_ms, }) # -- flusher 스레드 ------------------------------------------------------- def start(self): t = threading.Thread(target=self._flush_loop, name="event-gate-flusher", daemon=True) t.start() return t def _flush_loop(self): while not self._stop: time.sleep(1) now_ms = int(time.time() * 1000) with self._lock: due = [p for p in self._pending if now_ms >= p["due"]] for p in due: self._pending.remove(p) for p in due: n = self.flush_window(p["ts"], p["event_id"], p["event_type"]) if n: print(f"[EVENT-GATE] {self.stream_id}: event={p['event_type']} " f"id={str(p['event_id'])[:8]} window_frames={n}", flush=True) def flush_window(self, event_ts_ms: int, event_id: str, event_type: str) -> int: """이벤트 ±윈도우 프레임을 PG에 플러시. 반환=플러시된 프레임 수.""" if self.storage is None or self.FrameMeta is None or self.DetectionMeta is None: return 0 lo = event_ts_ms - self.pre_ms hi = event_ts_ms + self.post_ms flushed = 0 with self._lock: items = [x for x in list(self.buffer) if lo <= x[4] <= hi] for fn_key, f, w, h, ts, pkey in items: if not pkey or pkey in self._flushed_pkeys: continue try: frame_meta = self.FrameMeta( private_key=pkey, stream_id=self.stream_id, site_id=self.site_id, timestamp_utc_ms=ts, frame_sequence_id=int(fn_key), inference_fps=15.0, width=w, height=h, event_context={"event_id": str(event_id), "event_type": event_type}, ) self.storage.insert_frame(frame_meta) # det/pose/seg → DetectionMeta (StorageWriter.persist와 동일 변환) dets = [] for d in f.get("det", {}).get("items", []): cx, cy, bw, bh = (float(v) for v in d["box"]) dets.append(self.DetectionMeta( private_key=pkey, track_id=None, class_label=_coco_name(d["class_id"]), confidence=float(d["score"]), bbox=((cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h), bbox_normalized=(cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2), )) for p in f.get("pose", {}).get("items", []): cx, cy, bw, bh = (float(v) for v in p["box"]) dets.append(self.DetectionMeta( private_key=pkey, track_id=None, class_label="person", confidence=float(p["score"]), bbox=((cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h), bbox_normalized=(cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2), keypoints=p.get("keypoints"), source="pose", )) for sg in f.get("seg", {}).get("items", []): cx, cy, bw, bh = (float(v) for v in sg["box"]) total_px = int(f.get("seg", {}).get("mask_w", 144)) * int(f.get("seg", {}).get("mask_h", 144)) or 20736 dets.append(self.DetectionMeta( private_key=pkey, track_id=None, class_label=_coco_name(sg["class_id"]), confidence=float(sg["score"]), bbox=((cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h), bbox_normalized=(cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2), source="seg", seg={"fg_px": int(sg["fg_px"]), "fg_ratio": round(int(sg["fg_px"]) / total_px, 6), "mask_w": int(f.get("seg", {}).get("mask_w", 144)), "mask_h": int(f.get("seg", {}).get("mask_h", 144))}, )) if dets: self.storage.insert_detections(dets) with self._lock: self._flushed_pkeys.add(pkey) flushed += 1 except Exception as e: print(f"[EVENT-GATE] flush failed pkey={pkey} (non-fatal): {e}", flush=True) if flushed: with self._lock: self.stats["flushed_frames"] += flushed self.stats["events_flushed"] += 1 # set 유한 유지 if len(self._flushed_pkeys) > 5000: self._flushed_pkeys = set(list(self._flushed_pkeys)[-2000:]) return flushed def prune(self): """버퍼에서 이미 플러시된 항목 제거 (메모리 절약).""" with self._lock: self.buffer = deque( (x for x in self.buffer if not x[5] or x[5] not in self._flushed_pkeys), maxlen=self.buffer.maxlen, ) def close(self): """stop 시 호출 — flusher 스레드 정지 + pending 이벤트 즉시 플러시.""" self._stop = True with self._lock: due = list(self._pending) self._pending.clear() for p in due: try: self.flush_window(p["ts"], p["event_id"], p["event_type"]) except Exception as e: print(f"[EVENT-GATE] close flush failed (non-fatal): {e}", flush=True) def _coco_name(class_id: int) -> str: """COCO class id → 이름 (multi_pipeline과 동일 매핑).""" from app.multi_pipeline import _coco_name as _cn return _cn(class_id)