"""Bounded isolated Pose preparation; no engine execution or production mutation.""" import concurrent.futures import csv import datetime import hashlib import json import pathlib import shutil import subprocess import sys import time ROOT = pathlib.Path(__file__).resolve().parent.parent BASE = ROOT.parent BASELINE = BASE / '2026-10-28_contextra-pose-input-review-15clips_r01' SOURCE = BASE / '2026-10-29_contextra-self-filming-original01-clips_r01' PACKAGE = SOURCE / '아이서티_컨텍스트라_싸움_쓰러짐_자체촬영_원본01_클립자료' REMOTE = '/storage/contextra/experiments/action-model-eval/2026/' + ROOT.name ENGINE = '/storage/contextra/models/triton/RFDETRKeypointPreview/8' PREFIX = 'contextra-selfpose-1029-r01-' PROD = 'contextra-ai2-analytics-1' sys.path.insert(0, str(BASE / '2026-10-26_contextra-camera24-prep_r01/helpers')) from remote import run, SSH def digest(path): h = hashlib.sha256() with open(path, 'rb') as f: for block in iter(lambda: f.read(1024 * 1024), b''): h.update(block) return h.hexdigest() def save(path, value): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2)) def checked(args, **kwargs): p = run(args, **kwargs) if p.returncode: raise RuntimeError('Remote command failed: ' + p.stderr.decode()[-1000:]) return p.stdout def scp(src, dst): cmd = ['scp', '-q', '-r', '-i', SSH[2], '-o', SSH[4], '-o', 'StrictHostKeyChecking=yes', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', '-o', 'ServerAliveInterval=15', '-o', 'ServerAliveCountMax=3', '-P', '1022', str(src), str(dst)] for attempt in range(2): p = subprocess.run(cmd, capture_output=True, timeout=1800) if p.returncode == 0:return if attempt == 0:time.sleep(3) raise RuntimeError('SCP failed after bounded retry: ' + p.stderr.decode()[-1000:]) def resources(gpu=False, starting=False): code = "import json,os,pathlib,shutil;m={l.split()[0].rstrip(':'):int(l.split()[1])*1024 for l in pathlib.Path('/proc/meminfo').read_text().splitlines() if len(l.split())>2 and l.split()[2]=='kB'};print(json.dumps(dict(root_free=shutil.disk_usage('/').free,storage_free=shutil.disk_usage('/storage').free,ram_available=m['MemAvailable'],cpus=os.cpu_count(),load=os.getloadavg())))" r = json.loads(checked(['python3', '-c', code], timeout=30)) assert r['root_free'] >= 15 * 1024**3 and r['storage_free'] >= 30 * 1024**3 assert r['ram_available'] >= 12 * 1024**3 and r['load'][0] < r['cpus'] * .9 if gpu: r['gpu_free_mib'] = int(checked(['nvidia-smi', '--query-gpu=memory.free', '--format=csv,noheader,nounits']).decode().strip()) assert r['gpu_free_mib'] >= (8192 if starting else 4096) return r def exclusive_gpu(): production_pids = set(checked(['docker', 'top', PROD, '-eo', 'pid']).decode().split()[1:]) gpu_pids = set(checked(['nvidia-smi', '--query-compute-apps=pid', '--format=csv,noheader,nounits']).decode().split()) assert gpu_pids <= production_pids, 'Other GPU owner present; defer side trial' containers = checked(['docker', 'ps', '--format', '{{.Names}}']).decode().split() assert all(n.startswith('contextra-ai2-') for n in containers), 'Other experiment container present' return dict(gpu_pids=sorted(gpu_pids), containers=containers) def snapshot(label): names = checked(['docker', 'ps', '--format', '{{.Names}} {{.ID}} {{.Image}}']).decode().splitlines() config = checked(['docker', 'exec', PROD, 'sha256sum', '/var/lib/contextra/runtime/streams_runtime.yaml']).decode().split()[0] value = dict(timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), containers=names, config_sha256=config, resources=resources(gpu=True), process_pids=checked(['docker','top',PROD,'-eo','pid']).decode().split()[1:]) save(ROOT / 'reports' / ('production_' + label + '.json'), value) return value def checkpoint(stage, completed): save(ROOT / 'checkpoint.json', dict(stage=stage, completed=completed, goal='C01-C22 Pose/RGB/Tracking only', excluded=['L01', 'Event', 'engine evaluation', 'new review videos'], expected_observations=7060, source_frame='clip-local 1-based', next='resume first incomplete stage; never overwrite completed cache')) def setup(): assert shutil.disk_usage('/').free >= 15 * 1024**3 and shutil.disk_usage('/data').free >= 30 * 1024**3 ROOT.joinpath('reports').mkdir(exist_ok=True) assert not (ROOT / 'specs.json').exists() ownership = exclusive_gpu() save(ROOT / 'reports/gpu_ownership_before.json', ownership) resources(gpu=True, starting=True) checks = json.loads((PACKAGE / 'SHA256SUMS.json').read_text())['files'] assert all(digest(PACKAGE / p) == v['sha256'] for p, v in checks.items()) manifest = json.loads((PACKAGE / 'metadata/manifest.json').read_text()) expected = [122,104,95,199,221,159,99,406,407,417,323,406,547,487,264,399,537,350,381,350,285,502] specs = {} for c in manifest['clips']: if c['id'] == 'L01': continue key = c['id'] specs[key] = dict(name=c['name'], local_source=str(PACKAGE / c['output']), remote_source=REMOTE + '/sources/' + key + '.mp4', sha256=checks[c['output']]['sha256'], display_bounds=None, known_loop_boundaries=[], cache=REMOTE + '/clips/' + key, reused=False, expected_samples=expected[int(key[1:])-1], source_clip=c) assert len(specs) == 22 and sum(s['expected_samples'] for s in specs.values()) == 7060 save(ROOT / 'specs.json', specs) for name in ('pose.py', 'plan.py', 'sampling.py', 'verify_cache.py'): target = ROOT / 'helpers' / name assert not target.exists() shutil.copy2(BASELINE / 'helpers' / name, target) shutil.copy2(BASELINE / 'model_reference.json', ROOT / 'model_reference.json') ref = json.loads((ROOT / 'model_reference.json').read_text()) image = ref['image'] assert checked(['docker','image','inspect',image,'--format','{{.Id}}']).decode().strip() == image assert checked(['sha256sum', ENGINE + '/model.plan'], sudo=True).decode().split()[0] == ref['pose']['engine_sha256'] for name, sha in ref['pose']['source_files'].items(): assert checked(['docker','exec',PROD,'sha256sum','/opt/contextra/app/'+name]).decode().split()[0] == sha checkpoint('PREPARING_C01_GATE', []) checked(['test', '!', '-e', REMOTE]) checked(['mkdir', REMOTE], sudo=True) checked(['chown', 'si:si', REMOTE], sudo=True) checked(['chmod', '700', REMOTE]) dirs = [REMOTE + '/sources'] for key in specs: dirs.extend(REMOTE + '/clips/' + key + '/' + d for d in ('inputs','reports','pose_models/RFDETRKeypointPreview/8')) checked(['mkdir','-p']+dirs) scp(ROOT / 'helpers', 'si@58.233.207.31:' + REMOTE + '/') for name in ('specs.json','model_reference.json','checkpoint.json'): scp(ROOT / name, 'si@58.233.207.31:' + REMOTE + '/') for key,s in specs.items(): scp(s['local_source'], 'si@58.233.207.31:' + s['remote_source']) snapshot('before') print('SETUP_PASS 22 inputs; no GPU inference yet', flush=True) def trial(stage, key): specs = json.loads((ROOT / 'specs.json').read_text()); s=specs[key] image = json.loads((ROOT/'model_reference.json').read_text())['image'] gpu = stage == 'pose' resources(gpu=gpu, starting=gpu) if gpu: exclusive_gpu() name = PREFIX + stage.replace('_','-') + '-' + key.lower() cmd=['docker','run','--rm','--pull','never','--name',name,'--user','0:0','--read-only','--network','none', '--cap-drop','ALL','--cap-add','DAC_OVERRIDE','--security-opt','no-new-privileges','--cpus','2','--memory','8g', '--pids-limit','256','--shm-size','256m','--tmpfs','/tmp:rw,size=512m','-e','PYTHONDONTWRITEBYTECODE=1', '-e','OMP_NUM_THREADS=2','-e','OPENBLAS_NUM_THREADS=2','-e','CLIP='+key, '-v',s['cache']+':/eval','-v',s['cache']+':/cache:ro','-v',REMOTE+'/helpers:/helpers:ro', '-v',REMOTE+'/specs.json:/specs.json:ro','-v',REMOTE+'/model_reference.json:/model_reference.json:ro', '-v',s['remote_source']+':/source.mp4:ro'] if gpu: cmd += ['--gpus','device=0','-v',ENGINE+':/eval/pose_models/RFDETRKeypointPreview/8:ro'] else: cmd += ['--runtime','runc','-e','NVIDIA_VISIBLE_DEVICES=void'] cmd += ['--entrypoint','python3',image,'/helpers/'+stage+'.py'] print('START',key,stage,flush=True); samples=[] with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: future=pool.submit(run,cmd,timeout=1800) try: while True: try: p=future.result(timeout=30);break except concurrent.futures.TimeoutError: if future.done(): raise samples.append(resources(gpu=gpu)) print('MONITOR',key,stage,samples[-1].get('gpu_free_mib'),flush=True) except BaseException: run(['docker','stop','--time','15',name],timeout=40) raise (ROOT/'reports'/f'{key}_{stage}.log').write_bytes(p.stdout+p.stderr) save(ROOT/'reports'/f'{key}_{stage}_resources.json',samples) if p.returncode: raise RuntimeError(key+'/'+stage+': '+p.stderr.decode()[-1800:]) print('PASS',key,stage,flush=True) def resume_transfer(): scp(ROOT / 'helpers', 'si@58.233.207.31:' + REMOTE + '/') for name in ('specs.json','model_reference.json','checkpoint.json'): scp(ROOT / name, 'si@58.233.207.31:' + REMOTE + '/') for key,s in json.loads((ROOT/'specs.json').read_text()).items(): probe=run(['sha256sum',s['remote_source']]) if probe.returncode==0 and probe.stdout.decode().split()[0]==s['sha256']:continue scp(s['local_source'],'si@58.233.207.31:'+s['remote_source']) if not (ROOT/'reports/production_before.json').exists():snapshot('before') print('TRANSFER_PASS',flush=True) def infer(keys): for key in keys: s=json.loads((ROOT/'specs.json').read_text())[key] for stage,marker in [('plan','inputs/sampling_plan.json'),('pose','inputs/pose.json'),('verify_cache','reports/cache_validation.json'),('verify_extra','reports/compatibility.json')]: exists=run(['test','-f',s['cache']+'/'+marker]).returncode==0 if exists: print('REUSE',key,stage,flush=True) continue trial(stage,key) local=ROOT/'clips'/key;local.mkdir(parents=True,exist_ok=True) # Reports and compact cache are local; RGB remains intact at the remote cache root. for folder in ('reports','inputs'): (local/folder).mkdir(exist_ok=True) for file in ('pose.json','sampling.csv','source_map.csv','sampling_plan.json','pose_candidates_prefilter.jsonl'): scp('si@58.233.207.31:'+s['cache']+'/inputs/'+file,local/'inputs'/file) scp('si@58.233.207.31:'+s['cache']+'/reports/.',local/'reports') remote_hashes=json.loads((local/'reports/cache_hashes.json').read_text()) for rel,h in remote_hashes['files'].items(): if rel.startswith('inputs/frames/'):continue assert digest(local/rel)==h,rel assert json.loads((local/'reports/compatibility.json').read_text())['samples']==s['expected_samples'] print('TRANSFER_VERIFIED',key,flush=True) if keys==['C01']:checkpoint('C01_COMPATIBILITY_PASS',['C01']) else: checkpoint('ALL_CACHE_PASS',list(json.loads((ROOT/'specs.json').read_text()))) snapshot('after') if __name__=='__main__': if sys.argv[1]=='setup':setup() elif sys.argv[1]=='transfer':resume_transfer() elif sys.argv[1]=='first':infer(['C01']) elif sys.argv[1]=='rest':infer([f'C{i:02d}' for i in range(2,23)]) else:raise ValueError('Unknown stage')