"""Local JSON-only descriptive statistics. No media/model/runtime access.""" import bisect import collections import csv import datetime import hashlib import itertools import json import math import pathlib import statistics RUN = pathlib.Path(__file__).resolve().parent.parent SOURCE = RUN.parent/'2026-10-27_contextra-short-input-7clips_r01/Contextra_Action_Review_20261027' IDS = (23,49,50,51) PROBS = {'p25':.25,'p50':.5,'p75':.75,'p95':.95} def quantile(values,p): if not values:return None ordered=sorted(values);h=(len(ordered)-1)*p;i=math.floor(h) return ordered[i]+(ordered[min(i+1,len(ordered)-1)]-ordered[i])*(h-i) def stats(values): if not values:return dict(n=0,min=None,mean=None,p25=None,p50=None,p75=None,p95=None,max=None) result=dict(n=len(values),min=min(values),mean=statistics.fmean(values),max=max(values)) result.update({name:quantile(values,p) for name,p in PROBS.items()}) assert list(sorted([result[k] for k in ('min','p25','p50','p75','p95','max')]))==[result[k] for k in ('min','p25','p50','p75','p95','max')] if len(values)>1: reference=statistics.quantiles(values,n=100,method='inclusive') for name,p in PROBS.items():assert math.isclose(result[name],reference[round(p*100)-1],rel_tol=1e-12,abs_tol=1e-18) assert result['p50']==statistics.median(values) return result def bin_spec(values): lo,hi=min(values),max(values);iqr=quantile(values,.75)-quantile(values,.25) if lo==hi:return dict(lo=lo,hi=hi,bins=1,method='CONSTANT',uncapped_bins=1,cap_applied=False) if iqr>0: requested=math.ceil((hi-lo)/(2*iqr/(len(values)**(1/3)))) method='Freedman-Diaconis: ceil(range / (2*IQR*N^(-1/3)))' else: requested=math.ceil(math.log2(len(values))+1);method='Sturges fallback (IQR=0)' return dict(lo=lo,hi=hi,bins=max(1,min(65536,requested)),method=method,uncapped_bins=requested,cap_applied=requested>65536) def histogram(values,spec,bins=None): n=bins or spec['bins'];lo,hi=spec['lo'],spec['hi'] if lo==hi:edges=[lo,hi];n=1 else:edges=[lo+(hi-lo)*i/n for i in range(n)]+[hi] counts=[0]*n def locate(v): assert lo<=v<=hi return 0 if lo==hi else min(n-1,bisect.bisect_right(edges,v)-1) for v in values:counts[locate(v)]+=1 assert sum(counts)==len(values) modes=[] if values: peak_count=max(counts) for i,count in enumerate(counts): if count==peak_count: members=[v for v in values if locate(v)==i] modes.append(dict(index=i,lower=edges[i],upper=edges[i+1],upper_closed=i==n-1, count=count,share=count/len(values),within_bin_mean=statistics.fmean(members))) return dict(bins=n,width=(hi-lo)/n,counts=counts,modes=modes) def modal_summary(values,spec): sizes=sorted({max(1,spec['bins']//2),spec['bins'],min(131072,spec['bins']*2)}) if spec['lo']!=spec['hi'] else [1] variants=[histogram(values,spec,n) for n in sizes] base=next(v for v in variants if v['bins']==spec['bins']) unique=all(len(v['modes'])==1 for v in variants) overlap=unique and (spec['lo']==spec['hi'] or max(v['modes'][0]['lower'] for v in variants)=max(5,.15*m) and (i==0 or c>counts[i-1]) and (i==len(counts)-1 or c>counts[i+1])] prefix=list(itertools.accumulate(counts));candidates=[] for left,right in itertools.combinations(peaks,2): if right-left<3:continue j=min(range(left+1,right),key=lambda i:counts[i]) if counts[j]>.25*min(counts[left],counts[right]):continue if min(prefix[j],n-prefix[j])<.1*n:continue candidates.append(dict(left_peak=left,right_peak=right,valley_bin=j,valley_count=counts[j])) return dict(peaks=peaks,candidates=candidates) def load_inputs(): sha=json.loads((SOURCE/'SHA256SUMS.json').read_text());fingerprints={};videos={};all_labels=None def checked(path): rel=str(path.relative_to(SOURCE));digest=hashlib.sha256(path.read_bytes()).hexdigest() assert sha[rel]==digest,rel fingerprints[rel]=digest return path for path in sorted(SOURCE.glob('*/data/posec3d_raw.json')): folder=path.parent;clip=folder.parent.name raw=json.loads(checked(path).read_text());summary=json.loads(checked(folder/'posec3d_summary.json').read_text()) labels={int(r['output_index']):r['class_name'] for r in csv.DictReader(checked(folder/'classes_posec3d.csv').open())} assert set(labels)==set(range(60)) if all_labels is None:all_labels=labels assert labels==all_labels assert summary['status']=='PASS' and summary['temporal_samples']==16 and summary['observation_fps']==10 assert summary['prediction_count']==len(raw) and summary['evaluation_samples']==[r['sample'] for r in raw] excluded=collections.Counter();valid=[] for position,r in enumerate(raw): if position: assert r['sample']>raw[position-1]['sample'] and r['source_frame']>raw[position-1]['source_frame'] and r['seconds']>raw[position-1]['seconds'] items=[it for it in r['items'] if 'scores' in it] assert len(items)<=1 if not items: reason=r.get('status') or ('NO_ITEMS' if not r['items'] else r['items'][0]['status']) excluded[reason]+=1;continue item=items[0];scores=item['scores'] assert len(scores)==60 and all(math.isfinite(v) and 0<=v<=1 for v in scores) assert abs(math.fsum(scores)-1)<1e-5 assert item['input_shape'][3]==16 and r['window']['samples']==list(range(r['sample']-8,r['sample']+8)) valid.append(dict(clip=clip,sample=r['sample'],source_frame=r['source_frame'],pts=r['seconds'],scores=scores, top5=sorted(range(60),key=lambda k:(-scores[k],k))[:5])) assert len(valid)+sum(excluded.values())==len(raw) videos[clip]=dict(total=len(raw),valid=len(valid),excluded=dict(excluded),valid_ratio=len(valid)/len(raw),observations=valid) assert len(videos)==7 model=json.loads(checked(SOURCE/'00_model_manifest.json').read_text()) checked(SOURCE/'00_source_manifest.json') return videos,all_labels,fingerprints,model['weights']['posec3d'] def compute(): videos,labels,fingerprints,weight=load_inputs() pooled=[r for v in videos.values() for r in v['observations']] actual=dict(total=sum(v['total'] for v in videos.values()),valid=len(pooled)) assert actual==dict(total=4701,valid=4316),actual result=dict(timestamp=datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).isoformat(), source=str(SOURCE),fingerprints=fingerprints,weight=weight,actual=actual, validity={k:{x:y for x,y in v.items() if x!='observations'} for k,v in videos.items()},classes={}) groups={k:v['observations'] for k,v in videos.items()};groups['ALL']=pooled for class_id in IDS: values=[r['scores'][class_id] for r in pooled];spec=bin_spec(values) c=dict(name=labels[class_id],histogram_spec=spec,groups={},thresholds=[],valley_scans=[]) for clip,records in groups.items(): a=[r['scores'][class_id] for r in records];s=stats(a) maxima=[{k:r[k] for k in ('clip','sample','source_frame','pts')} for r in records if r['scores'][class_id]==s['max']] c['groups'][clip]=dict(stats=s,modal=modal_summary(a,spec),max_first=maxima[0] if maxima else None, max_ties=len(maxima),top5_count=sum(class_id in r['top5'] for r in records)) means=[g['stats']['mean'] for k,g in c['groups'].items() if k!='ALL' and g['stats']['n']] c['equal_video_mean']=statistics.fmean(means) if means else None;c['equal_video_count']=len(means) weighted=math.fsum(c['groups'][k]['stats']['mean']*c['groups'][k]['stats']['n'] for k in videos if c['groups'][k]['stats']['n'])/len(pooled) assert math.isclose(weighted,c['groups']['ALL']['stats']['mean'],rel_tol=1e-12) for label in ('p25','p50','p75'): cutoff=c['groups']['ALL']['stats'][label];existing=next((t for t in c['thresholds'] if t['cutoff']==cutoff),None) if existing:existing['labels'].append(label);continue threshold=dict(labels=[label],cutoff=cutoff,groups={}) for clip,records in groups.items(): a=[r['scores'][class_id] for r in records] count=sum(v>=cutoff for v in a);top=sum(r['scores'][class_id]>=cutoff and class_id in r['top5'] for r in records) assert count==len(a)-bisect.bisect_left(sorted(a),cutoff) and 0<=top<=count<=len(a) threshold['groups'][clip]=dict(n=len(a),pass_count=count,pass_ratio=count/len(a) if a else None, top5_pass_count=top,top5_pass_ratio=top/len(a) if a else None) assert threshold['groups']['ALL']['pass_count']==sum(threshold['groups'][k]['pass_count'] for k in videos) assert threshold['groups']['ALL']['top5_pass_count']==sum(threshold['groups'][k]['top5_pass_count'] for k in videos) c['thresholds'].append(threshold) for bins in sorted({max(1,spec['bins']//2),spec['bins'],min(131072,spec['bins']*2)}): h=histogram(values,spec,bins);scan=valley_scan(h) c['valley_scans'].append(dict(bins=bins,**scan)) c['valley_candidate_status']='NONE_UNLESS_STABLE_ACROSS_WIDTHS' result['classes'][str(class_id)]=c result['validation']=dict(status='PASS',source_files=len(fingerprints),quantiles='R7 cross-check statistics.quantiles inclusive', checks=['schema/range/softmax/16 observations','denominator and exclusions','common-bin counts', 'weighted pooled mean','raw >= cutoff vs independent bisect','per-video counts sum to pooled']) return result if __name__=='__main__': result=compute();target=RUN/'statistics.json';target.write_text(json.dumps(result,ensure_ascii=False,indent=2)) print(json.dumps(dict(status='PASS',actual=result['actual'],source_files=len(result['fingerprints']),output=str(target)),ensure_ascii=False)) for key,c in result['classes'].items(): g=c['groups']['ALL'];print(key,c['name'],'pooled',g['stats'],'equal_video_mean',c['equal_video_mean'], 'bins',c['histogram_spec']['bins'],'modal',g['modal']['base'],'sensitivity',g['modal']['flag'], 'valleys',[(v['bins'],len(v['candidates'])) for v in c['valley_scans']]) for t in c['thresholds']:print('THRESHOLD',key,t['labels'],t['cutoff'],t['groups'])