Useful Snippets

Various examples of Isaac Sim Replicator snippets that can be run as Standalone Applications or from the UI using the Script Editor.

Annotator and Custom Writer Data from Multiple Cameras

Example on how to access data from multiple cameras in a scene using annotators or custom writers. The standalone example can also be run directly (on Windows use python.bat instead of python.sh):

./python.sh standalone_examples/api/isaacsim.replicator.examples/multi_camera.py

Standalone Application

Annotator and Custom Writer Data from Multiple Cameras

from isaacsim import SimulationApp

simulation_app = SimulationApp(launch_config={"headless": False})

import os
import omni.usd
import omni.kit
import omni.replicator.core as rep
from omni.replicator.core import AnnotatorRegistry, Writer
from PIL import Image
from pxr import UsdGeom, Sdf

NUM_FRAMES = 5

# Save rgb image to file
def save_rgb(rgb_data, file_name):
    rgb_img = Image.fromarray(rgb_data, "RGBA")
    rgb_img.save(file_name + ".png")


# Randomize cube color every frame using a replicator randomizer
def cube_color_randomizer():
    cube_prims = rep.get.prims(path_pattern="Cube")
    with cube_prims:
        rep.randomizer.color(colors=rep.distribution.uniform((0, 0, 0), (1, 1, 1)))
    return cube_prims.node


# Access data through a custom replicator writer
class MyWriter(Writer):
    def __init__(self, rgb: bool = True):
        self._frame_id = 0
        if rgb:
            self.annotators.append(AnnotatorRegistry.get_annotator("rgb"))
        # Create writer output directory
        self.file_path = os.path.join(os.getcwd(), "_out_mc_writer", "")
        print(f"Writing writer data to {self.file_path}")
        dir = os.path.dirname(self.file_path)
        os.makedirs(dir, exist_ok=True)

    def write(self, data):
        for annotator in data.keys():
            annotator_split = annotator.split("-")
            if len(annotator_split) > 1:
                render_product_name = annotator_split[-1]
            if annotator.startswith("rgb"):
                save_rgb(data[annotator], f"{self.file_path}/{render_product_name}_frame_{self._frame_id}")
        self._frame_id += 1


rep.WriterRegistry.register(MyWriter)

# Create a new stage with a dome light
omni.usd.get_context().new_stage()
stage = omni.usd.get_context().get_stage()
dome_light = stage.DefinePrim("/World/DomeLight", "DomeLight")
dome_light.CreateAttribute("inputs:intensity", Sdf.ValueTypeNames.Float).Set(900.0)

# Create cube
cube_prim = stage.DefinePrim("/World/Cube", "Cube")
UsdGeom.Xformable(cube_prim).AddTranslateOp().Set((0.0, 5.0, 1.0))

# Register cube color randomizer to trigger on every frame
rep.randomizer.register(cube_color_randomizer)
with rep.trigger.on_frame():
    rep.randomizer.cube_color_randomizer()

# Create cameras
camera_prim1 = stage.DefinePrim("/World/Camera1", "Camera")
UsdGeom.Xformable(camera_prim1).AddTranslateOp().Set((0.0, 10.0, 20.0))
UsdGeom.Xformable(camera_prim1).AddRotateXYZOp().Set((-15.0, 0.0, 0.0))

camera_prim2 = stage.DefinePrim("/World/Camera2", "Camera")
UsdGeom.Xformable(camera_prim2).AddTranslateOp().Set((-10.0, 15.0, 15.0))
UsdGeom.Xformable(camera_prim2).AddRotateXYZOp().Set((-45.0, 0.0, 45.0))

# Create render products
rp1 = rep.create.render_product(str(camera_prim1.GetPrimPath()), resolution=(320, 320))
rp2 = rep.create.render_product(str(camera_prim2.GetPrimPath()), resolution=(640, 640))
rp3 = rep.create.render_product("/OmniverseKit_Persp", (1024, 1024))

# Access the data through a custom writer
writer = rep.WriterRegistry.get("MyWriter")
writer.initialize(rgb=True)
writer.attach([rp1, rp2, rp3])

# Access the data through annotators
rgb_annotators = []
for rp in [rp1, rp2, rp3]:
    rgb = rep.AnnotatorRegistry.get_annotator("rgb")
    rgb.attach(rp)
    rgb_annotators.append(rgb)

# Create annotator output directory
file_path = os.path.join(os.getcwd(), "_out_mc_annot", "")
print(f"Writing annotator data to {file_path}")
dir = os.path.dirname(file_path)
os.makedirs(dir, exist_ok=True)

# Data will be captured manually using step
rep.orchestrator.set_capture_on_play(False)

for i in range(NUM_FRAMES):
    # The step function provides new data to the annotators, triggers the randomizers and the writer
    rep.orchestrator.step(rt_subframes=4)
    for j, rgb_annot in enumerate(rgb_annotators):
        save_rgb(rgb_annot.get_data(), f"{dir}/rp{j}_step_{i}")

simulation_app.close()

Script Editor

Annotator and Custom Writer Data from Multiple Cameras

import asyncio
import os
import omni.usd
import omni.kit
import omni.replicator.core as rep
from omni.replicator.core import AnnotatorRegistry, Writer
from PIL import Image
from pxr import UsdGeom, Sdf

NUM_FRAMES = 5

# Save rgb image to file
def save_rgb(rgb_data, file_name):
    rgb_img = Image.fromarray(rgb_data, "RGBA")
    rgb_img.save(file_name + ".png")


# Randomize cube color every frame using a replicator randomizer
def cube_color_randomizer():
    cube_prims = rep.get.prims(path_pattern="Cube")
    with cube_prims:
        rep.randomizer.color(colors=rep.distribution.uniform((0, 0, 0), (1, 1, 1)))
    return cube_prims.node


# Access data through a custom replicator writer
class MyWriter(Writer):
    def __init__(self, rgb: bool = True):
        self._frame_id = 0
        if rgb:
            self.annotators.append(AnnotatorRegistry.get_annotator("rgb"))
        # Create writer output directory
        self.file_path = os.path.join(os.getcwd(), "_out_mc_writer", "")
        print(f"Writing writer data to {self.file_path}")
        dir = os.path.dirname(self.file_path)
        os.makedirs(dir, exist_ok=True)

    def write(self, data):
        for annotator in data.keys():
            annotator_split = annotator.split("-")
            if len(annotator_split) > 1:
                render_product_name = annotator_split[-1]
            if annotator.startswith("rgb"):
                save_rgb(data[annotator], f"{self.file_path}/{render_product_name}_frame_{self._frame_id}")
        self._frame_id += 1


rep.WriterRegistry.register(MyWriter)

# Create a new stage with a dome light
omni.usd.get_context().new_stage()
stage = omni.usd.get_context().get_stage()
dome_light = stage.DefinePrim("/World/DomeLight", "DomeLight")
dome_light.CreateAttribute("inputs:intensity", Sdf.ValueTypeNames.Float).Set(900.0)

# Create cube
cube_prim = stage.DefinePrim("/World/Cube", "Cube")
UsdGeom.Xformable(cube_prim).AddTranslateOp().Set((0.0, 5.0, 1.0))

# Register cube color randomizer to trigger on every frame
rep.randomizer.register(cube_color_randomizer)
with rep.trigger.on_frame():
    rep.randomizer.cube_color_randomizer()

# Create cameras
camera_prim1 = stage.DefinePrim("/World/Camera1", "Camera")
UsdGeom.Xformable(camera_prim1).AddTranslateOp().Set((0.0, 10.0, 20.0))
UsdGeom.Xformable(camera_prim1).AddRotateXYZOp().Set((-15.0, 0.0, 0.0))

camera_prim2 = stage.DefinePrim("/World/Camera2", "Camera")
UsdGeom.Xformable(camera_prim2).AddTranslateOp().Set((-10.0, 15.0, 15.0))
UsdGeom.Xformable(camera_prim2).AddRotateXYZOp().Set((-45.0, 0.0, 45.0))

# Create render products
rp1 = rep.create.render_product(str(camera_prim1.GetPrimPath()), resolution=(320, 320))
rp2 = rep.create.render_product(str(camera_prim2.GetPrimPath()), resolution=(640, 640))
rp3 = rep.create.render_product("/OmniverseKit_Persp", (1024, 1024))

# Access the data through a custom writer
writer = rep.WriterRegistry.get("MyWriter")
writer.initialize(rgb=True)
writer.attach([rp1, rp2, rp3])

# Access the data through annotators
rgb_annotators = []
for rp in [rp1, rp2, rp3]:
    rgb = rep.AnnotatorRegistry.get_annotator("rgb")
    rgb.attach(rp)
    rgb_annotators.append(rgb)

# Create annotator output directory
file_path = os.path.join(os.getcwd(), "_out_mc_annot", "")
print(f"Writing annotator data to {file_path}")
dir = os.path.dirname(file_path)
os.makedirs(dir, exist_ok=True)

# Data will be captured manually using step
rep.orchestrator.set_capture_on_play(False)

async def run_example_async():
    for i in range(NUM_FRAMES):
        # The step function provides new data to the annotators, triggers the randomizers and the writer
        await rep.orchestrator.step_async(rt_subframes=4)
        for j, rgb_annot in enumerate(rgb_annotators):
            save_rgb(rgb_annot.get_data(), f"{dir}/rp{j}_step_{i}")


asyncio.ensure_future(run_example_async())

Synthetic Data Access at Specific Simulation Timepoints

Example on how to access synthetic data (rgb, semantic segmentation) from multiple cameras in a simulation scene at specific events using annotators or writers. The standalone example can also be run directly (on Windows use python.bat instead of python.sh):

./python.sh standalone_examples/api/isaacsim.replicator.examples/simulation_get_data.py

Standalone Application

Synthetic Data Access at Specific Simulation Timepoints

from isaacsim import SimulationApp

simulation_app = SimulationApp(launch_config={"renderer": "RayTracedLighting", "headless": False})

import json
import os

import carb.settings
import numpy as np
import omni
import omni.replicator.core as rep
from isaacsim.core.api import World
from isaacsim.core.api.objects import DynamicCuboid
from isaacsim.core.utils.semantics import add_update_semantics
from PIL import Image


# Util function to save rgb annotator data
def write_rgb_data(rgb_data, file_path):
    rgb_img = Image.fromarray(rgb_data, "RGBA")
    rgb_img.save(file_path + ".png")


# Util function to save semantic segmentation annotator data
def write_sem_data(sem_data, file_path):
    id_to_labels = sem_data["info"]["idToLabels"]
    with open(file_path + ".json", "w") as f:
        json.dump(id_to_labels, f)
    sem_image_data = np.frombuffer(sem_data["data"], dtype=np.uint8).reshape(*sem_data["data"].shape, -1)
    sem_img = Image.fromarray(sem_image_data, "RGBA")
    sem_img.save(file_path + ".png")


# Create a new stage with the default ground plane
omni.usd.get_context().new_stage()

# Setup the simulation world
world = World()
world.scene.add_default_ground_plane()
world.reset()

# Setting capture on play to False will prevent the replicator from capturing data each frame
carb.settings.get_settings().set("/omni/replicator/captureOnPlay", False)

# Create a camera and render product to collect the data from
cam = rep.create.camera(position=(5, 5, 5), look_at=(0, 0, 0))
rp = rep.create.render_product(cam, (512, 512))

# Set the output directory for the data
out_dir = os.path.join(os.getcwd(), "_out_sim_event")
os.makedirs(out_dir, exist_ok=True)
print(f"Outputting data to {out_dir}..")

# Example of using a writer to save the data
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(
    output_dir=f"{out_dir}/writer", rgb=True, semantic_segmentation=True, colorize_semantic_segmentation=True
)
writer.attach(rp)

# Run a preview to ensure the replicator graph is initialized
rep.orchestrator.preview()

# Example of accessing the data directly from annotators
rgb_annot = rep.AnnotatorRegistry.get_annotator("rgb")
rgb_annot.attach(rp)
sem_annot = rep.AnnotatorRegistry.get_annotator("semantic_segmentation", init_params={"colorize": True})
sem_annot.attach(rp)

# Spawn and drop a few cubes, capture data when they stop moving
for i in range(5):
    cuboid = world.scene.add(DynamicCuboid(prim_path=f"/World/Cuboid_{i}", name=f"Cuboid_{i}", position=(0, 0, 10 + i)))
    add_update_semantics(cuboid.prim, "Cuboid")

    for s in range(500):
        world.step(render=False)
        vel = np.linalg.norm(cuboid.get_linear_velocity())
        if vel < 0.1:
            print(f"Cube_{i} stopped moving after {s} simulation steps, writing data..")
            # Trigger the writer and update the annotators with new data
            rep.orchestrator.step(rt_subframes=4, delta_time=0.0, pause_timeline=False)
            write_rgb_data(rgb_annot.get_data(), f"{out_dir}/Cube_{i}_step_{s}_rgb")
            write_sem_data(sem_annot.get_data(), f"{out_dir}/Cube_{i}_step_{s}_sem")
            break

simulation_app.close()

Script Editor

Synthetic Data Access at Specific Simulation Timepoints

import asyncio
import json
import os

import carb.settings
import numpy as np
import omni
import omni.replicator.core as rep
from isaacsim.core.api import World
from isaacsim.core.api.objects import DynamicCuboid
from isaacsim.core.utils.semantics import add_update_semantics
from PIL import Image


# Util function to save rgb annotator data
def write_rgb_data(rgb_data, file_path):
    rgb_img = Image.fromarray(rgb_data, "RGBA")
    rgb_img.save(file_path + ".png")


# Util function to save semantic segmentation annotator data
def write_sem_data(sem_data, file_path):
    id_to_labels = sem_data["info"]["idToLabels"]
    with open(file_path + ".json", "w") as f:
        json.dump(id_to_labels, f)
    sem_image_data = np.frombuffer(sem_data["data"], dtype=np.uint8).reshape(*sem_data["data"].shape, -1)
    sem_img = Image.fromarray(sem_image_data, "RGBA")
    sem_img.save(file_path + ".png")


# Create a new stage with the default ground plane
omni.usd.get_context().new_stage()

# Setup the simulation world
world = World()
world.scene.add_default_ground_plane()


# Setting capture on play to False will prevent the replicator from capturing data each frame
carb.settings.get_settings().set("/omni/replicator/captureOnPlay", False)

# Create a camera and render product to collect the data from
cam = rep.create.camera(position=(5, 5, 5), look_at=(0, 0, 0))
rp = rep.create.render_product(cam, (512, 512))

# Set the output directory for the data
out_dir = os.path.join(os.getcwd(), "_out_sim_event")
os.makedirs(out_dir, exist_ok=True)
print(f"Outputting data to {out_dir}..")

# Example of using a writer to save the data
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(
    output_dir=f"{out_dir}/writer", rgb=True, semantic_segmentation=True, colorize_semantic_segmentation=True
)
writer.attach(rp)

# Run a preview to ensure the replicator graph is initialized
rep.orchestrator.preview()

# Example of accessing the data directly from annotators
rgb_annot = rep.AnnotatorRegistry.get_annotator("rgb")
rgb_annot.attach(rp)
sem_annot = rep.AnnotatorRegistry.get_annotator("semantic_segmentation", init_params={"colorize": True})
sem_annot.attach(rp)


async def run_example_async():
    await world.initialize_simulation_context_async()
    await world.reset_async()

    # Spawn and drop a few cubes, capture data when they stop moving
    for i in range(5):
        cuboid = world.scene.add(
            DynamicCuboid(prim_path=f"/World/Cuboid_{i}", name=f"Cuboid_{i}", position=(0, 0, 10 + i))
        )
        add_update_semantics(cuboid.prim, "Cuboid")

        for s in range(500):
            await omni.kit.app.get_app().next_update_async()
            vel = np.linalg.norm(cuboid.get_linear_velocity())
            if vel < 0.1:
                print(f"Cube_{i} stopped moving after {s} simulation steps, writing data..")
                # Trigger the writer and update the annotators with new data
                await rep.orchestrator.step_async(rt_subframes=4, delta_time=0.0, pause_timeline=False)
                write_rgb_data(rgb_annot.get_data(), f"{out_dir}/Cube_{i}_step_{s}_rgb")
                write_sem_data(sem_annot.get_data(), f"{out_dir}/Cube_{i}_step_{s}_sem")
                break


asyncio.ensure_future(run_example_async())

Custom Event Randomization and Writing

The following example showcases the use of custom events to trigger randomizations and data writing at various times throughout the simulation. The standalone example can also be run directly (on Windows use python.bat instead of python.sh):

./python.sh standalone_examples/api/isaacsim.replicator.examples/custom_event_and_write.py

Standalone Application

Custom Event Randomization and Writing

from isaacsim import SimulationApp

simulation_app = SimulationApp(launch_config={"headless": False})

import os

import omni.replicator.core as rep
import omni.usd

omni.usd.get_context().new_stage()
distance_light = rep.create.light(rotation=(315, 0, 0), intensity=4000, light_type="distant")

large_cube = rep.create.cube(scale=1.25, position=(1, 1, 0))
small_cube = rep.create.cube(scale=0.75, position=(-1, -1, 0))
large_cube_prim = large_cube.get_output_prims()["prims"][0]
small_cube_prim = small_cube.get_output_prims()["prims"][0]

rp = rep.create.render_product("/OmniverseKit_Persp", (512, 512))
writer = rep.WriterRegistry.get("BasicWriter")
out_dir = os.path.join(os.getcwd(), "_out_custom_event")
print(f"Writing data to {out_dir}")
writer.initialize(output_dir=out_dir, rgb=True)
writer.attach(rp)

with rep.trigger.on_custom_event(event_name="randomize_large_cube"):
    with large_cube:
        rep.randomizer.rotation()

with rep.trigger.on_custom_event(event_name="randomize_small_cube"):
    with small_cube:
        rep.randomizer.rotation()


def run_example():
    print(f"Randomizing small cube")
    rep.utils.send_og_event(event_name="randomize_small_cube")
    print("Capturing frame")
    rep.orchestrator.step(rt_subframes=8)

    print("Moving small cube")
    small_cube_prim.GetAttribute("xformOp:translate").Set((-2, -2, 0))
    print("Capturing frame")
    rep.orchestrator.step(rt_subframes=8)

    print(f"Randomizing large cube")
    rep.utils.send_og_event(event_name="randomize_large_cube")
    print("Capturing frame")
    rep.orchestrator.step(rt_subframes=8)

    print("Moving large cube")
    large_cube_prim.GetAttribute("xformOp:translate").Set((2, 2, 0))
    print("Capturing frame")
    rep.orchestrator.step(rt_subframes=8)

    # Wait until all the data is saved to disk
    rep.orchestrator.wait_until_complete()


run_example()

simulation_app.close()

Script Editor

Custom Event Randomization and Writing

import asyncio
import os

import omni.replicator.core as rep
import omni.usd

omni.usd.get_context().new_stage()
distance_light = rep.create.light(rotation=(315, 0, 0), intensity=4000, light_type="distant")

large_cube = rep.create.cube(scale=1.25, position=(1, 1, 0))
small_cube = rep.create.cube(scale=0.75, position=(-1, -1, 0))
large_cube_prim = large_cube.get_output_prims()["prims"][0]
small_cube_prim = small_cube.get_output_prims()["prims"][0]

rp = rep.create.render_product("/OmniverseKit_Persp", (512, 512))
writer = rep.WriterRegistry.get("BasicWriter")
out_dir = os.path.join(os.getcwd(), "_out_custom_event")
print(f"Writing data to {out_dir}")
writer.initialize(output_dir=out_dir, rgb=True)
writer.attach(rp)

with rep.trigger.on_custom_event(event_name="randomize_large_cube"):
    with large_cube:
        rep.randomizer.rotation()

with rep.trigger.on_custom_event(event_name="randomize_small_cube"):
    with small_cube:
        rep.randomizer.rotation()


async def run_example_async():
    print(f"Randomizing small cube")
    rep.utils.send_og_event(event_name="randomize_small_cube")
    print("Capturing frame")
    await rep.orchestrator.step_async(rt_subframes=8)

    print("Moving small cube")
    small_cube_prim.GetAttribute("xformOp:translate").Set((-2, -2, 0))
    print("Capturing frame")
    await rep.orchestrator.step_async(rt_subframes=8)

    print(f"Randomizing large cube")
    rep.utils.send_og_event(event_name="randomize_large_cube")
    print("Capturing frame")
    await rep.orchestrator.step_async(rt_subframes=8)

    print("Moving large cube")
    large_cube_prim.GetAttribute("xformOp:translate").Set((2, 2, 0))
    print("Capturing frame")
    await rep.orchestrator.step_async(rt_subframes=8)

    # Wait until all the data is saved to disk
    await rep.orchestrator.wait_until_complete_async()


asyncio.ensure_future(run_example_async())

Motion Blur

This example demonstrates how to capture motion blur data using RTX Real-Time and RTX Interactive (Path Tracing) rendering modes. For the RTX - Real-Time mode, details on the motion blur parameters can be found here. For the RTX – Interactive (Path Tracing) mode, motion blur is achieved by rendering multiple subframes (/omni/replicator/pathTracedMotionBlurSubSamples) and combining them to create the effect. The example uses animated and physics-enabled assets with synchronized motion. Keyframe animated assets can be advanced at any custom delta time due to their interpolated motion, whereas physics-enabled assets require a custom physics FPS to ensure motion samples at any custom delta time. The example showcases how to compute the target physics FPS, change it if needed, and restore the original physics FPS after capturing the motion blur.

The standalone example can also be run directly (on Windows use python.bat instead of python.sh):

./python.sh standalone_examples/api/isaacsim.replicator.examples/motion_blur.py

Standalone Application

Motion Blur

from isaacsim import SimulationApp

simulation_app = SimulationApp({"headless": False})

import os

import carb.settings
import omni.kit.app
import omni.replicator.core as rep
import omni.timeline
import omni.usd
from isaacsim.storage.native import get_assets_root_path
from pxr import PhysxSchema, Sdf, UsdGeom, UsdPhysics

# Paths to the animated and physics-ready assets
PHYSICS_ASSET_URL = "/Isaac/Props/YCB/Axis_Aligned_Physics/003_cracker_box.usd"
ANIM_ASSET_URL = "/Isaac/Props/YCB/Axis_Aligned/003_cracker_box.usd"

# -z velocities and start locations of the animated (left side) and physics (right side) assets (stage units/s)
ASSET_VELOCITIES = [0, 5, 10]
ASSET_X_MIRRORED_LOCATIONS = [(0.5, 0, 0.3), (0.3, 0, 0.3), (0.1, 0, 0.3)]

# Used to calculate how many frames to animate the assets to maintain the same velocity as the physics assets
ANIMATION_DURATION = 10

# Create a new stage with animated and physics-enabled assets with synchronized motion
def setup_stage():
    # Create new stage
    omni.usd.get_context().new_stage()
    stage = omni.usd.get_context().get_stage()
    timeline = omni.timeline.get_timeline_interface()
    timeline.set_end_time(ANIMATION_DURATION)

    # Create lights
    dome_light = stage.DefinePrim("/World/DomeLight", "DomeLight")
    dome_light.CreateAttribute("inputs:intensity", Sdf.ValueTypeNames.Float).Set(100.0)
    distant_light = stage.DefinePrim("/World/DistantLight", "DistantLight")
    if not distant_light.GetAttribute("xformOp:rotateXYZ"):
        UsdGeom.Xformable(distant_light).AddRotateXYZOp()
    distant_light.GetAttribute("xformOp:rotateXYZ").Set((-75, 0, 0))
    distant_light.CreateAttribute("inputs:intensity", Sdf.ValueTypeNames.Float).Set(2500)

    # Setup the physics assets with gravity disabled and the requested velocity
    assets_root_path = get_assets_root_path()
    physics_asset_url = assets_root_path + PHYSICS_ASSET_URL
    for loc, vel in zip(ASSET_X_MIRRORED_LOCATIONS, ASSET_VELOCITIES):
        prim = stage.DefinePrim(f"/World/physics_asset_{int(abs(vel))}", "Xform")
        prim.GetReferences().AddReference(physics_asset_url)
        if not prim.GetAttribute("xformOp:translate"):
            UsdGeom.Xformable(prim).AddTranslateOp()
        prim.GetAttribute("xformOp:translate").Set(loc)
        prim.GetAttribute("physxRigidBody:disableGravity").Set(True)
        prim.GetAttribute("physxRigidBody:angularDamping").Set(0.0)
        prim.GetAttribute("physxRigidBody:linearDamping").Set(0.0)
        prim.GetAttribute("physics:velocity").Set((0, 0, -vel))

    # Setup animated assets maintaining the same velocity as the physics assets
    anim_asset_url = assets_root_path + ANIM_ASSET_URL
    for loc, vel in zip(ASSET_X_MIRRORED_LOCATIONS, ASSET_VELOCITIES):
        start_loc = (-loc[0], loc[1], loc[2])
        prim = stage.DefinePrim(f"/World/anim_asset_{int(abs(vel))}", "Xform")
        prim.GetReferences().AddReference(anim_asset_url)
        if not prim.GetAttribute("xformOp:translate"):
            UsdGeom.Xformable(prim).AddTranslateOp()
        anim_distance = vel * ANIMATION_DURATION
        end_loc = (start_loc[0], start_loc[1], start_loc[2] - anim_distance)
        end_keyframe = timeline.get_time_codes_per_seconds() * ANIMATION_DURATION
        # Timesampled keyframe (animated) translation
        prim.GetAttribute("xformOp:translate").Set(start_loc, time=0)
        prim.GetAttribute("xformOp:translate").Set(end_loc, time=end_keyframe)


# Capture motion blur frames with the given delta time step and render mode
def run_motion_blur_example(num_frames=3, custom_delta_time=None, use_path_tracing=True, pt_subsamples=8, pt_spp=64):
    # Create a new stage with the assets
    setup_stage()
    stage = omni.usd.get_context().get_stage()

    # Set replicator settings (capture only on request and enable motion blur)
    carb.settings.get_settings().set("/omni/replicator/captureOnPlay", False)
    carb.settings.get_settings().set("/omni/replicator/captureMotionBlur", True)

    # Set motion blur settings based on the render mode
    if use_path_tracing:
        print(f"[MotionBlur] Setting PathTracing render mode motion blur settings")
        carb.settings.get_settings().set("/rtx/rendermode", "PathTracing")
        # (int): Total number of samples for each rendered pixel, per frame.
        carb.settings.get_settings().set("/rtx/pathtracing/spp", pt_spp)
        # (int): Maximum number of samples to accumulate per pixel. When this count is reached the rendering stops until a scene or setting change is detected, restarting the rendering process. Set to 0 to remove this limit.
        carb.settings.get_settings().set("/rtx/pathtracing/totalSpp", pt_spp)
        carb.settings.get_settings().set("/rtx/pathtracing/optixDenoiser/enabled", 0)
        # Number of sub samples to render if in PathTracing render mode and motion blur is enabled.
        carb.settings.get_settings().set("/omni/replicator/pathTracedMotionBlurSubSamples", pt_subsamples)
    else:
        print(f"[MotionBlur] Setting RayTracedLighting render mode motion blur settings")
        carb.settings.get_settings().set("/rtx/rendermode", "RayTracedLighting")
        # 0: Disabled, 1: TAA, 2: FXAA, 3: DLSS, 4:RTXAA
        carb.settings.get_settings().set("/rtx/post/aa/op", 2)
        # (float): The fraction of the largest screen dimension to use as the maximum motion blur diameter.
        carb.settings.get_settings().set("/rtx/post/motionblur/maxBlurDiameterFraction", 0.02)
        # (float): Exposure time fraction in frames (1.0 = one frame duration) to sample.
        carb.settings.get_settings().set("/rtx/post/motionblur/exposureFraction", 1.0)
        # (int): Number of samples to use in the filter. A higher number improves quality at the cost of performance.
        carb.settings.get_settings().set("/rtx/post/motionblur/numSamples", 8)

    # Setup camera and writer
    camera = rep.create.camera(position=(0, 1.5, 0), look_at=(0, 0, 0), name="MotionBlurCam")
    render_product = rep.create.render_product(camera, (1280, 720))
    basic_writer = rep.WriterRegistry.get("BasicWriter")
    delta_time_str = "None" if custom_delta_time is None else f"{custom_delta_time:.4f}"
    render_mode_str = f"pt_subsamples_{pt_subsamples}_spp_{pt_spp}" if use_path_tracing else "rt"
    output_directory = os.path.join(os.getcwd(), f"_out_motion_blur_dt_{delta_time_str}_{render_mode_str}")
    print(f"[MotionBlur] Output directory: {output_directory}")
    basic_writer.initialize(output_dir=output_directory, rgb=True)
    basic_writer.attach(render_product)

    # Run a few updates to make sure all materials are fully loaded for capture
    for _ in range(50):
        simulation_app.update()

    # Use the physics scene to modify the physics FPS (if needed) to guarantee motion samples at any custom delta time
    physx_scene = None
    for prim in stage.Traverse():
        if prim.IsA(UsdPhysics.Scene):
            physx_scene = PhysxSchema.PhysxSceneAPI.Apply(prim)
            break
    if physx_scene is None:
        print(f"[MotionBlur] Creating a new PhysicsScene")
        physics_scene = UsdPhysics.Scene.Define(stage, "/PhysicsScene")
        physx_scene = PhysxSchema.PhysxSceneAPI.Apply(stage.GetPrimAtPath("/PhysicsScene"))

    # Check the target physics depending on the custom delta time and the render mode
    target_physics_fps = stage.GetTimeCodesPerSecond() if custom_delta_time is None else 1 / custom_delta_time
    if use_path_tracing:
        target_physics_fps *= pt_subsamples

    # Check if the physics FPS needs to be increased to match the custom delta time
    orig_physics_fps = physx_scene.GetTimeStepsPerSecondAttr().Get()
    if target_physics_fps > orig_physics_fps:
        print(f"[MotionBlur] Changing physics FPS from {orig_physics_fps} to {target_physics_fps}")
        physx_scene.GetTimeStepsPerSecondAttr().Set(target_physics_fps)

    # Start the timeline for physics updates in the step function
    timeline = omni.timeline.get_timeline_interface()
    timeline.play()

    # Capture frames
    for i in range(num_frames):
        print(f"[MotionBlur] \tCapturing frame {i}")
        rep.orchestrator.step(delta_time=custom_delta_time)

    # Restore the original physics FPS
    if target_physics_fps > orig_physics_fps:
        print(f"[MotionBlur] Restoring physics FPS from {target_physics_fps} to {orig_physics_fps}")
        physx_scene.GetTimeStepsPerSecondAttr().Set(orig_physics_fps)

    # Switch back to the raytracing render mode
    if use_path_tracing:
        print(f"[MotionBlur] Restoring render mode to RayTracedLighting")
        carb.settings.get_settings().set("/rtx/rendermode", "RayTracedLighting")

    # Wait until all the data is saved to disk
    rep.orchestrator.wait_until_complete()


def run_motion_blur_examples():
    motion_blur_step_duration = [None, 1 / 30, 1 / 60, 1 / 240]
    for custom_delta_time in motion_blur_step_duration:
        # RayTracing examples
        run_motion_blur_example(custom_delta_time=custom_delta_time, use_path_tracing=False)
        # PathTracing examples
        spps = [32, 128]
        motion_blur_sub_samples = [4, 16]
        for motion_blur_sub_sample in motion_blur_sub_samples:
            for spp in spps:
                run_motion_blur_example(
                    custom_delta_time=custom_delta_time,
                    use_path_tracing=True,
                    pt_subsamples=motion_blur_sub_sample,
                    pt_spp=spp,
                )


run_motion_blur_examples()

simulation_app.close()

Script Editor

Motion Blur

import asyncio
import os

import carb.settings
import omni.kit.app
import omni.replicator.core as rep
import omni.timeline
import omni.usd
from isaacsim.storage.native import get_assets_root_path
from pxr import PhysxSchema, Sdf, UsdGeom, UsdPhysics

# Paths to the animated and physics-ready assets
PHYSICS_ASSET_URL = "/Isaac/Props/YCB/Axis_Aligned_Physics/003_cracker_box.usd"
ANIM_ASSET_URL = "/Isaac/Props/YCB/Axis_Aligned/003_cracker_box.usd"

# -z velocities and start locations of the animated (left side) and physics (right side) assets (stage units/s)
ASSET_VELOCITIES = [0, 5, 10]
ASSET_X_MIRRORED_LOCATIONS = [(0.5, 0, 0.3), (0.3, 0, 0.3), (0.1, 0, 0.3)]

# Used to calculate how many frames to animate the assets to maintain the same velocity as the physics assets
ANIMATION_DURATION = 10

# Create a new stage with animated and physics-enabled assets with synchronized motion
def setup_stage():
    # Create new stage
    omni.usd.get_context().new_stage()
    stage = omni.usd.get_context().get_stage()
    timeline = omni.timeline.get_timeline_interface()
    timeline.set_end_time(ANIMATION_DURATION)

    # Create lights
    dome_light = stage.DefinePrim("/World/DomeLight", "DomeLight")
    dome_light.CreateAttribute("inputs:intensity", Sdf.ValueTypeNames.Float).Set(100.0)
    distant_light = stage.DefinePrim("/World/DistantLight", "DistantLight")
    if not distant_light.GetAttribute("xformOp:rotateXYZ"):
        UsdGeom.Xformable(distant_light).AddRotateXYZOp()
    distant_light.GetAttribute("xformOp:rotateXYZ").Set((-75, 0, 0))
    distant_light.CreateAttribute("inputs:intensity", Sdf.ValueTypeNames.Float).Set(2500)

    # Setup the physics assets with gravity disabled and the requested velocity
    assets_root_path = get_assets_root_path()
    physics_asset_url = assets_root_path + PHYSICS_ASSET_URL
    for loc, vel in zip(ASSET_X_MIRRORED_LOCATIONS, ASSET_VELOCITIES):
        prim = stage.DefinePrim(f"/World/physics_asset_{int(abs(vel))}", "Xform")
        prim.GetReferences().AddReference(physics_asset_url)
        if not prim.GetAttribute("xformOp:translate"):
            UsdGeom.Xformable(prim).AddTranslateOp()
        prim.GetAttribute("xformOp:translate").Set(loc)
        prim.GetAttribute("physxRigidBody:disableGravity").Set(True)
        prim.GetAttribute("physxRigidBody:angularDamping").Set(0.0)
        prim.GetAttribute("physxRigidBody:linearDamping").Set(0.0)
        prim.GetAttribute("physics:velocity").Set((0, 0, -vel))

    # Setup animated assets maintaining the same velocity as the physics assets
    anim_asset_url = assets_root_path + ANIM_ASSET_URL
    for loc, vel in zip(ASSET_X_MIRRORED_LOCATIONS, ASSET_VELOCITIES):
        start_loc = (-loc[0], loc[1], loc[2])
        prim = stage.DefinePrim(f"/World/anim_asset_{int(abs(vel))}", "Xform")
        prim.GetReferences().AddReference(anim_asset_url)
        if not prim.GetAttribute("xformOp:translate"):
            UsdGeom.Xformable(prim).AddTranslateOp()
        anim_distance = vel * ANIMATION_DURATION
        end_loc = (start_loc[0], start_loc[1], start_loc[2] - anim_distance)
        end_keyframe = timeline.get_time_codes_per_seconds() * ANIMATION_DURATION
        # Timesampled keyframe (animated) translation
        prim.GetAttribute("xformOp:translate").Set(start_loc, time=0)
        prim.GetAttribute("xformOp:translate").Set(end_loc, time=end_keyframe)


# Capture motion blur frames with the given delta time step and render mode
async def run_motion_blur_example_async(
    num_frames=3, custom_delta_time=None, use_path_tracing=True, pt_subsamples=8, pt_spp=64
):
    # Create a new stage with the assets
    setup_stage()
    stage = omni.usd.get_context().get_stage()

    # Set replicator settings (capture only on request and enable motion blur)
    carb.settings.get_settings().set("/omni/replicator/captureOnPlay", False)
    carb.settings.get_settings().set("/omni/replicator/captureMotionBlur", True)

    # Set motion blur settings based on the render mode
    if use_path_tracing:
        print(f"[MotionBlur] Setting PathTracing render mode motion blur settings")
        carb.settings.get_settings().set("/rtx/rendermode", "PathTracing")
        # (int): Total number of samples for each rendered pixel, per frame.
        carb.settings.get_settings().set("/rtx/pathtracing/spp", pt_spp)
        # (int): Maximum number of samples to accumulate per pixel. When this count is reached the rendering stops until a scene or setting change is detected, restarting the rendering process. Set to 0 to remove this limit.
        carb.settings.get_settings().set("/rtx/pathtracing/totalSpp", pt_spp)
        carb.settings.get_settings().set("/rtx/pathtracing/optixDenoiser/enabled", 0)
        # Number of sub samples to render if in PathTracing render mode and motion blur is enabled.
        carb.settings.get_settings().set("/omni/replicator/pathTracedMotionBlurSubSamples", pt_subsamples)
    else:
        print(f"[MotionBlur] Setting RayTracedLighting render mode motion blur settings")
        carb.settings.get_settings().set("/rtx/rendermode", "RayTracedLighting")
        # 0: Disabled, 1: TAA, 2: FXAA, 3: DLSS, 4:RTXAA
        carb.settings.get_settings().set("/rtx/post/aa/op", 2)
        # (float): The fraction of the largest screen dimension to use as the maximum motion blur diameter.
        carb.settings.get_settings().set("/rtx/post/motionblur/maxBlurDiameterFraction", 0.02)
        # (float): Exposure time fraction in frames (1.0 = one frame duration) to sample.
        carb.settings.get_settings().set("/rtx/post/motionblur/exposureFraction", 1.0)
        # (int): Number of samples to use in the filter. A higher number improves quality at the cost of performance.
        carb.settings.get_settings().set("/rtx/post/motionblur/numSamples", 8)

    # Setup camera and writer
    camera = rep.create.camera(position=(0, 1.5, 0), look_at=(0, 0, 0), name="MotionBlurCam")
    render_product = rep.create.render_product(camera, (1280, 720))
    basic_writer = rep.WriterRegistry.get("BasicWriter")
    delta_time_str = "None" if custom_delta_time is None else f"{custom_delta_time:.4f}"
    render_mode_str = f"pt_subsamples_{pt_subsamples}_spp_{pt_spp}" if use_path_tracing else "rt"
    output_directory = os.path.join(os.getcwd(), f"_out_motion_blur_dt_{delta_time_str}_{render_mode_str}")
    print(f"[MotionBlur] Output directory: {output_directory}")
    basic_writer.initialize(output_dir=output_directory, rgb=True)
    basic_writer.attach(render_product)

    # Run a few updates to make sure all materials are fully loaded for capture
    for _ in range(50):
        await omni.kit.app.get_app().next_update_async()

    # Use the physics scene to modify the physics FPS (if needed) to guarantee motion samples at any custom delta time
    physx_scene = None
    for prim in stage.Traverse():
        if prim.IsA(UsdPhysics.Scene):
            physx_scene = PhysxSchema.PhysxSceneAPI.Apply(prim)
            break
    if physx_scene is None:
        print(f"[MotionBlur] Creating a new PhysicsScene")
        physics_scene = UsdPhysics.Scene.Define(stage, "/PhysicsScene")
        physx_scene = PhysxSchema.PhysxSceneAPI.Apply(stage.GetPrimAtPath("/PhysicsScene"))

    # Check the target physics depending on the custom delta time and the render mode
    target_physics_fps = stage.GetTimeCodesPerSecond() if custom_delta_time is None else 1 / custom_delta_time
    if use_path_tracing:
        target_physics_fps *= pt_subsamples

    # Check if the physics FPS needs to be increased to match the custom delta time
    orig_physics_fps = physx_scene.GetTimeStepsPerSecondAttr().Get()
    if target_physics_fps > orig_physics_fps:
        print(f"[MotionBlur] Changing physics FPS from {orig_physics_fps} to {target_physics_fps}")
        physx_scene.GetTimeStepsPerSecondAttr().Set(target_physics_fps)

    # Start the timeline for physics updates in the step function
    timeline = omni.timeline.get_timeline_interface()
    timeline.play()

    # Capture frames
    for i in range(num_frames):
        print(f"[MotionBlur] \tCapturing frame {i}")
        await rep.orchestrator.step_async(delta_time=custom_delta_time)

    # Restore the original physics FPS
    if target_physics_fps > orig_physics_fps:
        print(f"[MotionBlur] Restoring physics FPS from {target_physics_fps} to {orig_physics_fps}")
        physx_scene.GetTimeStepsPerSecondAttr().Set(orig_physics_fps)

    # Switch back to the raytracing render mode
    if use_path_tracing:
        print(f"[MotionBlur] Restoring render mode to RayTracedLighting")
        carb.settings.get_settings().set("/rtx/rendermode", "RayTracedLighting")

    # Wait until all the data is saved to disk
    await rep.orchestrator.wait_until_complete_async()


async def run_motion_blur_examples_async():
    motion_blur_step_duration = [None, 1 / 30, 1 / 60, 1 / 240]
    for custom_delta_time in motion_blur_step_duration:
        # RayTracing examples
        await run_motion_blur_example_async(custom_delta_time=custom_delta_time, use_path_tracing=False)
        # PathTracing examples
        spps = [32, 128]
        motion_blur_sub_samples = [4, 16]
        for motion_blur_sub_sample in motion_blur_sub_samples:
            for spp in spps:
                await run_motion_blur_example_async(
                    custom_delta_time=custom_delta_time,
                    use_path_tracing=True,
                    pt_subsamples=motion_blur_sub_sample,
                    pt_spp=spp,
                )


asyncio.ensure_future(run_motion_blur_examples_async())

Subscribers and Events at Custom FPS

Examples of subscribing to various events (such as stage, physics, and render/app), setting custom update rates, and adjusting various related settings. The standalone example can also be run directly (on Windows use python.bat instead of python.sh):

./python.sh standalone_examples/api/isaacsim.replicator.examples/subscribers_and_events.py

Standalone Application

Subscribers and Events at Custom FPS

from isaacsim import SimulationApp

simulation_app = SimulationApp({"headless": False})

import asyncio
import math
import time

import carb.eventdispatcher
import carb.events
import carb.settings
import omni.kit.app
import omni.physx
import omni.timeline
import omni.usd
from pxr import PhysxSchema, UsdPhysics

# TIMELINE / STAGE
USE_CUSTOM_TIMELINE_SETTINGS = False
USE_FIXED_TIME_STEPPING = False
PLAY_EVERY_FRAME = True
PLAY_DELAY_COMPENSATION = 0.0
SUBSAMPLE_RATE = 1
STAGE_FPS = 30.0

# PHYSX
USE_CUSTOM_PHYSX_FPS = False
PHYSX_FPS = 60.0
MIN_SIM_FPS = 30

# Simulations can also be enabled/disabled at runtime
DISABLE_SIMULATIONS = False

# APP / RENDER
LIMIT_APP_FPS = False
APP_FPS = 120

# Duration after which to clear subscribers and print the cached events
SUBSCRIBER_WALL_TIME_LIMIT_SEC = 0.5
PRINT_EVENTS = True


def on_timeline_event(event: omni.timeline.TimelineEventType):
    global wall_start_time
    global timeline_sub
    global timeline_events
    elapsed_wall_time = time.time() - wall_start_time

    # Cache only time advance events
    if event.type == omni.timeline.TimelineEventType.CURRENT_TIME_TICKED.value:
        event_name = omni.timeline.TimelineEventType(event.type).name
        event_payload = event.payload
        timeline_events.append((elapsed_wall_time, event_name, event_payload))

    # Clear subscriber and print cached events
    if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
        if timeline_sub is not None:
            timeline_sub.unsubscribe()
            timeline_sub = None
        num_events = len(timeline_events)
        fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
        print(f"[timeline] captured {num_events} events with aprox {fps} FPS")
        if PRINT_EVENTS:
            for i, (wall_time, event_name, payload) in enumerate(timeline_events):
                print(f"\t[timeline][{i}]\ttime={wall_time:.4f};\tevent={event_name};\tpayload={payload}")


def on_physics_step(dt: float):
    global wall_start_time
    global physx_events
    global physx_sub
    elapsed_wall_time = time.time() - wall_start_time

    # Cache physics events
    physx_events.append((elapsed_wall_time, dt))

    # Clear subscriber and print cached events
    if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
        # Physics unsubscription needs to be defered from the callback function
        # see: '[Error] [omni.physx.plugin] Subscription cannot be changed during the event call'
        async def clear_physx_sub_async():
            global physx_sub
            if physx_sub is not None:
                physx_sub.unsubscribe()
                physx_sub = None

        asyncio.ensure_future(clear_physx_sub_async())
        num_events = len(physx_events)
        fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
        print(f"[physics] captured {num_events} events with aprox {fps} FPS")
        if PRINT_EVENTS:
            for i, (wall_time, dt) in enumerate(physx_events):
                print(f"\t[physics][{i}]\ttime={wall_time:.4f};\tdt={dt};")


def on_stage_render_event(event: carb.eventdispatcher.Event):
    global wall_start_time
    global stage_render_sub
    global stage_render_events
    elapsed_wall_time = time.time() - wall_start_time

    event_name = event.event_name
    event_payload = event.payload
    stage_render_events.append((elapsed_wall_time, event_name, event_payload))

    if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
        if stage_render_sub is not None:
            stage_render_sub.reset()
            stage_render_sub = None
        num_events = len(stage_render_events)
        fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
        print(f"[stage render] captured {num_events} events with aprox {fps} FPS")
        if PRINT_EVENTS:
            for i, (wall_time, event_name, payload) in enumerate(stage_render_events):
                print(f"\t[stage render][{i}]\ttime={wall_time:.4f};\tevent={event_name};\tpayload={payload}")


def on_app_update(event: carb.eventdispatcher.Event):
    global wall_start_time
    global app_sub
    global app_update_events
    elapsed_wall_time = time.time() - wall_start_time

    event_name = event.event_name
    event_payload = event.payload
    app_update_events.append((elapsed_wall_time, event_name, event_payload))

    if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
        if app_sub is not None:
            app_sub.reset()
            app_sub = None
        num_events = len(app_update_events)
        fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
        print(f"[app] captured {num_events} events with aprox {fps} FPS")
        if PRINT_EVENTS:
            for i, (wall_time, event_name, payload) in enumerate(app_update_events):
                print(f"\t[app][{i}]\ttime={wall_time:.4f};\tevent={event_name};\tpayload={payload}")


stage = omni.usd.get_context().get_stage()
timeline = omni.timeline.get_timeline_interface()


if USE_CUSTOM_TIMELINE_SETTINGS:
    # Ideal to make simulation and animation synchronized.
    # Default: True in editor, False in standalone.
    # NOTE:
    # - It may limit the frame rate (see 'timeline.set_play_every_frame') such that the elapsed wall clock time matches the frame's delta time.
    # - If the app runs slower than this, animation playback may slow down (see 'CompensatePlayDelayInSecs').
    # - For performance benchmarks, turn this off or set a very high target in `timeline.set_target_framerate`
    carb.settings.get_settings().set("/app/player/useFixedTimeStepping", USE_FIXED_TIME_STEPPING)

    # This compensates for frames that require more computation time than the frame's fixed delta time, by temporarily speeding up playback.
    # The parameter represents the length of these "faster" playback periods, which means that it must be larger than the fixed frame time to take effect.
    # Default: 0.0
    # NOTE:
    # - only effective if `useFixedTimeStepping` is set to True
    # - setting a large value results in long fast playback after a huge lag spike
    carb.settings.get_settings().set("/app/player/CompensatePlayDelayInSecs", PLAY_DELAY_COMPENSATION)

    # If set to True, no frames are skipped and in every frame time advances by `1 / TimeCodesPerSecond`.
    # Default: False
    # NOTE:
    # - only effective if `useFixedTimeStepping` is set to True
    # - simulation is usually faster than real-time and processing is only limited by the frame rate of the runloop
    # - useful for recording
    # - same as `carb.settings.get_settings().set("/app/player/useFastMode", PLAY_EVERY_FRAME)`
    timeline.set_play_every_frame(PLAY_EVERY_FRAME)

    # Timeline sub-stepping, i.e. how many times updates are called (update events are dispatched) each frame.
    # Default: 1
    # NOTE: same as `carb.settings.get_settings().set("/app/player/timelineSubsampleRate", SUBSAMPLE_RATE)`
    timeline.set_ticks_per_frame(SUBSAMPLE_RATE)

    # Time codes per second for the stage
    # NOTE: same as `stage.SetTimeCodesPerSecond(STAGE_FPS)` and `carb.settings.get_settings().set("/app/stage/timeCodesPerSecond", STAGE_FPS)`
    timeline.set_time_codes_per_second(STAGE_FPS)


# Create a PhysX scene to set the physics time step
if USE_CUSTOM_PHYSX_FPS:
    physx_scene = None
    for prim in stage.Traverse():
        if prim.IsA(UsdPhysics.Scene):
            physx_scene = PhysxSchema.PhysxSceneAPI.Apply(prim)
            break
    if physx_scene is None:
        physics_scene = UsdPhysics.Scene.Define(stage, "/PhysicsScene")
        physx_scene = PhysxSchema.PhysxSceneAPI.Apply(stage.GetPrimAtPath("/PhysicsScene"))

    # Time step for the physics simulation
    # Default: 60.0
    physx_scene.GetTimeStepsPerSecondAttr().Set(PHYSX_FPS)

    # Minimum simulation frequency to prevent clamping; if the frame rate drops below this,
    # physics steps are discarded to avoid app slowdown if the overall frame rate is too low.
    # Default: 30.0
    # NOTE: Matching `minFrameRate` with `TimeStepsPerSecond` ensures a single physics step per update.
    carb.settings.get_settings().set("/persistent/simulation/minFrameRate", MIN_SIM_FPS)


# Throttle Render/UI/Main thread update rate
if LIMIT_APP_FPS:
    # Enable rate limiting of the main run loop (UI, rendering, etc.)
    # Default: False
    carb.settings.get_settings().set("/app/runLoops/main/rateLimitEnabled", LIMIT_APP_FPS)

    # FPS limit of the main run loop (UI, rendering, etc.)
    # Default: 120
    # NOTE: disabled if `/app/player/useFixedTimeStepping` is False
    carb.settings.get_settings().set("/app/runLoops/main/rateLimitFrequency", int(APP_FPS))


# Simulations can be selectively disabled (or toggled at specific times)
if DISABLE_SIMULATIONS:
    carb.settings.get_settings().set("/app/player/playSimulations", False)


# Start the timeline
timeline.set_current_time(0)
timeline.set_end_time(SUBSCRIBER_WALL_TIME_LIMIT_SEC + 1)
timeline.set_looping(False)
timeline.play()
timeline.commit()
wall_start_time = time.time()

# Subscribe and cache various events for a limited duration
timeline_events = []
timeline_sub = timeline.get_timeline_event_stream().create_subscription_to_pop(on_timeline_event)
physx_events = []
physx_sub = omni.physx.get_physx_interface().subscribe_physics_step_events(on_physics_step)
stage_render_events = []
stage_render_sub = carb.eventdispatcher.get_eventdispatcher().observe_event(
    event_name=omni.usd.get_context().stage_rendering_event_name(omni.usd.StageRenderingEventType.NEW_FRAME, True),
    on_event=on_stage_render_event,
    observer_name="subscribers_and_events.on_stage_render_event",
)
app_update_events = []
app_sub = carb.eventdispatcher.get_eventdispatcher().observe_event(
    event_name=omni.kit.app.GLOBAL_EVENT_UPDATE,
    on_event=on_app_update,
    observer_name="subscribers_and_events.on_app_update",
)

# Run the application for a while to trigger the events, with a buffer to ensure subscribers have enough wall-clock time
num_app_updates = int(math.ceil(SUBSCRIBER_WALL_TIME_LIMIT_SEC * STAGE_FPS * 4))
for _ in range(num_app_updates):
    simulation_app.update()

print(f"Finished running the application for {num_app_updates} updates.")
print(f"Wall time: {time.time() - wall_start_time:.4f} seconds")
print(f"Number of timeline events: {len(timeline_events)}")
print(f"Number of physics events: {len(physx_events)}")
print(f"Number of stage render events: {len(stage_render_events)}")
print(f"Number of app update events: {len(app_update_events)}")

simulation_app.close()

Script Editor

Subscribers and Events at Custom FPS

import asyncio
import math
import time

import carb.eventdispatcher
import carb.events
import carb.settings
import omni.kit.app
import omni.physx
import omni.timeline
import omni.usd
from pxr import PhysxSchema, UsdPhysics

# TIMELINE / STAGE
USE_CUSTOM_TIMELINE_SETTINGS = False
USE_FIXED_TIME_STEPPING = False
PLAY_EVERY_FRAME = True
PLAY_DELAY_COMPENSATION = 0.0
SUBSAMPLE_RATE = 1
STAGE_FPS = 30.0

# PHYSX
USE_CUSTOM_PHYSX_FPS = False
PHYSX_FPS = 60.0
MIN_SIM_FPS = 30

# Simulations can also be enabled/disabled at runtime
DISABLE_SIMULATIONS = False

# APP / RENDER
LIMIT_APP_FPS = False
APP_FPS = 120

# Duration after which to clear subscribers and print the cached events
SUBSCRIBER_WALL_TIME_LIMIT_SEC = 0.5
PRINT_EVENTS = True


async def example_async():
    def on_timeline_event(event: omni.timeline.TimelineEventType):
        nonlocal wall_start_time
        nonlocal timeline_sub
        nonlocal timeline_events
        elapsed_wall_time = time.time() - wall_start_time

        # Cache only time advance events
        if event.type == omni.timeline.TimelineEventType.CURRENT_TIME_TICKED.value:
            event_name = omni.timeline.TimelineEventType(event.type).name
            event_payload = event.payload
            timeline_events.append((elapsed_wall_time, event_name, event_payload))

        # Clear subscriber and print cached events
        if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
            if timeline_sub is not None:
                timeline_sub.unsubscribe()
                timeline_sub = None
            num_events = len(timeline_events)
            fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
            print(f"[timeline] captured {num_events} events with aprox {fps} FPS")
            if PRINT_EVENTS:
                for i, (wall_time, event_name, payload) in enumerate(timeline_events):
                    print(
                        f"\t[timeline][{i}]\ttime={wall_time:.4f};\tevent={event_name};\tpayload={payload}"
                    )

    def on_physics_step(dt: float):
        nonlocal wall_start_time
        nonlocal physx_events
        nonlocal physx_sub
        elapsed_wall_time = time.time() - wall_start_time

        # Cache physics events
        physx_events.append((elapsed_wall_time, dt))

        # Clear subscriber and print cached events
        if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
            # Physics unsubscription needs to be deferred from the callback function
            # see: '[Error] [omni.physx.plugin] Subscription cannot be changed during the event call'
            async def clear_physx_sub_async():
                nonlocal physx_sub
                if physx_sub is not None:
                    physx_sub.unsubscribe()
                    physx_sub = None

            asyncio.ensure_future(clear_physx_sub_async())
            num_events = len(physx_events)
            fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
            print(f"[physics] captured {num_events} events with aprox {fps} FPS")
            if PRINT_EVENTS:
                for i, (wall_time, dt) in enumerate(physx_events):
                    print(f"\t[physics][{i}]\ttime={wall_time:.4f};\tdt={dt};")

    def on_stage_render_event(event: carb.eventdispatcher.Event):
        nonlocal wall_start_time
        nonlocal stage_render_sub
        nonlocal stage_render_events
        elapsed_wall_time = time.time() - wall_start_time

        event_name = event.event_name
        event_payload = event.payload
        stage_render_events.append((elapsed_wall_time, event_name, event_payload))

        if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
            if stage_render_sub is not None:
                stage_render_sub.reset
                stage_render_sub = None
            num_events = len(stage_render_events)
            fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
            print(f"[stage render] captured {num_events} events with aprox {fps} FPS")
            if PRINT_EVENTS:
                for i, (wall_time, event_name, payload) in enumerate(
                    stage_render_events
                ):
                    print(
                        f"\t[stage render][{i}]\ttime={wall_time:.4f};\tevent={event_name};\tpayload={payload}"
                    )

    def on_app_update(event: carb.eventdispatcher.Event):
        nonlocal wall_start_time
        nonlocal app_sub
        nonlocal app_update_events
        elapsed_wall_time = time.time() - wall_start_time

        # Cache app update events
        event_name = event.event_name
        event_payload = event.payload
        app_update_events.append((elapsed_wall_time, event_name, event_payload))

        if elapsed_wall_time > SUBSCRIBER_WALL_TIME_LIMIT_SEC:
            if app_sub is not None:
                app_sub.reset()
                app_sub = None
            num_events = len(app_update_events)
            fps = num_events / SUBSCRIBER_WALL_TIME_LIMIT_SEC
            print(f"[app] captured {num_events} events with aprox {fps} FPS")
            if PRINT_EVENTS:
                for i, (wall_time, event_name, payload) in enumerate(app_update_events):
                    print(
                        f"\t[app][{i}]\ttime={wall_time:.4f};\tevent={event_name};\tpayload={payload}"
                    )

    stage = omni.usd.get_context().get_stage()
    timeline = omni.timeline.get_timeline_interface()

    if USE_CUSTOM_TIMELINE_SETTINGS:
        # Ideal to make simulation and animation synchronized.
        # Default: True in editor, False in standalone.
        # NOTE:
        # - It may limit the frame rate (see 'timeline.set_play_every_frame') such that the elapsed wall clock time matches the frame's delta time.
        # - If the app runs slower than this, animation playback may slow down (see 'CompensatePlayDelayInSecs').
        # - For performance benchmarks, turn this off or set a very high target in `timeline.set_target_framerate`
        carb.settings.get_settings().set(
            "/app/player/useFixedTimeStepping", USE_FIXED_TIME_STEPPING
        )

        # This compensates for frames that require more computation time than the frame's fixed delta time, by temporarily speeding up playback.
        # The parameter represents the length of these "faster" playback periods, which means that it must be larger than the fixed frame time to take effect.
        # Default: 0.0
        # NOTE:
        # - only effective if `useFixedTimeStepping` is set to True
        # - setting a large value results in long fast playback after a huge lag spike
        carb.settings.get_settings().set(
            "/app/player/CompensatePlayDelayInSecs", PLAY_DELAY_COMPENSATION
        )

        # If set to True, no frames are skipped and in every frame time advances by `1 / TimeCodesPerSecond`.
        # Default: False
        # NOTE:
        # - only effective if `useFixedTimeStepping` is set to True
        # - simulation is usually faster than real-time and processing is only limited by the frame rate of the runloop
        # - useful for recording
        # - same as `carb.settings.get_settings().set("/app/player/useFastMode", PLAY_EVERY_FRAME)`
        timeline.set_play_every_frame(PLAY_EVERY_FRAME)

        # Timeline sub-stepping, i.e. how many times updates are called (update events are dispatched) each frame.
        # Default: 1
        # NOTE: same as `carb.settings.get_settings().set("/app/player/timelineSubsampleRate", SUBSAMPLE_RATE)`
        timeline.set_ticks_per_frame(SUBSAMPLE_RATE)

        # Time codes per second for the stage
        # NOTE: same as `stage.SetTimeCodesPerSecond(STAGE_FPS)` and `carb.settings.get_settings().set("/app/stage/timeCodesPerSecond", STAGE_FPS)`
        timeline.set_time_codes_per_second(STAGE_FPS)

    # Create a PhysX scene to set the physics time step
    if USE_CUSTOM_PHYSX_FPS:
        physx_scene = None
        for prim in stage.Traverse():
            if prim.IsA(UsdPhysics.Scene):
                physx_scene = PhysxSchema.PhysxSceneAPI.Apply(prim)
                break
        if physx_scene is None:
            physics_scene = UsdPhysics.Scene.Define(stage, "/PhysicsScene")
            physx_scene = PhysxSchema.PhysxSceneAPI.Apply(
                stage.GetPrimAtPath("/PhysicsScene")
            )

        # Time step for the physics simulation
        # Default: 60.0
        physx_scene.GetTimeStepsPerSecondAttr().Set(PHYSX_FPS)

        # Minimum simulation frequency to prevent clamping; if the frame rate drops below this,
        # physics steps are discarded to avoid app slowdown if the overall frame rate is too low.
        # Default: 30.0
        # NOTE: Matching `minFrameRate` with `TimeStepsPerSecond` ensures a single physics step per update.
        carb.settings.get_settings().set(
            "/persistent/simulation/minFrameRate", MIN_SIM_FPS
        )

    # Throttle Render/UI/Main thread update rate
    if LIMIT_APP_FPS:
        # Enable rate limiting of the main run loop (UI, rendering, etc.)
        # Default: False
        carb.settings.get_settings().set(
            "/app/runLoops/main/rateLimitEnabled", LIMIT_APP_FPS
        )

        # FPS limit of the main run loop (UI, rendering, etc.)
        # Default: 120
        # NOTE: disabled if `/app/player/useFixedTimeStepping` is False
        carb.settings.get_settings().set(
            "/app/runLoops/main/rateLimitFrequency", int(APP_FPS)
        )

    # Simulations can be selectively disabled (or toggled at specific times)
    if DISABLE_SIMULATIONS:
        carb.settings.get_settings().set("/app/player/playSimulations", False)

    # Start the timeline
    timeline.set_current_time(0)
    timeline.set_end_time(SUBSCRIBER_WALL_TIME_LIMIT_SEC + 1)
    timeline.set_looping(False)
    timeline.play()
    timeline.commit()
    wall_start_time = time.time()

    # Subscribe and cache various events for a limited duration
    timeline_events = []
    timeline_sub = timeline.get_timeline_event_stream().create_subscription_to_pop(
        on_timeline_event
    )
    physx_events = []
    physx_sub = omni.physx.get_physx_interface().subscribe_physics_step_events(
        on_physics_step
    )
    stage_render_events = []
    stage_render_sub = carb.eventdispatcher.get_eventdispatcher().observe_event(
        event_name=omni.usd.get_context().stage_rendering_event_name(
            omni.usd.StageRenderingEventType.NEW_FRAME, True
        ),
        on_event=on_stage_render_event,
        observer_name="subscribers_and_events.on_stage_render_event",
    )

    app_update_events = []
    app_sub = carb.eventdispatcher.get_eventdispatcher().observe_event(
        event_name=omni.kit.app.GLOBAL_EVENT_UPDATE,
        on_event=on_app_update,
        observer_name="subscribers_and_events.on_app_update",
    )

    # Run the application for a while to trigger the events, with a buffer to ensure subscribers have enough wall-clock time
    num_app_updates = int(math.ceil(SUBSCRIBER_WALL_TIME_LIMIT_SEC * STAGE_FPS * 4))
    for _ in range(num_app_updates):
        await omni.kit.app.get_app().next_update_async()

    print(f"Finished running the application for {num_app_updates} updates.")
    print(f"Wall time: {time.time() - wall_start_time:.4f} seconds")
    print(f"Number of timeline events: {len(timeline_events)}")
    print(f"Number of physics events: {len(physx_events)}")
    print(f"Number of stage render events: {len(stage_render_events)}")
    print(f"Number of app update events: {len(app_update_events)}")

    # Cleanup
    timeline.stop()
    if app_sub is not None:
        app_sub.reset()
        app_sub = None
    if stage_render_sub is not None:
        stage_render_sub.reset()
        stage_render_sub = None
    if physx_sub is not None:
        physx_sub.unsubscribe()
        physx_sub = None
    if timeline_sub is not None:
        timeline_sub.unsubscribe()
        timeline_sub = None


asyncio.ensure_future(example_async())

Accessing Writer and Annotator Data at Custom FPS

Example of how to trigger a writer and access annotator data at a custom FPS, with product rendering disabled when the data is not needed. The standalone example can also be run directly (on Windows use python.bat instead of python.sh):

./python.sh standalone_examples/api/isaacsim.replicator.examples/custom_fps_writer_annotator.py

Note

It is currently not possible to change timeline (stage) FPS after the replicator graph creation as it causes a graph reset. This issue is being addressed. As a workaround make sure you are setting the timeline (stage) parameters before creating the replicator graph.

Standalone Application

Accessing Writer and Annotator Data at Custom FPS

from isaacsim import SimulationApp

simulation_app = SimulationApp({"headless": False})

import os

import carb.settings
import omni.kit.app
import omni.replicator.core as rep
import omni.timeline
import omni.usd

# NOTE: To avoid FPS delta misses make sure the sensor framerate is divisible by the timeline framerate
STAGE_FPS = 100.0
SENSOR_FPS = 10.0
SENSOR_DT = 1.0 / SENSOR_FPS


def run_custom_fps_example(duration_seconds):
    # Create a new stage
    omni.usd.get_context().new_stage()

    # Disable capture on play (data will only be accessed at custom times)
    carb.settings.get_settings().set("/omni/replicator/captureOnPlay", False)

    # Make sure fixed time stepping is set (the timeline will be advanced with the same delta time)
    carb.settings.get_settings().set("/app/player/useFixedTimeStepping", True)

    # Set the timeline parameters
    timeline = omni.timeline.get_timeline_interface()
    timeline.set_looping(False)
    timeline.set_current_time(0.0)
    timeline.set_end_time(10)
    timeline.set_time_codes_per_second(STAGE_FPS)
    timeline.play()
    timeline.commit()

    # Create a light and a semantically annoated cube
    rep.create.light()
    rep.create.cube(semantics=[("class", "cube")])

    # Create a render product and disable it (it will re-enabled when data is needed)
    rp = rep.create.render_product("/OmniverseKit_Persp", (512, 512), name="rp")
    rp.hydra_texture.set_updates_enabled(False)

    # Create a writer and an annotator as different ways to access the data
    out_dir_rgb = os.path.join(os.getcwd(), "_out_writer_fps_rgb")
    print(f"Writer data will be written to: {out_dir_rgb}")
    writer_rgb = rep.WriterRegistry.get("BasicWriter")
    writer_rgb.initialize(output_dir=out_dir_rgb, rgb=True)
    writer_rgb.attach(rp)
    annot_depth = rep.AnnotatorRegistry.get_annotator("distance_to_camera")
    annot_depth.attach(rp)

    # Run the simulation for the given number of frames and access the data at the desired framerates
    written_frames = 0
    previous_time = timeline.get_current_time()
    elapsed_time = 0.0
    loop_iteration_count = 0
    while timeline.get_current_time() < duration_seconds:
        current_time = timeline.get_current_time()
        delta_time = current_time - previous_time
        elapsed_time += delta_time
        print(
            f"[{loop_iteration_count}] current_time={current_time:.4f}; delta_time={delta_time:.4f}; elapsed_time={elapsed_time:.4f}/{SENSOR_DT:.4f};"
        )

        # Check if enough time has passed to trigger the sensor
        if elapsed_time >= SENSOR_DT:
            # Reset the elapsed time with the difference to the optimal trigger time (when the timeline fps is not divisible by the sensor framerate)
            elapsed_time = elapsed_time - SENSOR_DT

            # Enable render products for data access
            rp.hydra_texture.set_updates_enabled(True)

            # Step needs to be called after scheduling the write
            rep.orchestrator.step(delta_time=0.0, pause_timeline=False)

            # After step, the annotator data is available and in sync with the stage
            annot_data = annot_depth.get_data()

            # Count the number of frames written
            print(f"\t Writing frame {written_frames}; annotator data shape={annot_data.shape};")
            written_frames += 1

            # Disable render products to avoid unnecessary rendering
            rp.hydra_texture.set_updates_enabled(False)

        previous_time = current_time
        # Advance the app (timeline) by one frame
        simulation_app.update()

    # Make sure the writer finishes writing the data to disk
    rep.orchestrator.wait_until_complete()


# Run the example.
# NOTE: The simulation duration is calculated to ensure 'target_num_writes' are captured.
# It includes the time for all sensor intervals plus a small buffer of a few stage frames.
target_num_writes = 6
duration_for_target_writes = (target_num_writes * SENSOR_DT) + (5.0 / STAGE_FPS)
run_custom_fps_example(duration_seconds=duration_for_target_writes)

# Close the application
simulation_app.close()

Script Editor

Accessing Writer and Annotator Data at Custom FPS

import asyncio
import os

import carb.settings
import omni.kit.app
import omni.replicator.core as rep
import omni.timeline
import omni.usd

# NOTE: To avoid FPS delta misses make sure the sensor framerate is divisible by the timeline framerate
STAGE_FPS = 100.0
SENSOR_FPS = 10.0
SENSOR_DT = 1.0 / SENSOR_FPS

async def run_custom_fps_example_async(duration_seconds):
    # Create a new stage
    await omni.usd.get_context().new_stage_async()

    # Disable capture on play (data will only be accessed at custom times)
    carb.settings.get_settings().set("/omni/replicator/captureOnPlay", False)

    # Make sure fixed time stepping is set (the timeline will be advanced with the same delta time)
    carb.settings.get_settings().set("/app/player/useFixedTimeStepping", True)

    # Set the timeline parameters
    timeline = omni.timeline.get_timeline_interface()
    timeline.set_looping(False)
    timeline.set_current_time(0.0)
    timeline.set_end_time(10)
    timeline.set_time_codes_per_second(STAGE_FPS)
    timeline.play()
    timeline.commit()

    # Create a light and a semantically annotated cube
    rep.create.light()
    rep.create.cube(semantics=[("class", "cube")])

    # Create a render product and disable it (it will re-enabled when data is needed)
    rp = rep.create.render_product("/OmniverseKit_Persp", (512, 512), name="rp")
    rp.hydra_texture.set_updates_enabled(False)

    # Create a writer and an annotator as different ways to access the data
    out_dir_rgb = os.path.join(os.getcwd(), "_out_writer_fps_rgb")
    print(f"Writer data will be written to: {out_dir_rgb}")
    writer_rgb = rep.WriterRegistry.get("BasicWriter")
    writer_rgb.initialize(output_dir=out_dir_rgb, rgb=True)
    writer_rgb.attach(rp)
    annot_depth = rep.AnnotatorRegistry.get_annotator("distance_to_camera")
    annot_depth.attach(rp)

    # Run the simulation for the given number of frames and access the data at the desired framerates
    written_frames = 0
    previous_time = timeline.get_current_time()
    elapsed_time = 0.0
    loop_iteration_count = 0
    while timeline.get_current_time() < duration_seconds:
        current_time = timeline.get_current_time()
        delta_time = current_time - previous_time
        elapsed_time += delta_time
        print(
            f"[{loop_iteration_count}] current_time={current_time:.4f}; delta_time={delta_time:.4f}; elapsed_time={elapsed_time:.4f}/{SENSOR_DT:.4f};"
        )

        # Check if enough time has passed to trigger the sensor
        if elapsed_time >= SENSOR_DT:
            # Reset the elapsed time with the difference to the optimal trigger time (when the timeline fps is not divisible by the sensor framerate)
            elapsed_time = elapsed_time - SENSOR_DT

            # Enable render products for data access
            rp.hydra_texture.set_updates_enabled(True)

            # Step needs to be called after scheduling the write
            await rep.orchestrator.step_async(delta_time=0.0, pause_timeline=False)

            # After step, the annotator data is available and in sync with the stage
            annot_data = annot_depth.get_data()

            # Count the number of frames written
            print(f"\t Writing frame {written_frames}; annotator data shape={annot_data.shape};")
            written_frames += 1

            # Disable render products to avoid unnecessary rendering
            rp.hydra_texture.set_updates_enabled(False)

        previous_time = current_time
        # Advance the app (timeline) by one frame
        await omni.kit.app.get_app().next_update_async()
        loop_iteration_count += 1

    # Make sure the writer finishes writing the data to disk
    await rep.orchestrator.wait_until_complete_async()

# Run the example.
# NOTE: The simulation duration is calculated to ensure 'target_num_writes' are captured.
# It includes the time for all sensor intervals plus a small buffer of a few stage frames.
target_num_writes = 6
duration_for_target_writes = (target_num_writes * SENSOR_DT) + (5.0 / STAGE_FPS)
await run_custom_fps_example_async(duration_seconds=duration_for_target_writes)

Cosmos Writer Example

This example demonstrates how to use the CosmosWriter for synthetic data generation. The CosmosWriter generates specialized data for NVIDIA’s Cosmos platform, producing key modalities including RGB, shaded instance segmentation, instance segmentation, depth, and edge maps. These visual features help Cosmos models better distinguish object boundaries and understand 3D form, enabling higher-quality synthetic data generation. For more information about using Cosmos for Synthetic Dataset Augmentation, refer to the detailed case study. The standalone example can also be run directly (on Windows use python.bat instead of python.sh):

./python.sh standalone_examples/api/isaacsim.replicator.examples/cosmos_writer_warehouse.py

Standalone Application

Cosmos Writer Example

from isaacsim import SimulationApp

simulation_app = SimulationApp(launch_config={"headless": False})

import os

import carb
import omni.replicator.core as rep
import omni.timeline
import omni.usd
from isaacsim.core.utils.stage import add_reference_to_stage
from isaacsim.storage.native import get_assets_root_path
from pxr import UsdGeom

STAGE_URL = "/Isaac/Samples/Replicator/Stage/full_warehouse_worker_and_anim_cameras.usd"
CARTER_NAV_ASSET_URL = "/Isaac/Samples/Replicator/OmniGraph/nova_carter_nav_only.usd"
CARTER_NAV_PATH = "/NavWorld/CarterNav"
CARTER_CHASSIS_PATH = f"{CARTER_NAV_PATH}/chassis_link"
CARTER_NAV_TARGET_PATH = f"{CARTER_NAV_PATH}/targetXform"
CARTER_CAMERA_PATH = f"{CARTER_NAV_PATH}/chassis_link/sensors/front_hawk/left/camera_left"
CARTER_NAV_POSITION = (-6, 4, 0)
CARTER_NAV_TARGET_POSITION = (3, 3, 0)


def advance_timeline_by_duration(duration: float, max_updates: int = 1000):
    timeline = omni.timeline.get_timeline_interface()
    current_time = timeline.get_current_time()
    target_time = current_time + duration

    if timeline.get_end_time() < target_time:
        timeline.set_end_time(1000000)

    if not timeline.is_playing():
        timeline.play()

    print(f"Advancing timeline from {current_time:.4f}s to {target_time:.4f}s")
    step_count = 0
    while current_time < target_time:
        if step_count >= max_updates:
            print(f"Max updates reached: {step_count}, finishing timeline advance.")
            break

        prev_time = current_time
        simulation_app.update()
        current_time = timeline.get_current_time()
        step_count += 1

        if step_count % 10 == 0:
            print(f"\tStep {step_count}, {current_time:.4f}s/{target_time:.4f}s")

        if current_time <= prev_time:
            print(f"Warning: Timeline did not advance at update {step_count} (time: {current_time:.4f}s).")
    print(f"Finished advancing timeline to {timeline.get_end_time():.4f}s in {step_count} steps")


def run_sdg_pipeline(camera_path, num_frames, capture_interval, use_instance_id=True, segmentation_mapping=None):
    rp = rep.create.render_product(camera_path, (1280, 720))
    cosmos_writer = rep.WriterRegistry.get("CosmosWriter")
    backend = rep.backends.get("DiskBackend")
    out_dir = os.path.join(os.getcwd(), f"_out_cosmos")
    print(f"output_directory: {out_dir}")
    backend.initialize(output_dir=out_dir)
    cosmos_writer.initialize(
        backend=backend, use_instance_id=use_instance_id, segmentation_mapping=segmentation_mapping
    )
    cosmos_writer.attach(rp)

    # Make sure the timeline is playing
    timeline = omni.timeline.get_timeline_interface()
    if not timeline.is_playing():
        timeline.play()

    print(f"Starting SDG pipeline. Capturing {num_frames} frames, every {capture_interval} simulation step(s).")
    frames_captured_count = 0
    simulation_step_index = 0
    while frames_captured_count < num_frames:
        print(f"Simulation step {simulation_step_index}")
        if simulation_step_index % capture_interval == 0:
            print(f"\t Capturing frame {frames_captured_count + 1}/{num_frames}")
            rep.orchestrator.step(pause_timeline=False)
            frames_captured_count += 1
        else:
            simulation_app.update()
        simulation_step_index += 1

    print("Waiting to finish processing and writing the data")
    rep.orchestrator.wait_until_complete()
    print(f"Finished SDG pipeline. Captured {frames_captured_count} frames")
    cosmos_writer.detach()
    rp.destroy()
    timeline.pause()


def run_example(num_frames, capture_interval=1, start_delay=None, use_instance_id=True, segmentation_mapping=None):
    assets_root_path = get_assets_root_path()
    stage_path = assets_root_path + STAGE_URL
    print(f"Opening stage: '{stage_path}'")
    omni.usd.get_context().open_stage(stage_path)
    stage = omni.usd.get_context().get_stage()

    # Enable script nodes
    carb.settings.get_settings().set_bool("/app/omni.graph.scriptnode/opt_in", True)

    # Disable capture on play on the new stage, data is captured manually using the step function
    rep.orchestrator.set_capture_on_play(False)

    # Load carter nova asset with its navigation graph
    carter_url_path = assets_root_path + CARTER_NAV_ASSET_URL
    print(f"Loading carter nova asset: '{carter_url_path}' at prim path: '{CARTER_NAV_PATH}'")
    carter_nav_prim = add_reference_to_stage(usd_path=carter_url_path, prim_path=CARTER_NAV_PATH)
    if not carter_nav_prim.GetAttribute("xformOp:translate"):
        UsdGeom.Xformable(carter_nav_prim).AddTranslateOp()
    carter_nav_prim.GetAttribute("xformOp:translate").Set(CARTER_NAV_POSITION)

    # Set the navigation target position
    carter_navigation_target_prim = stage.GetPrimAtPath(CARTER_NAV_TARGET_PATH)
    if not carter_navigation_target_prim.IsValid():
        print(f"Carter navigation target prim not found at path: {CARTER_NAV_TARGET_PATH}, exiting")
        return
    if not carter_navigation_target_prim.GetAttribute("xformOp:translate"):
        UsdGeom.Xformable(carter_navigation_target_prim).AddTranslateOp()
    carter_navigation_target_prim.GetAttribute("xformOp:translate").Set(CARTER_NAV_TARGET_POSITION)

    # Use the carter nova front hawk camera for capturing data
    camera_prim = stage.GetPrimAtPath(CARTER_CAMERA_PATH)
    if not camera_prim.IsValid():
        print(f"Camera prim not found at path: {CARTER_CAMERA_PATH}, exiting")
        return

    # Advance the timeline with the start delay if provided
    if start_delay is not None and start_delay > 0:
        advance_timeline_by_duration(start_delay)

    # Run the SDG pipeline
    run_sdg_pipeline(camera_prim.GetPath(), num_frames, capture_interval, use_instance_id, segmentation_mapping)


NUM_FRAMES = 120
CAPTURE_INTERVAL = 4
START_DELAY = 1.0
run_example(num_frames=NUM_FRAMES, capture_interval=CAPTURE_INTERVAL, start_delay=START_DELAY, use_instance_id=True)

Script Editor

Cosmos Writer Example

import asyncio
import os

import carb
import omni.replicator.core as rep
import omni.timeline
import omni.usd
from isaacsim.core.utils.stage import add_reference_to_stage
from isaacsim.storage.native import get_assets_root_path_async
from pxr import UsdGeom

STAGE_URL = "/Isaac/Samples/Replicator/Stage/full_warehouse_worker_and_anim_cameras.usd"
CARTER_NAV_ASSET_URL = "/Isaac/Samples/Replicator/OmniGraph/nova_carter_nav_only.usd"
CARTER_NAV_PATH = "/NavWorld/CarterNav"
CARTER_CHASSIS_PATH = f"{CARTER_NAV_PATH}/chassis_link"
CARTER_NAV_TARGET_PATH = f"{CARTER_NAV_PATH}/targetXform"
CARTER_CAMERA_PATH = f"{CARTER_NAV_PATH}/chassis_link/sensors/front_hawk/left/camera_left"
CARTER_NAV_POSITION = (-6, 4, 0)
CARTER_NAV_TARGET_POSITION = (3, 3, 0)


async def advance_timeline_by_duration(duration: float, max_updates: int = 1000):
    timeline = omni.timeline.get_timeline_interface()
    current_time = timeline.get_current_time()
    target_time = current_time + duration

    if timeline.get_end_time() < target_time:
        timeline.set_end_time(1000000)

    if not timeline.is_playing():
        timeline.play()

    print(f"Advancing timeline from {current_time:.4f}s to {target_time:.4f}s")
    step_count = 0
    while current_time < target_time:
        if step_count >= max_updates:
            print(f"Max updates reached: {step_count}, finishing timeline advance.")
            break

        prev_time = current_time
        await omni.kit.app.get_app().next_update_async()
        current_time = timeline.get_current_time()
        step_count += 1

        if step_count % 10 == 0:
            print(f"\tStep {step_count}, {current_time:.4f}s/{target_time:.4f}s")

        if current_time <= prev_time:
            print(f"Warning: Timeline did not advance at update {step_count} (time: {current_time:.4f}s).")
    print(f"Finished advancing timeline to {timeline.get_end_time():.4f}s in {step_count} steps")


async def run_sdg_pipeline(camera_path, num_frames, capture_interval, use_instance_id=True, segmentation_mapping=None):
    rp = rep.create.render_product(camera_path, (1280, 720))
    cosmos_writer = rep.WriterRegistry.get("CosmosWriter")
    backend = rep.backends.get("DiskBackend")
    out_dir = os.path.join(os.getcwd(), f"_out_cosmos")
    print(f"output_directory: {out_dir}")
    backend.initialize(output_dir=out_dir)
    cosmos_writer.initialize(
        backend=backend, use_instance_id=use_instance_id, segmentation_mapping=segmentation_mapping
    )
    cosmos_writer.attach(rp)

    # Make sure the timeline is playing
    timeline = omni.timeline.get_timeline_interface()
    if not timeline.is_playing():
        timeline.play()

    print(f"Starting SDG pipeline. Capturing {num_frames} frames, every {capture_interval} simulation step(s).")
    frames_captured_count = 0
    simulation_step_index = 0
    while frames_captured_count < num_frames:
        print(f"Simulation step {simulation_step_index}")
        if simulation_step_index % capture_interval == 0:
            print(f"\t Capturing frame {frames_captured_count + 1}/{num_frames}")
            await rep.orchestrator.step_async(pause_timeline=False)
            frames_captured_count += 1
        else:
            await omni.kit.app.get_app().next_update_async()
        simulation_step_index += 1

    print("Waiting to finish processing and writing the data")
    await rep.orchestrator.wait_until_complete_async()
    print(f"Finished SDG pipeline. Captured {frames_captured_count} frames")
    cosmos_writer.detach()
    rp.destroy()
    timeline.pause()


async def run_example_async(
    num_frames, capture_interval=1, start_delay=None, use_instance_id=True, segmentation_mapping=None
):
    assets_root_path = await get_assets_root_path_async()
    stage_path = assets_root_path + STAGE_URL
    print(f"Opening stage: '{stage_path}'")
    await omni.usd.get_context().open_stage_async(stage_path)
    stage = omni.usd.get_context().get_stage()

    # Enable script nodes
    carb.settings.get_settings().set_bool("/app/omni.graph.scriptnode/opt_in", True)

    # Disable capture on play on the new stage, data is captured manually using the step function
    rep.orchestrator.set_capture_on_play(False)

    # Load carter nova asset with its navigation graph
    carter_url_path = assets_root_path + CARTER_NAV_ASSET_URL
    print(f"Loading carter nova asset: '{carter_url_path}' at prim path: '{CARTER_NAV_PATH}'")
    carter_nav_prim = add_reference_to_stage(usd_path=carter_url_path, prim_path=CARTER_NAV_PATH)
    if not carter_nav_prim.GetAttribute("xformOp:translate"):
        UsdGeom.Xformable(carter_nav_prim).AddTranslateOp()
    carter_nav_prim.GetAttribute("xformOp:translate").Set(CARTER_NAV_POSITION)

    # Set the navigation target position
    carter_navigation_target_prim = stage.GetPrimAtPath(CARTER_NAV_TARGET_PATH)
    if not carter_navigation_target_prim.IsValid():
        print(f"Carter navigation target prim not found at path: {CARTER_NAV_TARGET_PATH}, exiting")
        return
    if not carter_navigation_target_prim.GetAttribute("xformOp:translate"):
        UsdGeom.Xformable(carter_navigation_target_prim).AddTranslateOp()
    carter_navigation_target_prim.GetAttribute("xformOp:translate").Set(CARTER_NAV_TARGET_POSITION)

    # Use the carter nova front hawk camera for capturing data
    camera_prim = stage.GetPrimAtPath(CARTER_CAMERA_PATH)
    if not camera_prim.IsValid():
        print(f"Camera prim not found at path: {CARTER_CAMERA_PATH}, exiting")
        return

    # Advance the timeline with the start delay if provided
    if start_delay is not None and start_delay > 0:
        await advance_timeline_by_duration(start_delay)

    # Run the SDG pipeline
    await run_sdg_pipeline(camera_prim.GetPath(), num_frames, capture_interval, use_instance_id, segmentation_mapping)


NUM_FRAMES = 120
CAPTURE_INTERVAL = 4
START_DELAY = 1.0
asyncio.ensure_future(
    run_example_async(num_frames=NUM_FRAMES, capture_interval=CAPTURE_INTERVAL, start_delay=START_DELAY, use_instance_id=True)
)