"""Run-scoped CPU jobs and read-only site evidence; never control live channels.""" import base64 import hashlib import json import os from pathlib import Path import shutil import sys os.umask(0o077) LOCAL = Path(__file__).resolve().parent.parent BASE = LOCAL.parent sys.path.insert(0, str(BASE / '2026-10-29_contextra-fight-b-refine_r01/helpers')) import control_refine as c c.LOCAL = LOCAL c.REMOTE = c.old.RB + '/' + LOCAL.name c.PREFIX = 'contextra-fall-1101-r01-' c.REPORT = LOCAL / 'reports' LEGACY = ['03_camera27_fall', '11_KISA_fall_C00_001_0018', '12_KISA_fall_C00_010_0004', '13_KISA_fall_C00_034_0005', '14_KISA_fall_C00_051_0002', '15_KISA_fall_C00_096_0008'] SITE_CODE = r''' import base64, hashlib, json, os, urllib.request from pathlib import Path import yaml names = ['app/fall_scenario.py','app/fall_config.py','app/fall_geometry.py', 'app/fall_runtime.py','app/byte_tracking.py','app/tracking_kernel.py', 'app/pose_nms.py','app/pose_config.py','app/contracts/frames.py', 'app/event_cooldown.py'] files = {n:base64.b64encode((Path('/opt/contextra')/n).read_bytes()).decode() for n in names} env = {k:v for k,v in os.environ.items() if k.startswith('CONTEXTRA_FALL_') or k == 'CONTEXTRA_EVENT_COOLDOWN_SECONDS'} configs = {} for k in ('STREAMS_CONFIG_PATH','STREAMS_RUNTIME_PATH','CONTEXTRA_CONTROL_TEMPLATE_FILE'): p = Path(os.environ[k]); raw = p.read_bytes() value = json.loads(raw) if p.suffix == '.json' else yaml.safe_load(raw) entries = [value] if k == 'CONTEXTRA_CONTROL_TEMPLATE_FILE' else value.get('added',value.get('streams',[])) configs[k] = dict(sha256=hashlib.sha256(raw).hexdigest(),channels=[dict(camera=x.get('camera_id',x.get('source_id')),fall=x.get('fall'),event_cooldowns=x.get('event_cooldowns'),tracking=x.get('tracking')) for x in entries]) runtime = {'camera27_present':False} token = Path(os.environ['CONTEXTRA_CONTROL_TOKEN_FILE']).read_text().strip() req = urllib.request.Request('http://127.0.0.1:5000/internal/secure-control/runtime-status',headers={'Authorization':'Bearer '+token}) with urllib.request.urlopen(req,timeout=5) as r: streams=json.load(r)['streams'] for stream in streams: if str(stream.get('cameraId')) != '27': continue with urllib.request.urlopen('http://127.0.0.1:5000/api/v1/streams/'+stream['streamId']+'/settings',timeout=5) as r: settings=json.load(r) runtime = dict(camera27_present=True,state=stream.get('state'),settings={k:{n:v.get(n) for n in ('fall','event_cooldowns','tracking')} if isinstance(v,dict) else v for k,v in settings.items() if k in ('configured','effective','restart_pending')}) print(json.dumps(dict(files=files,environment=env,configs=configs,runtime=runtime))) ''' def save(path, value): path.parent.mkdir(parents=True, exist_ok=True) assert not path.exists(), path path.write_text(json.dumps(value, ensure_ascii=False, indent=2)) def preflight(): c.REPORT.mkdir(exist_ok=True) if (LOCAL/'site_baseline.json').exists(): print('PREVIOUS_PREFLIGHT_PRESERVED'); return c.checked_resources() c.snapshot('before') assert c.old.checked(['docker','image','inspect',c.old.IMAGE,'--format','{{.Id}}']).decode().strip() == c.old.IMAGE names = c.old.checked(['docker','ps','--filter','label=com.docker.compose.service=analytics','--format','{{.Names}}']).decode().splitlines() assert len(names) == 1, 'Ambiguous analytics owner' result = json.loads(c.old.checked(['docker','exec','-i',names[0],'python3','-B','-'], data=SITE_CODE.encode(),timeout=35)) files = result.pop('files'); digests = {} for name, encoded in files.items(): raw = base64.b64decode(encoded) dest = LOCAL/'site_source'/name dest.parent.mkdir(parents=True,exist_ok=True) assert not dest.exists();dest.write_bytes(raw) digests[name] = hashlib.sha256(raw).hexdigest() result.update(source_sha256=digests,container=names[0],image=c.old.checked(['docker','inspect',names[0],'--format','{{.Image}}']).decode().strip(),runtime_control_calls=0) save(LOCAL/'site_baseline.json',result) print(json.dumps(dict(environment=result['environment'],runtime=result['runtime'],source_files=len(files)),ensure_ascii=False)) def plan(): prior=BASE/'2026-10-29_contextra-fight-b-refine_r01' existing=c.read(prior/'specs.json'); source=c.read(c.SOURCE/'specs.json') specs={k:dict(existing[k]) for k in LEGACY+['SELF_C%02d'%i for i in range(1,23)]} cat=c.read(c.SELF/'metadata/catalog.json'); by={x['clip_id']:x for x in cat['clips']} for name,s in specs.items(): s.pop('baseline',None) s['tracking_mode']='stored10' if name in (LEGACY[0],LEGACY[4]) else 'replay30' if name.startswith('SELF_'): item=by[s['clip_id']];s['core_review_range']=item['core_review_range'];s['case_ids']=item['case_ids'] else: s['display_bounds']=source[name]['display_bounds'];s['remote_source']=source[name]['remote_source'] save(LOCAL/'specs.json',specs) save(LOCAL/'self_maps.json',{k:v for k,v in c.read(prior/'self_maps.json').items() if k in specs}) save(LOCAL/'cases.json',c.read(c.SELF/'metadata/action_cases.json')) save(LOCAL/'references.json',dict( precision='Approximate user references, never exact GT or mandatory acceptance', self_fall_expectation='non_event_reference; core/context separated; deduplicate review Frame', legacy={ LEGACY[0]:dict(fall_motion=[476,512],down=[513,722],other_repeat='723~1083 repeated loop without separately supplied action anchors'), LEGACY[1]:dict(down=[250,2077]), LEGACY[2]:dict(down=[325,2179]), LEGACY[3]:dict(down=[697,None]), LEGACY[4]:dict(down=[139,1399]), LEGACY[5]:dict(down=[127,None])}, source='camera27 observation record and user feedback on the 15 Pose review videos; approximate boundaries')) save(LOCAL/'execution_contract.json',dict(candidate='FALL_SPATIAL_R1',inputs=28, displayed_observations=sum(len(s['selected_samples']) for s in specs.values()), person_score=.25,joint_score=.10,retracking_inputs=26,stored_tracking_inputs=2, cpus_per_job=2,memory_gib=8,max_parallel=2,timeout_seconds=1800, root_min_gib=15,storage_min_gib=30,ram_min_gib=12,image=c.old.IMAGE, gpu_jobs=0,production_control_calls=0,scope='offline candidate; no Runtime integration/deployment', baseline_head='83c08f15f3d8074407b9eca63a1f14d9d5746cbe')) print('PLAN',len(specs),sum(len(s['selected_samples']) for s in specs.values())) def setup(): c.checked_resources() c.old.checked(['test','!','-e',c.REMOTE]) c.old.checked(['mkdir',c.REMOTE],sudo=True) c.old.checked(['chown','si:si',c.REMOTE],sudo=True) c.old.checked(['chmod','700',c.REMOTE]) c.old.checked(['mkdir','-p',c.REMOTE+'/work',c.REMOTE+'/download/validation',c.REMOTE+'/reports']) sync() def sync(): for p in [LOCAL/'helpers',LOCAL/'site_source',*LOCAL.glob('*.json'),*LOCAL.glob('*.md')]: c.old.scp(str(p),'si@58.233.207.31:'+c.REMOTE+'/') if __name__=='__main__': stage=sys.argv[1] if stage=='preflight':preflight() elif stage=='plan':plan() elif stage=='setup':setup() elif stage=='sync':sync() elif stage=='one':c.run_job(sys.argv[2],sys.argv[3] if len(sys.argv)>3 else 'all') elif stage=='pull': # Trial containers write as root. Transfer ownership only inside this run. c.old.checked(['chown','-R','si:si',c.REMOTE+'/download',c.REMOTE+'/work',c.REMOTE+'/reports'],sudo=True) c.old.scp('si@58.233.207.31:'+c.REMOTE+'/download/.',str(LOCAL/'Contextra_Fall_Spatial_Review_20261101')) c.old.scp('si@58.233.207.31:'+c.REMOTE+'/reports/.',str(c.REPORT)) elif stage=='finish':c.finish() else:raise ValueError(stage)