#!/usr/bin/env python3 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 """Route a source asset by asking upstream converter references for support. Usage: python3 scripts/run.py [--report PATH] python3 scripts/run.py [--markdown-report PATH] Arguments: source_asset Path to the input file or directory to convert. output_directory Directory to write the generated USD artifact and reports. --report PATH Optional path to write the normalized conversion report (JSON). --markdown-report PATH Optional path to write the normalized conversion report (Markdown). Exit codes: 0 - conversion succeeded or source is already USD 1 - expected failure (unsupported source format or missing required converter) 2 - unexpected error (crash or malformed input) """ from __future__ import annotations import argparse from dataclasses import dataclass, field, replace import json from pathlib import Path import subprocess import sys import tempfile from typing import Any import zipfile SKILL = "convert-to-usd" NEXT_STEP = "validate-usd-minimum" PROBE_TIMEOUT_SECONDS = 30 CONVERSION_TIMEOUT_SECONDS = 1900 REFERENCE_ORDER = ( "urdf-usd-converter", "mujoco-usd-converter", "usd-convert-gsplat", "usd-convert-cad", ) @dataclass(frozen=True) class ProbeResult: converter_skill: str converter_tool: str source_format: str supported: bool warnings: list[str] = field(default_factory=list) errors: list[str] = field(default_factory=list) install_hint: str = "" @dataclass(frozen=True) class SourceSelection: source_asset: Path warnings: list[str] = field(default_factory=list) selected_probe: ProbeResult | None = None probes: list[ProbeResult] = field(default_factory=list) @dataclass(frozen=True) class ConversionReport: source_asset_path: str source_format: str converter_skill: str converter_tool: str converter_command: list[str] output_directory: str output_usd_path: str generated_files: list[str] converter_reference: str = "" sidecar_inputs: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) errors: list[str] = field(default_factory=list) diagnostics: list[dict[str, Any]] = field(default_factory=list) install_hint: str = "" next_step: str = NEXT_STEP @property def passed(self) -> bool: return not self.errors def to_dict(self) -> dict[str, Any]: payload = { "source_asset_path": self.source_asset_path, "source_format": self.source_format, "converter_skill": self.converter_skill, "converter_tool": self.converter_tool, "converter_command": self.converter_command, "output_directory": self.output_directory, "output_usd_path": self.output_usd_path, "generated_files": self.generated_files, "sidecar_inputs": self.sidecar_inputs, "warnings": self.warnings, "errors": self.errors, "next_step": self.next_step, } if self.converter_reference: payload["converter_reference"] = self.converter_reference if self.diagnostics: payload["diagnostics"] = self.diagnostics if self.install_hint: payload["install_hint"] = self.install_hint return payload def to_json(self) -> str: return json.dumps(self.to_dict(), indent=2, sort_keys=True) + "\n" def to_markdown(self) -> str: command = " ".join(self.converter_command) lines = [ "# Conversion Report", "", f"- Source asset: `{self.source_asset_path}`", f"- Source format: `{self.source_format}`", f"- Converter skill: `{self.converter_skill}`", f"- Converter tool: `{self.converter_tool}`", f"- Converter command: `{command}`", f"- Output directory: `{self.output_directory}`", f"- Output USD: `{self.output_usd_path}`", f"- Next step: `{self.next_step}`", "", "## Generated Files", "", ] lines.extend(f"- `{path}`" for path in self.generated_files) if not self.generated_files: lines.append("- None") lines.extend(["", "## Warnings", ""]) lines.extend(f"- {warning}" for warning in self.warnings) if not self.warnings: lines.append("- None") lines.extend(["", "## Errors", ""]) lines.extend(f"- {error}" for error in self.errors) if not self.errors: lines.append("- None") if self.install_hint: lines.extend(["", "## Install Hint", "", self.install_hint]) lines.append("") return "\n".join(lines) def reference_root() -> Path: return Path(__file__).resolve().parents[1] / "references" def reference_run_script(converter_skill: str) -> Path: return reference_root() / converter_skill / "scripts" / "run.py" def is_existing_usd(source_asset: Path) -> bool: if not source_asset.is_file(): return False try: with source_asset.open("rb") as file: header = file.read(16) except OSError: return False if header.startswith(b"#usda") or header.startswith(b"PXR-USDC"): return True if zipfile.is_zipfile(source_asset): try: with zipfile.ZipFile(source_asset) as archive: for name in archive.namelist(): with archive.open(name) as member: member_header = member.read(16) if member_header.startswith(b"#usda") or member_header.startswith(b"PXR-USDC"): return True return False except zipfile.BadZipFile: return False return False def already_usd_report(source_asset: Path, output_directory: Path) -> ConversionReport: return ConversionReport( source_asset_path=str(source_asset), source_format="usd", converter_skill=SKILL, converter_tool="none", converter_command=[], output_directory=str(output_directory), output_usd_path=str(source_asset), generated_files=[], warnings=["Source asset is already USD; conversion skipped"], ) def missing_source_report(source_asset: Path, output_directory: Path) -> ConversionReport: return ConversionReport( source_asset_path=str(source_asset), source_format="unknown", converter_skill=SKILL, converter_tool="none", converter_command=[], output_directory=str(output_directory), output_usd_path="", generated_files=[], errors=["Source asset does not exist"], ) def run_probe(source_asset: Path, converter_skill: str) -> ProbeResult: script = reference_run_script(converter_skill) if not script.exists(): return ProbeResult( converter_skill=converter_skill, converter_tool="none", source_format="unknown", supported=False, errors=[f"converter reference script is missing: {script}"], ) command = [sys.executable, str(script), str(source_asset), "--probe"] try: completed = subprocess.run( command, capture_output=True, text=True, timeout=PROBE_TIMEOUT_SECONDS, check=False, ) except subprocess.TimeoutExpired: return ProbeResult( converter_skill=converter_skill, converter_tool=converter_skill, source_format="unknown", supported=False, errors=[f"converter probe timed out after {PROBE_TIMEOUT_SECONDS}s: {converter_skill}"], ) try: payload = json.loads(completed.stdout) except json.JSONDecodeError: detail = completed.stderr.strip() or completed.stdout.strip() or f"exit {completed.returncode}" return ProbeResult( converter_skill=converter_skill, converter_tool=converter_skill, source_format="unknown", supported=False, errors=[f"converter probe did not return JSON for {converter_skill}: {detail}"], ) return ProbeResult( converter_skill=str(payload.get("converter_skill") or converter_skill), converter_tool=str(payload.get("converter_tool") or converter_skill), source_format=str(payload.get("source_format") or "unknown"), supported=bool(payload.get("supported")), warnings=[str(warning) for warning in payload.get("warnings", [])], errors=[str(error) for error in payload.get("errors", [])], install_hint=str(payload.get("install_hint") or ""), ) def probe_warnings(probes: list[ProbeResult]) -> list[str]: warnings: list[str] = [] for probe in probes: status = "supported" if probe.supported else "not supported" warnings.append(f"Probe {probe.converter_skill}: {status} ({probe.source_format})") warnings.extend(f"{probe.converter_skill}: {warning}" for warning in probe.warnings) warnings.extend(f"{probe.converter_skill}: {error}" for error in probe.errors) return warnings def select_converter(source_asset: Path) -> tuple[str | None, ProbeResult | None, list[ProbeResult]]: probes = [run_probe(source_asset, converter_skill) for converter_skill in REFERENCE_ORDER] supported = [probe for probe in probes if probe.supported] if not supported: return None, None, probes selected = supported[0] return selected.converter_skill, selected, probes def source_relative_label(path: Path, root: Path) -> str: try: return path.relative_to(root).as_posix() except ValueError: return str(path) def supported_source_label(selection: SourceSelection, root: Path) -> str: label = source_relative_label(selection.source_asset, root) if selection.selected_probe is None: return f"`{label}` (existing USD)" return f"`{label}` ({selection.selected_probe.converter_skill}, {selection.selected_probe.source_format})" def report_with_warnings(report: ConversionReport, warnings: list[str]) -> ConversionReport: if not warnings: return report return replace(report, warnings=[*warnings, *report.warnings]) def directory_inspection_report(source_directory: Path, output_directory: Path, error: str) -> ConversionReport: return ConversionReport( source_asset_path=str(source_directory), source_format="unknown", converter_skill=SKILL, converter_tool="none", converter_command=[], output_directory=str(output_directory), output_usd_path="", generated_files=[], errors=[error], ) def unsupported_directory_report( source_directory: Path, output_directory: Path, inspected_count: int, probes: list[ProbeResult], ) -> ConversionReport: install_hint = next((probe.install_hint for probe in probes if probe.install_hint), "") if inspected_count == 0: detail = "no files" error = "directory source is empty; expected exactly one supported source file" else: detail = f"{inspected_count} file(s)" error = ( "directory source does not contain a supported source file among " f"{inspected_count} inspected file(s)" ) return ConversionReport( source_asset_path=str(source_directory), source_format="unknown", converter_skill=SKILL, converter_tool="none", converter_command=[], output_directory=str(output_directory), output_usd_path="", generated_files=[], warnings=[f"Inspected directory source `{source_directory}` and found {detail}."], errors=[error], install_hint=install_hint, ) def ambiguous_directory_report( source_directory: Path, output_directory: Path, selections: list[SourceSelection], ) -> ConversionReport: candidates = ", ".join(supported_source_label(selection, source_directory) for selection in selections) return ConversionReport( source_asset_path=str(source_directory), source_format="unknown", converter_skill=SKILL, converter_tool="none", converter_command=[], output_directory=str(output_directory), output_usd_path="", generated_files=[], errors=[ "directory source is ambiguous because multiple supported source files were found: " f"{candidates}. Pass one source file explicitly." ], ) def select_directory_source( source_directory: Path, output_directory: Path, ) -> tuple[SourceSelection | None, ConversionReport | None]: try: files = sorted( (path for path in source_directory.rglob("*") if path.is_file()), key=lambda path: source_relative_label(path, source_directory), ) except OSError as exc: return None, directory_inspection_report( source_directory, output_directory, f"could not inspect directory source: {exc}", ) selections: list[SourceSelection] = [] probes: list[ProbeResult] = [] for path in files: if is_existing_usd(path): selections.append(SourceSelection(source_asset=path)) continue _, selected_probe, source_probes = select_converter(path) probes.extend(source_probes) if selected_probe is not None: selections.append( SourceSelection(source_asset=path, selected_probe=selected_probe, probes=source_probes) ) if not selections: return None, unsupported_directory_report(source_directory, output_directory, len(files), probes) if len(selections) > 1: return None, ambiguous_directory_report(source_directory, output_directory, selections) selection = selections[0] label = source_relative_label(selection.source_asset, source_directory) warning = f"Directory source contained exactly one supported source file; selected `{label}` for conversion." return replace(selection, warnings=[warning]), None def run_converter(source_asset: Path, output_directory: Path, converter_skill: str) -> ConversionReport: script = reference_run_script(converter_skill) with tempfile.TemporaryDirectory(prefix="convert-to-usd-") as temp_dir: report_path = Path(temp_dir) / "conversion.json" command = [ sys.executable, str(script), str(source_asset), str(output_directory), "--report", str(report_path), ] try: completed = subprocess.run( command, capture_output=True, text=True, timeout=CONVERSION_TIMEOUT_SECONDS, check=False, ) except subprocess.TimeoutExpired: return ConversionReport( source_asset_path=str(source_asset), source_format="unknown", converter_skill=converter_skill, converter_reference=converter_skill, converter_tool=converter_skill, converter_command=command, output_directory=str(output_directory), output_usd_path="", generated_files=[], errors=[f"converter timed out after {CONVERSION_TIMEOUT_SECONDS}s: {converter_skill}"], ) if report_path.exists(): try: payload = json.loads(report_path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: return malformed_converter_report(source_asset, output_directory, converter_skill, command, f"invalid JSON report: {exc}") else: detail = completed.stderr.strip() or completed.stdout.strip() or f"exit {completed.returncode}" return malformed_converter_report(source_asset, output_directory, converter_skill, command, f"missing converter report: {detail}") return report_from_payload(payload, converter_skill) def malformed_converter_report( source_asset: Path, output_directory: Path, converter_skill: str, command: list[str], error: str, ) -> ConversionReport: return ConversionReport( source_asset_path=str(source_asset), source_format="unknown", converter_skill=converter_skill, converter_reference=converter_skill, converter_tool=converter_skill, converter_command=command, output_directory=str(output_directory), output_usd_path="", generated_files=[], errors=[error], ) def report_from_payload(payload: dict[str, Any], converter_skill: str) -> ConversionReport: return ConversionReport( source_asset_path=str(payload.get("source_asset_path") or ""), source_format=str(payload.get("source_format") or "unknown"), converter_skill=str(payload.get("converter_skill") or converter_skill), converter_reference=str(payload.get("converter_reference") or payload.get("converter_skill") or converter_skill), converter_tool=str(payload.get("converter_tool") or converter_skill), converter_command=[str(part) for part in payload.get("converter_command", [])], output_directory=str(payload.get("output_directory") or ""), output_usd_path=str(payload.get("output_usd_path") or ""), generated_files=[str(path) for path in payload.get("generated_files", [])], sidecar_inputs=[str(path) for path in payload.get("sidecar_inputs", [])], warnings=[str(warning) for warning in payload.get("warnings", [])], errors=[str(error) for error in payload.get("errors", [])], diagnostics=[diagnostic for diagnostic in payload.get("diagnostics", []) if isinstance(diagnostic, dict)], install_hint=str(payload.get("install_hint") or ""), next_step=str(payload.get("next_step") or NEXT_STEP), ) def append_selection_warnings(report: ConversionReport, selected: ProbeResult, probes: list[ProbeResult]) -> ConversionReport: supported = [probe.converter_skill for probe in probes if probe.supported] warnings = list(report.warnings) warnings.append(f"Router selected `{selected.converter_skill}` from upstream converter capability probes.") if len(supported) > 1: warnings.append( "Multiple converter references reported support; selected by converter-reference priority: " + ", ".join(supported) ) return ConversionReport( source_asset_path=report.source_asset_path, source_format=report.source_format, converter_skill=report.converter_skill, converter_reference=report.converter_reference, converter_tool=report.converter_tool, converter_command=report.converter_command, output_directory=report.output_directory, output_usd_path=report.output_usd_path, generated_files=report.generated_files, sidecar_inputs=report.sidecar_inputs, warnings=warnings, errors=report.errors, diagnostics=report.diagnostics, install_hint=report.install_hint, next_step=report.next_step, ) def unsupported_report(source_asset: Path, output_directory: Path, probes: list[ProbeResult]) -> ConversionReport: install_hint = next((probe.install_hint for probe in probes if probe.install_hint), "") return ConversionReport( source_asset_path=str(source_asset), source_format="unknown", converter_skill=SKILL, converter_tool="none", converter_command=[], output_directory=str(output_directory), output_usd_path="", generated_files=[], warnings=probe_warnings(probes), errors=["no converter reference reported support for this source asset"], install_hint=install_hint, ) def convert_to_usd(source_asset: Path, output_directory: Path) -> ConversionReport: source_asset = source_asset.resolve() output_directory = output_directory.resolve() directory_warnings: list[str] = [] selected_probe: ProbeResult | None = None probes: list[ProbeResult] = [] if not source_asset.exists(): return missing_source_report(source_asset, output_directory) if source_asset.is_dir(): selection, report = select_directory_source(source_asset, output_directory) if report is not None: return report if selection is None: return unsupported_directory_report(source_asset, output_directory, 0, []) source_asset = selection.source_asset directory_warnings = selection.warnings selected_probe = selection.selected_probe probes = selection.probes if is_existing_usd(source_asset): return report_with_warnings(already_usd_report(source_asset, output_directory), directory_warnings) if selected_probe is None: converter_skill, selected_probe, probes = select_converter(source_asset) else: converter_skill = selected_probe.converter_skill if converter_skill is None or selected_probe is None: return unsupported_report(source_asset, output_directory, probes) report = run_converter(source_asset, output_directory, converter_skill) return report_with_warnings(append_selection_warnings(report, selected_probe, probes), directory_warnings) def emit_report( report: ConversionReport, *, report_path: Path | None = None, markdown_report_path: Path | None = None, ) -> None: report_json = report.to_json() if report_path is not None: report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(report_json, encoding="utf-8") if markdown_report_path is not None: markdown_report_path.parent.mkdir(parents=True, exist_ok=True) markdown_report_path.write_text(report.to_markdown(), encoding="utf-8") print(report_json, end="") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Route a source asset by querying upstream converter references.") parser.add_argument("source_asset", type=Path) parser.add_argument("output_directory", type=Path) parser.add_argument("--report", type=Path, help="Write a JSON report to this path.") parser.add_argument("--markdown-report", type=Path, help="Write a Markdown report to this path.") args = parser.parse_args(argv) report = convert_to_usd(args.source_asset, args.output_directory) emit_report(report, report_path=args.report, markdown_report_path=args.markdown_report) return 0 if report.passed else 1 if __name__ == "__main__": raise SystemExit(main())