"""Bounded offline preparation. Only this run's containers are disposable.""" import concurrent.futures import datetime import hashlib import json import pathlib import subprocess import sys import time LOCAL=pathlib.Path(__file__).resolve().parent.parent BASE=LOCAL.parent RB='/storage/contextra/experiments/action-model-eval/2026' REMOTE=RB+'/'+LOCAL.name PREP=BASE/'2026-10-26_contextra-camera24-prep_r01' SHORT=BASE/'2026-10-27_contextra-short-input-7clips_r01' NEG=BASE/'2026-10-27_contextra-posec3d-nonfight-4clips_r01' KISA=pathlib.Path('/data/playlist/uploads/260914_아이서티_컨텍스트라_테스트용_kisa_영상들_클립들/(검증용)kisa_data') PREFIX='contextra-pose-review-1028-r01-' IMAGE='sha256:671c7416f503481d2be2092a5aa0c035938a07fc514cfc0b45f94d78e86370bd' ENGINE='/storage/contextra/models/triton/RFDETRKeypointPreview/8' REPORT=LOCAL/'reports' sys.path.insert(0,str(PREP/'helpers')) from remote import run,SSH def checked(args,**kwargs): p=run(args,**kwargs) if p.returncode:raise RuntimeError(p.stderr.decode()[-1500:]+p.stdout.decode()[-1500:]) 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',src,dst] for attempt in range(2): p=subprocess.run(cmd,capture_output=True,timeout=1200) if p.returncode==0:return if attempt==0:time.sleep(2) raise RuntimeError('Bounded SCP failed: '+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())))" s=json.loads(checked(['python3','-c',code],timeout=40)) assert s['root_free']>=15*1024**3 and s['storage_free']>=30*1024**3 and s['ram_available']>=12*1024**3,s assert s['load'][0]=(8192 if starting else 4096),s return s def snapshot(label): raw=checked(['docker','exec','-i','contextra-ai2-analytics-1','python3','-'],data=(PREP/'helpers/snapshot.py').read_bytes(),timeout=90) state=json.loads(raw);assert state['health']['ready'] (REPORT/f'production_{label}.json').write_bytes(raw) (REPORT/f'containers_{label}.txt').write_bytes(checked(['docker','ps','--format','{{.Names}} {{.Image}} {{.ID}}'])) return state def specs():return json.loads((LOCAL/'specs.json').read_text()) def inventory(): short={v['source_file']:v for v in json.loads((SHORT/'reports/source_manifest.json').read_text())} neg=json.loads((NEG/'specs.json').read_text()) result={} camera=[('01_camera24_fight',PREP/'inputs/camera24_original.mp4',[242,723],RB+'/2026-10-26_contextra-camera24-eval_r01',short['camera24_original.mp4']['sha256'],'좌측 상단2명은 지정 반복 전체에서 싸움이라는 사용자 관찰. 필수 수용 기준은 미정.'), ('02_camera22_crowd',BASE/'2026-10-27_contextra-four-scenarios-prep_r01/camera22/inputs/camera22_original.mp4',[242,723],RB+'/'+NEG.name+'/clips/01_camera22_crowd',neg['01_camera22_crowd']['sha256'],'사용자가 싸움이 없는 비교자료로 설명. 군중 인원은 정확한 정답 수치가 아님.'), ('03_camera27_fall',BASE/'2026-10-27_contextra-camera27-prep_r01/inputs/camera27_original.mp4',[362,1083],RB+'/'+NEG.name+'/clips/02_camera27_fall',neg['02_camera27_fall']['sha256'],'사용자가 싸움이 없는 비교자료로 설명. 쓰러짐 경계는 대략적이며 판정 조건은 미정.'), ('04_camera29_fire',BASE/'2026-10-27_contextra-four-scenarios-prep_r01/camera29/inputs/camera29_original.mp4',[242,482],RB+'/'+NEG.name+'/clips/03_camera29_fire',neg['03_camera29_fire']['sha256'],'사용자가 싸움이 없는 비교자료로 설명. 화재 Event 정답 시점은 미정.')] for key,p,bounds,cache,digest,note in camera: assert hashlib.sha256(p.read_bytes()).hexdigest()==digest result[key]=dict(local_source=str(p),source_file=p.name,remote_source=RB+'/'+str(p.relative_to(BASE)),sha256=digest, display_bounds=bounds,known_loop_boundaries=[],cache=cache,reused=True,user_observation=note,acceptance='USER_UNSET') fights=sorted((KISA/'싸움/배포/clips').glob('*.mp4'));falls=sorted((KISA/'쓰러짐/배포/clips').glob('*.mp4')) assert len(fights)==6 and len(falls)==5 for i,p in enumerate(fights+falls,5): kind='fight' if p in fights else 'fall';key=f'{i:02d}_KISA_{kind}_'+p.stem.split('_s')[0] digest=hashlib.sha256(p.read_bytes()).hexdigest();cache=None;remote=None if kind=='fight': old=short[p.name];assert digest==old['sha256'];cache=RB+'/'+SHORT.name+'/clips/'+old['clip'];remote=RB+'/'+SHORT.name+'/sources/'+old['clip']+'.mp4' elif p.name.startswith('C00_051_0002_'): assert digest==neg['04_KISA_C00_051_0002']['sha256'];cache=RB+'/'+NEG.name+'/clips/04_KISA_C00_051_0002';remote=RB+'/'+NEG.name+'/sources/'+p.name result[key]=dict(local_source=str(p),source_file=p.name,remote_source=remote or REMOTE+'/sources/'+p.name, sha256=digest,display_bounds=None,known_loop_boundaries=[],cache=cache or REMOTE+'/clips/'+key,reused=cache is not None, user_observation=('대부분 구간이 싸움이라는 사용자 설명. 정확한 구간은 미지정.' if kind=='fight' else '싸움이 없는 비교자료라는 사용자 설명.' if cache else '쓰러짐 폴더의 확보 자료. 영상별 기대/구간은 사용자 지정 대기.'),acceptance='USER_UNSET') assert len(result)==15 and len({v['sha256'] for v in result.values()})==15 (LOCAL/'specs.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)) return result def sync(): scp(str(LOCAL/'helpers'),'si@58.233.207.31:'+REMOTE+'/') scp(str(LOCAL/'specs.json'),'si@58.233.207.31:'+REMOTE+'/') def setup(): REPORT.mkdir(exist_ok=True) rows=inventory();(REPORT/'resources_before.json').write_text(json.dumps(resources(gpu=True,starting=True),indent=2)) for script in ('test_contract.py','test_sampling.py'): p=subprocess.run([sys.executable,'-B',str(LOCAL/'helpers'/script),'-v'],capture_output=True,timeout=30) (REPORT/(script+'.txt')).write_bytes(p.stdout+p.stderr);assert p.returncode==0,p.stderr.decode() assert checked(['docker','image','inspect',IMAGE,'--format','{{.Id}}']).decode().strip()==IMAGE assert checked(['docker','inspect','contextra-ai2-analytics-1','--format','{{.Image}}']).decode().strip()==IMAGE reference=json.loads(checked(['cat',RB+'/'+SHORT.name+'/clips/02_KISA_C00_005_0018/reports/pose_summary.json'])) assert checked(['sha256sum',ENGINE+'/model.plan'],sudo=True).decode().split()[0]==reference['engine_sha256'] for name,digest in reference['source_files'].items(): assert checked(['docker','exec','contextra-ai2-analytics-1','sha256sum','/opt/contextra/app/'+name]).decode().split()[0]==digest (LOCAL/'model_reference.json').write_text(json.dumps(dict(image=IMAGE,pose=reference),indent=2)) checked(['test','!','-e',REMOTE]);checked(['mkdir',REMOTE],sudo=True);checked(['chown','si:si',REMOTE],sudo=True);checked(['chmod','700',REMOTE]) dirs=['helpers','sources','download/validation'] for key,v in rows.items(): dirs += ['clips/'+key+'/'+d for d in ('inputs','reports','pose_models/RFDETRKeypointPreview/8')] checked(['mkdir','-p']+[REMOTE+'/'+d for d in dirs]);sync() for p in ('model_reference.json','checkpoint.ko.md'):scp(str(LOCAL/p),'si@58.233.207.31:'+REMOTE+'/') for key,v in rows.items(): if not v['reused']:scp(v['local_source'],'si@58.233.207.31:'+REMOTE+'/sources/') checks=json.loads(checked(['python3','-c',"import pathlib,json,hashlib; d=json.loads(pathlib.Path("+repr(REMOTE+'/specs.json')+").read_text());print(json.dumps({k:hashlib.sha256(pathlib.Path(v['remote_source']).read_bytes()).hexdigest() for k,v in d.items()}))"],timeout=120)) assert all(checks[k]==v['sha256'] for k,v in rows.items()) snapshot('before');print('SETUP_PASS reused=11 new=4',flush=True) def args(stage,clip): s=specs()[clip];gpu=stage=='pose';name=PREFIX+stage+'-'+clip.lower().replace('_','-') 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='+clip, '-v',REMOTE+'/clips/'+clip+':/eval','-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', '-v',s['cache']+':/cache:ro','-v',REMOTE+'/download:/download', '-v',RB+'/'+SHORT.name+'/fonts/NotoSansCJK-Regular.ttc:/font.ttc: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'] return name,cmd def trial(stage,clip): gpu=stage=='pose';resources(gpu=gpu,starting=gpu);name,cmd=args(stage,clip) print('START',stage,clip,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 s=resources(gpu=gpu);samples.append(s);print('RESOURCE',stage,clip,s.get('gpu_free_mib'),flush=True) except BaseException: run(['docker','stop','--time','15',name],timeout=40);raise (REPORT/f'{stage}_{clip}.log').write_bytes(p.stdout+p.stderr) (REPORT/f'{stage}_{clip}_resources.json').write_text(json.dumps(samples,indent=2)) if p.returncode:raise RuntimeError(stage+'/'+clip+': '+p.stderr.decode()[-2000:]+p.stdout.decode()[-1000:]) print('DONE',stage,clip,p.stdout.decode()[-800:],flush=True) def infer(): for key,v in specs().items(): if not v['reused']: for stage in ('plan','pose'):trial(stage,key) snapshot('after_inference') def many(stage): with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool: for f in concurrent.futures.as_completed([pool.submit(trial,stage,key) for key in specs()]):f.result() if __name__=='__main__': REPORT.mkdir(exist_ok=True) if sys.argv[1]=='setup':setup() elif sys.argv[1]=='sync':sync() elif sys.argv[1]=='infer':infer() elif sys.argv[1] in ('verify_cache','render'):many(sys.argv[1]) elif sys.argv[1]=='one':trial(sys.argv[2],sys.argv[3]) else:raise ValueError('Unknown bounded stage')