tngo/product-recognition-demo
1
0
Fork 0
forked from vnguyen/product-recognition-demo
Code Pull requests Activity

perf: bound OCR so a walking scan stays real-time, and add walk pacing

Browse source 
macOS Vision holds the GIL for the ~124 ms it takes to read a rail band, so the
tag pass decides the live frame rate: with 2 bands every 15 frames a walking
session fell to 1.7 results/s, and 1 band every 30 frames brings it back to 7.2
with tags still settling in a second or two. Measured, not guessed - the same
walk, same method (orb), only the OCR duty cycle changed.

The walk also gained a `pan` parameter (default 0.06, clamped 0.005-0.5) so a
demo can choose its pace: pan=0.02 sweeps a row over ~8 s, which reads like a
person rather than a time-lapse.

Demo configuration measured end to end on the walk (20 s, 1280x720, one client):
orb, pan=0.02, 1 band/30 frames -> 7.2 results/s, 178 labelled tracks, 21
distinct SKUs, first label 0.7 s, 3 readable price tags.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kiyo9w 2026-09-24 18:14:32 +09:00
parent 00367a95c9
commit d38d863873
3 changed files with 34 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

21
recognization/code/app.py
View file

@ -245,6 +245,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)
demo_cameras: Dict[float, DemoCamera] = {}
sessions: Dict[str, LiveSession] = {}
sessions_lock = asyncio.Lock()
app.state.engine = engine
@ -328,22 +329,34 @@ def create_app(
raise _error("BAD_REQUEST", "window must be between 160x120 and 2048x2048", 400)
return win_w, win_h
def _demo_camera(pan: Optional[float]) -> DemoCamera:
"""One camera per pacing value (the walk geometry depends on it)."""
if pan is None:
return demo_camera
key = round(float(pan), 4)
camera = demo_cameras.get(key)
if camera is None:
camera = DemoCamera(samples, pan_overlap=key)
demo_cameras[key] = camera
return camera
@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]:
height: Optional[int] = None, pan: Optional[float] = 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))}
camera = _demo_camera(pan)
return {"status": "success", **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:
quality: int = 65, pan: Optional[float] = None) -> 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
@ -353,7 +366,7 @@ def create_app(
ids = _demo_samples(samples_param)
if not ids:
raise _error("NOT_FOUND", "no samples available", 404)
payload, position = demo_camera.frame(
payload, position = _demo_camera(pan).frame(
ids, step, _demo_window(width, height), max(30, min(95, int(quality)))
)
except DemoSourceError as exc: