import collections,json,os,subprocess,time,urllib.request
from pathlib import Path
os.umask(0o077)
ROOT=Path('/backup/contextra/deployment-inbox/icerti-provider-20260927-r01')
SID='camera-c64e6ce645495b73ab536746ec8714af'
def api(path,method='GET',body=None):
    r=urllib.request.Request('http://127.0.0.1:35000'+path,method=method,data=json.dumps(body).encode() if body is not None else None,headers={'Content-Type':'application/json'})
    return json.load(urllib.request.urlopen(r,timeout=30))
def save(name,data):
    with (ROOT/('canary-r02-'+name)).open('x') as f:json.dump(data,f,indent=2)
def row():return next(r for r in api('/api/v1/streams')['streams'] if r['stream_id']==SID)
def stop():
    api('/api/v1/streams/'+SID+'/stop','POST',{})
    until=time.monotonic()+100
    while time.monotonic()<until:
        r=row()
        if r['state']=='stopped' and not r.get('child_pid') and not r.get('cleanup_pending'):return r
        time.sleep(1)
    raise RuntimeError('CANARY_REAP')
def patch(changes):
    current=api('/api/v1/streams/'+SID+'/settings')['configured']
    return api('/api/v1/streams/'+SID+'/settings','PATCH',dict(expected_revision=current['revision'],**changes))
assert all(r['state']=='stopped' for r in api('/api/v1/streams')['streams'])
original=api('/api/v1/streams/'+SID+'/settings')['configured']
save('canary-settings-before.json',original)
fields=('pose','tracking','intrusion','loitering','fall','fight','fire','crowd')
reader='''import os,time,json,collections,hashlib,redis
r=redis.from_url(os.environ['CONTEXTRA_METADATA_REDIS_URL'],decode_responses=True,socket_timeout=2)
key=os.environ.get('CONTEXTRA_METADATA_NAMESPACE','contextra:metadata:v1')+':camera:'+hashlib.sha256(b'34').hexdigest()+':frames'
counts=collections.Counter();objects=collections.Counter();seen={};gaps=collections.Counter();bad=[];fields={};pts={};end=time.monotonic()+20;cursor='0-0'
while time.monotonic()<end:
 for _,messages in r.xread({key:cursor},count=100,block=500):
  for mid,data in messages:
   cursor=mid;p=json.loads(data['json']);provider=p.get('providerKey','missing');counts[provider]+=1
   fields[provider]=dict(providerModel=p.get('providerModel'),requiredProviders=p.get('requiredProviders'),keys=sorted(p))
   objects[provider]+=len(p.get('objects',[]))
   frame=p.get('frameSeq'); stamp=p.get('ptsNs');pts[provider]=stamp
   if frame is not None and provider in seen and frame<=seen[provider]:bad.append(provider)
   if frame is not None:seen[provider]=frame
print(json.dumps(dict(counts=counts,objects=objects,fields=fields,last_pts=pts,order_errors=bad,length=r.xlen(key),ttl=r.ttl(key))))
'''
try:
    for mode in ('pose','mixed'):
        changes={k:dict(original[k],enabled=True) for k in ('pose','tracking','intrusion','loitering','fall','fight')}
        changes['fire']=dict(original['fire'],enabled=mode=='mixed',scenario=dict(original['fire'].get('scenario',{}),enabled=mode=='mixed'))
        changes['crowd']=dict(original['crowd'],enabled=mode=='mixed')
        patch(changes)
        api('/api/v1/streams/'+SID+'/start','POST',{})
        deadline=time.monotonic()+90
        while time.monotonic()<deadline:
            current=row()
            if current['state']=='error':raise RuntimeError('CANARY_'+mode+'_ERROR')
            if current['state']=='running':break
            time.sleep(1)
        else:raise RuntimeError('CANARY_START_TIMEOUT')
        output=subprocess.check_output(['docker','exec','contextra-icerti-analytics-1','python3','-c',reader],text=True,timeout=30)
        result=json.loads(output)
        result['runtime']=row()
        save('canary-'+mode+'.json',result)
        expected={'pose'} if mode=='pose' else {'pose','fire','crowd'}
        assert set(result['counts'])==expected,(mode,'PROVIDER_OUTPUT_MISSING',result['counts'])
        assert not result['order_errors'],'FRAME_ORDER'
        for item in result['fields'].values():assert set(item['requiredProviders'])==expected and item['providerModel']
        result=stop();save('canary-'+mode+'-stop.json',result)
        print(json.dumps(dict(stage=mode,status='PASS')),flush=True)
finally:
    stop()
    restore={k:original[k] for k in fields}
    restore['fire']=dict(original['fire'],scenario=dict(original['fire'].get('scenario',{}),enabled=original['fire'].get('scenario',{}).get('enabled',False)))
    patch(restore)
    restored=api('/api/v1/streams/'+SID+'/settings')['configured']
    assert all(restored[k]==original[k] for k in fields if k!='fire'),'SETTINGS_RESTORE'
    assert restored['fire']['enabled']==original['fire']['enabled']
    save('canary-restored.json',dict(camera_id=34,stopped=True,settings_equal_except_materialized_fire=True,exact_saved_spec_restore_pending=True))