import argparse import ctypes as c import hashlib import json import os import time from pathlib import Path import numpy as np from PIL import Image, ImageDraw, ImageFont ROOT = Path('/evidence') FRAMES = 200 FRAME_BYTES = 1280*720*3 THRESHOLD = .5 os.umask(0o077) def infer(precision): plans = { 'fp32': ('/models/fp32.plan','2f80bc327d156c4ae5d1fa82c1c96b955122a10b581ab0733c7c119a06d1635e'), 'fp16': ('/models/fp16.plan','5fdf1e9c55ab947a3b7d5f7f6bdceaf1e36a9e52efdf86f5a15d3b99b8631d1e'), } plan, expected = plans[precision] assert hashlib.sha256(Path(plan).read_bytes()).hexdigest()==expected assert (ROOT/'frames.rgb').stat().st_size==FRAMES*FRAME_BYTES lib=c.CDLL(str(ROOT/'video_probe.so')) ptr=np.ctypeslib.ndpointer(dtype=np.float32, flags='C_CONTIGUOUS') lib.open_probe.argtypes=[c.c_char_p]; lib.open_probe.restype=c.c_void_p lib.infer_probe.argtypes=[c.c_void_p,ptr,ptr,np.ctypeslib.ndpointer(dtype=np.float64,flags='C_CONTIGUOUS')] lib.infer_probe.restype=c.c_int lib.close_probe.argtypes=[c.c_void_p] start=time.perf_counter(); handle=lib.open_probe(plan.encode()) if not handle: raise RuntimeError('engine load/contract failed') load_seconds=time.perf_counter()-start timing=[]; preprocessing=[]; counts=[]; person_counts=[]; digest=hashlib.sha256() begin=time.perf_counter() try: with (ROOT/'frames.rgb').open('rb') as frames, (ROOT/(precision+'.jsonl')).open('x') as output: for index in range(FRAMES): if time.perf_counter()-begin>280: raise TimeoutError('model window exceeded') raw=frames.read(FRAME_BYTES) assert len(raw)==FRAME_BYTES before=time.perf_counter() # Same RGB/mean/std/NCHW contract as model_tooling.reference; CPU offline resize only. im=Image.frombytes('RGB',(1280,720),raw).resize((576,576),Image.Resampling.BILINEAR) tensor=np.asarray(im,dtype=np.float32)/255 tensor=(tensor-np.array([.485,.456,.406],np.float32))/np.array([.229,.224,.225],np.float32) tensor=np.ascontiguousarray(tensor.transpose(2,0,1)) preprocessing.append((time.perf_counter()-before)*1000) digest.update(tensor.tobytes()) result=np.empty((300,6),np.float32); ms=np.empty(2,np.float64) n=lib.infer_probe(handle,tensor,result,ms) if not 0<=n<=300: raise RuntimeError('inference/finite/decoder failed') detections=result[:n].tolist() visible=[d for d in detections if d[1]>=THRESHOLD] counts.append(len(visible)); person_counts.append(sum(int(d[0])==0 for d in visible)) output.write(json.dumps({'frame':index,'source_time_s':20+index/10, 'detections':detections,'transfer_inference_ms':ms[0], 'finite_decode_ms':ms[1]})+'\n') timing.append(ms[0]) finally: lib.close_probe(handle) report={'precision':precision,'plan_sha256':expected,'frames':FRAMES,'warmup':0, 'tensor_sequence_sha256':digest.hexdigest(),'display_threshold':THRESHOLD, 'engine_load_seconds':load_seconds,'loop_seconds':time.perf_counter()-begin, 'transfer_inference_ms':{'first':timing[0],'median_excluding_first':float(np.median(timing[1:])), 'p95_excluding_first':float(np.percentile(timing[1:],95))}, 'cpu_preprocess_median_ms':float(np.median(preprocessing)), 'visible_detections_total':sum(counts),'visible_person_total':sum(person_counts), 'person_per_frame_min':min(person_counts),'person_per_frame_max':max(person_counts), 'accuracy_acceptance':'USER_REVIEW_WAIT_NO_GROUND_TRUTH'} (ROOT/(precision+'-summary.json')).write_text(json.dumps(report,indent=2)) print(json.dumps(report),flush=True) def render(): from rfdetr.assets.coco_classes import COCO_CLASSES names=[COCO_CLASSES[key] for key in sorted(COCO_CLASSES)] reports=[json.loads((ROOT/(p+'-summary.json')).read_text()) for p in ('fp32','fp16')] assert reports[0]['tensor_sequence_sha256']==reports[1]['tensor_sequence_sha256'] results=[[json.loads(line) for line in (ROOT/(p+'.jsonl')).read_text().splitlines()] for p in ('fp32','fp16')] assert all(len(rows)==FRAMES for rows in results) font=ImageFont.load_default(size=19); small=ImageFont.load_default(size=16) with (ROOT/'frames.rgb').open('rb') as frames, (ROOT/'comparison.rgb').open('xb') as output: for i in range(FRAMES): raw=frames.read(FRAME_BYTES); assert len(raw)==FRAME_BYTES base=Image.frombytes('RGB',(1280,720),raw).resize((960,540),Image.Resampling.BILINEAR) canvas=Image.new('RGB',(1920,608),(22,24,26)); draw=ImageDraw.Draw(canvas) for col,label in enumerate(('FP32 / TF32 OFF','FP16 ENABLED')): rows=results[col][i]; assert rows['frame']==i canvas.paste(base,(col*960,68)) visible=[d for d in rows['detections'] if d[1]>=THRESHOLD] people=sum(int(d[0])==0 for d in visible) draw.text((col*960+16,9),f'{label} | frame {i:03d} | t={20+i/10:.1f}s',font=font,fill='white') draw.text((col*960+16,37),f'Display score >= 0.50 | person: {people} | all: {len(visible)}',font=small,fill=(210,215,220)) for cls,score,x,y,w,h in visible: color=(60,235,145) if int(cls)==0 else (255,196,70) left=max(col*960,min(col*960+959,col*960+round(x*.75))) top=max(68,min(607,68+round(y*.75))) right=max(left,min(col*960+959,col*960+round((x+w)*.75))) bottom=max(top,min(607,68+round((y+h)*.75))) draw.rectangle((left,top,right,bottom),outline=color,width=2) text=f'{names[int(cls)]} {score:.2f}' bounds=draw.textbbox((0,0),text,font=small); tw=bounds[2]+6 tx=min(left,col*960+960-tw); ty=max(68,top-21) draw.rectangle((tx,ty,tx+tw,ty+21),fill=(18,20,22)) draw.text((tx+3,ty+2),text,font=small,fill=color) draw.line((959,0,959,607),fill=(255,255,255),width=2) output.write(canvas.tobytes()) if i==100: canvas.save(ROOT/'comparison-preview.png') print('render complete: 200 aligned frames, identical tensor sequence',flush=True) parser=argparse.ArgumentParser() parser.add_argument('stage',choices=['fp32','fp16','render']) args=parser.parse_args() render() if args.stage=='render' else infer(args.stage)