"""CPU rendering of immutable RGB and PoseC3D cache; no inference imports.""" import csv import hashlib import json import os import pathlib import shutil import time from collections import Counter from fractions import Fraction import av import numpy as np from PIL import Image, ImageDraw, ImageFont, ImageOps from contract import THRESHOLDS, NAMES, VARIANTS, WIDTH, HEIGHT, observation, display_state, score_text CLIP = os.environ['CLIP'] SRC = pathlib.Path('/source/clips') / CLIP OUT = pathlib.Path('/download') / CLIP DATA = OUT / 'data'; DATA.mkdir(parents=True, exist_ok=True) RAW_PATH = SRC / 'outputs/posec3d.json' RAW = json.loads(RAW_PATH.read_text()) ROWS = json.loads((SRC / 'inputs/pose.json').read_text()) SUMMARY = json.loads((SRC / 'reports/posec3d_summary.json').read_text()) assert SUMMARY['status'] == 'PASS' and SUMMARY['temporal_samples'] == 16 and SUMMARY['observation_fps'] == 10 assert SUMMARY['prediction_count'] == len(RAW) LABELS = {int(r['output_index']): r['class_name'] for r in csv.DictReader(open('/classes.csv'))} assert all(LABELS[k] == v for k, v in NAMES.items()) and len(LABELS) == 60 EXPECTED_HASHES = json.loads(pathlib.Path('/expected.json').read_text()) for filename, path in [('posec3d_raw.json', RAW_PATH), ('posec3d_summary.json', SRC/'reports/posec3d_summary.json'), ('classes_posec3d.csv', pathlib.Path('/classes.csv'))]: assert hashlib.sha256(path.read_bytes()).hexdigest() == EXPECTED_HASHES[f'{CLIP}/data/{filename}'] DECISIONS = [observation(r) for r in RAW] FONTS = {n: ImageFont.truetype('/font.ttc', n, index=1) for n in (28, 30, 32, 34, 40, 44)} BG, FG, MUTED, ACTIVE = '#111820', '#edf1f5', '#aeb8c2', '#ffae57' LABEL_STATE = {'SIGNALS': '개별 신호 확인 / 필터 미적용', 'PASS': '표시 조건 충족', 'NO_MATCH': '표시 조건 미충족', 'INPUT_INSUFFICIENT': '입력 구간 부족 / 점수 없음', 'NO_VALID_SKELETON': '입력 구간의 유효 skeleton 없음', 'NO_SCORES': '행동 점수 없음'} TITLES = {'signals_3levels': '개별 신호 / 3단계 기준', 'filtered_ge2': '기본값 충족 2종류 이상', 'filtered_ge3': '기본값 충족 3종류 이상'} def text(draw, boxes, xy, value, size=32, color=FG, right=3180): value = str(value); b = draw.textbbox(xy, value, font=FONTS[size]) assert 0 <= b[0] and 0 <= b[1] and b[2] <= right and b[3] <= HEIGHT, (value, b) assert not any(b[0] < a[2] and b[2] > a[0] and b[1] < a[3] and b[3] > a[1] for a in boxes), ('text overlap', value, b) boxes.append(b); draw.text(xy, value, font=FONTS[size], fill=color) def template(variant, shown): canvas = Image.new('RGB', (WIDTH, HEIGHT), BG); draw = ImageDraw.Draw(canvas); boxes = [] text(draw, boxes, (20, 18), f'{CLIP} | PoseC3D | {TITLES[variant]}', 44) for y, value in [(1450, '평가 단위: 화면 전체 / 특정 사람의 행동 판정 아님'), (1510, '조건은 원시 score 기준 / 반올림은 표시에만 적용'), (1570, '시간 누적·이전 결과 유지·Top5 조건 없음'), (1630, '주황 강조는 표시 조건 충족 / 싸움 확정·확률 아님'), (1690, '동일 10FPS 관측 / 원본 Frame·PTS 대응 유지')]: text(draw, boxes, (20, y), value, 30, MUTED, right=1880) if shown: for slot, (key, name) in enumerate(NAMES.items()): x, y = 1910, 150 + slot * 340 draw.rectangle((x, y, 3180, y + 320), outline='#43515f', width=2) text(draw, boxes, (x+18, y+18), f'{key:02d} {name}', 30, ACTIVE) if variant == 'signals_3levels': for col, title in enumerate(('낮은 기준', '기본값', '높은 기준')): xx = x+18+col*410 text(draw, boxes, (xx, y+130), title, 28, MUTED, right=xx+390) text(draw, boxes, (xx, y+176), f'>= {THRESHOLDS[key][col]:.5f}', 28, right=xx+390) else: text(draw, boxes, (x+18, y+145), f'기본값 기준: score >= {THRESHOLDS[key][1]:.5f}', 32) text(draw, boxes, (1910, 1550), 'NTU60 0-based ID / 주4개 고정 순서', 30, MUTED) text(draw, boxes, (1910, 1610), '기본값 충족 클래스만 각각 1회 집계', 30, MUTED) text(draw, boxes, (1910, 1670), '낮은/높은 기준은 개수 판정에 사용하지 않음', 30, MUTED) else: text(draw, boxes, (1930, 250), '현재 관측의 상세 점수 표시는 숨깁니다.', 32, MUTED) text(draw, boxes, (1930, 320), '원시 점수와 조건 근거는 data에 보존합니다.', 30, MUTED) text(draw, boxes, (1930, 390), '미충족·자료 없음은 정상 행동 판정이 아닙니다.', 30, MUTED) return canvas, boxes def draw_frame(variant, index, rgb, templates): r, d = RAW[index], DECISIONS[index] state = display_state(d, variant); shown = state in ('SIGNALS', 'PASS') base, static_boxes = templates[shown]; canvas = base.copy(); boxes = static_boxes.copy() canvas.paste(rgb, (20, 150)); draw = ImageDraw.Draw(canvas) text(draw, boxes, (20, 85), f"Source Frame {r['source_frame']} | PTS {r['seconds']:.3f}s | sample {r['sample']} | 10FPS", 34) color = ACTIVE if state == 'PASS' else FG draw.rectangle((18,148,1878,1196), outline=color if state == 'PASS' else '#43515f', width=4) text(draw, boxes, (20,1240), LABEL_STATE[state], 44, color, right=1880) if d['valid']: ids = ', '.join(str(k) for k in d['passed_ids']) or '없음' text(draw, boxes, (20,1310), f"기본값 충족 {d['count']}/4 | 클래스: {ids}", 32, right=1880) else: text(draw, boxes, (20,1310), '충족 개수: 계산하지 않음 / 점수를 0으로 채우지 않음', 32, MUTED, right=1880) w = r.get('window') window = f"입력16관측: {w['start_seconds']:.3f} ~ {w['end_seconds']:.3f}s / 미래 관측 포함" if w else '입력16관측 미확보 / 결과를 복제하지 않음' text(draw, boxes, (20,1380), window, 32, MUTED, right=1880) if shown: for slot, key in enumerate(NAMES): x, y = 1910, 150+slot*340; c = d['classes'][str(key)] text(draw, boxes, (x+18,y+68), 'score '+score_text(c['score']), 40) if variant == 'signals_3levels': for col, flag in enumerate(('low','basic','high')): xx = x+18+col*410 text(draw, boxes, (xx,y+232), '충족' if c[flag] else '미충족', 32, ACTIVE if c[flag] else MUTED, right=xx+390) else: text(draw, boxes, (x+18,y+220), '충족' if c['basic'] else '미충족', 40, ACTIVE if c['basic'] else MUTED) return canvas, state, shown def render_variant(variant): stem = 'PoseC3D_'+variant file, report = OUT/(stem+'.mp4'), DATA/(stem+'_validation.json') if report.exists(): check = json.loads(report.read_text()) assert check['status'] == 'PASS' and hashlib.sha256(file.read_bytes()).hexdigest() == check['sha256'] return check assert not file.exists(), 'Partial output exists; inspect before retry' started = time.monotonic(); templates = {b: template(variant,b) for b in (True,False)} samples = {0, len(RAW)-1}; states_seen = set() for i, d in enumerate(DECISIONS): state = display_state(d, variant) if state not in states_seen: samples.add(i); states_seen.add(state) with av.open(str(file), 'w', options={'movflags':'+faststart'}) as container: stream = container.add_stream('libx264', rate=10) stream.width, stream.height, stream.pix_fmt = WIDTH, HEIGHT, 'yuv420p' stream.options = {'preset':'veryfast','crf':'21'}; stream.codec_context.thread_count = 2 stream.time_base = stream.codec_context.time_base = Fraction(1,90000) mappings = []; rgb_hashes = {} for index, record in enumerate(RAW): row = ROWS[record['sample']] assert (row['source_frame'], row['seconds']) == (record['source_frame'], record['seconds']) if index: assert record['sample'] > RAW[index-1]['sample'] and record['seconds'] > RAW[index-1]['seconds'] if DECISIONS[index]['valid']: assert record['items'][0]['input_shape'] == [1,1,17,16,64,64] assert record['window']['samples'] == list(range(record['sample']-8,record['sample']+8)) path = (pathlib.Path('/camera/inputs/frames') if CLIP == '01_camera24' else SRC/'inputs/frames')/f"{record['sample']:04d}.png" rgb_hashes[path.name] = hashlib.sha256(path.read_bytes()).hexdigest() with Image.open(path) as image: assert image.size == (row['width'],row['height']) fitted = ImageOps.contain(image.convert('RGB'), (1856,1044)) rgb = Image.new('RGB',(1856,1044),'black') rgb.paste(fitted,((1856-fitted.width)//2,(1044-fitted.height)//2)) canvas, state, shown = draw_frame(variant,index,rgb,templates) if index in samples: canvas.save(DATA/f'{stem}_sample{index}.jpg',quality=85) frame = av.VideoFrame.from_ndarray(np.asarray(canvas),format='rgb24') frame.pts = round((record['seconds']-RAW[0]['seconds'])*90000); frame.time_base = Fraction(1,90000) for packet in stream.encode(frame): container.mux(packet) mappings.append(dict(output_frame=index+1,sample=record['sample'],source_frame=record['source_frame'], source_seconds=record['seconds'],output_pts_seconds=frame.pts/90000, state=state,detail_visible=shown,basic_count=DECISIONS[index]['count'])) for packet in stream.encode(): container.mux(packet) with av.open(str(file)) as video: frames = [float(f.pts*f.time_base) for f in video.decode(video=0)] assert len(frames) == len(RAW) error = max(abs(t-m['output_pts_seconds']) for t,m in zip(frames,mappings)); assert error < 1/90000 with (DATA/(stem+'_frame_map.csv')).open('w',newline='') as f: writer = csv.DictWriter(f,fieldnames=list(mappings[0])); writer.writeheader(); writer.writerows(mappings) result = dict(status='PASS',clip=CLIP,variant=variant,frames=len(frames),resolution=[WIDTH,HEIGHT], nominal_fps=10,last_pts=frames[-1],max_pts_error=error,text_overflow=0,text_overlap=0, state_counts=dict(Counter(m['state'] for m in mappings)),sample_indices=sorted(samples), sha256=hashlib.sha256(file.read_bytes()).hexdigest(),elapsed_seconds=round(time.monotonic()-started,3)) report.write_text(json.dumps(result,indent=2)) (DATA/'rgb_sha256.json').write_text(json.dumps(rgb_hashes,indent=2)) print(json.dumps(result),flush=True); return result if __name__ == '__main__': selected = ['filtered_ge2','filtered_ge3'] for variant in selected: render_variant(variant) for src, name in [(RAW_PATH,'posec3d_raw.json'),(SRC/'reports/posec3d_summary.json','posec3d_summary.json'), (pathlib.Path('/classes.csv'),'classes_posec3d.csv')]: shutil.copyfile(src,DATA/name) (DATA/'filter_decisions.json').write_text(json.dumps(DECISIONS,ensure_ascii=False,separators=(',',':'))) n = sum(d['valid'] for d in DECISIONS) result = dict(clip=CLIP,total=len(RAW),valid=n,excluded=len(RAW)-n, exclusions=dict(Counter(d['state'] for d in DECISIONS if not d['valid'])), count_histogram=dict(Counter(d['count'] for d in DECISIONS if d['valid'])), ge2=sum(d['ge2'] is True for d in DECISIONS),ge3=sum(d['ge3'] is True for d in DECISIONS), thresholds=THRESHOLDS,source_pose_sha256=hashlib.sha256((SRC/'inputs/pose.json').read_bytes()).hexdigest()) result.update(ge2_ratio=result['ge2']/n if n else None,ge3_ratio=result['ge3']/n if n else None) (DATA/'filter_summary.json').write_text(json.dumps(result,indent=2))