"""One simultaneous real-input window; bounded observation and guaranteed Stop."""
import concurrent.futures
import copy
import hashlib
import json
import os
from pathlib import Path
import re
import time
import urllib.request
import urllib.error

OUT=Path('/trial/runtime/acceptance')
OUT.mkdir(mode=0o700,exist_ok=True)
assert not (OUT/'started.json').exists(), 'REAL_INPUT_TEST_ALREADY_STARTED'
TOKEN=Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip()
REQUESTS=json.loads(Path('/etc/contextra/requests.json').read_text())
RESULTS={'scope':'ISOLATED_REAL_INPUT','cameras':[3,5],'responses':[],'observations':[],
         'checks':{},'absolute_time_validation':False}

def safe(value):
    if isinstance(value,dict):return {k:safe(v) for k,v in value.items() if k.lower() not in ('uri','rtspurl','password','token','authorization')}
    if isinstance(value,list):return [safe(v) for v in value]
    if isinstance(value,str):return re.sub(r'(?:rtsp|https?|redis|postgresql)://\S+','<redacted-endpoint>',value)
    return value

def save():
    p=OUT/'result.next';p.write_text(json.dumps(safe(RESULTS),ensure_ascii=False,indent=2));p.replace(OUT/'result.json')

def http(path,body=None):
    req=urllib.request.Request('http://127.0.0.1:5000'+path,data=json.dumps(body).encode() if body is not None else None,
        headers={'Authorization':'Bearer '+TOKEN,'Content-Type':'application/json'})
    try:r=urllib.request.urlopen(req,timeout=8)
    except urllib.error.HTTPError as error:r=error
    with r:return r.status,json.load(r)

def call(camera,kind,body):
    start=time.time();status,value=http('/internal/secure-control/'+kind,body)
    RESULTS['responses'].append(dict(camera=camera,kind=kind,http=status,response=value,at=start))
    return status,value

def get_status():
    status,data=http('/internal/secure-control/runtime-status')
    assert status==200
    return data

from app.multi_roi.adapter import normalize
from app.multi_roi.policy import PolicyStore
policy=PolicyStore(os.environ['CONTEXTRA_ROI_POLICY_FILE'])
assert policy.value['mode']=='multi_roi'
for value in REQUESTS.values():normalize(copy.deepcopy(value),policy.value)
import redis
REDIS=redis.from_url(os.environ['CONTEXTRA_METADATA_REDIS_URL'],decode_responses=True,
                     socket_connect_timeout=3,socket_timeout=3)
assert REDIS.ping()
assert get_status()['streamCounts']['total']==0
STARTED=time.time()
(OUT/'started.json').write_text(json.dumps({'at':STARTED,'cameras':[3,5]}))
seen={3:set(),5:set()}; run_ids={};duplicate_done=False;reject_done=False
try:
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
        jobs=[pool.submit(call,camera,'start',REQUESTS[str(camera)]) for camera in (3,5)]
        for job in jobs:
            status,value=job.result(timeout=12)
            assert status==202, safe(value)
    save()
    deadline=time.monotonic()+110
    last_sample=0;first_both=None
    while time.monotonic()<deadline:
        now=time.monotonic();state=get_status()
        current={int(s['cameraId']):s for s in state['streams']}
        obs=dict(at=time.time(),counts=state['streamCounts'],streams=state['streams'])
        RESULTS['observations'].append(obs)
        assert not state['streamCounts']['error'], 'LIVE_START_ERROR'
        for camera in (3,5):
            key='mroi-trial:camera:'+hashlib.sha256(str(camera).encode()).hexdigest()+':frames'
            for ident,fields in REDIS.xrevrange(key,count=16):
                snapshot=json.loads(fields['json']);seq=(snapshot.get('runId'),snapshot.get('frameSeq'))
                seen[camera].add(seq)
                if now-last_sample>=8:
                    target=OUT/f'camera-{camera}';target.mkdir(exist_ok=True)
                    (target/f'frame-{int(time.time()*1000)}.json').write_text(json.dumps(snapshot))
                    break
        if now-last_sample>=8:last_sample=now
        if all(seen.values()) and first_both is None:first_both=now
        if all(c in current and current[c]['observedState']=='RUNNING' for c in (3,5)):
            if not run_ids:run_ids={c:current[c]['runId'] for c in (3,5)}
            if not duplicate_done and now-(first_both or now)>=12:
                status,value=call(3,'start',REQUESTS['3'])
                assert status==202
                RESULTS['checks']['same_start_action']=value.get('change',{}).get('action')
                assert value.get('change',{}).get('action')=='RUN_REUSED', 'SAME_START_NOT_REUSED'
                duplicate_done=True
            if duplicate_done and not reject_done:
                invalid=copy.deepcopy(REQUESTS['3']);invalid['rules'][0]['roiId']=999999
                status,value=call(3,'start',invalid)
                assert status==422 and value.get('error')=='ROI_REFERENCE_NOT_FOUND'
                RESULTS['checks']['invalid_request']=value.get('error');reject_done=True
            if duplicate_done:assert all(current[c]['runId']==run_ids[c] for c in (3,5)), 'RUN_REPLACED_UNEXPECTEDLY'
        RESULTS['frames_observed']={str(c):len(v) for c,v in seen.items()}
        save()
        if first_both is not None and now-first_both>=65:break
        time.sleep(3)
    RESULTS['checks']['both_produced_frames']=all(seen.values())
    RESULTS['checks']['same_runs_after_rejection']=bool(run_ids) and reject_done
    assert all(seen.values()), 'FRAME_NOT_OBSERVED'
except Exception as error:
    RESULTS['error']=safe(type(error).__name__+': '+str(error))
finally:
    for camera in (3,5):
        try:call(camera,'stop',{'cameraId':camera})
        except Exception as error:RESULTS.setdefault('stop_errors',[]).append(type(error).__name__)
    end=time.monotonic()+90
    while time.monotonic()<end:
        status=get_status();RESULTS['after_stop']=status
        if status['streamCounts']['stopped']==status['streamCounts']['total']:break
        time.sleep(2)
    RESULTS['finishedAt']=time.time();save();REDIS.close()
print(json.dumps({'finished':True,'error':RESULTS.get('error'),'frames':RESULTS.get('frames_observed'),
                  'checks':RESULTS['checks'],'after_stop':RESULTS['after_stop']['streamCounts']}))