"""Bounded, read-only numeric Redis observation; never export payloads or credentials."""
import os,sys,time,json,hashlib,signal
import redis
MARKER='contextra-icerti-ac-reader-r01'
stop=False
def ending(*args):
    global stop
    stop=True
signal.signal(signal.SIGTERM,ending)
c=redis.from_url(os.environ['CONTEXTRA_METADATA_REDIS_URL'],decode_responses=True,socket_timeout=3,socket_connect_timeout=3)
ns=os.environ.get('CONTEXTRA_METADATA_NAMESPACE','contextra:metadata:v1');cursors={}
for camera in range(30,46):
    key=ns+':camera:'+hashlib.sha256(str(camera).encode()).hexdigest()+':frames'
    tail=c.xrevrange(key,count=1);cursors[key]=tail[0][0] if tail else '0-0'
print(json.dumps(dict(kind='ready',pid=os.getpid(),marker=MARKER,wall_ms=time.time_ns()//1000000,mono_ns=time.monotonic_ns())),flush=True)
begin=time.monotonic();count=0
try:
    while not stop and time.monotonic()-begin<2200:
        for key,entries in c.xread(cursors,count=100,block=200):
            for sid,fields in entries:
                cursors[key]=sid;s=json.loads(fields['json']);ft=s.get('frameTime') or {}
                if s.get('providerKey') not in ('pose','fire','crowd'):continue
                print(json.dumps(dict(kind='frame',provider=s['providerKey'],camera=s['cameraId'],run=s['runId'],seq=s['frameSeq'],pts=s['ptsNs'],redis_id=sid,
                    read_wall_ms=time.time_ns()//1000000,read_mono_ns=time.monotonic_ns(),ingest_ms=s.get('ingestTimeWallMs'),
                    rtp=ft.get('rtpTimestamp'),clock=ft.get('rtpClockRate'),sync=ft.get('syncStatus'),
                    visible=sum(bool(o.get('visible')) for o in s.get('objects',[])))),flush=True)
                count+=1
                if count>=500000:raise RuntimeError('OBSERVATION_CAP')
finally:
    c.close();print(json.dumps(dict(kind='finished',seconds=time.monotonic()-begin,frames=count)),flush=True)