import copy, fcntl, hashlib, json, math, os, shutil, subprocess, sys, time, traceback from pathlib import Path from urllib.parse import quote import yaml os.umask(0o077) R=Path(__file__).parent;R.chmod(0o700) REPO=Path('/workspace/contextra-analytics-runtime-lifecycle-refactor');sys.path.insert(0,str(REPO)) from tools.stream_benchmark import load_config, preflight, Benchmark from tools.stream_benchmark_runtime import Runtime, file_hash OLD=Path('/workspace/contextra-runtime-config/test-run/medium8-fp16-sharing.yaml') NEW=Path('/workspace/contextra-runtime-config/test-run/medium8-tracking-intrusion.yaml') ACTIVE=Path('/workspace/contextra-runtime-config/metadata-active/compose.yaml') ENV=Path('/workspace/contextra-runtime-config/runtime.env') def save(n,v): p=R/n;p.write_text(json.dumps(v,ensure_ascii=False,indent=2,allow_nan=False));p.chmod(0o600) def read(n):return json.loads((R/n).read_text()) def cmd(a):return subprocess.check_output(a,text=True,timeout=20) def gate(): assert not (R/'plan.json').exists() and not NEW.exists() cfg=load_config(OLD);rt=Runtime(cfg,R);plan=preflight(rt) assert len(plan['rows'])==16 and cfg['model']==dict(name='RFDETRMedium',version=4,role='det',instances=8,max_batch=4,queue_delay_us=5000) assert all(plan['states'][r['stream_id']]['state']==('running' if r['original'] else 'stopped') for r in plan['rows']) for r in plan['rows']: s=plan['settings'][r['stream_id']];i=s['intrusion'] assert i['roi']==[[.35,.3],[.85,.3],[.85,1],[.35,1]] and i['confidence']==.5 and i['dwell_seconds']==.5 assert not i['enabled'] and not s['tracking']['enabled'] for p,n in ((Path(cfg['registration_file']),'registry.before'),(OLD,'benchmark.before')):shutil.copy2(p,R/n) presets={str(p):dict(sha256=file_hash(p),data=yaml.safe_load(p.read_text())) for p in Path('/workspace/contextra-runtime-config/test-run/presets').rglob('*.yaml')} inspect=rt.inspect() plan.update(head=cmd(['git','-C',str(REPO),'rev-parse','HEAD']).strip(),started=time.time(),config=cfg,presets=presets, active_hash=file_hash(ACTIVE),env_hash=file_hash(ENV),container_id=inspect['Id'], redis_id=json.loads(cmd(['docker','inspect','contextra-redis-1']))[0]['Id'], sender=json.loads(cmd(['python3',str(REPO/'tools/rtsp_test.py'),'list'])), policy='one ascent; on failure stop all targets; no restore/restart; preserve registrations/presets') save('plan.json',plan);print('Gate PASS: approved partial ROI present in all16, baseline locked',flush=True) def prepare(): plan=read('plan.json');cfg=plan['config'];rt=Runtime(cfg,R) for row in plan['rows']: if row['original']:save('original-stop-'+row['stream_id']+'.json',rt.stop(row['stream_id'])) assert all(q['state']=='stopped' for q in rt.states().values()) for row in plan['rows']: sid=row['stream_id'];before=rt.settings(sid);assert before==plan['settings'][sid] change=dict(expected_revision=before['revision'],tracking=dict(enabled=True),intrusion=dict(enabled=True)) marker=R/('patch-'+hashlib.sha256(sid.encode()).hexdigest()[:16]+'.json') with marker.open('x') as f:json.dump(dict(sid=sid,request=change,before=before),f) response=rt.api.call('PATCH','/api/v1/streams/'+quote(sid,safe='')+'/settings',change) after=response['configured'];expected=copy.deepcopy(before);expected['tracking']['enabled']=True;expected['intrusion']['enabled']=True expected['revision']=after['revision'];assert expected==after and not response['restart_pending'] save('applied-'+hashlib.sha256(sid.encode()).hexdigest()[:16]+'.json',response) cfg=copy.deepcopy(cfg);cfg['settings_expected']['tracking']['enabled']=True;cfg['settings_expected']['intrusion']['enabled']=True cfg['failure_stop_all']=True for a in cfg['artifacts']: if a['name']=='registry':a['sha256']=file_hash(cfg['registration_file']) NEW.write_text(yaml.safe_dump(cfg,sort_keys=False));NEW.chmod(0o600) save('prepared.json',dict(config=str(NEW),settings={r['stream_id']:rt.settings(r['stream_id']) for r in plan['rows']})) print('All16 settings applied with CAS; no Start yet',flush=True) class Trial(Benchmark): def metrics(self,rows,enabled): states=self.runtime.states();out={} for row in rows: sid=row['stream_id'];q=states[sid] if q['state']!='running':continue try: out[sid]=self.runtime.api.call('POST','/api/v1/streams/'+quote(sid,safe='')+'/metrics', dict(run_generation=q['run_generation'],enabled=enabled,seconds=60)) except Exception as e:out[sid]=dict(error=type(e).__name__) self.runtime.save(f'metrics-{len(rows)}-'+('on' if enabled else 'off')+'.json',out) def metadata(self,rows): samples={};summaries={};states=self.runtime.states() for row in rows: sid=row['stream_id'];key='contextra:metadata:v1:camera:'+hashlib.sha256(row['source_id'].encode()).hexdigest()+':frames' entries=self.runtime.redis('XREVRANGE',key,'+','-','COUNT','2') values=[] for entry in entries: fields=dict(zip(entry[1][::2],entry[1][1::2]));v=json.loads(fields['json']) assert v['sourceId']==row['source_id'] and v['runId']==states[sid]['run_id'] and v['streamId']==sid assert v['coordinateSpace']=='SOURCE_FRAME' and v['normalized'] is True values.append(v) samples[sid]=values sets=[{str(o['trackId']) for o in v['objects'] if o.get('trackId') is not None} for v in values] objects=[o for v in values for o in v['objects']] summaries[sid]=dict(frames=len(values),frame_seq=[v['frameSeq'] for v in values], objects=[len(v['objects']) for v in values],tracked=[len(x) for x in sets], adjacent_track_ids=len(set.intersection(*sets)) if sets else 0, event_ids=len({e for o in objects for e in o.get('eventIds',[])}), red_objects=sum(o.get('color')=='#FF3030' for o in objects), events_created=states[sid].get('metadata_path',{}).get('events_created'), frame_time=[v['frameTime'] for v in values], detail_capabilities=states[sid].get('detail_capabilities')) if len(values)==2:assert values[0]['frameSeq']>values[1]['frameSeq'] self.runtime.save(f'metadata-{len(rows)}-private.json',samples) self.runtime.save(f'metadata-{len(rows)}-summary.json',summaries) def observe(self,rows,seconds): self.metrics(rows,True) try: out=super().observe(rows,seconds) self.metadata(rows) window=json.loads((self.runtime.evidence/f'window-{len(rows)}.json').read_text()) out['bounded_durations_last']={sid:v['durations'] for sid,v in window['samples'][-1]['channels'].items()} out['duration_scope']='Per-channel most recent up to256 samples within at-most60s diagnostics; not full-window mean.' return out finally:self.metrics(rows,False) def trial(): cfg=load_config(NEW);evidence=R/'ascent';evidence.mkdir(mode=0o700);rt=Runtime(cfg,evidence) with Path(cfg['lock_file']).open('a+') as f: fcntl.flock(f,fcntl.LOCK_EX|fcntl.LOCK_NB);plan=preflight(rt) rt.save('plan.json',plan);rt.save('conditions.json',dict(config=cfg,head=read('plan.json')['head'], tools={p.name:file_hash(p) for p in (REPO/'tools').glob('stream_benchmark*.py')})) save('ascent-start.json',dict(at=time.time(),count=1)) if Trial(rt).run(plan):raise RuntimeError('Ascent failed; all targets must remain stopped') def finish(error): cfg=load_config(NEW if NEW.exists() else OLD);rt=Runtime(cfg,R/'ascent' if (R/'ascent').exists() else R) plan=read('plan.json');cleanup={} if error: for row in reversed(plan['rows']): sid=row['stream_id'];q=rt.states()[sid] if not q.get('child_pid') and q['state'] in ('error','stopped'):cleanup[sid]=q;continue try:cleanup[sid]=rt.stop(sid) except Exception as e:cleanup[sid]=dict(error=type(e).__name__) states=rt.states();settings={r['stream_id']:rt.settings(r['stream_id']) for r in plan['rows']} save('final.json',dict(error=error,states=states,settings=settings,cleanup=cleanup, image=rt.inspect()['Image'],container_unchanged=rt.inspect()['Id']==plan['container_id'], model=rt.triton('v2/models/RFDETRMedium/config'),resource=rt.resources(), active_preserved=file_hash(ACTIVE)==plan['active_hash'],env_preserved=file_hash(ENV)==plan['env_hash'], presets_preserved=all(file_hash(p)==v['sha256'] for p,v in plan['presets'].items()))) print('Final failed=',bool(error),'active=',sum(bool(q.get('child_pid')) for q in states.values()),flush=True) if __name__=='__main__': error=None try:gate();prepare();trial() except Exception as e: error=dict(type=type(e).__name__,message=str(e),trace=traceback.format_exc());save('failure.json',error) finally: if (R/'plan.json').exists():finish(error)