"""Group per-frame document detections into capture events. Stable-window-only detection misses documents shown while hands are still adjusting them — motion scores stay above threshold even though a receipt is clearly visible and detectable frame-by-frame. This clusters consecutive frames where a document contour was found (with a time-gap to split separate placements) and keeps the sharpest frame from each cluster. """ from __future__ import annotations from dataclasses import dataclass import numpy as np from paperpod.capture.video import Frame from paperpod.vision.document import DocumentDetection from paperpod.vision.sharpness import sharpness_score @dataclass class DocumentEvent: event_id: int t_start: float t_end: float sample_count: int best_frame_t: float best_frame_index: int sharpness: float frame: Frame detection: DocumentDetection def _same_document( prev: DocumentDetection, curr: DocumentDetection, frame_shape: tuple[int, ...], area_tolerance: float = 0.12, centroid_tolerance: float = 0.12, ) -> bool: """True if two detections likely belong to the same physical document. On a black mat, the empty table is often detected as a large contour on almost every frame, which bridges separate receipts/forms into one giant cluster if we only use time gaps. A sudden change in area or position means a new document was placed (Costco receipt area ~0.26 vs mat ~0.65). """ if abs(prev.area_ratio - curr.area_ratio) > area_tolerance: return False h, w = frame_shape[:2] prev_c = prev.quad.mean(axis=0) curr_c = curr.quad.mean(axis=0) if float(np.linalg.norm(prev_c - curr_c)) > centroid_tolerance * max(w, h): return False return True def _plausible_capture(det: DocumentDetection) -> bool: """Skip mat/table false positives that sit between receipt and full-page sizes. On a black mat, empty-table contours usually land around area_ratio 0.45-0.57. Real receipts cluster below 0.40; full tax forms above 0.58. """ ar = det.area_ratio return ar <= 0.40 or ar >= 0.58 def find_document_events( samples: list[tuple[Frame, DocumentDetection]], gap_s: float = 2.0, min_samples: int = 2, ) -> list[DocumentEvent]: """Cluster consecutive detections; split when gap_s passes with no detection. samples must be in time order. Each sample is a (frame, detection) pair from a frame where detect_document() succeeded. """ if not samples: return [] clusters: list[tuple[float, float, int, float, Frame, DocumentDetection]] = [] active: tuple[float, float, int, float, Frame, DocumentDetection] | None = None def flush() -> None: nonlocal active if active is None: return t_start, t_end, count, sharp, frame, det = active if count >= min_samples: clusters.append(active) active = None for frame, det in samples: if active is None: active = (frame.t, frame.t, 1, sharpness_score(frame.image), frame, det) continue t_start, t_end, count, best_sharp, best_frame, best_det = active if frame.t - t_end > gap_s or not _same_document(best_det, det, frame.image.shape): flush() active = (frame.t, frame.t, 1, sharpness_score(frame.image), frame, det) continue sharp = sharpness_score(frame.image) if sharp > best_sharp: best_sharp, best_frame, best_det = sharp, frame, det active = (t_start, frame.t, count + 1, best_sharp, best_frame, best_det) flush() return [ DocumentEvent( event_id=i, t_start=t_start, t_end=t_end, sample_count=count, best_frame_t=frame.t, best_frame_index=frame.index, sharpness=sharp, frame=frame, detection=det, ) for i, (t_start, t_end, count, sharp, frame, det) in enumerate(clusters) ]