#!/usr/bin/env python3 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path, PurePosixPath import re import subprocess import sys from typing import Any from script_utils import check_result_with_code as _check, emit_json_report PACKAGING_DEFINITION_FILENAME = "com.nvidia.simready.packaging.json" METADATA_FOLDER = ".metadata" BOM_FILENAME = "com.nvidia.simready.packaging.bom.json" ROOT_USDS_FILENAME = "com.nvidia.simready.root_usds.json" STANDARD_FORMAT_VERSION = "1.0" SIMREADY_PROFILE_VERSION = "1.0.0" PACKAGE_ID_PREFIX = "com.nvidia.simready" PACKAGE_ROOT_USD_SUFFIXES = {".usd", ".usda", ".usdc"} SUPPORTED_REFERENCED_SUFFIXES = {".usd", ".usda", ".usdc", ".usdz", ".png", ".jpg", ".jpeg", ".exr", ".m4a", ".mp3", ".wav"} SUPPORTED_PACKAGE_SIDECAR_SUFFIXES = {".json", ".md", ".txt"} PROFILE_CHOICES = ("Package", "Package-NoBOM") def _skill_name() -> str: return Path(sys.argv[0]).resolve().parents[1].name def _phase(name: str, checks: list[dict[str, Any]], success_message: str, fail_message: str) -> dict[str, Any]: passed = not _errors_from_checks(checks) return { "name": name, "passed": passed, "status": "PASS" if passed else "FAIL", "message": success_message if passed else fail_message, "checks": checks, } def _errors_from_checks(checks: list[dict[str, Any]]) -> list[str]: return [check["message"] for check in checks if check["severity"] == "error" and not check["passed"]] def _warnings_from_checks(checks: list[dict[str, Any]]) -> list[str]: return [check["message"] for check in checks if check["severity"] == "warning" and not check["passed"]] def _read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def _write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") def _hash_bytes(data: bytes) -> dict[str, str]: return {"sha256": hashlib.sha256(data).hexdigest()} def _hash_file(path: Path) -> dict[str, str]: return _hash_bytes(path.read_bytes()) def _content_files(package_root: Path) -> list[Path]: files: list[Path] = [] for path in package_root.rglob("*"): if not path.is_file(): continue rel = path.relative_to(package_root).as_posix() if rel == PACKAGING_DEFINITION_FILENAME: continue if rel == METADATA_FOLDER or rel.startswith(f"{METADATA_FOLDER}/"): continue if rel.endswith(".wrapp"): continue files.append(path) return sorted(files, key=lambda value: value.relative_to(package_root).as_posix()) def _build_bom(package_root: Path) -> dict[str, Any]: items = [] for path in _content_files(package_root): items.append( { "relative_path": path.relative_to(package_root).as_posix(), "size": path.stat().st_size, "hash": _hash_file(path), } ) return { "format_version": STANDARD_FORMAT_VERSION, "content_root": str(package_root), "items": items, } def _compute_content_hash(items: list[dict[str, Any]]) -> dict[str, str] | None: buffer = bytearray() for item in sorted(items, key=lambda value: str(value["relative_path"]).encode("utf-8")): sha256 = item.get("hash", {}).get("sha256") if not sha256: return None buffer.extend(str(item["relative_path"]).encode("utf-8")) buffer.append(0) buffer.extend(bytes.fromhex(sha256)) return _hash_bytes(bytes(buffer)) def _compute_package_hash( package_id: str, license_id: str, content_hash: dict[str, str], metadata_entries: list[dict[str, Any]], ) -> dict[str, str]: buffer = bytearray() buffer.extend(package_id.encode("utf-8")) buffer.append(0) buffer.extend(license_id.encode("utf-8")) buffer.append(0) buffer.extend(bytes.fromhex(content_hash["sha256"])) for entry in sorted(metadata_entries, key=lambda value: str(value["name"]).encode("utf-8")): buffer.extend(str(entry["name"]).encode("utf-8")) buffer.append(0) buffer.extend(bytes.fromhex(entry["hash"]["sha256"])) return _hash_bytes(bytes(buffer)) def _is_valid_posix_relative_path(value: str) -> bool: pure = PurePosixPath(value) if "\\" in value or pure.is_absolute(): return False if not value or value.startswith("../") or value == ".." or "/../" in value: return False return True def _open_usd(path: Path) -> bool: try: from pxr import Usd except Exception: return False try: return Usd.Stage.Open(str(path)) is not None except Exception: return False def _validate_root_usds(package_root: Path, root_usds: list[str] | None) -> tuple[list[dict[str, Any]], list[str]]: checks: list[dict[str, Any]] = [] entries = sorted(root_usds or []) checks.append( _check( "root_usds_declared", bool(entries), "Root USD entries are declared" if entries else "No root USD entries were declared", code="PKG.CONF.002", ) ) valid_entries: list[str] = [] seen: set[str] = set() for entry in entries: duplicate = entry in seen seen.add(entry) checks.append(_check(f"root_usd_unique:{entry}", not duplicate, f"Root USD entry is unique: {entry}" if not duplicate else f"Duplicate root USD entry: {entry}", code="PKG.CONF.002")) rel_ok = _is_valid_posix_relative_path(entry) checks.append(_check(f"root_usd_relative:{entry}", rel_ok, f"Root USD entry is relative: {entry}" if rel_ok else f"Root USD entry must be a forward-slash relative package path: {entry}", code="PKG.CONF.002")) suffix_ok = Path(entry).suffix.lower() in PACKAGE_ROOT_USD_SUFFIXES checks.append(_check(f"root_usd_suffix:{entry}", suffix_ok, f"Root USD entry has supported suffix: {entry}" if suffix_ok else f"Root USD entry must end with .usd, .usda, or .usdc: {entry}", code="PKG.CONF.002")) if not rel_ok: continue path = package_root / PurePosixPath(entry) exists = path.is_file() checks.append(_check(f"root_usd_exists:{entry}", exists, f"Root USD exists: {entry}" if exists else f"Root USD does not exist: {entry}", code="PKG.CONF.002")) if exists: opens = _open_usd(path) checks.append(_check(f"root_usd_opens:{entry}", opens, f"Root USD opens: {entry}" if opens else f"Root USD cannot be opened: {entry}", code="PKG.CONF.002")) if opens: valid_entries.append(entry) return checks, valid_entries def _validate_sidecar_types(package_root: Path) -> list[dict[str, Any]]: unsupported: list[str] = [] for path in _content_files(package_root): suffix = path.suffix.lower() if suffix in SUPPORTED_REFERENCED_SUFFIXES or suffix in SUPPORTED_PACKAGE_SIDECAR_SUFFIXES: continue unsupported.append(path.relative_to(package_root).as_posix()) return [ _check( "package_content_file_types", not unsupported, "Package content file types are supported" if not unsupported else f"Package contains files outside the MVP supported type set: {unsupported}", severity="warning", code="AA.002", ) ] def validate_package_source(package_root: Path, root_usds: list[str] | None) -> dict[str, Any]: package_root = package_root.resolve() checks = [ _check( "source_is_directory", package_root.is_dir(), "Source package root exists" if package_root.is_dir() else "Source package root does not exist or is not a directory", ) ] if package_root.is_dir(): root_checks, _ = _validate_root_usds(package_root, root_usds) checks.extend(root_checks) checks.extend(_validate_sidecar_types(package_root)) return _phase("pre-validation", checks, "Source passed package candidate checks", "Source failed package candidate checks") def _write_root_usds_metadata(package_root: Path, root_usds: list[str]) -> Path: path = package_root / METADATA_FOLDER / ROOT_USDS_FILENAME _write_json(path, {"format_version": STANDARD_FORMAT_VERSION, "entries": sorted(root_usds)}) return path def _write_bom_metadata(package_root: Path, bom: dict[str, Any]) -> Path: path = package_root / METADATA_FOLDER / BOM_FILENAME _write_json(path, bom) return path def _write_conformance_metadata( package_root: Path, root_usds: list[str], pre_checks: list[dict[str, Any]], content_hash: dict[str, str] | None, ) -> Path: path = package_root / METADATA_FOLDER / f"com.nvidia.simready.conformance.Package-Candidate@{SIMREADY_PROFILE_VERSION}.json" payload: dict[str, Any] = { "format_version": STANDARD_FORMAT_VERSION, "profile": "Package-Candidate", "profile_version": SIMREADY_PROFILE_VERSION, "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "validator": "nv-core-package-sample/scripts/run.py", "assets": [ { "asset": entry, "features": [ { "portable_simready_package_preflight": { "version": "0.1.0", "passed": not _errors_from_checks(pre_checks), "failing_requirements": sorted({check["code"] for check in pre_checks if check["code"] and not check["passed"]}), "dependencies": [], } } ], } for entry in root_usds ], } if content_hash is not None: payload["content_hash"] = content_hash _write_json(path, payload) return path def _metadata_entry(path: Path) -> dict[str, Any]: return {"name": path.name, "hash": _hash_file(path)} def _write_package_definition( package_root: Path, name: str, version: str, license_id: str, metadata_files: list[Path], content_hash: dict[str, str] | None, overwrite: bool, ) -> Path: path = package_root / PACKAGING_DEFINITION_FILENAME if path.exists() and not overwrite: raise FileExistsError(f"Package definition already exists at {path}; pass --overwrite to replace it") metadata_entries = [_metadata_entry(path) for path in sorted(metadata_files, key=lambda value: value.name)] package_id = f"{PACKAGE_ID_PREFIX}.{name}.{version}" payload: dict[str, Any] = { "format_version": STANDARD_FORMAT_VERSION, "package_id": package_id, "license": license_id, "metadata": metadata_entries, } if content_hash is not None: payload["content_hash"] = content_hash payload["package_hash"] = _compute_package_hash(package_id, license_id, content_hash, metadata_entries) _write_json(path, payload) return path def _base_report(package_root: Path, skill: str, operation: str, backend: str, profile: str) -> dict[str, Any]: return { "package_root": str(package_root.resolve()), "package_definition_path": None, "skill": skill, "tool": "portable SimReady package script", "operation": operation, "backend": backend, "profile": profile, "passed": False, "status": "FAIL", "checks": [], "phases": [], "metadata": {}, "warnings": [], "errors": [], "command": [], "next_step": "report-validation-result", } def _finalize(report: dict[str, Any]) -> dict[str, Any]: checks = list(report.get("checks", [])) for phase in report.get("phases", []): checks.extend(phase.get("checks", [])) errors = list(dict.fromkeys(report.get("errors", []) + _errors_from_checks(checks))) warnings = list(dict.fromkeys(report.get("warnings", []) + _warnings_from_checks(checks))) report["errors"] = errors report["warnings"] = warnings report["passed"] = not errors report["status"] = "PASS" if not errors else report.get("status", "FAIL") if errors and report["status"] == "PASS": report["status"] = "FAIL" return report def _run_wrapp_sample(args: argparse.Namespace) -> dict[str, Any]: source = args.source.resolve() report = _base_report(source, "nv-core-package-sample", "package", "wrapp", "Package") sample_dir = args.upstream_sample_dir.resolve() if args.upstream_sample_dir else None checks = [ _check( "upstream_sample_dir_provided", sample_dir is not None, "WRAPP upstream package script directory provided" if sample_dir is not None else ( "WRAPP backend requires --upstream-scripts-dir pointing at " "skills/simready-foundation-create-package/assets/scripts" ), ) ] script = sample_dir / "create_simready_package.py" if sample_dir is not None else None checks.append( _check( "upstream_sample_script_exists", script is not None and script.is_file(), f"Found upstream package sample script: {script}" if script is not None and script.is_file() else "Upstream create_simready_package.py was not found", ) ) checks.append(_check("wrapp_repo_provided", args.repo is not None, "WRAPP repository path provided" if args.repo else "WRAPP backend requires --repo")) checks.append(_check("root_usds_provided", bool(args.root_usd), "Root USD entries provided" if args.root_usd else "WRAPP backend requires at least one --root-usd entry", code="PKG.CONF.002")) report["checks"] = checks if _errors_from_checks(checks): report["status"] = "BLOCKED" report["next_step"] = "provide-wrapp-package-inputs" return _finalize(report) command = [ sys.executable, str(script), args.name, args.version, args.license_id, str(source), str(args.repo.resolve()), ] for root_usd in args.root_usd: command.extend(["--root-usd", root_usd]) report["command"] = command env = os.environ.copy() if sample_dir.parts[-4:] == ("skills", "simready-foundation-create-package", "assets", "scripts"): env.setdefault("SIMREADY_FOUNDATIONS_ROOT", str(sample_dir.parents[3])) completed = subprocess.run( command, cwd=sample_dir, env=env, capture_output=True, text=True, timeout=600, check=False, ) report["metadata"] = {"stdout": completed.stdout, "stderr": completed.stderr, "returncode": completed.returncode} report["checks"].append(_check("upstream_sample_completed", completed.returncode == 0, "Upstream package sample completed" if completed.returncode == 0 else "Upstream package sample failed")) report["next_step"] = "nv-core-package-sample-validation" if completed.returncode == 0 else "fix-package-workflow-inputs" return _finalize(report) def create_package(args: argparse.Namespace) -> dict[str, Any]: if args.backend == "wrapp": return _run_wrapp_sample(args) source = args.source.resolve() report = _base_report(source, "nv-core-package-sample", "package", "local", "Package") report["command"] = [str(Path(__file__).name), str(source), "--name", args.name, "--version", args.version, "--license", args.license_id] pre_phase = validate_package_source(source, args.root_usd) if not args.skip_pre_validation: report["phases"].append(pre_phase) if not pre_phase["passed"]: report["next_step"] = "fix-package-candidate" return _finalize(report) try: normalized_roots = sorted(args.root_usd) root_metadata = _write_root_usds_metadata(source, normalized_roots) bom = _build_bom(source) content_hash = _compute_content_hash(bom["items"]) bom_path = _write_bom_metadata(source, bom) conformance_path = _write_conformance_metadata(source, normalized_roots, pre_phase["checks"], content_hash) metadata_files = [root_metadata, bom_path, conformance_path] package_definition = _write_package_definition( source, args.name, args.version, args.license_id, metadata_files, content_hash, args.overwrite, ) except Exception as exc: create_checks = [_check("package_definition_written", False, f"Package creation failed: {exc}")] report["phases"].append(_phase("create", create_checks, "Package definition and metadata were written", str(exc))) report["next_step"] = "fix-package-create-inputs" return _finalize(report) create_checks = [ _check("package_definition_written", package_definition.is_file(), f"Wrote package definition: {package_definition}"), _check("bom_written", bom_path.is_file(), f"Wrote BOM metadata: {bom_path}", code="PKG.BOM.001"), _check("root_usds_written", root_metadata.is_file(), f"Wrote root USD metadata: {root_metadata}", code="PKG.CONF.002"), ] report["phases"].append(_phase("create", create_checks, "Package definition and metadata were written", "Package creation failed")) report["package_definition_path"] = str(package_definition) report["metadata"] = { "package_id": f"{PACKAGE_ID_PREFIX}.{args.name}.{args.version}", "root_usds": normalized_roots, "metadata_files": [str(path) for path in metadata_files], "bom_item_count": len(bom["items"]), "content_hash": content_hash, } if not args.skip_post_validation: validation_report = validate_package(package_definition, profile="Package") report["phases"].append( { "name": "post-validation", "passed": validation_report["passed"], "status": validation_report["status"], "message": "Package post-validation passed" if validation_report["passed"] else "Package post-validation failed", "checks": validation_report["checks"], } ) report["metadata"]["post_validation"] = validation_report["metadata"] report["next_step"] = "publish-or-consume-package" return _finalize(report) def _validate_package_definition_fields(path: Path, payload: Any) -> list[dict[str, Any]]: checks: list[dict[str, Any]] = [] is_object = isinstance(payload, dict) checks.append(_check("package_definition_is_object", is_object, "Package definition is a JSON object" if is_object else "Package definition must be a JSON object", code="PKG.DEF.001")) if not isinstance(payload, dict): return checks missing = sorted({"format_version", "package_id", "license"} - set(payload)) checks.append(_check("package_definition_required_fields", not missing, "Package definition has required fields" if not missing else f"Package definition missing required fields: {missing}", code="PKG.DEF.001")) checks.append(_check("format_version_valid", isinstance(payload.get("format_version"), str) and re.fullmatch(r"\d+\.\d+", payload["format_version"]) is not None, "format_version is major.minor" if isinstance(payload.get("format_version"), str) else "format_version must be a major.minor string", code="PKG.DEF.001")) package_id = payload.get("package_id") forbidden = set('<>:"/\\|?*') package_id_ok = isinstance(package_id, str) and bool(package_id) and len(package_id) <= 255 and not any(ch.isspace() or ch in forbidden or ord(ch) < 32 or 127 <= ord(ch) <= 159 for ch in package_id) checks.append(_check("package_id_valid", package_id_ok, "package_id is valid" if package_id_ok else "package_id is empty, too long, or contains forbidden characters", code="PKG.DEF.001")) checks.append(_check("license_valid", isinstance(payload.get("license"), str) and bool(payload.get("license")), "license is present" if payload.get("license") else "license must be a non-empty string", code="PKG.DEF.001")) checks.append(_check("package_definition_canonical_location", path.name == PACKAGING_DEFINITION_FILENAME, "Package definition uses the canonical file name" if path.name == PACKAGING_DEFINITION_FILENAME else f"Package definition must be named {PACKAGING_DEFINITION_FILENAME}", code="PKG.DEF.001")) return checks def _validate_metadata_files(package_root: Path, payload: dict[str, Any], require_metadata_dir: bool) -> list[dict[str, Any]]: checks: list[dict[str, Any]] = [] metadata_dir = package_root / METADATA_FOLDER checks.append(_check("metadata_directory_exists", metadata_dir.is_dir(), "Metadata directory exists" if metadata_dir.is_dir() else "Metadata directory is missing", severity="error" if require_metadata_dir else "warning", code="PKG.DEF.001")) entries = payload.get("metadata", []) if not isinstance(entries, list): checks.append(_check("metadata_array_valid", False, "metadata must be an array when present", code="PKG.DEF.001")) return checks seen: set[str] = set() for entry in entries: entry_ok = isinstance(entry, dict) and isinstance(entry.get("name"), str) and isinstance(entry.get("hash"), dict) checks.append(_check(f"metadata_entry_valid:{entry}", entry_ok, "Metadata entry has name and hash" if entry_ok else f"Metadata entry must include name and hash: {entry}", code="PKG.DEF.001")) if not entry_ok: continue name = entry["name"] duplicate = name in seen seen.add(name) checks.append(_check(f"metadata_entry_unique:{name}", not duplicate, f"Metadata entry is unique: {name}" if not duplicate else f"Duplicate metadata entry: {name}", code="PKG.DEF.001")) metadata_file = metadata_dir / name exists = metadata_file.is_file() checks.append(_check(f"metadata_file_exists:{name}", exists, f"Metadata file exists: {name}" if exists else f"Metadata file listed but missing: {name}", code="PKG.META.001")) if exists: actual_hash = _hash_file(metadata_file).get("sha256") expected_hash = entry["hash"].get("sha256") checks.append(_check(f"metadata_hash_matches:{name}", bool(expected_hash) and actual_hash == expected_hash, f"Metadata file hash matches: {name}" if actual_hash == expected_hash else f"Metadata file hash mismatch: {name}", code="PKG.DEF.001")) return checks def _validate_bom(package_root: Path, payload: dict[str, Any], require_bom: bool) -> tuple[list[dict[str, Any]], dict[str, Any] | None]: checks: list[dict[str, Any]] = [] bom_path = package_root / METADATA_FOLDER / BOM_FILENAME exists = bom_path.is_file() checks.append(_check("bom_exists", exists, "BOM metadata exists" if exists else "BOM metadata is missing", severity="error" if require_bom else "warning", code="PKG.BOM.001")) if not exists: return checks, None try: bom = _read_json(bom_path) except Exception as exc: checks.append(_check("bom_json_valid", False, f"BOM is not valid JSON: {exc}", code="PKG.BOM.001")) return checks, None items = bom.get("items") if isinstance(bom, dict) else None checks.append(_check("bom_items_array", isinstance(items, list), "BOM items is an array" if isinstance(items, list) else "BOM items must be an array", code="PKG.BOM.001")) if not isinstance(items, list): return checks, bom if isinstance(bom, dict) else None seen: set[str] = set() for item in items: rel = str(item.get("relative_path", "")) if isinstance(item, dict) else "" path_ok = isinstance(item, dict) and _is_valid_posix_relative_path(rel) checks.append(_check(f"bom_relative_path:{rel}", path_ok, f"BOM item path is valid: {rel}" if path_ok else f"BOM item path must be a forward-slash relative path: {rel}", code="PKG.BOM.001")) duplicate = rel in seen seen.add(rel) checks.append(_check(f"bom_relative_path_unique:{rel}", not duplicate, f"BOM item path is unique: {rel}" if not duplicate else f"Duplicate BOM item path: {rel}", code="PKG.BOM.001")) if not path_ok: continue path = package_root / PurePosixPath(rel) file_exists = path.is_file() checks.append(_check(f"bom_file_exists:{rel}", file_exists, f"BOM item file exists: {rel}" if file_exists else f"BOM item file is missing: {rel}", code="PKG.BOM.001")) if file_exists: checks.append(_check(f"bom_file_size:{rel}", isinstance(item.get("size"), int) and item["size"] == path.stat().st_size, f"BOM item size matches: {rel}" if item.get("size") == path.stat().st_size else f"BOM item size mismatch: {rel}", code="PKG.BOM.001")) expected = item.get("hash", {}).get("sha256") if isinstance(item.get("hash"), dict) else None actual = _hash_file(path).get("sha256") checks.append(_check(f"bom_file_hash:{rel}", bool(expected) and expected == actual, f"BOM item hash matches: {rel}" if expected == actual else f"BOM item sha256 hash mismatch: {rel}", code="PKG.BOM.001")) actual_content = {path.relative_to(package_root).as_posix() for path in _content_files(package_root)} bom_content = {str(item.get("relative_path", "")) for item in items if isinstance(item, dict)} checks.append(_check("bom_complete", actual_content == bom_content, "BOM lists every package content file" if actual_content == bom_content else f"BOM content mismatch; missing={sorted(actual_content - bom_content)}, extra={sorted(bom_content - actual_content)}", code="PKG.BOM.001")) computed_content_hash = _compute_content_hash(items) content_hash = payload.get("content_hash") if isinstance(content_hash, dict) and computed_content_hash is not None: checks.append(_check("content_hash_matches", content_hash.get("sha256") == computed_content_hash.get("sha256"), "content_hash matches BOM items" if content_hash.get("sha256") == computed_content_hash.get("sha256") else "content_hash does not match BOM items", code="PKG.HASH.001")) return checks, bom def _root_usds_metadata(package_root: Path) -> tuple[list[dict[str, Any]], list[str]]: path = package_root / METADATA_FOLDER / ROOT_USDS_FILENAME if not path.is_file(): discovered = sorted(p.relative_to(package_root).as_posix() for p in package_root.rglob("*") if p.is_file() and p.suffix.lower() in PACKAGE_ROOT_USD_SUFFIXES and METADATA_FOLDER not in p.parts) return [_check("root_usds_metadata_exists", False, "Root USD metadata is missing", severity="warning", code="PKG.CONF.002")], discovered try: payload = _read_json(path) except Exception as exc: return [_check("root_usds_json_valid", False, f"Root USD metadata is not valid JSON: {exc}", code="PKG.CONF.002")], [] entries = payload.get("entries") if isinstance(payload, dict) else None if not isinstance(entries, list): return [_check("root_usds_entries_array", False, "Root USD metadata must include entries array", code="PKG.CONF.002")], [] return _validate_root_usds(package_root, [str(entry) for entry in entries]) def _validate_package_hash(payload: dict[str, Any]) -> list[dict[str, Any]]: package_hash = payload.get("package_hash") content_hash = payload.get("content_hash") entries = payload.get("metadata", []) if package_hash is None: return [_check("package_hash_available", False, "package_hash is not present", severity="warning", code="PKG.HASH.001")] can_compute = isinstance(payload.get("package_id"), str) and isinstance(payload.get("license"), str) and isinstance(content_hash, dict) and isinstance(content_hash.get("sha256"), str) and isinstance(entries, list) if not can_compute: return [_check("package_hash_computable", False, "package_hash cannot be computed from package definition fields", code="PKG.HASH.001")] computed = _compute_package_hash(payload["package_id"], payload["license"], content_hash, entries) return [_check("package_hash_matches", package_hash.get("sha256") == computed.get("sha256"), "package_hash matches package definition" if package_hash.get("sha256") == computed.get("sha256") else "package_hash does not match package definition", code="PKG.HASH.001")] def validate_package(package_definition: Path, *, profile: str) -> dict[str, Any]: package_definition = package_definition.resolve() package_root = package_definition.parent report = _base_report(package_root, "nv-core-package-sample-validation", "validate", "local", profile) report["package_definition_path"] = str(package_definition) checks = report["checks"] exists = package_definition.is_file() checks.append(_check("package_definition_exists", exists, "Package definition exists" if exists else "Package definition does not exist", code="PKG.DEF.001")) if not exists: return _finalize(report) try: payload = _read_json(package_definition) except Exception as exc: checks.append(_check("package_definition_json_valid", False, f"Package definition is not valid JSON: {exc}", code="PKG.DEF.001")) return _finalize(report) checks.append(_check("package_definition_json_valid", True, "Package definition is valid JSON", code="PKG.DEF.001")) checks.extend(_validate_package_definition_fields(package_definition, payload)) metadata = {"package_id": None, "metadata_entries": [], "bom_item_count": 0, "root_usds": []} if isinstance(payload, dict): metadata["package_id"] = payload.get("package_id") metadata["metadata_entries"] = [entry.get("name") for entry in payload.get("metadata", []) if isinstance(entry, dict)] checks.extend(_validate_metadata_files(package_root, payload, require_metadata_dir=profile == "Package")) bom_checks, bom = _validate_bom(package_root, payload, require_bom=profile == "Package") checks.extend(bom_checks) if bom and isinstance(bom.get("items"), list): metadata["bom_item_count"] = len(bom["items"]) root_checks, root_entries = _root_usds_metadata(package_root) checks.extend(root_checks) metadata["root_usds"] = root_entries checks.extend(_validate_sidecar_types(package_root)) checks.extend(_validate_package_hash(payload)) report["metadata"] = metadata report["next_step"] = "publish-or-consume-package" return _finalize(report) def _markdown(report: dict[str, Any]) -> str: lines = [ "# SimReady Package Report", "", f"- Package root: `{report['package_root']}`", f"- Package definition: `{report['package_definition_path']}`", f"- Skill: `{report['skill']}`", f"- Operation: `{report['operation']}`", f"- Backend: `{report['backend']}`", f"- Profile: `{report['profile']}`", f"- Passed: `{report['passed']}`", f"- Status: `{report['status']}`", f"- Next step: `{report['next_step']}`", "", "## Checks", "", ] checks = list(report.get("checks", [])) for phase in report.get("phases", []): checks.extend(phase.get("checks", [])) for check in checks: state = "PASS" if check["passed"] else "FAIL" code = f" `{check['code']}`" if check.get("code") else "" lines.append(f"- `{state}` `{check['name']}`{code}: {check['message']}") if report["errors"]: lines.extend(["", "## Errors", ""]) lines.extend(f"- {error}" for error in report["errors"]) if report["warnings"]: lines.extend(["", "## Warnings", ""]) lines.extend(f"- {warning}" for warning in report["warnings"]) lines.append("") return "\n".join(lines) def _emit(report: dict[str, Any], report_path: Path | None, markdown_report_path: Path | None) -> None: emit_json_report(report, report_path, markdown_report_path, _markdown(report)) def _package_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Create a SimReady package definition and metadata.") parser.add_argument("source", type=Path) parser.add_argument("--name", required=True) parser.add_argument("--version", default="1.0.0") parser.add_argument("--license", dest="license_id", required=True) parser.add_argument("--root-usd", action="append", default=[]) parser.add_argument("--backend", choices=("local", "wrapp"), default="local") parser.add_argument("--repo", type=Path) parser.add_argument( "--upstream-sample-dir", "--upstream-scripts-dir", dest="upstream_sample_dir", type=Path, ) parser.add_argument("--skip-pre-validation", action="store_true") parser.add_argument("--skip-post-validation", action="store_true") parser.add_argument("--overwrite", action="store_true") parser.add_argument("--report", type=Path) parser.add_argument("--markdown-report", type=Path) return parser def _validate_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Validate a SimReady package definition and metadata.") parser.add_argument("package_definition", type=Path) parser.add_argument("--profile", choices=PROFILE_CHOICES, default="Package") parser.add_argument("--report", type=Path) parser.add_argument("--markdown-report", type=Path) return parser def main(argv: list[str] | None = None) -> int: if _skill_name() == "nv-core-package-sample-validation": args = _validate_parser().parse_args(argv) report = validate_package(args.package_definition, profile=args.profile) else: args = _package_parser().parse_args(argv) report = create_package(args) _emit(report, args.report, args.markdown_report) return 0 if report["passed"] else 1 if __name__ == "__main__": raise SystemExit(main())