ShortVideoGenerator/tools/blender/smoke.py

"""작은 3D 영상과 같은 논리 프레임의 서로 다른 Scene을 실제로 확인한다."""
import argparse
import json
from pathlib import Path
import shutil
import subprocess

from PIL import Image

import run
from test_runner import fixture, production_fixture, endpoint_fixture, ROOT
from test_runner import write


def pixels(path):
    with Image.open(path) as image:
        return image.convert("RGB").tobytes()


def check_motion(evidence, ffmpeg):
    frames = evidence["logical_frames"]
    checked = [frames[0], frames[len(frames) // 2], frames[-1]]
    media = evidence["media"]
    attempt = (ROOT / media["path"]).parent
    rendered_samples = [pixels(attempt / ("frame-" + str(f).zfill(8) + ".png")) for f in checked]
    # H.264 프레임마다 달라질 수 있는 양자화 흔적을 원본 3D 움직임으로 오인하지 않는다.
    run.require(len(set(rendered_samples)) == 3, "디코드 PNG RGB 시작/중간/끝에서 3D 움직임이 없습니다: " + evidence["scene_name"])
    decoded = subprocess.run([ffmpeg, "-v", "error", "-xerror", "-i", str(ROOT / media["path"]),
        "-map", "0:v:0", "-pix_fmt", "rgb24", "-c:v", "rawvideo", "-f", "image2pipe", "-"], capture_output=True)
    run.require(decoded.returncode == 0, "MP4 RGB 디코드가 실패했습니다: " + decoded.stderr.decode("utf-8", errors="replace")[-1000:])
    size = media["width"] * media["height"] * 3
    run.require(len(decoded.stdout) == size * len(frames), "실제 RGB 디코드 프레임 수가 다릅니다")
    samples = [decoded.stdout[index * size:(index + 1) * size] for index in (0, len(frames) // 2, len(frames) - 1)]
    run.require(len(set(samples)) == 3, "디코드 MP4 RGB 시작/중간/끝에서 3D 움직임이 없습니다: " + evidence["scene_name"])
    return checked


def main():
    parser = argparse.ArgumentParser(description="설치 정보가 아닌 실제 렌더·디코드 검사")
    parser.add_argument("--blender", required=True)
    parser.add_argument("--ffmpeg", required=True)
    parser.add_argument("--ffprobe", required=True)
    parser.add_argument("--engine", choices=("CYCLES", "BLENDER_EEVEE"), default="CYCLES")
    parser.add_argument("--animation-only", action="store_true", help="50fps·5프레임으로 비활성 Scene의 실제 RGB 움직임만 검사")
    args = parser.parse_args()
    folder, project, artifacts, job = fixture()
    if args.engine == "BLENDER_EEVEE":
        artifacts["08-scenes"]["payload"]["preview"]["render_profile"].update(engine="BLENDER_EEVEE", device="GPU")
        write(ROOT / job["artifact"], artifacts["08-scenes"])
    if args.animation_only:
        artifacts["08-scenes"]["payload"]["preview"]["render_profile"]["fps"] = 50
        job["end"] = 5
        write(ROOT / job["artifact"], artifacts["08-scenes"])
    source = ROOT / job["source"]
    # fixture의 검사 전 바이트는 생성 원본이 아니다. 실패 흔적은 별도 파일로 보존한다.
    source.rename(source.with_suffix(".uninspected"))
    run.execute([args.blender, "--background", "--factory-startup", "--disable-autoexec", "--python-exit-code", "1", "--python",
                 Path(__file__).with_name("smoke_fixture.py"), "--", source])
    video = run.render(ROOT, job, args.blender, args.ffmpeg, args.ffprobe)
    require = run.require
    require(video["media"]["duration_ms"] == (100 if args.animation_only else 1000), "검사 영상 길이 불일치")
    second_video = run.render(ROOT, dict(job, scene="SceneB", require_object=["BallB"]), args.blender, args.ffmpeg, args.ffprobe)
    motion_checks, errors = [], []
    for rendered in (video, second_video):
        try:
            motion_checks.append(check_motion(rendered, args.ffmpeg))
        except ValueError as error:
            errors.append(str(error))
    require(not errors, "\n".join(errors))
    first_motion, second_motion = motion_checks
    still_job = dict(job, frame=4 if args.animation_only else 10, start=None, end=None)
    first = run.render(ROOT, still_job, args.blender, args.ffmpeg, args.ffprobe)
    second = run.render(ROOT, dict(still_job, scene="SceneB", require_object=["BallB"]), args.blender, args.ffmpeg, args.ffprobe)
    require(first["identity"] != second["identity"] and pixels(ROOT / first["media"]["path"]) != pixels(ROOT / second["media"]["path"]),
            "같은 논리 프레임의 다른 Scene 카메라가 혼입되었습니다")
    if args.animation_only:
        print(json.dumps(dict(first_video=video["media"], second_video=second_video["media"],
            first_rgb_frames=first_motion, second_rgb_frames=second_motion, first_scene=first["media"], second_scene=second["media"],
            renderer=video["render_profile"], source=video["scene_file"], result="초기 Idle Scene과 무관한 두 Scene의 디코드 RGB 움직임·5프레임·100ms·같은 시각의 Scene 구별 검사 통과"), ensure_ascii=False, indent=2))
        return
    for invalid in (dict(still_job, scene="Missing"), dict(still_job, require_object=["Missing"])):
        try:
            run.render(ROOT, invalid, args.blender, args.ffmpeg, args.ffprobe)
        except ValueError:
            pass
        else:
            raise ValueError("없는 Scene/오브젝트를 거부하지 않았습니다")
    for change in ("external", "short_scene"):
        bad_source = source.with_name(change + ".blend")
        options = ["--external-image", ROOT / first["media"]["path"]] if change == "external" else ["--short-scene"]
        run.execute([args.blender, "--background", "--factory-startup", "--disable-autoexec", "--python-exit-code", "1", "--python",
                     Path(__file__).with_name("smoke_fixture.py"), "--", bad_source, *options])
        artifacts["08-scenes"]["payload"]["preview"]["scene_file"]["path"] = bad_source.relative_to(ROOT).as_posix()
        write(ROOT / job["artifact"], artifacts["08-scenes"])
        try:
            run.render(ROOT, dict(job, source=bad_source.relative_to(ROOT).as_posix()), args.blender, args.ffmpeg, args.ffprobe)
        except ValueError:
            pass
        else:
            raise ValueError("외부 참조/실제 Scene 범위 오류를 거부하지 않았습니다")
    approved_folder, approved_project, approved_artifacts, approved_job = production_fixture()
    old_source = ROOT / approved_artifacts["08-scenes"]["payload"]["preview"]["scene_file"]["path"]
    old_source.rename(old_source.with_suffix(".uninspected"))
    shutil.copyfile(source, old_source)
    approved_artifacts["08-scenes"]["payload"]["preview"]["scene_file"].update(sha256=run.digest(old_source), blender_version=video["scene_file"]["blender_version"])
    write(approved_folder / "stages/08-scenes/r2/artifact.json", approved_artifacts["08-scenes"])
    production_source = ROOT / approved_job["source"]
    production_source.rename(production_source.with_suffix(".uninspected"))
    shutil.copyfile(source, production_source)
    production = run.render(ROOT, approved_job, args.blender, args.ffmpeg, args.ffprobe)
    require(production["media"]["kind"] == "image", "승인된 10 기준 PNG가 아닙니다")
    actual_motion = None
    for motion in (False, True):
        target_folder, target_project, target_artifacts, target_job = endpoint_fixture(motion)
        for stage in ("08-scenes", "10-references", "11-images"):
            item = target_artifacts[stage]["payload"]
            result = item["preview"] if stage == "08-scenes" else item["result"]
            metadata = result["scene_file"] if result else None
            path = ROOT / (metadata["path"] if metadata else item["request"]["scene_path"])
            path.rename(path.with_suffix(".uninspected"))
            shutil.copyfile(source, path)
            if metadata:
                metadata.update(sha256=run.digest(path), blender_version=video["scene_file"]["blender_version"])
            (item["preview"] if stage == "08-scenes" else item["request"])["render_profile"].update(
                engine=args.engine, device="GPU" if args.engine == "BLENDER_EEVEE" else "CPU")
        if motion:
            target_artifacts["12-motion"]["payload"]["request"]["render_profile"].update(
                engine=args.engine, device="GPU" if args.engine == "BLENDER_EEVEE" else "CPU")
        for pointer in target_project["stages"]:
            if pointer["stage_id"] in target_artifacts:
                write(ROOT / pointer["artifact_path"], target_artifacts[pointer["stage_id"]])
        actual = run.render(ROOT, target_job, args.blender, args.ffmpeg, args.ffprobe)
        require(actual["media"]["kind"] == ("video" if motion else "image"), "승인된 11/12 렌더 형식이 다릅니다")
        if motion:
            actual_motion = actual["media"]
    for failure in folder.rglob("failure.json"):
        require(not list(failure.parent.glob("*.png")), "실패 검사에서 출력 PNG를 썼습니다")
    print(json.dumps(dict(video=video["media"], first_scene=first["media"], second_scene=second["media"],
                         production_reference=production["media"],
                         production_clip=actual_motion,
                         renderer=video["render_profile"], source=video["scene_file"], result="실제 렌더·디코드·움직임·다중 Scene·승인된 기준 생성·외부 참조 거부 검사 통과"), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()