import csv
import json
import os
import pathlib
import subprocess
import urllib.request
from preflight import R,g,HEAD

os.umask(0o077)
run=json.loads((R/'run.json').read_text())
plan=json.loads((R/'plan.json').read_text())
built=json.loads((R/'build.json').read_text())
source=json.loads((R/'source.json').read_text())
assert run['final']['success'] and run['cleanup_verified']
c=g.inspect()
assert c['Image']==built['image_id'] and c['State']['Health']['Status']=='healthy'
assert c['HostConfig']['RestartPolicy']==plan['analytics']['HostConfig']['RestartPolicy']
assert g.others()==plan['other_containers']
assert all(g.sha(g.REPO/p)==h for p,h in source['hashes'].items())
assert g.sha(g.CFG)==plan['config_hash'] and g.sha(g.CFG.parent/'2/model.plan')==plan['plan_hash']
assert g.sha('/data/contextra/runtime-state/streams_runtime.yaml')==plan['overlay_hash']
env=g.load_env_file(str(g.ENV))
actual=dict(v.split('=',1) for v in c['Config']['Env'])
C=g.ENV.parent/'metadata-active'
active=g.yaml.safe_load((C/'compose.yaml').read_text())['services']['analytics']
assert active['image']==c['Image'] and env['CONTEXTRA_IMAGE']==built['tag']
assert active['restart']=='on-failure:3'
assert not actual.get('CONTEXTRA_METADATA_VISUAL_DIR') and not actual.get('CONTEXTRA_METADATA_SAMPLE_DIR')
assert all(actual[k]==active['environment'][k] for k in ('REDIS_URL','CONTEXTRA_METADATA_REDIS_URL','CONTEXTRA_METADATA_ENABLED'))
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open('http://127.0.0.1:'+env['FASTAPI_HOST_PORT']+'/api/v1/streams',timeout=3) as f:
    assert all(x['state']=='stopped' for x in json.load(f)['streams'])
script='''import json,pathlib
out=[]
for p in pathlib.Path('/proc').glob('[0-9]*'):
 try:
  cmd=(p/'cmdline').read_bytes().split(b'\\0')
  role='app' if b'app.main' in cmd and cmd[0].rsplit(b'/',1)[-1].startswith(b'python') else 'triton' if cmd and cmd[0].endswith(b'tritonserver') else None
  if role:
   env=dict(x.split(b'=',1) for x in (p/'environ').read_bytes().split(b'\\0') if b'=' in x)
   out.append(dict(role=role,blas=env.get(b'OPENBLAS_NUM_THREADS',b'unset').decode()))
 except (OSError,ValueError):pass
print(json.dumps(out))'''
processes=json.loads(g.sh(['docker','exec',g.NAME,'python3','-c',script]))
assert {'role':'app','blas':'1'} in processes and {'role':'triton','blas':'unset'} in processes
stats=json.loads(g.sh(['docker','exec',g.NAME,'python3','-c',
    'import json,urllib.request;print(json.dumps(json.load(urllib.request.urlopen("http://localhost:8000/v2/models/RFDETRMedium/stats",timeout=3))["model_stats"][0]["inference_stats"]))']))
spec=g.yaml.safe_load(pathlib.Path('/data/contextra/runtime-state/streams_runtime.yaml').read_text())['added'][0]
stream='contextra:metadata:v1:camera:'+g.hashlib.sha256(spec['source_id'].encode()).hexdigest()+':frames'
access=json.loads((C/'access.json').read_text())
password=json.loads((C/'secrets.json').read_text())['reader']
p=subprocess.run(['redis-cli','--json','-h',access['host'],'-p',str(access['port']),
    '--user','contextra_reader','XREVRANGE',stream,'+','-','COUNT','1'],
    env=dict(os.environ,REDISCLI_AUTH=password),capture_output=True,text=True,timeout=3)
assert p.returncode==0
entries=json.loads(p.stdout)
# Frame streams have a short TTL; post-Stop absence is not a failed live publish.
rtp={'timeSource':None,'syncStatus':None}
if entries:
    fields=entries[0][1]
    if isinstance(fields,list):fields=dict(zip(fields[::2],fields[1::2]))
    payload=json.loads(fields['json'])
    assert payload['runId']==run['run_id']
    rtp={k:payload.get(k) for k in ('rtpTimestamp','rtpClockRate','detectedAtWallMs','timeSource','syncStatus')}
phase=run['phases'][0];a,b=phase['after'],phase['before']
seconds=(a['sampled_ns']-b['sampled_ns'])/1e9
counts={k:v-b['counts'].get(k,0) for k,v in a['counts'].items()}
fps={k:v/seconds for k,v in counts.items()}
redis_count=phase['publisher_after']['sent_frames']-phase['publisher_before']['sent_frames']
rows=run.get('resources',[])
cpu=[(b['cpu_usage_usec']-a['cpu_usage_usec'])/((b['monotonic_ns']-a['monotonic_ns'])/1000)*100 for a,b in zip(rows,rows[1:])]
out=dict(measure_s=seconds,counts=counts,fps=fps,redis_frames=redis_count,redis_fps=redis_count/seconds,
    durations=a['durations'],triton_whole_run_mean_ms={k:v['ns']/v['count']/1e6 if v['count'] else None for k,v in stats.items()},
    first_result_s=(run['first_result_ns']-run['calls'][0]['requested_ns'])/1e9,
    cleanup_s=run['observations'][-1]['last_cleanup']['elapsed_s'],
    previous_fps=.9608955,target_fps=10,target_approximately_achieved=9.5<=fps['result']<=10.5,
    process_blas=processes,post_stop_redis_entries=len(entries),
    rtp_field_recheck='NOT_CAPTURED_BEFORE_TTL' if not entries else 'READ',
    time_source=rtp['timeSource'],sync_status=rtp['syncStatus'],
    gpu_peak_percent=max(x['gpu_percent'] for x in rows),gpu_peak_mib=max(x['gpu_mib'] for x in rows),
    gpu_idle_after_mib=g.gpu(),cpu_peak_percent=max(cpu),memory_peak_mib=max(x['memory_bytes'] for x in rows)/1024**2,
    publisher_final_metrics=phase['publisher_after'],
    image=built['image_id'],analytics='healthy',stream='stopped',detail_off=True,capture_off=True,
    overlay_unchanged=True,non_targets_unchanged=True,builds=1,live_windows=1,failed_groups=0)
(R/'summary.json').write_text(json.dumps(out,indent=2))
with (R/'summary.csv').open('w') as f:
    w=csv.writer(f);w.writerow(['stage','count','fps'])
    w.writerows((k,counts[k],fps[k]) for k in counts)
    w.writerow(['redis_sent',redis_count,redis_count/seconds])
assert not g.sh(['docker','ps','-aq','--filter','name=codex-receiver-']).strip()
print(json.dumps(out,indent=2))