docs: API contract, README for the server and live mode, real-model test

docs/API.md is the full contract: run flags and environment, auth, coordinate
convention, the identity vocabulary, every REST endpoint and WebSocket message
with examples, the backpressure rule, what a client may rely on, the design
decisions with their rejected alternatives, measured performance, known
limitations, and notes for the future mobile client.

README.md (Vietnamese) now documents the client-server shape, both modes, the
flags, the API, the test commands, and the source layout. It also repairs the
PowerShell escape damage in the original file (backtick-r had become a line
break, backtick-b a backspace, backtick-a a bell), which had left the code
fences and file paths mangled.

test_real_model.py is the opt-in integration test (DORAYAKI_REAL_MODEL=1): it
guards the measured baseline (170 detections, 105 recognized on image.jpg) and
proves a pan keeps labels stable with no unexplained changes.

run_app.sh / run_app.bat pass arguments through to the server.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kiyo9w 2026-09-24 12:46:54 +09:00
parent da2bed1c35
commit 4960553254
5 changed files with 568 additions and 77 deletions

View file

@ -0,0 +1,122 @@
"""
Opt-in integration test against the real model and the real catalog.
Skipped unless DORAYAKI_REAL_MODEL=1, because it loads the YOLO weights and
classifies a full shelf. It defends the end-to-end claim: the server detects the
whole shelf, accepts a meaningful share of the products, keeps a label stable
while the camera pans, and never names a SKU for a weak match.
cd recognization/code
DORAYAKI_REAL_MODEL=1 YOLO_OFFLINE=1 ../.venv/bin/python -m unittest test_real_model -v
"""
from __future__ import annotations
import math
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 cv2 # noqa: E402
from engine import EngineSettings, build_default_engine # noqa: E402
from live import LiveSession, LiveSettings # noqa: E402
ENABLED = os.environ.get("DORAYAKI_REAL_MODEL") == "1"
# Measured baseline on image.jpg (2026-09-24): 170 detections, 105 recognized.
BASELINE_DETECTIONS = 170
BASELINE_RECOGNIZED = 105
@unittest.skipUnless(ENABLED, "set DORAYAKI_REAL_MODEL=1 to run against the real model")
class RealModelTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.engine = build_default_engine(EngineSettings(device="cpu"))
cls.engine.warm_up()
@classmethod
def tearDownClass(cls):
cls.engine.shutdown()
def test_recognize_whole_shelf(self):
with open("image.jpg", "rb") as handle:
payload = handle.read()
image = self.engine.decode_image(payload)
result = self.engine.recognize(image, classify=True)
summary = result["summary"]
self.assertGreaterEqual(
summary["in_view"], int(BASELINE_DETECTIONS * 0.95),
"detection must not regress below the measured baseline",
)
self.assertGreaterEqual(
summary["recognized"], int(BASELINE_RECOGNIZED * 0.9),
"recognition must not regress below the measured baseline",
)
# every named SKU must exist in the catalog
catalog = {product["sku_id"] for product in self.engine.catalog()}
for detection in result["detections"]:
identity = detection["identity"]
if identity["state"] == "recognized":
self.assertIn(identity["sku_id"], catalog)
else:
self.assertIsNone(identity["sku_id"], "only an accepted match may name a SKU")
def test_pan_keeps_labels_stable_and_advances_them(self):
image = cv2.imread("image.jpg")
height, width = image.shape[:2]
frame_w, frame_h = 1280, 720
session = LiveSession("real", self.engine, settings=LiveSettings(method="hybrid"))
session.refresh_catalog()
labels: dict = {}
relabels: dict = {}
restarts: dict = {}
unexplained = []
recognized_frames = 0
frames = 16
for index in range(frames):
phase = index / (frames - 1)
x = int((width - frame_w) * (0.5 - 0.5 * math.cos(2 * math.pi * phase)))
y = max(0, (height - frame_h) // 2)
window = image[y:y + frame_h, x:x + frame_w]
ok, buffer = cv2.imencode(".jpg", window, [cv2.IMWRITE_JPEG_QUALITY, 75])
session.submit_frame(index, buffer.tobytes())
pending = session.take_pending(timeout=5.0)
self.assertIsNotNone(pending)
result = session.process_frame(*pending)
import time
time.sleep(0.4) # let classification land, as a live client's next frame would
if result["summary"]["recognized"]:
recognized_frames += 1
for track in result["tracks"]:
identity = track["identity"]
if identity["state"] != "recognized":
continue
track_id = track["track_id"]
sku = identity["sku_id"]
if restarts.get(track_id, 0) < track["restarts"]:
labels.pop(track_id, None)
if track_id in labels and labels[track_id] != sku:
if track["relabels"] <= relabels.get(track_id, 0):
unexplained.append((track_id, labels[track_id], sku))
labels[track_id] = sku
relabels[track_id] = track["relabels"]
restarts[track_id] = track["restarts"]
self.assertGreater(recognized_frames, frames // 2, "labels should appear on most frames")
self.assertGreater(len(labels), 0, "the pan should label products")
self.assertEqual(unexplained, [], "a label must never change without re-verification")
if __name__ == "__main__":
unittest.main()