"""Render frozen results; no feature extraction or decision invocation here.""" import bootstrap from bootstrap import SOURCE import csv from fractions import Fraction import hashlib import json import os from pathlib import Path import shutil import time import av import numpy as np from PIL import Image, ImageDraw, ImageOps from inputs import OUT, load, evidence from contract import observation, palette, edge_state, EDGES from layout import W,H,IW,IH,IX,IY,FONT,text,overlay,MUTED from render_candidates import hull COLORS={'RED_CURRENT':(255,65,65),'YELLOW_HOLD':(255,205,35),'BLUE_CANDIDATE':(40,145,255)} TB=Fraction(1,90000) def sha(path): h=hashlib.sha256() with path.open('rb') as f: for block in iter(lambda:f.read(1024*1024),b''):h.update(block) return h.hexdigest() def jsonl(path): with path.open() as f: for line in f:yield json.loads(line) def part_entries(clip, legacy): rows,chosen,spec,cache,maps=load(clip) rows={r['sample']:r for r in rows if r['sample'] in chosen} root=SOURCE if legacy else OUT trace=root/'download'/clip/'data/B_Integrated_diagnostics.jsonl' if legacy else OUT/'download/SELF_L01/data'/clip/'B_Integrated_diagnostics.jsonl' summary=json.loads((trace.parent/'summary.json').read_text()) assert sha(trace)==summary['sha256'] and summary['candidate_sha256']==sha(OUT/'candidates.json') views={r['sample']:{k:r[k] for k in ('sample','source_frame','seconds','bodies','pairs','review_frame')} for r in jsonl(root/'work'/clip/'enriched.jsonl') if r['displayed']} # Discard large per-pair feature histories after reading; geometry alone is needed. for v in views.values():v['pairs']=[dict(orders=p['orders']) for p in v['pairs']] results=[r for r in jsonl(trace) if r['displayed']] hashes=json.loads(evidence(spec['rgb_hashes_remote']).read_text()) colors=palette(list(rows.values())) entries=[] for r in results: row=rows[r['sample']];view=views[r['sample']] assert (row['source_frame'],row['seconds'])==(r['source_frame'],r['seconds'])==(view['source_frame'],view['seconds']) entries.append(dict(kind='observation',row=row,view=view,result=r,colors=colors, part_id=spec.get('part_id',''),path=cache/'frames'/f"{r['sample']:04d}.png", rgb_sha256=hashes[f"{r['sample']:04d}.png"],seconds=row['seconds'])) assert len(entries)==len(chosen) return entries,trace,summary def entries_for(clip): if clip!='SELF_L01': entries,trace,summary=part_entries(clip,True) dest=OUT/'download'/clip/'data';dest.mkdir(parents=True,exist_ok=True) for name in ('B_Integrated_diagnostics.jsonl','summary.json'): target=dest/name if not target.exists():shutil.copy2(trace.parent/name,target) assert sha(target)==sha(trace.parent/name) return entries names=json.loads((OUT/'plan.json').read_text())['parts'] first,_,_=part_entries(names[0],False);second,_,_=part_entries(names[1],False) assert (len(first),len(second))==(1044,7261) gap=second[0]['seconds']-first[-1]['seconds'] assert abs(gap-1.057729)<1e-6 # One neutral video slate occupies the missing interval, never an engine input. slate=dict(kind='gap',seconds=first[-1]['seconds']+.1,part_id='P01/P02', observation_gap=gap,next_seconds=second[0]['seconds']) assert first[-1]['seconds']2:d.polygon(poly,fill=color+(68,),outline=color+(255,),width=max(3,round(image.width/450))) image=Image.alpha_composite(image.convert('RGBA'),layer).convert('RGB') people=[p for p in observation(row,e['colors']) if p['visible']] for p in people: p['joint_states']=['BELOW' if s=='LOW' else s for s in p['joint_states']] p['edge_states']=[edge_state(p['joint_states'][a],p['joint_states'][b]) for a,b in EDGES] p['identity_color']=(176,185,194) if e['part_id']:p['identity']=e['part_id'].replace('L01_','')+'/'+p['identity'] # Existing collision-aware placement, neutral input marks, no rule highlighting. panel,meta=overlay(image,people) return panel,meta def page(e,clip,index,total): out=Image.new('RGB',(W,H),'#101820');d=ImageDraw.Draw(out);rects=[] text(d,rects,(20,16),clip+' | 싸움 통합 후보 · 공간 검토',34) if e['kind']=='gap': text(d,rects,(20,80),f"L01 PTS {e['seconds']:.6f} ~ {e['next_seconds']:.6f}s",30) text(d,rects,(280,550),'자료 공백 · 다음 파트에서 상태 초기화',40) text(d,rects,(280,650),f"선택 관측 사이 간격 {e['observation_gap']:.6f}초",32) text(d,rects,(280,750),'영상 안내 화면이며 관측·행동·Event 판정이 아닙니다.',30) text(d,rects,(280,850),'이전 공간색·bbox·skeleton·판정 이력을 이어 붙이지 않습니다.',30) return out,dict(kind='gap',state='GAP',group_states=[],label_count=0) row=e['row'];r=e['result'];ep=r['episode'];panel,meta=panel_for(e) out.paste(panel,(IX,IY)) text(d,rects,(20,78),f"입력 Frame {row['source_frame']} | PTS {row['seconds']:.3f}s | 관측 {index+1}/{total}"+(' | '+e['part_id'] if e['part_id'] else ''),28) color=COLORS['RED_CURRENT'] if r['hit'] else COLORS['YELLOW_HOLD'] if ep['display'] else COLORS['BLUE_CANDIDATE'] if ep['display']:d.rectangle((IX,IY,IX+IW-1,IY+IH-1),outline=color,width=8) state='충족' if r['hit'] else '미충족' if r['observable'] else '자료 부족 / 준비 중' x=1950 text(d,rects,(x,180),'현재 최종 조건',34) text(d,rects,(x,245),state,40,color if r['hit'] else MUTED) text(d,rects,(x,365),'표시 유지',34) hold='현재 충족 표시' if r['hit'] else '노랑 유지 중 · 현재 충족 아님' if ep['display'] else '없음' text(d,rects,(x,430),hold,32,color if ep['display'] else MUTED) text(d,rects,(x,550),'새 시험 Event',34) text(d,rects,(x,615),'발생' if ep['event'] else '이번 관측에는 없음',32) text(d,rects,(x,750),'공간 후보',34) text(d,rects,(x,815),f"{len(r['groups'])}개 그룹",32) uncertain=ep['display'] and not any(g['state'] in ('RED_CURRENT','YELLOW_HOLD') for g in r['groups']) if uncertain: text(d,rects,(x,915),'기여 공간 연결 불확실',30) text(d,rects,(x,965),'ROI 상태만 표시 · 옛 위치 사용 안 함',26,MUTED) text(d,rects,(20,1240),'청색: 공간 후보 빨강: 현재 최종 조건 충족 노랑: 직전 충족 뒤 1초 유지',30) text(d,rects,(20,1310),'빨강은 현재 조건입니다. 노랑은 현재 조건 충족을 뜻하지 않습니다.',28,MUTED) text(d,rects,(20,1380),'새 시험 Event는 별도 집계합니다. 쿨다운 중에도 빨간 판정은 표시합니다.',28,MUTED) text(d,rects,(20,1450),'bbox·skeleton은 입력 참고 표시입니다. 특정 사람의 싸움 확정 표기가 아닙니다.',28,MUTED) text(d,rects,(20,1520),'bbox >= 0.25 · 관절 >= 0.10 | T: 기존 추적 ID · q: 해당 관측의 미추적 칸',28,MUTED) text(d,rects,(20,1590),'기존 10FPS 관측 · 정상 속도 · 운영 미적용 | 상세 근거는 data 폴더',28,MUTED) if e['part_id']: text(d,rects,(20,1660),'L01: 파트별 이력 초기화 · T는 파트와 함께 해석 · 실제 PTS 사용',28,MUTED) elif clip.startswith(('05_','10_')): text(d,rects,(20,1660),'입력 한계 참고 사례 · 감지 성공을 필수로 요구하지 않음',28,MUTED) elif clip=='SELF_C22': text(d,rects,(20,1660),'사용자 감지 기대 판단 대기 · 자동 정답 지정 없음',28,MUTED) meta.update(kind='observation',state='RED' if r['hit'] else 'YELLOW' if ep['display'] else 'BLUE', group_states=[dict(id=g['id'],state=g['state']) for g in r['groups']], hit=r['hit'],display=ep['display'],event=ep['event'],observable=r['observable']) return out,meta def render(clip,entries,output,report,start_index=0,total=None): started=time.monotonic();output.parent.mkdir(parents=True,exist_ok=True) if report.exists(): assert sha(output)==json.loads(report.read_text())['sha256'];print('REUSED',output);return assert not output.exists() draft=output.with_name('.'+output.stem+'.partial.mp4');assert not draft.exists() origin=entries[0]['seconds'];total=total or len(entries) con=av.open(str(draft),'w',options={'movflags':'faststart'});stream=con.add_stream('libx264',rate=10) stream.width=W;stream.height=H;stream.pix_fmt='yuv420p';stream.options={'crf':'22','preset':'veryfast'} stream.codec_context.thread_count=1;stream.time_base=stream.codec_context.time_base=TB mapping=[];states=[];saved=set();last=None for i,e in enumerate(entries): out,meta=page(e,clip,start_index+i,total) state=meta['state'];sample=state not in saved or i in (0,len(entries)-1) or (last and last['kind']!=e['kind']) if sample: out.save(report.parent/(output.stem+f'_sample_{i:05d}.jpg'),quality=90);saved.add(state) pts=round((e['seconds']-origin)*90000) frame=av.VideoFrame.from_ndarray(np.asarray(out),format='rgb24');frame.pts=pts;frame.time_base=TB for packet in stream.encode(frame):con.mux(packet) mapping.append(dict(output_frame=i+1,kind=e['kind'],part_id=e['part_id'], sample=e.get('row',{}).get('sample'),source_frame=e.get('row',{}).get('source_frame'), source_seconds=e['seconds'],output_seconds=pts/90000, review_frame=e.get('result',{}).get('review_frame'))) states.append(dict(output_frame=i+1,**meta));last=e if (i+1)%200==0:print('FRAME',clip,start_index+i+1,total,flush=True) for packet in stream.encode():con.mux(packet) con.close();times=[] with av.open(str(draft)) as inp: inp.streams.video[0].codec_context.thread_count=1 for f in inp.decode(video=0): assert not f.is_corrupt and (f.width,f.height)==(W,H);times.append(float(f.pts*f.time_base)) assert len(times)==len(entries) and all(b>a for a,b in zip(times,times[1:])) error=max(abs(t-m['output_seconds']) for t,m in zip(times,mapping));assert error<1/90000 draft.rename(output) with (report.parent/(output.stem+'_frame_map.csv')).open('w',newline='') as f: w=csv.DictWriter(f,fieldnames=mapping[0]);w.writeheader();w.writerows(mapping) (report.parent/(output.stem+'_display.json')).write_text(json.dumps(states,separators=(',',':'))) value=dict(status='PASS',frames=len(entries),observations=sum(e['kind']=='observation' for e in entries), gap_slates=sum(e['kind']=='gap' for e in entries),origin_seconds=origin,last_source_seconds=entries[-1]['seconds'], sha256=sha(output),resolution=[W,H],nominal_fps=10,max_pts_error=error, text_overflow=0,text_overlap=0,rgb_hashes_verified=sum(e['kind']=='observation' for e in entries), elapsed_seconds=time.monotonic()-started) report.write_text(json.dumps(value,indent=2));print('VIDEO_PASS',clip,json.dumps(value),flush=True) if __name__=='__main__': clip=os.environ['CLIP'];assert clip in json.loads((OUT/'plan.json').read_text())['remaining'] dest=OUT/'download'/clip render(clip,entries_for(clip),dest/'B_Integrated_Space_Review.mp4',dest/'data/video_validation.json')