EVEngine/scripts/compare_render_backends.py

#!/usr/bin/env python3
"""Compare Vulkan and WebGPU render-parity artifact directories."""

from __future__ import annotations

import argparse
import json
import math
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from PIL import Image

try:
    from scripts.ci_result import CIResult, FailureKind
except ModuleNotFoundError:  # Direct execution sets sys.path to scripts/.
    from ci_result import CIResult, FailureKind
import utf8_stdio


class ArtifactContractError(ValueError):
    """Raised when render-parity producers emitted incomplete artifact sets."""


@dataclass(frozen=True)
class RenderContract:
    """Versioned, backend-neutral comparison semantics for one rendered scene."""

    color_space: str
    alpha_coverage_delta_max: float
    mean_rgb_error_max: float
    p99_rgb_error_max: float


def parse_manifest(path: Path) -> dict[str, Any]:
    try:
        manifest = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ArtifactContractError(f"manifest {path.name} is invalid: {error}") from error
    for field in ("schema", "version", "scene", "backend", "contract"):
        if field not in manifest:
            raise ArtifactContractError(
                f"manifest {path.name} missing required field: {field}"
            )
    if manifest["schema"] != "evengine.render-parity" or manifest["version"] != 1:
        raise ArtifactContractError(
            f"manifest {path.name} has unsupported schema/version: "
            f"{manifest['schema']!r}/{manifest['version']!r}"
        )
    return manifest


def parse_render_contract(manifest: dict[str, Any], path: Path) -> RenderContract:
    payload = manifest["contract"]
    required = (
        "color_space",
        "alpha_coverage_delta_max",
        "mean_rgb_error_max",
        "p99_rgb_error_max",
    )
    if not isinstance(payload, dict):
        raise ArtifactContractError(f"manifest {path.name} contract must be an object")
    missing = [field for field in required if field not in payload]
    if missing:
        raise ArtifactContractError(
            f"manifest {path.name} contract missing: {', '.join(missing)}"
        )
    if payload["color_space"] != "srgb-linearized":
        raise ArtifactContractError(
            f"manifest {path.name} has unsupported color space: {payload['color_space']!r}"
        )
    limits = [float(payload[field]) for field in required[1:]]
    if any(not math.isfinite(value) or value < 0.0 for value in limits):
        raise ArtifactContractError(f"manifest {path.name} has invalid tolerance")
    return RenderContract(str(payload["color_space"]), *limits)


def srgb_to_linear(value: int) -> float:
    channel = value / 255.0
    if channel <= 0.04045:
        return channel / 12.92
    return math.pow((channel + 0.055) / 1.055, 2.4)


def compare_scene(
    reference_dir: Path, candidate_dir: Path, manifest_path: Path
) -> list[str]:
    manifest = parse_manifest(manifest_path)
    contract = parse_render_contract(manifest, manifest_path)
    scene = manifest["scene"]
    reference = Image.open(reference_dir / f"{scene}.png").convert("RGBA")
    candidate = Image.open(candidate_dir / f"{scene}.png").convert("RGBA")
    failures: list[str] = []
    if reference.size != candidate.size:
        return [f"{scene}: dimensions differ: {reference.size} vs {candidate.size}"]

    ref_pixels = list(reference.get_flattened_data())
    candidate_pixels = list(candidate.get_flattened_data())
    alpha_different = sum(
        (ref[3] >= 128) != (candidate[3] >= 128)
        for ref, candidate in zip(ref_pixels, candidate_pixels)
    )
    alpha_ratio = alpha_different / len(ref_pixels)
    errors = sorted(
        abs(srgb_to_linear(ref[channel]) - srgb_to_linear(candidate[channel]))
        for ref, candidate in zip(ref_pixels, candidate_pixels)
        for channel in range(3)
    )
    mean_error = sum(errors) / len(errors)
    percentile_99 = errors[min(len(errors) - 1, math.ceil(len(errors) * 0.99) - 1)]

    if alpha_ratio > contract.alpha_coverage_delta_max:
        failures.append(
            f"{scene}: alpha coverage delta {alpha_ratio:.4%} > "
            f"{contract.alpha_coverage_delta_max:.4%}"
        )
    if mean_error > contract.mean_rgb_error_max:
        failures.append(
            f"{scene}: mean linear RGB error {mean_error:.6f} > "
            f"{contract.mean_rgb_error_max:.6f}"
        )
    if percentile_99 > contract.p99_rgb_error_max:
        failures.append(
            f"{scene}: p99 linear RGB error {percentile_99:.6f} > "
            f"{contract.p99_rgb_error_max:.6f}"
        )
    print(
        f"{scene}: alpha={alpha_ratio:.4%} mean={mean_error:.6f} "
        f"p99={percentile_99:.6f} color_space={contract.color_space}"
    )
    return failures


def validate_artifact_contract(
    reference_dir: Path, candidate_dir: Path
) -> list[Path]:
    """Validate that both backends emitted the same complete manifest/image set."""
    reference_manifests = {path.name: path for path in reference_dir.glob("*.json")}
    candidate_manifests = {path.name: path for path in candidate_dir.glob("*.json")}
    if not reference_manifests:
        raise ArtifactContractError(f"no manifests found in {reference_dir}")
    if not candidate_manifests:
        raise ArtifactContractError(f"no manifests found in {candidate_dir}")

    errors: list[str] = []
    missing = sorted(reference_manifests.keys() - candidate_manifests.keys())
    unexpected = sorted(candidate_manifests.keys() - reference_manifests.keys())
    if missing:
        errors.append("candidate missing manifests: " + ", ".join(missing))
    if unexpected:
        errors.append("candidate has unexpected manifests: " + ", ".join(unexpected))

    for backend, directory, manifests in (
        ("reference", reference_dir, reference_manifests),
        ("candidate", candidate_dir, candidate_manifests),
    ):
        for name, manifest_path in sorted(manifests.items()):
            try:
                manifest = parse_manifest(manifest_path)
                parse_render_contract(manifest, manifest_path)
                scene = manifest["scene"]
            except (ArtifactContractError, KeyError, TypeError, ValueError) as error:
                errors.append(f"{backend} manifest {name} is invalid: {error}")
                continue
            image_path = directory / f"{scene}.png"
            if not image_path.is_file():
                errors.append(
                    f"{backend} manifest {name} is missing image {image_path.name}"
                )

    for name in sorted(reference_manifests.keys() & candidate_manifests.keys()):
        try:
            reference_manifest = parse_manifest(reference_manifests[name])
            candidate_manifest = parse_manifest(candidate_manifests[name])
            if reference_manifest["scene"] != candidate_manifest["scene"]:
                errors.append(f"manifest {name} scene differs between backends")
            if reference_manifest["contract"] != candidate_manifest["contract"]:
                errors.append(f"manifest {name} contract differs between backends")
        except ArtifactContractError:
            # The backend-specific validation above already reports malformed input.
            pass

    if errors:
        raise ArtifactContractError("\n".join(errors))
    return [reference_manifests[name] for name in sorted(reference_manifests)]


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("reference", type=Path, help="Vulkan artifact directory")
    parser.add_argument("candidate", type=Path, help="WebGPU artifact directory")
    parser.add_argument("--result-json", type=Path, help="write a versioned CI result")
    args = parser.parse_args()

    try:
        manifests = validate_artifact_contract(args.reference, args.candidate)
    except ArtifactContractError as error:
        print("ARTIFACT CONTRACT FAILURE:", file=sys.stderr)
        print(error, file=sys.stderr)
        if args.result_json:
            CIResult.failure(
                FailureKind.CONTRACT,
                "RENDER_ARTIFACT_CONTRACT_INVALID",
                "render-parity",
                str(error),
            ).write(args.result_json)
        return 2
    failures: list[str] = []
    for manifest in manifests:
        failures.extend(compare_scene(args.reference, args.candidate, manifest))
    if failures:
        print("PIXEL PARITY FAILURE:", file=sys.stderr)
        print("\n".join(f"ERROR: {failure}" for failure in failures), file=sys.stderr)
        if args.result_json:
            CIResult.failure(
                FailureKind.PARITY,
                "RENDER_PARITY_EXCEEDED",
                "render-parity",
                "one or more scenes exceeded their render contract",
                {"failures": failures},
            ).write(args.result_json)
        return 1
    if args.result_json:
        CIResult.success("render-parity").write(args.result_json)
    return 0


if __name__ == "__main__":
    utf8_stdio.enable_utf8_stdio()
    raise SystemExit(main())