"""Independent saved-score/count/map audit; no semantic image analysis.""" import csv import hashlib import json import pathlib from collections import Counter from fractions import Fraction import av import numpy as np from PIL import Image, ImageOps ROOT = pathlib.Path('/download') SOURCE = pathlib.Path('/source') EXPECTED = json.loads(pathlib.Path('/expected.json').read_text()) LIMITS = {23:(.00003,.00006,.0001),49:(.00006,.0001,.0002),50:(.0001,.0002,.0004),51:(.002,.004,.008)} VARIANTS = {'filtered_ge2':2,'filtered_ge3':3} results = []; inventory = []; n_all = n_valid = 0 for folder in sorted(p for p in ROOT.iterdir() if p.is_dir() and p.name[:2].isdigit()): clip = folder.name; data = folder/'data' raw = json.loads((data/'posec3d_raw.json').read_text()) decisions = json.loads((data/'filter_decisions.json').read_text()) rows = json.loads((SOURCE/'clips'/clip/'inputs/pose.json').read_text()) assert len(raw) == len(decisions) plan=json.loads((SOURCE/'clips'/clip/'inputs/sampling_plan.json').read_text()) pose_summary=json.loads((SOURCE/'clips'/clip/'reports/pose_summary.json').read_text()) model_manifest=json.loads((ROOT/'00_model_manifest.json').read_text()) reference=model_manifest['current_trial']['pose_reference'] assert pose_summary['status']=='PASS' assert pose_summary['engine_sha256']==reference['engine_sha256'] assert pose_summary['source_files']==reference['source_files'] assert pose_summary['tracking']==reference['tracking'] assert pose_summary['person_threshold']==.15 and pose_summary['joint_threshold']==.3 assert [r['sample'] for r in raw]==plan['display_samples'] assert len(rows)==len(plan['target_seconds']) assert [r['source_frame'] for r in rows]==plan['source_frame_indices'] assert all(r['seconds']==t for r,t in zip(rows,plan['relative_seconds'])) assert len(set(plan['source_frame_indices']))==len(rows) model_summary=json.loads((data/'posec3d_summary.json').read_text()) assert model_summary['temporal_samples']==16 and model_summary['observation_fps']==10 assert model_summary['evaluation_samples']==plan['display_samples'] for src,name in [(SOURCE/'clips'/clip/'inputs/sampling_plan.json','sampling_plan.json'), (SOURCE/'clips'/clip/'inputs/source_map.csv','source_frame_map.csv'), (SOURCE/'clips'/clip/'reports/pose_summary.json','pose_summary.json'), (SOURCE/'clips'/clip/'reports/source_probe.json','source_probe.json')]: (data/name).write_bytes(src.read_bytes()) for filename in ('posec3d_raw.json','posec3d_summary.json','classes_posec3d.csv'): assert hashlib.sha256((data/filename).read_bytes()).hexdigest() == EXPECTED[f'{clip}/data/{filename}'] counts = []; states = []; valid = 0; ge2 = ge3 = 0 for r, d in zip(raw,decisions): assert all(r[k] == d[k] for k in ('sample','source_frame','seconds')) row = rows[r['sample']] assert plan['display_bounds'][0]<=r['source_frame']<=plan['display_bounds'][1] assert row['sample']==r['sample'] for p in row['persons']: assert p['score']>=.15 and len(p['keypoints'])==17 and len(p['bbox'])==4 kp=np.asarray(p['keypoints']) assert kp.shape==(17,3) and np.isfinite(kp).all() and np.isfinite(p['bbox']).all() assert p['valid_joints']==int((kp[:,2]>=.3).sum()) if r['window'] is not None: ids=list(range(r['sample']-8,r['sample']+8)) assert r['window']['samples']==ids and min(ids)>=0 and max(ids)=len(rows) assert (row['source_frame'],row['seconds']) == (r['source_frame'],r['seconds']) items = [it for it in r['items'] if it.get('scores') is not None] if items: assert len(items)==1 and len(items[0]['scores'])==60 scores = items[0]['scores']; valid += 1 assert np.isfinite(scores).all() and all(0<=v<=1 for v in scores) assert abs(sum(scores)-1)<1e-5 and items[0]['input_shape']==[1,1,17,16,64,64] flags = {k:[scores[k]>=t for t in ts] for k,ts in LIMITS.items()} passing = [k for k,v in flags.items() if v[1]]; count = len(passing) assert d['valid'] and d['state']=='VALID' and d['count']==count and d['passed_ids']==passing for k, flags_for_k in flags.items(): entry=d['classes'][str(k)] assert entry['score']==scores[k] assert [entry[t] for t in ('low','basic','high')]==flags_for_k assert d['ge2']==(count>=2) and d['ge3']==(count>=3) ge2 += count>=2; ge3 += count>=3; states.append('VALID'); counts.append(count) else: s = [r.get('status')] + [it.get('status') for it in r['items']] state='INPUT_INSUFFICIENT' if 'INSUFFICIENT_WINDOW' in s else 'NO_VALID_SKELETON' if 'NO_VALID_SKELETON' in s else 'NO_SCORES' assert not d['valid'] and d['state']==state and d['count'] is None and d['ge2'] is None and d['ge3'] is None assert not d['classes']; states.append(state); counts.append(None) assert ge3 <= ge2 <= valid <= len(raw) summary=json.loads((data/'filter_summary.json').read_text()) assert (summary['valid'],summary['ge2'],summary['ge3'])==(valid,ge2,ge3) rgb_hashes=json.loads((data/'rgb_sha256.json').read_text()) assert len(rgb_hashes)==len(raw) for filename,digest in rgb_hashes.items(): path=(pathlib.Path('/camera/inputs/frames') if clip=='01_camera24' else SOURCE/'clips'/clip/'inputs/frames')/filename assert hashlib.sha256(path.read_bytes()).hexdigest()==digest maps = []; sample_errors = [] for variant, minimum in VARIANTS.items(): stem='PoseC3D_'+variant; video=folder/(stem+'.mp4') report=json.loads((data/(stem+'_validation.json')).read_text()) assert report['status']=='PASS' and report['frames']==len(raw) and report['text_overlap']==report['text_overflow']==0 assert report['max_pts_error']<1/90000 and report['resolution']==[3200,1800] assert hashlib.sha256(video.read_bytes()).hexdigest()==report['sha256'] with av.open(str(video)) as container: stream=container.streams.video[0] assert (stream.width,stream.height)==(3200,1800) and stream.frames==len(raw) mapping=list(csv.DictReader((data/(stem+'_frame_map.csv')).open())); assert len(mapping)==len(raw) base_map=[] for i,(r,m) in enumerate(zip(raw,mapping)): state=states[i] if counts[i] is None else 'SIGNALS' if minimum is None else 'PASS' if counts[i]>=minimum else 'NO_MATCH' shown=state in ('SIGNALS','PASS') assert int(m['output_frame'])==i+1 and int(m['sample'])==r['sample'] and int(m['source_frame'])==r['source_frame'] assert float(m['source_seconds'])==r['seconds'] and m['state']==state and m['detail_visible']==str(shown) assert m['basic_count']==('' if counts[i] is None else str(counts[i])) assert float(m['output_pts_seconds'])==round((r['seconds']-raw[0]['seconds'])*90000)/90000 base_map.append(tuple(m[k] for k in ('output_frame','sample','source_frame','source_seconds','output_pts_seconds'))) maps.append(base_map) assert Counter(m['state'] for m in mapping)==report['state_counts'] # Compare only unchanged RGB area, not scene semantics or model correctness. index=report['sample_indices'][0]; r=raw[index] rgb=(pathlib.Path('/camera/inputs/frames') if clip=='01_camera24' else SOURCE/'clips'/clip/'inputs/frames')/f"{r['sample']:04d}.png" with Image.open(rgb) as image: fit=ImageOps.contain(image.convert('RGB'),(1856,1044)) canvas=Image.new('RGB',(1856,1044),'black');canvas.paste(fit,((1856-fit.width)//2,(1044-fit.height)//2)) reference=np.asarray(canvas).astype(np.float32) with Image.open(data/f'{stem}_sample{index}.jpg') as image: candidate=np.asarray(image.crop((20,150,1876,1194))).astype(np.float32) mae=float(np.mean(np.abs(reference[8:-8,8:-8]-candidate[8:-8,8:-8]))) assert mae < 12, (clip,variant,mae) sample_errors.append(dict(variant=variant,rgb_mean_absolute_error=mae)) inventory.append(dict(input=clip,file=f'{clip}/{stem}.mp4',frames=len(raw),valid=valid, passed='' if minimum is None else ge2 if minimum==2 else ge3, first_source_frame=raw[0]['source_frame'],last_source_frame=raw[-1]['source_frame'], source_first_pts=raw[0]['seconds'],source_last_pts=raw[-1]['seconds'],sha256=report['sha256'])) assert maps[0]==maps[1] results.append(dict(clip=clip,observations=len(raw),valid=valid,excluded=len(raw)-valid,ge2=ge2,ge3=ge3, ge2_ratio=ge2/valid if valid else None,ge3_ratio=ge3/valid if valid else None, count_histogram=dict(Counter(c for c in counts if c is not None)),rgb_checks=sample_errors,status='PASS')) n_all+=len(raw); n_valid+=valid assert len(results)==4 and len(inventory)==8 and n_all>0 and 0<=n_valid<=n_all source_manifest=EXPECTED for filename in ('00_model_manifest.json','00_source_manifest.json'): assert hashlib.sha256((ROOT/filename).read_bytes()).hexdigest()==source_manifest[filename] with (ROOT/'video_list.csv').open('w',newline='',encoding='utf-8-sig') as f: writer=csv.DictWriter(f,fieldnames=list(inventory[0]));writer.writeheader();writer.writerows(inventory) report=dict(status='PASS',inputs=4,videos=8,observations=n_all,valid=n_valid,output_frames=n_all*2,inputs_detail=results, class_thresholds=LIMITS,checks=['raw hashes','independent threshold/count audit','ge3 subset ge2','missing states', 'same Frame/PTS across variants','full decoding in renderer','text bounding boxes/overlap per frame', 'sample unchanged RGB numerical comparison','source manifests'], human_layout_review=json.loads((ROOT/'validation/layout_review.json').read_text())['status']) (ROOT/'validation/technical_audit.json').write_text(json.dumps(report,indent=2)) fingerprints={str(p.relative_to(ROOT)):hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(ROOT.rglob('*')) if p.is_file() and p.name!='SHA256SUMS.json'} (ROOT/'SHA256SUMS.json').write_text(json.dumps(fingerprints,indent=2)) print(json.dumps(dict(status='PASS',inputs=4,videos=8,frames=n_all*2,files=len(fingerprints),counts=[(r['clip'],r['valid'],r['ge2'],r['ge3']) for r in results])))