wehub-resource-sync
1279 lines
43 KiB
Python
1279 lines
43 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib
|
|
import inspect
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import re
|
|
import runpy
|
|
import statistics
|
|
import sys
|
|
import threading
|
|
import traceback
|
|
from contextlib import suppress
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from agent.feedback import build_compile_signal_bundle
|
|
from agent.models import CompileReport, CompileSignalBundle
|
|
from agent.mp_utils import get_mp_context
|
|
from agent.prompts import normalize_sdk_package
|
|
from sdk._core.v0.assets import activate_asset_session, asset_session_for_script
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_COMPILE_TARGETS = {"full", "visual"}
|
|
|
|
_EXCEPTION_PREFIX_RE = re.compile(r"^(?:[A-Za-z_][A-Za-z0-9_]*(?:Error|Exception)):\s*")
|
|
_VISUAL_OBJ_MESH_RE = re.compile(
|
|
r"<visual\b[\s\S]*?<mesh\b[^>]*filename=['\"][^'\"]+\.obj['\"]",
|
|
re.IGNORECASE,
|
|
)
|
|
_GEOMETRY_QC_MARKERS = (
|
|
"isolated parts detected",
|
|
"geometry overlap check reported overlaps",
|
|
"mesh connectivity check failed",
|
|
"visual connectivity check failed",
|
|
"fail_if_parts_overlap_in_sampled_poses(",
|
|
"expect_contact(",
|
|
"expect_gap(",
|
|
"expect_overlap(",
|
|
"expect_within(",
|
|
"expect_aabb_",
|
|
"expect_xy_distance",
|
|
"expect_above",
|
|
"expect_joint_motion_axis",
|
|
"urdf tests failed:",
|
|
)
|
|
_AUTOMATED_BASELINE_WARNING_CHECK_NAME = (
|
|
"warn_if_part_contains_disconnected_geometry_islands(tol=1e-06)"
|
|
)
|
|
_AUTOMATED_BASELINE_DEFAULT_CHECK_NAMES = frozenset(
|
|
{
|
|
"check_model_valid",
|
|
"check_single_root_part",
|
|
"check_mesh_assets_ready",
|
|
"fail_if_isolated_parts()",
|
|
_AUTOMATED_BASELINE_WARNING_CHECK_NAME,
|
|
"fail_if_parts_overlap_in_current_pose()",
|
|
}
|
|
)
|
|
_MODEL_EXECUTION_LOCK = threading.Lock()
|
|
|
|
|
|
def _import_sdk_module(sdk_package: str, module_suffix: str = "") -> Any:
|
|
package = normalize_sdk_package(sdk_package)
|
|
return importlib.import_module(f"{package}{module_suffix}")
|
|
|
|
|
|
def _env_float(name: str, default: float) -> float:
|
|
raw = os.environ.get(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return float(raw.strip())
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def compile_urdf(
|
|
script_path: Path,
|
|
*,
|
|
sdk_package: str = "sdk",
|
|
run_checks: bool = True,
|
|
ignore_geom_qc: bool = False,
|
|
target: str = "full",
|
|
rewrite_visual_glb: bool | None = None,
|
|
) -> str:
|
|
"""Execute a generated script and return the exported XML payload."""
|
|
report = compile_urdf_report_maybe_timeout(
|
|
script_path,
|
|
sdk_package=sdk_package,
|
|
run_checks=run_checks,
|
|
ignore_geom_qc=ignore_geom_qc,
|
|
target=target,
|
|
rewrite_visual_glb=rewrite_visual_glb,
|
|
)
|
|
for warning in report.warnings:
|
|
logger.warning("%s", warning)
|
|
return report.urdf_xml
|
|
|
|
|
|
def _normalize_compile_target(target: str) -> str:
|
|
target_key = str(target).strip().lower()
|
|
if target_key not in _COMPILE_TARGETS:
|
|
supported = ", ".join(sorted(_COMPILE_TARGETS))
|
|
raise ValueError(f"Unsupported compile target {target!r}. Expected one of: {supported}")
|
|
return target_key
|
|
|
|
|
|
def _extract_urdf_xml(
|
|
globals_dict: dict,
|
|
*,
|
|
sdk_package: str = "sdk",
|
|
target: str = "full",
|
|
validate_export: bool = True,
|
|
suppress_exceptions: bool = False,
|
|
) -> str | None:
|
|
target_key = _normalize_compile_target(target)
|
|
object_model = globals_dict.get("object_model")
|
|
urdf_xml: str | None = None
|
|
export_exc: Exception | None = None
|
|
try:
|
|
if object_model is not None:
|
|
compile_object_to_urdf_xml = getattr(
|
|
_import_sdk_module(sdk_package, ".v0._urdf_export"),
|
|
"compile_object_to_urdf_xml",
|
|
)
|
|
|
|
script_dir = globals_dict.get("__file__")
|
|
asset_root = Path(script_dir).resolve().parent if isinstance(script_dir, str) else None
|
|
try:
|
|
params = inspect.signature(compile_object_to_urdf_xml).parameters
|
|
except Exception:
|
|
params = {}
|
|
if "asset_root" in params:
|
|
urdf_xml = compile_object_to_urdf_xml(
|
|
object_model,
|
|
asset_root=asset_root,
|
|
include_physical_collisions=target_key != "visual",
|
|
validate=validate_export,
|
|
)
|
|
else:
|
|
urdf_xml = compile_object_to_urdf_xml(object_model)
|
|
globals_dict["urdf_xml"] = urdf_xml
|
|
except Exception as exc:
|
|
export_exc = exc
|
|
urdf_xml = None
|
|
|
|
if not isinstance(urdf_xml, str):
|
|
maybe_urdf_xml = globals_dict.get("urdf_xml")
|
|
if isinstance(maybe_urdf_xml, str):
|
|
urdf_xml = maybe_urdf_xml
|
|
if not isinstance(urdf_xml, str) and export_exc is not None and not suppress_exceptions:
|
|
raise export_exc
|
|
return urdf_xml
|
|
|
|
|
|
def _attach_compiled_urdf_on_failure(
|
|
exc: BaseException,
|
|
*,
|
|
urdf_xml: str | None,
|
|
warnings: list[str],
|
|
signal_bundle: CompileSignalBundle,
|
|
) -> BaseException:
|
|
wrapped = RuntimeError(f"{type(exc).__name__}: {exc}")
|
|
if isinstance(urdf_xml, str) and urdf_xml.strip():
|
|
setattr(wrapped, "compiled_urdf_xml", urdf_xml)
|
|
setattr(wrapped, "warnings", list(warnings))
|
|
test_report = getattr(exc, "test_report", None)
|
|
if test_report is not None:
|
|
setattr(wrapped, "test_report", test_report)
|
|
setattr(wrapped, "compile_signal_bundle", signal_bundle)
|
|
return wrapped
|
|
|
|
|
|
def _strip_exception_prefixes(message: str) -> str:
|
|
stripped = message.strip()
|
|
while True:
|
|
match = _EXCEPTION_PREFIX_RE.match(stripped)
|
|
if match is None:
|
|
return stripped
|
|
stripped = stripped[match.end() :].lstrip()
|
|
|
|
|
|
def _nonblocking_geometry_qc_warning_from_exception(exc: BaseException) -> str | None:
|
|
compiled_urdf_xml = getattr(exc, "compiled_urdf_xml", None)
|
|
if not isinstance(compiled_urdf_xml, str) or not compiled_urdf_xml.strip():
|
|
return None
|
|
|
|
message = _strip_exception_prefixes(str(exc))
|
|
lowered = message.lower()
|
|
if not any(marker in lowered for marker in _GEOMETRY_QC_MARKERS):
|
|
return None
|
|
|
|
if message.startswith("URDF compile failure ("):
|
|
header_end = message.find("):")
|
|
if header_end != -1:
|
|
context = message[len("URDF compile failure (") : header_end]
|
|
detail = message[header_end + 2 :].lstrip(": ").lstrip()
|
|
geometry_label = context.split(",", 1)[0].strip() or "geometry"
|
|
return f"URDF compile warning ({geometry_label}, non-blocking): {detail}"
|
|
|
|
return f"URDF compile warning (non-blocking): {message}"
|
|
|
|
|
|
def compile_urdf_report(
|
|
script_path: Path,
|
|
*,
|
|
sdk_package: str = "sdk",
|
|
run_checks: bool = True,
|
|
ignore_geom_qc: bool = False,
|
|
target: str = "full",
|
|
rewrite_visual_glb: bool | None = None,
|
|
) -> CompileReport:
|
|
session = asset_session_for_script(script_path)
|
|
with activate_asset_session(session):
|
|
return _compile_urdf_report_impl(
|
|
script_path,
|
|
sdk_package=sdk_package,
|
|
run_checks=run_checks,
|
|
ignore_geom_qc=ignore_geom_qc,
|
|
target=target,
|
|
rewrite_visual_glb=rewrite_visual_glb,
|
|
)
|
|
|
|
|
|
def _compile_urdf_report_impl(
|
|
script_path: Path,
|
|
*,
|
|
sdk_package: str = "sdk",
|
|
run_checks: bool = True,
|
|
ignore_geom_qc: bool = False,
|
|
target: str = "full",
|
|
rewrite_visual_glb: bool | None = None,
|
|
) -> CompileReport:
|
|
"""Execute a generated script and return export XML plus non-blocking warnings."""
|
|
globals_dict = load_model_globals(script_path, sdk_package=sdk_package)
|
|
warnings: list[str] = []
|
|
test_report = None
|
|
target_key = _normalize_compile_target(target)
|
|
script_path = script_path.resolve()
|
|
if run_checks:
|
|
try:
|
|
if target_key == "full":
|
|
authored_report = _run_required_tests(
|
|
globals_dict,
|
|
sdk_package=sdk_package,
|
|
)
|
|
baseline_report = _run_compiler_owned_baseline_tests(
|
|
globals_dict,
|
|
script_path=script_path,
|
|
sdk_package=sdk_package,
|
|
authored_report=authored_report,
|
|
)
|
|
test_report = _merge_test_reports(
|
|
authored_report,
|
|
baseline_report,
|
|
sdk_package=sdk_package,
|
|
)
|
|
_raise_for_failed_test_report(test_report)
|
|
except Exception as exc:
|
|
urdf_xml = _extract_urdf_xml(
|
|
globals_dict,
|
|
sdk_package=sdk_package,
|
|
target=target_key,
|
|
suppress_exceptions=True,
|
|
)
|
|
signal_bundle = build_compile_signal_bundle(
|
|
status="failure",
|
|
warnings=warnings,
|
|
test_report=getattr(exc, "test_report", None),
|
|
exc=exc,
|
|
)
|
|
wrapped = _attach_compiled_urdf_on_failure(
|
|
exc,
|
|
urdf_xml=urdf_xml,
|
|
warnings=warnings,
|
|
signal_bundle=signal_bundle,
|
|
)
|
|
if ignore_geom_qc:
|
|
nonblocking_warning = _nonblocking_geometry_qc_warning_from_exception(wrapped)
|
|
compiled_urdf_xml = getattr(wrapped, "compiled_urdf_xml", None)
|
|
if (
|
|
nonblocking_warning is not None
|
|
and isinstance(compiled_urdf_xml, str)
|
|
and compiled_urdf_xml.strip()
|
|
):
|
|
warning_lines = list(warnings)
|
|
warning_lines.append(nonblocking_warning)
|
|
return CompileReport(
|
|
urdf_xml=compiled_urdf_xml,
|
|
warnings=warning_lines,
|
|
signal_bundle=build_compile_signal_bundle(
|
|
status="success",
|
|
warnings=warning_lines,
|
|
test_report=getattr(wrapped, "test_report", None),
|
|
),
|
|
)
|
|
raise wrapped from exc
|
|
|
|
warnings.extend(str(item) for item in getattr(test_report, "warnings", ()) or ())
|
|
|
|
urdf_xml = _extract_urdf_xml(
|
|
globals_dict,
|
|
sdk_package=sdk_package,
|
|
target=target_key,
|
|
validate_export=not (run_checks and target_key == "full"),
|
|
)
|
|
if not isinstance(urdf_xml, str):
|
|
raise ValueError("object_model must compile into an exportable XML payload")
|
|
if _should_rewrite_visual_meshes_to_glb(
|
|
sdk_package=sdk_package,
|
|
rewrite_visual_glb=rewrite_visual_glb,
|
|
):
|
|
urdf_xml = rewrite_visual_meshes_to_glb(
|
|
urdf_xml,
|
|
sdk_package=sdk_package,
|
|
asset_root=script_path.parent,
|
|
warnings=warnings,
|
|
)
|
|
signal_bundle = build_compile_signal_bundle(
|
|
status="success",
|
|
warnings=warnings,
|
|
test_report=test_report,
|
|
)
|
|
return CompileReport(
|
|
urdf_xml=urdf_xml,
|
|
warnings=warnings,
|
|
signal_bundle=signal_bundle,
|
|
)
|
|
|
|
|
|
def rewrite_visual_meshes_to_glb(
|
|
urdf_xml: str,
|
|
*,
|
|
sdk_package: str,
|
|
asset_root: Path,
|
|
warnings: list[str],
|
|
) -> str:
|
|
if _VISUAL_OBJ_MESH_RE.search(urdf_xml) is None:
|
|
return urdf_xml
|
|
try:
|
|
convert_urdf_visual_meshes_to_glb = getattr(
|
|
_import_sdk_module(sdk_package, ".v0.viewer_assets"),
|
|
"convert_urdf_visual_meshes_to_glb",
|
|
)
|
|
converted_xml, conversion_warnings = convert_urdf_visual_meshes_to_glb(
|
|
urdf_xml,
|
|
asset_root=asset_root,
|
|
)
|
|
except Exception as exc:
|
|
warnings.append(f"Viewer mesh conversion warning: {exc}")
|
|
return urdf_xml
|
|
|
|
warnings.extend(str(item) for item in conversion_warnings)
|
|
return converted_xml
|
|
|
|
|
|
def _should_rewrite_visual_meshes_to_glb(
|
|
*,
|
|
sdk_package: str,
|
|
rewrite_visual_glb: bool | None,
|
|
) -> bool:
|
|
if rewrite_visual_glb is not None:
|
|
return rewrite_visual_glb
|
|
return False
|
|
|
|
|
|
def load_model_globals(
|
|
script_path: Path,
|
|
*,
|
|
sdk_package: str = "sdk",
|
|
) -> dict[str, Any]:
|
|
"""Execute a model script and return its globals."""
|
|
normalize_sdk_package(sdk_package)
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
script_path = script_path.resolve()
|
|
with _MODEL_EXECUTION_LOCK:
|
|
prev_cwd = Path.cwd()
|
|
os.chdir(script_path.parent)
|
|
sys.path.insert(0, str(repo_root))
|
|
try:
|
|
globals_dict = runpy.run_path(script_path.name)
|
|
finally:
|
|
os.chdir(prev_cwd)
|
|
if sys.path and sys.path[0] == str(repo_root):
|
|
sys.path.pop(0)
|
|
return globals_dict
|
|
|
|
|
|
def _warn_cwd_relative_asset_paths(*, script_path: Path, warnings: list[str]) -> None:
|
|
"""
|
|
Emit a non-blocking warning for path anti-patterns that make outputs depend on current cwd.
|
|
|
|
These scripts are expected to write assets relative to script location, not process cwd.
|
|
"""
|
|
try:
|
|
source = script_path.read_text(encoding="utf-8")
|
|
except Exception:
|
|
return
|
|
|
|
findings: list[str] = []
|
|
|
|
if re.search(r'^\s*HERE\s*=\s*Path\(\s*["\']\.\s*["\']\s*\)', source, re.MULTILINE):
|
|
findings.append("Detected `HERE = Path('.')` assignment.")
|
|
if re.search(r'asset_root\s*=\s*["\']\.\s*["\']', source):
|
|
findings.append("Detected `asset_root='.'` usage.")
|
|
if re.search(r'asset_root\s*=\s*Path\(\s*["\']\.\s*["\']\s*\)', source):
|
|
findings.append("Detected `asset_root=Path('.')` usage.")
|
|
|
|
if not findings:
|
|
return
|
|
|
|
warnings.append(
|
|
"URDF compile warning (non-blocking): cwd-relative asset paths detected.\n"
|
|
+ "\n".join(f"- {item}" for item in findings)
|
|
+ "\nUse managed mesh helpers instead: `mesh_from_geometry(..., name='part_name')`, "
|
|
"`mesh_from_input('existing_mesh')`, `mesh_from_cadquery(..., name='part_name')`, "
|
|
"and `TestContext(object_model)`."
|
|
)
|
|
|
|
|
|
def _env_int(name: str, default: int) -> int:
|
|
raw = os.environ.get(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return int(raw.strip())
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def _iter_model_links(object_model: object) -> list[object]:
|
|
links = getattr(object_model, "links", None)
|
|
if isinstance(links, list):
|
|
return links
|
|
parts = getattr(object_model, "parts", None)
|
|
if isinstance(parts, list):
|
|
return parts
|
|
return []
|
|
|
|
|
|
def _format_dims(dims: tuple[float, float, float]) -> str:
|
|
return "(" + ", ".join(f"{float(v):.4g}" for v in dims) + ")m"
|
|
|
|
|
|
def _warn_geometry_scale_anomalies(
|
|
globals_dict: dict,
|
|
*,
|
|
script_dir: Path,
|
|
warnings: list[str],
|
|
sdk_package: str = "sdk",
|
|
) -> None:
|
|
if os.environ.get("URDF_DISABLE_IMPORTANT_GEOMETRY_WARNINGS") in {"1", "true", "TRUE"}:
|
|
return
|
|
|
|
object_model = globals_dict.get("object_model")
|
|
if object_model is None:
|
|
return
|
|
|
|
links = _iter_model_links(object_model)
|
|
if not links:
|
|
return
|
|
|
|
absurd_dim_max = float(os.environ.get("URDF_IMPORTANT_GEOMETRY_MAX_DIM", "1000.0"))
|
|
outlier_ratio = float(os.environ.get("URDF_IMPORTANT_GEOMETRY_OUTLIER_RATIO", "100.0"))
|
|
outlier_abs_min = float(os.environ.get("URDF_IMPORTANT_GEOMETRY_OUTLIER_ABS_MIN", "10.0"))
|
|
max_findings = _env_int("URDF_IMPORTANT_GEOMETRY_MAX_FINDINGS", 10)
|
|
|
|
absurd_findings: list[str] = []
|
|
span_records: list[tuple[str, str, int, str, tuple[float, float, float], float]] = []
|
|
|
|
for link in links:
|
|
link_name = getattr(link, "name", None)
|
|
if not isinstance(link_name, str) or not link_name:
|
|
continue
|
|
|
|
for source_name in ("visuals", "collisions"):
|
|
items = getattr(link, source_name, None)
|
|
if not isinstance(items, list) or not items:
|
|
continue
|
|
|
|
for index, item in enumerate(items):
|
|
geometry = getattr(item, "geometry", None)
|
|
if geometry is None:
|
|
continue
|
|
try:
|
|
local_min, local_max = _geometry_local_aabb(
|
|
geometry,
|
|
script_dir=script_dir,
|
|
sdk_package=sdk_package,
|
|
)
|
|
except Exception:
|
|
continue
|
|
|
|
dims = (
|
|
float(local_max[0] - local_min[0]),
|
|
float(local_max[1] - local_min[1]),
|
|
float(local_max[2] - local_min[2]),
|
|
)
|
|
geom_type = type(geometry).__name__
|
|
max_dim = max(abs(d) for d in dims)
|
|
|
|
has_non_finite = any(not math.isfinite(v) for v in (*local_min, *local_max, *dims))
|
|
if has_non_finite or max_dim > absurd_dim_max:
|
|
reason = (
|
|
"non-finite dimensions" if has_non_finite else f"max_dim={max_dim:.4g}m"
|
|
)
|
|
absurd_findings.append(
|
|
f"- link={link_name!r} source={source_name[:-1]!r} index={index} "
|
|
f"geometry={geom_type!r} dims={_format_dims(dims)} reason={reason}"
|
|
)
|
|
|
|
if all(math.isfinite(v) for v in dims) and max_dim > 0.0:
|
|
span_records.append(
|
|
(link_name, source_name[:-1], index, geom_type, dims, max_dim)
|
|
)
|
|
|
|
if absurd_findings:
|
|
preview = "\n".join(absurd_findings[:max_findings])
|
|
more = (
|
|
""
|
|
if len(absurd_findings) <= max_findings
|
|
else f"\n... ({len(absurd_findings) - max_findings} more)"
|
|
)
|
|
warnings.append(
|
|
"IMPORTANT: URDF compile warning (non-blocking): non-finite or absurd geometry dimensions detected.\n"
|
|
f"{preview}{more}\n"
|
|
"These dimensions are likely numerically broken and will likely look wrong in the viewer. "
|
|
"Check for sign errors, bad normalization denominators, unit mistakes, or runaway procedural geometry."
|
|
)
|
|
|
|
if not span_records:
|
|
return
|
|
|
|
spans = [record[5] for record in span_records]
|
|
median_span = float(statistics.median(spans))
|
|
if not math.isfinite(median_span) or median_span <= 0.0:
|
|
return
|
|
|
|
outlier_findings: list[str] = []
|
|
for link_name, source_name, index, geom_type, dims, max_dim in span_records:
|
|
if max_dim < outlier_abs_min:
|
|
continue
|
|
ratio = max_dim / max(median_span, 1e-9)
|
|
if ratio < outlier_ratio:
|
|
continue
|
|
outlier_findings.append(
|
|
f"- link={link_name!r} source={source_name!r} index={index} geometry={geom_type!r} "
|
|
f"dims={_format_dims(dims)} max_dim={max_dim:.4g}m median_dim={median_span:.4g}m ratio={ratio:.4g}x"
|
|
)
|
|
|
|
if outlier_findings:
|
|
preview = "\n".join(outlier_findings[:max_findings])
|
|
more = (
|
|
""
|
|
if len(outlier_findings) <= max_findings
|
|
else f"\n... ({len(outlier_findings) - max_findings} more)"
|
|
)
|
|
warnings.append(
|
|
"IMPORTANT: URDF compile warning (non-blocking): geometry outlier dimensions detected.\n"
|
|
f"{preview}{more}\n"
|
|
"One or more members are dramatically larger than the rest of the object and are likely malformed. "
|
|
"Check for bad interpolation, reversed spans, or accidental scale explosions in authored geometry."
|
|
)
|
|
|
|
|
|
def _compile_worker(
|
|
script_path_str: str,
|
|
sdk_package: str,
|
|
run_checks: bool,
|
|
ignore_geom_qc: bool,
|
|
target: str,
|
|
rewrite_visual_glb: bool,
|
|
conn: object,
|
|
) -> None:
|
|
try:
|
|
report = compile_urdf_report(
|
|
Path(script_path_str),
|
|
sdk_package=sdk_package,
|
|
run_checks=run_checks,
|
|
ignore_geom_qc=ignore_geom_qc,
|
|
target=target,
|
|
rewrite_visual_glb=rewrite_visual_glb,
|
|
)
|
|
payload = {
|
|
"ok": True,
|
|
"urdf_xml": report.urdf_xml,
|
|
"warnings": report.warnings,
|
|
"signal_bundle": report.signal_bundle.to_dict(),
|
|
}
|
|
conn.send(payload) # type: ignore[attr-defined]
|
|
except BaseException as exc:
|
|
payload = {
|
|
"ok": False,
|
|
"error": f"{type(exc).__name__}: {exc}",
|
|
"error_type": type(exc).__name__,
|
|
"traceback": traceback.format_exc(),
|
|
}
|
|
compiled_urdf_xml = getattr(exc, "compiled_urdf_xml", None)
|
|
if isinstance(compiled_urdf_xml, str) and compiled_urdf_xml.strip():
|
|
payload["compiled_urdf_xml"] = compiled_urdf_xml
|
|
warnings = getattr(exc, "warnings", None)
|
|
if isinstance(warnings, list):
|
|
payload["warnings"] = [str(w) for w in warnings]
|
|
signal_bundle = getattr(exc, "compile_signal_bundle", None)
|
|
if isinstance(signal_bundle, CompileSignalBundle):
|
|
payload["signal_bundle"] = signal_bundle.to_dict()
|
|
conn.send(payload) # type: ignore[attr-defined]
|
|
finally:
|
|
with suppress(Exception):
|
|
conn.close() # type: ignore[attr-defined]
|
|
|
|
|
|
def compile_urdf_report_maybe_timeout(
|
|
script_path: Path,
|
|
*,
|
|
sdk_package: str = "sdk",
|
|
run_checks: bool = True,
|
|
ignore_geom_qc: bool = False,
|
|
target: str = "full",
|
|
rewrite_visual_glb: bool | None = None,
|
|
) -> CompileReport:
|
|
"""
|
|
Run `compile_urdf_report` with a hard timeout to prevent indefinite hangs.
|
|
|
|
Controlled by `URDF_COMPILE_TIMEOUT_SECONDS` (default: 300). Set to 0 to disable.
|
|
"""
|
|
timeout_seconds = float(_env_float("URDF_COMPILE_TIMEOUT_SECONDS", 300.0))
|
|
if timeout_seconds <= 0:
|
|
return compile_urdf_report(
|
|
script_path,
|
|
sdk_package=sdk_package,
|
|
run_checks=run_checks,
|
|
ignore_geom_qc=ignore_geom_qc,
|
|
target=target,
|
|
rewrite_visual_glb=rewrite_visual_glb,
|
|
)
|
|
|
|
ctx = get_mp_context()
|
|
parent_conn, child_conn = ctx.Pipe(duplex=False)
|
|
proc = ctx.Process(
|
|
target=_compile_worker,
|
|
args=(
|
|
str(script_path),
|
|
sdk_package,
|
|
run_checks,
|
|
ignore_geom_qc,
|
|
target,
|
|
rewrite_visual_glb,
|
|
child_conn,
|
|
),
|
|
daemon=True,
|
|
)
|
|
proc.start()
|
|
try:
|
|
with suppress(Exception):
|
|
child_conn.close()
|
|
|
|
if parent_conn.poll(timeout_seconds):
|
|
msg = parent_conn.recv()
|
|
else:
|
|
with suppress(Exception):
|
|
proc.terminate()
|
|
proc.join(timeout=2.0)
|
|
raise TimeoutError(
|
|
f"URDF compile timed out after {timeout_seconds:.0f}s. "
|
|
"This can happen if the generated script contains a long-running loop, "
|
|
"expensive mesh processing, or very slow overlap checks. "
|
|
"To adjust: set URDF_COMPILE_TIMEOUT_SECONDS, or reduce/disable overlap checks "
|
|
"(URDF_GEOMETRY_OVERLAP_MAX_SAMPLES / URDF_DISABLE_GEOMETRY_OVERLAP_CHECK=1)."
|
|
)
|
|
finally:
|
|
with suppress(Exception):
|
|
parent_conn.close()
|
|
|
|
proc.join(timeout=2.0)
|
|
if proc.is_alive():
|
|
with suppress(Exception):
|
|
proc.terminate()
|
|
proc.join(timeout=2.0)
|
|
|
|
if not isinstance(msg, dict):
|
|
raise RuntimeError(
|
|
f"URDF compile failed: worker returned non-dict payload ({type(msg).__name__})"
|
|
)
|
|
if msg.get("ok") is True:
|
|
urdf_xml = msg.get("urdf_xml")
|
|
warnings = msg.get("warnings")
|
|
signal_bundle_payload = msg.get("signal_bundle")
|
|
if not isinstance(urdf_xml, str):
|
|
raise RuntimeError("URDF compile failed: missing urdf_xml from worker")
|
|
if not isinstance(warnings, list):
|
|
warnings = []
|
|
if isinstance(signal_bundle_payload, dict):
|
|
signal_bundle = CompileSignalBundle.from_dict(signal_bundle_payload)
|
|
else:
|
|
signal_bundle = build_compile_signal_bundle(status="success", warnings=warnings)
|
|
return CompileReport(
|
|
urdf_xml=urdf_xml,
|
|
warnings=[str(w) for w in warnings],
|
|
signal_bundle=signal_bundle,
|
|
)
|
|
|
|
error_text = str(msg.get("error", "Unknown compile worker error")).strip()
|
|
error_type = str(msg.get("error_type", "")).strip()
|
|
tb_text = str(msg.get("traceback", "")).strip()
|
|
if tb_text:
|
|
logger.debug("Compile worker traceback:\n%s", tb_text)
|
|
exc = RuntimeError(error_text or "Unknown compile worker error")
|
|
if error_type:
|
|
setattr(exc, "remote_error_type", error_type)
|
|
if tb_text:
|
|
setattr(exc, "remote_traceback", tb_text)
|
|
compiled_urdf_xml = msg.get("compiled_urdf_xml")
|
|
if isinstance(compiled_urdf_xml, str) and compiled_urdf_xml.strip():
|
|
setattr(exc, "compiled_urdf_xml", compiled_urdf_xml)
|
|
warnings = msg.get("warnings")
|
|
if isinstance(warnings, list):
|
|
setattr(exc, "warnings", [str(w) for w in warnings])
|
|
signal_bundle_payload = msg.get("signal_bundle")
|
|
if isinstance(signal_bundle_payload, dict):
|
|
setattr(
|
|
exc,
|
|
"compile_signal_bundle",
|
|
CompileSignalBundle.from_dict(signal_bundle_payload),
|
|
)
|
|
raise exc
|
|
|
|
|
|
def _validate_mesh_connectivity(
|
|
globals_dict: dict,
|
|
*,
|
|
script_dir: Path,
|
|
sdk_package: str = "sdk",
|
|
) -> None:
|
|
if os.environ.get("URDF_DISABLE_MESH_CONNECTIVITY_CHECK") in {"1", "true", "TRUE"}:
|
|
return
|
|
|
|
tol = float(os.environ.get("URDF_MESH_CONNECTIVITY_TOL", "0.02"))
|
|
object_model = globals_dict.get("object_model")
|
|
if object_model is None:
|
|
return
|
|
|
|
find_joint_origin_distance_findings = getattr(
|
|
_import_sdk_module("sdk", "._core.v0.geometry_qc"),
|
|
"find_joint_origin_distance_findings",
|
|
)
|
|
findings = find_joint_origin_distance_findings(
|
|
object_model,
|
|
asset_root=script_dir,
|
|
tol=tol,
|
|
)
|
|
if not findings:
|
|
return
|
|
|
|
first = findings[0]
|
|
raise ValueError(
|
|
"Visual connectivity check failed: articulation origin is far from exact visual geometry. "
|
|
f"joint={first.joint!r} parent={first.parent!r} child={first.child!r} "
|
|
f"dist_parent={first.parent_distance:.4g} dist_child={first.child_distance:.4g} "
|
|
f"tol={first.tol:.4g}. "
|
|
"Fix by moving the joint origin and/or adjusting authored visuals so the parts touch where the joint is mounted."
|
|
)
|
|
|
|
|
|
def _geometry_local_aabb(
|
|
geometry: object,
|
|
*,
|
|
script_dir: Path,
|
|
sdk_package: str = "sdk",
|
|
) -> tuple[tuple[float, float, float], tuple[float, float, float]]:
|
|
sdk_mod = _import_sdk_module(sdk_package)
|
|
box = getattr(sdk_mod, "Box")
|
|
cylinder = getattr(sdk_mod, "Cylinder")
|
|
mesh = getattr(sdk_mod, "Mesh")
|
|
sphere = getattr(sdk_mod, "Sphere")
|
|
|
|
if isinstance(geometry, box):
|
|
sx, sy, sz = geometry.size
|
|
return (-sx / 2.0, -sy / 2.0, -sz / 2.0), (sx / 2.0, sy / 2.0, sz / 2.0)
|
|
if isinstance(geometry, cylinder):
|
|
radius = float(geometry.radius)
|
|
length = float(geometry.length)
|
|
return (-radius, -radius, -length / 2.0), (radius, radius, length / 2.0)
|
|
if isinstance(geometry, sphere):
|
|
radius = float(geometry.radius)
|
|
return (-radius, -radius, -radius), (radius, radius, radius)
|
|
if isinstance(geometry, mesh):
|
|
filename = getattr(geometry, "filename", "")
|
|
if not isinstance(filename, str) or not filename:
|
|
raise ValueError("Mesh geometry filename is missing")
|
|
mesh_path = (script_dir / filename).resolve()
|
|
if not mesh_path.exists():
|
|
raise ValueError(f"Mesh file not found: {mesh_path}")
|
|
local_min, local_max = _obj_aabb(mesh_path)
|
|
scale = getattr(geometry, "scale", None)
|
|
if scale:
|
|
sx, sy, sz = scale
|
|
local_min = (local_min[0] * sx, local_min[1] * sy, local_min[2] * sz)
|
|
local_max = (local_max[0] * sx, local_max[1] * sy, local_max[2] * sz)
|
|
return local_min, local_max
|
|
|
|
raise ValueError(f"Unsupported geometry type for connectivity check: {type(geometry).__name__}")
|
|
|
|
|
|
def _obj_aabb(path: Path) -> tuple[tuple[float, float, float], tuple[float, float, float]]:
|
|
min_x = min_y = min_z = float("inf")
|
|
max_x = max_y = max_z = float("-inf")
|
|
found = False
|
|
for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
|
if not line.startswith("v "):
|
|
continue
|
|
parts = line.split()
|
|
if len(parts) < 4:
|
|
continue
|
|
try:
|
|
x = float(parts[1])
|
|
y = float(parts[2])
|
|
z = float(parts[3])
|
|
except ValueError:
|
|
continue
|
|
found = True
|
|
min_x = min(min_x, x)
|
|
min_y = min(min_y, y)
|
|
min_z = min(min_z, z)
|
|
max_x = max(max_x, x)
|
|
max_y = max(max_y, y)
|
|
max_z = max(max_z, z)
|
|
|
|
if not found:
|
|
raise ValueError(f"OBJ contains no vertices: {path}")
|
|
return (min_x, min_y, min_z), (max_x, max_y, max_z)
|
|
|
|
|
|
def _transform_aabb(
|
|
local_min: tuple[float, float, float],
|
|
local_max: tuple[float, float, float],
|
|
*,
|
|
origin: object,
|
|
) -> tuple[tuple[float, float, float], tuple[float, float, float]]:
|
|
xyz = getattr(origin, "xyz", (0.0, 0.0, 0.0)) if origin is not None else (0.0, 0.0, 0.0)
|
|
rpy = getattr(origin, "rpy", (0.0, 0.0, 0.0)) if origin is not None else (0.0, 0.0, 0.0)
|
|
ox, oy, oz = (float(xyz[0]), float(xyz[1]), float(xyz[2]))
|
|
rr, rp, ry = (float(rpy[0]), float(rpy[1]), float(rpy[2]))
|
|
rot = _rpy_matrix(rr, rp, ry)
|
|
|
|
corners = []
|
|
for x in (local_min[0], local_max[0]):
|
|
for y in (local_min[1], local_max[1]):
|
|
for z in (local_min[2], local_max[2]):
|
|
corners.append((x, y, z))
|
|
|
|
min_x = min_y = min_z = float("inf")
|
|
max_x = max_y = max_z = float("-inf")
|
|
for x, y, z in corners:
|
|
tx, ty, tz = _mat_vec(rot, (x, y, z))
|
|
tx += ox
|
|
ty += oy
|
|
tz += oz
|
|
min_x = min(min_x, tx)
|
|
min_y = min(min_y, ty)
|
|
min_z = min(min_z, tz)
|
|
max_x = max(max_x, tx)
|
|
max_y = max(max_y, ty)
|
|
max_z = max(max_z, tz)
|
|
|
|
return (min_x, min_y, min_z), (max_x, max_y, max_z)
|
|
|
|
|
|
def _rpy_matrix(roll: float, pitch: float, yaw: float) -> tuple[tuple[float, float, float], ...]:
|
|
cr = math.cos(roll)
|
|
sr = math.sin(roll)
|
|
cp = math.cos(pitch)
|
|
sp = math.sin(pitch)
|
|
cy = math.cos(yaw)
|
|
sy = math.sin(yaw)
|
|
return (
|
|
(cy * cp, cy * sp * sr - sy * cr, cy * sp * cr + sy * sr),
|
|
(sy * cp, sy * sp * sr + cy * cr, sy * sp * cr - cy * sr),
|
|
(-sp, cp * sr, cp * cr),
|
|
)
|
|
|
|
|
|
def _mat_vec(
|
|
mat: tuple[tuple[float, float, float], ...],
|
|
vec: tuple[float, float, float],
|
|
) -> tuple[float, float, float]:
|
|
x, y, z = vec
|
|
return (
|
|
mat[0][0] * x + mat[0][1] * y + mat[0][2] * z,
|
|
mat[1][0] * x + mat[1][1] * y + mat[1][2] * z,
|
|
mat[2][0] * x + mat[2][1] * y + mat[2][2] * z,
|
|
)
|
|
|
|
|
|
def _point_aabb_distance(
|
|
point: tuple[float, float, float],
|
|
aabb: tuple[tuple[float, float, float], tuple[float, float, float]],
|
|
) -> float:
|
|
(min_x, min_y, min_z), (max_x, max_y, max_z) = aabb
|
|
px, py, pz = point
|
|
dx = 0.0
|
|
if px < min_x:
|
|
dx = min_x - px
|
|
elif px > max_x:
|
|
dx = px - max_x
|
|
|
|
dy = 0.0
|
|
if py < min_y:
|
|
dy = min_y - py
|
|
elif py > max_y:
|
|
dy = py - max_y
|
|
|
|
dz = 0.0
|
|
if pz < min_z:
|
|
dz = min_z - pz
|
|
elif pz > max_z:
|
|
dz = pz - max_z
|
|
|
|
return math.sqrt(dx * dx + dy * dy + dz * dz)
|
|
|
|
|
|
def _run_required_tests(
|
|
globals_dict: dict,
|
|
*,
|
|
sdk_package: str = "sdk",
|
|
) -> object:
|
|
run_tests = globals_dict.get("run_tests")
|
|
if not callable(run_tests):
|
|
raise ValueError(
|
|
"Missing required `run_tests()` in generated script. "
|
|
"Add a top-level `def run_tests() -> TestReport:` and return `ctx.report()`."
|
|
)
|
|
|
|
try:
|
|
test_report_type = getattr(_import_sdk_module(sdk_package), "TestReport")
|
|
except Exception as exc: # pragma: no cover
|
|
raise ValueError(f"Failed to import {sdk_package}.TestReport: {exc}") from exc
|
|
|
|
report = run_tests()
|
|
if not isinstance(report, test_report_type):
|
|
raise ValueError(
|
|
f"run_tests() must return {sdk_package}.TestReport (got {type(report).__name__})"
|
|
)
|
|
|
|
return report
|
|
|
|
|
|
def _report_warning_check_name(text: object) -> str | None:
|
|
warning_text = str(text).strip()
|
|
if not warning_text:
|
|
return None
|
|
check_name, has_separator, _detail = warning_text.partition(":")
|
|
if not has_separator:
|
|
return None
|
|
return check_name.strip() or None
|
|
|
|
|
|
def _build_test_report(
|
|
report_type: type,
|
|
*,
|
|
checks: tuple[str, ...],
|
|
failures: tuple[object, ...],
|
|
warnings: tuple[str, ...],
|
|
allowances: tuple[str, ...],
|
|
allowed_isolated_parts: tuple[str, ...],
|
|
allowed_overlaps: tuple[object, ...],
|
|
) -> object:
|
|
return report_type(
|
|
passed=not failures,
|
|
checks_run=len(checks),
|
|
checks=checks,
|
|
failures=failures,
|
|
warnings=warnings,
|
|
allowances=allowances,
|
|
allowed_isolated_parts=allowed_isolated_parts,
|
|
allowed_overlaps=allowed_overlaps,
|
|
)
|
|
|
|
|
|
def _filter_duplicate_automated_baseline_results(
|
|
authored_report: object, baseline_report: object
|
|
) -> object:
|
|
authored_check_names = {
|
|
str(name)
|
|
for name in getattr(authored_report, "checks", ())
|
|
if str(name) in _AUTOMATED_BASELINE_DEFAULT_CHECK_NAMES
|
|
}
|
|
if not authored_check_names:
|
|
return baseline_report
|
|
|
|
baseline_checks = tuple(
|
|
str(name)
|
|
for name in getattr(baseline_report, "checks", ())
|
|
if str(name) not in authored_check_names
|
|
)
|
|
baseline_failures = tuple(
|
|
failure
|
|
for failure in getattr(baseline_report, "failures", ())
|
|
if str(getattr(failure, "name", "")) not in authored_check_names
|
|
)
|
|
baseline_warnings = tuple(
|
|
str(warning)
|
|
for warning in getattr(baseline_report, "warnings", ())
|
|
if _report_warning_check_name(warning) not in authored_check_names
|
|
)
|
|
|
|
return _build_test_report(
|
|
type(baseline_report),
|
|
checks=baseline_checks,
|
|
failures=baseline_failures,
|
|
warnings=baseline_warnings,
|
|
allowances=tuple(str(item) for item in getattr(baseline_report, "allowances", ())),
|
|
allowed_isolated_parts=tuple(
|
|
str(item) for item in getattr(baseline_report, "allowed_isolated_parts", ())
|
|
),
|
|
allowed_overlaps=tuple(getattr(baseline_report, "allowed_overlaps", ())),
|
|
)
|
|
|
|
|
|
def _merge_test_reports(
|
|
authored_report: object, baseline_report: object, *, sdk_package: str
|
|
) -> object:
|
|
filtered_baseline_report = _filter_duplicate_automated_baseline_results(
|
|
authored_report,
|
|
baseline_report,
|
|
)
|
|
|
|
try:
|
|
test_report_type = getattr(_import_sdk_module(sdk_package), "TestReport")
|
|
except Exception as exc: # pragma: no cover
|
|
raise ValueError(f"Failed to import {sdk_package}.TestReport: {exc}") from exc
|
|
|
|
check_names: list[str] = []
|
|
seen_checks: set[str] = set()
|
|
for report in (authored_report, filtered_baseline_report):
|
|
for check_name in getattr(report, "checks", ()):
|
|
normalized = str(check_name)
|
|
if normalized in seen_checks:
|
|
continue
|
|
seen_checks.add(normalized)
|
|
check_names.append(normalized)
|
|
|
|
failures: list[object] = []
|
|
seen_failures: set[tuple[str, str]] = set()
|
|
for report in (authored_report, filtered_baseline_report):
|
|
for failure in getattr(report, "failures", ()):
|
|
key = (str(getattr(failure, "name", "")), str(getattr(failure, "details", "")))
|
|
if key in seen_failures:
|
|
continue
|
|
seen_failures.add(key)
|
|
failures.append(failure)
|
|
|
|
warnings: list[str] = []
|
|
seen_warnings: set[str] = set()
|
|
for report in (authored_report, filtered_baseline_report):
|
|
for warning in getattr(report, "warnings", ()):
|
|
text = str(warning)
|
|
if text in seen_warnings:
|
|
continue
|
|
seen_warnings.add(text)
|
|
warnings.append(text)
|
|
|
|
allowances: list[str] = []
|
|
seen_allowances: set[str] = set()
|
|
for report in (authored_report, filtered_baseline_report):
|
|
for allowance in getattr(report, "allowances", ()):
|
|
text = str(allowance)
|
|
if text in seen_allowances:
|
|
continue
|
|
seen_allowances.add(text)
|
|
allowances.append(text)
|
|
|
|
allowed_isolated_parts: list[str] = []
|
|
seen_isolated_parts: set[str] = set()
|
|
for report in (authored_report, filtered_baseline_report):
|
|
for part_name in getattr(report, "allowed_isolated_parts", ()):
|
|
normalized = str(part_name)
|
|
if normalized in seen_isolated_parts:
|
|
continue
|
|
seen_isolated_parts.add(normalized)
|
|
allowed_isolated_parts.append(normalized)
|
|
|
|
allowed_overlaps: list[object] = []
|
|
seen_overlaps: set[tuple[str, str, str | None, str | None, str]] = set()
|
|
for report in (authored_report, filtered_baseline_report):
|
|
for overlap in getattr(report, "allowed_overlaps", ()):
|
|
key = (
|
|
str(getattr(overlap, "link_a", "")),
|
|
str(getattr(overlap, "link_b", "")),
|
|
(
|
|
None
|
|
if getattr(overlap, "elem_a", None) is None
|
|
else str(getattr(overlap, "elem_a"))
|
|
),
|
|
(
|
|
None
|
|
if getattr(overlap, "elem_b", None) is None
|
|
else str(getattr(overlap, "elem_b"))
|
|
),
|
|
str(getattr(overlap, "reason", "")),
|
|
)
|
|
if key in seen_overlaps:
|
|
continue
|
|
seen_overlaps.add(key)
|
|
allowed_overlaps.append(overlap)
|
|
|
|
return _build_test_report(
|
|
test_report_type,
|
|
checks=tuple(check_names),
|
|
failures=tuple(failures),
|
|
warnings=tuple(warnings),
|
|
allowances=tuple(allowances),
|
|
allowed_isolated_parts=tuple(allowed_isolated_parts),
|
|
allowed_overlaps=tuple(allowed_overlaps),
|
|
)
|
|
|
|
|
|
def _check_model_has_single_root_part(ctx: object) -> bool:
|
|
root_parts = getattr(ctx.model, "root_parts", None)
|
|
if not callable(root_parts):
|
|
return bool(ctx.check("check_single_root_part", False, "model has no .root_parts()"))
|
|
|
|
try:
|
|
roots = root_parts()
|
|
except Exception as exc:
|
|
return bool(ctx.check("check_single_root_part", False, f"{type(exc).__name__}: {exc}"))
|
|
|
|
root_names = tuple(
|
|
str(getattr(part, "name", "")).strip()
|
|
for part in roots
|
|
if str(getattr(part, "name", "")).strip()
|
|
)
|
|
if len(root_names) != 1:
|
|
return bool(
|
|
ctx.check(
|
|
"check_single_root_part",
|
|
False,
|
|
f"Expected exactly one root part, found {len(root_names)}: {list(root_names)!r}",
|
|
)
|
|
)
|
|
return bool(ctx.check("check_single_root_part", True))
|
|
|
|
|
|
def _apply_authored_allowances_to_baseline_context(ctx: object, authored_report: object) -> None:
|
|
for part_name in getattr(authored_report, "allowed_isolated_parts", ()):
|
|
ctx.allow_isolated_part(
|
|
str(part_name),
|
|
reason="carried over from authored run_tests() allowance",
|
|
)
|
|
|
|
for overlap in getattr(authored_report, "allowed_overlaps", ()):
|
|
ctx.allow_overlap(
|
|
str(getattr(overlap, "link_a", "")),
|
|
str(getattr(overlap, "link_b", "")),
|
|
reason=str(getattr(overlap, "reason", "")),
|
|
elem_a=(
|
|
None
|
|
if getattr(overlap, "elem_a", None) is None
|
|
else str(getattr(overlap, "elem_a"))
|
|
),
|
|
elem_b=(
|
|
None
|
|
if getattr(overlap, "elem_b", None) is None
|
|
else str(getattr(overlap, "elem_b"))
|
|
),
|
|
)
|
|
|
|
|
|
def _run_compiler_owned_baseline_tests(
|
|
globals_dict: dict,
|
|
*,
|
|
script_path: Path,
|
|
sdk_package: str,
|
|
authored_report: object,
|
|
) -> object:
|
|
object_model = globals_dict.get("object_model")
|
|
if object_model is None:
|
|
raise ValueError("Generated script must define top-level `object_model`")
|
|
|
|
sdk_module = _import_sdk_module(sdk_package)
|
|
test_context_type = getattr(sdk_module, "TestContext")
|
|
ctx = test_context_type(object_model, asset_root=script_path.parent)
|
|
_apply_authored_allowances_to_baseline_context(ctx, authored_report)
|
|
ctx.check_model_valid()
|
|
_check_model_has_single_root_part(ctx)
|
|
preliminary_report = ctx.report()
|
|
if not bool(getattr(preliminary_report, "passed", False)):
|
|
return _build_test_report(
|
|
type(preliminary_report),
|
|
checks=tuple(str(name) for name in getattr(preliminary_report, "checks", ())),
|
|
failures=tuple(getattr(preliminary_report, "failures", ())),
|
|
warnings=tuple(str(item) for item in getattr(preliminary_report, "warnings", ())),
|
|
allowances=(),
|
|
allowed_isolated_parts=(),
|
|
allowed_overlaps=(),
|
|
)
|
|
ctx.check_mesh_assets_ready()
|
|
ctx.fail_if_isolated_parts()
|
|
ctx.warn_if_part_contains_disconnected_geometry_islands()
|
|
ctx.fail_if_parts_overlap_in_current_pose()
|
|
baseline_report = ctx.report()
|
|
return _build_test_report(
|
|
type(baseline_report),
|
|
checks=tuple(str(name) for name in getattr(baseline_report, "checks", ())),
|
|
failures=tuple(getattr(baseline_report, "failures", ())),
|
|
warnings=tuple(str(item) for item in getattr(baseline_report, "warnings", ())),
|
|
allowances=(),
|
|
allowed_isolated_parts=(),
|
|
allowed_overlaps=(),
|
|
)
|
|
|
|
|
|
def _raise_for_failed_test_report(report: object) -> None:
|
|
if bool(getattr(report, "passed", False)):
|
|
return
|
|
|
|
failures = getattr(report, "failures", ())
|
|
lines = ["URDF tests failed:"]
|
|
for failure in list(failures)[:10]:
|
|
name = getattr(failure, "name", "unknown")
|
|
details = getattr(failure, "details", "")
|
|
lines.append(f"- {name}: {details}")
|
|
if len(list(failures)) > 10:
|
|
lines.append(f"... ({len(list(failures)) - 10} more)")
|
|
exc = ValueError("\n".join(lines))
|
|
setattr(exc, "test_report", report)
|
|
raise exc
|
|
|
|
|
|
def update_manifest(outputs_root: Path) -> None:
|
|
entries = []
|
|
for path in outputs_root.rglob("*.urdf"):
|
|
if "viewer" in path.parts:
|
|
continue
|
|
rel = path.relative_to(outputs_root).as_posix()
|
|
name = path.stem
|
|
entries.append({"name": name, "path": rel})
|
|
|
|
entries.sort(key=lambda item: item["name"])
|
|
manifest_path = outputs_root / "manifest.json"
|
|
manifest_path.write_text(json.dumps({"generated": entries}, indent=2))
|
|
|
|
|
|
def persist_compile_success_artifacts(
|
|
*,
|
|
urdf_xml: str,
|
|
urdf_out: Path | None,
|
|
outputs_root: Path | None,
|
|
previous_sig: str | None = None,
|
|
) -> str | None:
|
|
"""
|
|
Persist the latest compile-success URDF and update manifest opportunistically.
|
|
|
|
Returns:
|
|
Content signature for deduping repeated writes.
|
|
"""
|
|
if not isinstance(urdf_xml, str):
|
|
return previous_sig
|
|
|
|
sig = hashlib.sha1(urdf_xml.encode("utf-8")).hexdigest()
|
|
if previous_sig and sig == previous_sig:
|
|
return previous_sig
|
|
|
|
if urdf_out is not None:
|
|
out_path = Path(urdf_out).resolve()
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
out_path.write_text(urdf_xml, encoding="utf-8")
|
|
logger.info("Wrote checkpoint URDF to %s", out_path)
|
|
|
|
if outputs_root is not None:
|
|
update_manifest(Path(outputs_root).resolve())
|
|
|
|
return sig
|