import hashlib,json,os,pathlib,signal,subprocess,time,urllib.parse,urllib.request
from apply import a,plan,built,C,R,g,cmd
os.umask(0o077)
env=g.load_env_file(str(g.ENV));base='http://127.0.0.1:'+env['FASTAPI_HOST_PORT']
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
access=json.loads((C/'access.json').read_text());passwords=json.loads((C/'secrets.json').read_text())
document=g.yaml.safe_load(a.overlay.read_text());spec=document['added'][0]
path='/api/v1/streams/'+urllib.parse.quote(spec['stream_id'],safe='')
key=hashlib.sha256(spec['source_id'].encode()).hexdigest()
stream='contextra:metadata:v1:camera:'+key+':frames'
status_key='contextra:metadata:v1:camera:'+key+':status'
visual=R/'visual';flag=visual/'capture.enabled'
result=dict(stage='preflight',calls=[],observations=[],first_frame_cap_s=60,live_after_first_s=8,cleanup_cap_s=70,frame_cap=1000)
def save():(R/'runtime.json').write_text(json.dumps(result,indent=2))
def api(p,method='GET',body=None,timeout=3):
 data=None if body is None else json.dumps(body).encode()
 req=urllib.request.Request(base+p,data=data,method=method,headers={'Content-Type':'application/json'})
 with opener.open(req,timeout=timeout) as f:return f.status,json.load(f)
def redis(role,*args):
 p=subprocess.run(['redis-cli','--json','-h',access['host'],'-p',str(access['port']),
   '--user','contextra_'+role,*args],env=dict(os.environ,REDISCLI_AUTH=passwords[role]),
   capture_output=True,text=True,timeout=3)
 assert p.returncode==0,'REDIS_READ_FAILED'
 return json.loads(p.stdout)
def resources():
 mem={l.split(':')[0]:int(l.split()[1]) for l in pathlib.Path('/proc/meminfo').read_text().splitlines()}
 assert mem['MemAvailable']>4*1024**2 and g.gpu()-plan['gpu_mib']<8192,'RESOURCE_CAP'
 c=g.inspect();assert c['Image']==built['image_id'] and c['State']['Status']=='running' and not c['RestartCount']
def status():
 _,v=api(path);resources()
 mp=v.get('metadata_path') or {}
 assert mp.get('processed',0)<=1000,'FRAME_CAP'
 result['observations'].append(dict(at=time.monotonic(),state=v['state'],cleanup_pending=v.get('cleanup_pending'),
  metadata_path=mp,last_cleanup=v.get('last_cleanup'),cleanup_error=bool(v.get('cleanup_error')),last_error=bool(v.get('last_error'))))
 save();return v
def command(action):
 row=dict(action=action,at=time.monotonic());result['calls'].append(row);save()
 code,v=api(path+'/'+action,'POST',timeout=10)
 row.update(http=code,returned=time.monotonic(),state=v.get('state',v.get('observed_state')));save()
def collect():
 entries=redis('reader','XRANGE',stream,'-','+','COUNT','30') or []
 seen={}
 for _,fields in entries:
  if isinstance(fields,list):fields=dict(zip(fields[::2],fields[1::2]))
  value=json.loads(fields['json'])
  assert value['sourceId']==spec['source_id'] and value['serverId']==spec['server_id']
  assert value['normalized'] and value['frameWidth']>0 and value['frameHeight']>0
  assert value['runId']==result['run_id']
  assert all(o['visible'] and o['confidence']>.1 for o in value['objects'])
  assert all(0<=o['bbox'][0]<o['bbox'][2]<=1 and 0<=o['bbox'][1]<o['bbox'][3]<=1 for o in value['objects'])
  seen[value['frameSeq']]=value
 result['redis_frames_read']=result.get('redis_frames_read',0)+len(entries)
 result['empty_frame_seen']=result.get('empty_frame_seen',False) or any(not v['objects'] for v in seen.values())
 for file in visual.glob('capture-*/frame-*.json'):
  pair=json.loads(file.read_text());seq=pair['frameSeq']
  if seq in seen:
   assert seen[seq]==pair,'FRAME_PAYLOAD_MISMATCH'
   target=R/'samples'/file.name
   if not target.exists():target.write_text(json.dumps(seen[seq],ensure_ascii=False))
 save()
def alarm(*unused):raise TimeoutError('HELPER_DEADLINE')
def main():
 assert not (R/'runtime.json').exists()
 assert json.loads((R/'apply.json').read_text())['stage']=='ready'
 signal.signal(signal.SIGALRM,alarm);signal.alarm(200)
 patched=False;started=False;before=None;after_hash=None
 save()
 try:
  assert g.sha(a.overlay)==plan['overlay_hash'] and g.sha(g.CFG)==plan['config_hash']
  assert g.sha(g.CFG.parent/'2/model.plan')==plan['plan_hash']
  assert status()['state']=='stopped'
  for p in ('/live','/ready','/health'):api(p)
  versions=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",timeout=3))["versions"]))']))
  assert versions==['2']
  for role in passwords:assert redis(role,'PING')=='PONG'
  assert redis('reader','HGETALL','contextra:metadata:v1:channels')
  for role,args in [('publisher',['XADD',stream,'*','json','{}']),('reader',['XREAD','STREAMS',stream,'0-0'])]:
   assert redis('admin','ACL','DRYRUN','contextra_'+role,*args)=='OK'
  # ACL DRYRUN validates denied commands without performing writes.
  for args in (['SET','contextra:metadata:v1:probe','x'],['GET','contextra:stream:frames'],['CONFIG','GET','maxmemory']):
   deny=redis('admin','ACL','DRYRUN','contextra_reader',*args)
   assert deny!='OK'
  cmd(['python3',str(C/'firewall.py'),'check'])
  result['auth_acl_host_lan']='PASS';result['windows_actual']='WAIT'
  _,before=api(path+'/settings');before=before['configured']
  _,changed=api(path+'/settings','PATCH',dict(expected_revision=before['revision'],display_score=.1))
  assert changed['configured']['display_score']==.1
  assert changed['configured']['downstream_score']==before['downstream_score']
  patched=True;after_hash=g.sha(a.overlay);result['overlay_after']=after_hash;save()
  (R/'samples').mkdir(mode=0o700);flag.write_text('enabled\n')
  started=True;command('start');deadline=time.monotonic()+60;first=None
  while time.monotonic()<deadline:
   v=status();assert not v.get('last_error') and not v.get('cleanup_error'),'PIPELINE_ERROR'
   if v['state']=='running' and (v.get('metadata_path') or {}).get('processed',0)>0:
    first=v;result['run_id']=v['run_id'];result['first_result_at']=time.monotonic();break
   time.sleep(.5)
  assert first is not None,'FIRST_RESULT_TIMEOUT'
  end=time.monotonic()+8
  while time.monotonic()<end:
   collect();v=status()
   assert not v.get('last_error') and not v.get('cleanup_error')
   time.sleep(.5)
  values=[json.loads(p.read_text()) for p in (R/'samples').glob('*.json')]
  result['visual_pairs']=len(list(visual.glob('capture-*/*.jpg')))
  result['redis_visual_matches']=len(values)
  assert 0<len(values)<=3 and result['visual_pairs']==len(values),'VISUAL_MATCH_INCOMPLETE'
  result['visible_counts']=[len(v['objects']) for v in values]
  tracks=[{o['trackId'] for o in v['objects'] if o['trackId'] is not None} for v in values]
  result['track_counts']=[len(x) for x in tracks]
  result['shared_track']=bool(set.intersection(*tracks)) if len(tracks)>1 else False
  result['publisher_before_stop']=api('/api/v1/metadata/status')[1]
  result['metadata_verified']=True;save()
 except Exception as e:
  result['error']=type(e).__name__
  result['error_code']=str(e) if str(e) in ('FIRST_RESULT_TIMEOUT','PIPELINE_ERROR','VISUAL_MATCH_INCOMPLETE','FRAME_PAYLOAD_MISMATCH','RESOURCE_CAP','REDIS_READ_FAILED') else 'BOUNDED_STEP_FAILED'
  save()
 finally:
  signal.alarm(0)
  if flag.exists():flag.unlink()
  if started:
   try:
    command('stop');end=time.monotonic()+70
    while time.monotonic()<end:
     v=status()
     if v.get('cleanup_error') or v.get('last_error'):break
     if v['state']=='stopped' and not v.get('cleanup_pending'):
      assert not v.get('runtime_observation') and (v.get('last_cleanup') or {}).get('completed')
      result['cleanup_verified']=True;result['cleanup']=v['last_cleanup'];break
     time.sleep(.5)
    state=redis('reader','GET',status_key)
    result['redis_stopped']=json.loads(state)['state']=='STOPPED' if state else False
   except Exception as e:result['stop_error']=type(e).__name__
  if patched:
   try:
    assert g.sha(a.overlay)==after_hash
    _,current=api(path+'/settings')
    _,restored=api(path+'/settings','PATCH',dict(expected_revision=current['configured']['revision'],display_score=before['display_score']))
    assert restored['configured']['display_score']==before['display_score']
    result['display_restored']=True;result['overlay_final']=g.sha(a.overlay)
   except Exception as e:result['display_restore_error']=type(e).__name__
  result['stage']='passed' if result.get('metadata_verified') and result.get('cleanup_verified') and result.get('display_restored') else 'failed'
  save()
  if result['stage']=='failed':
   cmd(['docker','stop','--timeout','20',g.NAME],40)
   result['analytics_safety_stop']=True;save()
  print(json.dumps({k:result.get(k) for k in ['stage','auth_acl_host_lan','metadata_verified','visual_pairs','redis_visual_matches','visible_counts','track_counts','shared_track','cleanup_verified','redis_stopped','error','error_code','display_restore_error']}))
if __name__=='__main__':main()