From d5a7a56dffcce2515c27175e736ee106430bf396 Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Mon, 28 Sep 2026 14:41:49 +0900 Subject: [PATCH] fix: bind tag projection to each frame origin --- recognization/code/live.py | 74 +++++++++++++++++------------ recognization/code/tags.py | 12 +++-- recognization/code/test_live.py | 74 +++++++++++++++++++++++++++++ recognization/code/test_ocr_tags.py | 21 ++++++++ 4 files changed, 146 insertions(+), 35 deletions(-) diff --git a/recognization/code/live.py b/recognization/code/live.py index 3850448..523a786 100644 --- a/recognization/code/live.py +++ b/recognization/code/live.py @@ -189,8 +189,9 @@ class LiveSession: self._catalog_by_sku: Dict[str, Dict[str, Any]] = {} self._catalog_version = "" self._tags = None - # Shelf position of this frame's origin. Updated from a content shift - # before the session lock; tag cadence keys rows by it. Not a shelf map. + # 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 @@ -337,22 +338,23 @@ class LiveSession: } def reset(self) -> None: - with self._lock: - self._generation += 1 - self.tracker.reset() - self._tracks.clear() - self._last_seen_seq.clear() - self._relabel_count.clear() - self._restart_count.clear() - self._slot_history.clear() - self._label_memory.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. + 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._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]: @@ -367,7 +369,6 @@ class LiveSession: empty_enabled = self.settings.empty_slots generation = self._generation - shelf_offset = self._note_frame_shift(image) boxes = self.engine.detect(image, conf=conf) @@ -495,7 +496,7 @@ class LiveSession: summary = self._summary_locked() stats = self._stats_locked() track_payload = [self._track_payload(state) for state in self._tracks.values()] - tag_summary = self._apply_tags(seq, track_payload, tag_bands) + 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"] @@ -996,16 +997,19 @@ class LiveSession: def _note_frame_shift(self, image: Any) -> Tuple[float, float]: """Shelf position of this frame's origin. Phase correlation only. - Runs before the session lock. 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. + 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. """ - if self._shift_map is None: - from shelf_map import 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) - with self._lock: + 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]) @@ -1034,19 +1038,27 @@ class LiveSession: LOGGER.warning("tag pass failed: %s", error) def _apply_tags( - self, seq: int, track_payloads: List[Dict[str, Any]], bands: Sequence[Any] + 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). + 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) + 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"]) diff --git a/recognization/code/tags.py b/recognization/code/tags.py index 2a7d8a0..0a97c5f 100644 --- a/recognization/code/tags.py +++ b/recognization/code/tags.py @@ -788,16 +788,20 @@ class TagService: Records live in shelf coordinates. They are projected into this frame before association, and a tag that does not overlap the frame is left out, so an old tag cannot bind a new facing that reused its screen box. - Returned boxes stay in screen pixels. `shelf_offset` defaults to the - origin of the latest `plan` (a still, or a caller that has not panned). + Returned boxes stay in screen pixels. A caller that omits `shelf_offset` + is a still: screen coordinates, never another frame's stored origin. Returns (tag_by_track_id, all_tags, state_counts). A track whose tag was never read gets state 'unreadable' when a tag was found near it, and 'absent' when nothing tag-shaped was seen there. """ with self._lock: - origin = _origin_of(shelf_offset if shelf_offset is not None else self._view_origin) - shape = image_shape if image_shape is not None else self._view_shape + origin = _origin_of(shelf_offset) + shape = ( + (int(image_shape[0]), int(image_shape[1])) + if image_shape is not None + else self._view_shape + ) visible = _visible_records(list(self._tags.values()), origin, shape) by_track: Dict[int, Dict[str, Any]] = {} diff --git a/recognization/code/test_live.py b/recognization/code/test_live.py index 3fbdc7d..5b1d02a 100644 --- a/recognization/code/test_live.py +++ b/recognization/code/test_live.py @@ -307,6 +307,80 @@ class TagPassIntegrationTests(unittest.TestCase): ) engine.shutdown() + def test_overlapping_frames_keep_their_own_origin(self): + """A later worker must not become the previous-frame baseline or steal origin.""" + session, engine, _ = self.make_tagged_session() + + class Ordered: + def __init__(self): + self.inside = 0 + self.max_inside = 0 + self.calls = 0 + self._inner = threading.Lock() + + def register(self, image): + with self._inner: + self.inside += 1 + self.max_inside = max(self.max_inside, self.inside) + self.calls += 1 + call = self.calls + time.sleep(0.03) + with self._inner: + self.inside -= 1 + return (0.0, -400.0) + + session._shift_map = Ordered() + errors = [] + + def run(frame_id): + try: + session.process_frame(frame_id, frame_bytes()) + except Exception as exc: + errors.append(exc) + + early = threading.Thread(target=run, args=(1,)) + late = threading.Thread(target=run, args=(2,)) + early.start() + late.start() + early.join(2.0) + late.join(2.0) + self.assertEqual(errors, []) + self.assertFalse(early.is_alive() or late.is_alive()) + self.assertEqual(session._shift_map.calls, 2) + self.assertEqual(session._shift_map.max_inside, 1) + self.assertEqual(session._frame_origin[1], 800.0) + engine.shutdown() + + def test_cached_tag_projects_at_this_frame_origin(self): + """A panned frame must not attach a tag that only overlaps the previous origin.""" + from tags import TagRecord + + session, engine, _ = self.make_tagged_session() + record = TagRecord( + tag_id="cached", box=[150, 700, 190, 760], band_index=0, name="cached", price=150, + flags=["tax_consistent"], + ) + record.settled = True + session._tags._tags[record.tag_id] = record + + class Panned: + def __init__(self): + self.calls = 0 + + def register(self, image): + self.calls += 1 + return (0.0, -400.0) if self.calls == 2 else (0.0, 0.0) + + session._shift_map = Panned() + first = session.process_frame(1, frame_bytes()) + self.assertFalse(any(track.get("price_tag", {}).get("id") for track in first["tracks"])) + second = session.process_frame(2, frame_bytes()) + bound = [track["price_tag"] for track in second["tracks"] if track.get("price_tag", {}).get("id")] + self.assertTrue(bound) + self.assertGreaterEqual(bound[0]["box"][1], 200) + self.assertLess(bound[0]["box"][1], 480) + engine.shutdown() + class FramePipelineTests(unittest.TestCase): def test_latest_frame_wins_and_drops_are_counted(self): diff --git a/recognization/code/test_ocr_tags.py b/recognization/code/test_ocr_tags.py index 26c6401..2d481da 100644 --- a/recognization/code/test_ocr_tags.py +++ b/recognization/code/test_ocr_tags.py @@ -385,6 +385,27 @@ class TagServiceTests(unittest.TestCase): self.assertEqual(by_track[9]["state"], TAG_STATE_ABSENT) self.assertIsNone(by_track[9]["id"]) + def test_attach_does_not_use_the_latest_plan_origin(self): + """A later plan must not project this attach through its stored origin.""" + service, _ = self.make(pass_interval_frames=0, settle_observations=1, settle_passes=1) + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + array = image() + first = service.plan(boxes, array.shape, shelf_offset=(0.0, 0.0)) + service.run_pass(array, first[0]) + service.plan(boxes, array.shape, shelf_offset=(0.0, 400.0)) + by_track, tags, _ = service.attach( + [{"track_id": 0, "box": [100, 100, 200, 300]}], + image_shape=array.shape, + ) + self.assertTrue(tags, "omitting shelf_offset is screen coordinates, not the panned plan origin") + self.assertEqual(by_track[0]["state"], TAG_STATE_READABLE) + by_track, tags, _ = service.attach( + [{"track_id": 0, "box": [100, 100, 200, 300]}], + shelf_offset=(0.0, 0.0), + image_shape=array.shape, + ) + self.assertEqual(by_track[0]["state"], TAG_STATE_READABLE) + class StorePolicyTests(unittest.TestCase): def test_a_missing_file_uses_the_defaults(self):