From 00367a95c9009a8cf282372429001653b65ca6ef Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Thu, 24 Sep 2026 18:09:23 +0900 Subject: [PATCH] feat: a synthetic store walk, and the live tuning it forced out /v1/demo/plan + /v1/demo/frame give a demo the one thing a still image cannot: motion. The virtual camera pans a shelf row left to right, steps up, and pans back - a store employee sweeping a shelf - over a real shelf photo, ~1 ms per frame, deterministic, and wrapping so an unattended demo loops. The client drives only `step`, so the walk's speed follows whatever frame rate it can sustain. Three real defects came out of measuring the walk rather than a still: - Bands were keyed by on-screen position, so a walking camera minted a new key every few pixels of pan and re-read the same shelf continuously. Keyed by row now; OCR stops chasing the window. - The classification budget was consumed by retries of products already leaving the frame. Retries now skip anything near the edge, and a track gets 2 attempts. - The pan advance was 0.22 of the window per step: a facing was in view for one or two frames, so nothing could ever be classified (9 labelled tracks in 20 s). 0.06 keeps a product in view for about a second. Measured on the walk (20 s, 1280x720 frames, one client): orb labels 296 tracks across 21 SKUs with the first label at 0.2 s, while hybrid manages 13 tracks / 4 SKUs at 5.4 s - so the live default for a demo is the fast method, with hybrid kept for tap-to-inspect. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- recognization/code/app.py | 81 ++++++++++- recognization/code/demo.py | 248 ++++++++++++++++++++++++++++++++ recognization/code/live.py | 17 ++- recognization/code/tags.py | 17 ++- recognization/code/test_demo.py | 119 +++++++++++++++ 5 files changed, 466 insertions(+), 16 deletions(-) create mode 100644 recognization/code/demo.py create mode 100644 recognization/code/test_demo.py diff --git a/recognization/code/app.py b/recognization/code/app.py index 4b267db..810c20b 100644 --- a/recognization/code/app.py +++ b/recognization/code/app.py @@ -70,6 +70,7 @@ from engine import ( # noqa: E402 identity_payload, settings_from_env, ) +from demo import DemoCamera, DemoSourceError # noqa: E402 from live import LiveSession, LiveSettings # noqa: E402 LOGGER = logging.getLogger("dorayaki") @@ -243,6 +244,7 @@ def create_app( scans = scans if scans is not None else build_legacy_scans(engine) samples = {item["id"]: item for item in get_available_shelf_images("add")} + demo_camera = DemoCamera(samples) sessions: Dict[str, LiveSession] = {} sessions_lock = asyncio.Lock() app.state.engine = engine @@ -295,12 +297,79 @@ def create_app( # ---------------------------------------------------------------- samples @app.get("/v1/samples", dependencies=[Depends(authorize)]) async def list_samples() -> Dict[str, Any]: - return { - "samples": [ - {"id": item["id"], "label": item["label"], "url": f"/v1/samples/{item['id']}"} - for item in samples.values() - ] - } + listing = [] + for item in samples.values(): + entry = {"id": item["id"], "label": item["label"], "url": f"/v1/samples/{item['id']}"} + try: # size is a convenience for clients that plan a demo walk + width, height = demo_camera._size(item["id"]) + entry["width"], entry["height"] = width, height + except DemoSourceError: + pass + listing.append(entry) + return {"samples": listing} + + # ---------------------------------------------------------------- demo camera + def _demo_samples(raw: Optional[str]) -> List[str]: + if raw: + wanted = [part.strip() for part in raw.split(",") if part.strip()] + unknown = [part for part in wanted if part not in samples] + if unknown: + raise _error("NOT_FOUND", f"unknown sample(s): {', '.join(unknown)}", 404) + return wanted + for preferred in ("image6", "main_shelf"): + if preferred in samples: + return [preferred] + return [next(iter(samples))] if samples else [] + + def _demo_window(width: Optional[int], height: Optional[int]) -> tuple: + win_w = int(width or 960) + win_h = int(height or 540) + if not (160 <= win_w <= 2048 and 120 <= win_h <= 2048): + raise _error("BAD_REQUEST", "window must be between 160x120 and 2048x2048", 400) + return win_w, win_h + + @app.get("/v1/demo/plan", dependencies=[Depends(authorize)]) + async def demo_plan(samples_param: Optional[str] = None, width: Optional[int] = None, + height: Optional[int] = None) -> Dict[str, Any]: + """Geometry of the virtual walk (demo-only capture source).""" + try: + ids = _demo_samples(samples_param) + if not ids: + raise _error("NOT_FOUND", "no samples available", 404) + return {"status": "success", **demo_camera.plan(ids, _demo_window(width, height))} + except DemoSourceError as exc: + raise _error("NOT_FOUND", str(exc), 404) + + @app.get("/v1/demo/frame", dependencies=[Depends(authorize)]) + async def demo_frame(step: int = 0, samples_param: Optional[str] = None, + width: Optional[int] = None, height: Optional[int] = None, + quality: int = 65) -> Response: + """One JPEG frame of the virtual walk: a store employee sweeping the shelf. + + Demo-only. The client advances `step` (it wraps), so the walk's speed + follows whatever frame rate the client can sustain. + """ + try: + ids = _demo_samples(samples_param) + if not ids: + raise _error("NOT_FOUND", "no samples available", 404) + payload, position = demo_camera.frame( + ids, step, _demo_window(width, height), max(30, min(95, int(quality))) + ) + except DemoSourceError as exc: + raise _error("NOT_FOUND", str(exc), 404) + return Response( + content=payload, + media_type="image/jpeg", + headers={ + "X-Demo-Sample": str(position["sample"]), + "X-Demo-Bay": str(position["bay_index"] + 1), + "X-Demo-Step": str(position["step"]), + "X-Demo-Steps": str(position["steps"]), + "X-Demo-X": str(position["x"]), + "X-Demo-Y": str(position["y"]), + }, + ) @app.get("/v1/samples/{sample_id}", dependencies=[Depends(authorize)]) async def get_sample(sample_id: str) -> Response: diff --git a/recognization/code/demo.py b/recognization/code/demo.py new file mode 100644 index 0000000..819e7f6 --- /dev/null +++ b/recognization/code/demo.py @@ -0,0 +1,248 @@ +""" +Synthetic camera for demos: a store employee walking a shelf. + +A simulator has no camera, and a still image makes a poor demo - nothing moves, so +nothing new is ever recognised. This module turns the demo shelf photos into a +virtual walk: a window pans left to right along a row, steps up to the next row, +and pans back the other way (serpentine, ascending), exactly the way a person +sweeps a shelf with a tablet. + +The client drives only `step`; all geometry lives here, so the web client, the +phone app, and the reference client see the same walk. `step` wraps, so an +unattended demo loops forever. + +Demo-only: this is a capture source, not a perception feature. `/v1/demo/*` is +documented as such. +""" + +from __future__ import annotations + +import math +import os +import threading +import time +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Sequence, Tuple + +DEFAULT_WINDOW = (960, 540) +DEFAULT_QUALITY = 65 +# How far the window advances per step, as a fraction of its own size. This is +# the demo's most important number: advance too fast and a product is in view for +# one or two frames, so the classifier never gets a chance and nothing is ever +# labelled (measured: 0.22 gave 9 labelled tracks in 20 s; the shelf swept past +# faster than a person could read it). 0.06 keeps a facing in view for roughly a +# second at demo frame rates - a deliberate, human sweep. +PAN_OVERLAP = 0.06 +ROW_OVERLAP = 0.25 +# Vertical steps spent moving between rows (the "look up, then keep going" beat). +TRANSITION_STEPS = 6 + + +class DemoSourceError(ValueError): + """Unknown sample or an impossible window.""" + + +@dataclass +class Bay: + """One shelf photo in the walk, with its geometry.""" + + sample_id: str + path: str + width: int + height: int + steps: int + + +class DemoCamera: + """Plans and renders the virtual walk over one or more shelf photos.""" + + def __init__(self, samples: Dict[str, Dict[str, str]], window: Tuple[int, int] = DEFAULT_WINDOW): + self._samples = samples + self._window = window + self._lock = threading.Lock() + self._images: Dict[str, Any] = {} + self._sizes: Dict[str, Tuple[int, int]] = {} + self._plans: Dict[Tuple[str, ...], List[Bay]] = {} + + # ------------------------------------------------------------------ planning + def _size(self, sample_id: str) -> Tuple[int, int]: + cached = self._sizes.get(sample_id) + if cached is not None: + return cached + item = self._samples.get(sample_id) + if item is None or not os.path.exists(item["path"]): + raise DemoSourceError(f"unknown sample '{sample_id}'") + import cv2 + + image = cv2.imread(item["path"]) + if image is None: + raise DemoSourceError(f"cannot read sample '{sample_id}'") + with self._lock: + self._images[sample_id] = image + self._sizes[sample_id] = (int(image.shape[1]), int(image.shape[0])) + return self._sizes[sample_id] + + def _image(self, sample_id: str) -> Any: + if sample_id not in self._images: + self._size(sample_id) + return self._images[sample_id] + + def steps_for(self, sample_id: str, window: Tuple[int, int]) -> int: + width, height = self._size(sample_id) + win_w, win_h = window + bands = self._bands(height, win_h) + per_band = self._pan_steps(width, win_w) + # Every band pans, and every band after the first needs the step-up. + return bands * per_band + max(0, bands - 1) * TRANSITION_STEPS + + def _bands(self, image_height: int, window_height: int) -> int: + if window_height >= image_height: + return 1 + stride = max(1, int(window_height * (1.0 - ROW_OVERLAP))) + return max(1, int(math.ceil((image_height - window_height) / stride)) + 1) + + def _pan_steps(self, image_width: int, window_width: int) -> int: + if window_width >= image_width: + return 1 + stride = max(1, int(window_width * PAN_OVERLAP)) + return max(1, int(math.ceil((image_width - window_width) / stride)) + 1) + + def plan(self, sample_ids: Sequence[str], window: Optional[Tuple[int, int]] = None) -> Dict[str, Any]: + window = window or self._window + key = (tuple(sample_ids), window) + bays = self._plans.get(key) + if bays is None: + bays = [] + for sample_id in sample_ids: + width, height = self._size(sample_id) + bays.append( + Bay( + sample_id=sample_id, + path=self._samples[sample_id]["path"], + width=width, + height=height, + steps=self.steps_for(sample_id, window), + ) + ) + self._plans[key] = bays + total = sum(bay.steps for bay in bays) + return { + "samples": [bay.sample_id for bay in bays], + "window": {"width": window[0], "height": window[1]}, + "bays": [ + {"sample": bay.sample_id, "width": bay.width, "height": bay.height, + "steps": bay.steps, "first_step": sum(b.steps for b in bays[:index])} + for index, bay in enumerate(bays) + ], + "steps": total, + "row_overlap": ROW_OVERLAP, + "pan_overlap": PAN_OVERLAP, + } + + # ------------------------------------------------------------------ rendering + def window_at( + self, sample_ids: Sequence[str], step: int, window: Optional[Tuple[int, int]] = None + ) -> Dict[str, Any]: + """Where the virtual camera is at this step (wraps around the walk).""" + window = window or self._window + plan = self.plan(sample_ids, window) + total = plan["steps"] + step = int(step) % max(1, total) + offset = 0 + for index, bay in enumerate(plan["bays"]): + if step < offset + bay["steps"]: + local = step - offset + x, y = self._position(local, bay["width"], bay["height"], window) + return { + "sample": bay["sample"], + "bay_index": index, + "step": step, + "steps": total, + "local_step": local, + "bay_steps": bay["steps"], + "x": x, + "y": y, + "width": min(window[0], bay["width"]), + "height": min(window[1], bay["height"]), + } + offset += bay["steps"] + raise DemoSourceError("step outside the walk") + + def _position(self, local_step: int, image_width: int, image_height: int, window: Tuple[int, int]) -> Tuple[int, int]: + """Serpentine: pan a row, step up, pan back.""" + win_w = min(window[0], image_width) + win_h = min(window[1], image_height) + bands = self._bands(image_height, win_h) + per_band = self._pan_steps(image_width, win_w) + cycle = per_band + TRANSITION_STEPS + band = min(bands - 1, local_step // cycle) + within = local_step - band * cycle + stride_y = 0 if bands == 1 else max(1, int(win_h * (1.0 - ROW_OVERLAP))) + # Ascending: band 0 is the bottom of the shelf. + y_bottom = min(image_height - win_h, band * stride_y) + y = max(0, image_height - win_h - y_bottom) + stride_x = max(1, int(win_w * PAN_OVERLAP)) + max_x = max(0, image_width - win_w) + if within < per_band: + pan = within + x = min(max_x, pan * stride_x) + else: + # The step-up beat: hold x at the end of the row and climb. + transition = min(TRANSITION_STEPS, max(1, within - per_band + 1)) + pan = per_band - 1 + x = min(max_x, pan * stride_x) + next_band = min(bands - 1, band + 1) + next_y_bottom = min(image_height - win_h, next_band * stride_y) + ratio = transition / float(TRANSITION_STEPS) + y = int(round(y - (next_y_bottom - y_bottom) * ratio)) + if band % 2 == 1 and within < per_band: + # Odd bands pan back the other way, like a person reversing direction. + x = max(0, max_x - x) + return int(max(0, min(max_x, x))), int(max(0, min(image_height - win_h, y))) + + def frame( + self, + sample_ids: Sequence[str], + step: int, + window: Optional[Tuple[int, int]] = None, + quality: int = DEFAULT_QUALITY, + ) -> Tuple[bytes, Dict[str, Any]]: + """The JPEG frame the virtual camera sees at this step, plus its geometry.""" + import cv2 + + position = self.window_at(sample_ids, step, window) + image = self._image(position["sample"]) + x, y = position["x"], position["y"] + w, h = position["width"], position["height"] + crop = image[y:y + h, x:x + w] + if crop.size == 0: + raise DemoSourceError("empty crop") + ok, buffer = cv2.imencode(".jpg", crop, [cv2.IMWRITE_JPEG_QUALITY, int(quality)]) + if not ok: + raise DemoSourceError("JPEG encode failed") + return bytes(buffer), position + + def warm(self, sample_ids: Sequence[str]) -> None: + """Decode the bays up front so the first demo frame is not the slow one.""" + for sample_id in sample_ids: + self._size(sample_id) + + +if __name__ == "__main__": # smoke: python code/demo.py [sample ...] + import sys + + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from app import get_available_shelf_images + + samples = {item["id"]: item for item in get_available_shelf_images("add")} + camera = DemoCamera(samples) + ids = sys.argv[1:] or ["image6"] + plan = camera.plan(ids) + print(f"walk over {plan['samples']}: {plan['steps']} steps, window {plan['window']}") + started = time.perf_counter() + for step in range(min(plan["steps"], 24)): + payload, position = camera.frame(ids, step) + print(f" step {step:3d} {position['sample']} x={position['x']:4d} y={position['y']:4d} " + f"{len(payload) // 1024} KB") + elapsed = (time.perf_counter() - started) * 1000 / max(1, min(plan["steps"], 24)) + print(f"{elapsed:.1f} ms per frame") diff --git a/recognization/code/live.py b/recognization/code/live.py index f6cb256..ecd6d1b 100644 --- a/recognization/code/live.py +++ b/recognization/code/live.py @@ -64,10 +64,13 @@ class LiveSettings: tags: bool = True min_hits: int = 2 min_crop_px: int = 40 - max_attempts: int = 3 + max_attempts: int = 2 retry_gap: int = 8 + # A product drifting out of the frame is not worth a retry: the budget belongs + # to what is still in front of the camera. + retry_edge_ratio: float = 0.06 area_growth: float = 1.3 - max_inflight_per_session: int = 12 + max_inflight_per_session: int = 16 edge_margin_ratio: float = 0.015 glare_fraction: float = 0.02 glare_pixel: int = 245 @@ -545,8 +548,14 @@ class LiveSession: continue if track.attempts >= self.settings.max_attempts: continue - if track.attempts > 0 and (seq - track.last_attempt_seq) < self.settings.retry_gap: - if track.area() < self.settings.area_growth * max(1, track.last_attempt_area): + if track.attempts > 0: + if (seq - track.last_attempt_seq) < self.settings.retry_gap: + if track.area() < self.settings.area_growth * max(1, track.last_attempt_area): + continue + retry_margin_x = int(frame_w * self.settings.retry_edge_ratio) + retry_margin_y = int(frame_h * self.settings.retry_edge_ratio) + if (x1 <= retry_margin_x or y1 <= retry_margin_y + or x2 >= frame_w - retry_margin_x or y2 >= frame_h - retry_margin_y): continue eligible.append(track) diff --git a/recognization/code/tags.py b/recognization/code/tags.py index 902f36c..36c3091 100644 --- a/recognization/code/tags.py +++ b/recognization/code/tags.py @@ -400,7 +400,13 @@ class TagService: # ------------------------------------------------------------------ frame loop def plan(self, boxes: Sequence[Dict[str, Any]], image_shape: Sequence[int]) -> List[RailBand]: - """Bands worth OCR-ing right now (must be called from the frame loop).""" + """Bands worth OCR-ing right now (must be called from the frame loop). + + A band is a shelf row, so its key is the row - not its on-screen position. + Keying by position meant a walking camera minted a fresh key every few + pixels of pan and re-read the same shelf continuously, which starved + classification of CPU (measured: 10x slower per crop while OCR ran). + """ if not self.settings.enabled or not self.engine.available(): return [] bands = rail_bands(boxes, image_shape) @@ -411,7 +417,7 @@ class TagService: seq = self._frame_seq chosen: List[RailBand] = [] for band in bands: - key = (band.row_index, band.box[0] // 32, band.box[2] // 32) + key = band.row_index if key in self._pending_bands: continue last = self._last_pass_seq.get(key) @@ -424,9 +430,8 @@ class TagService: if len(chosen) >= self.settings.max_bands_per_pass: break for band in chosen: - key = (band.row_index, band.box[0] // 32, band.box[2] // 32) - self._last_pass_seq[key] = seq - self._pending_bands.add(key) + self._last_pass_seq[band.row_index] = seq + self._pending_bands.add(band.row_index) return chosen def run_pass(self, image: Any, band: RailBand, scale_bias: float = 1.0) -> List[TagObservation]: @@ -436,7 +441,7 @@ class TagService: independent reads of the same tag are what makes the vote meaningful (identical pixels would produce identical output). """ - key = (band.row_index, band.box[0] // 32, band.box[2] // 32) + key = band.row_index try: x1, y1, x2, y2 = band.box crop = image[y1:y2, x1:x2] diff --git a/recognization/code/test_demo.py b/recognization/code/test_demo.py new file mode 100644 index 0000000..b62ca5c --- /dev/null +++ b/recognization/code/test_demo.py @@ -0,0 +1,119 @@ +""" +Contract tests for the demo camera (the synthetic store walk). + +The walk is what a demo shows when there is no camera, so its geometry has to be +predictable: inside the image, moving, serpentine, and wrapping forever. +""" + +from __future__ import annotations + +import os +import sys +import unittest + +current_dir = os.path.dirname(os.path.abspath(__file__)) +project_root = os.path.dirname(current_dir) +if os.path.exists(os.path.join(project_root, "stock")): + os.chdir(project_root) +if current_dir not in sys.path: + sys.path.insert(0, current_dir) + +import numpy as np # noqa: E402 + +from demo import DemoCamera, DemoSourceError # noqa: E402 + + +class FakeImageCamera(DemoCamera): + """A camera over a synthetic image, so the geometry is testable without files.""" + + def __init__(self, width=1500, height=2000, window=(960, 540)): + super().__init__({"synthetic": {"id": "synthetic", "path": "synthetic", "label": "synthetic"}}, window) + self._image_cache = np.zeros((height, width, 3), dtype="uint8") + self._size_cache = (width, height) + + def _size(self, sample_id): # noqa: D102 + if sample_id != "synthetic": + raise DemoSourceError(f"unknown sample '{sample_id}'") + return self._size_cache + + def _image(self, sample_id): # noqa: D102 + return self._image_cache + + +class WalkGeometryTests(unittest.TestCase): + def setUp(self): + self.camera = FakeImageCamera() + self.plan = self.camera.plan(["synthetic"]) + + def test_plan_is_deterministic_and_covers_the_shelf(self): + self.assertEqual(self.plan["steps"], self.camera.plan(["synthetic"])["steps"]) + self.assertGreater(self.plan["steps"], 20, "a shelf walk is not two steps long") + self.assertEqual(self.plan["bays"][0]["sample"], "synthetic") + + def test_window_never_leaves_the_image(self): + width, height = 1500, 2000 + for step in range(self.plan["steps"]): + position = self.camera.window_at(["synthetic"], step) + self.assertGreaterEqual(position["x"], 0) + self.assertGreaterEqual(position["y"], 0) + self.assertLessEqual(position["x"] + position["width"], width) + self.assertLessEqual(position["y"] + position["height"], height) + + def test_the_camera_actually_moves(self): + xs = [self.camera.window_at(["synthetic"], step)["x"] for step in range(6)] + ys = [self.camera.window_at(["synthetic"], step)["y"] for step in range(self.plan["steps"])] + self.assertGreater(len(set(xs)), 1, "the window must pan, not sit still") + self.assertGreater(len(set(ys)), 1, "the walk must climb rows") + + def test_it_ascends_the_shelf(self): + first = self.camera.window_at(["synthetic"], 0)["y"] + last_band = self.plan["steps"] - 1 + self.assertGreater(first, self.camera.window_at(["synthetic"], last_band)["y"], + "the walk starts at the bottom row and climbs") + + def test_step_wraps(self): + first = self.camera.window_at(["synthetic"], 0) + wrapped = self.camera.window_at(["synthetic"], self.plan["steps"]) + self.assertEqual((first["x"], first["y"]), (wrapped["x"], wrapped["y"])) + + def test_frames_are_deterministic_jpegs(self): + payload, position = self.camera.frame(["synthetic"], 3, quality=70) + again, _ = self.camera.frame(["synthetic"], 3, quality=70) + self.assertTrue(payload.startswith(b"\xff\xd8"), "JPEG magic") + self.assertEqual(payload, again, "the same step must render the same frame") + self.assertIn("sample", position) + + def test_unknown_sample_is_rejected(self): + with self.assertRaises(DemoSourceError): + self.camera.plan(["nope"]) + + def test_window_larger_than_the_image_is_clamped(self): + camera = FakeImageCamera(width=800, height=600, window=(960, 540)) + position = camera.window_at(["synthetic"], 0) + self.assertLessEqual(position["width"], 800) + self.assertLessEqual(position["height"], 600) + self.assertEqual(position["x"], 0) + # The walk starts at the bottom of the shelf, so the window is bottom-aligned. + self.assertEqual(position["y"], 600 - position["height"]) + + +class RealShelfTests(unittest.TestCase): + """One check against the actual demo asset, skipped when it is absent.""" + + def test_real_sample_walks(self): + from app import get_available_shelf_images + + samples = {item["id"]: item for item in get_available_shelf_images("add")} + if "image6" not in samples: + self.skipTest("demo sample image6 not present") + camera = DemoCamera(samples) + plan = camera.plan(["image6"], (1280, 720)) + self.assertGreater(plan["steps"], 20, "a 1500x2000 shelf gives a multi-row walk") + payload, position = camera.frame(["image6"], 5, (1280, 720)) + self.assertGreater(len(payload), 10_000) + self.assertLessEqual(position["x"] + position["width"], plan["bays"][0]["width"]) + self.assertLessEqual(position["y"] + position["height"], plan["bays"][0]["height"]) + + +if __name__ == "__main__": + unittest.main()