diff --git a/recognization/code/app.py b/recognization/code/app.py index 35eaaed..d4a3cc7 100644 --- a/recognization/code/app.py +++ b/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: diff --git a/recognization/code/record.py b/recognization/code/record.py new file mode 100644 index 0000000..c753459 --- /dev/null +++ b/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//000123.jpg the frame, exactly as the client sent it + DIR//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 diff --git a/recognization/code/test_record.py b/recognization/code/test_record.py new file mode 100644 index 0000000..140d23f --- /dev/null +++ b/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() diff --git a/recognization/docs/DATA.md b/recognization/docs/DATA.md new file mode 100644 index 0000000..81e1f60 --- /dev/null +++ b/recognization/docs/DATA.md @@ -0,0 +1,92 @@ +# Capturing real data + +Everything measured in this project so far comes from one or two photographs. That +is its largest weakness: the identity benchmark takes its gallery and its queries +from the same photo, so every accuracy number is an in-distribution number. + +This is the procedure for replacing that with data from a real store, recorded on +the device that will do the scanning. + +## 1. Record a walk + +Start the server with recording on. Frames are written every Nth processed frame, +and each one is written with a line describing what the server saw in it. + +```bash +cd recognization +YOLO_OFFLINE=1 .venv/bin/python code/app.py \ + --port 8090 --host 0.0.0.0 \ + --classify-workers 8 \ + --record-frames ~/dorayaki-data/2026-09-25-storeA \ + --record-every 3 +``` + +Then run the app and scan. Each scan session becomes its own directory: + +``` +~/dorayaki-data/2026-09-25-storeA// + 000000.jpg the frame, exactly as the client sent it + 000001.jpg + index.jsonl one line per recorded frame: + {"frame": 123, "path": "000001.jpg", "at": 1699..., + "summary": {...}, + "tracks": [{"box": [...], "identity": {...}, + "tag_identity": {...}, "tag": {...}}]} +``` + +Frames are sampled, not continuous, and the index keeps each frame's own number so +the sampling is visible rather than silent. `/v1/health` reports how many frames +have been written. + +## 2. What to capture, and why + +Two visits to the same shelf are worth more than one long visit, because the +benchmark needs a **cross-visit split**: the gallery comes from visit 1 and the +test set from visit 2. That is the difference between "it works on this photo" and +"it works on this shelf". + +Per shelf, capture: + +- a **slow walk along each rail** at working distance (about 1 m) — the live loop's + real input; +- a **close pass** (about 30 cm) over a few tags — the only distance where the + barcode is decodable (measured: 0.63 px per module at 1 m needs ~3× more pixels); +- the same shelf again **on a second visit**, ideally with something changed: a gap, + a moved facing, a new product. + +## 3. Turn a recording into labels + +The tag channel seeds the labels. A facing whose tag named a product, with a price +the tag's own arithmetic corroborates, is a candidate label rather than a guess: + +```bash +python -c " +import json, sys +sys.path.insert(0, 'code') +from record import labelling_candidates +seeds = labelling_candidates('~/dorayaki-data/2026-09-25-storeA//index.jsonl') +print(len(seeds), 'candidate labels') +print(json.dumps(seeds[:3], ensure_ascii=False, indent=1)) +" +``` + +A human then confirms or corrects them in the app's confirmation screen — the same +screen that closes the flywheel. Nothing here is a training set on its own; it is +the raw material for one. + +## 4. Freeze a benchmark + +Write the confirmed labels into a manifest that the evaluation reads, and keep it in +the repository so results are reproducible: + +``` +benchmarks/-/ + visit1//.jpg the gallery: 2-3 views per SKU + visit2/*.jpg the test frames + labels.json {"visit2/0007.jpg": [{"box": [...], "sku_id": "..."}]} +``` + +Then the identity benchmark (see `code/bench_identity.py`) can be run with a +cross-visit split, and every number it reports is a number worth acting on. Until +that exists, treat the current figures as what they are: measurements on a +photograph.