{len(spec.nodes)} nodes · {len(spec.tasks)} tasks · initial state: {html_lib.escape(initial_state.id)}
{html_lib.escape(initial_state.id)}){render_mermaid_graph(spec, initial_state)}
+ {html_lib.escape(constraint.subject)}{params}No unary constraints.
' + return ( + f'No tasks defined.
" + rows = [] + for index, task in enumerate(spec.tasks): + params_str = yaml.safe_dump(task.params, sort_keys=False).rstrip() if task.params else "(empty)" + description = html_lib.escape(task.description or "") + rows.append( + "{index}{html_lib.escape(params_str)}| # | kind | description | params |
|---|
{html_lib.escape(node_yaml)}
+
'
+ f'{html_lib.escape(node.name)}'
+ "No visualization yet — fix the YAML errors to auto-render.
+""" + +_DEFAULT_SAVE_PATH = "isaaclab_arena_environments/llm_generated/generated_spec.yaml" + + +@st.cache_resource(show_spinner="Booting Isaac Sim sidecar (≈30s first run, cached afterwards)…") +def _get_simapp_sidecar() -> SimAppSidecar | None: + """Spawn the SimApp sidecar once per Streamlit server process.""" + sidecar = SimAppSidecar() + try: + sidecar.start() + except SimAppSidecarError as exc: + print(f"[review_gui] SimApp sidecar failed to start: {exc}", flush=True) + return None + + atexit.register(sidecar.close) + return sidecar + + +def _ensure_sidecar() -> SimAppSidecar | None: + """Return a healthy sidecar, re-spawning if the cached one died.""" + sidecar = _get_simapp_sidecar() + if sidecar is not None and sidecar.is_alive(): + return sidecar + if sidecar is not None: + sidecar.close() + _get_simapp_sidecar.clear() + return _get_simapp_sidecar() + + +def _spec_from_sidecar_dict(spec_dict: dict[str, Any]) -> ArenaEnvInitialGraphSpec: + """Rebuild a validated spec locally without registry imports.""" + return ArenaEnvInitialGraphSpec.model_validate(spec_dict, context=_SKIP_REGISTRY_CONTEXT) + + +def _render_with_thumbnails(spec: ArenaEnvInitialGraphSpec) -> str: + """Render review HTML, asking the sidecar for live USD thumbnails.""" + sidecar = _ensure_sidecar() + if sidecar is None: + st.warning( + "Isaac Sim sidecar is unavailable — showing placeholder thumbnails. " + "Check the terminal where you launched the server for the underlying error.", + icon="⚠️", + ) + return render_dashboard_html(spec) + + try: + thumbnails = sidecar.render_spec(spec) + except SimAppSidecarError as exc: + st.error( + f"Sidecar render failed; showing placeholder thumbnails.\n\n```\n{exc}\n```", + icon="🛑", + ) + with st.spinner("Resetting the SimApp sidecar…"): + _get_simapp_sidecar.clear() + return render_dashboard_html(spec) + + return render_dashboard_html(spec, thumbnails=thumbnails if thumbnails else None) + + +@dataclass +class ValidationResult: + """Outcome of parsing and validating YAML text as an initial graph spec.""" + + spec: ArenaEnvInitialGraphSpec | None + error: str | None + + @property + def is_valid(self) -> bool: + return self.spec is not None + + +def validate_yaml_text(text: str) -> ValidationResult: + """Parse YAML and validate via the SimApp sidecar (registry lookups run there).""" + if not text.strip(): + return ValidationResult(spec=None, error=None) + + try: + raw = yaml.safe_load(text) + except Exception: + return ValidationResult(spec=None, error=traceback.format_exc()) + + if raw is None: + return ValidationResult(spec=None, error="YAML is empty") + if not isinstance(raw, dict): + return ValidationResult(spec=None, error=f"Expected mapping, got {type(raw).__name__}") + + sidecar = _ensure_sidecar() + if sidecar is None: + return ValidationResult( + spec=None, + error=( + "SimApp sidecar is unavailable — cannot validate registry entries. " + "Check the terminal where you launched the server." + ), + ) + + try: + response = sidecar.validate_yaml_text(text) + except SimAppSidecarError as exc: + _get_simapp_sidecar.clear() + return ValidationResult(spec=None, error=str(exc)) + + if not response.get("ok"): + err = response.get("error", "validation failed") + tb = response.get("traceback", "") + message = f"{err}\n\n{tb}" if tb else str(err) + return ValidationResult(spec=None, error=message) + + try: + spec = _spec_from_sidecar_dict(response["spec_dict"]) + except Exception: + return ValidationResult(spec=None, error=traceback.format_exc()) + + return ValidationResult(spec=spec, error=None) + + +@dataclass +class CatalogueBundle: + """Asset/relation/task vocabulary fetched from the SimApp sidecar.""" + + asset_catalogue: AssetCatalogue + relation_catalogue: RelationCatalogue + task_catalogue: TaskCatalogue + + +def _catalogues_from_sidecar_response(response: dict[str, Any]) -> CatalogueBundle: + asset_raw = response["asset_catalogue"] + relation_raw = response["relation_catalogue"] + task_raw = response["task_catalogue"] + return CatalogueBundle( + asset_catalogue=AssetCatalogue( + embodiments=list(asset_raw["embodiments"]), + backgrounds=list(asset_raw["backgrounds"]), + objects=list(asset_raw["objects"]), + ), + relation_catalogue=RelationCatalogue( + relations=[RelationCatalogueEntry(**entry) for entry in relation_raw["relations"]], + ), + task_catalogue=TaskCatalogue( + tasks=[TaskCatalogueEntry(**entry) for entry in task_raw["tasks"]], + ), + ) + + +def _get_catalogue_bundle() -> CatalogueBundle | None: + """Return cached catalogues built in the sidecar (registry lookups run there).""" + cached = st.session_state.get("catalogue_bundle") + if cached is not None: + return cached + + sidecar = _ensure_sidecar() + if sidecar is None: + return None + + try: + response = sidecar.build_catalogues() + except SimAppSidecarError as exc: + st.session_state["catalogue_error"] = str(exc) + _get_simapp_sidecar.clear() + return None + + if not response.get("ok"): + st.session_state["catalogue_error"] = response.get("error", "catalogue build failed") + return None + + bundle = _catalogues_from_sidecar_response(response) + st.session_state["catalogue_bundle"] = bundle + st.session_state.pop("catalogue_error", None) + return bundle + + +def _get_generation_agent() -> EnvironmentGenerationAgent | None: + """Lazy-init the LLM agent when ``NV_API_KEY`` is available.""" + if st.session_state.get("generation_agent_error"): + return None + agent = st.session_state.get("generation_agent") + if agent is not None: + return agent + try: + agent = EnvironmentGenerationAgent() + except AssertionError as exc: + st.session_state["generation_agent_error"] = str(exc) + return None + except Exception as exc: + st.session_state["generation_agent_error"] = f"{type(exc).__name__}: {exc}" + return None + st.session_state["generation_agent"] = agent + st.session_state.pop("generation_agent_error", None) + return agent + + +def _format_trace_lines(trace: list[dict[str, Any]], *, errors_only: bool = False) -> str: + error_stages = ASSET_ERROR_STAGES | IntentCompiler._ERROR_TRACE_STAGES + lines: list[str] = [] + for event in trace: + stage = event.get("stage", "") + if errors_only and stage not in error_stages: + continue + chosen = event.get("chosen") + chosen_str = chosen if chosen is not None else "