NINI
104 lines
7.0 KiB
Python
104 lines
7.0 KiB
Python
"""실제 끝점과 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))
|