""" Contextra AI Analytics — Python-side IoU 트래커 (MLOps 발판) 역할: - det 파이프라인의 프레임별 탐지에 안정적인 track_id를 부여한다. - 시나리오 엔진(배회/침입)이 track_id 기반 궤적·쿨다운을 전제로 동작하므로, 이 모듈 없이는 loitering이 검증 불가하다 (모든 사람이 track_id=None 공유). 설계 원칙: - 모델 독립적: bbox(xyxy 픽셀) + class_name만 요구. det/seg 어느 쪽에서든 사용 가능. - 파이프라인 무변경: GStreamer 요소 추가 없음, obj_meta 불필요. consume()의 tensor 파싱 결과에 track_id를 붙이는 후처리 단계로 삽입된다. - per-stream 인스턴스: StreamGroup/CLI 실행마다 하나씩 생성 → 스트림 간 상태 격리. - non-fatal: 트래킹 예외는 track_id=None으로 폴백 (기존 동작 유지). 알고리즘: - IoU 매칭 + 클래스 제약 (같은 클래스끼리만 매칭) - Hungarian 없이 greedy IoU 매칭 — CCTV 시나리오(15fps, 대상 수 <20)에서 충분하고 의존성 0. 대상 수가 늘면 scipy.linear_sum_assignment로 교체 가능 (인터페이스 동일). - max_age: N프레임 미매칭 시 track 폐기 (track_id 재사용 방지) - min_hits: 연속 M회 매칭된 track만 유효 (단일 프레임 노이즈 제거) """ from __future__ import annotations import itertools import logging from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple logger = logging.getLogger(__name__) def _iou(a: Tuple[float, float, float, float], b: Tuple[float, float, float, float]) -> float: """xyxy bbox 두 개의 IoU.""" ax1, ay1, ax2, ay2 = a bx1, by1, bx2, by2 = b ix1, iy1 = max(ax1, bx1), max(ay1, by1) ix2, iy2 = min(ax2, bx2), min(ay2, by2) iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1) inter = iw * ih if inter <= 0: return 0.0 area_a = max(0.0, ax2 - ax1) * max(0.0, ay2 - ay1) area_b = max(0.0, bx2 - bx1) * max(0.0, by2 - by1) union = area_a + area_b - inter return inter / union if union > 0 else 0.0 @dataclass class _Track: track_id: int class_name: str bbox: Tuple[float, float, float, float] hits: int = 1 # 연속 매칭 횟수 age: int = 0 # 마지막 매칭 후 경과 프레임 수 (미매칭 시 증가) class IoUTracker: """프레임별 탐지 목록에 track_id를 부여하는 경량 트래커. 사용법: tracker = IoUTracker() for det in frame_dets: # [{"bbox": xyxy, "class_name": str}, ...] tid = tracker.update(dets) # det 순서와 동일한 track_id 목록 반환 """ def __init__(self, iou_threshold: float = 0.3, max_age: int = 15, min_hits: int = 2): self.iou_threshold = iou_threshold self.max_age = max_age # ~1s @15fps — 미매칭 track 폐기 self.min_hits = min_hits # 2회 연속 매칭 시에만 유효 id 발급 self._tracks: Dict[int, _Track] = {} self._next_id: int = 1 def update(self, dets: List[dict]) -> List[Optional[int]]: """dets: [{"bbox": (x1,y1,x2,y2), "class_name": str}, ...] 반환: det 순서와 동일한 track_id 목록 (미매칭/미성숙은 None).""" if not dets: # 탐지 0개 — 모든 track age 증가 후 폐기 처리만 수행. self._age_out() return [] det_boxes: List[Tuple[float, float, float, float]] = [ (float(d["bbox"][0]), float(d["bbox"][1]), float(d["bbox"][2]), float(d["bbox"][3])) for d in dets] det_cls = [str(d.get("class_name", "unknown")) for d in dets] # 1) 기존 track과 IoU 행렬 (클래스 불일치는 0 처리) active_ids = list(self._tracks.keys()) best: Dict[int, Tuple[float, int]] = {} # det_idx -> (iou, track_id) for di, db in enumerate(det_boxes): row_best_iou, row_best_tid = -1.0, -1 for tid in active_ids: tr = self._tracks[tid] if tr.class_name != det_cls[di]: continue v = _iou(db, tr.bbox) if v > row_best_iou: row_best_iou, row_best_tid = v, tid if row_best_iou >= self.iou_threshold and row_best_tid >= 0: best[di] = (row_best_iou, row_best_tid) # 2) greedy 매칭 — IoU 큰 쌍부터. 한 track은 det 하나만 담당. pairs = sorted(best.items(), key=lambda kv: -kv[1][0]) matched_det: set[int] = set() matched_track: set[int] = set() for di, (iou_v, tid) in pairs: if di in matched_det or tid in matched_track: continue matched_det.add(di) matched_track.add(tid) tr = self._tracks[tid] tr.bbox = det_boxes[di] tr.hits += 1 tr.age = 0 # 3) 미매칭 det → 신규 track (id는 발급하지만 min_hits 도달 전까지 None 반환) new_track_for: Dict[int, int] = {} # det_idx -> tid for di in range(len(dets)): if di not in matched_det: tid = self._next_id self._next_id += 1 self._tracks[tid] = _Track( track_id=tid, class_name=det_cls[di], bbox=det_boxes[di], hits=1) new_track_for[di] = tid # 4) 미매칭 기존 track age 증가 + 폐기 for tid in active_ids: if tid not in matched_track: self._tracks[tid].age += 1 self._prune() # 5) 결과 — min_hits 미만은 None (노이즈 track의 id 유출 방지). # 신규 track도 hits=1로 평가 → min_hits=1이면 첫 프레임부터 id 발급. out: List[Optional[int]] = [None] * len(dets) for di in range(len(dets)): if di in matched_det: tid = best[di][1] elif di in new_track_for: tid = new_track_for[di] else: continue if self._tracks[tid].hits >= self.min_hits: out[di] = tid return out def _age_out(self): for tr in self._tracks.values(): tr.age += 1 self._prune() def _prune(self): dead = [tid for tid, tr in self._tracks.items() if tr.age > self.max_age] for tid in dead: del self._tracks[tid] # -- 관측용 ------------------------------------------------------------- @property def active_tracks(self) -> int: return len(self._tracks) def reset(self): self._tracks.clear() self._next_id = 1 # ============================================================================ # 모듈 레벨 헬퍼 — StorageWriter가 det items에 track_id를 붙이는 공용 경로 # ============================================================================ def assign_track_ids(tracker: Optional[IoUTracker], dets_out: List[dict]) -> None: """dets_out 항목(각 {"box","class_id","score"})에 "track_id" 키를 in-place 부여. tracker가 None이면 track_id=None으로 채운다 (기존 동작과 동일한 계약). 예외는 절대 전파하지 않는다 — 트래킹 실패는 시나리오만 degrade, 파이프라인은 유지. """ if not dets_out: return try: if tracker is None: ids = [None] * len(dets_out) else: ids = tracker.update([ {"bbox": _box_to_xyxy(d["box"]), "class_name": _cid_to_name(int(d.get("class_id", 0)))} for d in dets_out ]) except Exception as e: logger.warning("[tracker] update failed (non-fatal): %s", e) ids = [None] * len(dets_out) for d, tid in zip(dets_out, ids): d["track_id"] = tid def _box_to_xyxy(box) -> Tuple[float, float, float, float]: """(cx, cy, bw, bh) 정규화 → (x1, y1, x2, y2) 정규화. 트래커는 프레임 내 상대 위치로만 매칭하므로 픽셀 스케일 변환 불필요 — 해상도 변경 시에도 track이 유지된다 (zone 스케일링과 동일한 원리). """ cx, cy, bw, bh = (float(v) for v in box) return (cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2) def _cid_to_name(class_id: int) -> str: """COCO class id → 이름. multi_pipeline._coco_name과 동일한 매핑을 재사용.""" try: from app.multi_pipeline import _coco_name return _coco_name(class_id) except Exception: return f"class_{class_id}"