From aa25f99699cb5f431474aaca54d38f1d220e033f Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Mon, 28 Sep 2026 17:22:52 +0900 Subject: [PATCH] fix: bound and gate the crop evidence lane The crop lane asked the phone for a full-resolution crop of a facing the frame could not decide. Four defects made that unsafe or useless: * A capability was half-applied. `crop: true` without a usable `max_crop_bytes` enabled the lane on the client's say-so, leaving the two sides believing different limits. A capability is now believed whole or refused with BAD_MESSAGE, and the limit must be an integer in 1..1048576 (bool is an int in Python and is not a limit). * A small JPEG could claim enormous dimensions. The header is now read without decoding and anything over 4 megapixels is refused before the decode allocates it. * A request was single-use, so an expired one stranded a facing only the crop lane could decide. The lane now spends the same max_attempts budget the frame path spends, refilled on method change and on settled-label drift, and a phone that never answers still cannot loop. * A candidate could be asked about from a box the frame path would refuse: clipped by the frame edge, carried over from an earlier frame, or measured at a score the tracker does not trust. Both lanes now share one edge gate, and only a track whose box this frame actually measured is asked about. Drift and method change also cleared `identity` but left `confirmed = True`, which permanently blocked both the crop lane and label inheritance for that track; both now clear it, without which the budget reset was dead code. Blur and glare are refused as whole-capture faults, while `too_far` is exempt on purpose: a distant shelf is the one frame a native-resolution crop exists to decide. --- recognization/code/app.py | 53 ++++++- recognization/code/live.py | 165 +++++++++++++++++---- recognization/code/test_api.py | 57 ++++++- recognization/code/test_live.py | 253 +++++++++++++++++++++++++++++++- recognization/docs/API.md | 22 ++- 5 files changed, 504 insertions(+), 46 deletions(-) diff --git a/recognization/code/app.py b/recognization/code/app.py index deadce0..2735fb7 100644 --- a/recognization/code/app.py +++ b/recognization/code/app.py @@ -94,6 +94,37 @@ MAX_FRAME_BYTES = 4 * 1024 * 1024 DEFAULT_MAX_SESSIONS = 4 WS_IDLE_TIMEOUT_S = 120.0 CATALOG_VERSION_FALLBACK = "unknown" +MAX_CROP_PIXELS = 4_000_000 + + +def _jpeg_dimensions(data: bytes) -> Optional[Tuple[int, int]]: + """Read JPEG SOF dimensions without allocating a decoded image.""" + if not data.startswith(b"\xff\xd8"): + return None + offset = 2 + while offset + 4 <= len(data): + if data[offset] != 0xff: + return None + while offset < len(data) and data[offset] == 0xff: + offset += 1 + if offset >= len(data): + return None + marker = data[offset] + offset += 1 + if marker in (0xd8, 0x01) or 0xd0 <= marker <= 0xd7: + continue + if marker in (0xd9, 0xda) or offset + 2 > len(data): + return None + length = int.from_bytes(data[offset:offset + 2], "big") + if length < 2 or offset + length > len(data): + return None + if marker in (0xc0, 0xc1, 0xc2, 0xc3, 0xc5, 0xc6, 0xc7, 0xc9, 0xca, 0xcb, 0xcd, 0xce, 0xcf): + if length < 7: + return None + return (int.from_bytes(data[offset + 5:offset + 7], "big"), + int.from_bytes(data[offset + 3:offset + 5], "big")) + offset += length + return None # --------------------------------------------------------------------------- @@ -821,13 +852,20 @@ def create_app( """A client that can answer with a full-resolution crop says so here. Until this arrives the session keeps the protocol-v1 behaviour: no - result carries a crop request, so a v1 client is never asked. + result carries a crop request, so a v1 client is never asked. A + capability is believed whole or not at all: a message that does not + name the lane with a limit the server can honour is refused and the + lane stays off, rather than half-applied on the client's say-so. """ - try: - capabilities = session.enable_crops(message.get("max_crop_bytes")) - except (TypeError, ValueError): + if message.get("crop") is not True or "max_crop_bytes" not in message: await send({"type": "error", "code": "BAD_MESSAGE", - "message": "bad max_crop_bytes"}) + "message": "crop must be true and state max_crop_bytes"}) + return + try: + capabilities = session.enable_crops(message["max_crop_bytes"]) + except ValueError: + await send({"type": "error", "code": "BAD_MESSAGE", + "message": "max_crop_bytes must be 1..%d" % MAX_CROP_BYTES}) return await send({"type": "capabilities_ack", "capabilities": capabilities}) @@ -855,6 +893,11 @@ def create_app( await send({"type": "crop_error", "request_id": request_id, "code": "CROP_BAD_MESSAGE", "message": "jpeg is not valid base64"}) return + dimensions = _jpeg_dimensions(crop_bytes) + if dimensions is not None and dimensions[0] * dimensions[1] > MAX_CROP_PIXELS: + await send({"type": "crop_error", "request_id": request_id, + "code": "CROP_TOO_LARGE", "message": "crop dimensions exceed the limit"}) + return reply = await loop.run_in_executor( None, session.submit_crop, request_id, message.get("track_id"), message.get("frame_id"), crop_bytes, diff --git a/recognization/code/live.py b/recognization/code/live.py index 29ead37..f8d40fc 100644 --- a/recognization/code/live.py +++ b/recognization/code/live.py @@ -25,7 +25,7 @@ from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass, field LOGGER = logging.getLogger("dorayaki.live") -from typing import Any, Dict, List, Optional, Sequence, Tuple +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: @@ -66,6 +66,11 @@ 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 @@ -135,9 +140,11 @@ class TrackState: # 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 - # A crop is asked for at most once per object. A recycled id gets a fresh - # TrackState, so the new object may be asked again. - crop_requested: bool = False + # 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 @@ -162,6 +169,21 @@ def _iou(a: Sequence[int], b: Sequence[int]) -> float: 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. @@ -175,6 +197,21 @@ def _crop_off_product(requested: Sequence[int], current: Sequence[int]) -> bool: 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.""" @@ -376,6 +413,10 @@ class LiveSession: 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 = {} @@ -383,6 +424,10 @@ class LiveSession: 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 @@ -452,12 +497,18 @@ class LiveSession: 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 - if track.det_index is not None and track.det_index in by_id: + 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"]) @@ -513,6 +564,10 @@ class LiveSession: 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 @@ -556,7 +611,7 @@ class LiveSession: # 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(seq, int(frame_id)) + crop_requests = self._plan_crop_requests(image.shape, seq, int(frame_id), measured) summary = self._summary_locked() stats = self._stats_locked() @@ -669,9 +724,19 @@ class LiveSession: 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 = {} @@ -681,6 +746,7 @@ class LiveSession: 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]]: @@ -690,7 +756,6 @@ class LiveSession: 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) @@ -711,7 +776,7 @@ class LiveSession: 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: + if not _clear_of_edge(track.raw_box, image.shape, self.settings.edge_margin_ratio): continue if track.attempts >= self.settings.max_attempts: continue @@ -775,24 +840,27 @@ class LiveSession: 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; the server keeps the lower of the two.""" + 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 max_crop_bytes is not None: - try: - stated = int(max_crop_bytes) - except (TypeError, ValueError): - stated = 0 - self.settings.max_crop_bytes = max(1024, min(stated, MAX_CROP_BYTES)) + if stated is not None: + self.settings.max_crop_bytes = stated return self.capabilities() - def _plan_crop_requests(self, seq: int, frame_id: int) -> Dict[int, Dict[str, Any]]: + 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. One request per track object, - at most `max_crop_inflight` outstanding, and only for a client that + 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) @@ -801,16 +869,27 @@ class LiveSession: 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_requested or track.in_flight > 0: + 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 @@ -832,7 +911,7 @@ class LiveSession: "seq": seq, "method": self.settings.method, } - track.crop_requested = True + 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 @@ -899,14 +978,22 @@ class LiveSession: generation = record["generation"] method = record["method"] object_generation = record["object_generation"] - try: - future = self._executor.submit(self.engine.score_crop, crop, method) - except RuntimeError: # executor already shut down - with self._lock: + 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_UNAVAILABLE", "classifier is shutting down") - with self._lock: + return _crop_error("CROP_STALE", stale) + try: + future = self._executor.submit(self.engine.score_crop, 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, @@ -975,6 +1062,26 @@ class LiveSession: # 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, @@ -1000,6 +1107,12 @@ class LiveSession: 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 diff --git a/recognization/code/test_api.py b/recognization/code/test_api.py index 99dc93d..17cbd0e 100644 --- a/recognization/code/test_api.py +++ b/recognization/code/test_api.py @@ -366,8 +366,35 @@ class WebSocketProtocolTests(unittest.TestCase): self.assertEqual(ack["type"], "capabilities_ack") self.assertEqual(ack["capabilities"], {"crop": True, "max_crop_bytes": 65536}) + def test_a_capability_the_server_cannot_honour_is_refused(self): + """A half-stated or malformed capability turns nothing on.""" + for message in ( + {"type": "capabilities", "max_crop_bytes": 65536}, + {"type": "capabilities", "crop": False, "max_crop_bytes": 65536}, + {"type": "capabilities", "crop": "true", "max_crop_bytes": 65536}, + {"type": "capabilities", "crop": True}, + {"type": "capabilities", "crop": True, "max_crop_bytes": 0}, + {"type": "capabilities", "crop": True, "max_crop_bytes": -4096}, + {"type": "capabilities", "crop": True, "max_crop_bytes": True}, + {"type": "capabilities", "crop": True, "max_crop_bytes": "65536"}, + {"type": "capabilities", "crop": True, "max_crop_bytes": 64 * 1024 * 1024}, + ): + with self.subTest(capability=message): + with self.crop_app().websocket_connect("/v1/stream") as socket: + socket.receive_json() # hello + socket.send_json(message) + self.assertEqual(socket.receive_json()["code"], "BAD_MESSAGE") + # A refused capability leaves this session on protocol v1: the + # facing it cannot report is never asked about. + for frame_id in (1, 2, 3): + socket.send_bytes(frame_payload(frame_id)) + result = socket.receive_json() + self.assertEqual(result["stats"]["crops"]["requested"], 0) + for track in result["tracks"]: + self.assertNotIn("crop_request", track) + def test_a_v1_client_is_never_asked_for_a_crop(self): - with self.client.websocket_connect("/v1/stream") as socket: + with self.crop_app().websocket_connect("/v1/stream") as socket: socket.receive_json() # hello for frame_id in (1, 2, 3): socket.send_bytes(frame_payload(frame_id)) @@ -379,7 +406,7 @@ class WebSocketProtocolTests(unittest.TestCase): def test_a_crop_round_trip_labels_the_track(self): with self.crop_app().websocket_connect("/v1/stream") as socket: socket.receive_json() # hello - socket.send_json({"type": "capabilities", "crop": True}) + socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072}) self.assertEqual(socket.receive_json()["type"], "capabilities_ack") socket.send_bytes(frame_payload(1)) socket.receive_json() @@ -406,7 +433,7 @@ class WebSocketProtocolTests(unittest.TestCase): def test_a_crop_that_does_not_match_its_request_is_refused(self): with self.crop_app().websocket_connect("/v1/stream") as socket: socket.receive_json() # hello - socket.send_json({"type": "capabilities", "crop": True}) + socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072}) socket.receive_json() socket.send_bytes(frame_payload(1)) socket.receive_json() @@ -429,7 +456,7 @@ class WebSocketProtocolTests(unittest.TestCase): def test_a_crop_that_is_not_an_image_is_refused(self): with self.crop_app().websocket_connect("/v1/stream") as socket: socket.receive_json() # hello - socket.send_json({"type": "capabilities", "crop": True}) + socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072}) socket.receive_json() socket.send_bytes(frame_payload(1)) socket.receive_json() @@ -445,6 +472,28 @@ class WebSocketProtocolTests(unittest.TestCase): }) self.assertEqual(socket.receive_json()["code"], "CROP_UNDECODABLE") + def test_a_small_jpeg_claiming_huge_pixels_is_refused_before_decode(self): + with self.crop_app().websocket_connect("/v1/stream") as socket: + socket.receive_json() + socket.send_json({"type": "capabilities", "crop": True, "max_crop_bytes": 131072}) + socket.receive_json() + socket.send_bytes(frame_payload(1)) + socket.receive_json() + socket.send_bytes(frame_payload(2)) + request = socket.receive_json()["tracks"][0]["crop_request"] + jpeg = bytearray(jpeg_bytes(80, 120)) + sof = jpeg.index(b"\xff\xc0") + jpeg[sof + 5:sof + 9] = (16384).to_bytes(2, "big") * 2 + socket.send_json({ + "type": "crop", "request_id": request["request_id"], + "track_id": request["track_id"], "frame_id": request["frame_id"], + "jpeg": base64.b64encode(jpeg).decode("ascii"), + }) + refused = socket.receive_json() + self.assertEqual(refused["code"], "CROP_TOO_LARGE") + socket.send_json({"type": "ping", "t": 1}) + self.assertEqual(socket.receive_json(), {"type": "pong", "t": 1}) + def test_a_crop_without_an_image_is_refused(self): with self.client.websocket_connect("/v1/stream") as socket: socket.receive_json() # hello diff --git a/recognization/code/test_live.py b/recognization/code/test_live.py index 709d869..f43c883 100644 --- a/recognization/code/test_live.py +++ b/recognization/code/test_live.py @@ -25,7 +25,7 @@ import cv2 # noqa: E402 import numpy as np # noqa: E402 from engine import RecognitionEngine # noqa: E402 -from live import LiveSession, LiveSettings # noqa: E402 +from live import MAX_CROP_BYTES, LiveSession, LiveSettings # noqa: E402 class FakeDetector: @@ -109,14 +109,23 @@ def frame_bytes(width=640, height=480): return buffer.tobytes() -def process(session, frame_id=1): - payload = frame_bytes() +def process_payload(session, frame_id, payload): session.submit_frame(frame_id, payload) pending = session.take_pending(timeout=1.0) assert pending is not None, "the session dropped a frame it should have kept" return session.process_frame(*pending) +def process(session, frame_id=1): + return process_payload(session, frame_id, frame_bytes()) + + +def blurred_frame_bytes(width=640, height=480, value=120): + """A flat grey capture: the whole frame is out of focus, not just a facing.""" + ok, buffer = cv2.imencode(".jpg", np.full((height, width, 3), value, dtype="uint8")) + return buffer.tobytes() + + def wait_for(predicate, timeout=2.0): deadline = time.time() + timeout while time.time() < deadline: @@ -556,6 +565,34 @@ class GatedMatcher(FakeMatcher): return super().score_crop(crop, method) +class SecondAnswerWaits(FakeMatcher): + """Answers the first crop at once, holds the second one, and disagrees. + + Two facings that overlap this much are the only way a track can be handed + a settled label while its own answer is still in the classifier, so the + test answers the first crop, lets the track settle, and only then hands + over a second crop whose answer names a different product. + """ + + def __init__(self, sku="sku_a", other="other"): + super().__init__(sku=sku) + self.gate = threading.Event() + self.first_answered = threading.Event() + self.disagreeing = other + self._seen = 0 + + def score_crop(self, crop, method="hybrid"): + with self.lock: + self._seen += 1 + turn = self._seen + self.calls += 1 + if turn == 1: + self.first_answered.set() + return {self.sku: 40.0, "other": 0.1}, {} + self.gate.wait(2.0) + return {self.disagreeing: 40.0, self.sku: 0.1}, {} + + class CropLaneTests(unittest.TestCase): """A full-resolution crop is extra evidence for one track, nothing else.""" @@ -764,6 +801,216 @@ class CropLaneTests(unittest.TestCase): self.assertEqual(session.capabilities()["crop"], True) engine.shutdown() + def test_a_refused_limit_never_turns_the_lane_on(self): + session, engine, _ = make_crop_session() + session.settings.crops = False # a lane no client has advertised yet + for bad in (True, 0, -1, 4096.0, "4096", MAX_CROP_BYTES + 1): + with self.subTest(limit=bad): + with self.assertRaises(ValueError): + session.enable_crops(bad) + self.assertFalse(session.capabilities()["crop"], + "a refused limit must leave the lane off") + self.assertEqual(session.enable_crops(4096)["max_crop_bytes"], 4096) + engine.shutdown() + + def test_a_reset_while_the_crop_decodes_is_refused(self): + """An ack promises this crop decided that track; a reset says it cannot.""" + matcher = FakeMatcher() + session, engine, _ = make_crop_session(matcher=matcher) + process(session, 1) + request = requested_crop(process(session, 2)) + self.assertIsNotNone(request) + + decoding = threading.Event() + resumed = threading.Event() + decode = engine.decode_image + + def gated_decode(payload): + decoding.set() + resumed.wait(2.0) + return decode(payload) + + engine.decode_image = gated_decode + replies = {} + + def answer(): + replies["reply"] = session.submit_crop( + request["request_id"], request["track_id"], request["frame_id"], frame_bytes() + ) + + worker = threading.Thread(target=answer) + worker.start() + self.assertTrue(decoding.wait(2.0), "the crop must be decoding before the reset") + session.reset() + resumed.set() + worker.join(5.0) + self.assertIn("reply", replies, "the answer never came back") + self.assertEqual(replies["reply"]["code"], "CROP_STALE") + self.assertEqual(session._crops_accepted, 0) + self.assertEqual(matcher.calls, 0, "a refused crop never reaches the classifier") + engine.shutdown() + + def test_a_blurred_frame_is_not_asked_about(self): + """More pixels of an out-of-focus scene are not more evidence.""" + session, engine, _ = make_crop_session() + process(session, 1) + result = process_payload(session, 2, blurred_frame_bytes()) + self.assertEqual(result["quality"]["hint"], "blur", + "the frame under test must be the blurry one") + self.assertIsNone(requested_crop(result)) + self.assertEqual(result["stats"]["crops"]["requested"], 0) + engine.shutdown() + + def test_a_distant_shelf_is_still_asked_about(self): + """Too far to read at the frame's own resolution is the case a crop exists for.""" + session, engine, detector = make_crop_session( + settings=LiveSettings(method="hybrid", min_crop_px=60) + ) + detector.boxes[0].update({ + "box": [300, 380, 336, 430], "width": 36, "height": 50, "center": [318, 405], + }) + process(session, 1) + result = process(session, 2) + self.assertEqual(result["quality"]["hint"], "too_far", + "the frame under test must be the distant one") + self.assertIsNotNone(requested_crop(result)) + engine.shutdown() + + def test_a_facing_the_frame_clips_is_not_asked_about(self): + session, engine, detector = make_crop_session() + clipped = {"box": [0, 100, 36, 220], "width": 36, "height": 120, "center": [18, 160]} + detector.boxes[0].update(clipped) + process(session, 1) + result = process(session, 2) + self.assertIsNone(requested_crop(result), "half a facing is not worth the phone's pixels") + self.assertEqual(result["stats"]["crops"]["requested"], 0) + # The same facing, walked clear of the edge, is asked about: only the + # clip kept it out of the lane. + asked = None + for frame_id, x in enumerate([12, 24, 36], start=3): + detector.boxes[0].update({"box": [x, 100, x + 36, 220], "center": [x + 18, 160]}) + result = process(session, frame_id) + asked = asked or requested_crop(result) + if asked is not None: + break + self.assertIsNotNone(asked, "a whole facing is worth the phone's pixels") + engine.shutdown() + + def test_a_box_this_frame_did_not_measure_is_not_asked_about(self): + """The phone crops by the box it is sent, so a stale box crops the wrong thing.""" + session, engine, detector = make_crop_session() + process(session, 1) + detector.boxes[0]["confidence"] = 0.15 # too low for the tracker to trust + result = process(session, 2) + self.assertIsNone(result["quality"]["hint"], "only the detection is at fault here") + self.assertIsNone(requested_crop(result)) + detector.boxes[0]["confidence"] = 0.9 + self.assertIsNotNone(requested_crop(process(session, 3)), + "the same facing is asked about once a frame measures it again") + engine.shutdown() + + def test_a_method_change_lets_the_small_facing_ask_again(self): + """A crop minted under the old method is not an attempt under the new one.""" + session, engine, _ = make_crop_session() + process(session, 1) + first = requested_crop(process(session, 2)) + self.assertIsNotNone(first) + session.configure(method="orb") + second = requested_crop(process(session, 3)) + self.assertIsNotNone(second, "the new method gets a budget of its own") + self.assertNotEqual(second["request_id"], first["request_id"]) + self.assertEqual(process(session, 4)["stats"]["crops"]["requested"], 2) + late = session.submit_crop(first["request_id"], first["track_id"], + first["frame_id"], frame_bytes()) + self.assertEqual(late["code"], "CROP_UNKNOWN_REQUEST", + "the outstanding request was minted under settings the session left") + engine.shutdown() + + def test_an_expired_request_leaves_exactly_one_more_ask(self): + session, engine, _ = make_crop_session(settings=LiveSettings(crop_ttl_frames=1)) + process(session, 1) + first = requested_crop(process(session, 2)) + self.assertIsNotNone(first) + process(session, 3) + # The one-frame budget ran out on frame 4, so the facing gets one more + # ask rather than being left with a request nobody will ever answer. + second = requested_crop(process(session, 4)) + self.assertIsNotNone(second, "an expiry must not strand a facing only the crop can decide") + self.assertNotEqual(second["request_id"], first["request_id"]) + process(session, 5) + process(session, 6) + result = process(session, 7) + self.assertIsNone(requested_crop(result), "the budget is bounded, not a loop") + self.assertEqual(result["stats"]["crops"]["requested"], 2) + engine.shutdown() + + def test_a_drifted_label_lets_the_next_product_be_asked_about(self): + """The facing in the frame is a different product, so it gets its own ask.""" + session, engine, detector = make_crop_session() + panned(detector, 100) + process(session, 1) + request = requested_crop(process(session, 2)) + session.submit_crop(request["request_id"], request["track_id"], + request["frame_id"], frame_bytes()) + self.assertTrue(wait_for(lambda: session._tracks[request["track_id"]].state == "recognized")) + track_id = request["track_id"] + asked_again = None + for frame_id, x in enumerate([118, 136, 154, 172, 190, 208], start=3): + panned(detector, x) + result = process(session, frame_id) + self.assertIn(track_id, session._tracks, "the id must survive the pan") + again = requested_crop(result) + if again is not None and again["request_id"] != request["request_id"]: + asked_again = again + break + self.assertIsNotNone(asked_again, + "a drifted label must hand the new product its own crop ask") + self.assertGreater(asked_again["box"][0], request["box"][0], + "the new ask must be about where the product is now") + engine.shutdown() + + def test_a_late_crop_cannot_repaint_a_settled_label(self): + """A track that took a neighbour's label keeps it; a later crop does not overrule it.""" + matcher = SecondAnswerWaits() + session, engine, detector = make_crop_session( + matcher=matcher, settings=LiveSettings(method="hybrid", min_crop_px=60) + ) + # Two narrow facings that overlap by more than a label can be trusted + # across: that is the only shape in which one inherits the other's. + detector.boxes = [ + {"id": 0, "box": [100, 100, 150, 340], "confidence": 0.9, "class_id": 0, + "width": 50, "height": 240, "center": [125, 220]}, + {"id": 1, "box": [110, 100, 160, 340], "confidence": 0.9, "class_id": 0, + "width": 50, "height": 240, "center": [135, 220]}, + ] + process(session, 1) + requests = {row["track_id"]: row["crop_request"] + for row in process(session, 2)["tracks"] if "crop_request" in row} + self.assertEqual(len(requests), 2, "both facings are too small for the frame crop") + settled_id, waiting_id = list(requests) + + session.submit_crop(requests[settled_id]["request_id"], settled_id, + requests[settled_id]["frame_id"], frame_bytes()) + self.assertTrue(matcher.first_answered.wait(2.0), "the first crop must be decided") + session.submit_crop(requests[waiting_id]["request_id"], waiting_id, + requests[waiting_id]["frame_id"], frame_bytes()) + + inherited = process(session, 3) + row = next(t for t in inherited["tracks"] if t["track_id"] == waiting_id) + self.assertTrue(row["inherited"], "the facing takes the label its neighbour settled") + self.assertEqual(row["identity"]["sku_id"], "sku_a") + + matcher.gate.set() + self.assertTrue(wait_for(lambda: session._tracks[waiting_id].in_flight == 0)) + settled = next(t for t in process(session, 4)["tracks"] if t["track_id"] == waiting_id) + self.assertEqual(settled["identity"]["sku_id"], "sku_a", + "a crop that arrives late must not overrule a settled label") + self.assertTrue(settled["inherited"]) + self.assertEqual(settled["identity"]["source"], "inherited") + self.assertEqual(session._tracks[waiting_id].attempts, 0, + "an answer that could not decide is not an attempt") + engine.shutdown() + if __name__ == "__main__": unittest.main() diff --git a/recognization/docs/API.md b/recognization/docs/API.md index 6a37ee7..f3da7b2 100644 --- a/recognization/docs/API.md +++ b/recognization/docs/API.md @@ -272,24 +272,30 @@ classification, not recognized; subsequent results carry the updated identity an `identity.source` (`frame`, `crop`, or `inherited`). `stats.crops` has `requested`, `accepted`, and `rejected` counts. -The hello limit is 1 MiB of **decoded crop bytes**. The optional -`max_crop_bytes` in `capabilities` states the client's limit; the acknowledged -session limit is clamped to 1 KiB–1 MiB (default 128 KiB when omitted). Use the +The hello limit is 1 MiB of **decoded crop bytes**. The `max_crop_bytes` in +`capabilities` states the client's limit and must be an integer in 1..1048576; a +statement that is not one is refused with `BAD_MESSAGE` and the lane stays where +it was, so both sides keep one limit (128 KiB when the key is `null`). Use the acknowledged limit, not only hello's ceiling. At most four requests are outstanding -per session and each track object is requested at most once. A request expires +per session, and each track object is asked about at most `max_attempts` times +(the same budget the frame path spends, 2 by default); the budget refills when the +method changes or when a settled label drifts off the object. A request expires after more than six processed frames; clients keep a bounded recent-frame buffer and may miss a crop if its capture has already been evicted. A late response -cannot label a recycled track. Changing method or resetting invalidates requests. +cannot label a recycled track. Changing method, resetting, or dropping a drifted +label invalidates that object's outstanding requests. Crop failures are recoverable text messages with `type: "crop_error"`, `code`, and `message`. Pre-decode failures also echo `request_id`; session-level refusals do not currently echo it. Codes: `CROP_BAD_MESSAGE` (missing/invalid base64 JPEG -or missing IDs), `CROP_TOO_LARGE` (empty or oversized decoded crop, or excessive -encoded length), `CROP_UNKNOWN_REQUEST` (no outstanding request, including one -already expired), `CROP_MISMATCH` (track/frame differs), `CROP_STALE` (late or +or missing IDs), `CROP_TOO_LARGE` (empty or oversized decoded crop, excessive +encoded length, or a JPEG whose header claims more than 4 megapixels), +`CROP_UNKNOWN_REQUEST` (no outstanding request, including one already expired), +`CROP_MISMATCH` (track/frame differs), `CROP_STALE` (late or invalidated track/session), `CROP_UNDECODABLE` (unreadable image), and `CROP_UNAVAILABLE` (classifier shutting down). A request is single-use, including when its response is rejected; continue streaming frames rather than retrying it. +The server may ask again for the same object while its budget lasts. **Backpressure:** the server keeps one pending frame per session and processes the newest, counting the ones it replaced in `frames_dropped`. A client should keep one