Compare commits

...

4 commits

13 changed files with 306 additions and 60 deletions

View file

@ -293,8 +293,11 @@ def create_app(
scans = scans if scans is not None else build_legacy_scans(engine)
# A live session can be recorded so a real scan becomes data to label later.
recorder = FrameRecorder(settings.record_frames, every=settings.record_every) \
if settings.record_frames else None
recorder = FrameRecorder(
settings.record_frames,
every=settings.record_every,
identity_backend=engine.settings.identity_backend,
) if settings.record_frames else None
samples = {item["id"]: item for item in get_available_shelf_images("add")}
# path -> (mtime_ns, size, width, height). Dropped when the file changes.
sample_dimensions: Dict[str, Tuple[int, int, int, int]] = {}
@ -1138,7 +1141,8 @@ def main() -> None:
engine = build_default_engine(engine_settings)
if (engine_settings.identity_backend or "").lower() == "gallery":
if engine.identity_gallery() is None:
raise RuntimeError(
engine.shutdown()
raise SystemExit(
"gallery identity backend requires local DINOv2 weights; refusing to start"
)
warm_ms = engine.warm_up()

View file

@ -15,8 +15,11 @@ store was enrolled; scoring them is training-set accuracy.
Prints a human count table and writes the same report as JSON (default
<recording>/eval_live.json). Missing input exits 2 with {"error": ...} on
stdout. --identity gallery uses the same LiveSession path as the server and
exits 2 when that gallery cannot load; it does not score the matcher instead.
stdout. A recording that names identity_backend is scored as that backend unless
--identity repeats it. A recording that does not name one requires --identity;
omitting it does not score the matcher. --identity gallery uses the same
LiveSession path as the server and exits 2 when that gallery cannot load; it
does not score the matcher instead.
The report names the backend, whether the head loaded, and the weight pins.
The same pinned recording, catalog, weights, and labels produce the
same counts; timings are omitted because they are not.
@ -102,9 +105,9 @@ def load_index(recording_dir: str) -> List[Dict[str, Any]]:
def recorded_identity(rows: Sequence[Dict[str, Any]]) -> Optional[str]:
"""Backend named by the recording, when every present value agrees.
FrameRecorder does not write this. A later index may, on the first line or
on each frame, under identity_backend or identity.backend. Silence means
the caller must say --identity. A mix is not a default.
FrameRecorder writes identity_backend on each line when the server constructed
it with the active backend. Older indexes may omit it, or name it once under
identity.backend. Silence is not a backend. A mix is not a default.
"""
found: List[str] = []
for row in rows:
@ -124,13 +127,17 @@ def recorded_identity(rows: Sequence[Dict[str, Any]]) -> Optional[str]:
def resolve_identity(requested: Optional[str], recorded: Optional[str]) -> str:
"""CLI wins. A recording may fill the gap. Neither one stays the matcher."""
chosen = (requested or recorded or "matcher").strip().lower()
"""CLI and recording must name the same backend. Silence is not matcher."""
chosen = (requested or recorded or "").strip().lower()
if not chosen:
raise EvalInputError(
"recording does not name an identity backend; pass --identity matcher or --identity gallery"
)
if chosen not in ("matcher", "gallery"):
raise EvalInputError(f"identity must be matcher or gallery, not {requested or recorded}")
if recorded and requested and recorded != chosen:
raise EvalInputError(
f"recording identity is {recorded}; pass --identity {recorded} or omit --identity"
f"recording identity is {recorded}; pass --identity {recorded}"
)
return chosen
@ -219,7 +226,7 @@ def catalog_fingerprint(stock_dir: str, stock_csv: str = "stock.csv") -> Dict[st
def model_fingerprint(
weights_path: str,
*,
identity: str = "matcher",
identity: str,
gallery: Any = None,
) -> Dict[str, Any]:
"""YOLO pin plus, for gallery, the backbone and head that actually loaded."""
@ -545,7 +552,7 @@ def score_recording(
labels_path: Optional[str] = None,
iou_min: float = 0.5,
live_frames: Optional[Sequence[Dict[str, Any]]] = None,
identity: str = "matcher",
identity: str,
gallery: Any = None,
) -> Dict[str, Any]:
"""Score one session directory. live_frames, when given, are the replayed results.
@ -564,7 +571,9 @@ def score_recording(
source = list(live_frames) if live_frames is not None else rows
tracks = [track for frame in source for track in (frame.get("tracks") or [])]
decisions = score_tracks(tracks)
backend = (identity or "matcher").strip().lower()
backend = identity.strip().lower()
if backend not in ("matcher", "gallery"):
raise EvalInputError(f"identity must be matcher or gallery, not {identity}")
report = {
"recording": recording_dir,
"source": "live_replay" if live_frames is not None else "recorded_index",
@ -678,7 +687,7 @@ def format_report(report: Dict[str, Any]) -> str:
head_note = f" head {head}" if head else ""
lines = [
f"source: {report['source']}",
f"identity: {identity.get('backend', 'matcher')}{head_note}",
f"identity: {identity.get('backend')}{head_note}",
(
f"decisions: accepted {decisions['accepted']} ambiguous {decisions['ambiguous']} "
f"rejected {decisions['rejected']} unknown {decisions['unknown']} "
@ -782,7 +791,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
"--identity",
default=None,
choices=["matcher", "gallery"],
help="which mechanism names products (default: recorded backend, else matcher)",
help="which mechanism names products (required when the recording omits identity_backend)",
)
parser.add_argument("--device", default=None, help="cpu | mps | cuda | cuda:0")
args = parser.parse_args(argv)

View file

@ -132,12 +132,14 @@ def enrollment_report_text(report: dict) -> str:
def build_model(image_size: int = 518):
import timm
import torch
from gallery import load_dinov2
device = "mps" if torch.backends.mps.is_available() else "cpu"
model = timm.create_model("vit_small_patch14_dinov2.lvd142m", pretrained=True,
num_classes=0, img_size=image_size, dynamic_img_size=True)
# Same hash-pinned local file as inference. pretrained=True would fetch a
# different copy even with YOLO_OFFLINE=1.
model = load_dinov2(image_size)
return model.eval().to(device), device

View file

@ -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"])

View file

@ -15,7 +15,8 @@ Layout:
DIR/<session>/000123.jpg the frame, exactly as the client sent it
DIR/<session>/index.jsonl one line per frame:
{"frame": 123, "path": "...", "at": 1699..., "summary": {...},
{"frame": 123, "path": "...", "at": 1699..., "identity_backend": "gallery",
"summary": {...},
"tracks": [{"box": [...], "identity": {...}, "tag": {...}}]}
Frames are written at most every `every` processed frames (default 3) so a long
@ -35,10 +36,17 @@ from typing import Any, Dict, List, Optional
class FrameRecorder:
"""Write frames and their results to disk, for labelling later."""
def __init__(self, root: str, every: int = 3, max_frames_per_session: int = 4000) -> None:
def __init__(
self,
root: str,
every: int = 3,
max_frames_per_session: int = 4000,
identity_backend: Optional[str] = None,
) -> None:
self.root = root
self.every = max(1, int(every))
self.max_frames_per_session = int(max_frames_per_session)
self.identity_backend = (identity_backend or "").strip().lower() or None
self._lock = threading.Lock()
self._counts: Dict[str, int] = {}
self._written: Dict[str, int] = {}
@ -86,6 +94,8 @@ class FrameRecorder:
"summary": result.get("summary"),
"tracks": tracks,
}
if self.identity_backend:
line["identity_backend"] = self.identity_backend
try:
with open(os.path.join(directory, "index.jsonl"), "a", encoding="utf-8") as handle:
handle.write(json.dumps(line, ensure_ascii=False) + "\n")

View file

@ -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]] = {}

View file

@ -274,6 +274,33 @@ class CatalogTests(unittest.TestCase):
self.assertEqual(product["number_out"], 3)
self.assertIsNone(engine.update_stock("nope", -1))
class GalleryLoadTests(unittest.TestCase):
def test_a_failed_gallery_load_returns_none_and_is_not_retried(self):
engine, _, _, _ = make_engine(identity_backend="gallery", stock_dir="does-not-exist")
class BrokenGallery:
def __init__(self, *args, **kwargs):
pass
def load(self):
raise FileNotFoundError("DINOv2 weights not found")
import gallery as gallery_module
original = gallery_module.ProductGallery
gallery_module.ProductGallery = BrokenGallery
try:
self.assertIsNone(engine.identity_gallery())
self.assertIsNone(engine.identity_gallery())
finally:
gallery_module.ProductGallery = original
engine.shutdown()
self.assertIs(engine._gallery, False)
class CatalogImageTests(unittest.TestCase):
def test_catalog_image_path_is_resolved_through_inventory(self):
engine, _, _, _ = make_engine()
self.assertIsNone(engine.catalog_image_path("../../etc/passwd"))

View file

@ -23,6 +23,8 @@ from eval_live import ( # noqa: E402
labelled_top1,
leaked_labels,
load_index,
recorded_identity,
resolve_identity,
score_recording,
score_tracks,
track_churn,
@ -176,7 +178,7 @@ class LeakageTests(unittest.TestCase):
)
with open(os.path.join(recording, "labels.json"), "w", encoding="utf-8") as handle:
json.dump({"labels": [{"frame": 1, "box": [0, 0, 12, 12], "sku_id": "sku_a"}]}, handle)
report = score_recording(recording, stock_dir=stock)
report = score_recording(recording, stock_dir=stock, identity="matcher")
self.assertEqual(report["leakage"]["excluded"], 1)
self.assertEqual(report["labelled_top1"]["labels"], 0)
self.assertEqual(report["labelled_top1"]["top1"], None)
@ -220,8 +222,8 @@ class CliTests(unittest.TestCase):
handle.write(json.dumps(line) + "\n")
with open(os.path.join(recording, "labels.json"), "w", encoding="utf-8") as handle:
json.dump({"labels": [{"frame": 4, "box": [0, 0, 8, 8], "sku_id": "sku_a"}]}, handle)
first = score_recording(recording, stock_dir=stock)
second = score_recording(recording, stock_dir=stock)
first = score_recording(recording, stock_dir=stock, identity="matcher")
second = score_recording(recording, stock_dir=stock, identity="matcher")
self.assertEqual(first, second)
self.assertEqual(first["decisions"]["accepted"], 1)
self.assertEqual(first["decisions"]["ambiguous"], 1)
@ -273,5 +275,26 @@ class IdentityContractTests(unittest.TestCase):
self.assertFalse(os.path.exists(report_path))
self.assertNotIn("accepted_rate", completed.stdout)
def test_a_silent_recording_does_not_become_the_matcher(self):
with self.assertRaises(EvalInputError):
resolve_identity(None, None)
self.assertEqual(resolve_identity("matcher", None), "matcher")
self.assertEqual(resolve_identity(None, "gallery"), "gallery")
with self.assertRaises(EvalInputError):
resolve_identity("matcher", "gallery")
def test_recorded_backend_is_read_and_a_mix_is_refused(self):
self.assertEqual(recorded_identity([{"identity_backend": "Gallery"}]), "gallery")
self.assertIsNone(recorded_identity([{"frame": 1}]))
self.assertEqual(
recorded_identity([{"identity": {"backend": "matcher", "sku_id": "sku_a"}}]),
"matcher",
)
with self.assertRaises(EvalInputError):
recorded_identity([
{"identity_backend": "gallery"},
{"identity_backend": "matcher"},
])
if __name__ == "__main__":
unittest.main()

View file

@ -225,5 +225,43 @@ class HeadPinTests(unittest.TestCase):
self.assertEqual(gallery._head["class_ids"], subset)
self.assertIsNone(head_class_set(["sku_a", "sku_a", ""]))
class TrainerPinTests(unittest.TestCase):
def test_head_training_loads_the_pinned_file_and_keeps_preprocess(self):
import finetune_identity
import gallery as gallery_module
seen = []
class FakeBackbone:
def eval(self):
return self
def to(self, device):
return self
sentinel = FakeBackbone()
def fake_load(image_size):
seen.append(image_size)
return sentinel
original = finetune_identity.load_dinov2
finetune_identity.load_dinov2 = fake_load
try:
model, device = finetune_identity.build_model(224)
finally:
finetune_identity.load_dinov2 = original
self.assertIs(model, sentinel)
self.assertIn(device, ("cpu", "mps"))
self.assertEqual(seen, [224])
image = np.zeros((10, 20, 3), dtype=np.uint8)
self.assertEqual(finetune_identity.prepare(image, 32).shape, gallery_module._prepare(image, 32).shape)
self.assertTrue(np.allclose(
finetune_identity.prepare(image, 32),
gallery_module._prepare(image, 32),
))
if __name__ == "__main__":
unittest.main()

View file

@ -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):

View file

@ -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):

View file

@ -45,6 +45,21 @@ class RecorderTests(unittest.TestCase):
self.assertEqual([line["frame"] for line in lines], [0, 1, 2])
self.assertEqual(lines[0]["tracks"][0]["tag_identity"]["sku_id"], "sku_l05")
def test_the_active_backend_is_written_on_each_line(self):
recorder = FrameRecorder(self.root, every=1, identity_backend="gallery")
recorder.write("session-d", 0, b"x", a_result("sku_l05"))
recorder.write("session-d", 1, b"y", a_result())
with open(os.path.join(self.root, "session-d", "index.jsonl"), encoding="utf-8") as handle:
lines = [json.loads(line) for line in handle]
self.assertEqual([line["identity_backend"] for line in lines], ["gallery", "gallery"])
def test_an_unspecified_backend_is_omitted(self):
recorder = FrameRecorder(self.root, every=1)
recorder.write("session-e", 0, b"x", a_result())
with open(os.path.join(self.root, "session-e", "index.jsonl"), encoding="utf-8") as handle:
line = json.loads(handle.readline())
self.assertNotIn("identity_backend", line)
def test_sampling_keeps_the_frame_numbering_visible(self):
recorder = FrameRecorder(self.root, every=3)
for frame_id in range(9):

View file

@ -38,6 +38,7 @@ Then run the app and scan. Each scan session becomes its own directory:
000001.jpg
index.jsonl one line per recorded frame:
{"frame": 123, "path": "000001.jpg", "at": 1699...,
"identity_backend": "gallery",
"summary": {...},
"tracks": [{"box": [...], "identity": {...},
"tag_identity": {...}, "tag": {...}}]}
@ -97,9 +98,15 @@ predictions. For the second visit, run from `recognization` with the same frozen
SESSION="$HOME/dorayaki-data/storeA-visit2/REPLACE_WITH_SESSION"
YOLO_OFFLINE=1 .venv/bin/python code/eval_live.py "$SESSION" \
--stock-dir stock-lopia --labels "$SESSION/labels.json" \
--replay --identity gallery --device cpu
--replay --device cpu
```
A recording written by the current server names `identity_backend` on each
index line, and the evaluator uses that. `--identity gallery` (or `matcher`)
is required when the index omits it, including recordings from before that
field existed, and must match the recorded value when both are present. It
does not silently score the matcher.
Replace `REPLACE_WITH_SESSION` with the actual session name. The evaluator replays
recorded JPEGs through `LiveSession`, writes `eval_live.json` in that directory,
and reports backend, loaded head, weight/catalog fingerprints, decision counts,