ilia
a43ab20df6
- vision/refine.py: tighten crops to the paper band (removes mat margins and hands beside receipts) and inpaint border-connected skin regions so fingers disappear from output - llm/vision.py: identify documents with a local Ollama vision model (qwen2.5vl); extracts vendor/date/total/form code and flags quality issues (fingers, blur, glare); falls back to Tesseract when down - pipeline: drop blank pages, dedupe consecutive captures of the same document, record refine/LLM fields in report and export summary
125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
"""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)
|
|
]
|