Upload clashcr/core/recorder.py with huggingface_hub

clashcr/core/recorder.py

@@ -0,0 +1,149 @@
+"""Evidence recorder: save frames, timestamps, metadata, and optional labels.
+
+Produces a recording directory with:
+- frames/          raw screenshots (optional, can be lossless PNG or JPEG)
+- metadata.jsonl   per-frame metadata
+- labels.csv       user annotations (timestamp, frame, side, card_key, confidence, note)
+"""
+from __future__ import annotations
+
+import csv
+import json
+import logging
+import time
+from dataclasses import dataclass, field, asdict
+from pathlib import Path
+from typing import Optional, Dict, Any
+
+import cv2
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class FrameMetadata:
+    timestamp: float
+    frame_idx: int
+    window_title: str
+    window_hwnd: int
+    window_width: int
+    window_height: int
+    dpi_scale: float
+    battle_state: str
+    battle_confidence: float
+    mode_hint: str
+    arena_roi: Optional[tuple] = None
+    own_roi: Optional[tuple] = None
+    opponent_roi: Optional[tuple] = None
+
+
+@dataclass
+class LabelEntry:
+    timestamp: float
+    frame_idx: int
+    side: str  # 'opponent', 'own', 'unknown'
+    card_key: str
+    confidence: float
+    manual_note: str = ""
+    source: str = "manual"  # or 'auto' for model predictions
+
+
+class EvidenceRecorder:
+    def __init__(self, output_dir: str, save_frames: bool = True,
+                 jpeg_quality: int = 85, max_fps: float = 8.0):
+        self.output_dir = Path(output_dir)
+        self.save_frames = save_frames
+        self.jpeg_quality = jpeg_quality
+        self.max_fps = max_fps
+        self.min_interval = 1.0 / max_fps
+
+        self.frames_dir = self.output_dir / "frames"
+        self.metadata_path = self.output_dir / "metadata.jsonl"
+        self.labels_path = self.output_dir / "labels.csv"
+        self.predictions_path = self.output_dir / "predictions.csv"
+
+        self._frame_idx = 0
+        self._last_capture_time = 0.0
+        self._metadata_file = None
+        self._labels_file = None
+        self._labels_writer = None
+        self._predictions_file = None
+        self._predictions_writer = None
+        self._is_open = False
+
+    def open(self) -> None:
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        if self.save_frames:
+            self.frames_dir.mkdir(parents=True, exist_ok=True)
+
+        self._metadata_file = open(self.metadata_path, "w", encoding="utf-8")
+        self._labels_file = open(self.labels_path, "w", newline="", encoding="utf-8")
+        self._labels_writer = csv.writer(self._labels_file)
+        self._labels_writer.writerow(["timestamp", "frame_idx", "side", "card_key",
+                                       "confidence", "manual_note", "source"])
+
+        self._predictions_file = open(self.predictions_path, "w", newline="", encoding="utf-8")
+        self._predictions_writer = csv.writer(self._predictions_file)
+        self._predictions_writer.writerow(["timestamp", "frame_idx", "side", "card_key",
+                                            "confidence", "evidence", "resolver_reason"])
+        self._is_open = True
+        logger.info("Recording opened: %s", self.output_dir)
+
+    def close(self) -> None:
+        if self._metadata_file:
+            self._metadata_file.close()
+        if self._labels_file:
+            self._labels_file.close()
+        if self._predictions_file:
+            self._predictions_file.close()
+        self._is_open = False
+        logger.info("Recording closed: %s", self.output_dir)
+
+    def __enter__(self):
+        self.open()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+        return False
+
+    def record_frame(self, frame: np.ndarray, meta: FrameMetadata) -> bool:
+        if not self._is_open:
+            return False
+        now = time.monotonic()
+        if now - self._last_capture_time < self.min_interval:
+            return False
+        self._last_capture_time = now
+
+        if self.save_frames:
+            path = self.frames_dir / f"frame_{self._frame_idx:06d}.jpg"
+            cv2.imwrite(str(path), frame, [cv2.IMWRITE_JPEG_QUALITY, self.jpeg_quality])
+
+        meta.frame_idx = self._frame_idx
+        self._metadata_file.write(json.dumps(asdict(meta)) + "\n")
+        self._frame_idx += 1
+        return True
+
+    def add_label(self, label: LabelEntry) -> None:
+        if not self._is_open or self._labels_writer is None:
+            return
+        self._labels_writer.writerow([
+            f"{label.timestamp:.3f}", label.frame_idx, label.side,
+            label.card_key, f"{label.confidence:.3f}", label.manual_note, label.source
+        ])
+        self._labels_file.flush()
+
+    def add_prediction(self, timestamp: float, frame_idx: int, side: str,
+                       card_key: str, confidence: float, evidence: str,
+                       resolver_reason: str) -> None:
+        if not self._is_open or self._predictions_writer is None:
+            return
+        self._predictions_writer.writerow([
+            f"{timestamp:.3f}", frame_idx, side, card_key,
+            f"{confidence:.3f}", evidence, resolver_reason
+        ])
+        self._predictions_file.flush()
+
+    def get_frame_path(self, frame_idx: int) -> Path:
+        return self.frames_dir / f"frame_{frame_idx:06d}.jpg"