feat(detect): log detection progress about once a minute

CHANGELOG.md

@@ -8,6 +8,10 @@ All notable changes to this project are documented here. The format follows
 
 ### Added
 
+- **Detection progress heartbeat.** `detect_and_track(progress_every_s=60.0)` logs
+  frame, percentage, detection count and fps about once a minute (wall clock) plus a
+  final `detect: done` line; `0` disables it. A 54 197-frame pass was silent for 4.5 h.
+  Logging only, results unchanged.
 - **Per-class verification result.** `CoverageReport.classes` reports `cabin`, `face`,
   `plate` and `person` each as `verified`, `failed` or `unchecked` with the evidence
   basis and its limitation; only `cabin` can be verified today, so a cabin-only pass

README.md

@@ -485,6 +485,11 @@ run is unchanged; a resume re-decodes `checkpoint_overlap_frames` (default 90)
 before the resume point and unions those boxes over the checkpointed ones. A
 checkpoint belonging to another video, or a corrupt one, is ignored.
 
+`detect_and_track(progress_every_s=60.0)` logs one INFO line
+(`detect: frame N/TOTAL (P%), D detections so far, F fps`) per wall-clock interval and a
+final `detect: done` line, so a multi-hour pass shows it is alive; `0` disables it. It
+only logs; the returned boxes are unchanged.
+
 **Tracking** — `TrackManager`, `Track`.
 
 **Timeline** — `build_timeline()`, `BlurTimeline`, `dilate_box()`,

redactcam/detect.py

@@ -17,6 +17,7 @@
 import json
 import logging
 import os
+import time
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -884,6 +885,7 @@ def detect_and_track(
     checkpoint_path: str | Path | None = None,
     checkpoint_every_frames: int = 0,
     checkpoint_overlap_frames: int = CHECKPOINT_OVERLAP_FRAMES,
+    progress_every_s: float = 60.0,
 ) -> tuple[dict[int, list[Box]], int, int, int]:
     """One decode pass producing DENSE per-frame privacy boxes.
 
@@ -988,6 +990,12 @@ def detect_and_track(
     before the detector first scores them). Emission also runs anchor interpolation — a track's
     between-detection gaps are filled with the straight line between bracketing detections, since flow
     freezes a small low-texture object between samples but two detections pin the true endpoints.
+
+    ``progress_every_s`` (wall-clock seconds, default 60; 0 disables) logs one INFO line
+    ``detect: frame N/TOTAL (P%), D detections so far, F fps`` per interval, plus a final
+    ``detect: done`` line, so a multi-hour pass is distinguishable from a hung one. TOTAL
+    and P are omitted when the container reports no frame count. Logging only; the
+    returned boxes are unchanged.
     """
     from .track import TrackManager
 
@@ -1053,6 +1061,9 @@ def detect_and_track(
                 frames_done,
             )
     last_checkpoint = frame_idx
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
+    t_start = last_progress = time.monotonic()
+    progress_start_idx = last_progress_idx = frame_idx
     # Adaptive plate cadence: between sparse samples a plate accelerating through a
     # close pass moves farther per frame than flow can follow, so the box lags and
     # the plate escapes (readable). While a plate track is fast, force plate-only
@@ -1162,6 +1173,22 @@ def detect_and_track(
                 plate_seen=len(plates) > 0 or len(vehicles) > 0,
             )
             frame_idx += 1
+            if progress_every_s > 0:
+                now = time.monotonic()
+                if now - last_progress >= progress_every_s:
+                    fps_now = (frame_idx - last_progress_idx) / max(now - last_progress, 1e-9)
+                    if total_frames > 0:
+                        pct = 100 * frame_idx / total_frames
+                        where = f"{frame_idx}/{total_frames} ({pct:.0f}%)"
+                    else:
+                        where = f"{frame_idx}"
+                    logger.info(
+                        "detect: frame %s, %d detections so far, %.1f fps",
+                        where,
+                        n_face + n_plate,
+                        fps_now,
+                    )
+                    last_progress, last_progress_idx = now, frame_idx
             # Snapshot only — the tracker keeps running, so an uninterrupted
             # pass returns exactly what it would have without checkpointing.
             if (
@@ -1194,4 +1221,10 @@ def detect_and_track(
     )
     if resumed:
         frame_boxes = _union_boxes(resumed, frame_boxes)
+    if progress_every_s > 0:
+        logger.info(
+            "detect: done, %d frames in %.0fs",
+            frame_idx - progress_start_idx,
+            time.monotonic() - t_start,
+        )
     return frame_boxes, frame_idx, n_face, n_plate

tests/test_detect.py

@@ -941,3 +941,80 @@ def test_a_corrupt_checkpoint_degrades_to_a_full_pass(self, tmp_path):
             video, checkpoint_path=ckpt, **self.CKPT, **self.KW
         )
         assert n_frames == 24 and len(boxes) == 24
+
+
+class TestDetectionProgressHeartbeat:
+    """A multi-hour pass used to log nothing between "model loaded" and the end
+    (4.5 h of silence on a 54 197-frame day video, 2026-10-08), so an operator could
+    not tell "working" from "hung". ``progress_every_s`` makes the loop log about
+    once per wall-clock interval. The clock is faked: one second per ``monotonic``
+    call, so the cadence is exact and the test never sleeps.
+    """
+
+    KW = TestDetectionCheckpoint.KW
+
+    @staticmethod
+    def _fake_clock(monkeypatch):
+        ticks = {"t": 0.0}
+
+        def monotonic():
+            ticks["t"] += 1.0
+            return ticks["t"]
+
+        monkeypatch.setattr(detect, "time", SimpleNamespace(monotonic=monotonic))
+
+    @staticmethod
+    def _progress(caplog):
+        return [r.getMessage() for r in caplog.records if r.getMessage().startswith("detect:")]
+
+    def test_progress_lines_are_logged_at_the_interval(self, tmp_path, monkeypatch, caplog):
+        self._fake_clock(monkeypatch)
+        video = _make_moving_video(tmp_path)  # 24 frames
+        with caplog.at_level("INFO", logger="redactcam.detect"):
+            detect_and_track(video, progress_every_s=5.0, **self.KW)
+        msgs = self._progress(caplog)
+        frame_lines = [m for m in msgs if m.startswith("detect: frame ")]
+        assert 2 <= len(frame_lines) <= 6, msgs
+        assert "/24" in frame_lines[0] and "%" in frame_lines[0] and "fps" in frame_lines[0]
+        assert msgs[-1].startswith("detect: done, 24 frames in "), msgs
+
+    def test_progress_does_not_change_the_result(self, tmp_path, monkeypatch, caplog):
+        video = _make_moving_video(tmp_path)
+        plain = detect_and_track(video, progress_every_s=0, **self.KW)
+        self._fake_clock(monkeypatch)
+        with caplog.at_level("INFO", logger="redactcam.detect"):
+            on = detect_and_track(video, progress_every_s=5.0, **self.KW)
+        assert [m for m in self._progress(caplog) if m.startswith("detect: frame ")]
+        assert on == plain
+
+    def test_zero_emits_no_progress_lines(self, tmp_path, monkeypatch, caplog):
+        self._fake_clock(monkeypatch)
+        video = _make_moving_video(tmp_path)
+        with caplog.at_level("INFO", logger="redactcam.detect"):
+            detect_and_track(video, progress_every_s=0, **self.KW)
+        assert self._progress(caplog) == []
+
+    def test_unknown_frame_total_drops_the_percentage(self, tmp_path, monkeypatch, caplog):
+        self._fake_clock(monkeypatch)
+        video = _make_moving_video(tmp_path)
+
+        real = cv2.VideoCapture
+
+        class NoCount:
+            """Delegating wrapper (subclassing cv2.VideoCapture segfaults at exit)."""
+
+            def __init__(self, *a):
+                self._cap = real(*a)
+
+            def get(self, prop):
+                return 0.0 if prop == cv2.CAP_PROP_FRAME_COUNT else self._cap.get(prop)
+
+            def __getattr__(self, name):
+                return getattr(self._cap, name)
+
+        fake_cv2 = SimpleNamespace(**{**vars(cv2), "VideoCapture": NoCount})
+        monkeypatch.setattr(detect, "cv2", fake_cv2)
+        with caplog.at_level("INFO", logger="redactcam.detect"):
+            detect_and_track(video, progress_every_s=5.0, **self.KW)
+        lines = [m for m in self._progress(caplog) if m.startswith("detect: frame ")]
+        assert lines and all("%" not in m and "/" not in m for m in lines), lines