from ops import *
import sys,yaml
phase=sys.argv[1]
label=sys.argv[2] if len(sys.argv)>2 else phase
original=yaml.safe_load((PRIVATE/'runtime-original.yaml').read_text())['added']
by_camera={str(s.get('camera_id')):s for s in original}
sid3=by_camera['3']['stream_id']
background=[by_camera[str(c)]['stream_id'] for c in (10,11,12,13,14,15,16,17,19,20)]
targets=[sid3]+(background if phase=='mixed' else [])
settings_path=PRIVATE/'settings-original.json'
if not settings_path.exists():
    save('settings-original.json',{sid:api('/api/v1/streams/'+sid+'/settings')['configured'] for sid in [sid3]+background},private=True)
saved=json.loads(settings_path.read_text())

def settings_patch(sid,changes):
    current=api('/api/v1/streams/'+sid+'/settings')['configured']
    return request('/api/v1/streams/'+sid+'/settings',dict(expected_revision=current['revision'],**changes),method='PATCH')

def stop_targets():
    for sid in targets:request('/api/v1/streams/'+sid+'/stop',{})
    end=time.monotonic()+85
    while True:
        rows=[api('/api/v1/streams/'+sid) for sid in targets]
        if all(not r.get('cleanup_pending') and not r.get('child_pid') and
               all(not p.get('child_pid') for p in r.get('provider_processes',{}).values()) for r in rows):return rows
        if time.monotonic()>end:raise RuntimeError('TRIAL_REAP_FAILED')
        time.sleep(.5)

def sample():
    rows={sid:api('/api/v1/streams/'+sid) for sid in targets}
    for sid,r in rows.items():
        if r['state']=='error' or r.get('last_error') or r.get('degraded'):raise RuntimeError('CHANNEL_OR_PROVIDER_FAILED:'+sid)
    return dict(t=time.monotonic(),groups=rows,gpu=cmd(['nvidia-smi','--query-gpu=utilization.gpu,memory.used','--format=csv,noheader,nounits']).strip())

def metadata():
    code='''import os,json,redis,hashlib,sys
from urllib.parse import urlsplit
auth=json.load(sys.stdin);u=urlsplit(os.environ['CONTEXTRA_METADATA_REDIS_URL'])
r=redis.Redis(host=u.hostname,port=u.port or 6379,db=int(u.path.strip('/') or 0),username=auth['username'],password=auth['password'],decode_responses=True,socket_timeout=2)
key=os.environ.get('CONTEXTRA_METADATA_NAMESPACE','contextra:metadata:v1')+':camera:'+hashlib.sha256(b'3').hexdigest()+':frames'
out={}
entries=r.xread({key:'0-0'},count=100,block=500)
for _,fields in (entries[0][1] if entries else []):
 s=json.loads(fields['json']);p=s.get('providerKey','MISSING');v=out.setdefault(p,dict(count=0,seq=[],pts=[],models=[],required=[],events=0,refs_valid=True))
 v['count']+=1;v['seq'].append(s['frameSeq']);v['pts'].append(s.get('ptsNs'));v['models']=list(set(v['models']+[s.get('providerModel')]));v['required']=s.get('requiredProviders');v['events']+=len(s.get('events',[]))
 refs={str(o.get('objectRef')) for o in s.get('objects',[])}
 v['refs_valid']=v['refs_valid'] and all(str(e.get('objectRef')) in refs for e in s.get('events',[]))
print(json.dumps(out));r.close()'''
    authpath=Path('/workspace/contextra-runtime-config/metadata-active/reader/reader-credentials.json')
    assert not authpath.is_symlink() and authpath.stat().st_uid==0 and authpath.stat().st_mode&0o077==0
    return json.loads(cmd(['docker','exec','-i','contextra-analytics-1','python3','-B','-c',code],input=authpath.read_text(),timeout=8))

assert not inspect('contextra-ai-gateway-1')['State']['Running']
save(label+'-intent.json',dict(phase=phase,channels=len(targets),wall_ms=time.time_ns()//1000000))
samples=[];result={};started=time.time();error=None
try:
    settings_patch(sid3,dict(fire={'enabled':True,'scenario':{'enabled':True}},crowd={'enabled':phase!='two'}))
    if phase=='mixed':
        sid13=by_camera['13']['stream_id']
        settings_patch(sid13,dict(fall={'enabled':False},pose={'enabled':False,'show_joints':False,'show_edges':False}))
    for sid in targets:request('/api/v1/streams/'+sid+'/start',{},timeout=20)
    deadline=time.monotonic()+80
    required={'pose','fire'} if phase=='two' else {'pose','fire','crowd'}
    while True:
        a=sample();rows=a['groups'];r=rows[sid3]
        ready=(all(v['state']=='running' for v in rows.values()) and
            set(r.get('provider_processes',{}))==required and
            all(p.get('metadata_path',{}).get('providers',{}).get(k,{}).get('responses',0)>2
                for k,p in r.get('provider_processes',{}).items()))
        if ready:break
        if time.monotonic()>deadline:raise RuntimeError('WARMUP_BOUND')
        time.sleep(1)
    time.sleep(5)
    duration=30 if phase=='mixed' else 10
    begin=time.monotonic()
    while True:
        samples.append(sample())
        if time.monotonic()-begin>=duration:break
        time.sleep(min(5,max(0,begin+duration-time.monotonic())))
    result['metadata']=metadata()
    assert set(result['metadata'])==required
    for k,v in result['metadata'].items():
        assert v['count']>1 and v['seq']==sorted(set(v['seq'])) and v['refs_valid']
        assert set(v['required'])==required and len(v['models'])==1
    result['result']='PASS'
except Exception as exc:
    error=type(exc).__name__+':'+str(exc)[:200]
    result.update(result='FAIL',error=error)
finally:
    result['after_stop']=stop_targets()
    for sid in targets:
        fields=('fire','crowd') if sid==sid3 else ('fall','pose') if sid==by_camera['13']['stream_id'] else ()
        if fields:settings_patch(sid,{k:saved[sid][k] for k in fields})
    result.update(samples=samples,started=started,finished=time.time())
    save(label+'-result.json',result)
print(json.dumps({k:v for k,v in result.items() if k not in ('samples','after_stop')}),flush=True)
if error:raise RuntimeError(error)