tngo
0cf64a18bf
The live path abstained too much: a detected product whose best candidate missed the accept line was left blank, so a shelf where 59% of in-catalog facings were shown read as broken. Show the best candidate immediately as `provisional`, keep the track re-classifiable so a later attempt upgrades or corrects it, and never blank a label already on screen. - identity.py / identity_lanes.py: every decision now carries `provisional_sku` - the row the evidence supports well enough to show, separate from `best_sku`, which still means "cleared the gate". The accept line does not move: far-unknown false accepts stay at 0.066%. - live.py: a below-gate candidate labels the track with `identity_level: provisional` and `confirmed = False`, so the planner and the crop lane keep offering it. Accepted labels stay frozen. Inherited provisional labels stay re-classifiable. - Clients render provisional distinctly (amber chip, dashed box) via `LiveScanOverlayPolicy.isProvisional(identity)`, which now reads the wire level instead of only the state. - build_stock_all.py merges stock, stock-lopia, stock-mediya and stock-milk150 into stock-all (111 images, 60 SKUs). The server could only read one folder, so anything outside it came back blank. stock-demo is byte-identical to mediya+milk150 and is not a source. Measured on add/image6.jpg at live resolution: in-catalog facings shown 59.4% -> 86.0%, correct shown 38 -> 55, precision 97.4% -> 87.3%. On the wire at 8fps 0.5x: 65 of 79 tracks labelled, 742 accepted + 309 provisional observations, first label 1.27s. docs/research/2026-09-29-shelf-intelligence/ carries the experiments and the decision: group signals (combined AUC 0.991 sibling / 0.990 unrelated), OCR scheduling (group representatives cut calls 4.4x and wall 3.3x), and the finding that group propagation damages more labels than it rescues on this shelf, so groups schedule compute rather than overwrite labels. 331 tests pass.
70 lines
4.5 KiB
Python
70 lines
4.5 KiB
Python
"""Actual eight-worker local/OCR calls; fixed crops, one variable per arm."""
|
|
from pathlib import Path
|
|
import json
|
|
import sys
|
|
import time
|
|
import statistics
|
|
import threading
|
|
import concurrent.futures
|
|
import os
|
|
import psutil
|
|
|
|
ROOT=Path(__file__).resolve().parents[3]
|
|
sys.path.insert(0,str(ROOT/'code'))
|
|
import cv2
|
|
from identity_lanes import LocalLane,TextLane
|
|
from ocr import build_ocr_engine
|
|
from eval_identity import _catalog_rows
|
|
DIR=Path(__file__).resolve().parent; OUT=DIR/'evidence'
|
|
|
|
|
|
def percentile(v,n):
|
|
return round(sorted(v)[min(len(v)-1,int((len(v)-1)*n))],2)
|
|
|
|
|
|
def main():
|
|
cv2.setNumThreads(1)
|
|
image=cv2.imread(str(ROOT/'add/image6.jpg'))
|
|
scale=1280/max(image.shape[:2]);image=cv2.resize(image,(round(image.shape[1]*scale),round(image.shape[0]*scale)),interpolation=cv2.INTER_AREA)
|
|
evidence=json.loads((OUT/'observations.json').read_text());rows=evidence['resolutions']['live']['observations']
|
|
schedule=json.loads((OUT/'ocr-ablation.json').read_text())['arms']['live']
|
|
crops={o['id']:image[o['box'][1]:o['box'][3],o['box'][0]:o['box'][2]] for o in rows}
|
|
local=LocalLane(str(ROOT/'stock-lopia-clean'))
|
|
text=TextLane.from_catalog(_catalog_rows(str(ROOT/'stock-lopia-clean')),str(ROOT/'stock-lopia-clean'),build_ocr_engine('vision'),lock=threading.Lock())
|
|
report={}
|
|
local.rank(next(iter(crops.values())))
|
|
text.read(next(iter(crops.values())))
|
|
# These benchmarks do not decode, detect, track or transmit; they isolate
|
|
# classifier worker contention with the same 109 observed shelf crops.
|
|
modes=[('every_attempt',2,lambda i,a:True),('once_per_track',2,lambda i,a:a==0),('ambiguous_track',2,lambda i,a:a==0 and i in schedule['ambiguous_track']['ocr_ids']),('group_representative',2,lambda i,a:a==0 and i in schedule['group_representative']['ocr_ids'])]
|
|
proc=psutil.Process()
|
|
for name,attempts,ocr_on in modes:
|
|
tasks=[(i,a,ocr_on(i,a)) for a in range(attempts) for i in crops]
|
|
durations=[];first={};done={};ocr_count=0
|
|
cpu0=proc.cpu_times();started=time.perf_counter()
|
|
def work(item):
|
|
i,a,use_ocr=item;start=time.perf_counter();rank=local.rank(crops[i]);read=text.read(crops[i]) if use_ocr else ''
|
|
return i,a,round((time.perf_counter()-start)*1000,2),bool(read)
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
for f in concurrent.futures.as_completed([pool.submit(work,item) for item in tasks]):
|
|
i,a,elapsed,read=f.result();now=(time.perf_counter()-started)*1000
|
|
durations.append(elapsed);first.setdefault(i,now);done[i]=now;ocr_count+=bool(read)
|
|
wall=time.perf_counter()-started;cpu1=proc.cpu_times();cpu=cpu1.user+cpu1.system-cpu0.user-cpu0.system
|
|
report[name]={'tasks':len(tasks),'ocr_invocations':sum(int(k) for _,_,k in tasks),'ocr_nonempty_reads':ocr_count,'worker_ms_p50':percentile(durations,.5),'worker_ms_p95':percentile(durations,.95),'first_facing_ms_p50':percentile(list(first.values()),.5),'all_first_facings_ms':round(max(first.values()),1),'wall_s':round(wall,3),'throughput_tasks_s':round(len(tasks)/wall,1),'cpu_s':round(cpu,2),'first_results':{str(i):round(v,1) for i,v in first.items()}}
|
|
print(name,report[name]['wall_s'],report[name]['all_first_facings_ms'],report[name]['ocr_invocations'],flush=True)
|
|
(OUT/'scheduler-benchmark.json').write_text(json.dumps(report,indent=2))
|
|
# Separate visual completion from serialized text: no OCR lock occupies a
|
|
# local worker. Measure evidence arrival, not phone rendering or inference.
|
|
started=time.perf_counter();events=[]
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
pending={pool.submit(local.rank,crop):i for i,crop in crops.items()}
|
|
for future in concurrent.futures.as_completed(pending):
|
|
events.append({'id':pending[future],'kind':'local','ms':(time.perf_counter()-started)*1000,'rank':future.result()})
|
|
visual_done=(time.perf_counter()-started)*1000
|
|
for i in schedule['ambiguous_track']['ocr_ids']:
|
|
read=text.read(crops[i])
|
|
events.append({'id':i,'kind':'text','ms':(time.perf_counter()-started)*1000,'read':read,'text':text.scores_from_text(read)})
|
|
(OUT/'anytime-events.json').write_text(json.dumps({'provenance':evidence['provenance'],'visual_done_ms':visual_done,'complete_ms':(time.perf_counter()-started)*1000,'events':events},ensure_ascii=False,indent=2))
|
|
print('visual_first',round(visual_done,1),'ms to all visual results',flush=True)
|
|
|
|
if __name__=='__main__':main()
|