From 31efdd668197f74334b970f70328fd427d198c62 Mon Sep 17 00:00:00 2001 From: NINI Date: Fri, 9 Oct 2026 23:54:55 +0900 Subject: [PATCH] =?UTF-8?q?Blender=20=EC=9B=90=EB=B3=B8=20=EA=B2=80?= =?UTF-8?q?=EC=82=AC=EC=99=80=20=EC=8A=B9=EC=9D=B8=20=EA=B2=BD=EA=B3=84?= =?UTF-8?q?=EB=A5=BC=20=EC=A7=80=ED=82=A4=EB=8A=94=20=EB=A1=9C=EC=BB=AC=20?= =?UTF-8?q?=EB=A0=8C=EB=8D=94=20=EC=8B=A4=ED=96=89=EA=B8=B0=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/blender/AGENTS.md | 9 + tools/blender/README.md | 53 +++++ tools/blender/inside.py | 93 ++++++++ tools/blender/run.py | 411 +++++++++++++++++++++++++++++++++ tools/blender/smoke.py | 102 ++++++++ tools/blender/smoke_fixture.py | 49 ++++ tools/blender/test_runner.py | 375 ++++++++++++++++++++++++++++++ 7 files changed, 1092 insertions(+) create mode 100644 tools/blender/AGENTS.md create mode 100644 tools/blender/README.md create mode 100644 tools/blender/inside.py create mode 100644 tools/blender/run.py create mode 100644 tools/blender/smoke.py create mode 100644 tools/blender/smoke_fixture.py create mode 100644 tools/blender/test_runner.py diff --git a/tools/blender/AGENTS.md b/tools/blender/AGENTS.md new file mode 100644 index 0000000..2b059a7 --- /dev/null +++ b/tools/blender/AGENTS.md @@ -0,0 +1,9 @@ +# Blender 실행기 작업 기준 + +이 폴더는 기존 Blender·FFmpeg 실행파일로 저장소 안의 원본과 요청을 검사하고 PNG·무음 MP4를 렌더한다. 주제별 모델 생성은 담당 제작물에서 작성한다. `bpy`는 Blender 내부에서만 사용하고 Python·conda·Blender 전역 설정을 변경하지 않는다. + +`run.py`는 승인·현재 포인터·선행 버전·시간·실제 파일을 검사한다. `inside.py`는 원본을 실제 열고 Scene·카메라·객체·외부 의존성과 지원 설정을 확인한다. 논리 프레임에 1을 더해 Blender 프레임으로 평가하고 영상 범위는 `[start,end)`다. 다른 Scene의 같은 논리 프레임을 하나의 화면으로 취급하지 않는다. + +08 대표 렌더와 승인 후 정식 10–12 렌더를 분리한다. project.json·artifact.json·사용자 승인 기록은 변경하지 않는다. source 검사 결과와 요청된 PNG/MP4의 실제 결과도 분리한다. 기계 검사는 화면 품질·전체 재생·청취를 대체하지 않는다. + +결과는 새 시도 폴더에만 쓴다. 캐시를 재사용하거나 실패 파일을 삭제하지 않는다. .blend 해시·프로필·Scene·논리 프레임·실행 스크립트 해시를 시도 식별에 포함한다. 변경 후 `test_runner.py`, `contracts/check.py`와 해당 경로의 실제 `smoke.py`를 실행하고 명령·종료코드·한계를 보고한다. 테스트 결과와 .cache는 소스 커밋에 포함하지 않는다. diff --git a/tools/blender/README.md b/tools/blender/README.md new file mode 100644 index 0000000..6c37de3 --- /dev/null +++ b/tools/blender/README.md @@ -0,0 +1,53 @@ +# 저장소 안의 Blender 렌더 실행기 + +필수 제작 입력은 기존대로 주제·언어다. 아래 인자는 담당자가 작성한 요청을 실제 실행하기 위한 내부 값이다. 이미 설치된 Blender와 기존 Remotion 패키지에 들어 있는 FFmpeg·ffprobe를 지정한다. 설치·전역 설정 변경·외부 서비스 연결은 하지 않는다. Python 실행 환경에 기존 `jsonschema`가 있어야 하며 `bpy`는 Blender 내부에서 실행한다. + +## 원본 검사 + +저장소 루트에서 실행한다. 파일 경로는 저장소 상대 경로다. 실행파일 경로는 명령에만 넣고 제작 JSON에 기록하지 않는다. + +```powershell +C:/Users/baram/anaconda3/python.exe -B -X utf8 tools/blender/run.py inspect --blender 'C:/Program Files/Blender Foundation/Blender 5.2/blender.exe' --source projects/<제작물>/stages/08-scenes/r2/sample.blend --scene Droplet --require-object Droplet +``` + +원본을 실제 열어 SHA-256·Blender 버전·Scene·카메라·객체 이름과 외부 의존성을 확인한다. 필요한 객체 대신 컬렉션 이름을 지정할 수 있다. 외부 .blend 링크·포장되지 않은 이미지/글꼴·외부 미디어/볼륨·외부 캐시/영상 시퀀스는 거부한다. 외부 파일 없이 재현할 수 있는 모델과 키프레임이 대상이다. 생성 예정 원본은 `verified=false`로 시작하고, 실제 검사 뒤에 받은 `scene_file`을 담당자가 결과에 반영한다. 검사 전에 검증 메타데이터를 요구하는 순환 조건을 만들지 않는다. + +## 대표 렌더와 정식 제작 + +대표 렌더는 총괄이 새 08 초안의 현재 포인터를 설정한 뒤 실행한다. topic/story의 기존 유효 승인은 필요하지만 새 scenes 승인은 아직 필요하지 않다. 08의 `preview.scene_file`과 `preview.render_profile`을 먼저 작성한다. 대표 컷 1–3개만 지정할 수 있으며 한 영상은 5초 이하다. 전체 장면은 08 본문에 설계하되 대표 렌더는 제한된 품질 검토 근거다. + +```powershell +C:/Users/baram/anaconda3/python.exe -B -X utf8 tools/blender/run.py render --mode preview --project projects/<제작물>/project.json --artifact projects/<제작물>/stages/08-scenes/r2/artifact.json --output-revision 2 --source projects/<제작물>/stages/08-scenes/r2/sample.blend --scene Droplet --require-object Droplet --cut-id c01 --frame 0 --blender 'C:/Program Files/Blender Foundation/Blender 5.2/blender.exe' --ffmpeg '<기존 ffmpeg.exe>' --ffprobe '<기존 ffprobe.exe>' +``` + +이미지는 `--frame N`, 영상은 `--start N --end M`을 사용한다. 논리 0은 Blender 1이며 영상은 `[N,M)`다. 컷 끝점 PNG와 MP4의 마지막 프레임은 다르다. 이번 물 영상의 전체 논리 끝점은 5528이지만, 최종 MP4는 0–5527의 5528프레임이다. 인접 cut의 같은 논리 시각도 Scene 이름을 별도로 지정한다. + +정식 제작은 `--mode production`과 1.1·visual_engine=Blender인 현재 진행 기록 및 10/11/12 요청을 사용한다. 기존 1.0 선행 단계는 보존·참조한다. 현재 topic/story/scenes 승인·feedback·승인 버전, 유효한 선행 입력, 실제 단위/통합 WAV·컷·선행 .blend/PNG/MP4를 확인한다. pending/changes_requested/stale 또는 옛 08/09 승인으로 새 제작을 시작할 수 없다. 원본 해시 변경·Scene/객체 누락·실측과 다른 프레임·다른 제작물 파일·이전 시각 제작 버전의 파일·누락 참조·프로필 불일치도 거부한다. 08/10/11의 원본과 결과 미디어는 해당 artifact 자신의 버전 폴더 아래에 있어야 한다. 기존 WAV의 media/06-tts/r1 경로는 유지할 수 있다. + +| 대상 | 지정할 요청 항목 | +|---|---| +| 10 기준 렌더 | `--reference-id 기준ID --frame 논리프레임 --require-object 실제객체명` | +| 11 컷 끝점 PNG | `--frame-id 프레임ID --frame 요청frame_index` | +| 12 컷 영상 | `--cut-id 컷ID --start 요청start_frame --end 요청end_frame` | + +10 원본은 같은 새 단계 버전 아래에 있어야 한다. 11은 request.scene_path, 12는 검증된 11 result.scene_file을 그대로 사용한다. 11/12 Scene은 요청의 이름과 같아야 한다. cut 연결은 양쪽 Scene 이름을 구별한다. continuous 연결은 같은 Scene·논리 프레임을 쓰며, 완료된 11 결과의 두 경계 ID는 같은 실제 PNG 경로를 공유한다. CLI는 출력 경로를 자동으로 공유하지 않으므로 담당자가 동일 경계의 한 렌더를 양쪽 ID에 참조한다. + +렌더 엔진·장치·fps·해상도·샘플·seed는 해당 요청의 RenderProfile에서 읽고 실제 적용 값을 대조한다. EEVEE는 GPU, Cycles는 CPU/GPU다. 이 Blender의 EEVEE에는 임의 seed 설정 API가 없으므로 기본 seed=0만 허용하며 다른 seed는 실패한다. Cycles에서는 실제 seed를 적용한다. Cycles GPU는 기존 OPTIX 장치를 실제 확인하고 미지원이면 실패한다. 이 경로의 GPU 성공은 따로 검사해야 한다. 요청에 고정된 설정을 조용히 다른 엔진으로 바꾸지 않는다. H.264를 위해 해상도는 짝수이고 9:16이어야 한다. + +## 결과와 실패 기록 + +렌더는 해당 artifact.json과 같은 버전 폴더의 `render-attempts/<식별값>-<유일ID>/`에만 PNG·MP4·로그·검사 결과를 쓴다. 식별값은 원본 전체 SHA-256·프로필·Scene·프레임·Blender 내부 스크립트 해시를 포함한다. 이전 결과나 캐시를 재사용하지 않는다. 새 시도는 이전 시도를 덮어쓰지 않으며 실패/부분 파일도 보존한다. + +`evidence.json`에는 실제 검사한 scene_file·render_profile·scene_name·logical_frames·media와 `visual_review="not_checked"`가 있다. PNG/MP4를 실제 probe·전체 디코드하고 해상도·코덱·프레임 수·fps·길이를 확인한 뒤에만 media.verified=true를 쓴다. PNG는 실제 PNG, MP4는 단일 무음 H.264 트랙이다. 실패 시 failure.json에 이유를 남기며 성공 증거를 만들지 않는다. + +이 증거는 전체 단계 완료·사용자 승인·미적 품질 평가를 뜻하지 않는다. 실행기는 artifact.json과 project.json을 수정하지 않는다. 담당자가 실제 결과를 자기 새 버전 결과에 반영하고 총괄이 완료·검토 상태를 확인한다. 08 대표 결과로 10–12를 완료 처리하지 않는다. + +## 검증 + +```powershell +C:/Users/baram/anaconda3/python.exe -B -X utf8 tools/blender/test_runner.py +C:/Users/baram/anaconda3/python.exe -B -X utf8 contracts/check.py +C:/Users/baram/anaconda3/python.exe -B -X utf8 tools/blender/smoke.py --engine BLENDER_EEVEE --blender 'C:/Program Files/Blender Foundation/Blender 5.2/blender.exe' --ffmpeg 'C:/git/ShortVideoGenerator/projects/water-hydrogen-bond-surface-tension/renderer/node_modules/@remotion/compositor-win32-x64-msvc/ffmpeg.exe' --ffprobe 'C:/git/ShortVideoGenerator/projects/water-hydrogen-bond-surface-tension/renderer/node_modules/@remotion/compositor-win32-x64-msvc/ffprobe.exe' +``` + +`smoke.py`는 사용자 영상과 별개로 .cache에 작은 움직이는 구·두 Scene·모의 승인 기록을 만든다. 90×160·10fps·1초로 실제 .blend 재열기·PNG/MP4 전체 디코드·움직임·다른 Scene의 같은 논리 시각·승인된 10/11/12 요청 실행·누락 Scene/객체·외부 의존성·실제 Scene 범위 실패를 확인한다. `--engine CYCLES`는 CPU 경로다. 제작물·기존 실제 승인 기록은 변경하지 않으며, 이것을 사용자의 물 영상 제작 성공이나 시각 품질 승인으로 보고하지 않는다. diff --git a/tools/blender/inside.py b/tools/blender/inside.py new file mode 100644 index 0000000..3bc4344 --- /dev/null +++ b/tools/blender/inside.py @@ -0,0 +1,93 @@ +"""Blender 내부 전용. 사용자 startup 설정을 바꾸거나 원본을 저장하지 않는다.""" +import json +from pathlib import Path +import sys + +import bpy + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def main(): + job = json.loads(Path(sys.argv[sys.argv.index("--") + 1]).read_text(encoding="utf-8")) + require(not bpy.data.libraries, "외부 .blend 라이브러리 링크를 허용하지 않습니다") + for image in bpy.data.images: + require(image.source in ("GENERATED", "VIEWER") or image.packed_file is not None, + "외부 이미지/영상 참조를 허용하지 않습니다: " + image.name) + require(not bpy.data.movieclips and not bpy.data.sounds and not bpy.data.volumes, + "외부 미디어·볼륨을 포함한 원본을 허용하지 않습니다") + for font in bpy.data.fonts: + require(font.filepath == "" or font.packed_file is not None, "외부 글꼴은 원본에 포함해야 합니다") + scene = bpy.data.scenes.get(job["scene"]) if job["scene"] else next(iter(bpy.data.scenes), None) + require(scene is not None and scene.camera is not None, "실제 Scene과 카메라가 필요합니다") + for name in job["require_object"]: + collection = bpy.data.collections.get(name) + require(name in scene.objects or (collection and any(o.name in scene.objects for o in collection.all_objects)), + "Scene에 필요한 오브젝트/컬렉션이 없습니다: " + name) + for item in bpy.data.objects: + require(not any(m.type in ("MESH_CACHE", "MESH_SEQUENCE_CACHE", "FLUID") for m in item.modifiers), + "외부 캐시를 쓰는 오브젝트를 허용하지 않습니다") + require(not bpy.data.cache_files, "외부 캐시 파일을 허용하지 않습니다") + require(all(not s.sequence_editor or not s.sequence_editor.strips for s in bpy.data.scenes), "외부 영상 시퀀스를 허용하지 않습니다") + report = dict(blender_version=bpy.app.version_string, scene_name=scene.name, + objects=sorted(o.name for o in scene.objects), + frame_start=scene.frame_start, frame_end=scene.frame_end, + logical_to_blender="logical_frame + 1", external_links=False) + if job["operation"] == "render": + profile = job["profile"] + scene.render.engine = profile["engine"] + scene.render.resolution_x = profile["width"] + scene.render.resolution_y = profile["height"] + scene.render.resolution_percentage = 100 + scene.render.fps = profile["fps"] + scene.render.fps_base = 1 + scene.render.image_settings.file_format = "PNG" + scene.render.image_settings.color_mode = "RGB" + scene.render.image_settings.color_depth = "8" + scene.render.film_transparent = False + scene.render.use_file_extension = True + scene.render.use_compositing = False + scene.render.use_sequencer = False + if profile["engine"] == "CYCLES": + scene.cycles.seed = profile["seed"] + scene.cycles.samples = profile["samples"] + scene.cycles.use_adaptive_sampling = False + scene.cycles.use_animated_seed = False + scene.cycles.device = profile["device"] + if profile["device"] == "GPU": + preferences = bpy.context.preferences.addons["cycles"].preferences + preferences.compute_device_type = "OPTIX" + preferences.get_devices() + gpus = [d for d in preferences.devices if d.type == "OPTIX"] + require(gpus, "Cycles OPTIX GPU가 확인되지 않았습니다") + for device in preferences.devices: + device.use = device in gpus + report["gpu_devices"] = [d.name for d in gpus] + else: + require(profile["device"] == "GPU", "EEVEE는 GPU 경로입니다") + require(profile["seed"] == 0, "EEVEE는 기본 seed=0 경로만 지원합니다") + require(hasattr(scene, "eevee") and hasattr(scene.eevee, "taa_render_samples"), + "이 Blender의 EEVEE 샘플 설정을 확인할 수 없습니다") + scene.eevee.taa_render_samples = profile["samples"] + actual = dict(engine=scene.render.engine, device=scene.cycles.device if profile["engine"] == "CYCLES" else "GPU", + fps=scene.render.fps, width=scene.render.resolution_x, height=scene.render.resolution_y, + samples=scene.cycles.samples if profile["engine"] == "CYCLES" else scene.eevee.taa_render_samples, + seed=scene.cycles.seed if profile["engine"] == "CYCLES" else 0) + require(actual == profile, "요청 설정이 Blender에서 다르게 적용되었습니다") + require(all(scene.frame_start <= f + 1 <= scene.frame_end for f in job["frames"]), + "논리 프레임이 실제 Scene 범위 밖입니다") + for frame in job["frames"]: + scene.frame_set(frame + 1) + scene.render.filepath = str(Path(job["directory"]) / ("frame-" + str(frame).zfill(8) + ".png")) + bpy.ops.render.render(write_still=True, scene=scene.name) + report["render_profile"] = actual + report["seed_policy"] = "cycles_seed" if profile["engine"] == "CYCLES" else "eevee_default_no_seed_api" + report["rendered_logical_frames"] = job["frames"] + Path(job["result"]).write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/tools/blender/run.py b/tools/blender/run.py new file mode 100644 index 0000000..ac04fa1 --- /dev/null +++ b/tools/blender/run.py @@ -0,0 +1,411 @@ +"""저장소 안의 실제 Blender 원본과 승인된 요청을 검사·렌더한다.""" +import argparse +from fractions import Fraction +import hashlib +import json +from pathlib import Path +import subprocess +import sys +import uuid +import wave + +from jsonschema import Draft202012Validator + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) +from contracts.check import semantic_errors + +NEEDED = { + "08-scenes": {"03-script", "07-cuts"}, + "10-references": {"02-research", "08-scenes", "09-entities"}, + "11-images": {"02-research", "03-script", "07-cuts", "08-scenes", "09-entities", "10-references"}, + "12-motion": {"07-cuts", "08-scenes", "11-images"}, +} + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def read(path): + return json.loads(path.read_text(encoding="utf-8")) + + +def relative(root, value): + path = Path(value) + require(not path.is_absolute() and ":" not in value and ".." not in path.parts, + "저장소 상대 경로가 필요합니다: " + value) + result = (root / path).resolve() + require(result.is_relative_to(root.resolve()), "저장소 밖의 경로입니다") + return result + + +def digest(path): + with path.open("rb") as source: + return hashlib.file_digest(source, "sha256").hexdigest() + + +def check_source(root, metadata): + path = relative(root, metadata["path"]) + require(path.is_file() and path.suffix.lower() == ".blend", "실제 .blend 원본이 필요합니다") + actual = digest(path) + if metadata["verified"]: + require(actual == metadata["sha256"], "원본 SHA-256이 바뀌었습니다") + return path, actual + + +def prepare(root, job): + """쓰기 전에 현재 포인터·선행 참조·승인·시간과 요청 항목을 검사한다.""" + validate = {name: Draft202012Validator(read(root / "contracts" / (name + ".schema.json"))) + for name in ("project", "artifact")} + project_path = relative(root, job["project"]) + project = read(project_path) + validate["project"].validate(project) + require(project["schema_version"] == "1.1" and project["settings"].get("visual_engine") == "Blender", + "새 Blender 실행 대상 진행 기록은 1.1·visual_engine=Blender여야 합니다") + pointers = {stage["stage_id"]: stage for stage in project["stages"]} + context = {} + loading = set() + + def load(stage, complete=True): + if stage in context: + return context[stage] + pointer = pointers[stage] + require(pointer["artifact_path"] and pointer["revision"], "현재 선행 결과가 없습니다: " + stage) + require(pointer["status"] == "complete" if complete else pointer["status"] in + ("draft", "awaiting_input", "awaiting_review", "complete"), "현재 단계 상태가 유효하지 않습니다: " + stage) + path = relative(root, pointer["artifact_path"]) + expected = project_path.parent / "stages" / stage / ("r" + str(pointer["revision"])) / "artifact.json" + require(path == expected.resolve(), "현재 결과 포인터 경로가 불일치합니다: " + stage) + data = read(path) + validate["artifact"].validate(data) + require((data["project_id"], data["stage_id"], data["revision"]) == + (project["project_id"], stage, pointer["revision"]), "결과의 제작물·단계·버전이 불일치합니다") + require(stage not in loading, "선행 참조가 순환합니다") + loading.add(stage) + ids = [ref["stage_id"] for ref in data["inputs"]] + require(len(ids) == len(set(ids)), "선행 단계 참조가 중복되었습니다") + require(NEEDED.get(stage, set()) <= set(ids), "필수 선행 참조가 빠졌습니다: " + stage) + for ref in data["inputs"]: + require(ref["revision"] == pointers[ref["stage_id"]]["revision"], "오래된 선행 버전입니다") + load(ref["stage_id"]) + loading.remove(stage) + # preview 초안의 미검증 결과는 렌더 뒤 검사한다. 승인된 결과는 완료 의미 검사를 한다. + if complete and data["schema_version"] == "1.1": + require(not semantic_errors(data, context, complete=True), "완료 선행 결과의 의미 조건이 유효하지 않습니다: " + stage) + context[stage] = data + return data + + data = read(relative(root, job["artifact"])) + stage = data["stage_id"] + require(stage in NEEDED and data["schema_version"] == "1.1", "렌더 대상은 1.1의 08 또는 10–12입니다") + require(relative(root, job["artifact"]) == relative(root, pointers[stage]["artifact_path"]), "현재 결과가 아닌 요청입니다") + artifact = load(stage, complete=False) + require(job["output_revision"] == artifact["revision"], "출력 버전과 현재 요청 버전이 다릅니다") + mode = job["mode"] + require((mode == "preview" and stage == "08-scenes") or + (mode == "production" and stage in ("10-references", "11-images", "12-motion")), "대표 렌더와 정식 제작 단계를 구분해야 합니다") + + for review_id, stages in (("topic", {"01-topic"}), ("story", {"02-research", "03-script", "04-title"}), + ("scenes", {"07-cuts", "08-scenes", "09-entities"})): + if review_id == "scenes" and mode == "preview": + continue + review = next(r for r in project["reviews"] if r["review_id"] == review_id) + require(review["decision"] == "approved" and review["feedback"].strip(), "실제 현재 사용자 승인이 필요합니다: " + review_id) + refs = {r["stage_id"]: r["revision"] for r in review["artifact_refs"]} + require(stages <= refs.keys() and len(refs) == len(review["artifact_refs"]), "승인 참조가 누락·중복되었습니다") + for approved_stage, revision in refs.items(): + require(revision == pointers[approved_stage]["revision"], "승인은 이전 버전에 대한 것입니다") + load(approved_stage) + + payload = artifact["payload"] + request = payload["preview"] if mode == "preview" else payload["request"] + require(request is not None, "대표 원본과 RenderProfile을 먼저 작성해야 합니다") + profile = request["render_profile"] + require(profile["engine"] != "BLENDER_EEVEE" or profile["seed"] == 0, + "이 EEVEE 경로는 기본 seed=0만 지원합니다. 임의 seed API가 확인되지 않았습니다") + require(profile["width"] * 16 == profile["height"] * 9 and profile["width"] % 2 == 0 + and profile["height"] % 2 == 0, "H.264 출력을 위한 짝수 크기 9:16 화면이 필요합니다") + metadata = request["scene_file"] if mode == "preview" else dict(path=job["source"], verified=False, sha256=None, blender_version=None) + if stage == "11-images": + require(job["source"] == request["scene_path"], "11 원본 요청 경로가 불일치합니다") + if payload["result"]: + metadata = payload["result"]["scene_file"] + if stage == "12-motion": + previous = context["11-images"]["payload"] + metadata = previous["result"]["scene_file"] + require(metadata["verified"] and request["render_profile"] == previous["request"]["render_profile"], "11 검증 원본·동일 프로필이 필요합니다") + require(request["scene_path"] == metadata["path"], "11 원본 참조가 다릅니다") + require(job["source"] == metadata["path"], "원본 메타데이터 경로가 다릅니다") + source, source_hash = check_source(root, metadata) + require(source.is_relative_to(project_path.parent), "다른 제작물의 원본입니다") + if mode == "preview" or stage in ("10-references", "11-images"): + require(source.is_relative_to(relative(root, job["artifact"]).parent), "대표/기준 원본은 자기 새 버전 아래에 있어야 합니다") + + cuts_data = load("07-cuts")["payload"] + cuts = {cut["id"]: cut for cut in cuts_data["cuts"]} + require(len(cuts) == len(cuts_data["cuts"]) and cuts, "실측 컷이 없거나 중복되었습니다") + previous_end = 0 + for cut in cuts.values(): + require(cut["start_ms"] == previous_end and 0 < cut["end_ms"] - cut["start_ms"] <= 5000, + "실측 컷은 겹침·빈 구간 없이 5초 이내여야 합니다") + require(all(cut[k] * profile["fps"] % 1000 == 0 for k in ("start_ms", "end_ms")), "실측 컷이 프레임 격자에 맞지 않습니다") + previous_end = cut["end_ms"] + tts = load("06-tts")["payload"] + narration = tts["result"]["narration"] + require(narration["verified"] and previous_end == cuts_data["narration_duration_ms"] <= 180000, "실측 통합 음성·컷 전체 길이가 다릅니다") + def audio_measure(media): + path = relative(root, media["path"]) + require(media["verified"] and media["kind"] == "audio" and path.is_file() + and path.is_relative_to(project_path.parent), "실제 검증 단위/통합 WAV가 필요합니다") + with wave.open(str(path), "rb") as wav: + require(Fraction(wav.getnframes() * 1000, wav.getframerate()) == media["duration_ms"], "실제 WAV 길이가 다릅니다") + require(len(wav.readframes(wav.getnframes())) == wav.getnframes() * wav.getnchannels() * wav.getsampwidth(), "WAV PCM 데이터가 잘렸습니다") + return wav.getnchannels(), wav.getsampwidth(), wav.getframerate() + audio_format = audio_measure(narration) + files, ranges = tts["result"]["files"], tts["result"]["unit_ranges"] + ids = [x["id"] for x in tts["request"]["units"]] + require(ids and len(ids) == len(set(ids)) and ids == [x["unit_id"] for x in files] == [x["unit_id"] for x in ranges], + "06 단위 요청·실제 WAV·통합 구간 참조가 다릅니다") + end_ms = 0 + for item, unit_range in zip(files, ranges): + require(audio_measure(item["media"]) == audio_format, "단위/통합 WAV 형식이 다릅니다") + require(unit_range["start_ms"] == end_ms and unit_range["end_ms"] - end_ms == item["media"]["duration_ms"] > 0, + "06 단위 실측 구간이 다릅니다") + end_ms = unit_range["end_ms"] + require(end_ms == previous_end == narration["duration_ms"], "실제 단위/통합 WAV 전체 길이가 다릅니다") + require(all(set(c["unit_ids"]) <= set(ids) for c in cuts.values()), "07 발화 단위 참조가 다릅니다") + scenes_payload = artifact["payload"] if mode == "preview" else context["08-scenes"]["payload"] + scene_cuts = [s["cut_id"] for s in scenes_payload["scenes"]] + require(len(scene_cuts) == len(cuts) and set(scene_cuts) == cuts.keys(), "08 장면의 컷이 누락·중복되었습니다") + ordered = list(cuts) + require([(c["from_cut_id"], c["to_cut_id"]) for c in scenes_payload["connections"]] == + list(zip(ordered, ordered[1:])), "08 연결의 인접 컷 참조가 다릅니다") + if "09-entities" in context: + entities = context["09-entities"]["payload"] + appearances = entities["appearances"] + require(len(appearances) == len(cuts) and {a["cut_id"] for a in appearances} == cuts.keys(), "09 컷별 등장 참조가 누락·중복되었습니다") + for kind in ("character", "background"): + ids = [item["id"] for item in entities[kind + "s"]] + require(len(ids) == len(set(ids)) and all(set(a[kind + "_ids"]) <= set(ids) for a in appearances), "09 등장 요소 ID 참조가 다릅니다") + + frame, start, end = job.get("frame"), job.get("start"), job.get("end") + require((frame is not None and start is None and end is None) or + (frame is None and start is not None and end is not None and 0 <= start < end), "이미지 한 프레임 또는 영상 [start,end) 범위를 지정해야 합니다") + require(frame is not None or (end - start) * 1000 <= 5000 * profile["fps"], "대표/컷 영상은 5초 이내입니다") + require(frame is not None or (end - start) * 1000 % profile["fps"] == 0, "영상 길이는 정수 밀리초 격자여야 합니다") + frames = [frame] if frame is not None else list(range(start, end)) + require(frames and 0 <= frames[0] and (len(frames) * 1000 <= 5000 * profile["fps"] or frame is not None), "대표/컷 영상은 5초 이내입니다") + require(frames[-1] <= previous_end * profile["fps"] // 1000, "통합 음성 밖의 프레임입니다") + if mode == "preview": + selected = job.get("cut_id", []) + require(0 < len(selected) <= 3 and len(selected) == len(set(selected)) and set(selected) <= cuts.keys(), "대표 컷 1–3개를 지정해야 합니다") + require(all(any(cuts[c]["start_ms"] * profile["fps"] // 1000 <= f < + cuts[c]["end_ms"] * profile["fps"] // 1000 + (1 if frame is not None else 0) + for c in selected) for f in frames), "대표 프레임이 지정 컷 범위 밖입니다") + elif stage == "11-images": + matches = [f for f in request["frames"] if f["id"] == job.get("frame_id")] + require(len(matches) == 1 and frame is not None and (job["scene"], frame) == + (matches[0]["scene_name"], matches[0]["frame_index"]), "11 논리 프레임·Scene 참조가 다릅니다") + cut = cuts[matches[0]["cut_id"]] + require(frame == cut[matches[0]["position"] + "_ms"] * profile["fps"] // 1000, "11 끝점과 실측 컷이 다릅니다") + require(not semantic_errors(artifact, context), "11 요청의 참조가 유효하지 않습니다") + elif stage == "12-motion": + matches = [c for c in request["clips"] if c["cut_id"] in job.get("cut_id", [])] + require(len(job.get("cut_id", [])) == len(matches) == 1 and frame is None, + "12 요청의 컷 한 개를 지정해야 합니다") + require((job["scene"], start, end) == (matches[0]["scene_name"], matches[0]["start_frame"], matches[0]["end_frame"]), "12 Scene·프레임 범위 참조가 다릅니다") + require(not semantic_errors(artifact, context), "12 요청의 참조가 유효하지 않습니다") + else: + require(frame is not None and any(x["id"] == job.get("reference_id") for x in request["items"]), "10 기준 ID와 이미지 프레임이 필요합니다") + require(not semantic_errors(artifact, context), "10 요청의 등장 요소 참조가 유효하지 않습니다") + if stage in ("11-images", "12-motion"): + endpoints = payload if stage == "11-images" else context["11-images"]["payload"] + frames_by_position = {(x["cut_id"], x["position"]): x for x in endpoints["request"]["frames"]} + for (cut_id, position), target in frames_by_position.items(): + require(target["frame_index"] == cuts[cut_id][position + "_ms"] * profile["fps"] // 1000, + "11 전체 끝점과 실측 컷이 다릅니다") + media_by_id = {x["frame_id"]: x["media"] for x in (endpoints["result"] or {}).get("frames", [])} + for connection in context["08-scenes"]["payload"]["connections"]: + left = frames_by_position[(connection["from_cut_id"], "end")] + right = frames_by_position[(connection["to_cut_id"], "start")] + if connection["mode"] == "cut": + require(left["scene_name"] != right["scene_name"], "cut 경계는 서로 다른 Scene 이름이 필요합니다") + continue + require((left["scene_name"], left["frame_index"]) == (right["scene_name"], right["frame_index"]), + "continuous 경계의 Scene/논리 프레임이 다릅니다") + if media_by_id: + require(media_by_id[left["id"]]["path"] == media_by_id[right["id"]]["path"], + "continuous 끝점은 같은 실제 PNG 경로를 공유해야 합니다") + objects = list(job.get("require_object", [])) + source_inputs, media_inputs = [], [] + if mode == "production": + for input_stage in ("08-scenes", "10-references", "11-images"): + if input_stage not in context or input_stage == stage: + continue + prior = context[input_stage]["payload"] + result = prior["preview"] if input_stage == "08-scenes" else prior["result"] + require(result and result["scene_file"]["verified"], "실제 검증 선행 .blend가 필요합니다") + input_directory = relative(root, pointers[input_stage]["artifact_path"]).parent + require(relative(root, result["scene_file"]["path"]).is_relative_to(input_directory), "이전 버전/다른 단계의 선행 .blend입니다") + check_source(root, result["scene_file"]) + source_inputs.append(result["scene_file"]) + key = {"08-scenes": "samples", "10-references": "items", "11-images": "frames"}[input_stage] + input_profile = prior["preview"]["render_profile"] if input_stage == "08-scenes" else prior["request"]["render_profile"] + for item in result[key]: + media = item["media"] + require(media["verified"] and relative(root, media["path"]).is_file(), "실제 검증 선행 PNG/MP4가 필요합니다") + require(relative(root, media["path"]).is_relative_to(input_directory), "이전 버전/다른 단계의 선행 미디어입니다") + media_inputs.append(dict(media=media, profile=input_profile)) + if stage in ("11-images", "12-motion") and "10-references" in context: + endpoint_request = payload["request"] if stage == "11-images" else context["11-images"]["payload"]["request"] + selected_frames = [x for x in endpoint_request["frames"] if + (x["id"] == job.get("frame_id") if stage == "11-images" else x["cut_id"] in job["cut_id"])] + selected_ids = {ref for frame in selected_frames for ref in frame["reference_ids"]} + refs = context["10-references"]["payload"] + model_ids = {x["id"] for x in refs["request"]["items"] if x["kind"] != "style"} + objects += [x["object_name"] for x in refs["result"]["items"] if x["reference_id"] in selected_ids & model_ids] + require(objects, "검사할 실제 오브젝트/컬렉션 이름이 필요합니다") + return dict(job, profile=profile, frames=frames, source_hash=source_hash, + metadata=metadata, require_object=sorted(set(objects)), + source_inputs=source_inputs, media_inputs=media_inputs) + + +def identity(job, source_hash): + value = dict(source=source_hash, profile=job["profile"], scene=job["scene"], frames=job["frames"], + renderer=digest(Path(__file__).with_name("inside.py"))) + return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() + + +def execute(command, logfile=None): + result = subprocess.run([str(x) for x in command], capture_output=True, text=True, encoding="utf-8", errors="replace") + if logfile: + logfile.write_text(result.stdout + result.stderr, encoding="utf-8") + require(result.returncode == 0, "실행 실패 (종료코드 " + str(result.returncode) + "): " + (result.stderr or result.stdout)[-1500:]) + return result.stdout + + +def measure(path, profile, count, ffmpeg, ffprobe): + value = json.loads(execute([ffprobe, "-v", "error", "-count_frames", "-show_streams", "-show_format", "-of", "json", path])) + videos = [s for s in value["streams"] if s["codec_type"] == "video"] + require(len(videos) == 1 and len(value["streams"]) == 1, "무음 영상 트랙 하나가 필요합니다") + stream = videos[0] + require((stream["width"], stream["height"]) == (profile["width"], profile["height"]), "실제 출력 해상도가 다릅니다") + kind = "image" if path.suffix == ".png" else "video" + require(stream["codec_name"] == ("png" if kind == "image" else "h264"), "실제 출력 코덱이 다릅니다") + require(int(stream["nb_read_frames"]) == count, "실제 디코드 프레임 수가 다릅니다") + duration = None + if kind == "video": + require(Fraction(stream["avg_frame_rate"]) == profile["fps"], "실제 영상 fps가 다릅니다") + expected = Fraction(count * 1000, profile["fps"]) + require(expected.denominator == 1 and abs(Fraction(stream["duration"]) * 1000 - expected) <= 1, + "실제 영상 길이가 다릅니다") + duration = int(expected) + execute([ffmpeg, "-v", "error", "-xerror", "-i", path, "-map", "0:v:0", "-c:v", "rawvideo", "-f", "null", "-"]) + return dict(path=path.relative_to(ROOT).as_posix(), kind=kind, verified=True, + duration_ms=duration, width=stream["width"], height=stream["height"]) + + +def blender_call(blender, source, job, folder, operation): + jobpath = folder / (operation + "-job.json") + resultpath = folder / (operation + "-result.json") + value = dict(job, operation=operation, result=str(resultpath), directory=str(folder)) + jobpath.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8") + execute([blender, "--background", "--factory-startup", "--disable-autoexec", source, + "--python-exit-code", "1", "--python", Path(__file__).with_name("inside.py"), "--", jobpath], folder / (operation + ".log")) + require(resultpath.is_file(), "Blender 검사 결과가 없습니다") + return read(resultpath) + + +def render(root, job, blender, ffmpeg, ffprobe): + job = prepare(root, job) + source = relative(root, job["source"]) + artifact = relative(root, job["artifact"]) + key = identity(job, job["source_hash"]) + folder = artifact.parent / "render-attempts" / (key[:16] + "-" + uuid.uuid4().hex[:12]) + folder.mkdir(parents=True, exist_ok=False) + try: + for index, metadata in enumerate(job["source_inputs"]): + prior_source, prior_hash = check_source(root, metadata) + prior_folder = folder / ("input-source-" + str(index)) + prior_folder.mkdir() + prior_report = blender_call(blender, prior_source, + dict(scene=None, require_object=[]), prior_folder, "inspect") + require(prior_report["blender_version"] == metadata["blender_version"] + and digest(prior_source) == prior_hash, "선행 .blend 버전·해시가 다릅니다") + for item in job["media_inputs"]: + media, profile = item["media"], item["profile"] + count = 1 if media["kind"] == "image" else Fraction(media["duration_ms"] * profile["fps"], 1000) + require(isinstance(count, int) or count.denominator == 1, "선행 영상이 프레임 격자와 다릅니다") + actual = measure(relative(root, media["path"]), profile, int(count), ffmpeg, ffprobe) + require(actual == media, "선행 미디어의 실제 메타데이터가 기록과 다릅니다") + report = blender_call(blender, source, job, folder, "inspect") + if job["metadata"]["verified"]: + require(report["blender_version"] == job["metadata"]["blender_version"], "검증 원본의 Blender 버전이 다릅니다") + require(digest(source) == job["source_hash"], "검사 중 원본이 바뀌었습니다") + rendered = blender_call(blender, source, job, folder, "render") + require(rendered["render_profile"] == job["profile"], "실제 적용 프로필이 요청과 다릅니다") + require(digest(source) == job["source_hash"], "렌더 중 원본이 바뀌었습니다") + # 같은 시각을 다른 Scene으로 평가할 때도 원본·Scene·프로필·프레임이 서로 분리된다. + paths = [folder / ("frame-" + str(f).zfill(8) + ".png") for f in job["frames"]] + pngs = [measure(path, job["profile"], 1, ffmpeg, ffprobe) for path in paths] + if job.get("frame") is None: + output = folder / "clip.mp4" + execute([ffmpeg, "-v", "error", "-n", "-framerate", job["profile"]["fps"], "-start_number", job["start"], + "-i", folder / "frame-%08d.png", "-frames:v", len(paths), "-an", "-c:v", "libx264", "-crf", "18", "-pix_fmt", "yuv420p", "-movflags", "+faststart", output], folder / "encode.log") + media = measure(output, job["profile"], len(paths), ffmpeg, ffprobe) + else: + media = pngs[0] + evidence = dict(scene_file=dict(path=job["source"], verified=True, sha256=job["source_hash"], blender_version=report["blender_version"]), + render_profile=rendered["render_profile"], scene_name=job["scene"], logical_frames=job["frames"], + source_inspection=report, render_inspection=rendered, + media=media, identity=key, visual_review="not_checked") + (folder / "evidence.json").write_text(json.dumps(evidence, ensure_ascii=False, indent=2), encoding="utf-8") + return evidence + except Exception as error: + (folder / "failure.json").write_text(json.dumps(dict(error=str(error), identity=key), ensure_ascii=False), encoding="utf-8") + raise + + +def main(): + parser = argparse.ArgumentParser(description="실제 Blender 원본 검사와 제한 렌더. 제작 상태는 변경하지 않습니다.") + parser.add_argument("operation", choices=("inspect", "render")) + parser.add_argument("--blender", required=True) + parser.add_argument("--source", required=True) + parser.add_argument("--scene", required=True) + parser.add_argument("--require-object", action="append", default=[]) + parser.add_argument("--mode", choices=("preview", "production")) + parser.add_argument("--project") + parser.add_argument("--artifact") + parser.add_argument("--output-revision", type=int) + parser.add_argument("--frame", type=int) + parser.add_argument("--start", type=int) + parser.add_argument("--end", type=int) + parser.add_argument("--cut-id", action="append", default=[]) + parser.add_argument("--frame-id") + parser.add_argument("--reference-id") + parser.add_argument("--ffmpeg") + parser.add_argument("--ffprobe") + args = parser.parse_args() + job = vars(args) + try: + if args.operation == "inspect": + source, source_hash = check_source(ROOT, dict(path=args.source, verified=False)) + folder = ROOT / "tools/blender/.cache" / ("inspect-" + uuid.uuid4().hex) + folder.mkdir(parents=True) + result = blender_call(args.blender, source, job, folder, "inspect") + require(digest(source) == source_hash, "검사 중 원본이 바뀌었습니다") + result["scene_file"] = dict(path=args.source, verified=True, sha256=source_hash, blender_version=result["blender_version"]) + else: + require(all((args.mode, args.project, args.artifact, args.output_revision, args.ffmpeg, args.ffprobe)), "렌더의 내부 실행 인자가 빠졌습니다") + result = render(ROOT, job, args.blender, args.ffmpeg, args.ffprobe) + print(json.dumps(result, ensure_ascii=False, indent=2)) + except Exception as error: + print("검사/렌더 실패: " + str(error), file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/blender/smoke.py b/tools/blender/smoke.py new file mode 100644 index 0000000..32bf4cb --- /dev/null +++ b/tools/blender/smoke.py @@ -0,0 +1,102 @@ +"""작은 3D 영상과 같은 논리 프레임의 서로 다른 Scene을 실제로 확인한다.""" +import argparse +import json +from pathlib import Path +import shutil + +import run +from test_runner import fixture, production_fixture, endpoint_fixture, ROOT +from test_runner import write + + +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") + 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"]) + 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"] == 1000, "검사 영상 길이 불일치") + attempt = (ROOT / video["media"]["path"]).parent + require(run.digest(attempt / "frame-00000000.png") != run.digest(attempt / "frame-00000009.png"), "3D 움직임이 없습니다") + still_job = dict(job, frame=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"), args.blender, args.ffmpeg, args.ffprobe) + require(first["identity"] != second["identity"] and run.digest(ROOT / first["media"]["path"]) != run.digest(ROOT / second["media"]["path"]), + "같은 논리 프레임의 다른 Scene 카메라가 혼입되었습니다") + 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() diff --git a/tools/blender/smoke_fixture.py b/tools/blender/smoke_fixture.py new file mode 100644 index 0000000..efc7fca --- /dev/null +++ b/tools/blender/smoke_fixture.py @@ -0,0 +1,49 @@ +"""실행 검사 전용 3D 구와 두 카메라를 만든다. 제작 내용과 무관하다.""" +import sys + +import bpy +from mathutils import Vector + +bpy.ops.object.select_all(action="SELECT") +bpy.ops.object.delete(use_global=False) +scene = bpy.context.scene +scene.name = "SceneA" +scene.frame_start, scene.frame_end = 1, 21 +scene.world.color = (0.1, 0.1, 0.1) +bpy.ops.mesh.primitive_uv_sphere_add(segments=24, ring_count=16, radius=0.65, location=(-1, 0, 0)) +ball = bpy.context.object +ball.name = "Ball" +ball.keyframe_insert("location", frame=1) +ball.location.x = 1 +ball.keyframe_insert("location", frame=11) +material = bpy.data.materials.new("Blue") +material.diffuse_color = (0.03, 0.3, 0.8, 1) +ball.data.materials.append(material) +bpy.ops.object.light_add(type="AREA", location=(-3, -4, 6)) +light = bpy.context.object +light.data.energy = 900 +light.data.shape = "DISK" +light.data.size = 5 +light.rotation_euler = (Vector((0, 0, 0)) - light.location).to_track_quat("-Z", "Y").to_euler() +bpy.ops.object.camera_add(location=(0, -7, 4)) +camera = bpy.context.object +camera.rotation_euler = (-camera.location).to_track_quat("-Z", "Y").to_euler() +camera.data.type = "ORTHO" +camera.data.ortho_scale = 5 +scene.camera = camera +second = bpy.data.scenes.new("SceneB") +second.frame_start, second.frame_end = 1, 21 +second.world = scene.world +second.collection.objects.link(ball) +second.collection.objects.link(light) +other = camera.copy() +other.data = camera.data.copy() +other.location = (5, -7, 1) +other.rotation_euler = (-other.location).to_track_quat("-Z", "Y").to_euler() +second.collection.objects.link(other) +second.camera = other +if "--external-image" in sys.argv: + bpy.data.images.load(sys.argv[sys.argv.index("--external-image") + 1]).use_fake_user = True +if "--short-scene" in sys.argv: + scene.frame_end = second.frame_end = 1 +bpy.ops.wm.save_as_mainfile(filepath=sys.argv[sys.argv.index("--") + 1]) diff --git a/tools/blender/test_runner.py b/tools/blender/test_runner.py new file mode 100644 index 0000000..fafef13 --- /dev/null +++ b/tools/blender/test_runner.py @@ -0,0 +1,375 @@ +"""승인 경계·시간·원본 참조 회귀. 생성한 시도는 진단을 위해 남긴다.""" +from copy import deepcopy +import json +from pathlib import Path +import unittest +import uuid +import wave +import zlib +import struct +import shutil + +import run + + +ROOT = Path(__file__).resolve().parents[2] +WATER = ROOT / "projects/water-hydrogen-bond-surface-tension" +PROFILE = dict(engine="CYCLES", device="CPU", fps=10, width=90, height=160, samples=4, seed=0) + + +def write(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8") + + +def fixture(): + folder = ROOT / "tools/blender/.cache" / uuid.uuid4().hex / WATER.name + project = json.loads((WATER / "project.json").read_text(encoding="utf-8")) + project["schema_version"] = "1.1" + project["settings"].pop("image_model", None) + project["settings"]["visual_engine"] = "Blender" + artifacts = {} + for stage in project["stages"]: + if stage["revision"] is None: + continue + data = json.loads((ROOT / stage["artifact_path"]).read_text(encoding="utf-8")) + if stage["stage_id"] in ("08-scenes", "09-entities"): + data.update(schema_version="1.1", revision=2) + stage.update(status="draft", revision=2) + stage["artifact_path"] = (folder / "stages" / stage["stage_id"] / ("r" + str(stage["revision"])) / "artifact.json").relative_to(ROOT).as_posix() + artifacts[stage["stage_id"]] = data + cuts = artifacts["07-cuts"]["payload"] + cuts["narration_duration_ms"] = 2000 + cuts["cuts"] = [dict(id="c1", start_ms=0, end_ms=1000, unit_ids=["u1"], narration_text="첫 장면"), + dict(id="c2", start_ms=1000, end_ms=2000, unit_ids=["u1"], narration_text="다음 장면")] + cuts["captions"] = [] + source = folder / "stages/08-scenes/r2/sample.blend" + source.parent.mkdir(parents=True, exist_ok=True) + source.write_bytes("검사 전 원본".encode("utf-8")) + narration = folder / "narration.wav" + with wave.open(str(narration), "wb") as audio: + audio.setparams((1, 2, 24000, 0, "NONE", "not compressed")) + audio.writeframes(b"\0\0" * 48000) + artifacts["06-tts"]["payload"]["result"]["narration"].update( + path=narration.relative_to(ROOT).as_posix(), duration_ms=2000) + tts = artifacts["06-tts"]["payload"] + ids = [unit["id"] for unit in tts["request"]["units"]] + tts["result"]["files"] = [] + tts["result"]["unit_ranges"] = [] + for index, unit_id in enumerate(ids): + path = folder / (unit_id + ".wav") + with wave.open(str(path), "wb") as audio: + audio.setparams((1, 2, 24000, 0, "NONE", "not compressed")) + audio.writeframes(b"\0\0" * 6000) + tts["result"]["files"].append(dict(unit_id=unit_id, media=dict(path=path.relative_to(ROOT).as_posix(), kind="audio", verified=True, + duration_ms=250, width=None, height=None))) + tts["result"]["unit_ranges"].append(dict(unit_id=unit_id, start_ms=index * 250, end_ms=(index + 1) * 250)) + for cut in cuts["cuts"]: + cut["unit_ids"] = ids[:4] if cut["id"] == "c1" else ids[4:] + scenes = artifacts["08-scenes"] + scenes["payload"]["scenes"] = [dict(cut_id=cut["id"], description="움직이는 구", start_state="왼쪽", end_state="오른쪽", camera="고정") for cut in cuts["cuts"]] + scenes["payload"]["connections"] = [dict(from_cut_id="c1", to_cut_id="c2", mode="cut", reason="시점 변경")] + scenes["payload"]["preview"] = dict(scene_file=dict(path=source.relative_to(ROOT).as_posix(), verified=False, sha256=None, blender_version=None), render_profile=deepcopy(PROFILE), samples=[]) + for stage in project["stages"]: + if stage["stage_id"] in artifacts: + write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]]) + write(folder / "project.json", project) + job = dict(mode="preview", project=(folder / "project.json").relative_to(ROOT).as_posix(), + artifact=next(s["artifact_path"] for s in project["stages"] if s["stage_id"] == "08-scenes"), + output_revision=2, source=source.relative_to(ROOT).as_posix(), scene="SceneA", + frame=None, start=0, end=10, cut_id=["c1"], frame_id=None, reference_id=None, + require_object=["Ball"]) + return folder, project, artifacts, job + + +def production_fixture(): + folder, project, artifacts, job = fixture() + for stage in project["stages"]: + if stage["stage_id"] in ("08-scenes", "09-entities"): + stage["status"] = "complete" + artifacts["09-entities"]["inputs"][1]["revision"] = 2 + artifacts["09-entities"]["payload"]["appearances"] = [dict(cut_id=c, character_ids=[], background_ids=[]) for c in ("c1", "c2")] + preview = artifacts["08-scenes"]["payload"]["preview"] + preview["scene_file"].update(verified=True, sha256=run.digest(ROOT / job["source"]), blender_version="5.2.2 LTS") + preview["samples"] = [dict(id="sample", cut_ids=["c1"], media=dict(path=job["source"].replace(".blend", ".png"), + kind="image", verified=True, duration_ms=None, width=90, height=160))] + def chunk(kind, body): + return struct.pack(">I", len(body)) + kind + body + struct.pack(">I", zlib.crc32(kind + body)) + png = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", 90, 160, 8, 2, 0, 0, 0)) + png += chunk(b"IDAT", zlib.compress((b"\0" + b"\0" * 270) * 160)) + chunk(b"IEND", b"") + (ROOT / preview["samples"][0]["media"]["path"]).write_bytes(png) + review = next(r for r in project["reviews"] if r["review_id"] == "scenes") + review["artifact_refs"] = [dict(stage_id=s, revision=1 if s == "07-cuts" else 2) for s in ("07-cuts", "08-scenes", "09-entities")] + review["feedback"] = "회귀 검사 전용 승인 기록이며 실제 영상의 사용자 승인이 아닙니다" + reference = json.loads((ROOT / "templates/stages/10-references.json").read_text(encoding="utf-8")) + reference.update(project_id=project["project_id"], revision=2, + inputs=[dict(stage_id=s, revision=1 if s == "02-research" else 2) for s in ("02-research", "08-scenes", "09-entities")]) + reference["payload"]["request"]["render_profile"] = deepcopy(PROFILE) + reference["payload"]["request"]["items"] = [dict(id="ball", kind="object", entity_id=None, prompt="검사용 구")] + pointer = next(s for s in project["stages"] if s["stage_id"] == "10-references") + pointer.update(status="draft", revision=2, + artifact_path=(folder / "stages/10-references/r2/artifact.json").relative_to(ROOT).as_posix()) + artifacts["10-references"] = reference + source = folder / "stages/10-references/r2/model.blend" + source.parent.mkdir(parents=True, exist_ok=True) + source.write_bytes(b"uninspected fixture") + job.update(mode="production", artifact=pointer["artifact_path"], source=source.relative_to(ROOT).as_posix(), + frame=0, start=None, end=None, reference_id="ball", cut_id=[]) + for stage in project["stages"]: + if stage["stage_id"] in artifacts: + write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]]) + write(folder / "project.json", project) + return folder, project, artifacts, job + + +def endpoint_fixture(motion=False): + folder, project, artifacts, job = production_fixture() + reference = artifacts["10-references"] + reference["payload"]["result"] = dict(scene_file=dict(path=job["source"], verified=True, + sha256=run.digest(ROOT / job["source"]), blender_version="5.2.2 LTS"), + items=[dict(reference_id="ball", object_name="Ball", media=deepcopy(artifacts["08-scenes"]["payload"]["preview"]["samples"][0]["media"]))]) + reference_media = reference["payload"]["result"]["items"][0]["media"] + image = folder / "stages/10-references/r2/reference.png" + shutil.copyfile(ROOT / reference_media["path"], image) + reference_media["path"] = image.relative_to(ROOT).as_posix() + frames = [dict(id=cut + "-" + position, cut_id=cut, position=position, prompt="상태 설명", reference_ids=["ball"], + scene_name="SceneA" if cut == "c1" else "SceneB", frame_index=index) + for cut, start in (("c1", 0), ("c2", 10)) for position, index in (("start", start), ("end", start + 10))] + source = folder / "stages/11-images/r2/animation.blend" + source.parent.mkdir(parents=True, exist_ok=True) + source.write_bytes(b"uninspected endpoint fixture") + images = json.loads((ROOT / "templates/stages/11-images.json").read_text(encoding="utf-8")) + images.update(project_id=project["project_id"], revision=2, + inputs=[dict(stage_id=s, revision=1 if s in ("02-research", "03-script", "07-cuts") else 2) + for s in ("02-research", "03-script", "07-cuts", "08-scenes", "09-entities", "10-references")]) + images["payload"]["request"].update(frames=frames, scene_path=source.relative_to(ROOT).as_posix(), render_profile=deepcopy(PROFILE)) + artifacts["11-images"] = images + for stage in project["stages"]: + if stage["stage_id"] in ("10-references", "11-images"): + stage.update(revision=2, status="complete" if stage["stage_id"] == "10-references" or motion else "draft", + artifact_path=(folder / "stages" / stage["stage_id"] / "r2/artifact.json").relative_to(ROOT).as_posix()) + job.update(artifact=(folder / "stages/11-images/r2/artifact.json").relative_to(ROOT).as_posix(), + source=source.relative_to(ROOT).as_posix(), frame_id="c1-start", reference_id=None) + if motion: + images["payload"]["result"] = dict(scene_file=dict(path=job["source"], verified=True, + sha256=run.digest(source), blender_version="5.2.2 LTS"), + frames=[dict(frame_id=f["id"], media=deepcopy(reference["payload"]["result"]["items"][0]["media"])) for f in frames]) + for item in images["payload"]["result"]["frames"]: + image = source.parent / (item["frame_id"] + ".png") + shutil.copyfile(ROOT / item["media"]["path"], image) + item["media"]["path"] = image.relative_to(ROOT).as_posix() + clips = [dict(cut_id=c, start_frame_id=c + "-start", end_frame_id=c + "-end", duration_ms=1000, + motion_prompt="구의 움직임", scene_name="SceneA" if c == "c1" else "SceneB", start_frame=start, end_frame=start + 10) + for c, start in (("c1", 0), ("c2", 10))] + artifact = json.loads((ROOT / "templates/stages/12-motion.json").read_text(encoding="utf-8")) + artifact.update(project_id=project["project_id"], revision=2, + inputs=[dict(stage_id=s, revision=1 if s == "07-cuts" else 2) for s in ("07-cuts", "08-scenes", "11-images")]) + artifact["payload"]["request"].update(clips=clips, scene_path=job["source"], render_profile=deepcopy(PROFILE)) + artifacts["12-motion"] = artifact + pointer = next(s for s in project["stages"] if s["stage_id"] == "12-motion") + pointer.update(revision=2, status="draft", artifact_path=(folder / "stages/12-motion/r2/artifact.json").relative_to(ROOT).as_posix()) + job.update(artifact=pointer["artifact_path"], frame=None, start=0, end=10, cut_id=["c1"], frame_id=None) + for stage in project["stages"]: + if stage["stage_id"] in artifacts: + write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]]) + write(folder / "project.json", project) + return folder, project, artifacts, job + + +class RequestTests(unittest.TestCase): + def setUp(self): + self.folder, self.project, self.artifacts, self.job = fixture() + + def persist(self): + write(self.folder / "project.json", self.project) + for stage in self.project["stages"]: + if stage["stage_id"] in self.artifacts: + write(ROOT / stage["artifact_path"], self.artifacts[stage["stage_id"]]) + + def test_preview_requires_no_new_scenes_approval(self): + validated = run.prepare(ROOT, self.job) + self.assertEqual(validated["profile"], PROFILE) + self.assertEqual(validated["frames"], list(range(10))) + + def test_production_rejects_preview_and_old_scenes(self): + self.job["mode"] = "production" + with self.assertRaises(ValueError): + run.prepare(ROOT, self.job) + folder, project, artifacts, job = production_fixture() + self.assertEqual(run.prepare(ROOT, job)["frames"], [0]) + review = next(r for r in project["reviews"] if r["review_id"] == "scenes") + for decision in ("pending", "changes_requested", "stale", "old_approved", "no_feedback"): + review.update(decision="approved", feedback="회귀 검사 기록") + review["artifact_refs"][1]["revision"] = 2 + if decision == "old_approved": + review["artifact_refs"][1]["revision"] = 1 + elif decision == "no_feedback": + review["feedback"] = "" + else: + review["decision"] = decision + write(folder / "project.json", project) + with self.subTest(decision=decision), self.assertRaises(ValueError): + run.prepare(ROOT, job) + + def test_bad_pointers_inputs_and_revision(self): + for change in ("revision", "pointer", "stale", "input"): + folder, project, artifacts, job = fixture() + if change == "revision": + job["output_revision"] = 1 + elif change == "pointer": + project["stages"][7]["revision"] = 1 + elif change == "stale": + project["stages"][2]["status"] = "stale" + else: + artifacts["08-scenes"]["inputs"][0]["revision"] = 99 + write(ROOT / job["artifact"], artifacts["08-scenes"]) + write(folder / "project.json", project) + with self.subTest(change=change), self.assertRaises(ValueError): + run.prepare(ROOT, job) + + def test_ranges_aspect_grid_and_foreign_source(self): + for change in ("long", "negative", "unknown_cut", "foreign", "aspect", "grid"): + folder, project, artifacts, job = fixture() + if change == "long": + job["end"] = 51 + elif change == "negative": + job["start"] = -1 + elif change == "unknown_cut": + job["cut_id"] = ["foreign"] + elif change == "foreign": + job["source"] = "../foreign.blend" + elif change == "aspect": + artifacts["08-scenes"]["payload"]["preview"]["render_profile"]["width"] = 100 + else: + artifacts["07-cuts"]["payload"]["cuts"][0]["end_ms"] = 999 + for stage in project["stages"]: + if stage["stage_id"] in artifacts: + write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]]) + with self.subTest(change=change), self.assertRaises(ValueError): + run.prepare(ROOT, job) + + def test_source_hash_changed_is_rejected(self): + metadata = dict(path=self.job["source"], verified=True, sha256="0" * 64, blender_version="5.2.2 LTS") + with self.assertRaises(ValueError): + run.check_source(ROOT, metadata) + + def test_eevee_does_not_claim_an_unsupported_seed(self): + profile = self.artifacts["08-scenes"]["payload"]["preview"]["render_profile"] + profile.update(engine="BLENDER_EEVEE", device="GPU", seed=5) + self.persist() + with self.assertRaises(ValueError): + run.prepare(ROOT, self.job) + + def test_production_missing_or_foreign_preview_is_rejected(self): + for change in ("missing", "foreign", "hash"): + folder, project, artifacts, job = production_fixture() + preview = artifacts["08-scenes"]["payload"]["preview"] + if change == "missing": + preview["samples"][0]["media"]["path"] += ".missing" + elif change == "foreign": + preview["scene_file"]["path"] = self.job["source"] + preview["scene_file"]["sha256"] = run.digest(ROOT / self.job["source"]) + else: + preview["scene_file"]["sha256"] = "a" * 64 + write(folder / "stages/08-scenes/r2/artifact.json", artifacts["08-scenes"]) + with self.subTest(change=change), self.assertRaises(ValueError): + run.prepare(ROOT, job) + + def test_attempt_identity_includes_scene_hash_profile_and_range(self): + job = run.prepare(ROOT, self.job) + key = run.identity(job, "a" * 64) + for name, value in (("scene", "SceneB"), ("frames", [10]), ("profile", dict(PROFILE, seed=4))): + other = dict(job, **{name: value}) + self.assertNotEqual(key, run.identity(other, "a" * 64)) + self.assertNotEqual(key, run.identity(job, "b" * 64)) + + def test_endpoint_and_motion_requests_match_actual_cuts(self): + for motion in (False, True): + folder, project, artifacts, job = endpoint_fixture(motion) + self.assertEqual(len(run.prepare(ROOT, job)["frames"]), 10 if motion else 1) + for change in ("scene", "endpoint", "reference", "profile", "continuous", "cut_scene"): + folder, project, artifacts, job = endpoint_fixture(motion) + stage = "12-motion" if motion else "11-images" + request = artifacts[stage]["payload"]["request"] + if change == "scene": + job["scene"] = "ForeignScene" + elif change == "endpoint": + artifacts["11-images"]["payload"]["request"]["frames"][3]["frame_index"] = 19 + elif change == "reference": + artifacts["11-images"]["payload"]["request"]["frames"][0]["reference_ids"] = ["missing"] + elif change == "profile": + request["render_profile"]["fps"] = 20 + elif change == "continuous": + artifacts["08-scenes"]["payload"]["connections"][0]["mode"] = "continuous" + else: + for target in artifacts["11-images"]["payload"]["request"]["frames"]: + target["scene_name"] = "SceneA" + if motion: + artifacts["12-motion"]["payload"]["request"]["clips"][1]["scene_name"] = "SceneA" + for pointer in project["stages"]: + if pointer["stage_id"] in artifacts: + write(ROOT / pointer["artifact_path"], artifacts[pointer["stage_id"]]) + with self.subTest(motion=motion, change=change), self.assertRaises(ValueError): + run.prepare(ROOT, job) + + def test_current_10_entity_reference_must_exist(self): + folder, project, artifacts, job = production_fixture() + artifacts["10-references"]["payload"]["request"]["items"][0].update(kind="character", entity_id="ABSENT") + write(ROOT / job["artifact"], artifacts["10-references"]) + with self.assertRaises(ValueError): + run.prepare(ROOT, job) + + def test_new_renderer_rejects_legacy_project_configuration(self): + self.project.update(schema_version="1.0") + self.project["settings"].pop("visual_engine") + self.project["settings"]["image_model"] = "gpt-image-2.5-sunburst" + self.persist() + with self.assertRaises(ValueError): + run.prepare(ROOT, self.job) + + def test_complete_06_unit_files_need_actual_metadata(self): + for change in ("missing", "unverified", "duration", "range", "truncated", "id"): + folder, project, artifacts, job = fixture() + tts = artifacts["06-tts"]["payload"] + media = tts["result"]["files"][0]["media"] + if change == "missing": + media["path"] += ".missing" + elif change == "unverified": + media["verified"] = False + elif change == "duration": + media["duration_ms"] = 251 + elif change == "range": + tts["result"]["unit_ranges"][0]["end_ms"] = 251 + elif change == "id": + tts["result"]["files"][0]["unit_id"] = "foreign" + else: + path = ROOT / media["path"] + path.write_bytes(path.read_bytes()[:-2]) + write(folder / "stages/06-tts/r1/artifact.json", artifacts["06-tts"]) + with self.subTest(change=change), self.assertRaises(ValueError): + run.prepare(ROOT, job) + + def test_old_visual_revision_source_and_media_are_rejected(self): + for input_stage in ("08-scenes", "10-references", "11-images"): + for change in ("source", "media"): + folder, project, artifacts, job = endpoint_fixture(motion=True) + payload = artifacts[input_stage]["payload"] + result = payload["preview"] if input_stage == "08-scenes" else payload["result"] + key = {"08-scenes": "samples", "10-references": "items", "11-images": "frames"}[input_stage] + metadata = result["scene_file"] if change == "source" else result[key][0]["media"] + old = ROOT / metadata["path"].replace("/r2/", "/r1/") + old.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(ROOT / metadata["path"], old) + metadata["path"] = old.relative_to(ROOT).as_posix() + if input_stage == "11-images" and change == "source": + artifacts["12-motion"]["payload"]["request"]["scene_path"] = metadata["path"] + job["source"] = metadata["path"] + for pointer in project["stages"]: + if pointer["stage_id"] in artifacts: + write(ROOT / pointer["artifact_path"], artifacts[pointer["stage_id"]]) + with self.subTest(stage=input_stage, change=change), self.assertRaises(ValueError): + run.prepare(ROOT, job) + + +if __name__ == "__main__": + unittest.main()