orisys-app/tools/scene_assets/validate_scene.py

"""Validate a scene package, or import a legacy export and validate that.

Stage 2 deliverable (``docs/architecture/studio-3d-stages.md``): the checks the
loader and the compiler must agree on, runnable on a build machine and in CI.

Exit codes: 0 = every check passed, 2 = a check failed (with the reason printed),
1 = the arguments or the files were wrong.

Examples::

    # a package produced by compile_scene.py
    python tools/scene_assets/validate_scene.py build/finger.oryscene

    # the legacy export, before packaging (units are stated, never guessed)
    python tools/scene_assets/validate_scene.py \\
        --legacy-dir sensors/finger/assets/30fps-0703 \\
        --map       sensors/finger/assets/30fps-0703/unroll_map_uv_s0.4.npz \\
        --meters-per-source-unit 0.001
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path


def _bootstrap_paths() -> None:
    here = Path(__file__).resolve()
    repo = here.parents[2]
    for path in (repo / "apps" / "studio" / "src", repo / "shared" / "src"):
        if path.is_dir() and str(path) not in sys.path:
            sys.path.insert(0, str(path))


def _legacy_report(args) -> tuple[int, dict]:
    import numpy as np

    from orisys_app_shared.scene3d.contracts import ExpectedGeometry, ThreeDError, validate_assets
    from orisys_app_shared.scene3d.legacy import (
        LegacyFingerImporter,
        estimate_meters_per_source_unit,
        load_legacy_domain,
        split_at_uv_seams,
        uv_correspondence_residual,
    )

    report: dict = {"source": "legacy", "dir": str(args.legacy_dir), "checks": []}

    def record(name: str, ok: bool, detail: str = "") -> None:
        report["checks"].append({"check": name, "ok": bool(ok), "detail": detail})

    map_path = Path(args.map)
    domain = load_legacy_domain(map_path)
    report["field_domain"] = {
        "shape_hw": list(domain.shape_hw),
        "flip_v": domain.flip_v,
        "source_scale": domain.source_scale,
        "domain_id": domain.domain_id,
        "valid_ratio": round(domain.valid_ratio, 4),
    }
    record("field domain reads", True, f"{domain.shape_hw} flip_v={domain.flip_v}")

    with np.load(Path(args.legacy_dir) / "mesh.npz", allow_pickle=False) as handle:
        positions, uvs, _, _ = split_at_uv_seams(
            handle["vertices"], handle["faces"], handle["uv_data"], handle["loop_to_vertex"]
        )
    report["mesh"] = {
        "render_vertices": int(len(positions)),
        "source_vertices": int(len(np.load(Path(args.legacy_dir) / "mesh.npz")["vertices"])),
        "extent_source_units": [float(v) for v in np.ptp(positions, axis=0)],
    }
    if domain.world_xyz is not None:
        residual, samples = uv_correspondence_residual(positions, uvs, domain)
        report["uv_correspondence"] = {
            "relative_residual": round(residual, 4),
            "samples": int(samples),
        }
        record(
            "UV convention agrees with world_xyz",
            residual < 0.05,
            f"residual {residual:.4f} over {samples} samples",
        )
    measured = estimate_meters_per_source_unit(positions)
    report["scale"] = {
        "meters_per_source_unit_if_span_is_1m": round(measured, 6),
        "stated": args.meters_per_source_unit,
        "extent_m": (
            [float(v) * args.meters_per_source_unit for v in np.ptp(positions, axis=0)]
            if args.meters_per_source_unit
            else None
        ),
    }

    if not args.meters_per_source_unit:
        report["status"] = "needs_units"
        report["note"] = (
            "state --meters-per-source-unit: the importer never guesses what a source "
            "unit is. Compare 'extent_source_units' with the sensor's real size; a "
            "15.8-unit span is 15.8 mm on the shipped finger sensor, i.e. 0.001."
        )
        return 1, report

    try:
        assets = LegacyFingerImporter().convert(
            Path(args.legacy_dir),
            map_path,
            transform_sensor_from_source=np.eye(4),
            meters_per_source_unit=float(args.meters_per_source_unit),
        )
    except ThreeDError as exc:
        record("import", False, f"{exc.diagnostic.code}: {exc.diagnostic.message}")
        report["status"] = "rejected"
        return 2, report
    record("import", True, f"{assets.sensing_part().positions_sensor_m.shape[0]} vertices")
    try:
        validate_assets(assets)
        record("contracts", True)
    except ThreeDError as exc:
        record("contracts", False, f"{exc.diagnostic.code}: {exc.diagnostic.message}")
    if args.sensor_model:
        expected = ExpectedGeometry(
            sensor_model=args.sensor_model,
            hardware_revision=args.hardware_revision,
            geometry_id=assets.geometry_id,
            field_domain_id=assets.field_domain_id,
            shape_hw=assets.shape_hw,
        )
        try:
            validate_assets(assets, expected)
            record("device description", True, args.sensor_model)
        except ThreeDError as exc:
            record("device description", False, exc.diagnostic.code)
    report["geometry_id"] = assets.geometry_id
    report["asset_hash"] = assets.asset_hash
    report["binding"] = {
        "rows": int(assets.binding.sample_indices.shape[0]),
        "weight_sum_min": float(np.asarray(assets.binding.sample_weights).sum(axis=1).min()),
        "weight_sum_max": float(np.asarray(assets.binding.sample_weights).sum(axis=1).max()),
    }
    report["status"] = "ok" if all(item["ok"] for item in report["checks"]) else "failed"
    return (0 if report["status"] == "ok" else 2), report


def _load_keys(args):
    """The key provider and verification key the arguments describe, if any."""
    from orisys_app_shared.scene3d import protect

    provider = None
    if args.key_file is not None:
        provider = protect.load_key_file(Path(args.key_file), key_id=args.key_id)
    verify = None
    if args.verify_key is not None:
        path = Path(args.verify_key)
        if not path.is_file():
            raise SystemExit(f"the verification key is missing: {path}")
        verify = path.read_bytes()
    return provider, verify


def _package_report(args) -> tuple[int, dict]:
    from orisys_app_shared.scene3d.contracts import ExpectedGeometry, ThreeDError
    from orisys_app_shared.scene3d.package import (
        describe_envelope,
        describe_package,
        read_package,
        scene_from_contents,
    )

    report: dict = {"source": "package", "path": str(args.asset), "checks": []}

    def record(name: str, ok: bool, detail: str = "") -> None:
        report["checks"].append({"check": name, "ok": bool(ok), "detail": detail})

    try:
        provider, verify = _load_keys(args)
    except SystemExit as exc:
        record("key material", False, str(exc))
        report["status"] = "rejected"
        return 2, report

    # What the container says about itself, *without* a key: a support engineer can
    # report "sealed for key id X, signed by Y" without being able to read it.
    try:
        report["envelope"] = describe_envelope(Path(args.asset))
        record(
            "envelope reads",
            True,
            f"v{report['envelope'].get('envelope_version')} "
            f"{report['envelope'].get('payload')}"
            f"{' signed' if report['envelope'].get('signed') else ' unsigned'}",
        )
    except ThreeDError as exc:
        record("envelope reads", False, f"{exc.diagnostic.code}: {exc.diagnostic.message}")
        report["status"] = "rejected"
        return 2, report
    if report["envelope"].get("signed"):
        record("package is signed", True, str(report["envelope"].get("signing_key") or ""))
    else:
        # A warning by default, a hard failure when the caller says this is a release:
        # an engineering package is legitimately plain, and a release package that is
        # plain has no origin, which is the design document's step 2.
        record(
            "package is signed",
            not args.require_signature,
            "unsigned: integrity only, no origin (use --require-signature for a release "
            "build) "
            if not args.require_signature
            else "unsigned: a release package must be signed",
        )

    try:
        contents = read_package(Path(args.asset), keys=provider, verify_key_pem=verify)
        record("signature, envelope, manifest and every entry hash", True)
    except ThreeDError as exc:
        record(
            "signature, envelope, manifest and every entry hash",
            False,
            f"{exc.diagnostic.code}: {exc.diagnostic.message}",
        )
        report["status"] = "rejected"
        return 2, report
    report["manifest"] = describe_package(
        Path(args.asset), keys=provider, verify_key_pem=verify
    )
    lock = report["manifest"].get("platform_lock")
    if lock:
        record(
            "platform lock of this package",
            True,
            f"{lock.get('os_family') or 'any OS'} / {lock.get('machine') or 'any arch'}"
            f"{' / ' + str(lock.get('gl_profile')) if lock.get('gl_profile') else ''}",
        )
    try:
        expected = (
            ExpectedGeometry(
                args.sensor_model, args.hardware_revision,
                str(contents.manifest.get("geometry_id")),
                str((contents.manifest.get("field_domain") or {}).get("id")),
                tuple(int(v) for v in (contents.manifest.get("field_domain") or {}).get("shape_hw", (0, 0))),
            )
            if args.sensor_model
            else None
        )
        assets = scene_from_contents(contents, expected)
        record("scene builds and validates", True,
               f"{assets.sensing_part().positions_sensor_m.shape[0]} vertices")
    except ThreeDError as exc:
        record("scene builds and validates", False,
               f"{exc.diagnostic.code}: {exc.diagnostic.message}")
    report["status"] = "ok" if all(item["ok"] for item in report["checks"]) else "failed"
    return (0 if report["status"] == "ok" else 2), report


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("asset", nargs="?", type=Path, help="scene package to validate")
    parser.add_argument("--legacy-dir", type=Path, default=None, help="legacy NPZ export")
    parser.add_argument("--map", type=Path, default=None, help="legacy field map (npz)")
    parser.add_argument(
        "--meters-per-source-unit",
        type=float,
        default=None,
        help="stated unit scale; without it the legacy report stops at 'needs_units'",
    )
    parser.add_argument("--sensor-model", default="finger")
    parser.add_argument("--hardware-revision", default="*")
    parser.add_argument(
        "--key-file",
        type=Path,
        default=None,
        help="content key for a sealed package (hex, base64 or raw)",
    )
    parser.add_argument("--key-id", default="", help="key id that --key-file holds")
    parser.add_argument(
        "--verify-key",
        type=Path,
        default=None,
        help="Ed25519 public key (PEM) a signed package must verify against",
    )
    parser.add_argument(
        "--require-signature",
        action="store_true",
        help="treat an unsigned package as a failure: the release gate",
    )
    parser.add_argument("--json", action="store_true", help="machine-readable output")
    args = parser.parse_args(argv)

    _bootstrap_paths()
    if args.legacy_dir is not None:
        if args.map is None:
            print("--legacy-dir also needs --map", file=sys.stderr)
            return 1
        code, report = _legacy_report(args)
    elif args.asset is not None:
        code, report = _package_report(args)
    else:
        parser.print_help()
        return 1

    if args.json:
        print(json.dumps(report, indent=2, ensure_ascii=False, default=str))
    else:
        print(f"{report['status']}: {report.get('path') or report.get('dir')}")
        for item in report.get("checks", []):
            mark = "ok  " if item["ok"] else "FAIL"
            print(f"  {mark} {item['check']}  {item['detail']}")
        for key in ("envelope", "field_domain", "mesh", "uv_correspondence", "scale", "binding"):
            if key in report:
                print(f"  {key}: {json.dumps(report[key], ensure_ascii=False, default=str)}")
        if "note" in report:
            print(f"  note: {report['note']}")
    return code


if __name__ == "__main__":
    sys.exit(main())