e67ab8bfb4
From the deep review's hygiene list: - The demo replay (a still recognition replayed as if it were live tracking) moves out of app.py into demo_replay.py, and every payload it produces is marked `synthetic: true`. It had been committed inside an unrelated feature commit, which made a film-making device look like part of the server. - `/v1/health` reports the OCR engine and whether it is available. Without it an operator could not tell a broken OCR from a shelf with no readable tags. - A failed OCR pass is counted and logged instead of swallowed: `TagService.stats()` now carries `passes_failed` and `last_pass_error`. - The live classification budget goes back to the measured production defaults (16 in flight, 2 attempts); a demo walk over a nearly static shelf asks for more through `--live-inflight` / `--live-attempts` rather than silently changing the default for everyone. 162 tests pass. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
162 lines
6.9 KiB
Python
162 lines
6.9 KiB
Python
"""
|
|
Demo-only: replay a shelf's recognition as if it were live tracking.
|
|
|
|
`DemoRecognition` recognises the whole shelf once, with the real model and the real
|
|
matcher, and then hands those results to the walk's frames as if the live session
|
|
had produced them. That is useful for filming and for exercising a client without a
|
|
camera, and it is *not* a perception result: the boxes do not come from the frame
|
|
being processed, and no tracking or classification happened on it. Every payload it
|
|
touches is marked `synthetic: true` so a client can tell.
|
|
|
|
It lives here rather than in `app.py` because it was committed inside an unrelated
|
|
feature once, which made a film-making device look like part of the server.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from demo import DemoSourceError
|
|
|
|
|
|
class DemoRecognition:
|
|
"""The demo shortcut.
|
|
|
|
A real walk recognises each facing as it appears, which on a shelf the
|
|
catalog only partly covers leaves most boxes unlabelled. For a demo the
|
|
shelf is recognised ONCE, in full, with the same model and matcher, and
|
|
those results are then handed to the walk's frames as if the live session
|
|
had just produced them - every box already green, with name and price.
|
|
Demo-only: enabled by --demo-walk.
|
|
"""
|
|
|
|
_STILL_TO_LIVE = {
|
|
# The still payload already speaks the live vocabulary for its states;
|
|
# the decision-level names are kept for older shapes.
|
|
"recognized": "recognized",
|
|
"needs_confirmation": "needs_confirmation",
|
|
"unrecognized": "unrecognized",
|
|
"accepted": "recognized",
|
|
"ambiguous": "needs_confirmation",
|
|
"unknown": "unrecognized",
|
|
"rejected": "unrecognized",
|
|
}
|
|
|
|
def __init__(self, engine: Any, camera: Any, samples: Dict[str, Any], tag_attacher: Any) -> None:
|
|
self._engine = engine
|
|
self._camera = camera
|
|
self._samples = samples
|
|
self._tag_attacher = tag_attacher
|
|
self._catalog_by_sku = {product["sku_id"]: product for product in engine.catalog()}
|
|
self._cache: Dict[str, Any] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def steps(self, sample_id: str) -> int:
|
|
return self._camera.plan([sample_id], (1280, 720)).get("steps", 0)
|
|
|
|
def recognition(self, sample_id: str) -> Any:
|
|
cached = self._cache.get(sample_id)
|
|
if cached is not None:
|
|
return cached
|
|
import cv2
|
|
|
|
path = self._samples[sample_id]["path"]
|
|
image = cv2.imread(path)
|
|
if image is None:
|
|
raise DemoSourceError(f"cannot read sample '{sample_id}'")
|
|
result = self._engine.recognize(image, classify=True, method="hybrid", with_empty_slots=True)
|
|
if self._tag_attacher is not None:
|
|
result = self._tag_attacher(result, image)
|
|
with self._lock:
|
|
self._cache[sample_id] = (result, int(image.shape[1]), int(image.shape[0]))
|
|
return self._cache[sample_id]
|
|
|
|
def warm(self, sample_id: str) -> None:
|
|
self.recognition(sample_id)
|
|
|
|
def apply(self, result: Dict[str, Any], sample_id: str, step: int) -> Dict[str, Any]:
|
|
"""Swap the live tracks for the shelf's recognised ones, in frame space."""
|
|
still, _, _ = self.recognition(sample_id)
|
|
position = self._camera.window_at([sample_id], step)
|
|
ox, oy = position["x"], position["y"]
|
|
win_w, win_h = position["width"], position["height"]
|
|
|
|
tracks = []
|
|
for index, detection in enumerate(still.get("detections") or []):
|
|
box = [int(v) for v in detection.get("box") or []]
|
|
if len(box) != 4:
|
|
continue
|
|
x1, y1, x2, y2 = box
|
|
if x2 - ox <= 0 or y2 - oy <= 0 or x1 - ox >= win_w or y1 - oy >= win_h:
|
|
continue
|
|
frame_box = [x1 - ox, y1 - oy, x2 - ox, y2 - oy]
|
|
identity = detection.get("identity") or {}
|
|
state = self._STILL_TO_LIVE.get(identity.get("state") or "", "unrecognized")
|
|
sku_id = identity.get("sku_id")
|
|
product = self._catalog_by_sku.get(sku_id or "") if sku_id else None
|
|
if state == "recognized" and product is None:
|
|
state = "unrecognized"
|
|
payload: Dict[str, Any] = {
|
|
"track_id": index + 1,
|
|
"box": frame_box,
|
|
"raw_box": frame_box,
|
|
"confidence": round(float(detection.get("confidence") or 0.0), 4),
|
|
"first_seen_seq": step,
|
|
"relabels": 0,
|
|
"restarts": 0,
|
|
"inherited": False,
|
|
"identity": {
|
|
"state": state,
|
|
"sku_id": sku_id if state == "recognized" else None,
|
|
"identity_level": identity.get("identity_level"),
|
|
"confidence": identity.get("confidence"),
|
|
"score": identity.get("score"),
|
|
"candidates": list(identity.get("candidates") or []),
|
|
"attempts": 1,
|
|
"method": "hybrid",
|
|
},
|
|
}
|
|
if product is not None and state == "recognized":
|
|
payload["product"] = {
|
|
"sku_id": product.get("sku_id"),
|
|
"product_name": product.get("product_name"),
|
|
"price": product.get("price"),
|
|
}
|
|
tracks.append(payload)
|
|
|
|
slots = []
|
|
for slot in still.get("empty_slots") or []:
|
|
box = [int(v) for v in slot.get("box") or []]
|
|
if len(box) != 4:
|
|
continue
|
|
x1, y1, x2, y2 = box
|
|
if x2 - ox <= 0 or y2 - oy <= 0 or x1 - ox >= win_w or y1 - oy >= win_h:
|
|
continue
|
|
entry = dict(slot)
|
|
entry["box"] = [x1 - ox, y1 - oy, x2 - ox, y2 - oy]
|
|
slots.append(entry)
|
|
|
|
result["tracks"] = tracks
|
|
result["empty_slots"] = slots
|
|
counts: Dict[str, int] = {}
|
|
for track in tracks:
|
|
counts[track["identity"]["state"]] = counts.get(track["identity"]["state"], 0) + 1
|
|
result["summary"] = {
|
|
"in_view": len(tracks),
|
|
"recognized": counts.get("recognized", 0),
|
|
"needs_confirmation": counts.get("needs_confirmation", 0),
|
|
"unrecognized": counts.get("unrecognized", 0),
|
|
"detected": 0,
|
|
"unconfirmed": counts.get("unconfirmed", 0),
|
|
"identifying": 0,
|
|
"empty_slots": len(slots),
|
|
"price_tags": (still.get("price_tags") or {"readable": 0, "unreadable": 0}),
|
|
"reconciliation": (still.get("reconciliation") or {}),
|
|
"compliance": (still.get("compliance") or {"violations": 0, "rows": [], "hints": []}),
|
|
}
|
|
# A client must be able to tell these boxes did not come from the frame it
|
|
# just sent.
|
|
result["synthetic"] = True
|
|
result["demo"] = {"sample": sample_id, "step": step, "bay": position["bay_index"] + 1}
|
|
return result
|