diff --git a/recognization/code/identity.py b/recognization/code/identity.py new file mode 100644 index 0000000..2514f91 --- /dev/null +++ b/recognization/code/identity.py @@ -0,0 +1,106 @@ +"""Identity ranking for catalog scores. + +Separates raw matcher scores from accept/unknown/ambiguous/reject so a +zero-score crop cannot become a SKU. Thresholds are conservative defaults +for the 21-SKU closed set; they are not calibrated accuracy claims. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Sequence, Tuple + +IDENTITY_ACCEPTED = "accepted" +IDENTITY_AMBIGUOUS = "ambiguous" +IDENTITY_UNKNOWN = "unknown" +IDENTITY_REJECTED = "rejected" + +# ponytail: closed-set defaults. Recalibrate when held-out scores exist. +DEFAULT_MIN_SCORE = 1.0 +DEFAULT_MIN_MARGIN = 0.15 +DEFAULT_MIN_CONFIDENCE = 0.55 + + +def relative_confidence(top_score: float, runner_up_score: float) -> float: + """Existing matcher formula: top / (top + 0.5 * runner).""" + return round(min(1.0, top_score / max(1.0, (top_score + runner_up_score * 0.5))), 4) + + +def decide_identity( + scores: Dict[str, float], + *, + min_score: float = DEFAULT_MIN_SCORE, + min_margin: float = DEFAULT_MIN_MARGIN, + min_confidence: float = DEFAULT_MIN_CONFIDENCE, + details: Optional[Dict[str, Any]] = None, + method: str = "hybrid", +) -> Dict[str, Any]: + """Rank catalog scores into accepted / ambiguous / unknown / rejected. + + Zero or missing evidence is rejected. A weak but unique top score is + unknown. Two close scores are ambiguous. Only a unique, above-threshold + score is accepted. + """ + if not scores: + return { + "status": IDENTITY_REJECTED, + "method": method, + "best_sku": None, + "score": 0.0, + "confidence": 0.0, + "identity_level": IDENTITY_REJECTED, + "ranked_candidates": [], + "details": {}, + "message": "No templates available", + } + + sorted_candidates: List[Tuple[str, float]] = sorted( + scores.items(), key=lambda x: x[1], reverse=True + ) + top_sku, top_score = sorted_candidates[0] + runner_up_sku, runner_up_score = ( + sorted_candidates[1] if len(sorted_candidates) > 1 else (None, 0.0) + ) + confidence = relative_confidence(top_score, runner_up_score) + ranked = [{"sku_id": s, "score": round(sc, 2)} for s, sc in sorted_candidates[:5]] + top_details = (details or {}).get(top_sku, {}) + + if top_score <= 0.0: + level = IDENTITY_REJECTED + message = "No matching evidence" + best = None + elif ( + runner_up_sku is not None + and top_score > 0.0 + and (top_score - runner_up_score) < min_margin * max(top_score, 1.0) + ): + level = IDENTITY_AMBIGUOUS + message = f"Close scores: {top_sku} vs {runner_up_sku}" + best = None + elif top_score < min_score or confidence < min_confidence: + level = IDENTITY_UNKNOWN + message = "Top score below accept threshold" + best = None + else: + level = IDENTITY_ACCEPTED + message = "Accepted unique catalog match" + best = top_sku + + return { + "status": "success" if level == IDENTITY_ACCEPTED else level, + "method": method, + "best_sku": best, + "score": round(top_score, 2), + "confidence": confidence, + "identity_level": level, + "ranked_candidates": ranked, + "details": top_details, + "message": message, + } + + +def identity_from_ranked( + ranked_candidates: Sequence[Dict[str, Any]], + **kwargs: Any, +) -> Dict[str, Any]: + scores = {str(row["sku_id"]): float(row["score"]) for row in ranked_candidates} + return decide_identity(scores, **kwargs) diff --git a/recognization/code/observe.py b/recognization/code/observe.py new file mode 100644 index 0000000..255fbe4 --- /dev/null +++ b/recognization/code/observe.py @@ -0,0 +1,474 @@ +"""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 index cb766d2..403efe6 100644 --- a/recognization/code/pipeline.py +++ b/recognization/code/pipeline.py @@ -116,34 +116,31 @@ class ShelfRecognitionPipeline: # Classify with selected method cls_result = self.matcher.classify_crop(crop, method=method) - if cls_result.get("status") != "success": - return { - "status": "classification_failed", - "target_box": target_box, - "message": cls_result.get("message", "Feature matching failed.") - } + 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 - best_sku = cls_result["best_sku"] - product_info = self.inventory.get_product(best_sku) + vis_match = None + if best_sku: + vis_match = self.matcher.generate_match_visualization(crop, best_sku, method=method) - # Generate match visualization - vis_match = self.matcher.generate_match_visualization(crop, best_sku, method=method) - - # Generate annotated shelf image highlighting the selected item 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": "success", + "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, @@ -172,9 +169,10 @@ if __name__ == "__main__": res = pipeline.process_click(cx, cy, method=args.method) print("\n--- Recognition & Inventory Result ---") print(f"Status: {res['status']}") - if res['status'] == 'success': - prod = res['inventory'] - print(f"Matched SKU: {res['classification']['matched_sku']} (Score: {res['classification']['score']})") + 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']})") diff --git a/recognization/code/test_observe.py b/recognization/code/test_observe.py new file mode 100644 index 0000000..6dd46a4 --- /dev/null +++ b/recognization/code/test_observe.py @@ -0,0 +1,319 @@ +"""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/yolo_adapter.py b/recognization/code/yolo_adapter.py new file mode 100644 index 0000000..d864369 --- /dev/null +++ b/recognization/code/yolo_adapter.py @@ -0,0 +1,64 @@ +"""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", []) + }