From f0e2baaf3dd6439960a0e72af180a093b1bff2ba Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Mon, 28 Sep 2026 13:41:12 +0900 Subject: [PATCH] fix: anchor OCR passes and tags to shelf coordinates --- recognization/code/tags.py | 183 +++++++++++++++++++++------- recognization/code/test_ocr_tags.py | 50 ++++++++ 2 files changed, 191 insertions(+), 42 deletions(-) diff --git a/recognization/code/tags.py b/recognization/code/tags.py index b4ad79d..2a7d8a0 100644 --- a/recognization/code/tags.py +++ b/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()] - records = list(self._tags.values()) + 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 + 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): - by_track[int(track["track_id"])] = { - "id": None, - "box": None, - "state": TAG_STATE_UNREADABLE, - "price": None, - "name": None, - "confidence": 0.0, - "observations": 0, - "settled": False, - } + elif self._band_had_readable(record=None, box=box, records=records, screen_of=screen_of): + by_track[int(track["track_id"])] = _empty_tag(TAG_STATE_UNREADABLE) else: - by_track[int(track["track_id"])] = { - "id": None, - "box": None, - "state": TAG_STATE_ABSENT, - "price": None, - "name": None, - "confidence": 0.0, - "observations": 0, - "settled": False, - } + by_track[int(track["track_id"])] = _empty_tag(TAG_STATE_ABSENT) 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 - cy = (candidate.box[1] + candidate.box[3]) / 2.0 + shown = (screen_of or {}).get(candidate.tag_id, candidate.box) + 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]: - """The tag that belongs to this facing: below it, horizontally overlapping.""" +def _tag_below( + 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 import argparse diff --git a/recognization/code/test_ocr_tags.py b/recognization/code/test_ocr_tags.py index 25d911d..26c6401 100644 --- a/recognization/code/test_ocr_tags.py +++ b/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):