From eca03e0c9179e59a98f34309b949a3601c66c30f Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Mon, 28 Sep 2026 17:47:50 +0900 Subject: [PATCH] refactor: delete the legacy session model Two session models shared one process. `/v1/*` and `/v1/stream` are the live one; `/api/sessions*` and `/api/compare_live_vs_still` were the earlier model, still constructed on every `create_app()` through `build_legacy_scans` even though nothing called them. The web client and `stream_client.py` speak only `/v1`. Deleted: `observe.py` (ScanService/PerceptionService/PerceptionAdapter), `yolo_adapter.py` (imported only by the legacy builder), `pipeline.py` and `test_suite.py` (the old single-image CLI, reachable only from each other), `test_observe.py` (its Session* tests), the eight legacy route handlers, `build_legacy_scans` with its two facades, `create_fastapi_app`, and the `scans` parameter on `create_app`. `IdentityRankingTests` moved to `test_identity.py` before `test_observe.py` was deleted: those five ranking rules are the live path's contract, not the legacy model's. `test_api.py` keeps every `/v1` and WebSocket test. Verified: 246 tests OK (2 skipped), and a route dump of a freshly built app shows no `/api/*` path with all nine `/v1` routes and the WebSocket intact. --- recognization/README.md | 2 +- recognization/code/app.py | 169 ---------- recognization/code/observe.py | 474 ---------------------------- recognization/code/pipeline.py | 179 ----------- recognization/code/test_api.py | 51 +-- recognization/code/test_identity.py | 52 +++ recognization/code/test_observe.py | 319 ------------------- recognization/code/test_suite.py | 91 ------ recognization/code/yolo_adapter.py | 64 ---- recognization/docs/API.md | 1 - 10 files changed, 54 insertions(+), 1348 deletions(-) delete mode 100644 recognization/code/observe.py delete mode 100644 recognization/code/pipeline.py create mode 100644 recognization/code/test_identity.py delete mode 100644 recognization/code/test_observe.py delete mode 100644 recognization/code/test_suite.py delete mode 100644 recognization/code/yolo_adapter.py diff --git a/recognization/README.md b/recognization/README.md index 20e4f23..7acdfdd 100644 --- a/recognization/README.md +++ b/recognization/README.md @@ -174,7 +174,7 @@ code/ detector.py # YOLO detector feature_matcher.py # SURF/SIFT, ORB, BRISK, Hybrid identity.py # accepted / ambiguous / unknown / rejected - empty_shelf_detector.py, inventory.py, pipeline.py, observe.py + empty_shelf_detector.py, inventory.py, gallery.py, tags.py, tag_cards.py stream_client.py # client mẫu (camera / video / pan ảnh tĩnh) bench_engine.py # đo tốc độ test_*.py # unittest modules (discover -s code -p 'test_*.py') diff --git a/recognization/code/app.py b/recognization/code/app.py index 2735fb7..f6be125 100644 --- a/recognization/code/app.py +++ b/recognization/code/app.py @@ -250,39 +250,6 @@ def get_available_shelf_images(add_dir: str = "add") -> List[Dict[str, str]]: return items -def build_legacy_scans(engine: RecognitionEngine): - """The /api/sessions* routes kept from the earlier session model, now backed - by the real engine instead of the fake adapter.""" - from observe import PerceptionService, ScanService - from yolo_adapter import YoloMatcherAdapter - - class _DetectorFacade: - def __init__(self, engine: RecognitionEngine): - self.engine = engine - self.weights_path = engine.settings.weights_path - - def detect(self, image, conf_threshold: float = 0.25, iou_threshold: float = 0.45): - boxes = self.engine.detect(image, conf=conf_threshold) - return {"status": "success", "boxes": boxes, "total_detected": len(boxes)} - - def detect_empty_slots(self, image_shape, boxes, min_gap_ratio: float = 0.75): - return self.engine.empty_slots(image_shape, boxes) - - def crop_roi(self, image, box): - return self.engine.crop(image, box) - - class _MatcherFacade: - def __init__(self, engine: RecognitionEngine): - self.engine = engine - - def classify_crop(self, crop, method: str = "hybrid"): - scores = self.engine.score_crop(crop, method=method) - from identity import decide_identity - - return decide_identity(scores, method=method) - - adapter = YoloMatcherAdapter(_DetectorFacade(engine), _MatcherFacade(engine), method="hybrid") - return ScanService(PerceptionService(adapter)) # --------------------------------------------------------------------------- @@ -295,7 +262,6 @@ def create_app( settings: Optional[ServerSettings] = None, *, web_dir: Optional[str] = None, - scans: Any = None, ) -> FastAPI: settings = settings or server_settings_from_env() @asynccontextmanager @@ -323,7 +289,6 @@ def create_app( allow_headers=["Authorization", "Content-Type"], ) - scans = scans if scans is not None else build_legacy_scans(engine) # A live session can be recorded so a real scan becomes data to label later. recorder = FrameRecorder( settings.record_frames, @@ -994,127 +959,6 @@ def create_app( async with sessions_lock: sessions.pop(session_id, None) - # ---------------------------------------------------------------- legacy session routes - @app.post("/api/sessions", dependencies=[Depends(authorize)]) - async def legacy_start_session(payload: Dict[str, Any] = None) -> Dict[str, Any]: - payload = payload or {} - session = scans.start_session( - zone_id=str(payload.get("zone_id") or "zone-1"), - catalog_version=str(payload.get("catalog_version") or "stock-v1"), - capture_kind=str(payload.get("capture_kind") or "sampled"), - ) - return {"status": "success", "session": session.snapshot()} - - @app.post("/api/sessions/{session_id}/frames", dependencies=[Depends(authorize)]) - async def legacy_observe_frame(session_id: str, payload: Dict[str, Any] = None) -> Dict[str, Any]: - from observe import FrameRequest, ObserveError - - payload = payload or {} - try: - session = scans.get_session(session_id) - raw = payload.get("frame_bytes_b64") or payload.get("image_base64") or "" - if not isinstance(raw, str): - raw = "" - if "," in raw: - raw = raw.split(",", 1)[1] - # Base64 expands ~4/3. Reject the text before decoding it. - if len(raw) > (MAX_FRAME_BYTES * 4) // 3 + 8: - raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413) - loop = asyncio.get_running_loop() - try: - frame_bytes = ( - await loop.run_in_executor(None, lambda: base64.b64decode(raw, validate=True)) - if raw else b"" - ) - except (ValueError, base64.binascii.Error): - raise _error("BAD_REQUEST", "frame_bytes_b64 is not valid base64", 400) - if len(frame_bytes) > MAX_FRAME_BYTES: - raise _error("PAYLOAD_TOO_LARGE", f"frame larger than {MAX_FRAME_BYTES} bytes", 413) - request = FrameRequest( - frame_id=str(payload.get("frame_id") or ""), - frame_bytes=frame_bytes, - sequence=int(payload.get("sequence", 0)), - captured_at=float(payload.get("captured_at") or 0), - session_id=session.session_id, - zone_id=session.zone_id, - catalog_version=session.catalog_version, - capture_kind=str(payload.get("capture_kind") or session.capture_kind), - width=payload.get("width"), - height=payload.get("height"), - ) - if not request.frame_id: - raise _error("BAD_REQUEST", "frame_id required", 400) - observation = await loop.run_in_executor(None, scans.observe_frame, request) - return {"status": "success", "observation": observation.as_dict(), "session": session.snapshot()} - except ObserveError as exc: - raise _error("CONFLICT", str(exc), 409) - - @app.get("/api/sessions/{session_id}", dependencies=[Depends(authorize)]) - async def legacy_get_session(session_id: str) -> Dict[str, Any]: - from observe import ObserveError - - try: - return {"status": "success", "session": scans.get_session(session_id).snapshot()} - except ObserveError as exc: - raise _error("NOT_FOUND", str(exc), 404) - - @app.post("/api/sessions/{session_id}/review", dependencies=[Depends(authorize)]) - async def legacy_review(session_id: str, payload: Dict[str, Any] = None) -> Dict[str, Any]: - from observe import ObserveError - - payload = payload or {} - - try: - snapshot = scans.review( - session_id, - actor=str(payload.get("actor") or "operator"), - reason=str(payload.get("reason") or "manual-review"), - note=payload.get("note"), - ) - return {"status": "success", "session": snapshot} - except ObserveError as exc: - raise _error("CONFLICT", str(exc), 409) - - @app.post("/api/sessions/{session_id}/finalize", dependencies=[Depends(authorize)]) - async def legacy_finalize(session_id: str) -> Dict[str, Any]: - from observe import ObserveError - - try: - return {"status": "success", "session": scans.finalize(session_id)} - except ObserveError as exc: - raise _error("CONFLICT", str(exc), 409) - - @app.post("/api/sessions/{session_id}/cancel", dependencies=[Depends(authorize)]) - async def legacy_cancel(session_id: str) -> Dict[str, Any]: - from observe import ObserveError - - try: - return {"status": "success", "session": scans.cancel(session_id)} - except ObserveError as exc: - raise _error("CONFLICT", str(exc), 409) - - @app.post("/api/sessions/{session_id}/rescan", dependencies=[Depends(authorize)]) - async def legacy_rescan(session_id: str) -> Dict[str, Any]: - from observe import ObserveError - - try: - return {"status": "success", "session": scans.rescan(session_id).snapshot()} - except ObserveError as exc: - raise _error("CONFLICT", str(exc), 409) - - @app.post("/api/compare_live_vs_still", dependencies=[Depends(authorize)]) - async def legacy_compare(payload: Dict[str, Any] = None) -> Dict[str, Any]: - from observe import ObserveError - - payload = payload or {} - try: - comparison = scans.compare_live_vs_still( - str(payload.get("sampled_session_id")), str(payload.get("still_session_id")) - ) - return {"status": "success", "comparison": comparison} - except ObserveError as exc: - raise _error("CONFLICT", str(exc), 409) - # ---------------------------------------------------------------- errors @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse: @@ -1148,25 +992,12 @@ def create_app( if os.path.isdir(resolved_web_dir): app.mount("/", StaticFiles(directory=resolved_web_dir, html=True), name="web") - - return app -def create_fastapi_app( - engine: RecognitionEngine, - scans: Any = None, - settings: Optional[ServerSettings] = None, -) -> FastAPI: - """Back-compat entry point; the server entry point is create_app().""" - return create_app(engine, settings, scans=scans) - - # --------------------------------------------------------------------------- # entry point # --------------------------------------------------------------------------- - - def main() -> None: parser = argparse.ArgumentParser(description="Dorayaki - shelf recognition server") parser.add_argument("--host", type=str, default=None) diff --git a/recognization/code/observe.py b/recognization/code/observe.py deleted file mode 100644 index 255fbe4..0000000 --- a/recognization/code/observe.py +++ /dev/null @@ -1,474 +0,0 @@ -"""Session-local perception and one-zone assisted scan. - -Production path: observe(frame, catalog_version) returns an immutable -frame-local observation. Scan sessions own draft → finalized/cancelled -state. Detector/matcher stay replaceable adapters; tests inject fakes. - -ponytail: in-memory sessions only. Durable store when a successful -experiment needs one. -""" - -from __future__ import annotations - -import hashlib -import threading -import time -import uuid -from dataclasses import dataclass, field -from typing import Any, Callable, Dict, List, Optional, Tuple - -from identity import ( - IDENTITY_ACCEPTED, - IDENTITY_AMBIGUOUS, - IDENTITY_REJECTED, - IDENTITY_UNKNOWN, - decide_identity, -) - -COORDINATE_CONVENTION = "pixel-xyxy-origin-top-left" - -SESSION_DRAFT = "draft" -SESSION_FINALIZED = "finalized" -SESSION_CANCELLED = "cancelled" - -CAPTURE_STILL = "still" -CAPTURE_SAMPLED = "sampled" - -DEFAULT_MAX_IN_FLIGHT = 1 -DEFAULT_MAX_QUEUED = 2 -DEFAULT_SESSION_TTL_S = 30 * 60 - - -class ObserveError(ValueError): - """Caller or session contract violation.""" - - -def content_fingerprint(frame_bytes: bytes) -> str: - return hashlib.sha256(frame_bytes).hexdigest() - - -def _now() -> float: - return time.time() - - -@dataclass -class FrameRequest: - frame_id: str - frame_bytes: bytes - sequence: int - captured_at: float - session_id: str - zone_id: str - catalog_version: str - capture_kind: str = CAPTURE_SAMPLED - width: Optional[int] = None - height: Optional[int] = None - orientation: str = "upright" - - -@dataclass -class Observation: - frame_id: str - content_fingerprint: str - sequence: int - captured_at: float - observed_at: float - session_id: str - zone_id: str - catalog_version: str - capture_kind: str - width: int - height: int - orientation: str - coordinate_convention: str - boxes: List[Dict[str, Any]] - identities: List[Dict[str, Any]] - empty_slots: List[Dict[str, Any]] - model_version: str - matcher_method: str - feature_extractor: str - quality: Dict[str, Any] - selected: bool = False - - def as_dict(self) -> Dict[str, Any]: - return { - "frame_id": self.frame_id, - "content_fingerprint": self.content_fingerprint, - "sequence": self.sequence, - "captured_at": self.captured_at, - "observed_at": self.observed_at, - "session_id": self.session_id, - "zone_id": self.zone_id, - "catalog_version": self.catalog_version, - "capture_kind": self.capture_kind, - "width": self.width, - "height": self.height, - "orientation": self.orientation, - "coordinate_convention": self.coordinate_convention, - "boxes": list(self.boxes), - "identities": list(self.identities), - "empty_slots": list(self.empty_slots), - "model_version": self.model_version, - "matcher_method": self.matcher_method, - "feature_extractor": self.feature_extractor, - "quality": dict(self.quality), - "selected": self.selected, - } - - -class PerceptionAdapter: - """Replaceable detector + matcher. Default is a testable fake.""" - - def __init__( - self, - detect_fn: Optional[Callable[[bytes], Dict[str, Any]]] = None, - classify_fn: Optional[Callable[[Dict[str, Any]], Dict[str, float]]] = None, - model_version: str = "fake-detector", - matcher_method: str = "hybrid", - feature_extractor: str = "recorded", - ): - self._detect_fn = detect_fn - self._classify_fn = classify_fn - self.model_version = model_version - self.matcher_method = matcher_method - self.feature_extractor = feature_extractor - - def detect(self, frame_bytes: bytes) -> Dict[str, Any]: - if self._detect_fn is not None: - return self._detect_fn(frame_bytes) - return { - "width": 1, - "height": 1, - "boxes": [], - "empty_slots": [], - "quality": {"coverage": 0.0, "limitation": "no-detector"}, - } - - def classify_box(self, box: Dict[str, Any]) -> Dict[str, float]: - if self._classify_fn is not None: - return self._classify_fn(box) - return {} - - -class PerceptionService: - """Frame-local observe(). Shared adapter cache is allowed; results are not.""" - - def __init__(self, adapter: Optional[PerceptionAdapter] = None): - self.adapter = adapter or PerceptionAdapter() - - def observe(self, request: FrameRequest, catalog_version: str) -> Observation: - if request.catalog_version != catalog_version: - raise ObserveError("catalog_version mismatch") - fingerprint = content_fingerprint(request.frame_bytes) - detected = self.adapter.detect(request.frame_bytes) - width = int(request.width or detected.get("width") or 0) - height = int(request.height or detected.get("height") or 0) - boxes = list(detected.get("boxes") or []) - empty_slots = list(detected.get("empty_slots") or []) - identities = [] - for box in boxes: - scores = self.adapter.classify_box(box) - ranked = decide_identity(scores, method=self.adapter.matcher_method) - identities.append( - { - "box_id": box.get("id"), - "identity_level": ranked["identity_level"], - "best_sku": ranked["best_sku"], - "score": ranked["score"], - "confidence": ranked["confidence"], - "ranked_candidates": ranked["ranked_candidates"], - "message": ranked["message"], - } - ) - quality = dict(detected.get("quality") or {}) - if not boxes: - quality.setdefault("limitation", "no-visible-objects") - return Observation( - frame_id=request.frame_id, - content_fingerprint=fingerprint, - sequence=request.sequence, - captured_at=request.captured_at, - observed_at=_now(), - session_id=request.session_id, - zone_id=request.zone_id, - catalog_version=catalog_version, - capture_kind=request.capture_kind, - width=width, - height=height, - orientation=request.orientation, - coordinate_convention=COORDINATE_CONVENTION, - boxes=boxes, - identities=identities, - empty_slots=empty_slots, - model_version=self.adapter.model_version, - matcher_method=self.adapter.matcher_method, - feature_extractor=self.adapter.feature_extractor, - quality=quality, - ) - - -@dataclass -class ScanSession: - session_id: str - zone_id: str - catalog_version: str - capture_kind: str - created_at: float - expires_at: float - status: str = SESSION_DRAFT - selected_frame_id: Optional[str] = None - observations: Dict[str, Observation] = field(default_factory=dict) - in_flight: int = 0 - queued: int = 0 - dropped_stale: int = 0 - review: Optional[Dict[str, Any]] = None - finalized_at: Optional[float] = None - cancelled_at: Optional[float] = None - revision: int = 0 - lock: threading.Lock = field(default_factory=threading.Lock, repr=False, compare=False) - - def snapshot(self) -> Dict[str, Any]: - selected = self.observations.get(self.selected_frame_id) if self.selected_frame_id else None - return { - "session_id": self.session_id, - "zone_id": self.zone_id, - "catalog_version": self.catalog_version, - "capture_kind": self.capture_kind, - "status": self.status, - "created_at": self.created_at, - "expires_at": self.expires_at, - "selected_frame_id": self.selected_frame_id, - "selected_observation": selected.as_dict() if selected else None, - "observation_count": len(self.observations), - "in_flight": self.in_flight, - "queued": self.queued, - "dropped_stale": self.dropped_stale, - "review": dict(self.review) if self.review else None, - "finalized_at": self.finalized_at, - "cancelled_at": self.cancelled_at, - "revision": self.revision, - } - - -def _usable_score(observation: Observation) -> Tuple[float, float, int]: - """Prefer more accepted identities, then fewer unknowns, then newer sequence.""" - accepted = sum(1 for i in observation.identities if i["identity_level"] == IDENTITY_ACCEPTED) - unknown = sum( - 1 - for i in observation.identities - if i["identity_level"] in (IDENTITY_UNKNOWN, IDENTITY_AMBIGUOUS, IDENTITY_REJECTED) - ) - return (float(accepted), -float(unknown), int(observation.sequence)) - - -class ScanService: - """One-zone assisted scan: sampled frames, review, finalize, rescan.""" - - def __init__( - self, - perception: Optional[PerceptionService] = None, - *, - max_in_flight: int = DEFAULT_MAX_IN_FLIGHT, - max_queued: int = DEFAULT_MAX_QUEUED, - ttl_s: float = DEFAULT_SESSION_TTL_S, - clock: Callable[[], float] = _now, - ): - self.perception = perception or PerceptionService() - self.max_in_flight = max_in_flight - self.max_queued = max_queued - self.ttl_s = ttl_s - self._clock = clock - self._sessions: Dict[str, ScanSession] = {} - self._global_lock = threading.Lock() - - def start_session( - self, - zone_id: str, - catalog_version: str, - *, - capture_kind: str = CAPTURE_SAMPLED, - session_id: Optional[str] = None, - ) -> ScanSession: - now = self._clock() - session = ScanSession( - session_id=session_id or str(uuid.uuid4()), - zone_id=zone_id, - catalog_version=catalog_version, - capture_kind=capture_kind, - created_at=now, - expires_at=now + self.ttl_s, - ) - with self._global_lock: - self._sessions[session.session_id] = session - return session - - def get_session(self, session_id: str) -> ScanSession: - with self._global_lock: - session = self._sessions.get(session_id) - if session is None: - raise ObserveError(f"unknown session {session_id}") - if self._clock() >= session.expires_at and session.status == SESSION_DRAFT: - raise ObserveError("session expired") - return session - - def observe_frame(self, request: FrameRequest) -> Observation: - session = self.get_session(request.session_id) - if request.zone_id != session.zone_id: - raise ObserveError("zone mismatch") - if request.catalog_version != session.catalog_version: - raise ObserveError("catalog_version mismatch") - if session.status != SESSION_DRAFT: - raise ObserveError(f"session is {session.status}") - - fingerprint = content_fingerprint(request.frame_bytes) - with session.lock: - existing = session.observations.get(request.frame_id) - if existing is not None: - if existing.content_fingerprint != fingerprint: - raise ObserveError("frame_id reused with different content") - return existing - if session.in_flight >= self.max_in_flight: - if session.queued >= self.max_queued: - session.dropped_stale += 1 - raise ObserveError("backpressure") - session.queued += 1 - queued = True - else: - session.in_flight += 1 - queued = False - - try: - if queued: - with session.lock: - session.queued -= 1 - if session.in_flight >= self.max_in_flight: - session.dropped_stale += 1 - raise ObserveError("backpressure") - session.in_flight += 1 - observation = self.perception.observe(request, session.catalog_version) - finally: - with session.lock: - if session.in_flight > 0: - session.in_flight -= 1 - - with session.lock: - if session.status != SESSION_DRAFT: - raise ObserveError(f"session is {session.status}") - replay = session.observations.get(request.frame_id) - if replay is not None: - if replay.content_fingerprint != fingerprint: - raise ObserveError("frame_id reused with different content") - return replay - selected = session.observations.get(session.selected_frame_id) if session.selected_frame_id else None - if selected is not None and request.sequence < selected.sequence: - session.observations[request.frame_id] = observation - session.dropped_stale += 1 - return observation - session.observations[request.frame_id] = observation - if selected is None or _usable_score(observation) >= _usable_score(selected): - if selected is not None: - selected.selected = False - observation.selected = True - session.selected_frame_id = observation.frame_id - return observation - - def review( - self, - session_id: str, - *, - actor: str, - reason: str, - note: Optional[str] = None, - previous_value: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: - session = self.get_session(session_id) - with session.lock: - if session.status != SESSION_DRAFT: - raise ObserveError(f"session is {session.status}") - session.review = { - "actor": actor, - "reason": reason, - "note": note, - "previous_value": previous_value, - "reviewed_at": self._clock(), - } - session.revision += 1 - return session.snapshot() - - def finalize(self, session_id: str) -> Dict[str, Any]: - session = self.get_session(session_id) - with session.lock: - if session.status != SESSION_DRAFT: - raise ObserveError(f"session is {session.status}") - session.status = SESSION_FINALIZED - session.finalized_at = self._clock() - return session.snapshot() - - def cancel(self, session_id: str) -> Dict[str, Any]: - session = self.get_session(session_id) - with session.lock: - if session.status == SESSION_FINALIZED: - raise ObserveError("finalized sessions are immutable") - session.status = SESSION_CANCELLED - session.cancelled_at = self._clock() - return session.snapshot() - - def rescan(self, session_id: str) -> ScanSession: - parent = self.get_session(session_id) - if parent.status == SESSION_DRAFT: - raise ObserveError("cancel or finalize before rescan") - child = self.start_session( - parent.zone_id, - parent.catalog_version, - capture_kind=parent.capture_kind, - ) - child.revision = parent.revision + 1 - child.review = { - "actor": "system", - "reason": "rescan", - "previous_session_id": parent.session_id, - "reviewed_at": self._clock(), - } - return child - - def compare_live_vs_still( - self, sampled_session_id: str, still_session_id: str - ) -> Dict[str, Any]: - sampled = self.get_session(sampled_session_id) - still = self.get_session(still_session_id) - sampled_obs = sampled.observations.get(sampled.selected_frame_id) if sampled.selected_frame_id else None - still_obs = still.observations.get(still.selected_frame_id) if still.selected_frame_id else None - - def counts(obs: Optional[Observation]) -> Dict[str, int]: - if obs is None: - return {"accepted": 0, "unknown": 0, "ambiguous": 0, "rejected": 0, "boxes": 0} - levels = [i["identity_level"] for i in obs.identities] - return { - "accepted": levels.count(IDENTITY_ACCEPTED), - "unknown": levels.count(IDENTITY_UNKNOWN), - "ambiguous": levels.count(IDENTITY_AMBIGUOUS), - "rejected": levels.count(IDENTITY_REJECTED), - "boxes": len(obs.boxes), - } - - return { - "sampled": { - "session_id": sampled.session_id, - "capture_kind": sampled.capture_kind, - "observation_count": len(sampled.observations), - "dropped_stale": sampled.dropped_stale, - "selected": counts(sampled_obs), - }, - "still": { - "session_id": still.session_id, - "capture_kind": still.capture_kind, - "observation_count": len(still.observations), - "dropped_stale": still.dropped_stale, - "selected": counts(still_obs), - }, - "duplicate_frames_prevented": any( - len({o.content_fingerprint for o in s.observations.values()}) < len(s.observations) - for s in (sampled, still) - ), - "note": "Comparator of session mechanics, not measured shelf accuracy or latency.", - } diff --git a/recognization/code/pipeline.py b/recognization/code/pipeline.py deleted file mode 100644 index 403efe6..0000000 --- a/recognization/code/pipeline.py +++ /dev/null @@ -1,179 +0,0 @@ -""" -End-to-End Supermarket Shelf Recognition Pipeline -Integrates: -1. YOLO object detector for shelf product localization -2. SURF / ORB / BRISK / Hybrid local feature matcher for classification -3. Stock CSV inventory lookup for real-time pricing and stock quantities -""" - -import os -import sys -import cv2 -import numpy as np -from typing import Dict, Optional, Any - -current_dir = os.path.dirname(os.path.abspath(__file__)) -if current_dir not in sys.path: - sys.path.insert(0, current_dir) - -from detector import ShelfProductDetector -from feature_matcher import ProductFeatureMatcher -from inventory import InventoryManager - - -class ShelfRecognitionPipeline: - """Master pipeline managing interactive supermarket shelf recognition.""" - - def __init__( - self, - weights_path: str = "runs/detect/train/weights/best.pt", - stock_dir: str = "stock", - stock_csv: str = "stock.csv" - ): - print("[Pipeline] Initializing Inventory Manager...") - self.inventory = InventoryManager(stock_dir=stock_dir, csv_filename=stock_csv) - - print("[Pipeline] Initializing Feature Matcher...") - self.matcher = ProductFeatureMatcher(stock_dir=stock_dir) - - print("[Pipeline] Initializing Shelf Detector...") - self.detector = ShelfProductDetector(weights_path=weights_path) - - self.current_image_path: Optional[str] = None - self.current_image: Optional[np.ndarray] = None - self.current_detections: list = [] - self.current_empty_slots: list = [] - - def set_active_shelf_image(self, image_input: Any) -> Dict[str, Any]: - """Loads and runs detection on a new shelf image.""" - if isinstance(image_input, str): - self.current_image_path = image_input - self.current_image = cv2.imread(image_input) - else: - self.current_image_path = "memory_buffer" - self.current_image = image_input - - if self.current_image is None: - return {"status": "error", "message": "Invalid or unreadable image"} - - det_result = self.detector.detect(self.current_image) - self.current_detections = det_result.get("boxes", []) - - # Detect empty shelf spaces (out-of-stock gaps) - self.current_empty_slots = self.detector.detect_empty_slots( - self.current_image.shape, self.current_detections - ) - - return { - "status": "success", - "image_shape": self.current_image.shape, - "total_detected": len(self.current_detections), - "boxes": self.current_detections, - "empty_slots": self.current_empty_slots - } - - def process_click( - self, - x: int, - y: int, - method: str = "hybrid", - tolerance_px: int = 40 - ) -> Dict[str, Any]: - """ - Handles user click at coordinate (x, y) on the shelf image: - 1. Identifies the bounding box - 2. Crops product ROI - 3. Classifies via SURF/ORB/BRISK/Hybrid matching - 4. Queries inventory for price, quantity, and metadata - """ - if self.current_image is None: - return {"status": "error", "message": "No active shelf image loaded."} - - target_box = self.detector.find_box_at_point( - x, y, self.current_detections, tolerance_px=tolerance_px - ) - - if target_box is None: - # Check if user clicked an empty shelf space - for slot in getattr(self, "current_empty_slots", []): - sx1, sy1, sx2, sy2 = slot["box"] - if (sx1 - 10 <= x <= sx2 + 10) and (sy1 - 10 <= y <= sy2 + 10): - return { - "status": "empty_slot", - "clicked_point": (x, y), - "empty_slot": slot, - "message": f"Chỗ trống trên kệ (Khu vực hàng hết / Row {slot['row_index']}). Ước tính thiếu ~{slot['missing_items_est']} sản phẩm." - } - - return { - "status": "not_found", - "clicked_point": (x, y), - "message": f"No product detected near coordinate ({x}, {y}). Please click inside a product packaging." - } - - # Crop ROI - crop = self.detector.crop_roi(self.current_image, target_box["box"]) - - # Classify with selected method - cls_result = self.matcher.classify_crop(crop, method=method) - identity_level = cls_result.get("identity_level", "rejected") - best_sku = cls_result.get("best_sku") - product_info = self.inventory.get_product(best_sku) if best_sku else None - - vis_match = None - if best_sku: - vis_match = self.matcher.generate_match_visualization(crop, best_sku, method=method) - - annotated_shelf = self.detector.draw_detections( - self.current_image, self.current_detections, selected_box_id=target_box["id"] - ) - - click_status = "success" if identity_level == "accepted" else identity_level - return { - "status": click_status, - "clicked_point": (x, y), - "target_box": target_box, - "classification": { - "method": method, - "matched_sku": best_sku, - "identity_level": identity_level, - "confidence": cls_result["confidence"], - "score": cls_result["score"], - "top_candidates": cls_result["ranked_candidates"], - "message": cls_result.get("message"), - }, - "inventory": product_info, - "crop_image": crop, - "annotated_shelf": annotated_shelf, - "match_visualization": vis_match - } - - -if __name__ == "__main__": - import argparse - parser = argparse.ArgumentParser(description="Test Shelf Recognition Pipeline") - parser.add_argument("--image", type=str, default="runs/detect/train/train_batch0.jpg") - parser.add_argument("--method", type=str, default="hybrid", choices=["surf", "orb", "brisk", "hybrid"]) - args = parser.parse_args() - - pipeline = ShelfRecognitionPipeline() - print(f"Loading image: {args.image}") - init_res = pipeline.set_active_shelf_image(args.image) - print(f"Total detected items: {init_res['total_detected']}") - - if init_res['boxes']: - # Simulate click on the center of the first detected product - first_box = init_res['boxes'][0] - cx, cy = first_box['center'] - print(f"Simulating click on Product #{first_box['id']} at ({cx}, {cy})...") - res = pipeline.process_click(cx, cy, method=args.method) - print("\n--- Recognition & Inventory Result ---") - print(f"Status: {res['status']}") - cls = res.get("classification") or {} - print(f"Identity: {cls.get('identity_level')} SKU={cls.get('matched_sku')} score={cls.get('score')}") - prod = res.get("inventory") - if prod: - print(f"Product Name: {prod['product_name']}") - print(f"Price: {prod['price']} JPY") - print(f"In Stock: {prod['number_on']} units (Received: {prod['number_in']}, Sold: {prod['number_out']})") - print(f"Expiration Date: {prod['expiration_date']}") diff --git a/recognization/code/test_api.py b/recognization/code/test_api.py index 17cbd0e..fc70c08 100644 --- a/recognization/code/test_api.py +++ b/recognization/code/test_api.py @@ -30,33 +30,6 @@ from test_live import FakeDetector, FakeInventory, FakeMatcher # noqa: E402 from fastapi.testclient import TestClient # noqa: E402 -def make_scans(engine): - """Legacy session model wired to the fake engine (the real factory is exercised - against the real model in test_real_model).""" - from observe import PerceptionAdapter, PerceptionService, ScanService - - def detect(frame_bytes): - image = engine.decode_image(frame_bytes) - boxes = engine.detect(image) - return {"width": image.shape[1], "height": image.shape[0], "boxes": boxes, - "empty_slots": [], "quality": {"decoded": True}} - - def classify(box): - scores = engine.score_crop(engine.crop(engine._last_image, box["box"])) - return scores - - # keep the frame for the classify callback - def detect_with_frame(frame_bytes): - image = engine.decode_image(frame_bytes) - engine._last_image = image - boxes = engine.detect(image) - return {"width": image.shape[1], "height": image.shape[0], "boxes": boxes, - "empty_slots": [], "quality": {"decoded": True}} - - adapter = PerceptionAdapter(detect_fn=detect_with_frame, classify_fn=classify, - model_version="fake", matcher_method="hybrid") - return ScanService(PerceptionService(adapter)) - def make_app(*, token=None, max_sessions=4, detector=None): engine = RecognitionEngine( @@ -64,7 +37,7 @@ def make_app(*, token=None, max_sessions=4, detector=None): settings=EngineSettings(classify_workers=2), ) settings = ServerSettings(token=token, max_sessions=max_sessions) - app = create_app(engine, settings, web_dir="/nonexistent", scans=make_scans(engine)) + app = create_app(engine, settings, web_dir="/nonexistent") return TestClient(app), engine @@ -549,27 +522,5 @@ class WebSocketProtocolTests(unittest.TestCase): engine.shutdown() -class LegacyRouteTests(unittest.TestCase): - def test_legacy_sessions_still_respond(self): - """The earlier session model stays mounted; it is backed by the engine now.""" - client, engine = make_app() - try: - started = client.post("/api/sessions", json={"zone_id": "z1", "catalog_version": "v1"}) - self.assertEqual(started.status_code, 200) - session_id = started.json()["session"]["session_id"] - frames = client.post( - f"/api/sessions/{session_id}/frames", - json={ - "frame_id": "f1", - "sequence": 1, - "frame_bytes_b64": base64.b64encode(jpeg_bytes()).decode(), - }, - ) - self.assertEqual(frames.status_code, 200) - self.assertIn("observation", frames.json()) - finally: - engine.shutdown() - - if __name__ == "__main__": unittest.main() diff --git a/recognization/code/test_identity.py b/recognization/code/test_identity.py new file mode 100644 index 0000000..ef2904d --- /dev/null +++ b/recognization/code/test_identity.py @@ -0,0 +1,52 @@ +"""Ranking contract for decide_identity: what a score set is allowed to name. + +Moved out of test_observe.py when the legacy session model was deleted; the +ranking rules are the live path's, not the legacy one's. +""" + +from __future__ import annotations + +import os +import sys +import unittest + +current_dir = os.path.dirname(os.path.abspath(__file__)) +if current_dir not in sys.path: + sys.path.insert(0, current_dir) + +from identity import ( # noqa: E402 + IDENTITY_ACCEPTED, + IDENTITY_AMBIGUOUS, + IDENTITY_REJECTED, + IDENTITY_UNKNOWN, + decide_identity, +) + + +class IdentityRankingTests(unittest.TestCase): + def test_zero_scores_are_rejected_not_a_sku(self): + ranked = decide_identity({"sku_a": 0.0, "sku_b": 0.0}) + self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED) + self.assertIsNone(ranked["best_sku"]) + self.assertEqual(ranked["status"], IDENTITY_REJECTED) + + def test_empty_scores_are_rejected(self): + ranked = decide_identity({}) + self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED) + self.assertIsNone(ranked["best_sku"]) + + def test_weak_unique_score_is_unknown(self): + ranked = decide_identity({"sku_a": 0.4, "sku_b": 0.0}) + self.assertEqual(ranked["identity_level"], IDENTITY_UNKNOWN) + self.assertIsNone(ranked["best_sku"]) + + def test_close_scores_are_ambiguous(self): + ranked = decide_identity({"sku_a": 20.0, "sku_b": 19.5}) + self.assertEqual(ranked["identity_level"], IDENTITY_AMBIGUOUS) + self.assertIsNone(ranked["best_sku"]) + + def test_unique_strong_score_is_accepted(self): + ranked = decide_identity({"sku_a": 40.0, "sku_b": 5.0}) + self.assertEqual(ranked["identity_level"], IDENTITY_ACCEPTED) + self.assertEqual(ranked["best_sku"], "sku_a") + self.assertEqual(ranked["status"], "success") diff --git a/recognization/code/test_observe.py b/recognization/code/test_observe.py deleted file mode 100644 index 6dd46a4..0000000 --- a/recognization/code/test_observe.py +++ /dev/null @@ -1,319 +0,0 @@ -"""Contract tests for identity ranking and session-local observe(). - -No OpenCV, YOLO, or network. These tests defend replay, isolation, -stale frames, backpressure, unknown/reject, and live-vs-still mechanics. -They do not measure shelf accuracy or latency. -""" - -from __future__ import annotations - -import os -import sys -import threading -import unittest - -current_dir = os.path.dirname(os.path.abspath(__file__)) -if current_dir not in sys.path: - sys.path.insert(0, current_dir) - -from identity import ( - IDENTITY_ACCEPTED, - IDENTITY_AMBIGUOUS, - IDENTITY_REJECTED, - IDENTITY_UNKNOWN, - decide_identity, -) -from observe import ( - CAPTURE_SAMPLED, - CAPTURE_STILL, - FrameRequest, - ObserveError, - PerceptionAdapter, - PerceptionService, - ScanService, -) - - -def _req(session, frame_id, payload, sequence, captured_at=1.0, kind=CAPTURE_SAMPLED): - return FrameRequest( - frame_id=frame_id, - frame_bytes=payload, - sequence=sequence, - captured_at=captured_at, - session_id=session.session_id, - zone_id=session.zone_id, - catalog_version=session.catalog_version, - capture_kind=kind, - ) - - -class IdentityRankingTests(unittest.TestCase): - def test_zero_scores_are_rejected_not_a_sku(self): - ranked = decide_identity({"sku_a": 0.0, "sku_b": 0.0}) - self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED) - self.assertIsNone(ranked["best_sku"]) - self.assertEqual(ranked["status"], IDENTITY_REJECTED) - - def test_empty_scores_are_rejected(self): - ranked = decide_identity({}) - self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED) - self.assertIsNone(ranked["best_sku"]) - - def test_weak_unique_score_is_unknown(self): - ranked = decide_identity({"sku_a": 0.4, "sku_b": 0.0}) - self.assertEqual(ranked["identity_level"], IDENTITY_UNKNOWN) - self.assertIsNone(ranked["best_sku"]) - - def test_close_scores_are_ambiguous(self): - ranked = decide_identity({"sku_a": 20.0, "sku_b": 19.5}) - self.assertEqual(ranked["identity_level"], IDENTITY_AMBIGUOUS) - self.assertIsNone(ranked["best_sku"]) - - def test_unique_strong_score_is_accepted(self): - ranked = decide_identity({"sku_a": 40.0, "sku_b": 5.0}) - self.assertEqual(ranked["identity_level"], IDENTITY_ACCEPTED) - self.assertEqual(ranked["best_sku"], "sku_a") - self.assertEqual(ranked["status"], "success") - - -class SessionObserveTests(unittest.TestCase): - def setUp(self): - def detect(frame_bytes: bytes): - sku = frame_bytes.decode("ascii", errors="ignore") - box = {"id": 0, "box": [0, 0, 10, 10], "sku_hint": sku} - return { - "width": 10, - "height": 10, - "boxes": [box], - "empty_slots": [], - "quality": {"coverage": 1.0}, - } - - def classify(box): - hint = box.get("sku_hint", "") - if hint.startswith("unknown"): - return {"sku_a": 0.0, "sku_b": 0.0} - if hint.startswith("ambig"): - return {"sku_a": 20.0, "sku_b": 19.8} - if hint.startswith("weak"): - return {"sku_a": 0.4, "sku_b": 0.0} - if hint.startswith("sku_b"): - return {"sku_a": 2.0, "sku_b": 40.0} - return {"sku_a": 40.0, "sku_b": 2.0} - - adapter = PerceptionAdapter( - detect_fn=detect, - classify_fn=classify, - model_version="fake", - matcher_method="hybrid", - feature_extractor="fake", - ) - self.scans = ScanService(PerceptionService(adapter), max_in_flight=1, max_queued=1) - - def test_replay_same_frame_id_and_content_is_idempotent(self): - session = self.scans.start_session("zone-1", "stock-v1") - first = self.scans.observe_frame(_req(session, "f1", b"sku_a", 1)) - second = self.scans.observe_frame(_req(session, "f1", b"sku_a", 1)) - self.assertIs(first, second) - self.assertEqual(len(session.observations), 1) - self.assertEqual(first.identities[0]["best_sku"], "sku_a") - - def test_same_frame_id_different_content_is_rejected(self): - session = self.scans.start_session("zone-1", "stock-v1") - self.scans.observe_frame(_req(session, "f1", b"sku_a", 1)) - with self.assertRaises(ObserveError): - self.scans.observe_frame(_req(session, "f1", b"sku_b", 1)) - - def test_sessions_do_not_share_observations(self): - a = self.scans.start_session("zone-1", "stock-v1") - b = self.scans.start_session("zone-1", "stock-v1") - self.scans.observe_frame(_req(a, "f1", b"sku_a", 1)) - self.scans.observe_frame(_req(b, "f1", b"sku_b", 1)) - self.assertEqual(a.observations["f1"].identities[0]["best_sku"], "sku_a") - self.assertEqual(b.observations["f1"].identities[0]["best_sku"], "sku_b") - self.assertEqual(a.selected_frame_id, "f1") - self.assertEqual(b.selected_frame_id, "f1") - self.assertIsNot(a.observations["f1"], b.observations["f1"]) - - def test_stale_sequence_cannot_overwrite_selected_state(self): - session = self.scans.start_session("zone-1", "stock-v1") - newer = self.scans.observe_frame(_req(session, "f2", b"sku_a", 2)) - older = self.scans.observe_frame(_req(session, "f1", b"sku_b", 1)) - self.assertEqual(session.selected_frame_id, "f2") - self.assertTrue(newer.selected) - self.assertFalse(older.selected) - self.assertGreaterEqual(session.dropped_stale, 1) - self.assertIn("f1", session.observations) - - def test_better_later_frame_replaces_selected_best(self): - session = self.scans.start_session("zone-1", "stock-v1") - weak = self.scans.observe_frame(_req(session, "f1", b"weak", 1)) - strong = self.scans.observe_frame(_req(session, "f2", b"sku_a", 2)) - self.assertFalse(weak.selected) - self.assertTrue(strong.selected) - self.assertEqual(session.selected_frame_id, "f2") - - def test_unknown_zero_score_is_not_selected_as_sku(self): - session = self.scans.start_session("zone-1", "stock-v1") - obs = self.scans.observe_frame(_req(session, "f1", b"unknown", 1)) - self.assertEqual(obs.identities[0]["identity_level"], IDENTITY_REJECTED) - self.assertIsNone(obs.identities[0]["best_sku"]) - - def test_backpressure_rejects_when_queue_full(self): - started = threading.Event() - release = threading.Event() - - def slow_detect(frame_bytes: bytes): - started.set() - release.wait(timeout=2) - return { - "width": 1, - "height": 1, - "boxes": [], - "empty_slots": [], - "quality": {}, - } - - scans = ScanService( - PerceptionService(PerceptionAdapter(detect_fn=slow_detect)), - max_in_flight=1, - max_queued=0, - ) - session = scans.start_session("zone-1", "stock-v1") - errors = [] - - def run(): - try: - scans.observe_frame(_req(session, "slow", b"sku_a", 1)) - except ObserveError as exc: - errors.append(str(exc)) - - worker = threading.Thread(target=run) - worker.start() - self.assertTrue(started.wait(timeout=2)) - with self.assertRaises(ObserveError) as ctx: - scans.observe_frame(_req(session, "other", b"sku_b", 2)) - self.assertIn("backpressure", str(ctx.exception)) - release.set() - worker.join(timeout=2) - self.assertTrue(session.dropped_stale >= 1) - - def test_finalize_is_immutable_and_rescan_creates_revision(self): - session = self.scans.start_session("zone-1", "stock-v1") - self.scans.observe_frame(_req(session, "f1", b"sku_a", 1)) - self.scans.review(session.session_id, actor="op", reason="looks-right") - snap = self.scans.finalize(session.session_id) - self.assertEqual(snap["status"], "finalized") - with self.assertRaises(ObserveError): - self.scans.observe_frame(_req(session, "f2", b"sku_b", 2)) - with self.assertRaises(ObserveError): - self.scans.cancel(session.session_id) - child = self.scans.rescan(session.session_id) - self.assertEqual(child.status, "draft") - self.assertEqual(child.revision, session.revision + 1) - self.assertEqual(child.review["previous_session_id"], session.session_id) - self.assertEqual(len(child.observations), 0) - - def test_live_vs_still_comparator_does_not_sum_frames(self): - sampled = self.scans.start_session("zone-1", "stock-v1", capture_kind=CAPTURE_SAMPLED) - still = self.scans.start_session("zone-1", "stock-v1", capture_kind=CAPTURE_STILL) - self.scans.observe_frame(_req(sampled, "s1", b"sku_a", 1, kind=CAPTURE_SAMPLED)) - self.scans.observe_frame(_req(sampled, "s2", b"sku_a", 2, kind=CAPTURE_SAMPLED)) - self.scans.observe_frame(_req(still, "still1", b"sku_a", 1, kind=CAPTURE_STILL)) - comparison = self.scans.compare_live_vs_still(sampled.session_id, still.session_id) - self.assertEqual(comparison["sampled"]["observation_count"], 2) - self.assertEqual(comparison["still"]["observation_count"], 1) - self.assertEqual(comparison["sampled"]["selected"]["accepted"], 1) - self.assertEqual(comparison["still"]["selected"]["accepted"], 1) - self.assertEqual(sampled.selected_frame_id, "s2") - self.assertIn("not measured shelf accuracy", comparison["note"]) - - def test_expired_session_rejects_new_frames(self): - clock = {"t": 100.0} - - def now(): - return clock["t"] - - scans = ScanService(self.scans.perception, ttl_s=10, clock=now) - session = scans.start_session("zone-1", "stock-v1") - clock["t"] = 120.0 - with self.assertRaises(ObserveError): - scans.observe_frame(_req(session, "f1", b"sku_a", 1)) - - -class SessionHttpTests(unittest.TestCase): - def setUp(self): - from fastapi import FastAPI - from fastapi.testclient import TestClient - from observe import ScanService, PerceptionService, PerceptionAdapter, CAPTURE_SAMPLED - - def detect(frame_bytes: bytes): - return { - "width": 8, - "height": 8, - "boxes": [{"id": 0, "box": [0, 0, 8, 8], "sku_hint": frame_bytes.decode()}], - "empty_slots": [], - "quality": {}, - } - - def classify(box): - hint = box.get("sku_hint", "") - if hint == "sku_b": - return {"sku_a": 2.0, "sku_b": 40.0} - return {"sku_a": 40.0, "sku_b": 2.0} - - self.scans = ScanService( - PerceptionService(PerceptionAdapter(detect_fn=detect, classify_fn=classify)) - ) - app = FastAPI() - - @app.post("/api/sessions") - def start(payload: dict): - session = self.scans.start_session( - payload.get("zone_id", "zone-1"), - payload.get("catalog_version", "stock-v1"), - capture_kind=payload.get("capture_kind", CAPTURE_SAMPLED), - ) - return {"status": "success", "session": session.snapshot()} - - @app.post("/api/sessions/{session_id}/frames") - def frames(session_id: str, payload: dict): - import base64 - session = self.scans.get_session(session_id) - request = FrameRequest( - frame_id=payload["frame_id"], - frame_bytes=base64.b64decode(payload["frame_bytes_b64"]), - sequence=int(payload["sequence"]), - captured_at=1.0, - session_id=session.session_id, - zone_id=session.zone_id, - catalog_version=session.catalog_version, - capture_kind=session.capture_kind, - ) - obs = self.scans.observe_frame(request) - return {"status": "success", "observation": obs.as_dict(), "session": session.snapshot()} - - self.client = TestClient(app) - - def test_http_sessions_are_isolated(self): - import base64 - a = self.client.post("/api/sessions", json={"zone_id": "zone-1", "catalog_version": "stock-v1"}).json() - b = self.client.post("/api/sessions", json={"zone_id": "zone-1", "catalog_version": "stock-v1"}).json() - aid = a["session"]["session_id"] - bid = b["session"]["session_id"] - oa = self.client.post( - f"/api/sessions/{aid}/frames", - json={"frame_id": "f1", "sequence": 1, "frame_bytes_b64": base64.b64encode(b"sku_a").decode()}, - ).json() - ob = self.client.post( - f"/api/sessions/{bid}/frames", - json={"frame_id": "f1", "sequence": 1, "frame_bytes_b64": base64.b64encode(b"sku_b").decode()}, - ).json() - self.assertEqual(oa["observation"]["identities"][0]["best_sku"], "sku_a") - self.assertEqual(ob["observation"]["identities"][0]["best_sku"], "sku_b") - self.assertNotEqual(aid, bid) - - - -if __name__ == "__main__": - unittest.main() diff --git a/recognization/code/test_suite.py b/recognization/code/test_suite.py deleted file mode 100644 index 7a744a6..0000000 --- a/recognization/code/test_suite.py +++ /dev/null @@ -1,91 +0,0 @@ -""" -Unit and Integration Test Suite -Validates InventoryManager, ShelfProductDetector, ProductFeatureMatcher, and Pipeline. -""" - -import os -import sys -import cv2 - -current_dir = os.path.dirname(os.path.abspath(__file__)) -if current_dir not in sys.path: - sys.path.insert(0, current_dir) - -from inventory import InventoryManager -from feature_matcher import ProductFeatureMatcher -from detector import ShelfProductDetector -from pipeline import ShelfRecognitionPipeline - - -def test_inventory(): - print("[TEST] Testing InventoryManager...") - inv = InventoryManager("stock") - assert len(inv.products) >= 10, f"Expected products >= 10 in stock.csv, found {len(inv.products)}" - p1 = inv.get_product("sku_001") - assert p1 is not None, "sku_001 not found" - assert p1["product_name"] == "明治 おいしい牛乳" - assert p1["price"] == 258.0 - assert p1["number_on"] == 73 - print(" -> InventoryManager test: PASSED") - - -def test_feature_matcher(): - print("[TEST] Testing ProductFeatureMatcher...") - matcher = ProductFeatureMatcher("stock") - assert len(matcher.db.templates) >= 7, f"Expected templates >= 7, got {len(matcher.db.templates)}" - - for method in ["hybrid", "orb", "brisk"]: - tmpl_img = matcher.db.templates["sku_001"]["image"] - res = matcher.classify_crop(tmpl_img, method=method) - assert res["status"] == "success", f"Method {method} failed" - assert res["best_sku"] == "sku_001", f"Expected sku_001 with {method}, got {res['best_sku']}" - print(f" -> {method.upper()} classification test: PASSED (Score: {res['score']}, Conf: {res['confidence']})") - - -def test_detector(): - print("[TEST] Testing ShelfProductDetector...") - detector = ShelfProductDetector("runs/detect/train/weights/best.pt") - sample_shelf = "runs/detect/train/train_batch0.jpg" - res = detector.detect(sample_shelf) - assert res["status"] == "success", "Detection failed" - assert res["total_detected"] > 0, "No boxes detected" - print(f" -> Detector test: PASSED ({res['total_detected']} items localized)") - - first_box = res["boxes"][0] - cx, cy = first_box["center"] - hit = detector.find_box_at_point(cx, cy, res["boxes"]) - assert hit is not None, "Hit test failed on center coordinate" - assert hit["id"] == first_box["id"], "Hit test returned incorrect box" - print(" -> Spatial hit-test: PASSED") - - -def test_pipeline(): - print("[TEST] Testing End-to-End Pipeline...") - pipeline = ShelfRecognitionPipeline( - weights_path="runs/detect/train/weights/best.pt", - stock_dir="stock" - ) - init_res = pipeline.set_active_shelf_image("runs/detect/train/train_batch0.jpg") - assert init_res["status"] == "success" - assert len(init_res["boxes"]) > 0 - - first_box = init_res["boxes"][0] - cx, cy = first_box["center"] - click_res = pipeline.process_click(cx, cy, method="hybrid") - assert click_res["status"] == "success", f"Click processing failed: {click_res.get('message')}" - assert "inventory" in click_res - assert click_res["inventory"] is not None - print(f" -> End-to-End Pipeline test: PASSED (Matched: {click_res['classification']['matched_sku']} - {click_res['inventory']['product_name']})") - - -if __name__ == "__main__": - print("=" * 60) - print("STARTING TEST SUITE") - print("=" * 60) - test_inventory() - test_feature_matcher() - test_detector() - test_pipeline() - print("=" * 60) - print("ALL TESTS PASSED SUCCESSFULLY!") - print("=" * 60) diff --git a/recognization/code/yolo_adapter.py b/recognization/code/yolo_adapter.py deleted file mode 100644 index d864369..0000000 --- a/recognization/code/yolo_adapter.py +++ /dev/null @@ -1,64 +0,0 @@ -"""Optional YOLO + matcher adapter for PerceptionService. - -Imported only when OpenCV/YOLO are available. Tests use PerceptionAdapter -fakes instead. This does not keep process-global active images. -""" - -from __future__ import annotations - -import threading -from typing import Any, Dict, List - -import cv2 -import numpy as np - -from observe import PerceptionAdapter - - -class YoloMatcherAdapter(PerceptionAdapter): - def __init__(self, detector: Any, matcher: Any, method: str = "hybrid"): - super().__init__( - model_version=getattr(detector, "weights_path", "yolo-local"), - matcher_method=method, - feature_extractor="surf-or-sift+orb+brisk", - ) - self.detector = detector - self.matcher = matcher - # observe() calls detect() then classify_box() on the same thread; keep - # the decoded frame per thread so concurrent sessions cannot swap images. - self._thread_local = threading.local() - - def detect(self, frame_bytes: bytes) -> Dict[str, Any]: - array = np.frombuffer(frame_bytes, dtype=np.uint8) - image = cv2.imdecode(array, cv2.IMREAD_COLOR) - if image is None: - return { - "width": 0, - "height": 0, - "boxes": [], - "empty_slots": [], - "quality": {"limitation": "undecodable-frame"}, - } - self._thread_local.image = image - h, w = image.shape[:2] - det = self.detector.detect(image) - boxes: List[Dict[str, Any]] = det.get("boxes", []) - empty = self.detector.detect_empty_slots(image.shape, boxes) - return { - "width": w, - "height": h, - "boxes": boxes, - "empty_slots": empty, - "quality": {"decoded": True}, - } - - def classify_box(self, box: Dict[str, Any]) -> Dict[str, float]: - image = getattr(self._thread_local, "image", None) - if image is None: - return {} - crop = self.detector.crop_roi(image, box["box"]) - ranked = self.matcher.classify_crop(crop, method=self.matcher_method) - return { - row["sku_id"]: float(row["score"]) - for row in ranked.get("ranked_candidates", []) - } diff --git a/recognization/docs/API.md b/recognization/docs/API.md index f3da7b2..2cfec12 100644 --- a/recognization/docs/API.md +++ b/recognization/docs/API.md @@ -334,7 +334,6 @@ just drops frames and breaks the pan's continuity for the tracker. | CPU as the default device | Measured: 26-29 ms p50 vs 112 ms on MPS, with byte-identical boxes | MPS (slower here), CUDA (no NVIDIA GPU on this machine) | | Identity states instead of raw scores | "Never turn model uncertainty into business certainty" (project master) | Showing a best-guess SKU with a confidence number | | Token via env var, access log off | Keeps credentials out of shell history and logs | A `--token` flag, default uvicorn access log (prints the query string) | -| Legacy `/api/sessions*` kept, engine-backed | The earlier session model is still useful and was already tested | Deleting it (loses the work), leaving it on the fake adapter (silently returned nothing) | ## Measured performance