"""CPU-only review clips, same sampled frame and prediction, no GT overlays.""" import csv import hashlib import json import math import pathlib import textwrap from fractions import Fraction import av import numpy as np from PIL import Image,ImageDraw,ImageFont ROOT=pathlib.Path('/eval') rows=json.loads((ROOT/'inputs/pose.json').read_text()) font='/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf' fonts={n:ImageFont.truetype(font,n) for n in (20,22,24,28,32,36)} names={'stgcnpp':'ST-GCN++ / pair classification', 'posec3d':'PoseC3D / whole-scene classification', 'ava_slowonly':'AVA SlowOnly / person classification', 'yowov2':'YOWOv2 / box classification'} palette=['#00c2ff','#ffd166','#c593ff','#51daa8','#ff96b3','#c2e86b','#ffad66'] edges=[(5,6),(5,7),(7,9),(6,8),(8,10),(5,11),(6,12),(11,12),(11,13),(13,15),(12,14),(14,16)] validation=json.loads((ROOT/'reports/render_validation.json').read_text())['clips'] validation['posec3d']['file']='camera24_posec3d_top5.mp4' overflow=[] def text(draw,xy,string,size=24,fill='#e9edf1'): draw.text(xy,str(string),font=fonts[size],fill=fill) def identity_label(draw,x,y,label,color,occupied): width=math.ceil(draw.textlength(label,font=fonts[22]))+12 for step in range(30): bx=max(2,min(1918-width,x+(step//8)*125)) by=max(112,y-32-(step%8)*32) rect=(bx,by,bx+width,by+29) if not any(rect[0]o[0] and rect[1]o[1] for o in occupied): occupied.append(rect) draw.line((bx+width/2,by+29,x,y),fill=color,width=1) draw.rectangle(rect,fill='#111820') text(draw,(bx+5,by),label,22,color) return raise RuntimeError('No non-overlapping identity label position') for name,title in names.items(): if name=='posec3d': continue report=json.loads((ROOT/'reports'/(name+'_summary.json')).read_text()) if report['status']!='PASS': continue results=json.loads((ROOT/'outputs'/(name+'.json')).read_text()) selected=[r for r in results if 242<=r['source_frame']<=723] assert selected and all(rows[r['sample']]['source_frame']==r['source_frame'] for r in selected) max_items=max(len(r['items']) for r in selected) card_rows=max(1,math.ceil(max_items/2)) width=3200; height=max(1440,120+card_rows*256+50) height+=height%2 output=ROOT/'outputs'/('camera24_'+name+'_top5_v2.mp4') assert not output.exists(), 'Preserve prior rendered video' container=av.open(str(output),'w',options={'movflags':'+faststart'}) stream=container.add_stream('libx264',rate=10) stream.width=width;stream.height=height;stream.pix_fmt='yuv420p' stream.options={'crf':'20','preset':'veryfast'} stream.codec_context.thread_count=2 stream.time_base=Fraction(1,90000) stream.codec_context.time_base=Fraction(1,90000) first=selected[0]['seconds'] maps=[] for index,r in enumerate(selected): frame=rows[r['sample']] canvas=Image.new('RGB',(width,height),'#111820') image=Image.open(ROOT/'inputs/frames'/f"{r['sample']:04d}.png").convert('RGB') canvas.paste(image,(0,110)) d=ImageDraw.Draw(canvas) occupied=[] text(d,(24,16),title,36) text(d,(24,63),f"Original Frame {r['source_frame']} | Time {r['seconds']:.3f}s | camera24 | same-frame output",28) if name in ('stgcnpp','ava_slowonly','posec3d'): for person in frame['persons']: tid=person['track_id']; label='T'+str(tid) if tid is not None else 'Uq'+str(person['query']) color=palette[(int(tid or person['query'])//80)%len(palette)] x1,y1,x2,y2=person['bbox'] if name!='posec3d': d.rectangle((max(0,x1),max(110,y1+110),min(1919,x2),min(1189,y2+110)),outline=color,width=3) identity_label(d,x1,y1+110,label,color,occupied) if name in ('stgcnpp','posec3d'): kp=person['keypoints'] for a,b in edges: if min(kp[a][2],kp[b][2])>=.3: d.line((kp[a][0],kp[a][1]+110,kp[b][0],kp[b][1]+110),fill=color,width=2) elif name=='yowov2': for item in r['items']: x1,y1,x2,y2=item['bbox'];color='#00c2ff' d.rectangle((x1,y1+110,x2,y2+110),outline=color,width=3) identity_label(d,x1,y1+110,item['display_id'],color,occupied) text(d,(1940,20),'TOP 5 ORIGINAL ACTION CLASSES',28) text(d,(1940,61),'Scores are not comparable across models.',24,'#bbc6d3') if not r['items']: text(d,(1940,135),'NO ELIGIBLE TRACK PAIR / NO OUTPUT',28,'#ffd166') text(d,(1940,180),'This is not a normal-action decision.',24) for j,item in enumerate(r['items']): col=j%2;line=j//2;x=1940+col*625;y=120+line*256 d.rectangle((x-8,y-5,x+605,y+241),outline='#475362',width=1) if name=='stgcnpp': heading='Pair '+' + '.join('T'+str(t) for t in item['ids']) status='Observed frames: '+' / '.join(f'{int(round(c*100))}/100' for c in item.get('coverage',[])) elif name=='ava_slowonly': heading=('T'+str(item['track_id'])) if item['track_id'] is not None else ('Uq'+str(item['query'])+' (untracked)') status=f"Person detection: {item['detection_score']:.3f}" elif name=='yowov2': heading=item['display_id']+' (frame-local label)' status=f"Objectness: {item['objectness']:.3f} (not action score)" else: heading='Whole scene / all detected people' status='No actor attribution' text(d,(x,y),heading,24,'#00c2ff') text(d,(x,y+32),status,20,'#ffd166' if item['status']=='PARTIAL_TRACK' else '#bbc6d3') yy=y+63 for rank,t in enumerate(item.get('top5',[]),1): label=f"{rank}. {t['name']}" lines=textwrap.wrap(label,width=40,break_long_words=False) or [''] for k,s in enumerate(lines): if d.textlength(s,font=fonts[20])>495: overflow.append((name,r['sample'],s,'width')) text(d,(x,yy),s,20) if k==0: text(d,(x+505,yy),f"{t['score']:.3f}",20,'#ffffff') yy+=24 if yy>y+238: overflow.append((name,r['sample'],j,'height')) win=r['window'] text(d,(24,1210),f"Input window: {win['start_seconds']:.3f}s - {win['end_seconds']:.3f}s | {len(win['samples'])} observations",28) text(d,(24,1252),f"10 FPS observations/results | Future input: {'YES (offline)' if win['end_seconds']>r['seconds'] else 'NO'} | Loop cuts crossed: {win.get('known_loop_boundaries_crossed',[])}",24) if name=='stgcnpp': note='PARTIAL means missing track observations; no guessed identity stitching. Uq = untracked detection.' elif name=='posec3d': note='Skeleton heatmaps include all people; Top 5 does not identify who performed an action.' elif name=='ava_slowonly': note='Compatibility adapter: native MMCV numerical equivalence is unverified. Uq = untracked detection.' else: note='Y labels reset every frame, not Tracking IDs. Objectness and action sigmoid are separate.' text(d,(24,1293),note,22,'#ffd166') text(d,(24,1332),'No fight-class mapping or event decision. Colors identify inputs only. Same loop repeated, not independent cases.',22) pts=round((r['seconds']-first)*90000) vf=av.VideoFrame.from_ndarray(np.asarray(canvas),format='rgb24') vf.pts=pts;vf.time_base=Fraction(1,90000) for packet in stream.encode(vf): container.mux(packet) maps.append(dict(output_frame=index+1,source_frame=r['source_frame'],source_seconds=r['seconds'],output_pts_seconds=pts/90000,sample=r['sample'])) if r['sample'] in (101,150,200,250): canvas.save(ROOT/'reports'/(name+f"_v2_sample{r['sample']}.jpg"),quality=92) for packet in stream.encode():container.mux(packet) container.close() with (ROOT/'outputs'/(name+'_v2_frame_map.csv')).open('w') as f: writer=csv.DictWriter(f,fieldnames=list(maps[0]));writer.writeheader();writer.writerows(maps) decoded=[] with av.open(str(output)) as check: for f in check.decode(video=0): decoded.append(float(f.pts*f.time_base)) assert len(decoded)==len(maps) delta=max(abs(a-b['output_pts_seconds']) for a,b in zip(decoded,maps)) assert delta<.001 validation[name]=dict(frame_count=len(maps),source_first=maps[0]['source_frame'],source_last=maps[-1]['source_frame'], width=width,height=height,max_pts_error_seconds=delta,decode='PASS',bytes=output.stat().st_size, sha256=hashlib.sha256(output.read_bytes()).hexdigest(),file=output.name) print(json.dumps({name:validation[name]}),flush=True) assert not overflow, str(overflow[:10]) (ROOT/'reports/render_validation_r02.json').write_text(json.dumps(dict(clips=validation,text_overflow=overflow,identity_label_overlap=0),indent=2))