from ops import *
import ast,datetime,re,sys

phase=sys.argv[1]
enabled={'pose':{'FALL'},'fire':{'FIRE'},'pose-fire':{'FALL','FIRE'},'pose-fire-crowd':{'FALL','FIRE','CROWD'}}[phase]
keys={'pose':{'pose'},'fire':{'fire'},'pose-fire':{'pose','fire'},'pose-fire-crowd':{'pose','fire','crowd'}}[phase]
source=json.loads((PRIVATE/'selected-source.json').read_text());sid=source['stream_id'];camera=3
assert not inspect('contextra-ai-gateway-1')['State']['Running']
assert api('/api/v1/streams/'+sid)['state']=='stopped'
before=api('/api/v1/streams/'+sid+'/settings')['configured']
save(phase+'-before.json',before,private=True)
save(phase+'-intent.json',dict(camera=3,providers=sorted(keys),image=inspect('contextra-analytics-1')['Image']))
tree=ast.parse((OLD_HERE/'gpu.py').read_text())
exec(compile(ast.Module(body=[n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name=='secure'],type_ignores=[]),'control-helper','exec'))
configs=[]
for name in ('INTRUSION','LOITERING','FALL','FIGHT','FIRE','CROWD'):
    cfg=before[name.lower()]
    if name=='FIRE':cfg=cfg.get('scenario') or {}
    roi=cfg.get('roi') or []
    configs.append(dict(scenarioCode=name,enabled=name in enabled,parameters={},
        rois=[dict(label=cfg.get('roi_name',name+' ROI'),points=roi)] if roi else []))
body=dict(cameraId=3,rtspUrl=source['uri'],startWallTimeMs=time.time_ns()//1000000,rtpClockRate=90000,scenarios=configs)
for k,v in [('recServerId','rec_server_id'),('recCameraId','rec_camera_id')]:
    if source.get(v):body[k]=source[v]

def stop():
    request('/api/v1/streams/'+sid+'/stop',{},timeout=90)
    deadline=time.monotonic()+25;ack=False
    while time.monotonic()<deadline:
        s=api('/api/v1/streams/'+sid)
        if not s.get('child_pid') and not s.get('cleanup_pending'):
            if s['state']=='stopped':return {k:s.get(k) for k in ('state','last_cleanup','cleanup_error','child_pid','cleanup_pending')}
            if s['state']=='error' and not ack:
                request('/api/v1/streams/'+sid+'/stop',{},timeout=10);ack=True
        time.sleep(.5)
    raise RuntimeError('STOP_REAP_FAILED')

out=dict(phase=phase,pass_=False);samples=[]
since=datetime.datetime.now(datetime.timezone.utc).isoformat();begin=time.monotonic()
try:
    out['command']=secure('start',body)
    stable=None
    while time.monotonic()-begin<65:
        s=api('/api/v1/streams/'+sid);m=s.get('metadata_path',{});p=m.get('providers',{})
        samples.append({k:s.get(k) for k in ('state','run_id','run_generation','child_pid','last_error','roles','metadata_path')})
        kernel=cmd(['dmesg','--since','2 minutes ago'],timeout=5)
        fresh=[line for line in kernel.splitlines() if 'segfault' in line and
               (match:=re.match(r'\[\s*([0-9.]+)\]',line)) and float(match[1])>=begin]
        if fresh:out['native_failure']=fresh;raise RuntimeError('NATIVE_CRASH')
        if s['state']=='error':raise RuntimeError('RUNTIME_ERROR')
        if any(v['state']!='ACTIVE' for v in p.values()):raise RuntimeError('PROVIDER_UNAVAILABLE')
        if s['state']=='running' and set(p)==keys and all(v['responses']>=3 for v in p.values()):
            stable=stable or time.monotonic()
            if time.monotonic()-stable>=5:
                out['last']=samples[-1];break
        time.sleep(1)
    if 'last' not in out:raise RuntimeError('FIRST_RESULT_TIMEOUT')
    out['publisher']=api('/api/v1/metadata/status')
    # Read-only current run summary; no raw payload or image is retained.
    code='''import sys,json,hashlib,redis,os
from urllib.parse import urlsplit
from redis.retry import Retry
from redis.backoff import NoBackoff
x=json.load(sys.stdin);c=x['connection'];a=x['auth']
u=urlsplit(os.environ['CONTEXTRA_METADATA_REDIS_URL'])
r=redis.Redis(host=u.hostname,port=u.port or 6379,username=a['username'],password=a['password'],decode_responses=True,socket_timeout=2,socket_connect_timeout=2,retry=Retry(NoBackoff(),0))
key=c['namespace']+':camera:'+hashlib.sha256(b'3').hexdigest()+':frames'
rows=r.xread({key:'0-0'},count=5);out=[]
for _,entries in rows:
 for _,row in entries:
  d=json.loads(row['json']);item={k:d.get(k) for k in ('frameSeq','runId','runGeneration','ptsNs','frameTime')}
  objects=d.get('objects',[]);refs=[o['objectRef'] for o in objects]
  item.update(objects=len(objects),refs_unique=len(refs)==len(set(refs)),models={k:sum(o['modelKey']==k for o in objects) for k in set(o['modelKey'] for o in objects)})
  out.append(item)
print(json.dumps(out))'''
    root=Path('/workspace/contextra-runtime-config/metadata-active/reader')
    access=dict(connection=json.loads((root/'connection.json').read_text()),auth=json.loads((root/'reader-credentials.json').read_text()))
    out['metadata_frames']=json.loads(cmd(['docker','exec','-i','contextra-analytics-1','timeout','6','python3','-B','-c',code],input=json.dumps(access),timeout=8))
    if not out['metadata_frames']:raise RuntimeError('NO_METADATA_DELIVERED')
    assert all(x['runId']==s['run_id'] for x in out['metadata_frames'])
    assert all(x['objects']<=300 and x['refs_unique'] for x in out['metadata_frames'])
    seqs=[x['frameSeq'] for x in out['metadata_frames']]
    assert all(a<b for a,b in zip(seqs,seqs[1:]))
    code='''import sys,json,os,socket
from pathlib import Path
x=json.load(sys.stdin);pid=x['pid'];root=Path('/proc')/str(pid)
inodes={os.readlink(p)[8:-1] for p in (root/'fd').iterdir() if os.readlink(p).startswith('socket:[')}
target=socket.inet_aton(socket.gethostbyname(x['host']))[::-1].hex().upper()+':'+format(x['port'],'04X')
connections=[row for row in (root/'net/tcp').read_text().splitlines()[1:] if row.split()[2]==target and row.split()[3]=='01' and row.split()[9] in inodes]
names=[p.read_text().strip() for p in (root/'task').glob('*/comm')]
print(json.dumps({'rtsp_connections':len(connections),'pipeline_threads':{n:names.count(n) for n in set(names) if any(k in n for k in ('decoder','provider','batch_capture'))}}))
'''
    from urllib.parse import urlsplit
    u=urlsplit(source['uri'])
    out['runtime_structure']=json.loads(cmd(['docker','exec','-i','contextra-analytics-1','timeout','5','python3','-B','-c',code],input=json.dumps(dict(pid=s['child_pid'],host=u.hostname,port=u.port or 554)),timeout=7))
    assert out['runtime_structure']['rtsp_connections']==1
    out['pass_']=True
except Exception as e:
    out['error']=type(e).__name__+': '+str(e)[:200]
    print(json.dumps(dict(phase=phase,failed=True,error=out['error'])),flush=True)
finally:
    out['stop']=stop()
    current=api('/api/v1/streams/'+sid+'/settings')['configured']
    fields=('intrusion','loitering','fall','fight','fire','crowd','tracking','frame_time_anchor')
    changes={k:before[k] for k in fields if current[k]!=before[k]}
    if 'fire' in changes and 'scenario' not in changes['fire']:
        changes['fire']=dict(changes['fire'],scenario={'enabled':False})
    if changes:request('/api/v1/streams/'+sid+'/settings',dict(expected_revision=current['revision'],**changes),method='PATCH')
    save(phase+'-samples.json',samples)
    p=subprocess.run(['docker','logs','--since',since,'--tail','180','contextra-analytics-1'],capture_output=True,text=True,timeout=8)
    safe=re.sub(r'rtsp[s]?://[^\s\"\']+','<RTSP_REDACTED>',p.stdout+p.stderr)
    (HERE/(phase+'-runtime.log')).write_text(safe[-30000:])
    out['elapsed_seconds']=round(time.monotonic()-begin,2)
    # Global publisher status may contain endpoint values; retain only aggregate counters.
    pub=out.pop('publisher',None)
    if pub is not None:out['publisher_keys']=list(pub)
    save(phase+'-result.json',out)
    print(json.dumps(dict(phase=phase,pass_=out['pass_'],error=out.get('error'),
        providers=out.get('last',{}).get('metadata_path',{}).get('providers'),stop=out['stop'])),flush=True)
raise SystemExit(0 if out['pass_'] else 1)