""" 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 asyncio import collections import os import statistics import sys import logging import threading import time from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass, field LOGGER = logging.getLogger("dorayaki.live") from typing import Any, Dict, List, Optional, Sequence, Set, 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 # Label memory: a recognized product keeps its label for any track that covers # the same place soon after, so panning does not re-classify one product many # times just because the tracker re-issued its id. LABEL_MEMORY_IOU = 0.55 LABEL_MEMORY_FRAMES = 45 LABEL_MEMORY_MAX = 200 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" # Which evidence produced the newest observation on a track. A label is only ever # shown for an accepted match, so this says how that match was measured. SOURCE_FRAME = "frame" SOURCE_CROP = "crop" SOURCE_INHERITED = "inherited" # A crop the client returns is base64 inside a text control message, so it is # bounded on its own rather than riding the binary frame path. MAX_CROP_BYTES = 1024 * 1024 # Frame faults a full-resolution crop cannot undo. Blur and glare are optical # faults of the whole capture, so the phone would photograph the same bad # scene. "too_far" is exempt on purpose: a distant shelf is the one frame a # native-resolution crop exists to decide. CROP_BLIND_QUALITY = ("blur", "glare") @dataclass class LiveSettings: method: str = "hybrid" conf: float = 0.1 empty_slots: bool = True tags: bool = True min_hits: int = 2 min_crop_px: int = 40 max_attempts: int = 2 retry_gap: int = 8 # A product drifting out of the frame is not worth a retry: the budget belongs # to what is still in front of the camera. retry_edge_ratio: float = 0.06 area_growth: float = 1.3 max_inflight_per_session: int = 16 edge_margin_ratio: float = 0.015 glare_fraction: float = 0.02 glare_pixel: int = 245 too_far_height_ratio: float = 0.08 # Full-resolution crop lane. Off until a client advertises it, so a # protocol-v1 client is never asked for a crop and its traffic is unchanged. crops: bool = False max_crop_bytes: int = 128 * 1024 # A crop answered this many processed frames late describes a product that # may have moved on, so it is dropped instead of relabelling the track. crop_ttl_frames: int = 6 # Requests outstanding at once. Bounds what one phone can owe us. max_crop_inflight: int = 4 # Ask for a native-resolution crop while the box is still this small. At the # camera's own resolution it may still be large enough to classify. crop_below_px: int = 64 @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 observations: List[Dict[str, float]] = field(default_factory=list) candidates: List[Dict[str, Any]] = field(default_factory=list) identity: Optional[Dict[str, Any]] = None 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 # How many full-resolution crops have been asked for on this object. It is # the budget the frame path spends on attempts, so a phone that never # answers cannot hold a small facing in the lane for ever. A recycled id # gets a fresh TrackState, so a new object starts with a fresh budget. crop_attempts: int = 0 # Which evidence produced the newest observation: frame or crop. last_source: str = SOURCE_FRAME 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 def _clear_of_edge(box: Sequence[int], frame_shape: Sequence[int], margin_ratio: float) -> bool: """True when a box sits clear of the frame edge. A facing the frame clips is only part of a product, so neither lane may spend a decision on it. Both lanes ask this question, so a candidate the frame path would refuse cannot be asked about through the crop lane instead. """ frame_h, frame_w = frame_shape[:2] margin_x = int(frame_w * margin_ratio) margin_y = int(frame_h * margin_ratio) x1, y1, x2, y2 = box return x1 > margin_x and y1 > margin_y and x2 < frame_w - margin_x and y2 < frame_h - margin_y def _crop_off_product(requested: Sequence[int], current: Sequence[int]) -> bool: """True when a track no longer covers the box a crop was taken from. The measure a settled label already uses: below DRIFT_IOU the track has slid onto a different product, and a crop of the old one says nothing about what is there now. A box that was never measured cannot vouch either, so it counts as off product. """ if len(requested) != 4 or len(current) != 4: return True return _iou(current, requested) < DRIFT_IOU def _validate_crop_limit(value: Any) -> int: """The crop limit a client stated, or ValueError. The limit is a memory bound, so a value the server cannot honour is refused rather than clamped: quietly handing a client that asked for 512 B a 1 KiB floor, or one that asked for 4 GiB a 1 MiB ceiling, leaves the two sides believing different limits. bool is an int in Python and is not a limit. """ if isinstance(value, bool) or not isinstance(value, int): raise ValueError("max_crop_bytes must be an integer") if value <= 0 or value > MAX_CROP_BYTES: raise ValueError("max_crop_bytes must be 1..%d" % MAX_CROP_BYTES) return value class _CropRejected(Exception): """A returned crop that may not be used, carrying the reply to send back.""" def __init__(self, code: str, message: str): super().__init__(message) self.reply = _crop_error(code, message) def _crop_error(code: str, message: str) -> Dict[str, Any]: """The one shape a refused crop answers with, recoverable like any other.""" return {"type": "crop_error", "code": code, "message": message} 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() # One waiter, created on the caller's loop. None until the first async take. self._waiter: Optional[asyncio.Event] = None self._waiter_loop: Optional[asyncio.AbstractEventLoop] = None 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._tags = None # Shelf position of this frame's origin. Register and origin update # share one gate so overlapping workers cannot reorder either. self._shift_gate = threading.Lock() self._frame_origin = [0.0, 0.0] self._shift_map = None self._compliance_tracker = None # Full-resolution crop lane. A request pins the frame and the track # object it was minted for, so a late or borrowed crop cannot relabel a # product the client never photographed. self._crop_requests: Dict[str, Dict[str, Any]] = {} self._crop_counter = 0 self._crops_requested = 0 self._crops_accepted = 0 self._crops_rejected = 0 self._last_compliance: Dict[str, Any] = {"violations": 0, "by_rule": {}, "rows": []} self._label_memory: "collections.deque[Dict[str, Any]]" = collections.deque( maxlen=LABEL_MEMORY_MAX ) 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() self._wake_waiter_locked() return True def _pop_pending_locked(self) -> Optional[Tuple[int, bytes]]: pending = self._pending self._pending = None self._frame_ready.clear() return pending def _wake_waiter_locked(self) -> None: """Set the async event from this thread. Caller holds self._lock.""" waiter = self._waiter loop = self._waiter_loop if waiter is None or loop is None: return try: running = asyncio.get_running_loop() except RuntimeError: running = None if running is loop: waiter.set() return try: loop.call_soon_threadsafe(waiter.set) except RuntimeError: # Loop already closed. The sync event still wakes take_pending. pass 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: return self._pop_pending_locked() async def take_pending_async(self) -> Optional[Tuple[int, bytes]]: """Awaitable pop. None means the session closed with nothing left. Bind the waiter to the running loop on first use. A frame that arrives before that bind is still visible: the event starts set when one is queued. """ loop = asyncio.get_running_loop() with self._lock: if self._waiter is not None and self._waiter_loop is not loop: raise RuntimeError("take_pending_async is bound to another event loop") if self._waiter is None: self._waiter = asyncio.Event() self._waiter_loop = loop if self._pending is not None or self._closed: self._waiter.set() waiter = self._waiter while True: await waiter.wait() with self._lock: if self._pending is not None: pending = self._pop_pending_locked() if self._waiter is waiter and self._pending is None and not self._closed: self._waiter.clear() return pending if self._closed: return None if self._waiter is waiter: self._waiter.clear() def close(self) -> None: with self._lock: self._closed = True self._pending = None self._frame_ready.set() self._wake_waiter_locked() @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, tags: 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 tags is not None: self.settings.tags = bool(tags) if method_changed: self._generation += 1 for track in self._tracks.values(): track.state = STATE_DETECTED # Nothing has decided this track under the new method yet, # so it is unconfirmed again rather than a settled track # that happens to carry no label. track.confirmed = False track.identity = None track.candidates = [] track.score_sums = {} track.observations = [] track.attempts = 0 track.in_flight = 0 track.last_attempt_seq = 0 # A crop asked about under the old method is evidence measured # under settings the session no longer holds, so the new method # gets a budget of its own. track.crop_attempts = 0 track.last_attempt_area = 0 # Outstanding crops were minted against the old method and the # old generation; answering one now would be evidence measured # under settings the session no longer holds. self._crop_requests.clear() return self.config() def config(self) -> Dict[str, Any]: return { "method": self.settings.method, "conf": self.settings.conf, "empty_slots": self.settings.empty_slots, "tags": self.settings.tags, "tags_available": self._tags is not None, "ocr_engine": getattr(self._tags.engine, "name", "none") if self._tags else "none", } def reset(self) -> None: with self._shift_gate: 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._label_memory.clear() self._crop_requests.clear() self._slots = [] self._latest_image = None self._latest_frame_id = None self._frame_origin = [0.0, 0.0] self._shift_map = None # ponytail: do not touch _waiter here. asyncio.Event is loop-affine; # take_pending_async clears a stale wake on its own loop. # ------------------------------------------------------------------ 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 shelf_offset = self._note_frame_shift(image) 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} # Ids whose box this frame actually measured. A track the tracker # is carrying without a detection keeps its last measured box, and # that box must not be handed to the phone as if it described the # frame the request points at. measured: Set[int] = set() active: Dict[int, TrackState] = {} for track in tracks: state = self._tracks.get(track.track_id) raw_box = None confidence = track.score measured_now = track.det_index is not None and track.det_index in by_id if measured_now: detection = by_id[track.det_index] raw_box = list(detection["box"]) confidence = float(detection["confidence"]) if raw_box is None: # Low-score second pass, or a Kalman box with no detection. # Keep the last measured crop. Do not invent one from the prediction. raw_box = list(state.raw_box) if state is not None and state.raw_box else [] confidence = float(state.confidence) if state is not None else confidence low_score = bool(getattr(track, "low_score", False)) or ( track.det_index is None and raw_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) if raw_box and not low_score: state.raw_box = raw_box state.confidence = confidence if measured_now: # A detection in this frame's list, at a score the tracker # trusts: the box describes the frame the request names. measured.add(track.track_id) elif not state.raw_box and raw_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) self._inherit_labels(seq) # Price tags: plan a pass (cheap) under the lock, run it on the tag # worker, and attach whatever the cache already knows this frame. tag_bands = [] if self._tags is not None and self.settings.tags: tag_boxes = [ {"id": state.track_id, "box": state.box} for state in self._tracks.values() if state.box ] tag_bands = self._tags.plan(tag_boxes, image.shape, shelf_offset=shelf_offset) # 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) crop_requests = self._plan_crop_requests(image.shape, seq, int(frame_id), measured) summary = self._summary_locked() stats = self._stats_locked() track_payload = [self._track_payload(state) for state in self._tracks.values()] if crop_requests: for row in track_payload: request = crop_requests.get(row["track_id"]) if request is not None: row["crop_request"] = request tag_summary = self._apply_tags(seq, track_payload, tag_bands, shelf_offset, image.shape) if tag_summary is not None: summary["price_tags"] = tag_summary["tags"] summary["compliance"] = tag_summary["compliance"] self._submit_classification(submissions, method, generation) self._run_tag_passes(image, tag_bands) 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 _inherit_labels(self, seq: int) -> None: """Give a fresh track the label of a recently recognized product it covers. Only a measured label, still overlapping its own box, may be copied. An inherited label is not a measurement, so it cannot spread to a neighbour. Blur, glare, and a too-far frame skip inheritance: the overlap is not evidence when the frame itself failed the quality gate. """ if not self._label_memory: return if (self._last_quality or {}).get("hint"): return for track in self._tracks.values(): if track.state != STATE_DETECTED or track.confirmed: continue for entry in reversed(self._label_memory): if seq - entry["seq"] > LABEL_MEMORY_FRAMES: break if entry.get("inherited"): continue if _iou(entry["box"], entry.get("label_box") or entry["box"]) < DRIFT_IOU: continue if _iou(track.box, entry["box"]) < LABEL_MEMORY_IOU: continue track.state = STATE_RECOGNIZED track.confirmed = True track.label_box = list(track.box) track.last_source = SOURCE_INHERITED track.identity = { "sku_id": entry["sku_id"], "confidence": entry["confidence"], "score": entry["score"], "attempts": 0, "inherited": True, } track.relabels = self._relabel_count.get(track.track_id, 0) + 1 self._relabel_count[track.track_id] = track.relabels break def _remember_label(self, track: TrackState, box: List[int], seq: int) -> None: """Must be called with the session lock held.""" if track.state != STATE_RECOGNIZED or not track.identity: return self._label_memory.append( { "box": list(box), "label_box": list(track.label_box or box), "sku_id": track.identity.get("sku_id"), "confidence": track.identity.get("confidence"), "score": track.identity.get("score"), "seq": seq, "inherited": bool(track.identity.get("inherited")), } ) 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 # A crop asked about the product this box has now left says nothing # about the neighbour in its place, so the answer goes with the label # and the new product is asked about on a budget of its own. for request_id in [rid for rid, rec in self._crop_requests.items() if rec["track_id"] == track.track_id]: self._crops_rejected += 1 self._release_crop_slot(self._crop_requests.pop(request_id)) 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 # The label went with the product, so the track is undecided again: # a settled track that carries no label can never be asked about. track.confirmed = False track.identity = None track.candidates = [] track.score_sums = {} track.observations = [] track.attempts = 0 track.in_flight = 0 track.last_attempt_seq = 0 track.last_attempt_area = 0 track.label_box = [] track.crop_attempts = 0 # ------------------------------------------------------------------ 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_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] = [] poor_frame = bool((self._last_quality or {}).get("hint")) for track in self._tracks.values(): if track.state != STATE_DETECTED or track.in_flight > 0: continue if poor_frame: continue if track.hits < self.settings.min_hits: continue if len(track.raw_box) != 4: continue x1, y1, x2, y2 = track.raw_box if min(x2 - x1, y2 - y1) < self.settings.min_crop_px: continue if not _clear_of_edge(track.raw_box, image.shape, self.settings.edge_margin_ratio): continue if track.attempts >= self.settings.max_attempts: continue if track.attempts > 0: if (seq - track.last_attempt_seq) < self.settings.retry_gap: if track.area() < self.settings.area_growth * max(1, track.last_attempt_area): continue retry_margin_x = int(frame_w * self.settings.retry_edge_ratio) retry_margin_y = int(frame_h * self.settings.retry_edge_ratio) if (x1 <= retry_margin_x or y1 <= retry_margin_y or x2 >= frame_w - retry_margin_x or y2 >= frame_h - retry_margin_y): continue eligible.append(track) # Newest first: while a camera pans, the products still worth labelling # are the ones that just entered the frame. Area is only the tie-break. eligible.sort(key=lambda track: (-track.first_seq, -track.area())) 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_lanes, 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 ) ) # ------------------------------------------------------------------ crop lane def capabilities(self) -> Dict[str, Any]: """What this client is allowed to ask the phone for. A client that never sends `capabilities` keeps the protocol-v1 defaults: no crop lane, so no result ever carries a crop request. """ return { "crop": bool(self.settings.crops), "max_crop_bytes": int(self.settings.max_crop_bytes), } def enable_crops(self, max_crop_bytes: Optional[int] = None) -> Dict[str, Any]: """Turn the lane on for a client that advertised it. The client states the largest crop it will produce; a limit that is not one is refused and the lane stays where it was, so a bad statement enables nothing.""" stated = None if max_crop_bytes is None else _validate_crop_limit(max_crop_bytes) with self._lock: self.settings.crops = True if stated is not None: self.settings.max_crop_bytes = stated return self.capabilities() def _plan_crop_requests( self, frame_shape: Tuple[int, int], seq: int, frame_id: int, measured: Set[int] ) -> Dict[int, Dict[str, Any]]: """Ask the phone for a fuller crop of tracks the frame crop could not use. The box is in the pixel space of `frame_id`, the frame the client just sent, so the phone can answer from the buffer behind that exact frame rather than from whatever is on screen now. That is also why the box has to be one this frame measured and has to sit clear of the edge: the phone crops by these numbers, so a carried-over or clipped box buys a photograph of the wrong thing. At most `max_attempts` requests per object, `max_crop_inflight` outstanding, and only for a client that advertised the lane. """ self._expire_crop_requests(seq) if not self.settings.crops: return {} room = self.settings.max_crop_inflight - len(self._crop_requests) if room <= 0: return {} if (self._last_quality or {}).get("hint") in CROP_BLIND_QUALITY: # A blurred or glared frame photographs the same bad scene at the # camera's own resolution, so there is nothing to ask the phone for. return {} planned: Dict[int, Dict[str, Any]] = {} for track in self._tracks.values(): if len(planned) >= room: break if track.crop_attempts >= self.settings.max_attempts or track.in_flight > 0: continue if track.state != STATE_DETECTED or track.confirmed: continue if track.hits < self.settings.min_hits or len(track.raw_box) != 4: continue if track.track_id not in measured: # The box is a carry-over: either no detection matched this # frame, or the one that did was too low a score to move it. # Either way it points at where the product used to be. continue if not _clear_of_edge(track.raw_box, frame_shape, self.settings.edge_margin_ratio): continue x1, y1, x2, y2 = track.raw_box # The frame crop is too small to say anything about this product. At # the camera's own resolution it may still be worth a decision, and # that is the only thing a crop can add here. if min(x2 - x1, y2 - y1) >= self.settings.crop_below_px: continue self._crop_counter += 1 request_id = "c%d" % self._crop_counter # The detection box this crop is a photograph of. The answer is only # about that product, so the box travels with the request. requested_box = list(track.raw_box) self._crop_requests[request_id] = { "request_id": request_id, "track_id": track.track_id, "frame_id": int(frame_id), "box": requested_box, "object_generation": track.object_generation, "generation": self._generation, "seq": seq, "method": self.settings.method, } track.crop_attempts += 1 # Hold the track while the phone answers, so the frame path does not # spend a second attempt on the same box at the same time. track.in_flight += 1 self._crops_requested += 1 planned[track.track_id] = { "request_id": request_id, "track_id": track.track_id, "frame_id": int(frame_id), "box": list(requested_box), } return planned def _expire_crop_requests(self, seq: int) -> None: """Drop requests the phone did not answer in time. A request holds the track busy while it is out, so an unanswered one has to give that slot back or the track could never be classified at all. """ stale = [ request_id for request_id, record in self._crop_requests.items() if seq - record["seq"] > self.settings.crop_ttl_frames ] for request_id in stale: record = self._crop_requests.pop(request_id) self._crops_rejected += 1 self._release_crop_slot(record) def _release_crop_slot(self, record: Dict[str, Any]) -> None: """Give back the busy slot a crop request took. Lock must be held.""" track = self._tracks.get(record["track_id"]) if track is not None and track.object_generation == record["object_generation"]: track.in_flight = max(0, track.in_flight - 1) def submit_crop( self, request_id: Any, track_id: Any, frame_id: Any, payload: Optional[bytes], ) -> Dict[str, Any]: """Score a crop the client returned for a track we asked about. The crop is evidence about one track object, in one frame, decided under one configuration. Anything that breaks that binding is refused before a single pixel reaches the classifier, and the frame crop stays the only evidence for the track. """ try: record, box, payload = self._prepare_crop(request_id, track_id, frame_id, payload) except _CropRejected as refused: with self._lock: self._crops_rejected += 1 return refused.reply try: crop = self.engine.decode_image(payload) except Exception: # The busy slot is the request's to give back, whoever failed on it. with self._lock: self._release_crop_slot(record) self._crops_rejected += 1 return _crop_error("CROP_UNDECODABLE", "crop is not a readable image") submitted_at = time.perf_counter() track_id_out = record["track_id"] generation = record["generation"] method = record["method"] object_generation = record["object_generation"] with self._lock: # Decoding is the slow part and the session keeps running through it, # so the binding checked before the decode is re-checked here. The # check and the submit share one critical section: an ack has to mean # this crop was measured against the track object it answers. stale = self._crop_stale_reason_locked(record) if stale is not None: self._release_crop_slot(record) self._crops_rejected += 1 return _crop_error("CROP_STALE", stale) try: future = self._executor.submit(self.engine.score_crop_lanes, crop, method) except RuntimeError: # executor already shut down self._release_crop_slot(record) self._crops_rejected += 1 return _crop_error("CROP_UNAVAILABLE", "classifier is shutting down") self._crops_accepted += 1 future.add_done_callback( lambda fut, tid=track_id_out, gen=generation, m=method, started=submitted_at, obj=object_generation, box=box: self._on_classified( fut, tid, gen, m, started, obj, box, SOURCE_CROP ) ) return { "type": "crop_ack", "request_id": record["request_id"], "track_id": track_id_out, "frame_id": record["frame_id"], } def _prepare_crop( self, request_id: Any, track_id: Any, frame_id: Any, payload: Optional[bytes] ) -> Tuple[Dict[str, Any], List[int], bytes]: """Check a returned crop against the request it claims to answer. Raises [_CropRejected] with the reply to send. The request is consumed either way: it is single use, so a rejected crop cannot be retried against a track that has since moved. """ with self._lock: record = self._crop_requests.pop(str(request_id), None) if request_id else None if record is None: raise _CropRejected("CROP_UNKNOWN_REQUEST", "no crop is waiting for that request id") try: answered_track = int(track_id) answered_frame = int(frame_id) except (TypeError, ValueError): self._release_crop_slot(record) raise _CropRejected("CROP_BAD_MESSAGE", "track_id and frame_id are required") if answered_track != record["track_id"] or answered_frame != record["frame_id"]: # A crop of some other track, or of another frame, says nothing # about this one however well it matches. self._release_crop_slot(record) raise _CropRejected("CROP_MISMATCH", "crop does not match the request it answers") if self._seq - record["seq"] > self.settings.crop_ttl_frames: self._release_crop_slot(record) raise _CropRejected("CROP_STALE", "crop answered too late to describe the track") if record["generation"] != self._generation: self._release_crop_slot(record) raise _CropRejected("CROP_STALE", "the session changed after the request") track = self._tracks.get(record["track_id"]) if track is None or track.object_generation != record["object_generation"]: self._release_crop_slot(record) raise _CropRejected("CROP_STALE", "the track id was recycled after the request") # The sender keeps streaming while the requested crop is encoded. # A newer frame is normal; object generation and the bounded frame # TTL, not exact latest-sequence equality, guard against reuse. if track.last_seq < record["seq"]: self._release_crop_slot(record) raise _CropRejected("CROP_STALE", "the track is no longer in the requested frame") if _crop_off_product(record["box"], track.raw_box): # The crop is of the product the phone was asked to photograph. # If the track has since slid onto a neighbour the answer would # be a label for a product nobody cropped. self._release_crop_slot(record) raise _CropRejected("CROP_STALE", "the track moved off the product that was cropped") limit = self.settings.max_crop_bytes if not payload or len(payload) > limit: self._release_crop_slot(record) raise _CropRejected("CROP_TOO_LARGE", "crop is empty or over %d bytes" % limit) # The label anchors to the box the crop was taken from, not to where # the track is by the time the answer lands. return record, list(record["box"]), payload def _crop_stale_reason_locked(self, record: Dict[str, Any]) -> Optional[str]: """Why a decoded crop can no longer speak for its track, or None. Run after the decode, under the lock that submits the classification. A reset or a method change, a recycled track id, a pan that carried the track off the photographed product, or another attempt that already settled the track can all land while the JPEG is being decoded, and an ack is a promise that this crop decided that track. Lock must be held. """ if record["generation"] != self._generation: return "the session changed while the crop was decoded" track = self._tracks.get(record["track_id"]) if track is None or track.object_generation != record["object_generation"]: return "the track id was recycled while the crop was decoded" if _crop_off_product(record["box"], track.raw_box): return "the track moved off the product while the crop was decoded" if track.state != STATE_DETECTED or track.confirmed: return "the track is no longer waiting to be decided" return None def _on_classified( self, future: Future, track_id: int, generation: int, method: str, submitted_at: float, object_generation: int, submitted_box: List[int], source: str = SOURCE_FRAME, ) -> None: try: scores, text_rank = future.result() except Exception: scores, text_rank = {}, [] 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) if track.state != STATE_DETECTED or track.confirmed: # Other evidence settled this track while the answer sat in the # classifier - a label inherited from a labelled neighbour, say. # A late answer must not repaint a label the client is already # showing, so the track keeps the one it settled on. return if source == SOURCE_CROP and _crop_off_product(submitted_box, track.raw_box): # The answer was measured on the box the track has since left. # Dropping it leaves the track to be decided from a later frame # crop instead of putting the old product's label on whatever # moved into its place. return track.attempts += 1 # A crop is an attempt against the same budget as a frame crop, so # the lane cannot be used to re-roll the dice on one product. track.last_source = source observation = { str(sku_id): float(score) for sku_id, score in (scores or {}).items() } track.observations.append(observation) track.observations = track.observations[-3:] for sku_id, score in observation.items(): track.score_sums[sku_id] = track.score_sums.get(sku_id, 0.0) + float(score) if observation: track.score_n += 1 self._classify_times.append(finished_at - submitted_at) if not any(track.observations) and not text_rank: 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 aggregate_votes, decide_identity from identity_lanes import DecisionPolicy mean_scores = aggregate_votes(track.observations) scale = getattr(self.engine, "identity_scale", "inlier") if scale == "cosine": # score_crop only returns scores the gallery already accepted, so # the aggregate must not be re-gated at COSINE_ACCEPT: that line is # above the head's own 0.40 and refused matches the gallery had # accepted. It still has to clear the gallery's own line, because # aggregate_votes divides by the whole window - one rejected crop # beside one accepted 0.45 would otherwise average to 0.225 and be # labelled at a score the gallery calls unknown. gallery = self.engine.identity_gallery() line = 0.0 if gallery is not None: line = gallery.head_threshold if gallery.head_loaded() else gallery.threshold decision = decide_identity( mean_scores, method=method, scale="cosine", min_score=line, min_confidence=line, ) elif scale == "local": # The local lane's order and value are different quantities: the # order comes from its rank key, the value is the margin over the # runner-up. `score_crop` returns that order, so it is used as-is; # re-sorting by value would name a different SKU than the lane # chose. The newest crop is the evidence, not the mean: averaging # in an empty observation halves a good confidence below the # accept line, which is the dilution the cosine branch guards too. decision = DecisionPolicy().decide(list(scores.items()), text_rank) else: decision = decide_identity(mean_scores, method=method, scale=scale) 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, } self._remember_label(track, submitted_box, self._seq) 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, # Which evidence produced the newest observation. A client that only # reads this can tell a label measured on the frame it sent from one # measured on a crop it sent back. "source": track.last_source, } 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, "inherited": bool(track.identity and track.identity.get("inherited")), "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 { "crops": { "requested": self._crops_requested, "accepted": self._crops_accepted, "rejected": self._crops_rejected, }, "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 len(state.raw_box) == 4] 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 ) ) # A black frame has Laplacian 0. That is not blur; a real shelf is not black. blur = sharpness < self._blur_threshold() and float(gray.mean()) > 1.0 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() # ------------------------------------------------------------------ price tags def enable_tags(self, settings: Optional[Any] = None) -> bool: """Attach a tag pass when an OCR engine is available. Returns whether it runs.""" ocr_engine = getattr(self.engine, "ocr_engine", None) if ocr_engine is None or not getattr(ocr_engine, "available", lambda: False)(): return False from tags import TagService, TagSettings with self._lock: self._tags = TagService(ocr_engine, settings=settings or TagSettings()) return True def _note_frame_shift(self, image: Any) -> Tuple[float, float]: """Shelf position of this frame's origin. Phase correlation only. Register and origin update share one per-session gate, so overlapping workers cannot reorder the previous-frame baseline or publish a later origin first. Correlation stays outside the session lock and outside OCR. A failed registration keeps the last origin rather than inventing a jump. The content shift moves the origin the other way, same sign as ShelfMap. """ with self._shift_gate: if self._shift_map is None: from shelf_map import ShelfMap self._shift_map = ShelfMap() shift = self._shift_map.register(image) if shift is not None: self._frame_origin[0] -= float(shift[0]) self._frame_origin[1] -= float(shift[1]) return (self._frame_origin[0], self._frame_origin[1]) def _run_tag_passes(self, image: Any, bands: Sequence[Any]) -> None: """Submit tag passes to the engine's tag worker (never the frame loop).""" service = self._tags if service is None or not bands: return for band in bands: try: future = self.engine.tag_executor.submit(service.run_pass, image, band) except RuntimeError: # executor shut down return future.add_done_callback(self._on_tag_pass_done) def _on_tag_pass_done(self, future: Any) -> None: try: future.result() except Exception as error: # A failed OCR pass must not disturb the stream, but it must not be # invisible either: counted and reported through stats(), so an # operator can tell a broken OCR from a shelf with no tags. self._tags.note_pass_failure(error) LOGGER.warning("tag pass failed: %s", error) def _apply_tags( self, seq: int, track_payloads: List[Dict[str, Any]], bands: Sequence[Any], shelf_offset: Sequence[float], image_shape: Sequence[int], ) -> Optional[Dict[str, Any]]: """Attach tag + reconciliation to each track and compute compliance. Must be called with the session lock held (it reads the tag cache under its own lock, which is never taken the other way round). The origin and shape are this frame's, not whatever `plan` last stored. """ service = self._tags if service is None or not self.settings.tags: return None from reconcile import ComplianceTracker, evaluate_compliance, reconcile, summarize_findings tag_by_track, all_tags, counts = service.attach( track_payloads, shelf_offset=shelf_offset, image_shape=image_shape ) reconciliations: Dict[int, Dict[str, Any]] = {} for payload in track_payloads: track_id = int(payload["track_id"]) tag = tag_by_track.get(track_id) identity = payload.get("identity") product = self._catalog_by_sku.get((identity or {}).get("sku_id") or "") decision = reconcile(identity, tag, product) if tag is not None: payload["price_tag"] = tag if decision is not None: payload["reconciliation"] = decision reconciliations[track_id] = decision if self._compliance_tracker is None: self._compliance_tracker = ComplianceTracker() findings = evaluate_compliance( track_payloads, tag_by_track, all_tags, bands, reconciliations ) stable = self._compliance_tracker.update(findings) summary = summarize_findings(stable) rows = [ { "row_index": finding.row_index, "rule": finding.rule, "box": finding.box, "missing": finding.count, "skus": finding.skus, } for finding in stable if finding.severity == "violation" ] self._last_compliance = {"violations": summary["violations"], "by_rule": summary["by_rule"], "rows": rows, "hints": summary["hints"]} return {"tags": counts, "compliance": self._last_compliance} 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), "tags": self._tags.stats() if self._tags else None, "compliance": dict(self._last_compliance), "created_at": self._created_at, }