from ops import *
import sys,signal,uuid,traceback
label=sys.argv[1];assert label in ('pose-a','pose-a-resume','mixed-a','pose-c','mixed-c','mixed-c-16')
mode,condition=label.split('-')[:2];KEYS=('pose',) if mode=='pose' else ('pose','fire','crowd')
LEVELS=(16,) if label=='mixed-c-16' else range(12,17) if label=='pose-a-resume' else range(1,17)
HERE=BASE/label;HERE.mkdir(mode=0o700)
ORDER=list(range(30,46));sids={c:'camera-'+uuid.uuid5(uuid.NAMESPACE_URL,str(c)).hex for c in ORDER}
FIELDS=('intrusion','loitering','fall','fight','fire','crowd','tracking')
def save(name,data):
    with (HERE/name).open('x') as f:json.dump(data,f,indent=2)
def event(name,**values):
    row=dict(event=name,t=time.monotonic(),wall_ms=time.time_ns()//1000000);row.update(values)
    with (HERE/'journal.jsonl').open('a') as f:f.write(json.dumps(row)+'\n')
def interrupt(*args):raise RuntimeError('TRIAL_INTERRUPTED')
signal.signal(signal.SIGTERM,interrupt);signal.signal(signal.SIGINT,interrupt)
launched=[];touched=[];starts={};initial_oom=None;control_pid=None;reader=None;reader_pid=None
saved={c:api('/api/v1/streams/'+sids[c]+'/settings')['configured'] for c in ORDER}
save('settings-before.json',saved)
def patch(c,changes):
    revision=api('/api/v1/streams/'+sids[c]+'/settings')['configured']['revision']
    return request('/api/v1/streams/'+sids[c]+'/settings',dict(expected_revision=revision,**changes),method='PATCH')
def rows():
    out={}
    for c in launched:
        begin=time.monotonic();r=api('/api/v1/streams/'+sids[c]);end=time.monotonic()
        out[str(c)]={k:r.get(k) for k in ('state','run_id','run_generation','last_error','cleanup_error','cleanup_pending','degraded','scenarios','publisher_metrics')}
        out[str(c)].update(read_begin=begin,read_end=end)
        providers=r.get('provider_processes') or ({KEYS[0]:r} if len(KEYS)==1 else {})
        out[str(c)]['providers']={p:{k:v.get(k) for k in ('state','child_pid','child_sampled_ns','last_error','cleanup_error','last_cleanup','metadata_path','publisher_metrics')} for p,v in providers.items()}
    return out
def safe(current):
    if shutil.disk_usage('/').free<15*1024**3:raise RuntimeError('ROOT_FREE_BOUND')
    mem=int(next(x for x in Path('/proc/meminfo').read_text().splitlines() if x.startswith('MemAvailable:')).split()[1])*1024
    if mem<2*1024**3:raise RuntimeError('RAM_FREE_BOUND')
    for c,r in current.items():
        if r['last_error'] or r['cleanup_error'] or r['degraded']:raise RuntimeError('CHANNEL_ERROR_'+c)
        for p,v in r['providers'].items():
            if v['last_error'] or v['cleanup_error'] or v['state']=='error':raise RuntimeError('PROVIDER_ERROR_'+c+'_'+p)
def persist(current,where):
    with (HERE/'startup.jsonl').open('a') as f:f.write(json.dumps(dict(t=time.monotonic(),where=where,channels=current))+'\n')
def sample():
    global initial_oom,control_pid
    begin=time.monotonic();current=rows();mapping={str(v['child_pid']):c+':'+p for c,r in current.items() for p,v in r['providers'].items() if v.get('child_pid')}
    if control_pid is None and mapping:
        code="from pathlib import Path;import sys;print(next(x.split()[1] for x in (Path('/proc')/sys.argv[1]/'status').read_text().splitlines() if x.startswith('PPid:')))"
        control_pid=cmd(['docker','exec',CONTAINER,'python3','-B','-c',code,next(iter(mapping))],timeout=5).strip()
    if control_pid:mapping[control_pid]='channel-control'
    resources=probe('resources',mapping)
    a=api('/api/v1/metadata/status')['artifacts'];train=training()
    value=dict(t=begin,finished=time.monotonic(),wall_ms=time.time_ns()//1000000,channels=current,resources=resources,
        gpu=cmd(['nvidia-smi','--query-gpu=utilization.gpu,utilization.memory,memory.used,power.draw,clocks.sm,clocks.mem','--format=csv,noheader,nounits'],timeout=6).strip(),
        artifacts={k:a.get(k) for k in ('rendering','active','activity','counts','peaks','error','storage_warning','ledger')},training=train.get('status'))
    oom=dict(x.split() for x in resources['cgroup'].get('memory.events','').splitlines())
    if initial_oom is None:initial_oom=oom
    if any(int(oom.get(k,0))>int(initial_oom.get(k,0)) for k in ('oom','oom_kill')):raise RuntimeError('CGROUP_OOM')
    if a.get('error'):raise RuntimeError('ARTIFACT_SERVICE_ERROR')
    return value
def contract(current):
    for c,r in current.items():
        assert set(r['providers'])==set(KEYS),'PROVIDER_COUNT'
        for p,v in r['providers'].items():
            path=v['metadata_path'];art=path.get('artifacts') or {}
            assert path['execution_path']=='provider-linear-trial' and path['source_latest_trial'] and not path['freshness_enabled'] and path['queue_max_age_s']==1
            if condition=='c':assert art.get('mode')=='test_no_capture' and not any(art.get(k) for k in ('delivery_enabled','encoded_cache_enabled','training_enabled','pending','queued'))
            elif p!='crowd':
                assert art.get('mode')!='test_no_capture'
                if art.get('error') or art.get('unaccepted_events'):raise RuntimeError('EVENT_HANDOFF_'+c+'_'+p)
            d=path['performance']['pipeline_diagnostic']['providers'][p]
            assert d['collecting'] and d.get('input_timeline_limit')==2,'DIAGNOSTIC_WINDOW'
def start(c):
    safe(rows());launched.append(c);sent=time.monotonic();ack=request('/api/v1/streams/'+sids[c]+'/start',{},timeout=20)
    starts[c]=dict(sent=sent,wall_ms=time.time_ns()//1000000,ack_at=time.monotonic(),request_state=ack.get('state'));event('start_ack',camera=c,**starts[c])
    until=time.monotonic()+90
    while time.monotonic()<until:
        current=rows();persist(current,'start-'+str(c));safe(current);r=current[str(c)]
        if set(r['providers'])==set(KEYS) and all(v.get('metadata_path',{}).get('providers',{}).get(p,{}).get('responses',0)>=2 for p,v in r['providers'].items()):
            starts[c]['ready_observed_at']=time.monotonic();return
        time.sleep(1)
    raise RuntimeError('INPUT_OR_PROVIDER_RESPONSE_'+str(c))
assert all(r['state']=='stopped' and not r.get('cleanup_pending') for r in api('/api/v1/streams')['streams'])
assert not inspect('contextra-icerti-ai-gateway-1')['State']['Running']
result=dict(mode=label,stages=[],starts=starts,errors=[],cleanup=None)
try:
    models=probe('models');save('model-configs.json',models);before=ledger();save('capture-before.json',before);save('training-before.json',training())
    conditions=[]
    for c in ORDER:
        desired={k:dict(enabled=True) for k in ('intrusion','loitering','fall','fight','tracking')}
        desired.update(fire=dict(enabled=mode=='mixed',scenario=dict(enabled=mode=='mixed')),crowd=dict(enabled=mode=='mixed'))
        touched.append(c);cfg=patch(c,desired)['configured'];assert [p['key'] for p in cfg['execution_plan']['providers']]==list(KEYS) and cfg['analysisFpsTarget']==10
        conditions.append(dict(camera=c,original_revision=saved[c]['revision'],trial_revision=cfg['revision'],scenarios={k:cfg[k] for k in FIELDS},pose=cfg['pose']))
    save('conditions.json',dict(order=ORDER,levels=list(LEVELS),duration_s=30,warmup_s=5,condition=condition,conditions=conditions,native_window_seconds=2400,timeline_limit=2,
        artifact_admission_rejection='record_and_continue_user_approved',resume_new_runs=label=='pose-a-resume'))
    with (BASE/(label+'-redis.jsonl')).open('x') as log,(HERE/'reader.stderr').open('x') as err:
        reader=subprocess.Popen(['docker','exec',CONTAINER,'python3','-u','-B','-c',(BASE/'reader.py').read_text()],stdout=log,stderr=err)
        until=time.monotonic()+10
        while time.monotonic()<until:
            with (BASE/(label+'-redis.jsonl')).open() as f:line=f.readline()
            if line:ready=json.loads(line);assert ready['kind']=='ready';reader_pid=ready['pid'];break
            if reader.poll() is not None:raise RuntimeError('READER_START')
            time.sleep(.1)
        else:raise RuntimeError('READER_TIMEOUT')
        for level in LEVELS:
            stage=dict(level=level,index=level-1,seconds=30,samples=[],status='RUNNING');result['stages'].append(stage)
            event('stage_begin',level=level)
            for c in ORDER[:level]:
                if c not in launched:start(c)
            time.sleep(5);safe(rows());begin=time.monotonic()
            stage.update(started_mono=begin,started_wall_ms=time.time_ns()//1000000)
            for i in range(7):
                time.sleep(max(0,begin+5*i-time.monotonic()));value=sample();stage['samples'].append(value)
                with (HERE/'samples.jsonl').open('a') as f:f.write(json.dumps(dict(level=level,**value))+'\n')
                safe(value['channels']);contract(value['channels'])
                if reader.poll() is not None:raise RuntimeError('READER_EXIT')
            assert models==probe('models')
            now=ledger();stage['capture_ledger']=now
            if condition=='c':assert now==before,'C_LEDGER_CHANGED'
            stage['status']='PASS';stage['finished_wall_ms']=time.time_ns()//1000000;save('stage-'+str(level).zfill(2)+'.json',stage)
            a,b=stage['samples'][0],stage['samples'][-1];seconds=b['t']-a['t'];means={}
            for p in KEYS:
                deltas=[b['channels'][str(x)]['providers'][p]['publisher_metrics'].get('sent_frames',0)-a['channels'][str(x)]['providers'][p]['publisher_metrics'].get('sent_frames',0) for x in ORDER[:level]]
                means[p]=dict(mean=round(sum(deltas)/level/seconds,2),minimum=round(min(deltas)/seconds,2))
            print(json.dumps(dict(mode=label,level=level,status='PASS',redis_fps=means,sample_seconds=round(seconds,2))),flush=True)
    save('capture-after.json',ledger());save('training-after.json',training())
except Exception as e:
    result['errors'].append(dict(type=type(e).__name__,reason=str(e)[:160],call_path=[dict(function=x.name,line=x.lineno,file=Path(x.filename).name) for x in traceback.extract_tb(e.__traceback__)]))
    event('trial_error',**result['errors'][-1]);print(json.dumps(dict(mode=label,error=result['errors'][-1])),flush=True)
finally:
    if reader is not None and reader.poll() is None and reader_pid:
        code="import os,signal;from pathlib import Path;p="+str(reader_pid)+";assert b'contextra-icerti-ac-reader-r01' in (Path('/proc')/str(p)/'cmdline').read_bytes();os.kill(p,signal.SIGTERM)"
        try:cmd(['docker','exec',CONTAINER,'python3','-B','-c',code],timeout=5);reader.wait(timeout=10)
        except Exception:result['errors'].append(dict(reason='READER_CLEANUP'))
    for c in launched:
        try:request('/api/v1/streams/'+sids[c]+'/stop',{},timeout=20)
        except Exception:result['errors'].append(dict(reason='STOP_REQUEST_'+str(c)))
    until=time.monotonic()+120;current={};reaped=False
    while time.monotonic()<until:
        current=rows();reaped=all(not r['cleanup_pending'] and all(not v.get('child_pid') for v in r['providers'].values()) for r in current.values())
        if reaped:break
        time.sleep(1)
    result['cleanup']=dict(reaped=reaped,states=current,reader_exit=reader.returncode if reader else None)
    if reaped:
        for c in touched:
            try:
                if c in current and current[c]['state']!='stopped':raise RuntimeError('RESTORE_ERROR_STATE')
                if str(c) in current and current[str(c)]['state']!='stopped':raise RuntimeError('RESTORE_ERROR_STATE')
                changes={k:saved[c][k] for k in FIELDS}
                if 'scenario' not in changes['fire']:changes['fire']=dict(changes['fire'],scenario=dict(enabled=False))
                patch(c,changes)
            except Exception:result['errors'].append(dict(reason='RESTORE_SETTINGS_'+str(c)))
    save('trial-result.json',result);print(json.dumps(dict(mode=label,completed=[s['level'] for s in result['stages'] if s['status']=='PASS'],errors=result['errors'],reaped=reaped)),flush=True)
if result['errors'] or not reaped:raise SystemExit(1)