4960553254
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>
122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
"""
|
|
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()
|