feat: record a live session so it can become data

The deep review's first recommendation was real data, because every accuracy number
so far comes from one or two photographs - the identity benchmark even takes its
gallery and its queries from the same photo. This is the mechanism for replacing
that.

`--record-frames DIR` writes each processed frame, sampled every Nth, with a JSON
line describing what the server saw in it: the boxes, the identity channels and the
tag reading. The lines are the labelling input - `labelling_candidates()` returns
the facings the tag channel named with a corroborated price, which are seeds for the
confirmation screen rather than guesses. Nothing here is a training set on its own.

Sampling keeps each frame's own number in the index, so a sampled recording says so
instead of hiding it, and `/v1/health` reports how much has been written. Measured
on a 163-frame walk: 32 frames and their lines.

`docs/DATA.md` is the runbook: what to capture (a slow walk per rail, a close pass
for the barcode, and the same shelf on a second visit), how to turn it into labels,
and how to freeze a benchmark with a cross-visit split.

167 tests pass.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

recognization/code/app.py

@@ -72,6 +72,7 @@ from engine import (  # noqa: E402
     settings_from_env,
 )
 from demo import DemoCamera, DemoSourceError  # noqa: E402
+from record import FrameRecorder  # noqa: E402
 from live import LiveSession, LiveSettings  # noqa: E402
 
 LOGGER = logging.getLogger("dorayaki")
@@ -99,6 +100,9 @@ class ServerSettings:
     live_conf: float = 0.1
     live_tags: bool = True
     demo_walk: bool = False
+    # A live session can be recorded so it becomes data to label (record.py).
+    record_frames: Optional[str] = None
+    record_every: int = 3
     log_level: str = "info"
     ssl_certfile: Optional[str] = None
     ssl_keyfile: Optional[str] = None
@@ -278,6 +282,7 @@ def create_app(
                 "engine": getattr(engine.ocr_engine, "name", "none"),
                 "available": bool(getattr(engine.ocr_engine, "available", lambda: False)()),
             },
+            "recording": recorder.stats() if recorder is not None else None,
             "sessions_active": len(sessions),
             "uptime_s": round(engine.uptime_s(), 1),
         }
@@ -322,6 +327,8 @@ def create_app(
     # still recognition mapped into each walk frame, which is a film-making device,
     # not a perception result. Every payload it produces is marked synthetic.
     demo_recognition = None
+    recorder = FrameRecorder(settings.record_frames, every=settings.record_every) \
+        if settings.record_frames else None
 
 
 
@@ -693,6 +700,10 @@ def create_app(
                     await send({"type": "error", "code": "INTERNAL", "frame_id": frame_id,
                                 "message": "frame processing failed"})
                     continue
+                if recorder is not None:
+                    # Record what the client sent and what we saw in it, so the walk
+                    # can be labelled later instead of only measured on photographs.
+                    recorder.write(session_id, frame_id, payload, result)
                 if demo_recognition is not None and demo_sample_ids:
                     # The walk's frames are served from the shelf's full recognition,
                     # so every product is already labelled with name and price.
@@ -980,6 +991,10 @@ def main() -> None:
                         help="classifications in flight per session (demo walks want a larger budget)")
     parser.add_argument("--live-attempts", type=int, default=None,
                         help="classification attempts per track")
+    parser.add_argument("--record-frames", type=str, default=None,
+                        help="write processed frames and their results under this directory")
+    parser.add_argument("--record-every", type=int, default=None,
+                        help="record one frame in every N (default 3)")
     parser.add_argument("--demo-walk", action="store_true",
                         help="serve the demo walk's pre-recognised shelf as live tracks")
     parser.add_argument("--no-tags", action="store_true", help="disable price-tag OCR")
@@ -1017,6 +1032,10 @@ def main() -> None:
         server.log_level = args.log_level
 
     engine_settings = settings_from_env()
+    if args.record_frames is not None:
+        server.record_frames = args.record_frames
+    if args.record_every is not None:
+        server.record_every = args.record_every
     if args.live_inflight is not None:
         engine_settings.live_inflight = args.live_inflight
     if args.live_attempts is not None:

recognization/code/record.py

@@ -0,0 +1,137 @@
+"""
+Record a live session so it can become data.
+
+Everything measured so far comes from one or two photographs, which is the largest
+weakness in the project. The fix is a store walk recorded from the device that will
+do the scanning, with enough context to label it afterwards.
+
+`--record-frames DIR` writes each processed frame plus a JSON line describing what
+the server saw in it. The lines are the labelling input: a facing that the tag
+channel named is a candidate label, and the confirmation screen is where a human
+accepts or corrects it. Nothing here is a training set on its own - it is the raw
+material for one.
+
+Layout:
+
+    DIR/<session>/000123.jpg          the frame, exactly as the client sent it
+    DIR/<session>/index.jsonl         one line per frame:
+        {"frame": 123, "path": "...", "at": 1699..., "summary": {...},
+         "tracks": [{"box": [...], "identity": {...}, "tag": {...}}]}
+
+Frames are written at most every `every` processed frames (default 3) so a long
+walk does not fill a disk, and the index keeps the frame's own numbering so the
+sampling is visible rather than silent.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+import time
+from typing import Any, Dict, List, Optional
+
+
+class FrameRecorder:
+    """Write frames and their results to disk, for labelling later."""
+
+    def __init__(self, root: str, every: int = 3, max_frames_per_session: int = 4000) -> None:
+        self.root = root
+        self.every = max(1, int(every))
+        self.max_frames_per_session = int(max_frames_per_session)
+        self._lock = threading.Lock()
+        self._counts: Dict[str, int] = {}
+        self._written: Dict[str, int] = {}
+        os.makedirs(self.root, exist_ok=True)
+
+    def _session_dir(self, session_id: str) -> str:
+        path = os.path.join(self.root, session_id)
+        os.makedirs(path, exist_ok=True)
+        return path
+
+    def write(self, session_id: str, frame_id: int, payload: bytes, result: Dict[str, Any]) -> None:
+        """Record one processed frame and what was seen in it."""
+        with self._lock:
+            seen = self._counts.get(session_id, 0) + 1
+            self._counts[session_id] = seen
+            if seen % self.every:
+                return
+            written = self._written.get(session_id, 0)
+            if written >= self.max_frames_per_session:
+                return
+            self._written[session_id] = written + 1
+
+        directory = self._session_dir(session_id)
+        name = f"{written:06d}.jpg"
+        try:
+            with open(os.path.join(directory, name), "wb") as handle:
+                handle.write(payload)
+        except OSError:
+            return
+
+        tracks: List[Dict[str, Any]] = []
+        for track in result.get("tracks") or []:
+            tracks.append(
+                {
+                    "box": track.get("box"),
+                    "identity": track.get("identity"),
+                    "tag_identity": track.get("tag_identity"),
+                    "tag": track.get("price_tag"),
+                }
+            )
+        line = {
+            "frame": frame_id,
+            "path": name,
+            "at": round(time.time(), 3),
+            "summary": result.get("summary"),
+            "tracks": tracks,
+        }
+        try:
+            with open(os.path.join(directory, "index.jsonl"), "a", encoding="utf-8") as handle:
+                handle.write(json.dumps(line, ensure_ascii=False) + "\n")
+        except OSError:
+            return
+
+    def stats(self) -> Dict[str, Any]:
+        with self._lock:
+            return {
+                "root": self.root,
+                "every": self.every,
+                "sessions": len(self._counts),
+                "frames_written": sum(self._written.values()),
+            }
+
+
+def labelling_candidates(index_path: str) -> List[Dict[str, Any]]:
+    """Turn a recorded index into candidate labels for a human to confirm.
+
+    A facing that the tag channel named carries a product name and a corroborated
+    price; those are the seeds. The rest are candidates grouped by position, to be
+    labelled in the confirmation screen rather than by hand from scratch.
+    """
+    seeds: List[Dict[str, Any]] = []
+    with open(index_path, "r", encoding="utf-8") as handle:
+        for line in handle:
+            line = line.strip()
+            if not line:
+                continue
+            entry = json.loads(line)
+            for track in entry.get("tracks") or []:
+                tag_identity = track.get("tag_identity") or {}
+                tag = track.get("tag") or {}
+                if not tag_identity.get("sku_id"):
+                    continue
+                seeds.append(
+                    {
+                        "frame": entry["frame"],
+                        "box": track.get("box"),
+                        "sku_id": tag_identity["sku_id"],
+                        "name_similarity": tag_identity.get("score"),
+                        "tag_price": tag.get("price"),
+                        "corroborated": bool(
+                            {"tax_consistent", "tax_pair_consistent", "discount_consistent"}
+                            & set(tag.get("flags") or [])
+                        ),
+                    }
+                )
+    return seeds

recognization/code/test_record.py

@@ -0,0 +1,81 @@
+"""Tests for recording a live session as data to label."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+if current_dir not in sys.path:
+    sys.path.insert(0, current_dir)
+
+from record import FrameRecorder, labelling_candidates  # noqa: E402
+
+
+def a_result(sku=None, flags=None):
+    return {
+        "summary": {"in_view": 1, "recognized": 1 if sku else 0},
+        "tracks": [
+            {
+                "box": [10, 100, 100, 300],
+                "identity": {"state": "recognized", "sku_id": sku} if sku else {"state": "unconfirmed"},
+                "tag_identity": {"sku_id": sku, "score": 0.7} if sku else {},
+                "price_tag": {"state": "readable", "price": 269, "flags": flags or []},
+            }
+        ],
+    }
+
+
+class RecorderTests(unittest.TestCase):
+    def setUp(self):
+        self.root = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
+
+    def test_frames_and_an_index_are_written(self):
+        recorder = FrameRecorder(self.root, every=1)
+        for frame_id in range(3):
+            recorder.write("session-a", frame_id, b"\xff\xd8jpeg", a_result("sku_l05"))
+        directory = os.path.join(self.root, "session-a")
+        self.assertEqual(sorted(os.listdir(directory)), ["000000.jpg", "000001.jpg", "000002.jpg", "index.jsonl"])
+        lines = [json.loads(line) for line in open(os.path.join(directory, "index.jsonl"), encoding="utf-8")]
+        self.assertEqual([line["frame"] for line in lines], [0, 1, 2])
+        self.assertEqual(lines[0]["tracks"][0]["tag_identity"]["sku_id"], "sku_l05")
+
+    def test_sampling_keeps_the_frame_numbering_visible(self):
+        recorder = FrameRecorder(self.root, every=3)
+        for frame_id in range(9):
+            recorder.write("session-b", frame_id, b"x", a_result())
+        lines = [json.loads(line) for line in open(os.path.join(self.root, "session-b", "index.jsonl"), encoding="utf-8")]
+        self.assertEqual([line["frame"] for line in lines], [2, 5, 8],
+                         "every third frame, recorded under its own number")
+
+    def test_sessions_are_separate(self):
+        recorder = FrameRecorder(self.root, every=1)
+        recorder.write("one", 0, b"a", a_result())
+        recorder.write("two", 0, b"b", a_result())
+        self.assertTrue(os.path.exists(os.path.join(self.root, "one", "000000.jpg")))
+        self.assertTrue(os.path.exists(os.path.join(self.root, "two", "000000.jpg")))
+
+    def test_a_session_is_capped(self):
+        recorder = FrameRecorder(self.root, every=1, max_frames_per_session=2)
+        for frame_id in range(5):
+            recorder.write("capped", frame_id, b"x", a_result())
+        self.assertEqual(recorder.stats()["frames_written"], 2)
+
+    def test_labelling_candidates_come_from_the_tag_channel(self):
+        recorder = FrameRecorder(self.root, every=1)
+        recorder.write("session-c", 0, b"x", a_result("sku_l05", ["tax_consistent"]))
+        recorder.write("session-c", 1, b"x", a_result())
+        seeds = labelling_candidates(os.path.join(self.root, "session-c", "index.jsonl"))
+        self.assertEqual(len(seeds), 1, "only the tag-named facing is a seed")
+        self.assertEqual(seeds[0]["sku_id"], "sku_l05")
+        self.assertTrue(seeds[0]["corroborated"])
+        self.assertEqual(seeds[0]["tag_price"], 269)
+
+
+if __name__ == "__main__":
+    unittest.main()