ykcheungab
340 lines
13 KiB
Python
340 lines
13 KiB
Python
"""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,
|
|
source_geometry_id=assets.source_geometry_id,
|
|
)
|
|
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["field_domain_id"] = assets.field_domain_id
|
|
report["source_geometry_id"] = assets.source_geometry_id
|
|
report["compatible"] = {
|
|
"sensor_model": assets.compatible_sensor_model,
|
|
"hardware_revisions": list(assets.compatible_hardware_revisions),
|
|
}
|
|
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))),
|
|
str(((contents.manifest.get("source_geometry") or {}) or {}).get("id") or ""),
|
|
)
|
|
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")
|
|
record(
|
|
"source geometry recorded",
|
|
# Not a failure: a package without the record is still publishable, it just
|
|
# cannot be compared with a device's own geometry. ``--require-signature``
|
|
# is where a release gets to demand more.
|
|
True,
|
|
assets.source_geometry_id[:16] if assets.source_geometry_id else
|
|
"not recorded (this package predates the record: a device reports it as "
|
|
"unverified rather than as a match)",
|
|
)
|
|
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())
|