356f7aa502
Keeps the model's uncertainty out of the UI: decide_identity() separates raw matcher scores into accepted / ambiguous / unknown / rejected, and only an accepted result can name a SKU. Zero evidence is rejected, never a product. observe.py adds frame-local observations and one-zone scan sessions (draft, finalize, cancel, rescan) with replay, staleness and backpressure rules. pipeline.process_click() now reports the identity level instead of failing when the match is not accepted. Incorporates earlier uncommitted session work in this repository (identity.py, observe.py, yolo_adapter.py, test_observe.py, and the pipeline hunk those needed); yolo_adapter keeps its decoded frame per thread so concurrent sessions cannot swap images. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
"""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", [])
|
|
}
|