"""실제 끝점과 MP4 디코드의 모든 컷 상태를 모바일 비교 화면으로 만든다.""" import argparse import hashlib import json from pathlib import Path import subprocess from PIL import Image,ImageChops,ImageStat,ImageDraw ROOT=Path(__file__).resolve().parents[3] PROJECT=ROOT/'projects/water-hydrogen-bond-surface-tension' FFMPEG='C:/git/ShortVideoGenerator/projects/water-hydrogen-bond-surface-tension/renderer/node_modules/@remotion/compositor-win32-x64-msvc/ffmpeg.exe' parser=argparse.ArgumentParser() parser.add_argument('--stage',choices=['11-images','12-motion'],required=True) args=parser.parse_args() BASE=PROJECT/f'stages/{args.stage}/r2' artifact=json.loads((BASE/'artifact.json').read_text(encoding='utf-8'))['payload'] folder=BASE/'generated-source/inspection' folder.mkdir(exist_ok=True) records_path=BASE/'visual-cli-commands.json' records=json.loads(records_path.read_text(encoding='utf-8')) if records_path.exists() else [] def sheet(paths,labels,out,columns): rows=(len(paths)+columns-1)//columns result=Image.new('RGB',(360*columns,664*rows),'white') draw=ImageDraw.Draw(result) for index,(path,label) in enumerate(zip(paths,labels)): x,y=(index%columns)*360,(index//columns)*664 draw.text((x+8,y+5),label,fill='black') result.paste(Image.open(path).convert('RGB').resize((360,640),Image.Resampling.LANCZOS),(x,y+24)) result.save(out) return dict(path=out.relative_to(ROOT).as_posix(),sha256=hashlib.sha256(out.read_bytes()).hexdigest()) views=[] if args.stage=='11-images': frames={f['id']:f for f in artifact['request']['frames']} groups={} seen=set() for item in artifact['result']['frames']: frame=frames[item['frame_id']] if item['media']['path'] in seen: continue seen.add(item['media']['path']) groups.setdefault(frame['scene_name'],[]).append((ROOT/item['media']['path'],str(frame['frame_index']))) for scene,images in groups.items(): path=folder/(scene+'.png') asset=sheet([p for p,label in images],['F'+label for p,label in images],path,min(4,len(images))) views.append(dict(scene_name=scene,unique_frames=len(images),sheet=asset,visual_review='not_checked')) else: requests={c['cut_id']:c for c in artifact['request']['clips']} endpoints=json.loads((PROJECT/'stages/11-images/r2/artifact.json').read_text(encoding='utf-8'))['payload'] endpoint_media={f['frame_id']:ROOT/f['media']['path'] for f in endpoints['result']['frames']} for item in artifact['result']['clips']: clip=requests[item['cut_id']] video=ROOT/item['media']['path'] sha=hashlib.sha256(video.read_bytes()).hexdigest() count=clip['end_frame']-clip['start_frame'] indices=[0,count//2,count-1] decoded=[video.parent/f'decoded-review-{index:03}.png' for index in [1,2,3]] if not all(p.is_file() for p in decoded): commands=[ [str(Path(FFMPEG).with_name('ffprobe.exe')),'-v','error','-count_frames','-show_streams','-of','json',str(video)], [FFMPEG,'-hide_banner','-loglevel','error','-xerror','-i',str(video),'-map','0:v:0','-c:v','rawvideo','-f','null','-']] commands += [[FFMPEG,'-hide_banner','-loglevel','error','-n','-i',str(video),'-ss',str(index/50),'-frames:v','1',str(path)] for index,path in zip(indices,decoded) if not path.is_file()] for command_index,command in enumerate(commands): value=subprocess.run(command,capture_output=True,text=True,encoding='utf-8') record=dict(command=command,exit_code=value.returncode,media_sha256=sha) if value.returncode: record['error']=value.stderr records.append(record) records_path.write_text(json.dumps(records,ensure_ascii=False,indent=2)+'\n',encoding='utf-8') assert value.returncode==0,value.stderr if command_index==0: streams=json.loads(value.stdout)['streams'] assert len(streams)==1 and streams[0]['codec_name']=='h264' assert int(streams[0]['nb_read_frames'])==count record['decoded_frames']=count record['expected_frames']=count records_path.write_text(json.dumps(records,ensure_ascii=False,indent=2)+'\n',encoding='utf-8') logical=[clip['start_frame']+index for index in indices] originals=[video.parent/f'frame-{frame:08}.png' for frame in logical] errors=[sum(ImageStat.Stat(ImageChops.difference(Image.open(a).convert('RGB'),Image.open(b).convert('RGB'))).mean)/3 for a,b in zip(decoded,originals)] assert max(errors)<5, 'MP4 실제 디코드와 원본 렌더 화면 불일치' start_png=endpoint_media[clip['start_frame_id']] end_png=endpoint_media[clip['end_frame_id']] start_error=sum(ImageStat.Stat(ImageChops.difference(Image.open(start_png).convert('RGB'),Image.open(originals[0]).convert('RGB'))).mean)/3 assert start_error<.1, '11 시작 PNG와 12 실제 시작 상태 불일치' end_difference=sum(ImageStat.Stat(ImageChops.difference(Image.open(end_png).convert('RGB'),Image.open(originals[-1]).convert('RGB'))).mean)/3 path=folder/(clip['cut_id']+'.png') labels=[clip['cut_id']+' F'+str(frame) for frame in logical]+['11 end F'+str(clip['end_frame'])] asset=sheet(decoded+[end_png],labels,path,4) views.append(dict(cut_id=clip['cut_id'],scene_name=clip['scene_name'],media_sha256=sha,logical_frames=logical,end_png_logical_frame=clip['end_frame'],last_video_frame_delta=1,last_video_time_delta_ms=20,start_endpoint_png_mean_error=start_error,end_boundary_last_render_mean_difference=end_difference,decoded_frames=[dict(path=p.relative_to(ROOT).as_posix(),sha256=hashlib.sha256(p.read_bytes()).hexdigest()) for p in decoded],render_frames=[dict(path=p.relative_to(ROOT).as_posix(),sha256=hashlib.sha256(p.read_bytes()).hexdigest()) for p in originals],mean_decode_render_pixel_error=errors,sheet=asset,visual_review='not_checked')) previous=json.loads((BASE/'visual-check.json').read_text(encoding='utf-8')) if (BASE/'visual-check.json').exists() else {} for view in views: old=next((v for v in previous.get('views',[]) if v.get('sheet')==view['sheet']),None) if old and old.get('visual_review')=='checked': view['visual_review']='checked' view['observation']=old['observation'] source=artifact['result']['scene_file'] if args.stage=='11-images' else json.loads((PROJECT/'stages/11-images/r2/artifact.json').read_text(encoding='utf-8'))['payload']['result']['scene_file'] complete=len(artifact['result']['frames'])==62 if args.stage=='11-images' else len(views)==31 value=dict(result='READY_FOR_VISUAL_REVIEW' if complete else 'PARTIAL',source_sha256=source['sha256'],width=1080,height=1920,panel_width=360,panel_height=640,views=views) (BASE/'visual-check.json').write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n',encoding='utf-8') print('실제 관찰 자료',args.stage,len(views))