import hashlib,json,os,signal,subprocess,time,traceback,urllib.request,uuid,shutil
from pathlib import Path
os.umask(0o077)
BASE=Path('/backup/contextra/deployment-inbox/icerti-crowd-20260928-r01')
HERE=BASE/'a13';HERE.mkdir(mode=0o700)
CONTAINER='contextra-icerti-analytics-1'
ORDER=list(range(30,43));KEYS=('pose','fire','crowd')
FIELDS=('intrusion','loitering','fall','fight','fire','crowd','tracking')
sids={c:'camera-'+uuid.uuid5(uuid.NAMESPACE_URL,str(c)).hex for c in ORDER}
def cmd(args,timeout=20,input=None):
    p=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout)
    if p.returncode:raise RuntimeError('COMMAND_'+Path(args[0]).name+'_'+str(p.returncode))
    return p.stdout
def api(path):return json.load(urllib.request.urlopen('http://127.0.0.1:35000'+path,timeout=8))
def request(path,body,method='POST'):
    r=urllib.request.Request('http://127.0.0.1:35000'+path,data=json.dumps(body).encode(),method=method,headers={'Content-Type':'application/json'})
    return json.load(urllib.request.urlopen(r,timeout=20))
def save(name,value):
    with (HERE/name).open('x') as f:json.dump(value,f,indent=2)
def event(name,**values):
    with (HERE/'journal.jsonl').open('a') as f:f.write(json.dumps(dict(event=name,wall_ms=time.time_ns()//1000000,**values))+'\n')
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),'PATCH')['configured']
def probe(mode,mapping=None):
    return json.loads(cmd(['docker','exec','-i',CONTAINER,'python3','-B','-c',(BASE/'kit/tools/provider_scale_probe.py').read_text(),mode],input=json.dumps(mapping or {})))
def rows():
    out={}
    for c in launched:
        r=api('/api/v1/streams/'+sids[c])
        out[str(c)]={k:r.get(k) for k in ('state','run_id','run_generation','last_error','cleanup_error','cleanup_pending','degraded')}
        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 r.get('provider_processes',{}).items()}
    return out
def safe(current):
    assert shutil.disk_usage('/').free>15*1024**3,'ROOT_FREE_BOUND'
    mem=int(next(x for x in Path('/proc/meminfo').read_text().splitlines() if x.startswith('MemAvailable:')).split()[1])
    assert mem>2*1024**2,'RAM_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)
            art=(v.get('metadata_path') or {}).get('artifacts') or {}
            if art.get('unaccepted_events'):raise RuntimeError('EVENT_HANDOFF_'+c+'_'+p)
def contract(current):
    for c,r in current.items():
        assert set(r['providers'])==set(KEYS),'PROVIDER_SET_'+c
        for p,v in r['providers'].items():
            path=v['metadata_path'];gate=path.get('event_cooldown') or {};art=path.get('artifacts') or {}
            assert path['execution_path']=='provider-linear-trial' and path['source_latest_trial'] and not path['freshness_enabled']
            assert path['queue_max_age_s']==1 and path['capacity']==32
            assert gate['seconds']==10 and all(x==10 for x in gate.get('overrides',{}).values()),'COOLDOWN'
            assert art.get('mode')!='test_no_capture'
def sample():
    begin=time.monotonic();current=rows();safe(current);contract(current)
    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')}
    resource=probe('resources',mapping)
    oom=dict(x.split() for x in resource['cgroup'].get('memory.events','').splitlines())
    assert not int(oom.get('oom_kill',0)) and not int(oom.get('oom',0)),'OOM'
    art=api('/api/v1/metadata/status')['artifacts']
    if art.get('error'):raise RuntimeError('ARTIFACT_SERVICE_ERROR')
    result=dict(t=begin,finished=time.monotonic(),wall_ms=time.time_ns()//1000000,channels=current,resources=resource,
        gpu=cmd(['nvidia-smi','--query-gpu=utilization.gpu,utilization.memory,memory.used','--format=csv,noheader,nounits'],8).strip(),
        artifacts={k:art.get(k) for k in ('active','rendering','counts','error','ledger')})
    with (HERE/'samples.jsonl').open('a') as f:f.write(json.dumps(result)+'\n')
    return result
def interrupt(*args):raise RuntimeError('TRIAL_INTERRUPTED')
signal.signal(signal.SIGTERM,interrupt);signal.signal(signal.SIGINT,interrupt)
launched=[];touched=[];reader=None;samples=[]
result=dict(errors=[],started=launched,touched=touched)
saved={c:api('/api/v1/streams/'+sids[c]+'/settings')['configured'] for c in ORDER}
save('settings-before.json',saved)
assert all(r['state']=='stopped' and not r.get('cleanup_pending') for r in api('/api/v1/streams')['streams'])
assert not json.loads(cmd(['docker','inspect','contextra-icerti-ai-gateway-1']))[0]['State']['Running']
try:
    result['begin_wall']=time.time();models=probe('models');save('models-before.json',models)
    for c in ORDER:
        touched.append(c)
        desired={k:dict(enabled=True) for k in ('intrusion','loitering','fall','fight','tracking')}
        desired.update(fire=dict(enabled=True,scenario=dict(enabled=True)),crowd=dict(enabled=True,event_enabled=True,
            reference_count=50,safe_ratio=.25,caution_ratio=.5,warning_ratio=.75,danger_ratio=.9,suppress_repeated_danger=True))
        cfg=patch(c,desired);assert [p['key'] for p in cfg['execution_plan']['providers']]==list(KEYS) and cfg['analysisFpsTarget']==10
        save('conditions-'+str(c)+'.json',{k:cfg[k] for k in (*FIELDS,'pose','revision','analysisFpsTarget')})
        launched.append(c);request('/api/v1/streams/'+sids[c]+'/start',{})
        deadline=time.monotonic()+90
        while time.monotonic()<deadline:
            current=rows();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()):break
            time.sleep(1)
        else:raise RuntimeError('FIRST_RESPONSE_'+str(c))
        event('ready',camera=c,channel_count=len(launched));print(json.dumps(dict(ready=c,channels=len(launched))),flush=True)
    time.sleep(5);current=rows();safe(current);contract(current)
    for c in ORDER:request('/api/v1/streams/'+sids[c]+'/metrics',dict(run_generation=current[str(c)]['run_generation'],enabled=True,seconds=60))
    with (HERE/'redis.jsonl').open('x') as out,(HERE/'reader.stderr').open('x') as err:
        reader=subprocess.Popen(['docker','exec',CONTAINER,'python3','-u','-B','-c',(BASE/'reader.py').read_text()],stdout=out,stderr=err)
        begin=time.monotonic();result['measurement_start_wall']=time.time()
        for i in range(7):
            time.sleep(max(0,begin+5*i-time.monotonic()));samples.append(sample())
            print(json.dumps(dict(sample=i,elapsed=time.monotonic()-begin)),flush=True)
        result['sample_seconds']=samples[-1]['t']-samples[0]['t'];result['measurement_end_wall']=time.time()
        assert models==probe('models'),'MODEL_CHANGED'
        assert reader.wait(timeout=15)==0,'READER_FAILED'
    result['status']='MEASURED'
except Exception as error:
    result['errors'].append(dict(type=type(error).__name__,reason=str(error)[:160],trace=[dict(file=Path(x.filename).name,line=x.lineno,function=x.name) for x in traceback.extract_tb(error.__traceback__)]))
    event('error',**result['errors'][-1]);print(json.dumps(result['errors'][-1]),flush=True)
finally:
    if reader is not None and reader.poll() is None:
        try:reader.wait(timeout=45)
        except subprocess.TimeoutExpired:result['errors'].append(dict(reason='READER_REAP_FAILED'))
    for c in launched:
        try:request('/api/v1/streams/'+sids[c]+'/stop',{})
        except Exception:result['errors'].append(dict(reason='STOP_REQUEST_'+str(c)))
    deadline=time.monotonic()+120
    while True:
        current=rows();reaped=all(r['state']=='stopped' and not r['cleanup_pending'] and all(not v.get('child_pid') for v in r['providers'].values()) for r in current.values())
        if reaped or time.monotonic()>deadline:break
        time.sleep(1)
    result['cleanup']=dict(reaped=reaped,channels=current)
    if reaped:
        for c in touched:
            try:
                changes={k:saved[c][k] for k in FIELDS}
                if 'scenario' not in changes['fire']:changes['fire']=dict(changes['fire'],scenario=dict(enabled=False))
                cfg=patch(c,changes);assert all(cfg[k]==saved[c][k] for k in FIELDS),'SETTINGS_VIEW_DRIFT'
            except Exception as e:result['errors'].append(dict(reason='RESTORE_'+str(c),type=type(e).__name__))
        result['settings_restored']=not any(e['reason'].startswith('RESTORE_') for e in result['errors'])
    save('trial-result.json',result)
    print(json.dumps(dict(status=result.get('status'),errors=result['errors'],reaped=reaped)),flush=True)
if result['errors'] or not reaped:raise SystemExit(1)