Upload clashcr/core/event_detector.py with huggingface_hub

clashcr/core/event_detector.py

@@ -0,0 +1,237 @@
+"""Event detector: find candidate card-play moments from temporal changes.
+
+For normal live view:
+- Temporal frame differencing around deployment moments
+- Suppress continuous combat motion (persistent units)
+- Require clear new spawn / effect event
+- Separate opponent-side evidence from own-side evidence
+- Save debug crops/masks for every candidate event
+"""
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+import cv2
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CandidateEvent:
+    timestamp: float
+    frame_idx: int
+    side: str  # 'opponent', 'own', 'unknown'
+    bbox: Tuple[int, int, int, int]  # x, y, w, h in full-frame coords
+    diff_mask: np.ndarray = field(repr=False)
+    crop: np.ndarray = field(repr=False)
+    reason: str = ""  # why this event was triggered
+    suppressed: bool = False
+    suppression_reason: str = ""
+
+
+class EventDetector:
+    """Detects deployment events using temporal differencing and motion suppression.
+
+    Pipeline:
+    1. Convert frame to grayscale and downscale for speed.
+    2. Compute absolute diff with previous frame.
+    3. Threshold diff to get motion mask.
+    4. Remove persistent motion (units already tracked for >N frames).
+    5. Find connected components in residual mask.
+    6. Classify components by side (opponent top, own bottom).
+    7. Filter by size, shape, and temporal consistency.
+    8. Return candidate events.
+    """
+
+    def __init__(self,
+                 diff_threshold: int = 25,
+                 min_event_area: int = 200,
+                 max_event_area: int = 50000,
+                 persistence_frames: int = 15,
+                 opponent_y_ratio: float = 0.45,
+                 own_y_ratio: float = 0.55,
+                 temporal_confirmation: int = 2,
+                 debug_dir: Optional[str] = None):
+        self.diff_threshold = diff_threshold
+        self.min_event_area = min_event_area
+        self.max_event_area = max_event_area
+        self.persistence_frames = persistence_frames
+        self.opponent_y_ratio = opponent_y_ratio
+        self.own_y_ratio = own_y_ratio
+        self.temporal_confirmation = temporal_confirmation
+        self.debug_dir = Path(debug_dir) if debug_dir else None
+
+        self._prev_gray: Optional[np.ndarray] = None
+        self._prev_mask: Optional[np.ndarray] = None
+        self._persistence_map: Optional[np.ndarray] = None
+        self._frame_idx = 0
+        self._pending_events: List[dict] = []
+
+    def _preprocess(self, frame: np.ndarray) -> np.ndarray:
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        # Downscale for speed (keep aspect)
+        h, w = gray.shape
+        if w > 640:
+            scale = 640 / w
+            gray = cv2.resize(gray, None, fx=scale, fy=scale)
+        return gray
+
+    def _update_persistence(self, motion_mask: np.ndarray) -> np.ndarray:
+        """Track how long motion has persisted in each pixel.
+        Returns mask of newly appearing motion."""
+        if self._persistence_map is None:
+            self._persistence_map = np.zeros_like(motion_mask, dtype=np.uint8)
+
+        # Increment where motion is active, decay where it is not
+        self._persistence_map = np.where(motion_mask > 0,
+                                          self._persistence_map + 1,
+                                          np.maximum(self._persistence_map - 2, 0))
+
+        # New motion: pixels that are active but have low persistence
+        new_motion = (motion_mask > 0) & (self._persistence_map <= self.persistence_frames)
+        return new_motion.astype(np.uint8) * 255
+
+    def _classify_side(self, cy: float, frame_h: int) -> str:
+        """Classify event side based on vertical position.
+
+        In standard portrait Clash Royale:
+        - Opponent side is top ~45% of arena
+        - Own side is bottom ~55% of arena
+        """
+        ratio = cy / frame_h
+        if ratio < self.opponent_y_ratio:
+            return "opponent"
+        elif ratio > self.own_y_ratio:
+            return "own"
+        return "unknown"
+
+    def process(self, frame: np.ndarray, battle_result) -> List[CandidateEvent]:
+        h, w = frame.shape[:2]
+        gray = self._preprocess(frame)
+        gh, gw = gray.shape
+
+        candidates: List[CandidateEvent] = []
+
+        if self._prev_gray is not None and self._prev_gray.shape == gray.shape:
+            diff = cv2.absdiff(gray, self._prev_gray)
+            _, motion_mask = cv2.threshold(diff, self.diff_threshold, 255, cv2.THRESH_BINARY)
+
+            # Morphological cleanup
+            kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+            motion_mask = cv2.morphologyEx(motion_mask, cv2.MORPH_OPEN, kernel)
+            motion_mask = cv2.morphologyEx(motion_mask, cv2.MORPH_CLOSE, kernel)
+
+            # Remove persistent motion
+            new_motion = self._update_persistence(motion_mask)
+
+            # Scale mask back to original resolution for ROI extraction
+            if new_motion.shape != (h, w):
+                new_motion_full = cv2.resize(new_motion, (w, h), interpolation=cv2.INTER_NEAREST)
+            else:
+                new_motion_full = new_motion
+
+            # Find connected components
+            num_labels, labels, stats, centroids = cv2.connectedComponentsWithStats(new_motion_full, connectivity=8)
+
+            for i in range(1, num_labels):
+                x, y, cw, ch, area = stats[i]
+                if not (self.min_event_area <= area <= self.max_event_area):
+                    continue
+
+                cx, cy = centroids[i]
+                side = self._classify_side(cy, h)
+
+                # Skip own-side placements (we only care about opponent)
+                if side == "own":
+                    continue
+
+                # Extract crop and mask
+                pad = 20
+                x1 = max(0, x - pad)
+                y1 = max(0, y - pad)
+                x2 = min(w, x + cw + pad)
+                y2 = min(h, y + ch + pad)
+                crop = frame[y1:y2, x1:x2].copy()
+                mask = new_motion_full[y1:y2, x1:x2].copy()
+
+                event = CandidateEvent(
+                    timestamp=time.monotonic(),
+                    frame_idx=self._frame_idx,
+                    side=side,
+                    bbox=(x1, y1, x2 - x1, y2 - y1),
+                    diff_mask=mask,
+                    crop=crop,
+                    reason=f"new_motion_area={area}_side={side}",
+                )
+                candidates.append(event)
+
+        self._prev_gray = gray
+        self._frame_idx += 1
+
+        # Temporal confirmation: require candidate to appear in consecutive frames
+        confirmed = self._temporal_confirm(candidates)
+
+        # Save debug crops
+        if self.debug_dir:
+            self.debug_dir.mkdir(parents=True, exist_ok=True)
+            for ev in confirmed:
+                if ev.suppressed:
+                    continue
+                ts = int(ev.timestamp * 1000)
+                crop_path = self.debug_dir / f"event_{ts}_{ev.side}.jpg"
+                mask_path = self.debug_dir / f"event_{ts}_{ev.side}_mask.png"
+                cv2.imwrite(str(crop_path), ev.crop)
+                cv2.imwrite(str(mask_path), ev.diff_mask)
+
+        return confirmed
+
+    def _temporal_confirm(self, candidates: List[CandidateEvent]) -> List[CandidateEvent]:
+        """Require events to persist across multiple frames near the same location."""
+        # Simple implementation: store pending and match by IoU
+        new_pending = []
+        confirmed = []
+
+        for cand in candidates:
+            matched = False
+            for pending in self._pending_events:
+                if self._iou(cand.bbox, pending["bbox"]) > 0.3:
+                    pending["frames"] += 1
+                    pending["last"] = cand
+                    matched = True
+                    if pending["frames"] >= self.temporal_confirmation:
+                        confirmed.append(cand)
+                    break
+            if not matched:
+                new_pending.append({"bbox": cand.bbox, "frames": 1, "last": cand})
+
+        self._pending_events = new_pending + [p for p in self._pending_events if p["frames"] < self.temporal_confirmation]
+        return confirmed
+
+    @staticmethod
+    def _iou(a: Tuple[int, int, int, int], b: Tuple[int, int, int, int]) -> float:
+        ax, ay, aw, ah = a
+        bx, by, bw, bh = b
+        inter_x1 = max(ax, bx)
+        inter_y1 = max(ay, by)
+        inter_x2 = min(ax + aw, bx + bw)
+        inter_y2 = min(ay + ah, by + bh)
+        inter_area = max(0, inter_x2 - inter_x1) * max(0, inter_y2 - inter_y1)
+        union_area = aw * ah + bw * bh - inter_area
+        return inter_area / union_area if union_area > 0 else 0.0
+
+    def reset(self) -> None:
+        self._prev_gray = None
+        self._prev_mask = None
+        self._persistence_map = None
+        self._frame_idx = 0
+        self._pending_events = []
+
+    def suppress_own_overlay(self, frame: np.ndarray, own_roi: Tuple[int, int, int, int]) -> None:
+        """Mask out own card overlay / hand card region to prevent false positives."""
+        # This is a hint; actual suppression happens in process() by side classification
+        pass