[WIP] Add Visualizers and Scene Data Providers

docs/source/experimental-features/newton-physics-integration/visualization.rst

@@ -213,8 +213,8 @@ Newton Visualizer
         enable_wireframe=False,                   # Enable wireframe mode
 
         # Color customization
-        background_color=(0.53, 0.81, 0.92),     # Sky/background color (RGB [0,1])
-        ground_color=(0.18, 0.20, 0.25),         # Ground plane color (RGB [0,1])
+        sky_upper_color=(0.53, 0.81, 0.92),      # Sky upper color (RGB [0,1])
+        sky_lower_color=(0.18, 0.20, 0.25),      # Sky lower color (RGB [0,1])
         light_color=(1.0, 1.0, 1.0),             # Directional light color (RGB [0,1])
     )
 

source/isaaclab/isaaclab/app/app_launcher.py

@@ -115,6 +115,7 @@ def __init__(self, launcher_args: argparse.Namespace | dict | None = None, **kwa
         self._livestream: Literal[0, 1, 2]  # 0: Disabled, 1: WebRTC public, 2: WebRTC private
         self._offscreen_render: bool  # 0: Disabled, 1: Enabled
         self._sim_experience_file: str  # Experience file to load
+        self._visualizer: list[str] | None  # Visualizer backends to use
 
         # Exposed to train scripts
         self.device_id: int  # device ID for GPU simulation (defaults to 0)
@@ -304,6 +305,16 @@ def add_app_launcher_args(parser: argparse.ArgumentParser) -> None:
             default=AppLauncher._APPLAUNCHER_CFG_INFO["device"][1],
             help='The device to run the simulation on. Can be "cpu", "cuda", "cuda:N", where N is the device ID',
         )
+        arg_group.add_argument(
+            "--visualizer",
+            type=str,
+            nargs="+",
+            default=None,
+            help=(
+                "Visualizer backend(s) to use. Valid values: newton, rerun, omniverse."
+                " Multiple visualizers can be specified: --visualizer rerun newton"
+            ),
+        )
         # Add the deprecated cpu flag to raise an error if it is used
         arg_group.add_argument("--cpu", action="store_true", help=argparse.SUPPRESS)
         arg_group.add_argument(
@@ -389,6 +400,7 @@ def add_app_launcher_args(parser: argparse.ArgumentParser) -> None:
         "device": ([str], "cuda:0"),
         "experience": ([str], ""),
         "rendering_mode": ([str], "balanced"),
+        "visualizer": ([list, type(None)], None),
     }
     """A dictionary of arguments added manually by the :meth:`AppLauncher.add_app_launcher_args` method.
 
@@ -488,6 +500,7 @@ def _config_resolution(self, launcher_args: dict):
         self._resolve_headless_settings(launcher_args, livestream_arg, livestream_env)
         self._resolve_camera_settings(launcher_args)
         self._resolve_xr_settings(launcher_args)
+        self._resolve_visualizer_settings(launcher_args)
         self._resolve_viewport_settings(launcher_args)
 
         # Handle device and distributed settings
@@ -777,6 +790,46 @@ def _resolve_anim_recording_settings(self, launcher_args: dict):
                 )
             sys.argv += ["--enable", "omni.physx.pvd"]
 
+    def _resolve_visualizer_settings(self, launcher_args: dict) -> None:
+        """Resolve visualizer related settings."""
+        visualizers = launcher_args.pop("visualizer", AppLauncher._APPLAUNCHER_CFG_INFO["visualizer"][1])
+        valid_visualizers = {"newton", "rerun", "omniverse"}
+        if visualizers is not None and len(visualizers) > 0:
+            invalid = [v for v in visualizers if v not in valid_visualizers]
+            if invalid:
+                raise ValueError(
+                    f"Invalid visualizer(s) specified: {invalid}. Valid options are: {sorted(valid_visualizers)}"
+                )
+        self._visualizer = visualizers if visualizers and len(visualizers) > 0 else None
+
+        # Auto-adjust headless based on requested visualizers (parity with feature/newton behavior).
+        if self._visualizer is None:
+            if not self._headless and self._livestream not in {1, 2}:
+                self._headless = True
+                launcher_args["headless"] = True
+                print(
+                    "[INFO][AppLauncher]: No visualizers specified. "
+                    "Automatically enabling headless mode. Use --visualizer <type> to enable GUI."
+                )
+            return
+
+        if "omniverse" in self._visualizer:
+            if self._headless:
+                self._headless = False
+                launcher_args["headless"] = False
+                print(
+                    "[INFO][AppLauncher]: Omniverse visualizer requested. "
+                    "Forcing headless=False for GUI."
+                )
+        else:
+            if not self._headless and self._livestream not in {1, 2}:
+                self._headless = True
+                launcher_args["headless"] = True
+                print(
+                    f"[INFO][AppLauncher]: Visualizer(s) {self._visualizer} requested. "
+                    "Enabling headless mode for SimulationApp (visualizers run independently)."
+                )
+
     def _resolve_kit_args(self, launcher_args: dict):
         """Resolve additional arguments passed to Kit."""
         # Resolve additional arguments passed to Kit
@@ -867,6 +920,12 @@ def _load_extensions(self):
         # for example: the `Camera` sensor class
         carb_settings_iface.set_bool("/isaaclab/render/rtx_sensors", False)
 
+        # store visualizer selection for SimulationContext
+        if self._visualizer is not None:
+            carb_settings_iface.set_string("/isaaclab/visualizer", ",".join(self._visualizer))
+        else:
+            carb_settings_iface.set_string("/isaaclab/visualizer", "")
+
         # set fabric update flag to disable updating transforms when rendering is disabled
         carb_settings_iface.set_bool("/physics/fabricUpdateTransformations", self._rendering_enabled())
 

source/isaaclab/isaaclab/envs/manager_based_env.py

@@ -142,6 +142,14 @@ def __init__(self, cfg: ManagerBasedEnvCfg):
                 self.scene = InteractiveScene(self.cfg.scene)
                 attach_stage_to_usd_context()
         print("[INFO]: Scene manager: ", self.scene)
+        from isaaclab.sim.utils import find_matching_prim_paths
+
+        env_prim_paths = find_matching_prim_paths("/World/envs/env_.*", stage=self.scene.stage)
+        print(
+            "[SceneDebug] env prims after InteractiveScene: "
+            f"num_envs_setting={self.cfg.scene.num_envs}, env_prims={len(env_prim_paths)}"
+        )
+        # import ipdb; ipdb.set_trace()
 
         # set up camera viewport controller
         # viewport is not available in other rendering modes so the function will throw a warning

source/isaaclab/isaaclab/scene/interactive_scene.py

@@ -120,6 +120,10 @@ def __init__(self, cfg: InteractiveSceneCfg):
         cfg.validate()
         # store inputs
         self.cfg = cfg
+
+        # TODO(mtrepte): remove
+        self.cfg.clone_in_fabric = False
+
         # initialize scene elements
         self._terrain = None
         self._articulations = dict()
@@ -133,6 +137,11 @@ def __init__(self, cfg: InteractiveSceneCfg):
         self.sim = SimulationContext.instance()
         self.stage = get_current_stage()
         self.stage_id = get_current_stage_id()
+        # publish num_envs for consumers outside the scene
+        try:
+            self.sim.set_setting("/isaaclab/scene/num_envs", int(self.cfg.num_envs))
+        except Exception:
+            pass
         # physics scene path
         self._physics_scene_path = None
         # prepare cloner for environment replication
@@ -270,6 +279,18 @@ def clone_environments(self, copy_from_source: bool = False):
         if self._default_env_origins is None:
             self._default_env_origins = torch.tensor(env_origins, device=self.device, dtype=torch.float32)
 
+        # publish env origins for consumers that cannot read USD (e.g., Fabric clones)
+        try:
+            if hasattr(env_origins, "flatten"):
+                origins_list = env_origins.flatten().tolist()
+            else:
+                origins_list = []
+                for origin in env_origins:
+                    origins_list.extend(list(origin))
+            self.sim.set_setting("/isaaclab/scene/env_origins", origins_list)
+        except Exception:
+            pass
+
     def filter_collisions(self, global_prim_paths: list[str] | None = None):
         """Filter environments collisions.
 

source/isaaclab/isaaclab/sim/scene_data_providers/__init__.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+# Copyright (c) 2022-2026, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Scene data providers for visualizers and renderers."""
+
+from .newton_scene_data_provider import NewtonSceneDataProvider
+from .ov_scene_data_provider import OVSceneDataProvider
+from .scene_data_provider import SceneDataProvider
+
+__all__ = [
+    "SceneDataProvider",
+    "NewtonSceneDataProvider",
+    "OVSceneDataProvider",
+]

source/isaaclab/isaaclab/sim/scene_data_providers/newton_scene_data_provider.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+# Copyright (c) 2022-2026, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Newton-backed scene data provider."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class NewtonSceneDataProvider:
+    """Scene data provider for Newton Warp physics backend.
+
+    Native (cheap): Newton Model/State from NewtonManager
+    Adapted (future): USD stage (would need Newton→USD sync for OV visualizer)
+    """
+
+    def __init__(self, visualizer_cfgs: list[Any] | None) -> None:
+        self._has_ov_visualizer = False
+        self._metadata: dict[str, Any] = {}
+
+        if visualizer_cfgs:
+            for cfg in visualizer_cfgs:
+                if getattr(cfg, "visualizer_type", None) == "omniverse":
+                    self._has_ov_visualizer = True
+
+        try:
+            from isaaclab.sim._impl.newton_manager import NewtonManager
+
+            self._metadata = {
+                "physics_backend": "newton",
+                "num_envs": NewtonManager._num_envs if NewtonManager._num_envs is not None else 0,
+                "gravity_vector": NewtonManager._gravity_vector,
+                "clone_physics_only": NewtonManager._clone_physics_only,
+            }
+        except Exception:
+            self._metadata = {"physics_backend": "newton"}
+
+    def update(self) -> None:
+        """No-op for Newton backend (state updated by Newton solver)."""
+        pass
+
+    def get_newton_model(self) -> Any | None:
+        """NATIVE: Newton Model from NewtonManager."""
+        try:
+            from isaaclab.sim._impl.newton_manager import NewtonManager
+            return NewtonManager._model
+        except Exception:
+            return None
+
+    def get_newton_state(self) -> Any | None:
+        """NATIVE: Newton State from NewtonManager."""
+        try:
+            from isaaclab.sim._impl.newton_manager import NewtonManager
+            return NewtonManager._state_0
+        except Exception:
+            return None
+
+    def get_usd_stage(self) -> None:
+        """UNAVAILABLE: Newton backend doesn't provide USD (future: Newton→USD sync)."""
+        return None
+
+    def get_metadata(self) -> dict[str, Any]:
+        return dict(self._metadata)
+
+    def get_transforms(self) -> dict[str, Any] | None:
+        """Extract transforms from Newton state (future work)."""
+        return None
+
+    def get_velocities(self) -> dict[str, Any] | None:
+        try:
+            from isaaclab.sim._impl.newton_manager import NewtonManager
+            if NewtonManager._state_0 is None:
+                return None
+            return {"body_qd": NewtonManager._state_0.body_qd}
+        except Exception:
+            return None
+
+    def get_contacts(self) -> dict[str, Any] | None:
+        try:
+            from isaaclab.sim._impl.newton_manager import NewtonManager
+            if NewtonManager._contacts is None:
+                return None
+            return {"contacts": NewtonManager._contacts}
+        except Exception:
+            return None
+
+    def get_mesh_data(self) -> dict[str, Any] | None:
+        """ADAPTED: Extract mesh data from Newton shapes (future work)."""
+        return None