"""Exercise the complete LiteRT cascade against frozen L4 reference outputs.""" from pathlib import Path import hashlib,io,json,time,zipfile import numpy as np import torch from inference import load_model,cascade from PIL import Image from scipy.optimize import linear_sum_assignment from cascade_litert import Cascade ROOT=Path('/content/drive/MyDrive/FireViewer/trainv2.5final1') OUT=ROOT/'litert/20261007' def match(a,b,key): assert len(a)==len(b),('Output count differs',len(a),len(b),key) if not a:return [] aa=np.asarray([key(x) for x in a]);bb=np.asarray([key(x) for x in b]) cost=np.abs(aa[:,None,:]-bb[None,:,:]).mean(2) _,order=linear_sum_assignment(cost) return [b[i] for i in order] def run(): selection=json.loads((ROOT/'cascade_benchmark/natural-selection.json').read_text()) records=sorted(selection['records'],key=lambda x:x['id']) chosen=[] for group in ('flame','smoke','negative'): families=set() for r in records: if r['group']!=group or r['visual_family_id'] in families:continue families.add(r['visual_family_id']);chosen.append(r) if len(families)==2:break references={} for line in (ROOT/'cascade_benchmark/results-20261006-234144/predictions.jsonl').read_text().splitlines(): value=json.loads(line);references[value['id']]=value['output'] recipe=dict(ids=[x['id'] for x in chosen],selection='First two sorted unique visual families per stratum; no prediction-dependent selection', tolerances=dict(normalized_box_atol=.001,score_atol=.002,normalized_point_atol=.005,rtol=.001), semantics='numerical routing/regression, not a new independent benchmark',device='CPU LiteRT vs same-CPU PyTorch',test_used=False) (OUT/'cascade-parity-plan.json').write_text(json.dumps(recipe,indent=2)) torch.set_num_threads(4);torch.set_grad_enabled(False) release=json.loads((ROOT/'cascade_benchmark/release-manifest.json').read_text()) native={role:load_model(Path('/content/fireviewer-litert-execution/models')/m['repository'].split('/')[-1],'cpu') for role,m in release['models'].items()} for role in ('proposal','visible'):native[role][0].model.model.eval() model=Cascade(OUT,threads=4) checks=[];predictions=[];began=time.perf_counter() with zipfile.ZipFile('/content/fireviewer-litert-execution/natural-selected.zip') as packed: for r in chosen: payload=packed.read(r['member']);assert hashlib.sha256(payload).hexdigest()==r['sha256'] image=Image.open(io.BytesIO(payload)).convert('RGB');scale=np.array([image.width,image.height]*2) started=time.perf_counter();actual=model(image);elapsed=time.perf_counter()-started expected=cascade(image,native['proposal'],native['visible'],native['point'],native['smoke'],device='cpu') gpu=references[r['id']] gpu_comparison=dict(flame_count_reference=len(gpu['visible_flames']),flame_count_litert=len(actual['visible_flames']), smoke_count_reference=len(gpu['smoke_compatibility']),smoke_count_litert=len(actual['smoke_compatibility']), image_alarm_reference=bool(gpu['visible_flames']) or any(x['status']=='fire_smoke_probable' for x in gpu['smoke_compatibility']), image_alarm_litert=bool(actual['visible_flames']) or any(x['status']=='fire_smoke_probable' for x in actual['smoke_compatibility']), normalized_boxes=[],normalized_points=[],smoke_probability_errors=[],strict_cross_backend_numeric_pass=True) try: gpu_flames=match(gpu['visible_flames'],actual['visible_flames'],lambda x:x['bbox_xyxy']) for e,a in zip(gpu['visible_flames'],gpu_flames): eb=np.asarray(e['bbox_xyxy'])/scale;ab=np.asarray(a['bbox_xyxy'])/scale gpu_comparison['normalized_boxes'].append(float(np.abs(eb-ab).max())) gpu_comparison['strict_cross_backend_numeric_pass'] &= bool(np.allclose(eb,ab,atol=.001,rtol=.001)) and bool(np.isclose(e['score'],a['score'],atol=.002,rtol=.001)) ep=np.asarray(e['point']['point_xy_original'])/np.array(image.size);ap=np.asarray(a['point']['point_xy_original'])/np.array(image.size) gpu_comparison['normalized_points'].append(float(np.abs(ep-ap).max())) gpu_comparison['strict_cross_backend_numeric_pass'] &= bool(np.allclose(ep,ap,atol=.005,rtol=.001)) gpu_smoke=match(gpu['smoke_compatibility'],actual['smoke_compatibility'],lambda x:x['proposal_bbox_xyxy']) for e,a in zip(gpu['smoke_compatibility'],gpu_smoke): error=abs(e['probability']-a['probability']);gpu_comparison['smoke_probability_errors'].append(error) gpu_comparison['strict_cross_backend_numeric_pass'] &= e['status']==a['status'] and bool(np.isclose(e['probability'],a['probability'],atol=.002,rtol=.001)) except AssertionError as error: gpu_comparison['strict_cross_backend_numeric_pass']=False;gpu_comparison['difference']=str(error) print('L4_VS_CPU_LITERT_DIAGNOSTIC',json.dumps(dict(id=r['id'],**gpu_comparison)),flush=True) (OUT/('cascade-parity-debug-'+r['id']+'.json')).write_text(json.dumps(dict(id=r['id'],expected=expected,actual=actual),indent=2)) flames=match(expected['visible_flames'],actual['visible_flames'],lambda x:x['bbox_xyxy']) box_max=score_max=point_max=0 for e,a in zip(expected['visible_flames'],flames): eb=np.asarray(e['bbox_xyxy'])/scale;ab=np.asarray(a['bbox_xyxy'])/scale box_max=max(box_max,float(np.abs(eb-ab).max()));assert np.allclose(eb,ab,atol=.001,rtol=.001) score_max=max(score_max,abs(e['score']-a['score']));assert np.isclose(e['score'],a['score'],atol=.002,rtol=.001) ep=np.asarray(e['point']['point_xy_original'])/np.array(image.size) ap=np.asarray(a['point']['point_xy_original'])/np.array(image.size) point_max=max(point_max,float(np.abs(ep-ap).max()));assert np.allclose(ep,ap,atol=.005,rtol=.001) smoke=match(expected['smoke_compatibility'],actual['smoke_compatibility'],lambda x:x['proposal_bbox_xyxy']) smoke_max=0 for e,a in zip(expected['smoke_compatibility'],smoke): assert e['status']==a['status'],('Smoke triage changed',r['id'],e['status'],a['status']) smoke_max=max(smoke_max,abs(e['probability']-a['probability'])) assert np.isclose(e['probability'],a['probability'],atol=.002,rtol=.001),dict(id=r['id'],expected_smoke=e,actual_smoke=a) assert a['source_location'] is None check=dict(id=r['id'],group=r['group'],flames=len(flames),smoke_ROIs=len(smoke), normalized_box_max=box_max,score_max=score_max,normalized_point_max=point_max, smoke_probability_max=smoke_max,latency_seconds=elapsed,status='passed',l4_cross_backend=gpu_comparison) checks.append(check);predictions.append(dict(id=r['id'],output=actual)) print('CASCADE_LITERT_PARITY',json.dumps(check),flush=True) result=dict(status='passed',samples=len(checks),plan=recipe,checks=checks, total_seconds=time.perf_counter()-began,reference_backend='Frozen source weights in eager PyTorch FP32 on the same CPU', l4_reference_backend='Original 680-image PyTorch FP32 benchmark on L4', l4_exact_numerical_parity=all(x['l4_cross_backend']['strict_cross_backend_numeric_pass'] for x in checks), l4_alarm_decisions_equal=all(x['l4_cross_backend']['image_alarm_reference']==x['l4_cross_backend']['image_alarm_litert'] for x in checks), backend='CPU LiteRT',full_680_litert_semantic_rebenchmark=False, stage_calls_present=dict(visible=any(x['flames'] for x in checks),point=any(x['flames'] for x in checks),smoke=any(x['smoke_ROIs'] for x in checks)), android_device_tested=False,test_used=False) assert all(result['stage_calls_present'].values()),'All branches must be exercised' (OUT/'cascade-parity.json').write_text(json.dumps(result,indent=2)) (OUT/'cascade-parity-predictions.jsonl').write_text('\n'.join(json.dumps(x) for x in predictions)+'\n') print('FULL_LITERT_CASCADE_PARITY_PASSED',len(checks),flush=True) if __name__=='__main__':run()