vnguyen/product-recognition-demo
1
0
Fork 1
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions
product-recognition-demo/recognization/code/live.py
kiyo9w afb59b487f feat: live recognition session (track, classify once, label) 
One LiveSession per streaming client. Each frame runs detection and tracking,
then classification is scheduled for products that are new, big enough, fully in
frame and not already settled; results are cached on the track so detection
never waits for classification. Empty-slot gaps are smoothed over frames and
quality hints (glare, blur, too far) come from a 320 px downscale.

Correctness work that the acceptance run forced out:

- Classification submissions happen OUTSIDE the session lock. add_done_callback
  runs the callback inline when the future already completed, so attaching it
  while holding the lock deadlocked the session on the second frame.
- A label is dropped only when its box has drifted off the product for two
  consecutive frames (panning slides boxes onto neighbours), and the anchor is
  the box the crop came from, not wherever the track moved by the time
  classification finished.
- An id that disappears and comes back is a restart, not the same object: the
  old label is cleared instead of silently labelling a neighbour. relabels and
  restarts counters are monotonic per id so clients can tell a re-verification
  from a silent flip.
- Track ids are tracked per session (tracking.py) and counters survive pruning.

Measured on the synthetic pan (30 frames, 248 label observations): 20
re-verifications, 0 unexplained label changes, detection 20 ms/frame.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-24 12:21:44 +09:00

795 lines
32 KiB
Python
Raw Blame History

"""
Live recognition session: one per streaming client.
Per frame it runs detection and tracking, then schedules classification of each
physical product ONCE in a background pool and remembers the result on the
track. Labels therefore fill in as fast as the hardware allows while detection
keeps the frame rate.
Pure Python: no FastAPI imports, no globals. Everything a session owns is behind
one lock, and the heavy per-frame work (decode/detect/track) runs outside that
lock so classification callbacks can land concurrently.
"""
from __future__ import annotations
import collections
import os
import statistics
import sys
import threading
import time
from concurrent.futures import Future, ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Tuple
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
from engine import ( # noqa: E402
STATE_DETECTED,
STATE_IDENTIFYING,
STATE_NEEDS_CONFIRMATION,
STATE_RECOGNIZED,
STATE_UNRECOGNIZED,
RecognitionEngine,
)
from tracking import SessionTracker # noqa: E402
EMPTY_SLOT_HISTORY = 5
EMPTY_SLOT_MIN_HITS = 3
EMPTY_SLOT_IOU = 0.3
STATS_WINDOW = 30
# A recognized label is only trustworthy while the box still covers the product
# it was measured on. Below this IoU against the box at label time, the track has
# slid onto a different product (typical when the camera pans) and must be
# re-classified rather than keep a stale label. 0.5 means more than half the box
# still overlaps the labelled product; a neighbour-slide crosses it.
DRIFT_IOU = 0.5
STATE_UNCONFIRMED = "unconfirmed"
@dataclass
class LiveSettings:
method: str = "hybrid"
conf: float = 0.1
empty_slots: bool = True
min_hits: int = 2
min_crop_px: int = 40
max_attempts: int = 3
retry_gap: int = 8
area_growth: float = 1.3
max_inflight_per_session: int = 6
edge_margin_ratio: float = 0.015
glare_fraction: float = 0.02
glare_pixel: int = 245
too_far_height_ratio: float = 0.08
@dataclass
class TrackState:
track_id: int
first_seq: int
last_seq: int
hits: int = 1
box: List[int] = field(default_factory=list)
raw_box: List[int] = field(default_factory=list)
confidence: float = 0.0
state: str = STATE_DETECTED
attempts: int = 0
in_flight: int = 0
last_attempt_seq: int = 0
last_attempt_area: int = 0
score_sums: Dict[str, float] = field(default_factory=dict)
score_n: int = 0
identity: Optional[Dict[str, Any]] = None
candidates: List[Dict[str, Any]] = field(default_factory=list)
confirmed: bool = False
label_box: List[int] = field(default_factory=list)
drift_frames: int = 0
# Bumped every time the tracker starts this id over. A classification that
# finishes after the id was recycled belongs to the old object and must not
# label the new one.
object_generation: int = 0
# How many times a settled label was dropped and re-verified (drift or id
# recycle). A label change without a relabel would be a defect; the counter
# makes that auditable from the result payload.
relabels: int = 0
# Monotonic: how many times this id was started over (absent then re-detected,
# or pruned then re-created). Restarting resets the label, so it is always a
# re-verification rather than an unexplained label change.
restarts: int = 0
def area(self) -> int:
if not self.raw_box:
return 0
return max(0, self.raw_box[2] - self.raw_box[0]) * max(0, self.raw_box[3] - self.raw_box[1])
def _iou(a: Sequence[int], b: Sequence[int]) -> float:
ax1, ay1, ax2, ay2 = a
bx1, by1, bx2, by2 = b
ix1, iy1 = max(ax1, bx1), max(ay1, by1)
ix2, iy2 = min(ax2, bx2), min(ay2, by2)
iw, ih = max(0, ix2 - ix1), max(0, iy2 - iy1)
inter = iw * ih
if inter <= 0:
return 0.0
area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1)
area_b = max(0, bx2 - bx1) * max(0, by2 - by1)
union = area_a + area_b - inter
return inter / union if union > 0 else 0.0
class LiveSession:
"""Stateful live session for one streaming client."""
def __init__(
self,
session_id: str,
engine: RecognitionEngine,
*,
settings: Optional[LiveSettings] = None,
tracker: Optional[SessionTracker] = None,
executor: Optional[ThreadPoolExecutor] = None,
):
self.session_id = session_id
self.engine = engine
self.settings = settings or LiveSettings()
self.tracker = tracker or SessionTracker()
self._executor = executor or engine.executor
self._lock = threading.Lock()
self._generation = 0
self._tracks: Dict[int, TrackState] = {}
self._last_seen_seq: Dict[int, int] = {}
# Monotonic per-id counters that outlive track-state pruning, so a client
# can always tell a label change caused by re-verification from a silent
# flip (see TrackState.relabels / .restarts).
self._relabel_count: Dict[int, int] = {}
self._restart_count: Dict[int, int] = {}
self._latest_image: Optional[Any] = None
self._latest_seq = 0
self._latest_frame_id: Optional[int] = None
self._pending: Optional[Tuple[int, bytes]] = None
self._frame_ready = threading.Event()
self._closed = False
self._seq = 0
self._frames_received = 0
self._frames_processed = 0
self._frames_dropped = 0
self._process_times: "collections.deque[float]" = collections.deque(maxlen=STATS_WINDOW)
self._classify_times: "collections.deque[float]" = collections.deque(maxlen=STATS_WINDOW)
self._slot_history: "collections.deque[List[Dict[str, Any]]]" = collections.deque(
maxlen=EMPTY_SLOT_HISTORY
)
self._slots: List[Dict[str, Any]] = []
self._slot_counter = 0
self._catalog_by_sku: Dict[str, Dict[str, Any]] = {}
self._catalog_version = ""
self._last_quality: Dict[str, Any] = {}
self._created_at = time.time()
# ------------------------------------------------------------------ control
@property
def generation(self) -> int:
with self._lock:
return self._generation
def submit_frame(self, frame_id: int, payload: bytes) -> bool:
"""Latest-frame-wins: replaces any frame not yet picked up by the loop."""
with self._lock:
if self._closed:
return False
self._frames_received += 1
if self._pending is not None:
self._frames_dropped += 1
self._pending = (int(frame_id), payload)
self._frame_ready.set()
return True
def take_pending(self, timeout: float = 0.2) -> Optional[Tuple[int, bytes]]:
"""Blocking pop for the processing loop; None on timeout/close."""
if not self._frame_ready.wait(timeout):
return None
with self._lock:
pending = self._pending
self._pending = None
self._frame_ready.clear()
return pending
def close(self) -> None:
with self._lock:
self._closed = True
self._pending = None
self._frame_ready.set()
@property
def closed(self) -> bool:
with self._lock:
return self._closed
def configure(
self,
*,
method: Optional[str] = None,
conf: Optional[float] = None,
empty_slots: Optional[bool] = None,
) -> Dict[str, Any]:
with self._lock:
method_changed = method is not None and method != self.settings.method
if method is not None:
self.settings.method = method
if conf is not None:
self.settings.conf = float(conf)
if empty_slots is not None:
self.settings.empty_slots = bool(empty_slots)
if method_changed:
self._generation += 1
for track in self._tracks.values():
track.state = STATE_DETECTED
track.identity = None
track.candidates = []
track.score_sums = {}
track.score_n = 0
track.attempts = 0
track.in_flight = 0
track.last_attempt_seq = 0
track.last_attempt_area = 0
return self.config()
def config(self) -> Dict[str, Any]:
return {
"method": self.settings.method,
"conf": self.settings.conf,
"empty_slots": self.settings.empty_slots,
}
def reset(self) -> None:
with self._lock:
self._generation += 1
self.tracker.reset()
self._tracks.clear()
self._last_seen_seq.clear()
self._relabel_count.clear()
self._restart_count.clear()
self._slot_history.clear()
self._slots = []
self._latest_image = None
self._latest_frame_id = None
# ------------------------------------------------------------------ frame path
def process_frame(self, frame_id: int, payload: bytes) -> Dict[str, Any]:
"""Decode -> detect -> track -> schedule classification -> result dict."""
started = time.perf_counter()
image = self.engine.decode_image(payload)
decoded_at = time.perf_counter()
with self._lock:
conf = self.settings.conf
method = self.settings.method
empty_enabled = self.settings.empty_slots
generation = self._generation
boxes = self.engine.detect(image, conf=conf)
detected_at = time.perf_counter()
tracks = self.tracker.update(
[box["box"] for box in boxes], [box["confidence"] for box in boxes]
)
tracked_at = time.perf_counter()
with self._lock:
self._seq += 1
seq = self._seq
self._frames_processed += 1
self._latest_image = image
self._latest_seq = seq
self._latest_frame_id = int(frame_id)
self._process_times.append(tracked_at)
by_id = {box["id"]: box for box in boxes}
active: Dict[int, TrackState] = {}
for track in tracks:
state = self._tracks.get(track.track_id)
raw_box = None
confidence = track.score
if track.det_index is not None and track.det_index in by_id:
detection = by_id[track.det_index]
raw_box = list(detection["box"])
confidence = float(detection["confidence"])
if raw_box is None:
raw_box = list(track.box)
previous_seq = self._last_seen_seq.get(track.track_id)
if state is None:
# An id that comes back after an absence may be a recycled id:
# the tracker reuses numbers after it drops a track, so the old
# label cannot be trusted and the object counts as restarted.
recycled = previous_seq is not None and seq - previous_seq > 1
if recycled:
self._restart_count[track.track_id] = (
self._restart_count.get(track.track_id, 0) + 1
)
self._relabel_count[track.track_id] = (
self._relabel_count.get(track.track_id, 0) + 1
)
state = TrackState(
track_id=track.track_id,
first_seq=seq,
last_seq=seq,
object_generation=1 if recycled else 0,
relabels=self._relabel_count.get(track.track_id, 0),
restarts=self._restart_count.get(track.track_id, 0),
)
elif seq - state.last_seq > 1:
# The id was absent and the tracker started it over. The
# previous label belonged to the old object, so start clean
# rather than keep a label that may now name a neighbour.
self._restart_count[track.track_id] = self._restart_count.get(track.track_id, 0) + 1
self._relabel_count[track.track_id] = self._relabel_count.get(track.track_id, 0) + 1
state = TrackState(
track_id=track.track_id,
first_seq=state.first_seq,
last_seq=seq,
hits=1,
object_generation=state.object_generation + 1,
relabels=self._relabel_count[track.track_id],
restarts=self._restart_count[track.track_id],
)
else:
state.hits += 1
state.last_seq = seq
self._last_seen_seq[track.track_id] = seq
state.box = list(track.box)
state.raw_box = raw_box
state.confidence = confidence
self._drop_stale_label(state)
active[track.track_id] = state
self._tracks = active
if len(self._last_seen_seq) > 5000:
live_ids = set(self._tracks)
keep = {
tid: seen for tid, seen in self._last_seen_seq.items()
if tid in live_ids or seq - seen <= 300
}
self._last_seen_seq = keep
self._relabel_count = {tid: c for tid, c in self._relabel_count.items() if tid in keep}
self._restart_count = {tid: c for tid, c in self._restart_count.items() if tid in keep}
if empty_enabled:
slot_boxes = [
{"box": state.box} for state in self._tracks.values() if state.box
]
raw_slots = self.engine.empty_slots(image.shape, slot_boxes)
else:
raw_slots = []
self._slots = self._persist_slots(raw_slots)
quality = self._quality(image, self._tracks)
# Collect the classification work under the lock; submit it after
# releasing it. add_done_callback can run the callback inline when the
# future already finished, and the callback takes this same lock.
submissions = self._plan_classification(image, seq)
summary = self._summary_locked()
stats = self._stats_locked()
track_payload = [self._track_payload(state) for state in self._tracks.values()]
self._submit_classification(submissions, method, generation)
result = {
"type": "result",
"frame_id": int(frame_id),
"seq": seq,
"width": int(image.shape[1]),
"height": int(image.shape[0]),
"tracks": track_payload,
"empty_slots": self._slots,
"quality": quality,
"summary": summary,
"timing_ms": {
"decode": round((decoded_at - started) * 1000.0, 2),
"detect": round((detected_at - decoded_at) * 1000.0, 2),
"track": round((tracked_at - detected_at) * 1000.0, 2),
"total": round((tracked_at - started) * 1000.0, 2),
},
"stats": stats,
}
return result
def _drop_stale_label(self, track: TrackState) -> None:
"""Reset a settled label when the box has drifted off its product.
Panning cameras make trackers slide a box onto the neighbouring product.
A label measured on the old box would then name the wrong product, so the
track goes back to detected and is classified again.
Two consecutive drifted frames are required: one noisy frame (a bad
detection or a blurred capture) must not churn a good label.
"""
if track.state not in (STATE_RECOGNIZED, STATE_NEEDS_CONFIRMATION, STATE_UNRECOGNIZED):
return
if track.label_box and _iou(track.box, track.label_box) < DRIFT_IOU:
track.drift_frames += 1
else:
track.drift_frames = 0
if track.drift_frames < 2:
return
track.drift_frames = 0
self._relabel_count[track.track_id] = self._relabel_count.get(track.track_id, 0) + 1
track.relabels = self._relabel_count[track.track_id]
track.state = STATE_DETECTED
track.identity = None
track.candidates = []
track.score_sums = {}
track.score_n = 0
track.attempts = 0
track.in_flight = 0
track.last_attempt_seq = 0
track.last_attempt_area = 0
track.label_box = []
# ------------------------------------------------------------------ classification
def _plan_classification(self, image: Any, seq: int) -> List[Tuple[TrackState, Any, int]]:
"""Pick what to classify now. Must be called with the session lock held.
Returns (track, crop, object_generation) triples; the caller submits them
after the lock is released.
"""
frame_h, frame_w = image.shape[:2]
margin_x = int(frame_w * self.settings.edge_margin_ratio)
margin_y = int(frame_h * self.settings.edge_margin_ratio)
inflight = sum(track.in_flight for track in self._tracks.values())
capacity = max(0, self.settings.max_inflight_per_session - inflight)
if capacity <= 0:
return []
eligible: List[TrackState] = []
for track in self._tracks.values():
if track.state != STATE_DETECTED or track.in_flight > 0:
continue
if track.hits < self.settings.min_hits:
continue
x1, y1, x2, y2 = track.raw_box
if min(x2 - x1, y2 - y1) < self.settings.min_crop_px:
continue
if x1 <= margin_x or y1 <= margin_y or x2 >= frame_w - margin_x or y2 >= frame_h - margin_y:
continue
if track.attempts >= self.settings.max_attempts:
continue
if track.attempts > 0 and (seq - track.last_attempt_seq) < self.settings.retry_gap:
if track.area() < self.settings.area_growth * max(1, track.last_attempt_area):
continue
eligible.append(track)
eligible.sort(key=lambda track: track.area(), reverse=True)
planned: List[Tuple[TrackState, Any, int]] = []
for track in eligible[:capacity]:
crop = self.engine.crop(image, track.raw_box)
track.in_flight += 1
track.last_attempt_seq = seq
track.last_attempt_area = track.area()
planned.append((track, crop, track.object_generation))
return planned
def _submit_classification(
self, planned: List[Tuple[TrackState, Any, int]], method: str, generation: int
) -> None:
"""Submit planned crops. Called WITHOUT the session lock held."""
for track, crop, object_generation in planned:
submitted_at = time.perf_counter()
# The label anchors to the displayed (smoothed) box, which is what the
# drift check later compares against.
submitted_box = list(track.box)
track_id = track.track_id
try:
future = self._executor.submit(self.engine.score_crop, crop, method)
except RuntimeError: # executor already shut down
with self._lock:
track.in_flight = max(0, track.in_flight - 1)
continue
future.add_done_callback(
lambda fut, tid=track_id, gen=generation, m=method, started=submitted_at,
obj=object_generation, box=submitted_box: self._on_classified(
fut, tid, gen, m, started, obj, box
)
)
def _on_classified(
self,
future: Future,
track_id: int,
generation: int,
method: str,
submitted_at: float,
object_generation: int,
submitted_box: List[int],
) -> None:
try:
scores = future.result()
except Exception:
scores = {}
finished_at = time.perf_counter()
with self._lock:
track = self._tracks.get(track_id)
if track is None or generation != self._generation:
return
if track.object_generation != object_generation:
# The id was recycled while this crop was being classified.
track.in_flight = max(0, track.in_flight - 1)
return
track.in_flight = max(0, track.in_flight - 1)
track.attempts += 1
for sku_id, score in (scores or {}).items():
track.score_sums[sku_id] = track.score_sums.get(sku_id, 0.0) + float(score)
if scores:
track.score_n += 1
self._classify_times.append(finished_at - submitted_at)
if not scores:
if track.attempts >= self.settings.max_attempts:
track.state = STATE_UNRECOGNIZED
track.confirmed = True
track.label_box = list(submitted_box)
track.identity = None
track.candidates = []
return
from identity import decide_identity
mean_scores = {
sku_id: total / max(1, track.score_n) for sku_id, total in track.score_sums.items()
}
decision = decide_identity(mean_scores, method=method)
track.candidates = [
{"sku_id": row.get("sku_id"), "score": row.get("score")}
for row in (decision.get("ranked_candidates") or [])[:3]
]
if decision.get("identity_level") == "accepted":
track.state = STATE_RECOGNIZED
track.confirmed = True
# The label belongs to the box the crop came from, not to wherever
# the track has moved by the time classification finishes.
track.label_box = list(submitted_box)
track.drift_frames = 0
track.identity = {
"sku_id": decision.get("best_sku"),
"confidence": decision.get("confidence"),
"score": decision.get("score"),
"attempts": track.attempts,
}
elif track.attempts >= self.settings.max_attempts:
track.state = (
STATE_NEEDS_CONFIRMATION
if decision.get("identity_level") in ("ambiguous", "unknown")
else STATE_UNRECOGNIZED
)
track.confirmed = True
track.label_box = list(submitted_box)
track.drift_frames = 0
track.identity = None
else:
track.state = STATE_DETECTED
track.confirmed = False
track.identity = None
# ------------------------------------------------------------------ payloads
def _track_payload(self, track: TrackState) -> Dict[str, Any]:
wire_state = track.state
if not track.confirmed and wire_state == STATE_DETECTED:
wire_state = STATE_UNCONFIRMED
identity: Dict[str, Any] = {
"state": wire_state,
"sku_id": None,
"identity_level": None,
"confidence": None,
"score": None,
"candidates": list(track.candidates),
"attempts": track.attempts,
"method": self.settings.method,
}
product = None
if track.state == STATE_RECOGNIZED and track.identity:
identity["sku_id"] = track.identity.get("sku_id")
identity["confidence"] = track.identity.get("confidence")
identity["score"] = track.identity.get("score")
identity["identity_level"] = "accepted"
product = self._catalog_by_sku.get(identity["sku_id"] or "")
elif track.state in (STATE_NEEDS_CONFIRMATION, STATE_UNRECOGNIZED):
identity["identity_level"] = (
"ambiguous" if track.state == STATE_NEEDS_CONFIRMATION else "rejected"
)
payload = {
"track_id": track.track_id,
# box: smoothed track box, for drawing. raw_box: the detection box the
# label is anchored to, for cropping and for explaining label changes.
"box": list(track.box),
"raw_box": list(track.raw_box),
"confidence": round(float(track.confidence), 4),
"first_seen_seq": track.first_seq,
"relabels": track.relabels,
"restarts": track.restarts,
"identity": identity,
}
if product is not None:
payload["product"] = {
"sku_id": product.get("sku_id"),
"product_name": product.get("product_name"),
"price": product.get("price"),
}
return payload
def _summary_locked(self) -> Dict[str, Any]:
counts = {
STATE_RECOGNIZED: 0,
STATE_NEEDS_CONFIRMATION: 0,
STATE_UNRECOGNIZED: 0,
STATE_DETECTED: 0,
STATE_IDENTIFYING: 0,
STATE_UNCONFIRMED: 0,
}
facings: Dict[str, int] = {}
for track in self._tracks.values():
if track.state == STATE_DETECTED and not track.confirmed:
counts[STATE_UNCONFIRMED] += 1
continue
state = track.state if track.state in counts else STATE_DETECTED
counts[state] += 1
if state == STATE_RECOGNIZED and track.identity:
sku_id = track.identity.get("sku_id")
if sku_id:
facings[sku_id] = facings.get(sku_id, 0) + 1
return {
"in_view": len(self._tracks),
"recognized": counts[STATE_RECOGNIZED],
"identifying": counts[STATE_IDENTIFYING],
"needs_confirmation": counts[STATE_NEEDS_CONFIRMATION],
"unrecognized": counts[STATE_UNRECOGNIZED],
"detected": counts[STATE_DETECTED],
"unconfirmed": counts[STATE_UNCONFIRMED],
"empty_slots": len(self._slots),
"by_sku": [
{"sku_id": sku, "facings": count}
for sku, count in sorted(facings.items(), key=lambda kv: (-kv[1], kv[0]))
],
}
def _stats_locked(self) -> Dict[str, Any]:
times = list(self._process_times)
fps = 0.0
if len(times) >= 2 and times[-1] > times[0]:
fps = round((len(times) - 1) / (times[-1] - times[0]), 2)
classify_p50 = None
if self._classify_times:
classify_p50 = round(statistics.median(self._classify_times) * 1000.0, 2)
return {
"frames_received": self._frames_received,
"frames_processed": self._frames_processed,
"frames_dropped": self._frames_dropped,
"server_fps": fps,
"classify_pending": sum(track.in_flight for track in self._tracks.values()),
"classify_ms_p50": classify_p50,
}
# ------------------------------------------------------------------ empty slots
def _persist_slots(self, raw_slots: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Keep only slots seen repeatedly, with stable ids across frames."""
previous = list(self._slots)
matched_ids: Dict[int, int] = {}
used_previous = set()
for index, slot in enumerate(raw_slots):
best_id = None
best_iou = EMPTY_SLOT_IOU
for prev_index, prev in enumerate(previous):
if prev_index in used_previous:
continue
score = _iou(slot["box"], prev["box"])
if score >= best_iou:
best_iou = score
best_id = prev.get("id")
best_prev_index = prev_index
if best_id is not None:
matched_ids[index] = best_id
used_previous.add(best_prev_index)
assigned: List[Dict[str, Any]] = []
for index, slot in enumerate(raw_slots):
slot_id = matched_ids.get(index)
if slot_id is None:
self._slot_counter += 1
slot_id = f"empty_{self._slot_counter}"
entry = dict(slot)
entry["id"] = slot_id
assigned.append(entry)
self._slot_history.append(assigned)
if len(self._slot_history) < EMPTY_SLOT_MIN_HITS:
return []
counts: Dict[str, int] = collections.Counter()
latest_by_id: Dict[str, Dict[str, Any]] = {}
for frame_slots in self._slot_history:
for slot in frame_slots:
counts[slot["id"]] += 1
latest_by_id[slot["id"]] = slot
return [
latest_by_id[slot_id]
for slot_id, count in counts.items()
if count >= EMPTY_SLOT_MIN_HITS and slot_id in latest_by_id
]
# ------------------------------------------------------------------ quality
def _quality(self, image: Any, tracks: Dict[int, TrackState]) -> Dict[str, Any]:
import cv2
import numpy as np
height, width = image.shape[:2]
target_width = 320
scale = target_width / float(width) if width > target_width else 1.0
small = cv2.resize(image, (max(1, int(width * scale)), max(1, int(height * scale))),
interpolation=cv2.INTER_AREA)
gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY)
sharpness = float(cv2.Laplacian(gray, cv2.CV_64F).var())
glare = float(np.mean(np.all(small > self.settings.glare_pixel, axis=2)))
heights = [state.raw_box[3] - state.raw_box[1] for state in tracks.values() if state.raw_box]
median_height = float(np.median(heights)) if heights else 0.0
too_far = bool(
heights
and (
median_height < self.settings.too_far_height_ratio * height
or sum(1 for h in heights if h < self.settings.min_crop_px) > len(heights) / 2
)
)
blur = sharpness < self._blur_threshold()
hint = None
if glare > self.settings.glare_fraction:
hint = "glare"
elif blur:
hint = "blur"
elif too_far:
hint = "too_far"
quality = {
"sharpness": round(sharpness, 1),
"blur": blur,
"glare": round(glare, 4),
"too_far": too_far,
"hint": hint,
}
self._last_quality = quality
return quality
def _blur_threshold(self) -> float:
return float(os.environ.get("DORAYAKI_BLUR_THRESHOLD", "45"))
# ------------------------------------------------------------------ inspect
def inspect(self, track_id: int, method: Optional[str] = None) -> Dict[str, Any]:
with self._lock:
track = self._tracks.get(track_id)
image = self._latest_image
chosen_method = (method or self.settings.method).lower()
if track is None or image is None:
raise KeyError("TRACK_NOT_IN_VIEW")
box = list(track.raw_box)
return self.engine.inspect(image, box, method=chosen_method)
# ------------------------------------------------------------------ catalog cache
def refresh_catalog(self) -> None:
products = self.engine.catalog()
with self._lock:
self._catalog_by_sku = {row["sku_id"]: row for row in products}
self._catalog_version = self.engine.catalog_version()
def snapshot(self) -> Dict[str, Any]:
with self._lock:
return {
"session_id": self.session_id,
"config": self.config(),
"tracks": len(self._tracks),
"seq": self._seq,
"stats": self._stats_locked(),
"quality": dict(self._last_quality),
"created_at": self._created_at,
}
Reference in a new issue View git blame Copy permalink