gig-line/Tools/review/e2e_asset_check.py

"""End-to-end asset check: Blender render -> Unreal build -> captures -> judge -> README.

    python3 Tools/review/e2e_asset_check.py                # everything
    python3 Tools/review/e2e_asset_check.py --skip-blender # reuse the existing Blender renders
    python3 Tools/review/e2e_asset_check.py --skip-judge   # no API calls, just the images

The chain, in order, each step failing loudly with its log under /tmp/gigline/:

1. **Blender renders** - `Art/blender/*/` assets rendered by `Tools/blender/blender_screenshots.py`, so the
   source the imports came from has an image of its own.
2. **Empty maps** - `Tools/unreal/ue_recreate_maps.py` rebuilds both maps, so a run never inherits an actor
   from an experiment. The builders also clear the level they load; this is the belt-and-braces step.
3. **Unreal build** - `Tools/unreal/ue_build_demo_map.py` (game map) and `Tools/unreal/ue_build_review_map.py`
   (staging map for individual assets).
4. **Captures** - `Tools/review/render_screenshots.sh`, which renders both maps into `Screenshots/`.
5. **Judge** - every shot in `Tools/review/visual_targets.json` judged by **two models** through
   `Tools/review/judge_screenshot.py --shots`, which reconciles agreement and reports disagreement.
6. **Review file** - `Tools/review/build_comparison_md.py` rewrites the Blender-vs-UE block in
   `Screenshots/README.md`, then prettier formats it.

The report lands in `/tmp/gigline/e2e-report.txt`; the exit code is non-zero if any step failed.
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
import time

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
ENGINE = os.environ.get("ENGINE", "/run/media/system/DATA/Engines/UE_5.8.4")
UPROJECT = os.path.join(ROOT, "Game", "GigLine", "GigLine.uproject")
BLENDER = os.environ.get(
    "BLENDER_FILES", "/var/lib/flatpak/app/org.blender.Blender/x86_64/stable/active/files")
UNREAL = os.path.join(ENGINE, "Engine", "Binaries", "Linux", "UnrealEditor-Cmd")
REPORT = "/tmp/gigline/e2e-report.txt"


def run(step, cmd, env=None, timeout=None, result_file=None, expect=None):
    """Run one step and record it. A step fails on a non-zero exit or a missing marker.

    The child's output streams to its own file under /tmp/gigline/e2e/, not into a pipe: a captured pipe
    hides progress for a fifteen-minute Unreal build and deadlocks if the engine spawns something that
    outlives it.
    """
    started = time.time()
    print(f"== {step}", flush=True)
    # A result file left by an earlier run would make this step look successful without doing anything.
    if result_file and os.path.exists(result_file):
        os.remove(result_file)
    safe = step.replace("/", "-").replace(" ", "_")
    os.makedirs("/tmp/gigline/e2e", exist_ok=True)
    log_path = f"/tmp/gigline/e2e/{safe}.log"
    with open(log_path, "w") as out:
        try:
            proc = subprocess.run(cmd, cwd=ROOT, env={**os.environ, **(env or {})},
                                  stdout=out, stderr=subprocess.STDOUT, timeout=timeout)
        except subprocess.TimeoutExpired:
            return step, False, f"timed out after {timeout}s (log {log_path})"
    elapsed = time.time() - started
    if proc.returncode != 0:
        tail = ""
        try:
            tail = " / ".join(open(log_path, errors="replace").read().strip().splitlines()[-3:])
        except OSError:
            pass
        return step, False, f"exit {proc.returncode} ({elapsed:.0f}s): {tail}"
    if result_file and expect:
        if not os.path.exists(result_file):
            return step, False, f"no result file {result_file}"
        text = open(result_file, errors="replace").read()
        if expect not in text:
            return step, False, f"{os.path.basename(result_file)} lacks {expect!r}"
    return step, True, f"{elapsed:.0f}s"


def unreal_script(script, timeout=1800):
    return [UNREAL, UPROJECT, "-run=pythonscript", f"-script={os.path.join(ROOT, script)}",
            "-nullrhi", "-unattended", "-nopause", "-stdout"]


def main():
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--skip-blender", action="store_true", help="reuse the Blender renders")
    parser.add_argument("--skip-judge", action="store_true", help="no API calls")
    parser.add_argument("--skip-recreate", action="store_true",
                        help="keep the existing maps (the builders still clear their levels)")
    args = parser.parse_args()

    # A second editor fighting the first is what makes an Unreal commandlet hang instead of fail.
    import shutil
    if shutil.which("pgrep"):
        others = subprocess.run(["pgrep", "-f", "UnrealEditor-Cmd"], capture_output=True, text=True)
        if others.returncode == 0 and others.stdout.strip():
            sys.exit("error: an UnrealEditor process is already running; close it before the check "
                     "(two editors on one project hang, they do not fail)")

    steps = []
    if not args.skip_blender:
        steps.append(run("1/6 blender renders",
                         [os.path.join(BLENDER, "blender", "blender"), "-b", "--factory-startup",
                          "--python", os.path.join(ROOT, "Tools", "blender", "blender_screenshots.py"),
                          "--", "--mode", "both", "--out", os.path.join(ROOT, "Screenshots", "blender")],
                         env={"LD_LIBRARY_PATH": os.path.join(BLENDER, "blender", "lib")},
                         timeout=1800))
    else:
        steps.append(("1/6 blender renders", True, "skipped, reusing Screenshots/blender"))

    if not args.skip_recreate:
        steps.append(run("2/6 empty maps", unreal_script("Tools/unreal/ue_recreate_maps.py"),
                         timeout=900, result_file="/tmp/gigline/recreate-maps.txt",
                         expect="VERDICT: PASS"))
    else:
        steps.append(("2/6 empty maps", True, "skipped"))

    steps.append(run("3/6 unreal build: demo map", unreal_script("Tools/unreal/ue_build_demo_map.py"),
                     timeout=1800, result_file="/tmp/gigline/demo-map.txt", expect="save_map"))
    steps.append(run("3/6 unreal build: review map", unreal_script("Tools/unreal/ue_build_review_map.py"),
                     timeout=1800, result_file="/tmp/ue-review-map-result.txt", expect="save_map"))

    steps.append(run("4/6 captures", ["bash", os.path.join(ROOT, "Tools", "review", "render_screenshots.sh")],
                     timeout=2400))

    if not args.skip_judge:
        steps.append(run("5/6 judge (two models)",
                         [sys.executable, os.path.join(ROOT, "Tools", "review", "judge_screenshot.py"), "--shots"],
                         timeout=1800))
    else:
        steps.append(("5/6 judge (two models)", True, "skipped"))

    steps.append(run("6/6 review file",
                     [sys.executable, os.path.join(ROOT, "Tools", "review", "build_comparison_md.py")],
                     timeout=600))

    lines = [f"e2e asset check {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}", ""]
    ok = True
    for step, passed, detail in steps:
        lines.append(f"{'PASS' if passed else 'FAIL'}  {step}: {detail}")
        ok = ok and passed
    lines += ["", f"VERDICT: {'PASS' if ok else 'FAIL'}",
              "Review: Screenshots/README.md (status table at the top)."]
    os.makedirs(os.path.dirname(REPORT), exist_ok=True)
    with open(REPORT, "w") as handle:
        handle.write("\n".join(lines) + "\n")
    print()
    print("\n".join(lines))
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(main())