import os,json,pathlib,configparser,threading,collections,time,sys
os.umask(0o077);R=pathlib.Path('/evidence')
os.environ['DATABASE_URL']='postgresql://u:p'+chr(64)+'localhost:1/db'
os.environ['REDIS_STREAMS_ENABLED']='false'
from app.multi_pipeline import _DecodedFrameObserver,_pipeline_message_callback,_make_retriever,ResultCollector
from app.stream_lifecycle import RoleRunHandle
from app.source_adapters.rtsp import RtspSourceAdapter
from pyservicemaker import Pipeline,Flow,Probe
cfg=list(pathlib.Path('/configs').glob('*.preprocess'))[0];c=configparser.ConfigParser();c.read(cfg);uid=int(c['property']['unique-id'])
source=json.loads(pathlib.Path('/source.json').read_text())
stages=collections.Counter();lock=threading.Lock();first=threading.Event()
def update(*args):
    with lock:
        if args[2]=='source_stage':stages[args[3][0]]+=1
        if args[2]=='frame':first.set()
handle=RoleRunHandle('det',1,update);collector=ResultCollector(['det']);state={'n':0}
p=Pipeline('live-isolation');state['pipe']=p
prepared=RtspSourceAdapter(source['uri'],source['options'],config_root='/tmp/source-config').prepare('det',1)
probes=[Probe('mux-observer',_DecodedFrameObserver(handle)),Probe('pre-observer',_DecodedFrameObserver(handle,'preprocess_output')),
        Probe('infer-observer',_DecodedFrameObserver(handle,'infer_output'))]
flow=prepared.build_capture(Flow(p),p,decoded_probe=probes[0])
flow=flow.preprocess(str(cfg),target_unique_ids=uid).attach(probes[1]).infer(str(cfg)[:-11],with_triton=True).attach(probes[2])
Retrieval=_make_retriever('det',str(cfg)[:-11],source['uri'],collector,0,state,score_threshold=source['threshold'],
                        role_handle=handle,component_id=uid,is_current=lambda:True)
flow.retrieve(Retrieval())
import ctypes
gst=ctypes.CDLL('libgstreamer-1.0.so.0')
gst.gst_debug_set_threshold_for_name.argtypes=[ctypes.c_char_p,ctypes.c_int]
gst.gst_debug_set_threshold_for_name(b'nvinferserver',5)
handle.bind_pipeline(p);result={'phase':'prepared','finite':False,'frame_sink':False,'tracker':False}
def save():
    with lock:result['stages']=dict(stages)
    result.update(collector=dict(collector.per_task),consumed=state['n'],first_frame=first.is_set(),failure=handle.failure.as_dict() if handle.failure else None)
    (R/'live.json').write_text(json.dumps(result,indent=2))
origin=time.monotonic()
result['timeline']={}
def mark(key):
    result['timeline'][key]=round(time.monotonic()-origin,6)
def worker():
    try:
        callback=_pipeline_message_callback(handle,finite=False)
        def observed(message):
            if type(message).__name__=='EOSMessage':mark('bus_eos')
            callback(message)
        mark('start_enter')
        waiter=handle.start_pipeline(observed)
        result['start_returned']=True;mark('start_returned')
        if waiter:waiter.wait()
        result['wait_returned']=True;mark('wait_returned')
    except Exception as e:result['worker_exception']=type(e).__name__
t=threading.Thread(target=worker,daemon=True);handle.bind_thread(t)
start=time.monotonic();t.start();result['phase']='started';save()
try:
    arrived=first.wait(60)
    result['first_wait_s']=round(time.monotonic()-start,3)
    if arrived:
        end=time.monotonic()+5
        while collector.per_task.get('det',0)<20 and time.monotonic()<end:time.sleep(.05)
    result['phase']='before_stop';save()
    # Stop request return and wait-thread exit are independent facts.
    def stop():
        mark('stop_enter');failure=handle.stop();result['stop_returned']=True;mark('stop_returned')
        result['stop_failure']=failure.failure_stage if failure else None
    stopper=threading.Thread(target=stop,daemon=True);stopper.start();stopper.join(10);t.join(5)
    result.update(phase='finished',stop_thread_alive=stopper.is_alive(),wait_thread_alive=t.is_alive())
    result['pass']=arrived and not stopper.is_alive() and not t.is_alive() and handle.failure is None
    save()
finally:prepared.cleanup()
sys.exit(0 if result.get('pass') else 4)