fix: anchor OCR passes and tags to shelf coordinates

recognization/code/tags.py

@@ -84,11 +84,19 @@ _TAX_RE = re.compile(r"税(込|抜|不|別)")
 
 @dataclass
 class RailBand:
-    """One horizontal strip of shelf rail, derived from a product row."""
+    """One horizontal strip of shelf rail, derived from a product row.
+
+    `box` is this frame's pixels (the crop `run_pass` reads). `origin` is the
+    shelf position of the frame origin at the moment `plan` chose the band, and
+    `key` is the rail that origin names. Both travel with the band: a later
+    frame's origin must not relabel a pass that is still queued.
+    """
 
     box: List[int]
     row_index: int
     product_ids: List[int] = field(default_factory=list)
+    origin: Optional[Tuple[float, float]] = None
+    key: Optional[Tuple[str, int]] = None
 
 
 # The field model lives with the card reader; the pipeline consumes it unchanged.
@@ -439,6 +447,9 @@ class TagService:
         self.passes_failed = 0
         self.last_pass_error: Optional[str] = None
         self.passes_skipped = 0
+        # Latest frame only. A queued pass must not read these; its origin is on the band.
+        self._view_origin: Optional[Tuple[float, float]] = None
+        self._view_shape: Optional[Tuple[int, int]] = None
 
     # ------------------------------------------------------------------ frame loop
     def plan(
@@ -462,7 +473,8 @@ class TagService:
         width = float(image_shape[1])
         bands = [band for band in bands if (band.box[2] - band.box[0]) >= self.settings.min_band_width_ratio * width]
         with self._lock:
-            self._shelf_offset = shelf_offset
+            self._view_origin = _origin_of(shelf_offset)
+            self._view_shape = (int(image_shape[0]), int(image_shape[1]))
             self._frame_seq += 1
             seq = self._frame_seq
             chosen: List[RailBand] = []
@@ -481,6 +493,8 @@ class TagService:
                     break
             for band in chosen:
                 key = band_key(band, shelf_offset)
+                band.origin = _origin_of(shelf_offset)
+                band.key = key
                 self._last_pass_seq[key] = seq
                 self._pending_bands.add(key)
         return chosen
@@ -492,7 +506,7 @@ class TagService:
         independent reads of the same tag are what makes the vote meaningful
         (identical pixels would produce identical output).
         """
-        key = band_key(band, getattr(self, "_shelf_offset", None))
+        key = _planned_key(band)
         try:
             x1, y1, x2, y2 = band.box
             crop = image[y1:y2, x1:x2]
@@ -632,13 +646,19 @@ class TagService:
         return all(tag.settled for tag in row_tags)
 
     def _absorb(self, candidate: Dict[str, Any], band: RailBand, pass_id: int) -> TagObservation:
-        """Fold one clustered tag into the cache; must be called with the lock held."""
+        """Fold one clustered tag into the cache; must be called with the lock held.
+
+        The candidate box is this frame's pixels. The record keeps shelf pixels,
+        using the origin `plan` stamped on the band, so a later pan cannot make
+        this read land on a different rail.
+        """
         box = candidate["box"]
         lines = candidate["lines"]
         fields = candidate["fields"]
-        shelf_key = band_key(band, getattr(self, "_shelf_offset", None))
+        shelf_key = _planned_key(band)
+        shelf_box = _to_shelf(box, band.origin)
         observation = TagObservation(
-            box=list(box),
+            box=list(shelf_box),
             lines=list(lines),
             fields=fields,
             band_index=band.row_index,
@@ -646,11 +666,11 @@ class TagService:
             observed_at=time.time(),
         )
 
-        record = self._match_record(box, shelf_key)
+        record = self._match_record(shelf_box, shelf_key)
         if record is None:
             record = TagRecord(
                 tag_id=f"tag_{uuid.uuid4().hex[:8]}",
-                box=list(box),
+                box=list(shelf_box),
                 band_index=band.row_index,
                 shelf_key=shelf_key,
                 first_seen=observation.observed_at,
@@ -658,7 +678,7 @@ class TagService:
             if len(self._tags) >= self.settings.max_tags:
                 return observation
             self._tags[record.tag_id] = record
-        record.box = list(box)
+        record.box = list(shelf_box)
         record.last_seen = observation.observed_at
         record.observations.append(observation)
         record.passes += 1
@@ -758,52 +778,47 @@ class TagService:
 
     # ------------------------------------------------------------------ attach
     def attach(
-        self, tracks: Sequence[Dict[str, Any]]
+        self,
+        tracks: Sequence[Dict[str, Any]],
+        shelf_offset: Optional[Sequence[float]] = None,
+        image_shape: Optional[Sequence[int]] = None,
     ) -> Tuple[Dict[int, Dict[str, Any]], List[Dict[str, Any]], Dict[str, int]]:
         """Give each track its tag for this frame.
 
+        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).
+
         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:
-            tags = [record.as_dict() for record in self._tags.values()]
+            origin = _origin_of(shelf_offset if shelf_offset is not None else self._view_origin)
-            records = list(self._tags.values())
+            shape = image_shape 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]] = {}
         used: set = set()
+        records = [record for record, _screen in visible]
+        screen_of = {record.tag_id: screen for record, screen in visible}
+        tags = [_screen_dict(record, screen_of[record.tag_id]) for record in records]
         for track in tracks:
             box = track.get("box") or track.get("raw_box")
             if not box:
                 continue
-            record = _tag_below(box, records, used)
+            record = _tag_below(box, records, used, screen_of)
             if record is not None:
                 used.add(record.tag_id)
-                payload = record.as_dict()
+                payload = _screen_dict(record, screen_of[record.tag_id])
                 payload["associated"] = True
                 by_track[int(track["track_id"])] = payload
-            elif self._band_had_readable(record=None, box=box, records=records):
+            elif self._band_had_readable(record=None, box=box, records=records, screen_of=screen_of):
-                by_track[int(track["track_id"])] = {
+                by_track[int(track["track_id"])] = _empty_tag(TAG_STATE_UNREADABLE)
-                    "id": None,
-                    "box": None,
-                    "state": TAG_STATE_UNREADABLE,
-                    "price": None,
-                    "name": None,
-                    "confidence": 0.0,
-                    "observations": 0,
-                    "settled": False,
-                }
             else:
-                by_track[int(track["track_id"])] = {
+                by_track[int(track["track_id"])] = _empty_tag(TAG_STATE_ABSENT)
-                    "id": None,
-                    "box": None,
-                    "state": TAG_STATE_ABSENT,
-                    "price": None,
-                    "name": None,
-                    "confidence": 0.0,
-                    "observations": 0,
-                    "settled": False,
-                }
 
         counts = {
             TAG_STATE_READABLE: 0,
@@ -815,13 +830,20 @@ class TagService:
             counts[payload["state"]] = counts.get(payload["state"], 0) + 1
         return by_track, tags, counts
 
-    def _band_had_readable(self, record, box: Sequence[int], records: Sequence[TagRecord]) -> bool:
+    def _band_had_readable(
+        self,
+        record,
+        box: Sequence[int],
+        records: Sequence[TagRecord],
+        screen_of: Optional[Dict[str, List[int]]] = None,
+    ) -> bool:
         """Was there any tag-shaped evidence near this facing (readable or not)?"""
         x1, y1, x2, y2 = box
         width = max(1, x2 - x1)
         for candidate in records:
-            cx = (candidate.box[0] + candidate.box[2]) / 2.0
+            shown = (screen_of or {}).get(candidate.tag_id, candidate.box)
-            cy = (candidate.box[1] + candidate.box[3]) / 2.0
+            cx = (shown[0] + shown[2]) / 2.0
+            cy = (shown[1] + shown[3]) / 2.0
             if x1 - width * 0.3 <= cx <= x2 + width * 0.3 and y1 <= cy <= y2 + (y2 - y1) * 0.5:
                 return True
         return False
@@ -858,7 +880,8 @@ class TagService:
             self._pending_bands.clear()
             self._pass_id = 0
             self._frame_seq = 0
-            self._shelf_offset = None
+            self._view_origin = None
+            self._view_shape = None
 
 
 def _iou(a: Sequence[int], b: Sequence[int]) -> float:
@@ -876,8 +899,17 @@ def _iou(a: Sequence[int], b: Sequence[int]) -> float:
     return inter / union if union > 0 else 0.0
 
 
-def _tag_below(box: Sequence[int], records: Sequence[TagRecord], used: set) -> Optional[TagRecord]:
+def _tag_below(
-    """The tag that belongs to this facing: below it, horizontally overlapping."""
+    box: Sequence[int],
+    records: Sequence[TagRecord],
+    used: set,
+    screen_of: Optional[Dict[str, List[int]]] = None,
+) -> Optional[TagRecord]:
+    """The tag that belongs to this facing: below it, horizontally overlapping.
+
+    `screen_of` is the record projected into the current frame. Without it the
+    stored shelf box is used, which is the screen box when no origin was known.
+    """
     x1, y1, x2, y2 = box
     width = max(1, x2 - x1)
     height = max(1, y2 - y1)
@@ -886,7 +918,7 @@ def _tag_below(box: Sequence[int], records: Sequence[TagRecord], used: set) -> O
     for record in records:
         if record.tag_id in used:
             continue
-        rx1, ry1, rx2, ry2 = record.box
+        rx1, ry1, rx2, ry2 = (screen_of or {}).get(record.tag_id, record.box)
         center_x = (rx1 + rx2) / 2.0
         overlap = max(0, min(x2, rx2) - max(x1, rx1))
         overlap_ratio = overlap / max(1, rx2 - rx1)
@@ -907,6 +939,73 @@ def _tag_below(box: Sequence[int], records: Sequence[TagRecord], used: set) -> O
     return best
 
 
+def _origin_of(shelf_offset: Optional[Sequence[float]]) -> Optional[Tuple[float, float]]:
+    if not shelf_offset or len(shelf_offset) < 2:
+        return None
+    return (float(shelf_offset[0]), float(shelf_offset[1]))
+
+
+def _planned_key(band: RailBand) -> Tuple[str, int]:
+    """The rail `plan` chose. A band built without `plan` falls back to its screen row."""
+    if band.key is not None:
+        return band.key
+    return band_key(band, band.origin)
+
+
+def _shift_box(box: Sequence[int], dx: float, dy: float) -> List[int]:
+    return [int(round(box[0] + dx)), int(round(box[1] + dy)), int(round(box[2] + dx)), int(round(box[3] + dy))]
+
+
+def _to_shelf(box: Sequence[int], origin: Optional[Tuple[float, float]]) -> List[int]:
+    if origin is None:
+        return [int(v) for v in box]
+    return _shift_box(box, origin[0], origin[1])
+
+
+def _to_screen(box: Sequence[int], origin: Optional[Tuple[float, float]]) -> List[int]:
+    if origin is None:
+        return [int(v) for v in box]
+    return _shift_box(box, -origin[0], -origin[1])
+
+
+def _overlaps_frame(box: Sequence[int], image_shape: Sequence[int]) -> bool:
+    height, width = int(image_shape[0]), int(image_shape[1])
+    return box[2] > 0 and box[0] < width and box[3] > 0 and box[1] < height
+
+
+def _visible_records(
+    records: Sequence[TagRecord],
+    origin: Optional[Tuple[float, float]],
+    image_shape: Optional[Sequence[int]],
+) -> List[Tuple[TagRecord, List[int]]]:
+    shown: List[Tuple[TagRecord, List[int]]] = []
+    for record in records:
+        screen = _to_screen(record.box, origin)
+        if image_shape is not None and not _overlaps_frame(screen, image_shape):
+            continue
+        shown.append((record, screen))
+    return shown
+
+
+def _screen_dict(record: TagRecord, screen_box: Sequence[int]) -> Dict[str, Any]:
+    payload = record.as_dict()
+    payload["box"] = list(screen_box)
+    return payload
+
+
+def _empty_tag(state: str) -> Dict[str, Any]:
+    return {
+        "id": None,
+        "box": None,
+        "state": state,
+        "price": None,
+        "name": None,
+        "confidence": 0.0,
+        "observations": 0,
+        "settled": False,
+    }
+
+
 if __name__ == "__main__":  # smoke: python code/tags.py <image>
     import argparse
 

recognization/code/test_ocr_tags.py

@@ -335,6 +335,56 @@ class TagServiceTests(unittest.TestCase):
         self.assertGreaterEqual(service.stats()["passes_run"], 1)
         self.assertIsNone(service.stats()["last_pass_error"])
 
+    def test_a_queued_pass_keeps_the_shelf_it_was_planned_on(self):
+        """OCR that finishes after a pan must clear its own rail, not the new one."""
+        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()
+        planned = service.plan(boxes, array.shape, shelf_offset=(0.0, 0.0))
+        self.assertEqual(service.stats()["pending_bands"], 1)
+        moved = service.plan(boxes, array.shape, shelf_offset=(0.0, 400.0))
+        self.assertTrue(moved, "the new shelf is a different rail and must be queued too")
+        self.assertEqual(service.stats()["pending_bands"], 2)
+        service.run_pass(array, planned[0])
+        self.assertEqual(service.stats()["pending_bands"], 1, "the finished pass clears only its own key")
+        self.assertEqual(service._tags[next(iter(service._tags))].shelf_key[0], "shelf")
+        self.assertNotEqual(
+            service._tags[next(iter(service._tags))].shelf_key,
+            moved[0].key,
+            "the late read must not be filed under the shelf now in frame",
+        )
+        service.run_pass(array, moved[0])
+        self.assertEqual(service.stats()["pending_bands"], 0)
+        self.assertEqual(len({tag.shelf_key for tag in service._tags.values()}), 2)
+
+    def test_an_offscreen_tag_does_not_bind_the_new_facing(self):
+        """A pan that reuses screen coordinates must not hand the old tag to the new facing."""
+        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])
+        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)
+        self.assertEqual(by_track[0]["price"], 269)
+        self.assertGreaterEqual(tags[0]["box"][1], 0)
+        self.assertLess(tags[0]["box"][1], array.shape[0])
+
+        again = service.plan(boxes, array.shape, shelf_offset=(0.0, 400.0))
+        self.assertTrue(again)
+        by_track, tags, _ = service.attach(
+            [{"track_id": 9, "box": [100, 100, 200, 300]}],
+            shelf_offset=(0.0, 400.0),
+            image_shape=array.shape,
+        )
+        self.assertEqual(tags, [], "the old tag is above this frame")
+        self.assertEqual(by_track[9]["state"], TAG_STATE_ABSENT)
+        self.assertIsNone(by_track[9]["id"])
+
 
 class StorePolicyTests(unittest.TestCase):
     def test_a_missing_file_uses_the_defaults(self):