diff --git a/recognization/code/live.py b/recognization/code/live.py new file mode 100644 index 0000000..676661b --- /dev/null +++ b/recognization/code/live.py @@ -0,0 +1,795 @@ +""" +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, + }