"""Render saved observations only; no inference or identity association.""" import colorsys import csv import hashlib import json import pathlib import time from fractions import Fraction import av import numpy as np from PIL import Image, ImageDraw, ImageFont ROOT = pathlib.Path('/trial') OLD = pathlib.Path('/old') cache = OLD / 'inputs/pose.json' cache_hash = hashlib.sha256(cache.read_bytes()).hexdigest() rows = [r for r in json.loads(cache.read_text()) if 1 <= r['source_frame'] <= 723] assert len(rows) == 301 and [r['sample'] for r in rows] == list(range(301)) assert (rows[0]['source_frame'], rows[-1]['source_frame']) == (1, 722) assert (rows[0]['seconds'], rows[-1]['seconds']) == (0.0, 29.998) ids = sorted({p['track_id'] for r in rows for p in r['persons'] if p['track_id'] is not None}) palette = {tid: tuple(round(c*255) for c in colorsys.hsv_to_rgb((i*.61803398875)%1, .65, .98)) for i, tid in enumerate(ids)} assert len(set(palette.values())) == len(ids) fonts = {n: ImageFont.truetype(str(ROOT/'inputs/NotoSansCJK-Regular.ttc'), n, index=1) for n in (22, 26, 30, 34, 42)} EDGES = [(5,6),(5,7),(7,9),(6,8),(8,10),(5,11),(6,12),(11,12),(11,13),(13,15),(12,14),(14,16)] CROP = (240, 40, 1680, 640) GRAY = (190,190,190) W, H = 3200, 1800 overflow = [] def text(d, xy, s, size=30, color='#f0f3f6', limit=None): bounds = d.textbbox(xy, str(s), font=fonts[size]) if bounds[0]<0 or bounds[1]<0 or bounds[2]>W or bounds[3]>H or (limit and bounds[2]>limit): overflow.append([str(s), bounds]) d.text(xy, str(s), font=fonts[size], fill=color) def dashed(d, box, color): x1,y1,x2,y2 = box for a,b in [((x1,y1),(x2,y1)),((x2,y1),(x2,y2)),((x2,y2),(x1,y2)),((x1,y2),(x1,y1))]: length = max(abs(b[0]-a[0]), abs(b[1]-a[1])) for offset in range(0, int(length), 14): u, v = offset/max(length,1), min(offset+8,length)/max(length,1) d.line((a[0]+(b[0]-a[0])*u,a[1]+(b[1]-a[1])*u,a[0]+(b[0]-a[0])*v,a[1]+(b[1]-a[1])*v), fill=color, width=3) def overlay(image, people, crop, scale): panel = image.crop(crop).resize((round((crop[2]-crop[0])*scale), round((crop[3]-crop[1])*scale))) d = ImageDraw.Draw(panel) def point(x,y): return ((x-crop[0])*scale, (y-crop[1])*scale) for p in people: c = palette[p['track_id']] if p['track_id'] is not None else GRAY x1,y1,x2,y2 = p['bbox'] b = (*point(x1,y1), *point(x2,y2)) if p['track_id'] is None: dashed(d,b,c) else: d.rectangle(b,outline=c,width=3) kp = p['keypoints'] for a,b in EDGES: if kp[a][2]>=.3 and kp[b][2]>=.3: d.line((*point(*kp[a][:2]), *point(*kp[b][:2])),fill=c,width=3) for x,y,score in kp: if score>=.3: px,py = point(x,y) d.ellipse((px-3,py-3,px+3,py+3),fill=c) return panel out = ROOT/'outputs/camera24_tracking_skeleton_10fps.mp4' assert not out.exists() started = time.monotonic() container = av.open(str(out),'w',options={'movflags':'+faststart'}) stream = container.add_stream('libx264',rate=10) stream.width=W; stream.height=H; stream.pix_fmt='yuv420p' stream.options={'crf':'20','preset':'veryfast'} stream.codec_context.thread_count=2 stream.time_base=Fraction(1,90000); stream.codec_context.time_base=Fraction(1,90000) seen = {}; previous = set(); maps=[]; facts=[]; input_hashes=[]; crossings=[] samples = {0,100,118,126,150,200,201,250,300} for i,r in enumerate(rows): path = OLD/'inputs/frames'/f"{r['sample']:04d}.png" raw = path.read_bytes() input_hashes.append({'sample':r['sample'],'sha256':hashlib.sha256(raw).hexdigest()}) image = Image.open(path).convert('RGB') assert image.size==(r['width'],r['height'])==(1920,1080) people = sorted(r['persons'],key=lambda p:(p['track_id'] is None,p['track_id'] if p['track_id'] is not None else p['query'])) assert len(people)<=9 current = [p['track_id'] for p in people if p['track_id'] is not None] assert len(current)==len(set(current)) canvas = Image.new('RGB',(W,H),'#111820'); d=ImageDraw.Draw(canvas) text(d,(24,12),'camera24 | Tracking ID · skeleton 검토',42) text(d,(24,74),f"원본 Frame {r['source_frame']} | 상대 PTS {r['seconds']:.3f}s | 관측 {i+1}/301 | 10FPS 표본 · 정상 속도",30) loop = '반복 시작 전 구간' if r['source_frame']<242 else ('확인 반복 1: 원본 242~482' if r['source_frame']<483 else '확인 반복 2: 원본 483~723') text(d,(2240,22),loop,30,limit=3180) for boundary in (242,483): if r['source_frame']>=boundary and (i==0 or rows[i-1]['source_frame']=.3 for k in p['keypoints']) assert valid==p['valid_joints'] and p['score']>=.15 label = f'T{tid}' if tid is not None else f'미추적(q{q})' color = palette[tid] if tid is not None else GRAY state = '미추적 · q는 이 Frame의 출력 칸 번호' gap = None if tid is not None: if tid not in seen: state='이 검토 범위에서 처음 관측' elif tid in previous: state='직전 관측에도 같은 ID 있음' else: gap=i-seen[tid]-1 state=f'같은 ID 재등장 · 중간 {gap}개 관측에서 없음' seen[tid]=i # Separate fixed label rail keeps all identity text outside the people region. lx=24+j*240 text(d,(lx,142),label,30,color,limit=lx+232) cx=(p['bbox'][0]+p['bbox'][2])/2; cy=p['bbox'][1] if CROP[0]<=cx<=CROP[2] and CROP[1]<=cy<=CROP[3]: d.line((lx+65,188,20+(cx-CROP[0])*1.5,220+(cy-CROP[1])*1.5),fill=color,width=1) y=205+j*160 d.line((2240,y-10,3170,y-10),fill='#43505f',width=1) text(d,(2240,y),label,34,color) text(d,(2600,y),f"score {p['score']:.3f} 관절 {valid}/17",30,limit=3180) text(d,(2240,y+55),state,26,limit=3180) text(d,(2240,y+96),f'query {q} · 원본 bbox / skeleton 그대로',22,'#b5c0cb') observation.append({'track_id':tid,'query':q,'score':p['score'],'valid_joints':valid,'state':state,'gap_observations':gap}) for k,line in enumerate([ '읽는 법', 'T숫자 = cache의 실제 Tracking ID', '색은 ID별 고정, 정체성 판단은 숫자로', '회색 점선 / 회색 skeleton = 미추적', 'q 반복은 같은 사람 추적을 뜻하지 않음', '같은 ID 재등장 ≠ 동일인 복구 확인', '미관측 ≠ Tracker 종료 확인', '사람 score ≥ 0.15 · 관절 score ≥ 0.30', '추론·Tracking 재실행 / 관절 보간 없음', '색이 비슷하면 반드시 T숫자로 비교', ]): text(d,(1020,1200+k*49),line,26 if k else 34) assert not overflow, overflow if i in samples: canvas.save(ROOT/'reports'/f'sample_{i:03d}_frame_{r["source_frame"]}.jpg',quality=88) vf=av.VideoFrame.from_ndarray(np.asarray(canvas),format='rgb24') vf.pts=round(r['seconds']*90000); vf.time_base=Fraction(1,90000) for packet in stream.encode(vf): container.mux(packet) maps.append({'output_frame_1based':i+1,'sample':r['sample'],'source_frame_1based':r['source_frame'],'source_relative_pts_seconds':r['seconds'],'output_pts_seconds':vf.pts/90000,'tracked':len(current),'untracked':len(people)-len(current)}) facts.append({'sample':r['sample'],'source_frame':r['source_frame'],'persons':observation}) previous=set(current) if i%50==0: print('rendered',i+1,flush=True) for packet in stream.encode(): container.mux(packet) container.close() with av.open(str(out)) as video: times=[float(f.pts*f.time_base) for f in video.decode(video=0)] assert len(times)==301 and all(abs(t-r['seconds'])<.001 for t,r in zip(times,rows)) assert all(b>a for a,b in zip(times,times[1:])) assert hashlib.sha256(cache.read_bytes()).hexdigest()==cache_hash with (ROOT/'outputs/frame_map.csv').open('w',newline='') as f: writer=csv.DictWriter(f,fieldnames=maps[0]); writer.writeheader(); writer.writerows(maps) (ROOT/'outputs/observation_states.json').write_text(json.dumps(facts,ensure_ascii=False,indent=2)) (ROOT/'reports/input_hashes.json').write_text(json.dumps(input_hashes,indent=2)) report={'status':'PASS','frames':len(times),'first_source_frame':1,'last_source_frame':722,'first_pts':times[0],'last_pts':times[-1], 'max_pts_error':max(abs(t-r['seconds']) for t,r in zip(times,rows)), 'nominal_fps':10,'preserve_sample_pts':True, 'crop':CROP,'resolution':[W,H],'font':fonts[30].getname(),'unique_ids':ids,'palette':palette,'crossings':crossings, 'text_overflow':overflow,'label_rail_slots':9,'max_persons':max(len(r['persons']) for r in rows), 'pose_cache_sha256':cache_hash,'seconds_elapsed':round(time.monotonic()-started,2),'cpu_only':True, 'untracked_history_association':False,'video_sha256':hashlib.sha256(out.read_bytes()).hexdigest()} (ROOT/'reports/validation.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)) print(json.dumps(report,ensure_ascii=False),flush=True)