from ops import *
import signal
import urllib.request
import yaml

ORDER = (3,8)
FIELDS = ('intrusion','loitering','fall','fight','fire','crowd','tracking')
original = yaml.safe_load((PRIVATE/'streams.yaml').read_text())['added']
specs = {c:next(s for s in original if str(s.get('camera_id'))==str(c)) for c in ORDER}
sids = {c:specs[c]['stream_id'] for c in ORDER}
saved = {c:api('/api/v1/streams/'+sids[c]+'/settings')['configured'] for c in ORDER}
save('settings-before-trial.json',saved,private=True)
started, touched, samples = [], [], []
result = dict(errors=[], started=started, touched=touched)

def request(path, body, method='POST'):
    req = urllib.request.Request('http://127.0.0.1:35000'+path,data=json.dumps(body).encode(),
        headers={'Content-Type':'application/json'},method=method)
    with urllib.request.urlopen(req, timeout=20) as r:
        return json.load(r)

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

def rows():
    result = {}
    for c in started:
        r = api('/api/v1/streams/'+sids[c])
        result[str(c)] = {k:r.get(k) for k in ('state','run_id','run_generation','last_error','cleanup_error',
            'cleanup_pending','degraded','publisher_metrics')}
        result[str(c)]['providers'] = {k:{a:v.get(a) for a in ('state','child_pid','last_error','cleanup_error',
            'last_cleanup','metadata_path','publisher_metrics')} for k,v in r.get('provider_processes',{}).items()}
    return result

def check(current):
    assert shutil.disk_usage('/').free > 15*1024**3
    available = int(next(s for s in Path('/proc/meminfo').read_text().splitlines() if s.startswith('MemAvailable:')).split()[1])
    assert available > 2*1024**2, 'HOST_RAM_BOUND'
    for c,r in current.items():
        if r['last_error'] or r['cleanup_error'] or r['degraded']:
            raise RuntimeError('CHANNEL_ERROR_'+c)
        for k,v in r['providers'].items():
            if v['last_error'] or v['cleanup_error'] or v['state']=='error':
                raise RuntimeError('PROVIDER_ERROR_'+c+'_'+k)

def sample():
    now = time.monotonic()
    current = rows()
    check(current)
    mapping = {str(v['child_pid']):c+':'+k for c,r in current.items() for k,v in r['providers'].items() if v['child_pid']}
    resources = json.loads(cmd(['docker','exec','-i','contextra-analytics-1','python3','-B',
        '/opt/contextra/tools/provider_scale_probe.py','resources'], input=json.dumps(mapping),timeout=15))
    assert not int(dict(s.split() for s in resources['cgroup'].get('memory.events','').splitlines()).get('oom_kill',0))
    artifact = api('/api/v1/metadata/status')['artifacts']
    value = dict(t=now,wall_ms=time.time_ns()//1000000,channels=current,resources=resources,
        gpu=cmd(['nvidia-smi','--query-gpu=utilization.gpu,utilization.memory,memory.used','--format=csv,noheader,nounits'],timeout=5).strip(),
        artifacts={k:artifact.get(k) for k in ('active','rendering','counts','error','ledger')})
    with (HERE/'samples.jsonl').open('a') as f:
        f.write(json.dumps(value)+'\n')
    return value

def interrupt(*_):
    raise RuntimeError('TRIAL_INTERRUPTED')

signal.signal(signal.SIGTERM,interrupt)
signal.signal(signal.SIGINT,interrupt)
probe = None
probe_out = None
assert not inspect('contextra-ai-gateway-1')['State']['Running']
try:
    result['begin_wall'] = time.time()
    save('models-before.json', json.loads(cmd(['docker','exec','contextra-analytics-1','python3','-B',
        '/opt/contextra/tools/provider_scale_probe.py','models'],timeout=15)))
    for c in ORDER:
        desired = {k:dict(enabled=True) for k in ('intrusion','loitering','fall','fight','tracking')}
        desired.update(fire=dict(enabled=True,scenario=dict(enabled=True)),crowd=dict(enabled=True,event_enabled=True,
            reference_count=1,safe_ratio=.25,caution_ratio=.5,warning_ratio=.75,danger_ratio=.9,suppress_repeated_danger=True))
        touched.append(c)
        effective = settings(c,desired)['configured']
        assert [p['key'] for p in effective['execution_plan']['providers']]==['pose','fire','crowd']
        started.append(c)
        request('/api/v1/streams/'+sids[c]+'/start',{})
        deadline = time.monotonic()+90
        while True:
            current = rows();check(current)
            r = current[str(c)]
            if len(r['providers'])==3 and all(v.get('metadata_path',{}).get('providers',{}).get(k,{}).get('responses',0)>=2 for k,v in r['providers'].items()):
                break
            if time.monotonic()>deadline:
                raise RuntimeError('NO_RESPONSE_'+str(c))
            time.sleep(1)
        print(json.dumps(dict(camera=c,providers='ready')),flush=True)
    time.sleep(5)
    current = rows();check(current)
    for c in ORDER:
        request('/api/v1/streams/'+sids[c]+'/metrics',dict(run_generation=current[str(c)]['run_generation'],enabled=True,seconds=60))
    target = '/tmp/codex/ltb1021/2026-09-28_crowd-events_r01/redis_probe.py'
    cmd(['docker','exec','contextra-analytics-1','mkdir','-p',str(Path(target).parent)])
    cmd(['docker','cp',str(HERE/'redis_probe.py'),'contextra-analytics-1:'+target])
    credentials = Path('/workspace/contextra-runtime-config/metadata-active/reader/reader-credentials.json')
    assert not credentials.is_symlink() and not credentials.stat().st_mode & 0o077
    probe_out = (HERE/'redis-observation.json').open('x')
    probe = subprocess.Popen(['docker','exec','-i','contextra-analytics-1','python3','-B',target,'3,8','35'],
        stdin=subprocess.PIPE,stdout=probe_out,stderr=subprocess.PIPE,text=True)
    probe.stdin.write(credentials.read_text());probe.stdin.close()
    start = time.monotonic()
    for i in range(7):
        time.sleep(max(0,start+5*i-time.monotonic()))
        samples.append(sample())
        print(json.dumps(dict(sample=i,elapsed=time.monotonic()-start)),flush=True)
    assert probe.wait(15)==0, 'REDIS_PROBE_FAILED'
    probe_out.close();probe_out=None
    result['sample_seconds']=samples[-1]['t']-samples[0]['t']
    result['status']='MEASURED'
except Exception as error:
    import traceback
    result['errors'].append(dict(type=type(error).__name__,reason=str(error)[:160],
        trace=[dict(file=Path(t.filename).name,line=t.lineno,function=t.name) for t in traceback.extract_tb(error.__traceback__)]))
finally:
    if probe is not None and probe.poll() is None:
        probe.wait(65)
    if probe_out is not None:
        probe_out.close()
    for c in started:
        try:
            request('/api/v1/streams/'+sids[c]+'/stop',{})
        except Exception as error:
            result['errors'].append(dict(type=type(error).__name__,reason='STOP_'+str(c)))
    deadline=time.monotonic()+120
    while True:
        current=rows()
        reaped=all(r['state']=='stopped' and not r['cleanup_pending'] and
            all(not v.get('child_pid') for v in r['providers'].values()) for r in current.values())
        if reaped or time.monotonic()>deadline:
            break
        time.sleep(1)
    result['cleanup']=dict(reaped=reaped,channels=current)
    if reaped:
        for c in touched:
            changes={k:saved[c][k] for k in FIELDS}
            if 'scenario' not in changes['fire']:
                changes['fire']=dict(changes['fire'],scenario={'enabled':False})
            settings(c,changes)
        result['settings_restored']=True
    save('trial-result.json',result)
    print(json.dumps(dict(status=result.get('status'),errors=result['errors'],reaped=reaped)),flush=True)
if result['errors'] or not result['cleanup']['reaped']:
    raise SystemExit(1)