Improve identity lanes: text evidence floor, motion-compensated labels

The text lane now requires shared bigrams between the read and the winning
row's name before it will answer, and the local lane passes the catalog
signatures through so the decision can weigh text evidence. On the labeled
shelf this moves the local lane from 66% to 80% end-to-end at full
resolution (91% precision) and 66% at live resolution (95% precision, 2
false accepts), against 57%/35% for the matcher alone.

Live fixes:
- `_plan_classification` counts `in_flight` again. The increment had been
  dropped while removing a `label_frame_id` write, so the guard never
  blocked, every frame re-planned the same tracks, and the executor queue
  grew without bound until the server stopped answering.
- Label memory boxes carry the shelf origin they were measured against and
  are shifted before the IoU test, so a panning camera no longer slides a
  stale box onto its neighbour.
- The text lane is resolved before the session lock is taken, removing a
  lock-ordering hazard against the lane's own build lock.
- A native crop always reads; the frame crop skips OCR only when it already
  has a reading.

Stream (20 s pan, 4 runs): 71-79% coverage at 76-84% precision, against
62-71% at 73-75% before.
This commit is contained in:
tngo 2026-09-29 02:08:23 +09:00
parent 771e9285d7
commit bf9c04bce4
7 changed files with 293 additions and 31 deletions

View file

@ -74,6 +74,99 @@ class TextLaneSignatureTests(unittest.TestCase):
decision = DecisionPolicy().decide([], scores, text_read="マヨネーズ")
self.assertIsNone(decision["best_sku"])
def test_a_signature_never_pairs_across_two_words(self):
"""A catalog name must not claim text it does not contain.
"キューピー アマニ油" normalizes to "キユピ アマニ油"; with the space
dropped it was "キユピアマニ油", which contains "ピア" - a pair neither
word has. That pair is rare across the catalog, so IDF weighted it as
distinctive and the garbage read "キユピアク5C" was accepted as l02.
"""
from rebuild_catalog import signature_bigrams
name = "キューピー アマニ油マヨネーズ 200g"
self.assertNotIn("ピア", signature_bigrams(name))
self.assertIn("アマ", signature_bigrams(name))
def test_a_read_keeps_matching_across_the_spaces_ocr_inserts(self):
"""The read is the noisy side, so its spaces must not break a match.
Vision prints "ピュア セレクト" where the catalog name has
"ピュアセレクト". Refusing to pair across the read's spaces dropped
two correct l10 labels below the accept line, so the read pairs
across them and the signature is the side that stays strict.
"""
from rebuild_catalog import bigrams
self.assertIn("アセ", bigrams("ピュア セレクト マヨネーズ"))
def test_a_garbage_read_does_not_name_a_row(self):
"""The read that the fake pair used to accept must not name a row.
It shares only "キユ" and "ユピ" with l02's name, both of which every
Kewpie row prints. Bounded to the catalog's own vocabulary the read
keeps just those two pairs, so it scores the same against every
Kewpie row and the margin gate refuses it - the same mechanism that
already handles a read of shared brand copy alone.
"""
from identity_lanes import DecisionPolicy, TextLane
rows = self.ROWS + [{"sku_id": "sku_l02", "product_name": "キューピー アマニ油マヨネーズ 200g"}]
read = "キユピアク5C"
scores = TextLane.from_catalog(rows, "no-such-dir", ocr_engine=None).scores_from_text(read)
ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
self.assertEqual(ranked[0][1], ranked[1][1], "the read cannot tell the Kewpie rows apart")
decision = DecisionPolicy().decide([], ranked, text_read=read)
self.assertIsNone(decision["best_sku"])
def test_a_read_of_real_text_still_names_its_row(self):
"""Bounding the denominator must not cost a correct read its answer.
This is the facing the strict-read variant broke: l10's own name read
with OCR's spacing, which has to clear the accept line by margin.
"""
from identity_lanes import TEXT_ACCEPT, TEXT_MARGIN, TextLane
rows = self.ROWS + [{"sku_id": "sku_l02", "product_name": "キューピー アマニ油マヨネーズ 200g"}]
scores = TextLane.from_catalog(rows, "no-such-dir", ocr_engine=None).scores_from_text(
"213日以月A ピュア セレクト マヨネーズ"
)
ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
self.assertEqual(ranked[0][0], "sku_l10")
self.assertGreaterEqual(ranked[0][1], TEXT_ACCEPT)
self.assertGreaterEqual(ranked[0][1] - ranked[1][1], TEXT_MARGIN)
def test_one_shared_pair_is_not_enough_to_name_a_row(self):
"""A ratio of 1.0 on a single pair is not evidence.
"XXイス" contains exactly one pair the catalog knows, and the row that
contains it scores 1.0 with a full margin over every other row. On
image6 that shape named sku_l08 from the read "サソース". The floor
asks how much of the read the row actually explains, so one pair is
refused while the same read with two is accepted.
"""
from identity_lanes import DecisionPolicy, TextLane
rows = [
{"sku_id": "sku_l08", "product_name": "オイスターソース"},
{"sku_id": "sku_l01", "product_name": "キューピー マヨネーズ"},
]
lane = TextLane.from_catalog(rows, "no-such-dir", ocr_engine=None)
thin = sorted(lane.scores_from_text("XXイス").items(), key=lambda kv: kv[1], reverse=True)
self.assertEqual(thin[0][1], 1.0, "the ratio alone cannot see the difference")
self.assertIsNone(
DecisionPolicy(signatures=lane.signatures).decide([], thin, text_read="XXイス")["best_sku"]
)
enough = sorted(lane.scores_from_text("ZZソース").items(), key=lambda kv: kv[1], reverse=True)
self.assertEqual(
DecisionPolicy(signatures=lane.signatures).decide(
[], enough, text_read="ZZソース"
)["best_sku"],
"sku_l08",
)
class FakeDetector:
"""Records concurrency and returns fixed boxes."""