Source code for isaaclab_newton.renderers.newton_warp_renderer

# 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 Warp renderer for tiled camera rendering."""

from __future__ import annotations

import logging
from dataclasses import dataclass
from functools import partial
from typing import TYPE_CHECKING, Any, NoReturn

import newton
import warp as wp

from isaaclab.renderers import BaseRenderer, RenderBufferKind, RenderBufferSpec
from isaaclab.renderers.camera_render_spec import CameraRenderSpec
from isaaclab.scene_data import REQUIRES_STAGE_AND_MODEL, SceneDataFormat
from isaaclab.sim import SimulationContext
from isaaclab.utils.warp.warp_math import replace_background_depth_wp

from ..physics import NewtonBackendCfg, NewtonQueries
from .newton_warp_renderer_cfg import NewtonWarpRendererCfg
from .segmentation import NewtonSegmentationMapper, NewtonSegmentationMapping

if TYPE_CHECKING:
    from isaaclab_ppisp import PpispPipeline

    from isaaclab.sensors.camera.camera_data import CameraData
    from isaaclab.utils.warp import ProxyArray

logger = logging.getLogger(__name__)


@wp.kernel(enable_backward=False)
def _check_shared_intrinsics(intrinsics: wp.array(dtype=wp.mat33f), different: wp.array(dtype=wp.int32)):
    i = wp.tid()
    for row in range(3):
        for column in range(3):
            if wp.abs(intrinsics[i][row, column] - intrinsics[0][row, column]) > 1.0e-4:
                wp.atomic_or(different, 0, 1)


@wp.kernel(enable_backward=False)
def _update_camera_rays(intrinsics: wp.array(dtype=wp.mat33f), rays: wp.array4d(dtype=wp.vec3f)):
    y, x = wp.tid()
    matrix = intrinsics[0]
    direction = wp.vec3f(
        (float(x) + 0.5 - matrix[0, 2]) / matrix[0, 0],
        -(float(y) + 0.5 - matrix[1, 2]) / matrix[1, 1],
        -1.0,
    )
    rays[0, y, x, 0] = wp.vec3f(0.0)
    rays[0, y, x, 1] = wp.normalize(direction)


_PPISP_IMPORT_ERROR_MESSAGE = (
    "isaaclab_ppisp is required when CameraCfg.isp_cfg is set. "
    "It ships with the Isaac Lab wheel (`pip install isaaclab`); otherwise install the "
    "isaaclab-ppisp extension from the Isaac Lab source checkout."
)


def _raise_missing_ppisp_error(exc: ModuleNotFoundError) -> NoReturn:
    # Only translate missing isaaclab_ppisp imports into the optional-dependency hint;
    # unrelated missing modules should surface unchanged for easier debugging.
    if exc.name != "isaaclab_ppisp" and not (exc.name and exc.name.startswith("isaaclab_ppisp.")):
        raise exc
    raise ModuleNotFoundError(_PPISP_IMPORT_ERROR_MESSAGE, name="isaaclab_ppisp") from exc


class RenderData:
    # Back-compat alias for callers of ``RenderData.OutputNames``.
    OutputNames = RenderBufferKind

    # Maps each supported RenderBufferKind to (CameraOutputs field name, Newton warp dtype).
    # Newton reinterprets the allocated buffer memory: e.g. RGBA is allocated as (N,H,W,4) uint8
    # but the Newton sensor API consumes it as (world_count,1,H,W) uint32 (same bytes, packed view).
    #
    # The depth family (``distance_to_camera`` / ``distance_to_image_plane`` / ``depth``) is handled
    # separately in :meth:`set_outputs` because the two planar-depth names share one native output.
    #
    # The segmentation family (``semantic_segmentation`` / ``instance_segmentation``) is likewise
    # handled separately: Newton emits a single per-shape index buffer that is remapped into each
    # requested segmentation output by
    # :class:`~isaaclab_newton.renderers.segmentation.NewtonSegmentationMapper`.
    _OUTPUT_MAP: dict[str, tuple[str, type]] = {
        str(RenderBufferKind.RGBA): ("color_image", wp.uint32),
        str(RenderBufferKind.RGB_HDR): ("hdr_color_image", wp.vec3f),
        str(RenderBufferKind.ALBEDO): ("albedo_image", wp.uint32),
        str(RenderBufferKind.NORMALS): ("normals_image", wp.vec3f),
    }

    # Newton's native ``depth_image`` is the ray-hit (euclidean) distance from the camera optical
    # center, which is Isaac Lab's ``distance_to_camera``.
    _RAY_DEPTH_KIND: str = str(RenderBufferKind.DISTANCE_TO_CAMERA)
    # Planar-depth outputs (distance along the camera's forward axis). ``depth`` is Isaac Lab's alias
    # for ``distance_to_image_plane``. Both are derived from the ray depth via
    # ``convert_ray_depth_to_forward_depth``.
    _PLANE_DEPTH_KINDS: frozenset[str] = frozenset(
        {
            str(RenderBufferKind.DEPTH),
            str(RenderBufferKind.DISTANCE_TO_IMAGE_PLANE),
        }
    )

    @dataclass
    class CameraOutputs:
        color_image: wp.array(dtype=wp.uint32, ndim=4) = None
        hdr_color_image: wp.array(dtype=wp.vec3f, ndim=4) = None
        albedo_image: wp.array(dtype=wp.uint32, ndim=4) = None
        # Native ray-hit distance and planar depth, bound directly to the requested outputs.
        depth_image: wp.array(dtype=wp.float32, ndim=4) = None
        forward_depth_image: wp.array(dtype=wp.float32, ndim=4) = None
        normals_image: wp.array(dtype=wp.vec3f, ndim=4) = None
        # Buffer Newton fills with the per-pixel shape index; the source for all segmentation outputs.
        shape_index_image: wp.array(dtype=wp.uint32, ndim=4) = None

    def __init__(
        self,
        newton_sensor: newton.sensors.SensorTiledCamera,
        spec: CameraRenderSpec,
        seg_mapper: NewtonSegmentationMapper | None = None,
        renderer_cfg: NewtonWarpRendererCfg | None = None,
    ):
        self.newton_sensor = newton_sensor
        # Shared, scene-static segmentation lookup builder (``None`` until segmentation is requested).
        self._seg_mapper = seg_mapper
        self._renderer_cfg = renderer_cfg

        self.num_cameras = 1

        self.camera_rays: wp.array(dtype=wp.vec3f, ndim=4) = None
        self.camera_transforms: wp.array(dtype=wp.transformf, ndim=2) = None
        self._intrinsic_status = wp.zeros(1, dtype=wp.int32, device=newton_sensor.model.device)
        self.graph = None
        self.outputs = RenderData.CameraOutputs()
        # Requested depth-family destination views keyed by data-type name. Each view aliases the
        # caller's output buffer as ``(world_count, 1, H, W)`` float32.
        self._depth_dests: dict[str, wp.array] = {}
        # Requested segmentation outputs keyed by data-type name -> (destination view, mapping). Each view
        # aliases the caller's output buffer as ``(world_count, 1, H, W)`` uint32.
        self._seg_dests: dict[str, tuple[wp.array, NewtonSegmentationMapping]] = {}
        self.width = getattr(spec.cfg, "width", 100)
        self.height = getattr(spec.cfg, "height", 100)
        # Camera clipping planes [m] from ``spawn.clipping_range`` (``[0]`` near, ``[1]`` far).
        # Newton's ray tracer has no near-plane parameter, so only the far plane is enforced (through
        # the sensor's ``max_distance``); ``near_clip`` is captured for consumers but not applied.
        spawn = spec.cfg.spawn
        clipping_range = getattr(spawn, "clipping_range", None)
        self.near_clip: float | None = float(clipping_range[0]) if clipping_range is not None else None
        self.far_clip: float | None = float(clipping_range[1]) if clipping_range is not None else None

        # ABGR clear color packed as uint32 — Newton's SensorTiledCamera reads the low byte as R,
        # next as G, next as B, high byte as A (little-endian RGBA in memory). Default is 93% gray
        # (0xFFEEEEEE), matching the RTX renderer background and improving visibility of dark objects.
        background_color = getattr(spec.cfg, "background_color", None)
        if background_color is not None:
            r, g, b = (max(0, min(255, round(c * 255))) for c in background_color)
            self.clear_color: int = (0xFF << 24) | (b << 16) | (g << 8) | r
        else:
            self.clear_color = 0xFFEEEEEE

        # OpenCV lens-distortion model (``spawn.distortion``), consumed by :meth:`_build_distortion_rays`
        # to trace distorted per-pixel rays instead of the centered, square-pixel pinhole field.
        self._distortion = spawn.distortion if spawn is not None else None
        # Post-render PPISP pipeline composed when ``spec.cfg.isp_cfg`` is set.
        # ``isp_cfg`` is already fully normalized by ``prepare_cameras`` by the time it reaches here.
        self.ppisp_pipeline: PpispPipeline | None = None
        if spec.cfg.isp_cfg is not None:
            try:
                from isaaclab_ppisp import PpispPipeline
            except ModuleNotFoundError as exc:
                _raise_missing_ppisp_error(exc)

            self.ppisp_pipeline = PpispPipeline(spec.cfg.isp_cfg)
        self._hdr_scratch_wp: wp.array | None = None
        """Internal HDR scratch buffer allocated when PPISP is composed but the
        user did not request ``"rgb_hdr"`` in ``data_types``. Also exposed to
        the Newton sensor through :attr:`CameraOutputs.hdr_color_image` as a
        vec3f reinterpretation of this same backing storage."""
        self._ppisp_hdr_source: wp.array | None = None
        """PPISP HDR source bound once in :meth:`set_outputs` from the caller's
        ``rgb_hdr`` output or :attr:`_hdr_scratch_wp`."""
        self._ppisp_rgba_dest: wp.array | None = None
        """PPISP LDR destination bound once in :meth:`set_outputs` from the
        caller's ``rgba`` output."""

    def _view(self, proxy: ProxyArray, dtype: type, shape: tuple[int, ...]) -> wp.array:
        """Alias the caller's output buffer as a ``(world_count, 1, H, W)`` warp array of ``dtype``.

        Newton reinterprets the backing memory in place (no copy), so the sensor writes directly
        into the camera's output buffer.
        """
        wp_arr = proxy.warp
        return wp.array(ptr=wp_arr.ptr, dtype=dtype, shape=shape, device=wp_arr.device, copy=False)

    def set_outputs(self, output_data: dict[str, ProxyArray]):
        shape = (self.newton_sensor.model.world_count, self.num_cameras, self.height, self.width)
        self._depth_dests = {}
        self._seg_dests = {}
        self.outputs.shape_index_image = None
        self.outputs.depth_image = None
        self.outputs.forward_depth_image = None
        for output_name, proxy in output_data.items():
            # Bind one destination for each native depth output; copy additional planar aliases after rendering.
            if output_name == self._RAY_DEPTH_KIND or output_name in self._PLANE_DEPTH_KINDS:
                dest = self._view(proxy, wp.float32, shape)
                self._depth_dests[output_name] = dest
                if output_name == self._RAY_DEPTH_KIND:
                    self.outputs.depth_image = dest
                elif self.outputs.forward_depth_image is None:
                    self.outputs.forward_depth_image = dest
                continue
            # Segmentation family: bind each requested output to a destination view — colorized RGBA
            # (uint32 packed) or raw int32 ids (matching the Isaac RTX / OVRTX contract).  Newton
            # fills only the shape-index scratch (uint32), which is remapped into each output in
            # :meth:`_convert_segmentation`.
            if output_name == RenderBufferKind.SEMANTIC_SEGMENTATION:
                colorize = bool(self._renderer_cfg.colorize_semantic_segmentation)
            elif output_name == RenderBufferKind.INSTANCE_SEGMENTATION:
                colorize = bool(self._renderer_cfg.colorize_instance_segmentation)
            else:
                colorize = None
            if colorize is not None:
                if self._seg_mapper is None:
                    raise RuntimeError(
                        f"Output '{output_name}' requires a segmentation mapper, but none was created. "
                        "Ensure the camera's data_types includes the segmentation output."
                    )
                seg_mapping = self._seg_mapper.get_mapping(output_name, colorize)
                dest = self._view(proxy, wp.uint32 if colorize else wp.int32, shape)
                self._seg_dests[output_name] = (dest, seg_mapping)
                continue
            mapping = self._OUTPUT_MAP.get(output_name)
            if mapping is None:
                if output_name != str(RenderBufferKind.RGB):
                    logger.warning(f"NewtonWarpRenderer - output type {output_name} is not yet supported")
                continue
            field_name, dtype = mapping
            setattr(self.outputs, field_name, self._view(proxy, dtype, shape))
        # Allocate the shape-index buffer Newton fills when any segmentation output is requested; all
        # requested segmentation outputs are remapped from this single buffer in :meth:`_convert_segmentation`.
        if self._seg_dests:
            self.outputs.shape_index_image = wp.zeros(shape, dtype=wp.uint32, device=self.newton_sensor.model.device)
        # When PPISP is composed but the user did not request the raw HDR AOV,
        # allocate an internal HDR scratch buffer and route a vec3f-shaped view
        # of it as the Newton sensor's ``hdr_color_image`` so the renderer
        # fills it directly.
        if self.ppisp_pipeline is not None and self.outputs.hdr_color_image is None:
            ref_proxy = next(iter(output_data.values()))
            self._hdr_scratch_wp = wp.zeros(
                (self.newton_sensor.model.world_count, self.height, self.width, 3),
                dtype=wp.float32,
                device=ref_proxy.device,
            )
            self.outputs.hdr_color_image = wp.array(
                ptr=self._hdr_scratch_wp.ptr,
                dtype=wp.vec3f,
                shape=shape,
                device=self._hdr_scratch_wp.device,
                copy=False,
            )
        # Bind the two warp arrays the per-frame PPISP dispatch needs.
        if self.ppisp_pipeline is not None:
            if str(RenderBufferKind.RGBA) not in output_data:
                raise ValueError(
                    "Newton renderer ISP requires 'rgba' (or 'rgb', which aliases into rgba) as the"
                    " LDR output destination, but neither was provided. Add 'rgb' or 'rgba' to"
                    " Camera.cfg.data_types when isp_cfg is set."
                )
            hdr_proxy = output_data.get(str(RenderBufferKind.RGB_HDR))
            self._ppisp_hdr_source = hdr_proxy.warp if hdr_proxy is not None else self._hdr_scratch_wp
            self._ppisp_rgba_dest = output_data[str(RenderBufferKind.RGBA)].warp

    def get_output(self, output_name: str) -> wp.array:
        if output_name in self._depth_dests:
            return self._depth_dests[output_name]
        elif output_name in self._seg_dests:
            return self._seg_dests[output_name][0]
        elif output_name == RenderBufferKind.RGBA:
            return self.outputs.color_image
        elif output_name == RenderBufferKind.RGB_HDR:
            return self.outputs.hdr_color_image
        elif output_name == RenderBufferKind.ALBEDO:
            return self.outputs.albedo_image
        elif output_name == RenderBufferKind.NORMALS:
            return self.outputs.normals_image
        return None

    def _convert_segmentation(self):
        """Remap Newton's shape-index buffer into each requested segmentation output.

        Newton emits a single per-pixel shape index (:attr:`CameraOutputs.shape_index_image`);
        ``semantic_segmentation`` / ``instance_segmentation`` are each derived from it by a
        :class:`~isaaclab_newton.renderers.segmentation.NewtonSegmentationMapping`.
        No-op when no segmentation output was requested.
        """
        if self.outputs.shape_index_image is None:
            return
        for dest, seg_mapping in self._seg_dests.values():
            seg_mapping.convert_shape_index_to_output(self.outputs.shape_index_image, dest)

    def segmentation_info(self) -> dict[str, dict]:
        """Per-output ``idToLabels`` / ``idToSemantics`` info for the requested segmentation outputs."""
        return {name: seg_mapping.info for name, (_dest, seg_mapping) in self._seg_dests.items()}

    def _copy_plane_depth(self):
        """Copy native planar depth when both ``depth`` and ``distance_to_image_plane`` were requested."""
        for output_name, dest in self._depth_dests.items():
            if output_name in self._PLANE_DEPTH_KINDS and dest.ptr != self.outputs.forward_depth_image.ptr:
                wp.copy(dest, self.outputs.forward_depth_image)

    def _apply_depth_clipping(self, behavior: str):
        """Apply the renderer's depth-clipping behavior to the depth-family outputs.

        Newton writes ``0.0`` for rays that miss all geometry or fall beyond the far plane
        (``max_distance``), so ``"none"`` and ``"zero"`` both leave that ``0.0`` background. ``"max"``
        replaces the background with the far clip [m] to mirror the RTX renderer's
        :attr:`~isaaclab_physx.renderers.IsaacRtxRendererCfg.depth_clipping_behavior`. No-op when no
        depth output was requested or the camera did not provide a clipping range.

        """
        if behavior != "max" or self.far_clip is None:
            return
        for dest in self._depth_dests.values():
            replace_background_depth_wp(dest, self.far_clip, device=dest.device)

    def update(self, positions: ProxyArray, orientations: ProxyArray, intrinsics: ProxyArray):
        # Buffers are persistent: the sensor manager graph captures the render
        # launch against `camera_transforms`, so it must be updated in place.
        if self.camera_transforms is None:
            self.camera_transforms = wp.empty(
                (1, self.newton_sensor.model.world_count),
                dtype=wp.transformf,
                device=self.newton_sensor.model.device,
            )
        wp.launch(
            RenderData._update_transforms,
            self.newton_sensor.model.world_count,
            [positions, orientations, self.camera_transforms],
            device=self.newton_sensor.model.device,
        )

        if self.camera_rays is None:
            if self._distortion is not None:
                self.camera_rays = self._build_distortion_rays()
            else:
                self.camera_rays = wp.empty(
                    (1, self.height, self.width, 2), dtype=wp.vec3f, device=self.newton_sensor.model.device
                )
                wp.launch(
                    _update_camera_rays,
                    (self.height, self.width),
                    [intrinsics.warp, self.camera_rays],
                    device=self.newton_sensor.model.device,
                )

    def _build_distortion_rays(self) -> wp.array(dtype=wp.vec3f, ndim=4):
        """Build the ``(1, H, W, 2)`` camera-space ray field for an OpenCV lens-distortion camera.

        Uses Newton's native OpenCV pinhole and fisheye ray helpers. Both paths honor calibrated
        ``fx/fy/cx/cy`` (non-square, off-center) intrinsics. When
        :attr:`OpenCvDistortionCfg.apply_lens_distortion` is ``False``, the coefficients are treated
        as zero while the calibrated intrinsics remain active, matching the RTX/OVRTX behavior.
        """
        cfg = self._distortion
        image_width, image_height = float(cfg.image_size[0]), float(cfg.image_size[1])

        def _coefficient(value: float) -> float:
            return float(value) if cfg.apply_lens_distortion else 0.0

        if cfg.model == "opencvFisheye":
            return self.newton_sensor.utils.compute_camera_rays_fisheye_opencv(
                self.width,
                self.height,
                float(cfg.fx),
                float(cfg.fy),
                float(cfg.cx),
                float(cfg.cy),
                image_width=image_width,
                image_height=image_height,
                k1=_coefficient(cfg.k1),
                k2=_coefficient(cfg.k2),
                k3=_coefficient(cfg.k3),
                k4=_coefficient(cfg.k4),
                max_fov=cfg.max_fov,
            )

        return self.newton_sensor.utils.compute_camera_rays_pinhole_opencv(
            self.width,
            self.height,
            float(cfg.fx),
            float(cfg.fy),
            float(cfg.cx),
            float(cfg.cy),
            image_width=image_width,
            image_height=image_height,
            k1=_coefficient(cfg.k1),
            k2=_coefficient(cfg.k2),
            k3=_coefficient(cfg.k3),
            k4=_coefficient(cfg.k4),
            k5=_coefficient(cfg.k5),
            k6=_coefficient(cfg.k6),
            p1=_coefficient(cfg.p1),
            p2=_coefficient(cfg.p2),
            s1=_coefficient(cfg.s1),
            s2=_coefficient(cfg.s2),
            s3=_coefficient(cfg.s3),
            s4=_coefficient(cfg.s4),
        )

    @wp.kernel
    def _update_transforms(
        positions: wp.array(dtype=wp.vec3f),
        orientations: wp.array(dtype=wp.quatf),
        output: wp.array(dtype=wp.transformf, ndim=2),
    ):
        tid = wp.tid()
        # Convert world camera axes (+X forward, +Z up) to OpenGL (-Z forward, +Y up).
        orientation = orientations[tid] * wp.quatf(0.5, -0.5, -0.5, 0.5)
        output[0, tid] = wp.transformf(positions[tid], orientation)


[docs] class NewtonWarpRenderer(BaseRenderer): """Newton Warp backend for tiled camera rendering.""" RenderData = RenderData
[docs] def __init__(self, cfg: NewtonWarpRendererCfg): """Pre-physics initialization.""" self.cfg = cfg self.newton_sensor: newton.sensors.SensorTiledCamera | None = None # USD stage captured in ``prepare_cameras``; used by the segmentation mapper to read semantics. self._stage: Any = None # Shared, scene-static segmentation lookup builder, created lazily in ``create_render_data``. self._seg_mapper: NewtonSegmentationMapper | None = None sim = SimulationContext.instance() self.newton_cfg = NewtonBackendCfg(physics_cfg=sim.cfg.physics, device=sim.device) requires_stage, requires_model = REQUIRES_STAGE_AND_MODEL["newton_warp"] sim.requires_usd_stage |= requires_stage sim.requires_newton_model |= requires_model
def initialize(self) -> None: """Acquire the clone-built native resource and bind its SDP layout.""" sim = SimulationContext.instance() self.backend = sim.get_or_create_backend(self.newton_cfg) self._scene_data_provider = sim.get_scene_data_provider() self._transform_mapping = self._scene_data_provider.create_mapping(list(self.backend.model.body_label)) self.newton_sensor = newton.sensors.SensorTiledCamera( self.backend.model, default_render_config=newton.sensors.SensorTiledCamera.RenderConfig( enable_textures=self.cfg.enable_textures, enable_shadows=self.cfg.enable_shadows, enable_ambient_lighting=self.cfg.enable_ambient_lighting, enable_backface_culling=self.cfg.enable_backface_culling, max_distance=self.cfg.max_distance, render_order=newton.sensors.SensorTiledCamera.RenderOrder.TILED, tile_width=self.cfg.tile_rendering_width, tile_height=self.cfg.tile_rendering_height, ), ) if self.cfg.render_order == "pixel_priority": self.newton_sensor.default_render_config.render_order = ( newton.sensors.SensorTiledCamera.RenderOrder.PIXEL_PRIORITY ) elif self.cfg.render_order == "view_priority": self.newton_sensor.default_render_config.render_order = ( newton.sensors.SensorTiledCamera.RenderOrder.VIEW_PRIORITY ) if self.cfg.create_default_light: self.newton_sensor.utils.create_default_light(enable_shadows=self.cfg.enable_shadows) @property def visual_material_writer(self): """Return the shared Newton model color-writer factory.""" return self.backend.create_visual_material_writer def supported_output_types(self) -> dict[RenderBufferKind, RenderBufferSpec]: """Publish the per-output layout this Newton Warp backend writes. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.supported_output_types`.""" return self.cfg.supported_output_types() def prepare_cameras(self, stage: Any, spec: CameraRenderSpec) -> None: """Resolve the camera's PPISP cfg before rendering. :mod:`isaaclab.sensors.camera` does not depend on PPISP; the renderer owns the sentinel-resolution + cfg-normalization step. Newton has no USD-side overrides to author beyond this. Also captures the USD ``stage`` so the segmentation mapper can read the scene's :class:`UsdSemantics.LabelsAPI` labels when a segmentation output is requested. """ self._stage = stage if spec.cfg.isp_cfg is None: return try: from isaaclab_ppisp import resolve_and_normalize except ModuleNotFoundError as exc: _raise_missing_ppisp_error(exc) camera_prim_path = spec.camera_prim_paths[0] if spec.camera_prim_paths else None spec.cfg.isp_cfg = resolve_and_normalize(spec.cfg.isp_cfg, stage, camera_prim_path) def prepare_stage(self, stage: Any, num_envs: int) -> None: """No-op for Newton Warp - uses Newton scene directly without stage export. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.prepare_stage`.""" pass def create_render_data(self, spec: CameraRenderSpec) -> RenderData: """Create render data for the Newton tiled camera. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.create_render_data`.""" # Build the shared segmentation mapper and its per-kind lookup tables up-front for all # requested segmentation outputs. has_semantic = RenderBufferKind.SEMANTIC_SEGMENTATION in spec.cfg.data_types has_instance = RenderBufferKind.INSTANCE_SEGMENTATION in spec.cfg.data_types if (has_semantic or has_instance) and self._seg_mapper is None: plan = SimulationContext.instance().get_clone_plan() self._seg_mapper = NewtonSegmentationMapper(self.newton_sensor.model, self._stage, self.cfg, plan) if has_semantic: self._seg_mapper.build_mapping( RenderBufferKind.SEMANTIC_SEGMENTATION, bool(self.cfg.colorize_semantic_segmentation) ) if has_instance: self._seg_mapper.build_mapping( RenderBufferKind.INSTANCE_SEGMENTATION, bool(self.cfg.colorize_instance_segmentation) ) return RenderData(self.newton_sensor, spec, seg_mapper=self._seg_mapper, renderer_cfg=self.cfg) def set_outputs(self, render_data: RenderData, output_data: dict[str, ProxyArray]): """Store output buffers. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.set_outputs`.""" render_data.set_outputs(output_data) def update_transforms(self) -> None: """No-op: the shared sensor pipeline refreshes transforms immediately before rendering.""" pass def update_geometries(self) -> None: """No-op for Newton Warp - geometry is read directly from Newton state during render. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.update_geometries`.""" pass def update_camera( self, render_data: RenderData, positions: ProxyArray, orientations: ProxyArray, intrinsics: ProxyArray, ): """Update camera poses and initialize the ray field on first use. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.update_camera`.""" render_data.update(positions, orientations, intrinsics) def update_camera_intrinsics(self, render_data: RenderData, intrinsics: wp.array, parameters: wp.array): """Regenerate the shared native ray field in place, preserving captured render pointers.""" if render_data._distortion is not None: return # The camera retains its fixed OpenCV calibration. if intrinsics.shape[0] > 1: render_data._intrinsic_status.zero_() wp.launch( _check_shared_intrinsics, intrinsics.shape[0], [intrinsics, render_data._intrinsic_status], device=intrinsics.device, ) if render_data._intrinsic_status.numpy()[0]: raise ValueError( "Newton Warp requires identical camera intrinsics across environments: its native ray field " "is shared across worlds. Use a uniform calibration batch or a renderer with per-view intrinsics." ) wp.launch( _update_camera_rays, (render_data.height, render_data.width), [intrinsics, render_data.camera_rays], device=intrinsics.device, ) def render(self, render_data: RenderData): """Render and write to output buffers. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.render`.""" backend, provider = self.backend, self._scene_data_provider poses = SceneDataFormat.Transform() if provider.get_transforms(poses, mapping=self._transform_mapping, count=backend.model.body_count): backend.state_0.body_q = poses.transforms if backend.geometry_offsets: provider.get_geometry_points(output=backend.state_0.particle_q, offsets=backend.geometry_offsets) render_data.graph = NewtonQueries.run_query( self.backend, provider.backend.transforms_timestamp + provider.backend.geometry_timestamp, partial(self._launch_render, render_data), render_data.graph, # Native triangle-mesh updates read back indices and may allocate after pointer swaps. use_cuda_graph=self.cfg.use_cuda_graph and backend.model.tri_count == 0, ) # Post-render PPISP: HDR scene-linear → LDR RGBA. Source/destination # tensors were bound once in ``set_outputs``. if render_data.ppisp_pipeline is not None: render_data.ppisp_pipeline.apply( render_data._ppisp_hdr_source, render_data._ppisp_rgba_dest, ) def _launch_render(self, render_data: RenderData) -> None: """Launch the tiled-camera render kernels for sensor graph capture.""" # default_render_config is shared state across all Newton sensors, so set max_distance # immediately before each render call rather than once in create_render_data. self.newton_sensor.default_render_config.max_distance = ( render_data.far_clip if render_data.far_clip is not None else self.cfg.max_distance ) # Use the renderer's clear value to fill distance_to_camera background when it is the only # depth output requested. This avoids a post-render kernel pass for that common case. # Planar-depth outputs retain the clipping pass for non-positive projected depths as well as misses. _depth_kinds = set(render_data._depth_dests) _use_depth_clear = ( self.cfg.depth_clipping_behavior == "max" and render_data.far_clip is not None and render_data._RAY_DEPTH_KIND in _depth_kinds and not (_depth_kinds & render_data._PLANE_DEPTH_KINDS) ) self.newton_sensor.update( self.backend.state_0, render_data.camera_transforms, render_data.camera_rays, color_image=render_data.outputs.color_image, hdr_color_image=render_data.outputs.hdr_color_image, albedo_image=render_data.outputs.albedo_image, depth_image=render_data.outputs.depth_image, forward_depth_image=render_data.outputs.forward_depth_image, normal_image=render_data.outputs.normals_image, shape_index_image=render_data.outputs.shape_index_image, # ARGB 93% gray to improve visibility of dark objects and align with RTX renderer background clear_data=newton.sensors.SensorTiledCamera.ClearData( clear_color=render_data.clear_color, **({"clear_depth": render_data.far_clip} if _use_depth_clear else {}), ), kernel_block_dim=self.cfg.kernel_block_dim, ) if _depth_kinds & render_data._PLANE_DEPTH_KINDS: render_data._copy_plane_depth() render_data._apply_depth_clipping(self.cfg.depth_clipping_behavior) # Remap the shape-index buffer into the requested segmentation outputs. render_data._convert_segmentation() def read_output(self, render_data: RenderData, camera_data: CameraData) -> None: """Copy rendered outputs to the camera data buffers. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.read_output`.""" for output_name in camera_data.output: if output_name == "rgb": continue image_data = render_data.get_output(output_name) if image_data is not None: output_wp = camera_data.output[output_name].warp if image_data.ptr != output_wp.ptr: wp.copy(output_wp, image_data) # Publish the segmentation id-to-label metadata (idToLabels / idToSemantics) alongside the # pixel buffers. for output_name, info in render_data.segmentation_info().items(): camera_data.info[output_name] = info def cleanup(self, render_data: RenderData | None): """Release the camera's buffers and captured query. See :meth:`~isaaclab.renderers.base_renderer.BaseRenderer.cleanup`.""" if render_data: render_data.graph = None render_data.newton_sensor = None def close(self) -> None: """Release borrowed native handles and SDP bindings after camera cleanup.""" self.newton_sensor = self.backend = self._scene_data_provider = self._transform_mapping = None