from ops import *
import csv,math,statistics as st,sys
mode=sys.argv[1];folder=HERE/mode
logical_cpus=json.loads((HERE/'hardware.json').read_text())['logical_cpus']
def stats(v):
    v=sorted(v)
    return dict(n=len(v),mean=st.mean(v),median=st.median(v),p95=v[math.ceil(.95*len(v))-1],maximum=v[-1],minimum=v[0]) if v else None
def slope(rows,key):
    if len(rows)<2:return None
    t=st.mean(x['t'] for x in rows);v=st.mean(x[key] for x in rows)
    den=sum((x['t']-t)**2 for x in rows)
    return 60*sum((x['t']-t)*(x[key]-v) for x in rows)/den if den else None
def diag(path,p):return path.get('performance',{}).get('pipeline_diagnostic',{}).get('providers',{}).get(p,{})
def delta(a,b,k):return b.get(k,0)-a.get(k,0)
def delivery(v,p):return next(iter(v.get('publisher_metrics',{}).get('delivery',{}).values()),{}).get(p,{})
stages=[json.loads(p.read_text()) for p in sorted(folder.glob('stage-*.json'))]
records={};reset=[]
for name in ('startup.jsonl','samples.jsonl'):
    path=folder/name
    if not path.exists():continue
    with path.open() as f:
        for line in f:
            s=json.loads(line)
            for c,r in s['channels'].items():
                for p,v in r['providers'].items():
                    for x in diag(v.get('metadata_path') or {},p).get('input_timeline',[]):
                        records[(c,r['run_id'],p,x['pts'])]=dict(camera=c,run=r['run_id'],provider=p,**x)
points=[];anchors={};last={};segments={}
for row in sorted(records.values(),key=lambda x:x['input']['ingress_ns']):
    x=row['input'];key=(row['camera'],row['run'],row['provider']);old=last.get(key)
    ticks=(x['rtp']-old['rtp']+2**31)%2**32-2**31 if old else 0
    if old and (x['ssrc']!=old['ssrc'] or x['rate']!=old['rate'] or ticks<0):
        reset.append(dict(boundary='native',camera=row['camera'],provider=row['provider'],run=row['run'],pts=row['pts']))
        segments[key]=segments.get(key,0)+1;old=None
    seg=(key,segments.get(key,0));unwrapped=old['unwrapped']+ticks if old else x['rtp']
    last[key]=dict(x,unwrapped=unwrapped)
    if seg not in anchors:anchors[seg]=dict(x,unwrapped=unwrapped)
    origin=anchors[seg];media=(unwrapped-origin['unwrapped'])/x['rate']
    point=dict(camera=row['camera'],run=row['run'],provider=row['provider'],pts=row['pts'],segment=seg[1],
        t=x['ingress_ns']/1e9,rtp=x['rtp'],ssrc=x['ssrc'],rate=x['rate'],media_elapsed_s=media)
    for k in ('depay_first_ns','source_exit_ns','ingress_ns'):
        point[k+'_relative_s']=(x[k]-origin[k])/1e9-media
    point['depay_source_ms']=(x['source_exit_ns']-x['depay_first_ns'])/1e6
    points.append(point)
lookup={(r['camera'],r['run'],r['provider'],r['pts']):r for r in points}
redis_rows=[];rfirst={};rlast={};rsegments={};unmatched=0;missing_rtp=0
with (HERE/(mode+'-redis.jsonl')).open() as f:
    for line in f:
        row=json.loads(line)
        if row.get('kind')!='frame':continue
        if type(row.get('rtp')) is not int or not row.get('clock'):missing_rtp+=1;continue
        key=(str(row['camera']),row['run'],row['provider']);old=rlast.get(key)
        ticks=(row['rtp']-old['rtp']+2**31)%2**32-2**31 if old else 0
        if old and (ticks<0 or row['clock']!=old['clock']):
            reset.append(dict(boundary='redis',camera=key[0],provider=key[2],run=key[1],pts=row['pts']))
            rsegments[key]=rsegments.get(key,0)+1;old=None
        seg=(key,rsegments.get(key,0));uw=old['uw']+ticks if old else row['rtp']
        rlast[key]=dict(rtp=row['rtp'],clock=row['clock'],uw=uw)
        wall=int(row['redis_id'].split('-')[0]);mono=row['read_mono_ns']/1e9+(wall-row['read_wall_ms'])/1000
        if seg not in rfirst:rfirst[seg]=dict(t=mono,uw=uw)
        origin=rfirst[seg];native=lookup.get((*key,row['pts']))
        unmatched+=int(native is None)
        redis_rows.append(dict(camera=key[0],run=key[1],provider=key[2],pts=row['pts'],seq=row['seq'],t=mono,
            redis_relative_s=mono-origin['t']-(uw-origin['uw'])/row['clock'],
            ingress_to_redis_ms=wall-row['ingest_ms'] if row.get('ingest_ms') is not None else None,
            native_joined=native is not None,visible=row['visible']))
lanes=[];summaries=[];resources=[];windows=[];startup=[];startup_seen=set()
trial_result=json.loads((folder/'trial-result.json').read_text()) if (folder/'trial-result.json').exists() else None
for stage in stages:
    a,b=stage['samples'][0],stage['samples'][-1];seconds=b['t']-a['t'];level=stage['level']
    local=[]
    for c,br in b['channels'].items():
        for p,bv in br['providers'].items():
            av=a['channels'][c]['providers'][p];ap=av['metadata_path'];bp=bv['metadata_path'];ad,bd=diag(ap,p),diag(bp,p)
            duration={}
            for k,v in bd.get('durations',{}).items():
                old=ad.get('durations',{}).get(k,{});n=delta(old,v,'count');ns=delta(old,v,'sum_ns')
                duration[k]=dict(n=n,sum_ns=ns,mean_ms=ns/n/1e6 if n else None)
            pp=[x for x in points if x['camera']==c and x['provider']==p and x['run']==br['run_id'] and a['t']<=x['t']<=b['t']]
            rr=[x for x in redis_rows if x['camera']==c and x['provider']==p and x['run']==br['run_id'] and a['t']<=x['t']<=b['t']]
            times=sorted(x['t'] for x in rr);cuts=[a['t'],*times,b['t']]
            gaps=[v-u for u,v in zip(cuts,cuts[1:])]
            pub0,pub1=delivery(av,p),delivery(bv,p)
            if (c,p) not in startup_seen and trial_result:
                startup_seen.add((c,p))
                sent=trial_result['starts'][c]['sent'];provider=bp['providers'][p]
                startup.append(dict(mode=mode,camera=c,provider=p,run=br['run_id'],start_mono=sent,
                    first_input_s=provider['first_input_at']-sent if provider.get('first_input_at') else None,
                    first_infer_entry_s=provider['first_request_at']-sent if provider.get('first_request_at') else None,
                    first_response_s=provider['first_response_at']-sent if provider.get('first_response_at') else None,
                    first_redis_success_s=pub1['first_at_ns']/1e9-sent if pub1.get('first_at_ns') else None))
            counts0,counts1=ap['performance']['counts'],bp['performance']['counts']
            cohort_reset=mode in ('mixed-resume','mixed-compact') and level==8
            lane=dict(mode=mode,level=level,camera=c,provider=p,cohort_reset=cohort_reset,
                new_channel=cohort_reset or int(c)==[3,5,6,8,9,10,11,12,13,14,15,16,17,18,19,20][level-1],seconds=seconds,
                selected_fps=delta(counts0,counts1,'selected')/seconds,
                requested_fps=delta(ap['providers'][p],bp['providers'][p],'requests')/seconds,
                complete_fps=delta(ap['providers'][p],bp['providers'][p],'responses')/seconds,
                accepted_fps=delta(ap,bp,'accepted_result')/seconds,finalized_fps=delta(ap,bp,'finalized')/seconds,
                redis_fps=delta(pub0,pub1,'count')/seconds,redis_read_fps=len(rr)/seconds,
                queue_expired=delta(ap,bp,'completed_queue_expired'),capacity_drop=delta(ap,bp,'completed_queue_drop'),
                over_100ms=delta(ap,bp,'over_freshness_'+p),queue_depth_max_sampled=max(s['channels'][c]['providers'][p]['metadata_path']['queued'] for s in stage['samples']),
                queue_age_max_sampled=max(s['channels'][c]['providers'][p]['metadata_path']['completed_queue_age_s'] for s in stage['samples']),
                provider_response_age_max_sampled=max(s['channels'][c]['providers'][p]['metadata_path']['providers'][p].get('last_response_age_s') or 0 for s in stage['samples']),
                sender={k:delta(av['publisher_metrics'],bv['publisher_metrics'],k) for k in ('replaced','expired','send_error')},
                gap_markers=delta(counts0,counts1,'source_queue_gap'),analysis_gap_markers=delta(counts0,counts1,'source_analysis_gap'),
                c_suppressed=delta(ap['artifacts'],bp['artifacts'],'event_candidates_suppressed'),
                max_redis_silence_s=max(gaps),redis_interval_s=stats([v-u for u,v in zip(times,times[1:])]),
                redis_drift_s=stats([x['redis_relative_s'] for x in rr]),redis_drift_s_per_min=slope(rr,'redis_relative_s'),
                ingress_drift_s=stats([x['ingress_ns_relative_s'] for x in pp]),ingress_drift_s_per_min=slope(pp,'ingress_ns_relative_s'),
                source_drift_s=stats([x['source_exit_ns_relative_s'] for x in pp]),
                ingress_to_redis_ms=stats([x['ingress_to_redis_ms'] for x in rr if x['ingress_to_redis_ms'] is not None]),
                source_samples=len(pp),redis_samples=len(rr),native_redis_joined=sum(x['native_joined'] for x in rr),durations=duration)
            local.append(lane);lanes.append(lane)
            for i in range(6):
                lo=a['t']+i*5;hi=min(b['t'],lo+5);rs=[x for x in rr if lo<=x['t']<hi];ps=[x for x in pp if lo<=x['t']<hi]
                windows.append(dict(mode=mode,level=level,camera=c,provider=p,window=i,
                    redis_count=len(rs),redis_relative_mean_s=st.mean(x['redis_relative_s'] for x in rs) if rs else None,
                    ingress_relative_mean_s=st.mean(x['ingress_ns_relative_s'] for x in ps) if ps else None))
    ra,rb=a['resources'],b['resources'];hx=[int(v) for v in ra['host_cpu'].split()[1:9]];hy=[int(v) for v in rb['host_cpu'].split()[1:9]];hd=[v-u for u,v in zip(hx,hy)]
    ca=dict(l.split() for l in ra['cgroup']['cpu.stat'].splitlines());cb=dict(l.split() for l in rb['cgroup']['cpu.stat'].splitlines())
    cpu=[]
    for group,tasks in rb['cpu'].items():
        old=ra['cpu'].get(group,{});pairs=[(old[t],v) for t,v in tasks.items() if t in old]
        cpu.append(dict(group=group,cpu_s=sum(v[0]-u[0] for u,v in pairs)/1e9,
            runnable_wait_thread_s=sum(v[1]-u[1] for u,v in pairs)/1e9,matched_threads=len(pairs)))
    resource=dict(mode=mode,level=level,cpu_busy_pct=100*(sum(hd)-hd[3]-hd[4])/sum(hd),cpu_steal_pct=100*hd[7]/sum(hd),
        cpu=cpu,quota=rb['cgroup']['cpu.max'],throttled_us=int(cb.get('throttled_usec',0))-int(ca.get('throttled_usec',0)),
        gpu_util_pct=st.mean(float(x['gpu'].split(',')[0]) for x in stage['samples']),
        gpu_mem_max_mib=max(float(x['gpu'].split(',')[2]) for x in stage['samples']),
        ram_max_bytes=max(int(x['resources']['cgroup']['memory.current']) for x in stage['samples']),
        renderer_peak_sampled=max(x['artifacts']['rendering'] for x in stage['samples']),
        artifact_counts={k:delta(a['artifacts']['counts'],b['artifacts']['counts'],k) for k in b['artifacts']['counts']})
    resource['analytics_cpu_core_equivalents']=(int(cb['usage_usec'])-int(ca['usage_usec']))/1e6/seconds
    resource['host_busy_core_equivalents']=resource['cpu_busy_pct']/100*logical_cpus
    resource['outside_analytics_cpu_core_equivalents_approx']=resource['host_busy_core_equivalents']-resource['analytics_cpu_core_equivalents']
    resources.append(resource)
    models={'pose':'RFDETRKeypointPreview','fire':'RFDETRFireSmoke','crowd':'Awiros_person_and_head_detection'}
    for p in sorted(set(x['provider'] for x in local)):
        group=[x for x in local if x['provider']==p];ta=ra['triton'][models[p]];tb=rb['triton'][models[p]]
        batch={};before={int(x['batch_size']):int(x['compute_infer']['count']) for x in ta['batch_stats']}
        for x in tb['batch_stats']:batch[int(x['batch_size'])]=int(x['compute_infer']['count'])-before.get(int(x['batch_size']),0)
        triton={}
        for k,v in tb['inference_stats'].items():
            old=ta['inference_stats'][k];n=int(v['count'])-int(old['count']);ns=int(v['ns'])-int(old['ns'])
            triton[k]=dict(count=n,mean_ms=ns/n/1e6 if n else None)
        durations={}
        for row in group:
            for k,v in row['durations'].items():
                d=durations.setdefault(k,dict(n=0,sum_ns=0));d['n']+=v['n'];d['sum_ns']+=v['sum_ns']
        for d in durations.values():d['mean_ms']=d['sum_ns']/d['n']/1e6 if d['n'] else None
        def cohort(new):
            g=[x for x in group if x['new_channel']==new]
            return {k:st.mean(x[k] for x in g) for k in ('complete_fps','redis_fps')} if g else None
        summaries.append(dict(mode=mode,level=level,provider=p,seconds=seconds,
            **{k:st.mean(x[k] for x in group) for k in ('selected_fps','requested_fps','complete_fps','accepted_fps','finalized_fps','redis_fps')},
            redis_min=min(x['redis_fps'] for x in group),redis_std=st.pstdev(x['redis_fps'] for x in group),
            total_redis_fps=sum(x['redis_fps'] for x in group),complete_min=min(x['complete_fps'] for x in group),
            queue_expired=sum(x['queue_expired'] for x in group),capacity_drop=sum(x['capacity_drop'] for x in group),
            max_redis_silence_s=max(x['max_redis_silence_s'] for x in group),
            source_drift_mean_s=st.mean(x['source_drift_s']['mean'] for x in group if x['source_drift_s']),
            redis_drift_mean_s=st.mean(x['redis_drift_s']['mean'] for x in group if x['redis_drift_s']),
            sender={k:sum(x['sender'][k] for x in group) for k in ('replaced','expired','send_error')},
            c_suppressed=sum(x['c_suppressed'] for x in group),batch_executions=batch,triton=triton,durations=durations,
            existing=cohort(False),new=cohort(True),cpu_busy_pct=resource['cpu_busy_pct'],gpu_util_pct=resource['gpu_util_pct']))
result=dict(mode=mode,summary=summaries,lanes=lanes,resources=resources,resets=reset,startup=startup,
    native_points=len(points),redis_points=len(redis_rows),unmatched_native=unmatched,missing_rtp=missing_rtp,
    notes=['Provider is part of every native/Redis join and anchor key.',
      'Redis drift uses all valid reader samples; native timeline exports at most8 (compact2)/8KiB and not every Frame is joined.',
      'Redis SSRC is not public: native sampled SSRC changes and Redis clock/backward RTP changes are recorded separately.',
      'Gap markers are not raw frame drop counts. CPU runnable wait sums thread time, not channel walltime.',
      'Redis entry wall time mapped using read wall/monotonic pair; clock steps and reader sampling remain limitations.',
      '30s per stage does not establish long-run stability or hardware maximum.'])
for name,value in [('analysis.json',result),('input-points.json',points),('redis-points.json',redis_rows),('windows.json',windows)]:
    with (folder/name).open('w') as f:json.dump(value,f,indent=2)
for name,value,keys in [('summary.csv',summaries,('mode','level','provider','seconds','complete_fps','redis_fps','redis_min','redis_std','max_redis_silence_s','source_drift_mean_s','redis_drift_mean_s','queue_expired','cpu_busy_pct','gpu_util_pct')),
                       ('windows.csv',windows,tuple(windows[0]) if windows else ('mode',))]:
    with (folder/name).open('w',newline='') as f:
        w=csv.DictWriter(f,fieldnames=keys,extrasaction='ignore');w.writeheader();w.writerows(value)
print(json.dumps([{k:r[k] for k in ('mode','level','provider','complete_fps','redis_fps','redis_min','queue_expired','cpu_busy_pct')} for r in summaries]))