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>
This commit is contained in:
kiyo9w
parent
a18a80a21e
commit
00367a95c9
5 changed files with 466 additions and 16 deletions
|
|
@ -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:
|
||||
|
|
|
|||
248
recognization/code/demo.py
Normal file
248
recognization/code/demo.py
Normal file
|
|
@ -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")
|
||||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
119
recognization/code/test_demo.py
Normal file
119
recognization/code/test_demo.py
Normal file
|
|
@ -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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue