"""Three synchronized views, one decode per cached RGB observation.""" import csv from fractions import Fraction import hashlib import json import os from pathlib import Path import time import av import numpy as np from PIL import Image, ImageDraw from review_data import verify_baseline, load_rows from review_draw import image_panel from layout import W,H,IX,IY,IW,IH,text,MUTED KINDS={'groups':('B_Space_Groups_Review.mp4','공간 그룹 / 표시용 가정'), 'motion':('B_Motion_Evidence_Review.mp4','동작 근거 / 기존 B 조건'), 'association':('B_Association_Difference_Review.mp4','연결 차이 / 정답 판정 아님')} ASSOC={'AGREEMENT':'두 연결 방식 일치','B_RESET_SAME_T_PAIR':'같은 T쌍 / B 연결 새 시작', 'T_SWAP_B_CONTINUES':'T 순서 교환 / B 공간 연결 유지','T_CHANGE_B_CONTINUES':'T 변경 / B 공간 연결 유지', 'UNCOMPARABLE_T_MISSING':'T 미확보 / 비교 불가','OBSERVATION_GAP':'관측 공백 / 차이 판정 제외', 'NEW_PAIR_OR_ELIGIBILITY_CHANGE':'새 쌍 또는 입력 조건 변화'} STATUS={'ALERT':'새 시험 이벤트','RULE_HIT_LATCHED':'현재 조건 충족 / 같은 episode', 'RULE_HIT_COOLDOWN':'현재 조건 충족 / cooldown','DISPLAY_HOLD':'0.4초 이내 표시 유지', 'UNKNOWN_LATCHED':'근거 부족 / episode 보존','QUIET_REARM_WAIT':'재무장 대기', 'MONITORING':'현재 규칙 미충족','INSUFFICIENT':'준비 또는 근거 부족'} def page(image,row,v,palette,kind,index,total): panel,meta=image_panel(image,row,v,palette,kind) out=Image.new('RGB',(W,H),'#101820');d=ImageDraw.Draw(out);rects=[] out.paste(panel,(IX,IY)) text(d,rects,(20,16),clip+' | '+KINDS[kind][1],34) text(d,rects,(20,78),f"원본 Frame {row['source_frame']} | PTS {row['seconds']:.3f}s | 관측 {index+1}/{total} | 10FPS 정상 속도",28) if kind=='motion' and v['episode']['display']:d.rectangle((IX,IY,IX+IW-1,IY+IH-1),outline='#ff6666',width=7) x=1910 text(d,rects,(x,150),'검토 정보 / bbox 0.25 · 관절 0.10',30) if kind=='groups': groups=v['groups'];entries=groups[:3] text(d,rects,(x,207),f"유효 몸통 {len(v['bodies'])}명 / 근접 쌍 {len(v['candidate_edges'])} / 그룹 {len(groups)}",26) for n,g in enumerate(entries): y=300+n*350;lim=(x,y,3180,y+330);d.rectangle(lim,outline='#43515e',width=2) bs={b['order']:b for b in v['bodies']};names=[bs[o]['ref'] for o in g['orders']] text(d,rects,(x+20,y+18),g['group']+f" | {len(g['orders'])}명 / {len(g['edges'])}근접 관계",30,limit=lim) for j in range(0,min(len(names),18),6):text(d,rects,(x+20,y+78+j//6*45),' / '.join(names[j:j+6]),24,limit=lim) if len(names)>18:text(d,rects,(x+20,y+225),f"추가 {len(names)-18}명은 화면·data에 유지",24,limit=lim) text(d,rects,(x+20,y+275),'근접 관계를 연결한 표시 그룹 / 싸움 판정 아님',24,MUTED,lim) if not groups:text(d,rects,(x+20,320),'2인 이상 근접 그룹 없음 / 정상 행동 확정 아님',26) legend=['반투명 다각형 = bbox 외곽의 표시용 그룹','가는 선 = 실제 근접 관계 / 빗금 = 도형 겹침', 'G번호·그룹색은 현재 Frame에서만 의미가 있습니다.', 'P-Q, Q-R이 연결되면 P-R이 멀어도 같은 그룹입니다.', '표시용 그룹은 기존 B 판정과 상태에 영향을 주지 않습니다.'] summary=f"그룹 {len(groups)}개 / 겹침 {meta['group_overlap_pixels']} 원본픽셀" omitted=max(0,len(groups)-3) else: differences=[p for p in v['pairs'] if p['association']['difference']] if kind=='motion': entries=sorted(v['pairs'],key=lambda p:(not bool(p['highlight']),-p['level'],p['token']))[:3] summary=STATUS[v['episode']['status']] text(d,rects,(x,207),f"현재 규칙 {','.join(v['current_rules']) or '없음'} / 새 시험Event {len(v['events'])}",26) legend=['얇은 작은 링 = 부분 조건 / 굵은 링 = 순간 조건', '가장 큰 링 = 반복·지속시간까지 누적 규칙 충족', '부분 조건에는 손의 근접만 충족한 경우도 포함됩니다.', '사람색 skeleton = 규칙 기여 / 쌍색 선·S번호 = 쌍 구분', '몸통 중심·기준선은 계산 근거 / 발차기 규칙은 없습니다.'] omitted=max(0,len(v['pairs'])-3) else: entries=differences[:3];summary=f"연결 차이 {len(differences)}쌍 / 어느 쪽 오류인지 확정하지 않음" unavailable=sum(p['association']['kind']=='UNCOMPARABLE_T_MISSING' for p in v['pairs']) text(d,rects,(x,207),f"차이 {len(differences)} / T 비교 불가 {unavailable} / 후보쌍 {len(v['pairs'])}",26) legend=['분홍 bbox·굵은 skeleton = 연결 방식이 다른 객체', '쌍색·S번호 = 쌍 / 이전→현재 선은 실제 연결 유지 때만 표시', 'T 변화 또는 B 연결 단절은 실제 오류 확정이 아닙니다.', 'T 없음·새 진입·근접 이탈·관측 공백은 별도로 구분합니다.', 'q는 Frame별 출력 칸 / S는 짧은 쌍 상태이며 실제 ID 아님'] omitted=max(0,len(differences)-3) for n,p in enumerate(entries): y=300+n*390;lim=(x,y,3180,y+370);d.rectangle(lim,outline='#43515e',width=2) text(d,rects,(x+20,y+15),p['token']+' | '+' + '.join(p['refs']),30,limit=lim) if kind=='motion': text(d,rects,(x+20,y+65),'규칙: '+(','.join(p['rules']) or '미충족')+' / '+p['reason'],24,limit=lim) text(d,rects,(x+20,y+110),f"강조 단계 {p['level']}/3 | pulse {p['pulses']}/2 | 근접 {p['close_seconds']:.2f}s",26,limit=lim) text(d,rects,(x+20,y+155),f"접촉 {p['contact_seconds']:.2f}/0.40s | 동적 {p['dynamic_seconds']:.2f}/0.20s ({p['dynamic_samples']}관측)",24,limit=lim) arms=sorted(p['evidence'],key=lambda e:(-e['level'],-e['values']['relative_motion'])) if arms: e=arms[0];a=e['values'];ref=p['refs'][e['actor']]+' '+e['side'] text(d,rects,(x+20,y+200),f"{ref}: 변화 {a['relative_motion']:.3f} / 타격0.18·접촉0.07",24,limit=lim) text(d,rects,(x+20,y+245),f"방향 {a['toward']:.3f}/0.10 접근 {a['approach']:.3f}/0.04 거리 {a['hand_distance']:.3f}",24,limit=lim) failed='/'.join(e['failed_strike']) or '없음' text(d,rects,(x+20,y+295),'타격 미달: '+failed,24,MUTED,lim) dynamic_failed='/'.join(e['failed_dynamic']) or '없음' text(d,rects,(x+20,y+335),f"몸통 변화 {p['features']['pair_relative_change']:.3f}/0.08 | 접촉 미달: "+dynamic_failed,22,MUTED,lim) else:text(d,rects,(x+20,y+220),'팔 비교 자료 없음 / 0점 또는 정상 판정 아님',24,MUTED,lim) else: a=p['association'] text(d,rects,(x+20,y+65),ASSOC[a['kind']],26,limit=lim) text(d,rects,(x+20,y+115),f"이전 {a['previous_token']} | T {a['previous_ids']}",26,limit=lim) text(d,rects,(x+20,y+165),f"현재 {a['token']} | T {a['current_ids']}",26,limit=lim) elapsed='--' if a['elapsed'] is None else f"{a['elapsed']:.3f}s" text(d,rects,(x+20,y+215),'관측 간격 '+elapsed+' / 손목 소속 정답 검증 아님',24,MUTED,lim) costs=a['costs'] or {} def number(value):return '--' if value is None else f'{value:.3f}' text(d,rects,(x+20,y+265),'현재순서 비용 '+number(costs.get('actual_orientation'))+' / 반대 '+number(costs.get('reversed_orientation')),24,limit=lim) if a['kind']=='B_RESET_SAME_T_PAIR': text(d,rects,(x+20,y+315),'같은 T의 과거 참조 / B가 선택한 연결 아님',22,MUTED,lim) if not entries:text(d,rects,(x+20,320),'표시할 연결 차이 없음' if kind=='association' else '유효한 근접 쌍 없음 / 근거 부족',28) text(d,rects,(20,1220),summary,30) text(d,rects,(20,1270),f"원본 표시 유지 / 입력 bbox {len(meta['drawn_people'])} / 유효 몸통 {len(v['bodies'])} / 누적 시험Event {v['event_count']}",26) for n,line in enumerate(legend):text(d,rects,(20,1340+n*65),line,28,MUTED) text(d,rects,(1910,1570),f"상세 {len(entries)}개 / {omitted}개 상세 생략 (전체 data 보존)",24,MUTED) text(d,rects,(1910,1630),'모든 대상 표시 유지 / 운영 Event 저장·전송 없음',24,MUTED) text(d,rects,(1910,1690),'사람색·쌍색·조건 단계는 서로 다른 의미입니다.',24,MUTED) return out,meta if __name__=='__main__': clip=os.environ['CLIP'];verify_baseline() rows,chosen,check,previous=load_rows(clip) rows=[r for r in rows if r['sample'] in chosen] dest=Path('/download')/clip;data=dest/'data' summary=json.loads((data/'engine_summary.json').read_text()) raw=(data/'review_diagnostics.jsonl').read_bytes() assert hashlib.sha256(raw).hexdigest()==summary['files']['review_diagnostics.jsonl'] views=[json.loads(line) for line in raw.splitlines()];views=[v for v in views if v['displayed']] assert len(views)==len(rows) palette={int(k):tuple(v) for k,v in json.loads((previous/'identity_palette.json').read_text()).items()} rgb_hashes=json.loads((previous/'input_rgb_sha256.json').read_text()) active={} for kind,(name,_) in KINDS.items(): out=dest/name;report=data/(kind+'_video_validation.json') if out.exists() and report.exists(): assert hashlib.sha256(out.read_bytes()).hexdigest()==json.loads(report.read_text())['sha256'] continue assert not out.exists(),('Incomplete output needs explicit resume',out) draft=dest/('.'+name+'.partial.mp4') if draft.exists(): archive=Path('/partials')/clip archive.mkdir(parents=True,exist_ok=True) draft.rename(archive/(name+'.'+str(time.time_ns())+'.partial.mp4')) container=av.open(str(draft),'w',options={'movflags':'+faststart'}) stream=container.add_stream('libx264',rate=10);stream.width=W;stream.height=H;stream.pix_fmt='yuv420p' stream.options={'crf':'21','preset':'veryfast'};stream.codec_context.thread_count=1 stream.time_base=stream.codec_context.time_base=Fraction(1,90000) active[kind]=dict(out=out,report=report,draft=draft,container=container,stream=stream, frames=0,max_labels=0,max_overlap=0,max_pairs=0) if not active: print('REUSED',clip);raise SystemExit(0) samples={0,len(rows)-1,max(range(len(rows)),key=lambda i:len(views[i]['pairs']))} for predicate in (lambda v:bool(v['current_rules']),lambda v:any(p['association']['difference'] for p in v['pairs']),lambda v:bool(v['held_support'])): idx=next((i for i,v in enumerate(views) if predicate(v)),None) if idx is not None:samples.add(idx) mapping=[];started=time.monotonic() for i,(row,v) in enumerate(zip(rows,views)): assert (row['sample'],row['source_frame'],row['seconds'])==(v['sample'],v['source_frame'],v['seconds']) path=Path('/cache/inputs/frames')/f"{row['sample']:04d}.png" assert hashlib.sha256(path.read_bytes()).hexdigest()==rgb_hashes[path.name] with Image.open(path) as im:image=im.convert('RGB') assert image.size==(row['width'],row['height']) pts=round((row['seconds']-rows[0]['seconds'])*90000) for kind,job in active.items(): frame,meta=page(image,row,v,palette,kind,i,len(rows)) if i in samples:frame.save(data/(kind+f'_sample_{i:04d}.jpg'),quality=88) expected=({p['token'] for p in v['pairs'] if p['level'] or p['highlight']} if kind=='motion' else {p['token'] for p in v['pairs'] if p['association']['difference']} if kind=='association' else set()) assert set(meta['visible_pairs'])==expected job['max_labels']=max(job['max_labels'],meta['label_count']);job['max_overlap']=max(job['max_overlap'],meta['group_overlap_pixels']) job['max_pairs']=max(job['max_pairs'],len(meta['visible_pairs'])) vf=av.VideoFrame.from_ndarray(np.asarray(frame),format='rgb24');vf.pts=pts;vf.time_base=Fraction(1,90000) for packet in job['stream'].encode(vf):job['container'].mux(packet) job['frames']+=1 mapping.append(dict(output_frame=i+1,sample=row['sample'],source_frame=row['source_frame'],source_seconds=row['seconds'],output_pts_seconds=pts/90000)) if (i+1)%200==0:print('FRAME',clip,i+1,len(rows),flush=True) for kind,job in active.items(): for packet in job['stream'].encode():job['container'].mux(packet) job['container'].close() times=[] with av.open(str(job['draft'])) as inp: inp.streams.video[0].codec_context.thread_count=1 for frame in inp.decode(video=0): assert not frame.is_corrupt and (frame.width,frame.height)==(W,H) times.append(float(frame.pts*frame.time_base)) assert len(times)==len(rows) and all(b>a for a,b in zip(times,times[1:])) error=max(abs(t-m['output_pts_seconds']) for t,m in zip(times,mapping));assert error<1/90000 job['draft'].rename(job['out']) result=dict(status='PASS',kind=kind,frames=len(rows),nominal_fps=10,resolution=[W,H], max_pts_error=error,text_overlap=0,text_overflow=0,label_overlap=0, max_labels=job['max_labels'],max_group_overlap_pixels=job['max_overlap'],max_visible_pairs=job['max_pairs'], sample_indices=sorted(samples),sha256=hashlib.sha256(job['out'].read_bytes()).hexdigest(), elapsed_seconds=time.monotonic()-started) job['report'].write_text(json.dumps(result,indent=2));print(json.dumps(result),flush=True) with (data/'frame_map.csv').open('w',newline='') as f: w=csv.DictWriter(f,fieldnames=mapping[0]);w.writeheader();w.writerows(mapping)