vnguyen/product-recognition-demo
1
0
Fork 1
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

fix: include latest shelf tag in live inspection

Browse source
This commit is contained in:
kiyo9w 2026-09-28 13:11:06 +09:00
parent 2894979a35
commit 06f9fb4819
2 changed files with 65 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

28
recognization/code/live.py
View file

@ -191,6 +191,10 @@ class LiveSession:
self._shift_map = None
self._compliance_tracker = None
self._last_compliance: Dict[str, Any] = {"violations": 0, "by_rule": {}, "rows": []}
# Settled tag + reconciliation from the last result, keyed by track.
# Inspect copies these; it does not re-run OCR.
self._settled_tags: Dict[int, Dict[str, Any]] = {}
self._settled_tag_seq = 0
self._label_memory: "collections.deque[Dict[str, Any]]" = collections.deque(
maxlen=LABEL_MEMORY_MAX
)
@ -293,6 +297,8 @@ class LiveSession:
self._latest_frame_id = None
self._frame_origin = [0.0, 0.0]
self._shift_map = None
self._settled_tags = {}
self._settled_tag_seq = 0
# ------------------------------------------------------------------ frame path
def process_frame(self, frame_id: int, payload: bytes) -> Dict[str, Any]:
@ -439,6 +445,17 @@ class LiveSession:
if tag_summary is not None:
summary["price_tags"] = tag_summary["tags"]
summary["compliance"] = tag_summary["compliance"]
settled: Dict[int, Dict[str, Any]] = {}
for payload in track_payload:
snapshot: Dict[str, Any] = {}
if "price_tag" in payload:
snapshot["price_tag"] = payload["price_tag"]
if "reconciliation" in payload:
snapshot["reconciliation"] = payload["reconciliation"]
if snapshot:
settled[int(payload["track_id"])] = snapshot
self._settled_tags = settled
self._settled_tag_seq = seq
self._submit_classification(submissions, method, generation)
self._run_tag_passes(image, tag_bands)
@ -912,7 +929,16 @@ class LiveSession:
if track is None or image is None:
raise KeyError("TRACK_NOT_IN_VIEW")
box = list(track.raw_box)
return self.engine.inspect(image, box, method=chosen_method)
# Snapshot is the frame current at tap time. engine.inspect runs
# outside the lock, so a newer frame may land first; the reply still
# carries the tag from the frame the tap matched, not that later one.
settled = None
if self._settled_tag_seq == self._latest_seq:
settled = self._settled_tags.get(track_id)
result = self.engine.inspect(image, box, method=chosen_method)
if settled:
result.update(settled)
return result
# ------------------------------------------------------------------ catalog cache
def refresh_catalog(self) -> None: