Randomization Snippets#
These examples demonstrate how to randomize Isaac Sim scenes for synthetic data generation (SDG) when default replicator randomizers are not sufficient or applicable. They cover light, material, object placement, camera, physics-based stacking, SimReady asset randomization, and object-reconstruction asset randomization, with both Script Editor snippets and standalone applications.
The randomizations use Replicator functional APIs together with Isaac Sim Core APIs for scene setup, transforms, bounds, materials, prim utilities, and physics helpers. The snippets show the exact function calls. USD and PhysX APIs are used only where direct scene or physics authoring is required.
Each example has two forms:
Script Editor snippets run inside an active Isaac Sim session with
asyncio.ensure_future(...).Standalone Application scripts run with
python.sh(orpython.baton Windows), create their ownSimulationApp, and can be used in automation and tests.
Standalone applications define the same configuration constants as the Script Editor snippets, such as NUM_FRAMES and WRITE_DATA. Command-line flags such as --num-frames and --no-write-data override those defaults when the script is run from a terminal. Use --help on any standalone example for the full list of options.
Prerequisites:
Familiarity with Isaac Sim Replicator and subframes.
Ability to execute code from the Script Editor or run standalone examples with
python.sh.Basic understanding of USD concepts for direct scene authoring operations.
Randomizing Light Sources#
This snippet creates a simple scene with a cube, sphere, ground plane, and multiple light sources. Each frame, it randomizes light poses and visual properties, and can optionally write RGB output with BasicWriter.
The standalone application can be run directly (on Windows use python.bat instead of python.sh):
./python.sh standalone_examples/api/isaacsim.replicator.examples/randomizing_light_sources.py
Optional flags include --num-frames, --num-lights, --delay, and --no-write-data. The Script Editor snippet exposes the same values through NUM_FRAMES, NUM_LIGHTS, WRITE_DATA, and DELAY constants.
Randomizing Light Sources
import asyncio
import os
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import omni.replicator.core as rep
NUM_FRAMES = 10
NUM_LIGHTS = 10
WRITE_DATA = True
DELAY = 0.2
SEED = 42
async def run_randomizations_async(num_frames, lights, write_data=True, delay=None, rng=None):
if rng is None:
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_rand_lights")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
cam = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), name="Camera")
rp = rep.create.render_product(cam, resolution=(512, 512))
writer.attach(rp)
for _ in range(num_frames):
for light in lights:
rep.functional.modify.pose(
light,
position_value=(
gen.uniform(-5, 5),
gen.uniform(-5, 5),
gen.uniform(4, 6),
),
scale_value=gen.uniform(0.5, 1.5),
)
rep.functional.modify.attribute(light, "inputs:colorTemperature", gen.normal(4500, 1500))
rep.functional.modify.attribute(light, "inputs:intensity", gen.normal(25000, 5000))
rep.functional.modify.attribute(
light,
"inputs:color",
(
gen.uniform(0.1, 0.9),
gen.uniform(0.1, 0.9),
gen.uniform(0.1, 0.9),
),
)
if write_data:
await rep.orchestrator.step_async(rt_subframes=16)
else:
await app_utils.update_app_async()
# Optional delay between frames to better visualize the randomization in the viewport
if delay is not None and delay > 0:
await asyncio.sleep(delay)
# Wait for the data to be written to disk and cleanup writer and render product
if write_data:
await rep.orchestrator.wait_until_complete_async()
writer.detach()
rp.destroy()
async def run_example_async(num_frames, num_lights, write_data, delay=None, rng=None):
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
await stage_utils.create_new_stage_async()
rep.functional.create.xform(name="World")
rep.functional.create.sphere(
parent="/World", name="Sphere", position=(0.0, 1.0, 1.0), semantics={"class": "sphere"}
)
rep.functional.create.cube(parent="/World", name="Cube", position=(0.0, -2.0, 2.0), semantics={"class": "cube"})
rep.functional.create.plane(parent="/World", name="Plane", scale=(10, 10, 1))
rep.functional.create.scope(name="Lights", parent="/World")
lights = rep.functional.create_batch.sphere_light(
count=num_lights,
parent="/World/Lights",
enable_color_temperature=True,
radius=0.5,
)
await run_randomizations_async(num_frames=num_frames, lights=lights, write_data=write_data, delay=delay, rng=rng)
asyncio.ensure_future(run_example_async(NUM_FRAMES, NUM_LIGHTS, WRITE_DATA, delay=DELAY))
Randomizing Light Sources
"""Randomize light properties and capture synthetic images."""
from isaacsim import SimulationApp
simulation_app = SimulationApp(launch_config={"headless": False})
import argparse
import os
import time
from typing import Any
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import omni.replicator.core as rep
NUM_FRAMES = 10
NUM_LIGHTS = 10
WRITE_DATA = True
DELAY = 0.2
SEED = 42
parser = argparse.ArgumentParser()
parser.add_argument("--num-frames", type=int, default=NUM_FRAMES, help="Number of randomization frames to run.")
parser.add_argument("--num-lights", type=int, default=NUM_LIGHTS, help="Number of sphere lights to create.")
parser.add_argument(
"--write-data",
action=argparse.BooleanOptionalAction,
default=WRITE_DATA,
help="Write captured output to disk.",
)
parser.add_argument("--delay", type=float, default=DELAY, help="Delay in seconds between frames (0 to disable).")
args, _ = parser.parse_known_args()
def run_randomizations(
num_frames: int,
lights: list[Any],
write_data: bool = True,
delay: float | None = None,
rng: Any = None,
) -> None:
"""Randomize light poses and attributes over multiple frames.
Args:
num_frames: Number of randomization frames to run.
lights: Sphere-light prims to randomize.
write_data: Whether to capture and write RGB images.
delay: Delay between frames, or None to run without a delay.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
if rng is None:
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_rand_lights")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
cam = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), name="Camera")
rp = rep.create.render_product(cam, resolution=(512, 512))
writer.attach(rp)
for _ in range(num_frames):
for light in lights:
rep.functional.modify.pose(
light,
position_value=(
gen.uniform(-5, 5),
gen.uniform(-5, 5),
gen.uniform(4, 6),
),
scale_value=gen.uniform(0.5, 1.5),
)
rep.functional.modify.attribute(light, "inputs:colorTemperature", gen.normal(4500, 1500))
rep.functional.modify.attribute(light, "inputs:intensity", gen.normal(25000, 5000))
rep.functional.modify.attribute(
light,
"inputs:color",
(
gen.uniform(0.1, 0.9),
gen.uniform(0.1, 0.9),
gen.uniform(0.1, 0.9),
),
)
if write_data:
rep.orchestrator.step(rt_subframes=16)
else:
app_utils.update_app()
if delay is not None and delay > 0:
time.sleep(delay)
if write_data:
rep.orchestrator.wait_until_complete()
writer.detach()
rp.destroy()
def run_example(
num_frames: int,
num_lights: int,
write_data: bool,
delay: float | None = None,
rng: Any = None,
) -> None:
"""Build the light-randomization scene and run the capture loop.
Args:
num_frames: Number of randomization frames to run.
num_lights: Number of sphere lights to create.
write_data: Whether to capture and write RGB images.
delay: Delay between frames, or None to run without a delay.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
stage_utils.create_new_stage()
rep.functional.create.xform(name="World")
rep.functional.create.sphere(
parent="/World", name="Sphere", position=(0.0, 1.0, 1.0), semantics={"class": "sphere"}
)
rep.functional.create.cube(parent="/World", name="Cube", position=(0.0, -2.0, 2.0), semantics={"class": "cube"})
rep.functional.create.plane(parent="/World", name="Plane", scale=(10, 10, 1))
rep.functional.create.scope(name="Lights", parent="/World")
lights = rep.functional.create_batch.sphere_light(
count=num_lights,
parent="/World/Lights",
enable_color_temperature=True,
radius=0.5,
)
run_randomizations(num_frames=num_frames, lights=lights, write_data=write_data, delay=delay, rng=rng)
run_example(
num_frames=args.num_frames,
num_lights=args.num_lights,
write_data=args.write_data,
delay=args.delay,
)
Randomizing Textures#
This snippet creates a small scene with multiple objects and materials, then randomizes texture inputs over a sequence of frames. After the randomizations, the original materials are restored.
The standalone application can be run directly (on Windows use python.bat instead of python.sh):
./python.sh standalone_examples/api/isaacsim.replicator.examples/randomizing_textures.py
Optional flags include --num-frames, --num-cubes, --delay, and --no-write-data. The Script Editor snippet exposes the same values through NUM_FRAMES, NUM_CUBES, WRITE_DATA, and DELAY constants.
Randomizing Textures
import asyncio
import os
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import omni.replicator.core as rep
from isaacsim.storage.native import get_assets_root_path_async
from pxr import UsdShade
NUM_FRAMES = 10
NUM_CUBES = 10
WRITE_DATA = True
DELAY = 0.2
SEED = 42
async def run_randomizations_async(num_frames, shapes, materials, write_data=True, delay=None, rng=None):
if rng is None:
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
assets_root_path = await get_assets_root_path_async()
textures = [
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/aggregate_exposed_diff.jpg",
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/gravel_track_ballast_diff.jpg",
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/gravel_track_ballast_multi_R_rough_G_ao.jpg",
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/rough_gravel_rough.jpg",
]
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_rand_textures")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
cam = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), name="Camera")
rp = rep.create.render_product(cam, resolution=(512, 512))
writer.attach(rp)
# Apply the new materials and store the initial ones to reassign later
initial_materials = {}
for i, shape in enumerate(shapes):
cur_mat, _ = UsdShade.MaterialBindingAPI(shape).ComputeBoundMaterial()
initial_materials[shape] = cur_mat
rep.functional.modify.material(shape, materials[i])
for _ in range(num_frames):
for mat in materials:
rep.functional.modify.attribute(mat, "inputs:diffuse_texture", gen.choice(textures))
project_uvw = gen.choice([True, False], p=[0.9, 0.1])
rep.functional.modify.attribute(mat, "inputs:project_uvw", bool(project_uvw))
texture_scale = gen.uniform(0.1, 1)
rep.functional.modify.attribute(mat, "inputs:texture_scale", (texture_scale, texture_scale))
rep.functional.modify.attribute(mat, "inputs:texture_rotate", gen.uniform(0, 45))
if write_data:
await rep.orchestrator.step_async(rt_subframes=16)
else:
await app_utils.update_app_async()
# Optional delay between frames to better visualize the randomization in the viewport
if delay is not None and delay > 0:
await asyncio.sleep(delay)
# Wait for the data to be written to disk and cleanup writer and render product
if write_data:
await rep.orchestrator.wait_until_complete_async()
writer.detach()
rp.destroy()
# Reassign the initial materials
for shape, mat in initial_materials.items():
if mat:
rep.functional.modify.material(shape, mat.GetPrim())
else:
UsdShade.MaterialBindingAPI(shape).UnbindAllBindings()
async def run_example_async(num_frames, num_cubes, write_data, delay=None, rng=None):
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
await stage_utils.create_new_stage_async()
rep.functional.create.xform(name="World")
rep.functional.create.scope(name="Looks", parent="/World")
rep.functional.create.dome_light(intensity=1000, parent="/World")
sphere = rep.functional.create.sphere(
parent="/World", name="Sphere", position=(0.0, 0.0, 1.0), semantics={"class": "sphere"}
)
cubes = rep.functional.create_batch.cube(count=num_cubes, parent="/World", semantics={"class": "cube"})
for cube in cubes:
scale_rand = gen.uniform(0.25, 0.5)
rep.functional.modify.pose(
cube,
position_value=(gen.uniform(-3.5, 3.5), gen.uniform(-3.5, 3.5), 1),
scale_value=scale_rand,
)
rep.functional.create.plane(parent="/World", name="Plane", scale=(10, 10, 1))
shapes = [sphere] + list(cubes)
materials = rep.functional.create_batch.material(mdl="OmniPBR.mdl", count=len(shapes), parent="/World/Looks")
await run_randomizations_async(num_frames, shapes, materials, write_data, delay, rng=rng)
asyncio.ensure_future(run_example_async(NUM_FRAMES, NUM_CUBES, WRITE_DATA, delay=DELAY))
Randomizing Textures
"""Randomize material textures and capture synthetic images."""
from isaacsim import SimulationApp
simulation_app = SimulationApp(launch_config={"headless": False})
import argparse
import os
import time
from typing import Any
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import omni.replicator.core as rep
from isaacsim.storage.native import get_assets_root_path
from pxr import UsdShade
NUM_FRAMES = 10
NUM_CUBES = 10
WRITE_DATA = True
DELAY = 0.2
SEED = 42
parser = argparse.ArgumentParser()
parser.add_argument("--num-frames", type=int, default=NUM_FRAMES, help="Number of randomization frames to run.")
parser.add_argument("--num-cubes", type=int, default=NUM_CUBES, help="Number of cubes to create.")
parser.add_argument(
"--write-data",
action=argparse.BooleanOptionalAction,
default=WRITE_DATA,
help="Write captured output to disk.",
)
parser.add_argument("--delay", type=float, default=DELAY, help="Delay in seconds between frames (0 to disable).")
args, _ = parser.parse_known_args()
def run_randomizations(
num_frames: int,
shapes: list[Any],
materials: list[Any],
write_data: bool = True,
delay: float | None = None,
rng: Any = None,
) -> None:
"""Randomize bound material textures over multiple frames.
Args:
num_frames: Number of randomization frames to run.
shapes: Shape prims whose materials are randomized.
materials: Materials to bind and randomize.
write_data: Whether to capture and write RGB images.
delay: Delay between frames, or None to run without a delay.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
if rng is None:
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
assets_root_path = get_assets_root_path()
textures = [
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/aggregate_exposed_diff.jpg",
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/gravel_track_ballast_diff.jpg",
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/gravel_track_ballast_multi_R_rough_G_ao.jpg",
assets_root_path + "/NVIDIA/Materials/vMaterials_2/Ground/textures/rough_gravel_rough.jpg",
]
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_rand_textures")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
cam = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), name="Camera")
rp = rep.create.render_product(cam, resolution=(512, 512))
writer.attach(rp)
initial_materials = {}
for i, shape in enumerate(shapes):
cur_mat, _ = UsdShade.MaterialBindingAPI(shape).ComputeBoundMaterial()
initial_materials[shape] = cur_mat
rep.functional.modify.material(shape, materials[i])
for _ in range(num_frames):
for mat in materials:
rep.functional.modify.attribute(mat, "inputs:diffuse_texture", gen.choice(textures))
project_uvw = gen.choice([True, False], p=[0.9, 0.1])
rep.functional.modify.attribute(mat, "inputs:project_uvw", bool(project_uvw))
texture_scale = gen.uniform(0.1, 1)
rep.functional.modify.attribute(mat, "inputs:texture_scale", (texture_scale, texture_scale))
rep.functional.modify.attribute(mat, "inputs:texture_rotate", gen.uniform(0, 45))
if write_data:
rep.orchestrator.step(rt_subframes=16)
else:
app_utils.update_app()
if delay is not None and delay > 0:
time.sleep(delay)
if write_data:
rep.orchestrator.wait_until_complete()
writer.detach()
rp.destroy()
for shape, mat in initial_materials.items():
if mat:
rep.functional.modify.material(shape, mat.GetPrim())
else:
UsdShade.MaterialBindingAPI(shape).UnbindAllBindings()
def run_example(
num_frames: int,
num_cubes: int,
write_data: bool,
delay: float | None = None,
rng: Any = None,
) -> None:
"""Build the texture-randomization scene and run the capture loop.
Args:
num_frames: Number of randomization frames to run.
num_cubes: Number of cubes to create.
write_data: Whether to capture and write RGB images.
delay: Delay between frames, or None to run without a delay.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
stage_utils.create_new_stage()
rep.functional.create.xform(name="World")
rep.functional.create.scope(name="Looks", parent="/World")
rep.functional.create.dome_light(intensity=1000, parent="/World")
sphere = rep.functional.create.sphere(
parent="/World", name="Sphere", position=(0.0, 0.0, 1.0), semantics={"class": "sphere"}
)
cubes = rep.functional.create_batch.cube(count=num_cubes, parent="/World", semantics={"class": "cube"})
for cube in cubes:
scale_rand = gen.uniform(0.25, 0.5)
rep.functional.modify.pose(
cube,
position_value=(gen.uniform(-3.5, 3.5), gen.uniform(-3.5, 3.5), 1),
scale_value=scale_rand,
)
rep.functional.create.plane(parent="/World", name="Plane", scale=(10, 10, 1))
shapes = [sphere] + list(cubes)
materials = rep.functional.create_batch.material(mdl="OmniPBR.mdl", count=len(shapes), parent="/World/Looks")
run_randomizations(num_frames, shapes, materials, write_data, delay, rng=rng)
run_example(
num_frames=args.num_frames,
num_cubes=args.num_cubes,
write_data=args.write_data,
delay=args.delay,
)
Sequential Randomizations#
This snippet shows a chained randomization workflow where one randomized result determines the next. For every frame, it cycles dome light textures, moves a pallet, places a bin fully on top of the pallet, and moves a camera along points sampled on a sphere so it faces the bin.
The standalone application can be run directly (on Windows use python.bat instead of python.sh):
./python.sh standalone_examples/api/isaacsim.replicator.examples/sequential_randomizations.py
Optional flags include --num-frames, --delay, and --no-write-data. The Script Editor snippet exposes the same values through NUM_FRAMES, WRITE_DATA, and DELAY constants.
Sequential Randomizations
import asyncio
import itertools
import os
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.bounds as bounds_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import isaacsim.core.experimental.utils.transform as transform_utils
import isaacsim.core.experimental.utils.xform as xform_utils
import numpy as np
import omni.replicator.core as rep
from isaacsim.storage.native import get_assets_root_path_async
NUM_FRAMES = 90
WRITE_DATA = True
DELAY = 0.2
SEED = 42
FORKLIFT_PATH = "/Isaac/Props/Forklift/forklift.usd"
PALLET_PATH = "/Isaac/Props/Pallet/pallet.usd"
BIN_PATH = "/Isaac/Props/KLT_Bin/small_KLT_visual.usd"
DOME_TEXTURES = [
"/NVIDIA/Assets/Skies/Cloudy/champagne_castle_1_4k.hdr",
"/NVIDIA/Assets/Skies/Clear/evening_road_01_4k.hdr",
"/NVIDIA/Assets/Skies/Clear/mealie_road_4k.hdr",
"/NVIDIA/Assets/Skies/Clear/qwantani_4k.hdr",
]
# Fibonacci sphere algorithm: https://arxiv.org/pdf/0912.4540
def next_point_on_sphere(idx, num_points, radius=1, origin=(0, 0, 0)):
offset = 2.0 / num_points
inc = np.pi * (3.0 - np.sqrt(5.0))
z = ((idx * offset) - 1) + (offset / 2)
phi = ((idx + 1) % num_points) * inc
r = np.sqrt(1 - pow(z, 2))
y = np.cos(phi) * r
x = np.sin(phi) * r
return [(x * radius) + origin[0], (y * radius) + origin[1], (z * radius) + origin[2]]
async def run_randomizations_async(
num_frames, forklift_path, pallet_path, bin_path, dome_textures, write_data=True, delay=None, rng=None
):
if rng is None:
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
assets_root_path = await get_assets_root_path_async()
await stage_utils.create_new_stage_async()
rep.functional.create.xform(name="World")
rep.functional.create.scope(name="Lights", parent="/World")
dome_light = rep.functional.create.dome_light(intensity=1000, parent="/World/Lights")
forklift = rep.functional.create.reference(
usd_path=assets_root_path + forklift_path,
parent="/World",
name="Forklift",
position=(-4.5, -4.5, 0),
)
pallet = rep.functional.create.reference(
usd_path=assets_root_path + pallet_path,
parent="/World",
name="Pallet",
)
bin_prim = rep.functional.create.reference(
usd_path=assets_root_path + bin_path,
parent="/World",
name="Bin",
)
view_cam = rep.functional.create.camera(parent="/World", name="Camera")
dome_textures_full = [assets_root_path + tex for tex in dome_textures]
textures_cycle = itertools.cycle(dome_textures_full)
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_rand_sphere_scan")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
persp_cam = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), name="PerspCamera")
rp_persp = rep.create.render_product(persp_cam, (512, 512), name="PerspView")
rp_view = rep.create.render_product(view_cam, (512, 512), name="SphereView")
writer.attach([rp_view, rp_persp])
bb_cache = bounds_utils.create_bbox_cache()
pallet_aabb = bounds_utils.compute_aabb(pallet, bbox_cache=bb_cache)
bin_aabb = bounds_utils.compute_aabb(bin_prim, bbox_cache=bb_cache)
pallet_size = pallet_aabb[3:6] - pallet_aabb[0:3]
bin_size = bin_aabb[3:6] - bin_aabb[0:3]
pallet_length = float(np.linalg.norm(pallet_size))
for i in range(num_frames):
# Set next background texture every nth frame and run an app update
if i % 5 == 0:
rep.functional.modify.attribute(dome_light, "inputs:texture:file", next(textures_cycle))
await app_utils.update_app_async()
# Randomize pallet pose
rand_z_rot = gen.uniform(-90, 90)
rep.functional.modify.pose(
pallet,
position_value=(gen.uniform(-1.5, 1.5), gen.uniform(-1.5, 1.5), 0),
rotation_value=(0, 0, rand_z_rot),
)
pallet_pos, pallet_quat = xform_utils.get_world_pose(pallet)
pallet_pos = tuple(map(float, pallet_pos.numpy()))
pallet_euler = transform_utils.quaternion_to_euler_angles(pallet_quat.numpy(), degrees=True).numpy().flatten()
# Randomize bin position on top of the rotated pallet area making sure the bin is fully on the pallet
rand_transl_x = gen.uniform(-pallet_size[0] / 2 + bin_size[0] / 2, pallet_size[0] / 2 - bin_size[0] / 2)
rand_transl_y = gen.uniform(-pallet_size[1] / 2 + bin_size[1] / 2, pallet_size[1] / 2 - bin_size[1] / 2)
# Adjust bin position to account for the random rotation of the pallet
rand_z_rot_rad = np.deg2rad(rand_z_rot)
rot_adjusted_transl_x = rand_transl_x * np.cos(rand_z_rot_rad) - rand_transl_y * np.sin(rand_z_rot_rad)
rot_adjusted_transl_y = rand_transl_x * np.sin(rand_z_rot_rad) + rand_transl_y * np.cos(rand_z_rot_rad)
rep.functional.modify.pose(
bin_prim,
position_value=(
float(pallet_pos[0] + rot_adjusted_transl_x),
float(pallet_pos[1] + rot_adjusted_transl_y),
float(pallet_pos[2] + pallet_size[2] + bin_size[2] / 2),
),
rotation_value=tuple(map(float, pallet_euler)),
)
# Get next camera position on a sphere looking at the bin with a randomized distance
rand_radius = gen.normal(3, 0.5) * pallet_length
bin_pos, _ = xform_utils.get_world_pose(bin_prim)
bin_pos = tuple(map(float, bin_pos.numpy()))
cam_pos = tuple(map(float, next_point_on_sphere(i, num_points=num_frames, radius=rand_radius, origin=bin_pos)))
rep.functional.modify.pose(view_cam, position_value=cam_pos, look_at_value=bin_pos, look_at_up_axis=(0, 0, 1))
if write_data:
await rep.orchestrator.step_async(rt_subframes=8, delta_time=0.0)
else:
await app_utils.update_app_async()
# Optional delay between frames to better visualize the randomization in the viewport
if delay is not None and delay > 0:
await asyncio.sleep(delay)
# Wait for the data to be written to disk and cleanup writer and render products
if write_data:
await rep.orchestrator.wait_until_complete_async()
writer.detach()
rp_persp.destroy()
rp_view.destroy()
async def run_example_async(num_frames, write_data, delay=None, rng=None):
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
await run_randomizations_async(
num_frames, FORKLIFT_PATH, PALLET_PATH, BIN_PATH, DOME_TEXTURES, write_data, delay, rng=rng
)
asyncio.ensure_future(run_example_async(NUM_FRAMES, WRITE_DATA, delay=DELAY))
Sequential Randomizations
"""Run sequential scene and camera randomizations for synthetic data generation."""
from isaacsim import SimulationApp
simulation_app = SimulationApp(launch_config={"headless": False})
import argparse
import itertools
import os
import time
from typing import Any
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.bounds as bounds_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import isaacsim.core.experimental.utils.transform as transform_utils
import isaacsim.core.experimental.utils.xform as xform_utils
import numpy as np
import omni.replicator.core as rep
from isaacsim.storage.native import get_assets_root_path
NUM_FRAMES = 90
WRITE_DATA = True
DELAY = 0.2
SEED = 42
parser = argparse.ArgumentParser()
parser.add_argument("--num-frames", type=int, default=NUM_FRAMES, help="Number of randomization frames to run.")
parser.add_argument(
"--write-data",
action=argparse.BooleanOptionalAction,
default=WRITE_DATA,
help="Write captured output to disk.",
)
parser.add_argument("--delay", type=float, default=DELAY, help="Delay in seconds between frames (0 to disable).")
args, _ = parser.parse_known_args()
# Fibonacci sphere algorithm: https://arxiv.org/pdf/0912.4540
def next_point_on_sphere(
idx: int,
num_points: int,
radius: float = 1,
origin: tuple[float, float, float] = (0, 0, 0),
) -> list[float]:
"""Compute the next evenly distributed point on a sphere.
Args:
idx: Index of the point to compute.
num_points: Total number of points distributed over the sphere.
radius: Sphere radius.
origin: Sphere center.
Returns:
The computed point in Cartesian coordinates.
"""
offset = 2.0 / num_points
inc = np.pi * (3.0 - np.sqrt(5.0))
z = ((idx * offset) - 1) + (offset / 2)
phi = ((idx + 1) % num_points) * inc
r = np.sqrt(1 - pow(z, 2))
y = np.cos(phi) * r
x = np.sin(phi) * r
return [(x * radius) + origin[0], (y * radius) + origin[1], (z * radius) + origin[2]]
FORKLIFT_PATH = "/Isaac/Props/Forklift/forklift.usd"
PALLET_PATH = "/Isaac/Props/Pallet/pallet.usd"
BIN_PATH = "/Isaac/Props/KLT_Bin/small_KLT_visual.usd"
DOME_TEXTURES = [
"/NVIDIA/Assets/Skies/Cloudy/champagne_castle_1_4k.hdr",
"/NVIDIA/Assets/Skies/Clear/evening_road_01_4k.hdr",
"/NVIDIA/Assets/Skies/Clear/mealie_road_4k.hdr",
"/NVIDIA/Assets/Skies/Clear/qwantani_4k.hdr",
]
def run_randomizations(
num_frames: int,
forklift_path: str,
pallet_path: str,
bin_path: str,
dome_textures: list[str],
write_data: bool = True,
delay: float | None = None,
rng: Any = None,
) -> None:
"""Run sequential asset, lighting, and camera randomizations.
Args:
num_frames: Number of randomization frames to run.
forklift_path: Forklift asset path relative to the assets root.
pallet_path: Pallet asset path relative to the assets root.
bin_path: Bin asset path relative to the assets root.
dome_textures: Dome texture paths relative to the assets root.
write_data: Whether to capture and write RGB images.
delay: Delay between frames, or None to run without a delay.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
if rng is None:
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
assets_root_path = get_assets_root_path()
stage_utils.create_new_stage()
rep.functional.create.xform(name="World")
rep.functional.create.scope(name="Lights", parent="/World")
dome_light = rep.functional.create.dome_light(intensity=1000, parent="/World/Lights")
rep.functional.create.reference(
usd_path=assets_root_path + forklift_path,
parent="/World",
name="Forklift",
position=(-4.5, -4.5, 0),
)
pallet = rep.functional.create.reference(
usd_path=assets_root_path + pallet_path,
parent="/World",
name="Pallet",
)
bin_prim = rep.functional.create.reference(
usd_path=assets_root_path + bin_path,
parent="/World",
name="Bin",
)
view_cam = rep.functional.create.camera(parent="/World", name="Camera")
dome_textures_full = [assets_root_path + tex for tex in dome_textures]
textures_cycle = itertools.cycle(dome_textures_full)
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_rand_sphere_scan")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
persp_cam = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), name="PerspCamera")
rp_persp = rep.create.render_product(persp_cam, (512, 512), name="PerspView")
rp_view = rep.create.render_product(view_cam, (512, 512), name="SphereView")
writer.attach([rp_view, rp_persp])
bb_cache = bounds_utils.create_bbox_cache()
pallet_aabb = bounds_utils.compute_aabb(pallet, bbox_cache=bb_cache)
bin_aabb = bounds_utils.compute_aabb(bin_prim, bbox_cache=bb_cache)
pallet_size = pallet_aabb[3:6] - pallet_aabb[0:3]
bin_size = bin_aabb[3:6] - bin_aabb[0:3]
pallet_length = float(np.linalg.norm(pallet_size))
for i in range(num_frames):
if i % 5 == 0:
rep.functional.modify.attribute(dome_light, "inputs:texture:file", next(textures_cycle))
app_utils.update_app()
rand_z_rot = gen.uniform(-90, 90)
rep.functional.modify.pose(
pallet,
position_value=(gen.uniform(-1.5, 1.5), gen.uniform(-1.5, 1.5), 0),
rotation_value=(0, 0, rand_z_rot),
)
pallet_pos, pallet_quat = xform_utils.get_world_pose(pallet)
pallet_pos = tuple(map(float, pallet_pos.numpy()))
pallet_euler = transform_utils.quaternion_to_euler_angles(pallet_quat.numpy(), degrees=True).numpy().flatten()
rand_transl_x = gen.uniform(-pallet_size[0] / 2 + bin_size[0] / 2, pallet_size[0] / 2 - bin_size[0] / 2)
rand_transl_y = gen.uniform(-pallet_size[1] / 2 + bin_size[1] / 2, pallet_size[1] / 2 - bin_size[1] / 2)
rand_z_rot_rad = np.deg2rad(rand_z_rot)
rot_adjusted_transl_x = rand_transl_x * np.cos(rand_z_rot_rad) - rand_transl_y * np.sin(rand_z_rot_rad)
rot_adjusted_transl_y = rand_transl_x * np.sin(rand_z_rot_rad) + rand_transl_y * np.cos(rand_z_rot_rad)
rep.functional.modify.pose(
bin_prim,
position_value=(
float(pallet_pos[0] + rot_adjusted_transl_x),
float(pallet_pos[1] + rot_adjusted_transl_y),
float(pallet_pos[2] + pallet_size[2] + bin_size[2] / 2),
),
rotation_value=tuple(map(float, pallet_euler)),
)
rand_radius = gen.normal(3, 0.5) * pallet_length
bin_pos, _ = xform_utils.get_world_pose(bin_prim)
bin_pos = tuple(map(float, bin_pos.numpy()))
cam_pos = tuple(map(float, next_point_on_sphere(i, num_points=num_frames, radius=rand_radius, origin=bin_pos)))
rep.functional.modify.pose(view_cam, position_value=cam_pos, look_at_value=bin_pos, look_at_up_axis=(0, 0, 1))
if write_data:
rep.orchestrator.step(rt_subframes=8, delta_time=0.0)
else:
app_utils.update_app()
if delay is not None and delay > 0:
time.sleep(delay)
if write_data:
rep.orchestrator.wait_until_complete()
writer.detach()
rp_persp.destroy()
rp_view.destroy()
def run_example(num_frames: int, write_data: bool, delay: float | None = None, rng: Any = None) -> None:
"""Run the sequential-randomization example.
Args:
num_frames: Number of randomization frames to run.
write_data: Whether to capture and write RGB images.
delay: Delay between frames, or None to run without a delay.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
run_randomizations(num_frames, FORKLIFT_PATH, PALLET_PATH, BIN_PATH, DOME_TEXTURES, write_data, delay, rng=rng)
run_example(args.num_frames, args.write_data, args.delay)
Physics-based Randomized Volume Filling#
This snippet combines Replicator randomization with Isaac Sim Core physics helpers. It spawns pallets and boxes, creates temporary collision walls, configures colliders and physics materials, and uses PhysX APIs for direct rigid body toggles and force application.
The workflow randomly spawns pallets, drops physically simulated boxes on top of them, creates a temporary collision volume to keep the boxes contained, applies forces to settle the stack, and then removes the collision volume. To allow easier sliding into stable positions, friction is temporarily reduced during the simulation and restored afterward.
The standalone application can be run directly (on Windows use python.bat instead of python.sh):
./python.sh standalone_examples/api/isaacsim.replicator.examples/physics_based_randomized_volume_filling.py
Optional flags include --num-pallets, --env-url (use none for an empty stage), and --no-write-data. The Script Editor snippet exposes the same values through NUM_PALLETS, ENV_URL, and WRITE_DATA constants.
Physics-based Randomized Volume Filling
import asyncio
import os
from itertools import chain
import carb
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.bounds as bounds_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import isaacsim.core.experimental.utils.transform as transform_utils
import isaacsim.core.experimental.utils.xform as xform_utils
import numpy as np
import omni.physx
import omni.replicator.core as rep
from isaacsim.core.experimental.materials import RigidBodyMaterial
from isaacsim.core.experimental.objects import Cube
from isaacsim.core.experimental.prims import GeomPrim, XformPrim
from isaacsim.storage.native import get_assets_root_path_async
from pxr import PhysicsSchemaTools, UsdUtils
NUM_PALLETS = 6
ENV_URL = "/Isaac/Environments/Simple_Warehouse/warehouse.usd"
WRITE_DATA = True
SEED = 42
def resolve_scene_root_path(stage) -> str | None:
"""Return /World, /Root, or the stage default prim path for spawning content."""
if stage is None:
return None
for root_path in ("/World", "/Root"):
if stage.GetPrimAtPath(root_path).IsValid():
return root_path
default_prim = stage.GetDefaultPrim()
if default_prim.IsValid():
return str(default_prim.GetPath())
return None
def add_rigid_body_dynamics(prim, disable_gravity=False, angular_damping=None):
# This flow adds bodies at runtime, so author the APIs without creating a persistent tensor view.
rigid_body_attributes = {
"rigidBodyEnabled": True,
"disableGravity": disable_gravity,
"maxDepenetrationVelocity": 3.0,
"maxLinearVelocity": float("inf"),
}
if angular_damping is not None:
rigid_body_attributes["angularDamping"] = angular_damping
rep.functional.physics.apply_rigid_body(prim, with_collider=False, **rigid_body_attributes)
def create_asset(asset_url, path, *, translations=None, orientations=None, scales=None):
prim_path = stage_utils.generate_next_free_path(path, prepend_default_prim=False)
prim = stage_utils.add_reference_to_stage(usd_path=asset_url, path=prim_path)
XformPrim(
prim_path,
translations=translations,
orientations=orientations,
scales=scales,
reset_xform_op_properties=True,
)
return prim
def create_asset_with_colliders(asset_url, path, *, translations=None, orientations=None, scales=None):
prim = create_asset(asset_url, path, translations=translations, orientations=orientations, scales=scales)
pallet_geom = GeomPrim(f"{prim.GetPath()}/.*", apply_collision_apis=True)
pallet_geom.set_collision_approximations("convexHull")
return prim
def place_collision_walls(
pallet_prim,
walls_root: str,
bbox_cache,
height=2,
thickness=0.3,
material=None,
visible=False,
):
bbox_cache.Clear()
aabb = bounds_utils.compute_aabb(pallet_prim, bbox_cache=bbox_cache, space="untransformed")
width, depth, local_height = aabb[3:] - aabb[:3]
mid = (aabb[:3] + aabb[3:]) * 0.5 + np.array([0.0, 0.0, local_height / 2])
walls = [
("floor", (mid[0], mid[1], mid[2] - thickness / 2), (width, depth, thickness)),
("ceiling", (mid[0], mid[1], mid[2] + height + thickness / 2), (width, depth, thickness)),
(
"left_wall",
(mid[0] - (width + thickness) / 2, mid[1], mid[2] + height / 2),
(thickness, depth, height),
),
(
"right_wall",
(mid[0] + (width + thickness) / 2, mid[1], mid[2] + height / 2),
(thickness, depth, height),
),
(
"front_wall",
(mid[0], mid[1] + (depth + thickness) / 2, mid[2] + height / 2),
(width, thickness, height),
),
(
"back_wall",
(mid[0], mid[1] - (depth + thickness) / 2, mid[2] + height / 2),
(width, thickness, height),
),
]
pallet_position, pallet_orientation = xform_utils.get_world_pose(pallet_prim)
stage = stage_utils.get_current_stage(backend="usd")
for name, location, size in walls:
wall_path = f"{walls_root}/{name}"
wall_scale = (size[0] / 2.0, size[1] / 2.0, size[2] / 2.0)
world_pos = tuple(
transform_utils.transform_local_to_world(location, pallet_position, pallet_orientation).numpy()
)
if stage.GetPrimAtPath(wall_path).IsValid():
wall = XformPrim(
wall_path,
positions=world_pos,
orientations=pallet_orientation.numpy(),
scales=wall_scale,
reset_xform_op_properties=True,
)
wall_geom = GeomPrim(wall_path)
else:
wall = Cube(
wall_path,
sizes=2.0,
positions=world_pos,
orientations=pallet_orientation.numpy(),
scales=wall_scale,
reset_xform_op_properties=True,
)
wall_geom = GeomPrim(wall_path, apply_collision_apis=True)
wall_geom.set_collision_approximations("convexHull")
if material is not None:
wall_geom.apply_physics_materials(material, weaker_than_descendants=[True])
wall.set_visibilities([visible])
async def apply_forces_async(boxes, pallet, strength=550, strength_center_multiplier=2):
stage = stage_utils.get_current_stage(backend="usd")
app_utils.play()
pallet_center, pallet_orientation = xform_utils.get_world_pose(pallet)
force_forward = transform_utils.rotate_vectors_by_quaternion([1.0, 0.0, 0.0], pallet_orientation).numpy() * strength
force_right = transform_utils.rotate_vectors_by_quaternion([0.0, 1.0, 0.0], pallet_orientation).numpy() * strength
physx_simulation_interface = omni.physx.get_physx_simulation_interface()
stage_id = UsdUtils.StageCache.Get().GetId(stage).ToLongInt()
directional_forces = [force_forward, force_right, -force_forward, -force_right]
for box_prim in boxes:
body_path = PhysicsSchemaTools.sdfPathToInt(box_prim.GetPath())
for force in chain(directional_forces, directional_forces):
box_position, _ = xform_utils.get_world_pose(box_prim)
box_position = carb.Float3(*box_position.numpy())
physx_simulation_interface.apply_force_at_pos(stage_id, body_path, carb.Float3(*force), box_position)
await app_utils.update_app_async(steps=10)
for box_prim in boxes:
body_path = PhysicsSchemaTools.sdfPathToInt(box_prim.GetPath())
box_location, _ = xform_utils.get_world_pose(box_prim)
box_location = box_location.numpy()
force_to_center = (pallet_center.numpy() - box_location) * strength * strength_center_multiplier
physx_simulation_interface.apply_force_at_pos(
stage_id,
body_path,
carb.Float3(*force_to_center),
carb.Float3(*box_location),
)
await app_utils.update_app_async(steps=20)
app_utils.pause()
async def stack_boxes_on_pallet_async(
pallet_prim, walls_root, boxes_urls_and_weights, num_boxes, drop_height=1.5, drop_margin=0.2, gen=None
):
pallet_path = pallet_prim.GetPath()
print(f"[BoxStacking] Running scenario for pallet {pallet_path} with {num_boxes} boxes..")
bbox_cache = bounds_utils.create_bbox_cache()
physics_material = RigidBodyMaterial(
f"{pallet_path}/Looks/PhysicsMaterial",
static_frictions=[0.01],
dynamic_frictions=[0.01],
restitutions=[0.0],
)
GeomPrim(f"{pallet_path}/.*").apply_physics_materials(physics_material, weaker_than_descendants=[True])
place_collision_walls(
pallet_prim, walls_root, bbox_cache, height=drop_height + drop_margin, material=physics_material
)
box_urls, box_weights = zip(*boxes_urls_and_weights)
box_weights_arr = np.asarray(box_weights, dtype=float)
box_weights_arr /= box_weights_arr.sum()
rand_boxes_urls = gen.choice(box_urls, size=num_boxes, p=box_weights_arr)
boxes = [create_asset(box_url, f"{pallet_path}_Boxes/Box_{i}") for i, box_url in enumerate(rand_boxes_urls)]
boxes.sort(
key=lambda box: bounds_utils.compute_bound_volume(box, bbox_cache=bbox_cache, space="local"),
reverse=True,
)
spawn_midpoint, spawn_size = bounds_utils.compute_bound_range(
pallet_prim, bbox_cache=bbox_cache, space="untransformed"
)
pallet_width, pallet_depth, pallet_height = spawn_size
spawn_center = spawn_midpoint + np.array([0.0, 0.0, pallet_height / 2 + drop_height])
spawn_width, spawn_depth = pallet_width / 2 - drop_margin, pallet_depth / 2 - drop_margin
pallet_position, pallet_orientation = xform_utils.get_world_pose(pallet_prim)
for box_prim in boxes:
local_loc = spawn_center + np.array(
[gen.uniform(-spawn_width, spawn_width), gen.uniform(-spawn_depth, spawn_depth), 0.0]
)
angles = [
gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-3, 3),
gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-3, 3),
gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-3, 3),
]
local_orientation = np.array([1.0, 0.0, 0.0, 0.0], dtype=np.float32)
for axis_index, angle in enumerate(angles):
axis_angles = [0.0, 0.0, 0.0]
axis_angles[axis_index] = angle
axis_quat = transform_utils.euler_angles_to_quaternion(axis_angles, degrees=True).numpy().astype(np.float32)
local_orientation = transform_utils.quaternion_multiplication(local_orientation, axis_quat).numpy()
world_loc = transform_utils.transform_local_to_world(local_loc, pallet_position, pallet_orientation).numpy()
world_orientation = transform_utils.quaternion_multiplication(pallet_orientation, local_orientation).numpy()
box_path = str(box_prim.GetPath())
XformPrim(
box_path,
positions=tuple(world_loc),
orientations=world_orientation,
reset_xform_op_properties=True,
)
box_geom = GeomPrim(f"{box_path}/.*", apply_collision_apis=True)
box_geom.set_collision_approximations("convexHull")
add_rigid_body_dynamics(box_prim, angular_damping=0.9)
box_geom.apply_physics_materials(physics_material, weaker_than_descendants=[True])
await app_utils.update_app_async()
app_utils.play()
await app_utils.update_app_async(steps=20)
app_utils.pause()
await apply_forces_async(boxes, pallet_prim)
# Flush the velocity changes while simulation is still enabled. Writing velocities after disabling a body
# causes PhysX to reject the operation, and a persistent RigidPrim view would be invalidated by the next batch.
zero_velocities = np.zeros((len(boxes), 3), dtype=np.float32)
rep.functional.modify.attribute(boxes, "physics:velocity", zero_velocities)
rep.functional.modify.attribute(boxes, "physics:angularVelocity", zero_velocities)
omni.physx.get_physx_simulation_interface().flush_changes()
rep.functional.modify.attribute(boxes, "physics:rigidBodyEnabled", False)
omni.physx.get_physx_simulation_interface().flush_changes()
physics_material.set_friction_coefficients(static_frictions=[0.9], dynamic_frictions=[0.9])
return boxes
async def run_box_stacking_scenarios_async(num_pallets, env_url=None, write_data=True, rng=None):
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
assets_root_path = await get_assets_root_path_async()
pallets_urls_and_weights = [
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_PaletteA_01.usd", 0.25),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_PaletteA_02.usd", 0.75),
]
boxes_urls_and_weights = [
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxA_01.usd", 0.02),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxB_01.usd", 0.06),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxC_01.usd", 0.12),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxD_01.usd", 0.80),
]
scene_root = "/World"
if env_url is not None:
env_path = env_url if env_url.startswith("omniverse://") else assets_root_path + env_url
success, _stage = await stage_utils.open_stage_async(env_path)
if not success or _stage is None:
carb.log_error(f"[BoxStacking] Failed to open stage: {env_path}")
return
await app_utils.update_app_async()
resolved_root = resolve_scene_root_path(_stage)
if resolved_root is None:
carb.log_error(
"[BoxStacking] Could not find a valid scene root in the opened environment "
"(expected /World or /Root, or a default prim on the stage)."
)
return
scene_root = resolved_root
print(f"[BoxStacking] Using scene root: {scene_root}")
else:
await stage_utils.create_new_stage_async(template="empty")
rep.functional.create.scope(name="Lights", parent="/World")
rep.functional.create.distant_light(
intensity=400, parent="/World/Lights", name="DistantLight", rotation=(0, 60, 0)
)
rep.functional.create.dome_light(intensity=500, parent="/World/Lights", name="DomeLight")
ground_plane = rep.functional.create.plane(parent=scene_root, name="GroundPlane", scale=(10, 10, 1))
rep.functional.physics.apply_collider(ground_plane)
await app_utils.update_app_async()
pallets = []
pallets_urls, pallets_weights = zip(*pallets_urls_and_weights)
pallet_weights_arr = np.asarray(pallets_weights, dtype=float)
pallet_weights_arr /= pallet_weights_arr.sum()
rand_pallet_urls = gen.choice(pallets_urls, size=num_pallets, p=pallet_weights_arr)
custom_pallet_locations = [
(-9.3, 5.3, 1.3),
(-9.3, 7.3, 1.3),
(-9.3, -0.6, 1.3),
]
gen.shuffle(custom_pallet_locations)
for i, pallet_url in enumerate(rand_pallet_urls):
if env_url is not None:
if i % 2 == 0 and custom_pallet_locations:
rand_loc = custom_pallet_locations.pop()
else:
rand_loc = (
-6.5 + gen.uniform(-0.2, 0.2),
i * 1.75 + gen.uniform(0, 0.2),
gen.uniform(0, 0.2),
)
else:
rand_loc = (
i * 1.5 + gen.uniform(0, 0.2),
gen.uniform(-0.2, 0.2),
0.0,
)
rand_rot = (0, 0, gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-15, 15))
pallet_prim = create_asset_with_colliders(
pallet_url,
f"{scene_root}/Pallet_{i}",
translations=rand_loc,
orientations=transform_utils.euler_angles_to_quaternion(rand_rot, degrees=True).numpy(),
)
pallets.append(pallet_prim)
walls_root = f"{scene_root}/_CollisionWalls"
stage_utils.define_prim(walls_root, "Xform")
total_boxes = []
for pallet in pallets:
drop_height = 1.0 if env_url is not None else 1.5
rand_num_boxes = int(gen.integers(8, 16) if env_url is not None else gen.integers(12, 21))
stacked_boxes = await stack_boxes_on_pallet_async(
pallet,
walls_root,
boxes_urls_and_weights,
num_boxes=rand_num_boxes,
drop_height=drop_height,
gen=gen,
)
total_boxes.extend(stacked_boxes)
if stage_utils.get_current_stage(backend="usd").GetPrimAtPath(walls_root).IsValid():
stage_utils.delete_prim(walls_root)
rep.functional.modify.attribute(total_boxes, "physics:rigidBodyEnabled", True)
omni.physx.get_physx_simulation_interface().flush_changes()
app_utils.play()
await app_utils.update_app_async(steps=200)
app_utils.pause()
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_box_stacking")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
cam = rep.functional.create.camera(position=(5, -5, 2), look_at=(0, 0, 0), name="PalletCamera")
rp = rep.create.render_product(cam, resolution=(512, 512))
writer.attach(rp)
await rep.orchestrator.step_async(rt_subframes=8)
await rep.orchestrator.wait_until_complete_async()
writer.detach()
rp.destroy()
async def run_example_async(num_pallets, env_url, write_data, rng=None):
await run_box_stacking_scenarios_async(num_pallets=num_pallets, env_url=env_url, write_data=write_data, rng=rng)
asyncio.ensure_future(run_example_async(NUM_PALLETS, ENV_URL, WRITE_DATA))
Physics-based Randomized Volume Filling
"""Fill randomized pallet volumes with physics-enabled box assets."""
from isaacsim import SimulationApp
simulation_app = SimulationApp(launch_config={"headless": False})
import argparse
import os
from itertools import chain
from typing import Any
import carb
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.bounds as bounds_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import isaacsim.core.experimental.utils.transform as transform_utils
import isaacsim.core.experimental.utils.xform as xform_utils
import numpy as np
import omni.kit.app
import omni.physx
import omni.replicator.core as rep
from isaacsim.core.experimental.materials import RigidBodyMaterial
from isaacsim.core.experimental.objects import Cube
from isaacsim.core.experimental.prims import GeomPrim, XformPrim
from isaacsim.storage.native import get_assets_root_path
from pxr import PhysicsSchemaTools, UsdUtils
NUM_PALLETS = 6
ENV_URL = "/Isaac/Environments/Simple_Warehouse/warehouse.usd"
WRITE_DATA = True
SEED = 42
def _parse_env_url(env_url: str) -> str | None:
return None if env_url.lower() in {"none", "null"} else env_url
parser = argparse.ArgumentParser()
parser.add_argument("--num-pallets", type=int, default=NUM_PALLETS, help="Number of pallets to spawn.")
parser.add_argument(
"--env-url",
type=_parse_env_url,
default=ENV_URL,
help="Environment USD path relative to the assets root, or none for an empty stage.",
)
parser.add_argument(
"--write-data",
action=argparse.BooleanOptionalAction,
default=WRITE_DATA,
help="Write captured output to disk.",
)
args, _ = parser.parse_known_args()
def resolve_scene_root_path(stage: Any) -> str | None:
"""Return a valid root path for spawning scene content.
Args:
stage: USD stage to inspect.
Returns:
The scene root path, or None if the stage has no usable root.
"""
if stage is None:
return None
for root_path in ("/World", "/Root"):
if stage.GetPrimAtPath(root_path).IsValid():
return root_path
default_prim = stage.GetDefaultPrim()
if default_prim.IsValid():
return str(default_prim.GetPath())
return None
def add_rigid_body_dynamics(prim: Any, disable_gravity: bool = False, angular_damping: float | None = None) -> None:
"""Apply rigid-body dynamics to a prim.
Args:
prim: Prim to configure.
disable_gravity: Whether to disable gravity on the body.
angular_damping: Angular damping to apply, or None to use the default.
"""
# This flow adds bodies at runtime, so author the APIs without creating a persistent tensor view.
rigid_body_attributes = {
"rigidBodyEnabled": True,
"disableGravity": disable_gravity,
"maxDepenetrationVelocity": 3.0,
"maxLinearVelocity": float("inf"),
}
if angular_damping is not None:
rigid_body_attributes["angularDamping"] = angular_damping
rep.functional.physics.apply_rigid_body(prim, with_collider=False, **rigid_body_attributes)
def create_asset(
asset_url: str,
path: str,
*,
translations: Any = None,
orientations: Any = None,
scales: Any = None,
) -> Any:
"""Create a referenced asset with the requested transform.
Args:
asset_url: URL of the USD asset to reference.
path: Desired prim path for the asset.
translations: Local translation values, or None to preserve the asset values.
orientations: Local orientation values, or None to preserve the asset values.
scales: Local scale values, or None to preserve the asset values.
Returns:
The created asset prim.
"""
prim_path = stage_utils.generate_next_free_path(path, prepend_default_prim=False)
prim = stage_utils.add_reference_to_stage(usd_path=asset_url, path=prim_path)
XformPrim(
prim_path,
translations=translations,
orientations=orientations,
scales=scales,
reset_xform_op_properties=True,
)
return prim
def create_asset_with_colliders(
asset_url: str,
path: str,
*,
translations: Any = None,
orientations: Any = None,
scales: Any = None,
) -> Any:
"""Create a referenced asset and apply convex-hull colliders.
Args:
asset_url: URL of the USD asset to reference.
path: Desired prim path for the asset.
translations: Local translation values, or None to preserve the asset values.
orientations: Local orientation values, or None to preserve the asset values.
scales: Local scale values, or None to preserve the asset values.
Returns:
The created asset prim.
"""
prim = create_asset(asset_url, path, translations=translations, orientations=orientations, scales=scales)
pallet_geom = GeomPrim(f"{prim.GetPath()}/.*", apply_collision_apis=True)
pallet_geom.set_collision_approximations("convexHull")
return prim
def place_collision_walls(
pallet_prim: Any,
walls_root: str,
bbox_cache: Any,
height: float = 2,
thickness: float = 0.3,
material: RigidBodyMaterial | None = None,
visible: bool = False,
) -> None:
"""Place collision walls around a pallet.
Args:
pallet_prim: Pallet prim whose bounds define the wall layout.
walls_root: Parent path for the collision walls.
bbox_cache: Bounding-box cache used to measure the pallet.
height: Wall height above the pallet.
thickness: Wall thickness.
material: Physics material to apply, or None to leave the walls unbound.
visible: Whether to render the collision walls.
"""
bbox_cache.Clear()
# Untransformed (pallet-local) bounds so the walls can follow the pallet's position and orientation.
aabb = bounds_utils.compute_aabb(pallet_prim, bbox_cache=bbox_cache, space="untransformed")
width, depth, local_height = aabb[3:] - aabb[:3]
mid = (aabb[:3] + aabb[3:]) * 0.5 + np.array([0.0, 0.0, local_height / 2])
walls = [
("floor", (mid[0], mid[1], mid[2] - thickness / 2), (width, depth, thickness)),
("ceiling", (mid[0], mid[1], mid[2] + height + thickness / 2), (width, depth, thickness)),
(
"left_wall",
(mid[0] - (width + thickness) / 2, mid[1], mid[2] + height / 2),
(thickness, depth, height),
),
(
"right_wall",
(mid[0] + (width + thickness) / 2, mid[1], mid[2] + height / 2),
(thickness, depth, height),
),
(
"front_wall",
(mid[0], mid[1] + (depth + thickness) / 2, mid[2] + height / 2),
(width, thickness, height),
),
(
"back_wall",
(mid[0], mid[1] - (depth + thickness) / 2, mid[2] + height / 2),
(width, thickness, height),
),
]
# Map pallet-local wall poses into the world frame so the shared wall set matches each pallet's pose.
pallet_position, pallet_orientation = xform_utils.get_world_pose(pallet_prim)
stage = stage_utils.get_current_stage(backend="usd")
for name, location, size in walls:
wall_path = f"{walls_root}/{name}"
wall_scale = (size[0] / 2.0, size[1] / 2.0, size[2] / 2.0)
world_pos = tuple(
transform_utils.transform_local_to_world(location, pallet_position, pallet_orientation).numpy()
)
if stage.GetPrimAtPath(wall_path).IsValid():
wall = XformPrim(
wall_path,
positions=world_pos,
orientations=pallet_orientation.numpy(),
scales=wall_scale,
reset_xform_op_properties=True,
)
wall_geom = GeomPrim(wall_path)
else:
wall = Cube(
wall_path,
sizes=2.0,
positions=world_pos,
orientations=pallet_orientation.numpy(),
scales=wall_scale,
reset_xform_op_properties=True,
)
wall_geom = GeomPrim(wall_path, apply_collision_apis=True)
wall_geom.set_collision_approximations("convexHull")
if material is not None:
wall_geom.apply_physics_materials(material, weaker_than_descendants=[True])
wall.set_visibilities([visible])
def apply_forces(boxes: list[Any], pallet: Any, strength: float = 550, strength_center_multiplier: float = 2) -> None:
"""Apply settling forces to boxes on a pallet.
Args:
boxes: Box prims to push from each horizontal direction.
pallet: Pallet prim that defines the final centering force.
strength: Magnitude of each directional force.
strength_center_multiplier: Multiplier for the final centering force.
"""
stage = stage_utils.get_current_stage(backend="usd")
app_utils.play()
pallet_center, pallet_orientation = xform_utils.get_world_pose(pallet)
force_forward = transform_utils.rotate_vectors_by_quaternion([1.0, 0.0, 0.0], pallet_orientation).numpy() * strength
force_right = transform_utils.rotate_vectors_by_quaternion([0.0, 1.0, 0.0], pallet_orientation).numpy() * strength
physx_simulation_interface = omni.physx.get_physx_simulation_interface()
stage_id = UsdUtils.StageCache.Get().GetId(stage).ToLongInt()
directional_forces = [force_forward, force_right, -force_forward, -force_right]
for box_prim in boxes:
body_path = PhysicsSchemaTools.sdfPathToInt(box_prim.GetPath())
for force in chain(directional_forces, directional_forces):
box_position, _ = xform_utils.get_world_pose(box_prim)
box_position = carb.Float3(*box_position.numpy())
physx_simulation_interface.apply_force_at_pos(stage_id, body_path, carb.Float3(*force), box_position)
app_utils.update_app(steps=10)
for box_prim in boxes:
body_path = PhysicsSchemaTools.sdfPathToInt(box_prim.GetPath())
box_location, _ = xform_utils.get_world_pose(box_prim)
box_location = box_location.numpy()
force_to_center = (pallet_center.numpy() - box_location) * strength * strength_center_multiplier
physx_simulation_interface.apply_force_at_pos(
stage_id,
body_path,
carb.Float3(*force_to_center),
carb.Float3(*box_location),
)
app_utils.update_app(steps=20)
app_utils.pause()
def stack_boxes_on_pallet(
pallet_prim: Any,
walls_root: str,
boxes_urls_and_weights: list[tuple[str, float]],
num_boxes: int,
drop_height: float = 1.5,
drop_margin: float = 0.2,
gen: Any = None,
) -> list[Any]:
"""Stack randomized boxes on one pallet.
Args:
pallet_prim: Pallet prim on which to stack boxes.
walls_root: Parent path for temporary collision walls.
boxes_urls_and_weights: Box asset URLs paired with sampling weights.
num_boxes: Number of boxes to create.
drop_height: Vertical offset of the temporary spawn volume above the pallet.
drop_margin: Horizontal inset from each pallet edge used for spawning.
gen: NumPy-compatible random generator used to sample box assets and poses. Must not be None.
Returns:
The stacked box prims.
"""
pallet_path = pallet_prim.GetPath()
print(f"[BoxStacking] Running scenario for pallet {pallet_path} with {num_boxes} boxes..")
bbox_cache = bounds_utils.create_bbox_cache()
physics_material = RigidBodyMaterial(
f"{pallet_path}/Looks/PhysicsMaterial",
static_frictions=[0.01],
dynamic_frictions=[0.01],
restitutions=[0.0],
)
GeomPrim(f"{pallet_path}/.*").apply_physics_materials(physics_material, weaker_than_descendants=[True])
place_collision_walls(
pallet_prim, walls_root, bbox_cache, height=drop_height + drop_margin, material=physics_material
)
box_urls, box_weights = zip(*boxes_urls_and_weights)
box_weights_arr = np.asarray(box_weights, dtype=float)
box_weights_arr /= box_weights_arr.sum()
rand_boxes_urls = gen.choice(box_urls, size=num_boxes, p=box_weights_arr)
boxes = [create_asset(box_url, f"{pallet_path}_Boxes/Box_{i}") for i, box_url in enumerate(rand_boxes_urls)]
boxes.sort(
key=lambda box: bounds_utils.compute_bound_volume(box, bbox_cache=bbox_cache, space="local"),
reverse=True,
)
spawn_midpoint, spawn_size = bounds_utils.compute_bound_range(
pallet_prim, bbox_cache=bbox_cache, space="untransformed"
)
pallet_width, pallet_depth, pallet_height = spawn_size
spawn_center = spawn_midpoint + np.array([0.0, 0.0, pallet_height / 2 + drop_height])
spawn_width, spawn_depth = pallet_width / 2 - drop_margin, pallet_depth / 2 - drop_margin
pallet_position, pallet_orientation = xform_utils.get_world_pose(pallet_prim)
for box_prim in boxes:
local_loc = spawn_center + np.array(
[gen.uniform(-spawn_width, spawn_width), gen.uniform(-spawn_depth, spawn_depth), 0.0]
)
angles = [
gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-3, 3),
gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-3, 3),
gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-3, 3),
]
local_orientation = np.array([1.0, 0.0, 0.0, 0.0], dtype=np.float32)
for axis_index, angle in enumerate(angles):
axis_angles = [0.0, 0.0, 0.0]
axis_angles[axis_index] = angle
axis_quat = transform_utils.euler_angles_to_quaternion(axis_angles, degrees=True).numpy().astype(np.float32)
local_orientation = transform_utils.quaternion_multiplication(local_orientation, axis_quat).numpy()
world_loc = transform_utils.transform_local_to_world(local_loc, pallet_position, pallet_orientation).numpy()
world_orientation = transform_utils.quaternion_multiplication(pallet_orientation, local_orientation).numpy()
box_path = str(box_prim.GetPath())
XformPrim(
box_path,
positions=tuple(world_loc),
orientations=world_orientation,
reset_xform_op_properties=True,
)
box_geom = GeomPrim(f"{box_path}/.*", apply_collision_apis=True)
box_geom.set_collision_approximations("convexHull")
add_rigid_body_dynamics(box_prim, angular_damping=0.9)
box_geom.apply_physics_materials(physics_material, weaker_than_descendants=[True])
app_utils.update_app()
app_utils.play()
app_utils.update_app(steps=20)
app_utils.pause()
apply_forces(boxes, pallet_prim)
# Flush the velocity changes while simulation is still enabled. Writing velocities after disabling a body
# causes PhysX to reject the operation, and a persistent RigidPrim view would be invalidated by the next batch.
zero_velocities = np.zeros((len(boxes), 3), dtype=np.float32)
rep.functional.modify.attribute(boxes, "physics:velocity", zero_velocities)
rep.functional.modify.attribute(boxes, "physics:angularVelocity", zero_velocities)
omni.physx.get_physx_simulation_interface().flush_changes()
rep.functional.modify.attribute(boxes, "physics:rigidBodyEnabled", False)
omni.physx.get_physx_simulation_interface().flush_changes()
physics_material.set_friction_coefficients(static_frictions=[0.9], dynamic_frictions=[0.9])
return boxes
def run_box_stacking_scenarios(
num_pallets: int, env_url: str | None = None, write_data: bool = True, rng: Any = None
) -> None:
"""Run randomized box-stacking scenarios.
Args:
num_pallets: Number of pallets to populate.
env_url: Environment URL, or None to create an empty stage.
write_data: Whether to capture and write an RGB image.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
if rng is None:
rep.set_global_seed(SEED)
rng = rep.rng.ReplicatorRNG(seed=SEED)
gen = rng.generator
assets_root_path = get_assets_root_path()
pallets_urls_and_weights = [
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_PaletteA_01.usd", 0.25),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_PaletteA_02.usd", 0.75),
]
boxes_urls_and_weights = [
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxA_01.usd", 0.02),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxB_01.usd", 0.06),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxC_01.usd", 0.12),
(assets_root_path + "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxD_01.usd", 0.80),
]
scene_root = "/World"
if env_url is not None:
env_path = env_url if env_url.startswith("omniverse://") else assets_root_path + env_url
success, _stage = stage_utils.open_stage(env_path)
if not success or _stage is None:
carb.log_error(f"[BoxStacking] Failed to open stage: {env_path}")
return
app_utils.update_app()
resolved_root = resolve_scene_root_path(_stage)
if resolved_root is None:
carb.log_error(
"[BoxStacking] Could not find a valid scene root in the opened environment "
"(expected /World or /Root, or a default prim on the stage)."
)
return
scene_root = resolved_root
print(f"[BoxStacking] Using scene root: {scene_root}")
else:
stage_utils.create_new_stage(template="empty")
rep.functional.create.scope(name="Lights", parent="/World")
rep.functional.create.distant_light(
intensity=400, parent="/World/Lights", name="DistantLight", rotation=(0, 60, 0)
)
rep.functional.create.dome_light(intensity=500, parent="/World/Lights", name="DomeLight")
ground_plane = rep.functional.create.plane(parent=scene_root, name="GroundPlane", scale=(10, 10, 1))
rep.functional.physics.apply_collider(ground_plane)
app_utils.update_app()
pallets = []
pallets_urls, pallets_weights = zip(*pallets_urls_and_weights)
pallet_weights_arr = np.asarray(pallets_weights, dtype=float)
pallet_weights_arr /= pallet_weights_arr.sum()
rand_pallet_urls = gen.choice(pallets_urls, size=num_pallets, p=pallet_weights_arr)
custom_pallet_locations = [
(-9.3, 5.3, 1.3),
(-9.3, 7.3, 1.3),
(-9.3, -0.6, 1.3),
]
gen.shuffle(custom_pallet_locations)
for i, pallet_url in enumerate(rand_pallet_urls):
if env_url is not None:
if i % 2 == 0 and custom_pallet_locations:
rand_loc = custom_pallet_locations.pop()
else:
rand_loc = (
-6.5 + gen.uniform(-0.2, 0.2),
i * 1.75 + gen.uniform(0, 0.2),
gen.uniform(0, 0.2),
)
else:
rand_loc = (
i * 1.5 + gen.uniform(0, 0.2),
gen.uniform(-0.2, 0.2),
0.0,
)
rand_rot = (0, 0, gen.choice([180, 90, 0, -90, -180]) + gen.uniform(-15, 15))
pallet_prim = create_asset_with_colliders(
pallet_url,
f"{scene_root}/Pallet_{i}",
translations=rand_loc,
orientations=transform_utils.euler_angles_to_quaternion(rand_rot, degrees=True).numpy(),
)
pallets.append(pallet_prim)
walls_root = f"{scene_root}/_CollisionWalls"
stage_utils.define_prim(walls_root, "Xform")
total_boxes = []
for pallet in pallets:
drop_height = 1.0 if env_url is not None else 1.5
rand_num_boxes = int(gen.integers(8, 16) if env_url is not None else gen.integers(12, 21))
stacked_boxes = stack_boxes_on_pallet(
pallet,
walls_root,
boxes_urls_and_weights,
num_boxes=rand_num_boxes,
drop_height=drop_height,
gen=gen,
)
total_boxes.extend(stacked_boxes)
if stage_utils.get_current_stage(backend="usd").GetPrimAtPath(walls_root).IsValid():
stage_utils.delete_prim(walls_root)
rep.functional.modify.attribute(total_boxes, "physics:rigidBodyEnabled", True)
omni.physx.get_physx_simulation_interface().flush_changes()
app_utils.play()
app_utils.update_app(steps=200)
app_utils.pause()
if write_data:
out_dir = os.path.join(os.getcwd(), "_out_box_stacking")
print(f"Writing data to {out_dir}..")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=out_dir)
writer = rep.WriterRegistry.get("BasicWriter")
writer.initialize(backend=backend, rgb=True)
cam = rep.functional.create.camera(position=(5, -5, 2), look_at=(0, 0, 0), name="PalletCamera")
rp = rep.create.render_product(cam, resolution=(512, 512))
writer.attach(rp)
rep.orchestrator.step(rt_subframes=8)
rep.orchestrator.wait_until_complete()
writer.detach()
rp.destroy()
def run_example(num_pallets: int, env_url: str | None, write_data: bool, rng: Any = None) -> None:
"""Run the pallet volume-filling example.
Args:
num_pallets: Number of pallets to populate.
env_url: Environment URL, or None to create an empty stage.
write_data: Whether to capture and write an RGB image.
rng: Replicator random-number generator, or None to create one from the example seed.
"""
run_box_stacking_scenarios(num_pallets=num_pallets, env_url=env_url, write_data=write_data, rng=rng)
run_example(args.num_pallets, args.env_url, args.write_data)
SimReady Assets SDG Example#
This example uses SimReady Assets to assemble randomized table-top scenes. SimReady Assets are physically accurate 3D objects with realistic properties, behavior, and data connections that are optimized for simulation.
Note
The Script Editor snippet runs in async mode and requires the SimReady Explorer window to be enabled to process the search requests. The standalone application enables the extension before running the SDG pipeline.
The example creates an SDG randomization and capture pipeline with a table, a plate, and a number of items on top of the plate. The Script Editor version always writes the captured images to disk. The standalone application also writes captures with its default writer settings.
The standalone application can be run directly (on Windows use python.bat instead of python.sh):
./python.sh standalone_examples/api/isaacsim.replicator.examples/simready_assets_sdg.py
Optional flags include --num-scenarios. The Script Editor snippet exposes the same value through NUM_SCENARIOS.
SimReady Assets SDG Example
import asyncio
import os
import time
import carb.settings
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.bounds as bounds_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import numpy as np
import omni.replicator.core as rep
from isaacsim.core.experimental.utils.semantics import upgrade_prim_semantics_to_labels
from isaacsim.core.simulation_manager import SimulationManager
from pxr import Sdf, Usd, UsdPhysics
NUM_SCENARIOS = 5
SEED = 34
if not app_utils.is_extension_enabled("omni.simready.explorer"):
app_utils.enable_extension("omni.simready.explorer")
import omni.simready.explorer as sre
def enable_simready_explorer() -> None:
"""Enable the SimReady Explorer window if not already open."""
if sre.get_instance().browser_model is None:
import omni.kit.actions.core as actions
actions.execute_action("omni.simready.explorer", "toggle_window")
def set_prim_variants(prim: Usd.Prim, variants: dict[str, str]) -> None:
"""Set variant selections on a prim from a dictionary of variant set names to values."""
vsets = prim.GetVariantSets()
for name, value in variants.items():
vset = vsets.GetVariantSet(name)
if vset:
vset.SetVariantSelection(value)
async def search_assets_async() -> tuple[list, list, list]:
"""Search for SimReady assets (tables, dishes, items) asynchronously."""
print(f"[SDG] Searching for SimReady assets...")
start_time = time.time()
tables = await sre.find_assets(["table", "furniture"])
print(f"[SDG] - Found {len(tables)} tables ({time.time() - start_time:.2f}s)")
start_time = time.time()
plates = await sre.find_assets(["plate"])
print(f"[SDG] - Found {len(plates)} plates ({time.time() - start_time:.2f}s)")
start_time = time.time()
bowls = await sre.find_assets(["bowl"])
print(f"[SDG] - Found {len(bowls)} bowls ({time.time() - start_time:.2f}s)")
dishes = plates + bowls
start_time = time.time()
fruits = await sre.find_assets(["fruit"])
print(f"[SDG] - Found {len(fruits)} fruits ({time.time() - start_time:.2f}s)")
start_time = time.time()
vegetables = await sre.find_assets(["vegetable"])
print(f"[SDG] - Found {len(vegetables)} vegetables ({time.time() - start_time:.2f}s)")
items = fruits + vegetables
return tables, dishes, items
async def run_simready_randomization_async(
stage: Usd.Stage,
camera_prim: Usd.Prim,
render_product,
tables: list,
dishes: list,
items: list,
rng: np.random.Generator = None,
) -> None:
"""Randomize a scene with SimReady assets, run physics, and capture the result."""
if rng is None:
rng = np.random.default_rng()
print(f"[SDG] Creating anonymous variation layer for the randomizations...")
root_layer = stage.GetRootLayer()
variation_layer = Sdf.Layer.CreateAnonymous("variation")
root_layer.subLayerPaths.insert(0, variation_layer.identifier)
stage.SetEditTarget(variation_layer)
variants = {"PhysicsVariant": "RigidBody"}
rep.functional.create.scope(name="Assets")
print(f"[SDG] Loading assets...")
table_asset = tables[rng.integers(len(tables))]
start_time = time.time()
table_prim = rep.functional.create.reference(usd_path=table_asset.main_url, parent="/Assets", name=table_asset.name)
set_prim_variants(table_prim, variants)
upgrade_prim_semantics_to_labels(table_prim)
print(f"[SDG] - Table: '{table_asset.name}' ({time.time() - start_time:.2f}s)")
await app_utils.update_app_async()
UsdPhysics.RigidBodyAPI(table_prim).GetRigidBodyEnabledAttr().Set(False)
bbox_cache = bounds_utils.create_bbox_cache()
table_aabb = bounds_utils.compute_aabb(table_prim, bbox_cache=bbox_cache, space="world")
table_extent = table_aabb[3:] - table_aabb[:3]
dish_asset = dishes[rng.integers(len(dishes))]
start_time = time.time()
dish_prim = rep.functional.create.reference(usd_path=dish_asset.main_url, parent="/Assets", name=dish_asset.name)
set_prim_variants(dish_prim, variants)
upgrade_prim_semantics_to_labels(dish_prim)
print(f"[SDG] - Dish: '{dish_asset.name}' ({time.time() - start_time:.2f}s)")
await app_utils.update_app_async()
dish_aabb = bounds_utils.compute_aabb(dish_prim, bbox_cache=bbox_cache, space="world")
dish_extent = dish_aabb[3:] - dish_aabb[:3]
center_region_scale = 0.75
dish_range_x = max(0, (table_extent[0] - dish_extent[0]) / 2 * center_region_scale)
dish_range_y = max(0, (table_extent[1] - dish_extent[1]) / 2 * center_region_scale)
dish_position = (
rng.uniform(-dish_range_x, dish_range_x) if dish_range_x > 0 else 0,
rng.uniform(-dish_range_y, dish_range_y) if dish_range_y > 0 else 0,
table_extent[2] + dish_extent[2] / 2,
)
rep.functional.modify.pose(dish_prim, position_value=dish_position)
num_items = rng.integers(2, 5)
item_prims = []
for _ in range(num_items):
item_asset = items[rng.integers(len(items))]
start_time = time.time()
item_prim = rep.functional.create.reference(
usd_path=item_asset.main_url, parent="/Assets", name=item_asset.name
)
set_prim_variants(item_prim, variants)
upgrade_prim_semantics_to_labels(item_prim)
print(f"[SDG] - Item: '{item_asset.name}' ({time.time() - start_time:.2f}s)")
item_prims.append(item_prim)
await app_utils.update_app_async()
print(f"[SDG] Positioning assets on table...")
stack_height = dish_position[2]
item_scatter_radius = max(0, dish_extent[0] / 4)
for item_prim in item_prims:
item_aabb = bounds_utils.compute_aabb(item_prim, bbox_cache=bbox_cache, space="world")
item_extent = item_aabb[3:] - item_aabb[:3]
scatter_x = rng.uniform(-item_scatter_radius, item_scatter_radius) if item_scatter_radius > 0 else 0
scatter_y = rng.uniform(-item_scatter_radius, item_scatter_radius) if item_scatter_radius > 0 else 0
item_position = (
dish_position[0] + scatter_x,
dish_position[1] + scatter_y,
stack_height + item_extent[2] / 2,
)
rep.functional.modify.pose(item_prim, position_value=item_position)
stack_height += item_extent[2]
num_sim_steps = 25
print(f"[SDG] Running physics simulation ({num_sim_steps} steps)...")
SimulationManager.invalidate_physics()
SimulationManager.initialize_physics()
SimulationManager.step(steps=num_sim_steps)
print(f"[SDG] Setting edit target to root layer...")
stage.SetEditTarget(root_layer)
print(f"[SDG] Positioning camera and capturing frame...")
camera_position = (
dish_position[0] + rng.uniform(-0.5, 0.5),
dish_position[1] + rng.uniform(-0.5, 0.5),
dish_position[2] + 1.5 + rng.uniform(-0.5, 0.5),
)
rep.functional.modify.pose(
camera_prim, position_value=camera_position, look_at_value=dish_prim, look_at_up_axis=(0, 0, 1)
)
render_product.hydra_texture.set_updates_enabled(True)
await rep.orchestrator.step_async(delta_time=0.0, rt_subframes=16)
render_product.hydra_texture.set_updates_enabled(False)
print(f"[SDG] Removing temp variation layer...")
variation_layer.Clear()
root_layer.subLayerPaths.remove(variation_layer.identifier)
async def run_simready_randomizations_async(num_scenarios: int, output_dir: str | None = None) -> None:
"""Run multiple SimReady randomization scenarios and capture the results."""
print(f"[SDG] Initializing scene...")
stage = await stage_utils.create_new_stage_async()
rng = np.random.default_rng(SEED)
rep.set_global_seed(SEED)
rep.orchestrator.set_capture_on_play(False)
SimulationManager.set_physics_dt(1.0 / 60.0)
carb.settings.get_settings().set("rtx/post/dlss/execMode", 2)
print(f"[SDG] Setting up lighting...")
rep.functional.create.xform(name="World")
rep.functional.create.dome_light(intensity=500, parent="/World", name="DomeLight")
rep.functional.create.distant_light(intensity=2500, parent="/World", name="DistantLight", rotation=(-75, 0, 0))
enable_simready_explorer()
tables, dishes, items = await search_assets_async()
if output_dir is None:
output_dir = os.path.join(os.getcwd(), "_out_simready_assets")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=output_dir)
writer = rep.writers.get("BasicWriter")
print(f"[SDG] Initializing writer, output directory: {output_dir}...")
writer.initialize(backend=backend, rgb=True)
print(f"[SDG] Creating camera and render product...")
camera_prim = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), parent="/World", name="Camera")
rp = rep.create.render_product(camera_prim, (512, 512))
rp.hydra_texture.set_updates_enabled(False)
writer.attach(rp)
for i in range(num_scenarios):
print(f"[SDG] Scenario {i + 1}/{num_scenarios}")
await run_simready_randomization_async(
stage=stage, camera_prim=camera_prim, render_product=rp, tables=tables, dishes=dishes, items=items, rng=rng
)
print("[SDG] Wait for the data to be written and cleanup render products...")
await rep.orchestrator.wait_until_complete_async()
writer.detach()
rp.destroy()
print(f"[SDG] Starting SDG pipeline with {NUM_SCENARIOS} scenarios...")
asyncio.ensure_future(run_simready_randomizations_async(NUM_SCENARIOS))
SimReady Assets SDG Example
"""Demonstrate synthetic data generation using SimReady assets with randomized scenes."""
from isaacsim import SimulationApp
simulation_app = SimulationApp(launch_config={"headless": False})
import argparse
import asyncio
import os
import sys
import time
from typing import Any
import carb.settings
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.bounds as bounds_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import numpy as np
import omni.replicator.core as rep
from isaacsim.core.experimental.utils.semantics import upgrade_prim_semantics_to_labels
from isaacsim.core.simulation_manager import SimulationManager
from pxr import Sdf, Usd, UsdPhysics
NUM_SCENARIOS = 5
SEED = 34
parser = argparse.ArgumentParser()
parser.add_argument(
"--num-scenarios", type=int, default=NUM_SCENARIOS, help="Number of randomization scenarios to create"
)
args, _ = parser.parse_known_args()
num_scenarios = args.num_scenarios
if not app_utils.is_extension_enabled("omni.simready.explorer"):
app_utils.enable_extension("omni.simready.explorer")
import omni.simready.explorer as sre
def enable_simready_explorer() -> None:
"""Enable the SimReady Explorer window if not already open."""
if sre.get_instance().browser_model is None:
import omni.kit.actions.core as actions
actions.execute_action("omni.simready.explorer", "toggle_window")
def set_prim_variants(prim: Usd.Prim, variants: dict[str, str]) -> None:
"""Set variant selections on a prim from a dictionary of variant set names to values.
Args:
prim: USD prim containing the variant sets to update.
variants: Desired selection keyed by variant-set name.
"""
vsets = prim.GetVariantSets()
for name, value in variants.items():
vset = vsets.GetVariantSet(name)
if vset:
vset.SetVariantSelection(value)
async def search_assets_async() -> tuple[list, list, list]:
"""Search for SimReady assets (tables, dishes, items) asynchronously.
Returns:
Separate table, dish, and food-item search result lists.
"""
print(f"[SDG] Searching for SimReady assets...")
start_time = time.time()
tables = await sre.find_assets(["table", "furniture"])
print(f"[SDG] - Found {len(tables)} tables ({time.time() - start_time:.2f}s)")
start_time = time.time()
plates = await sre.find_assets(["plate"])
print(f"[SDG] - Found {len(plates)} plates ({time.time() - start_time:.2f}s)")
start_time = time.time()
bowls = await sre.find_assets(["bowl"])
print(f"[SDG] - Found {len(bowls)} bowls ({time.time() - start_time:.2f}s)")
dishes = plates + bowls
start_time = time.time()
fruits = await sre.find_assets(["fruit"])
print(f"[SDG] - Found {len(fruits)} fruits ({time.time() - start_time:.2f}s)")
start_time = time.time()
vegetables = await sre.find_assets(["vegetable"])
print(f"[SDG] - Found {len(vegetables)} vegetables ({time.time() - start_time:.2f}s)")
items = fruits + vegetables
return tables, dishes, items
def search_assets() -> tuple[list, list, list]:
"""Run SimReady asset search while pumping the app (required in standalone SimulationApp).
Returns:
Separate table, dish, and food-item search result lists.
"""
search_task = asyncio.ensure_future(search_assets_async())
while not search_task.done():
app_utils.update_app()
return search_task.result()
def run_simready_randomization(
stage: Usd.Stage,
camera_prim: Usd.Prim,
render_product: Any,
tables: list,
dishes: list,
items: list,
rng: np.random.Generator | None = None,
) -> None:
"""Arrange SimReady tableware, settle it with physics, and capture an RGB frame.
Args:
stage: Stage on which to author the temporary randomized asset layer.
camera_prim: Camera to position above the selected dish.
render_product: Render product to enable while capturing the scenario.
tables: SimReady table search results from which to select one asset.
dishes: SimReady dish search results from which to select one asset.
items: SimReady food-item search results from which to select two to four assets.
rng: NumPy random generator used for asset and pose selection, or None to create an unseeded generator.
"""
if rng is None:
rng = np.random.default_rng()
print(f"[SDG] Creating anonymous variation layer for the randomizations...")
root_layer = stage.GetRootLayer()
variation_layer = Sdf.Layer.CreateAnonymous("variation")
root_layer.subLayerPaths.insert(0, variation_layer.identifier)
stage.SetEditTarget(variation_layer)
variants = {"PhysicsVariant": "RigidBody"}
rep.functional.create.scope(name="Assets")
print(f"[SDG] Loading assets...")
table_asset = tables[rng.integers(len(tables))]
start_time = time.time()
table_prim = rep.functional.create.reference(usd_path=table_asset.main_url, parent="/Assets", name=table_asset.name)
set_prim_variants(table_prim, variants)
upgrade_prim_semantics_to_labels(table_prim)
print(f"[SDG] - Table: '{table_asset.name}' ({time.time() - start_time:.2f}s)")
app_utils.update_app()
UsdPhysics.RigidBodyAPI(table_prim).GetRigidBodyEnabledAttr().Set(False)
bbox_cache = bounds_utils.create_bbox_cache()
table_aabb = bounds_utils.compute_aabb(table_prim, bbox_cache=bbox_cache, space="world")
table_extent = table_aabb[3:] - table_aabb[:3]
dish_asset = dishes[rng.integers(len(dishes))]
start_time = time.time()
dish_prim = rep.functional.create.reference(usd_path=dish_asset.main_url, parent="/Assets", name=dish_asset.name)
set_prim_variants(dish_prim, variants)
upgrade_prim_semantics_to_labels(dish_prim)
print(f"[SDG] - Dish: '{dish_asset.name}' ({time.time() - start_time:.2f}s)")
app_utils.update_app()
dish_aabb = bounds_utils.compute_aabb(dish_prim, bbox_cache=bbox_cache, space="world")
dish_extent = dish_aabb[3:] - dish_aabb[:3]
center_region_scale = 0.75
dish_range_x = max(0, (table_extent[0] - dish_extent[0]) / 2 * center_region_scale)
dish_range_y = max(0, (table_extent[1] - dish_extent[1]) / 2 * center_region_scale)
dish_position = (
rng.uniform(-dish_range_x, dish_range_x) if dish_range_x > 0 else 0,
rng.uniform(-dish_range_y, dish_range_y) if dish_range_y > 0 else 0,
table_extent[2] + dish_extent[2] / 2,
)
rep.functional.modify.pose(dish_prim, position_value=dish_position)
num_items = rng.integers(2, 5)
item_prims = []
for _ in range(num_items):
item_asset = items[rng.integers(len(items))]
start_time = time.time()
item_prim = rep.functional.create.reference(
usd_path=item_asset.main_url, parent="/Assets", name=item_asset.name
)
set_prim_variants(item_prim, variants)
upgrade_prim_semantics_to_labels(item_prim)
print(f"[SDG] - Item: '{item_asset.name}' ({time.time() - start_time:.2f}s)")
item_prims.append(item_prim)
app_utils.update_app()
print(f"[SDG] Positioning assets on table...")
stack_height = dish_position[2]
item_scatter_radius = max(0, dish_extent[0] / 4)
for item_prim in item_prims:
item_aabb = bounds_utils.compute_aabb(item_prim, bbox_cache=bbox_cache, space="world")
item_extent = item_aabb[3:] - item_aabb[:3]
scatter_x = rng.uniform(-item_scatter_radius, item_scatter_radius) if item_scatter_radius > 0 else 0
scatter_y = rng.uniform(-item_scatter_radius, item_scatter_radius) if item_scatter_radius > 0 else 0
item_position = (
dish_position[0] + scatter_x,
dish_position[1] + scatter_y,
stack_height + item_extent[2] / 2,
)
rep.functional.modify.pose(item_prim, position_value=item_position)
stack_height += item_extent[2]
num_sim_steps = 25
print(f"[SDG] Running physics simulation ({num_sim_steps} steps)...")
SimulationManager.invalidate_physics()
SimulationManager.initialize_physics()
SimulationManager.step(steps=num_sim_steps)
print(f"[SDG] Setting edit target to root layer...")
stage.SetEditTarget(root_layer)
print(f"[SDG] Positioning camera and capturing frame...")
camera_position = (
dish_position[0] + rng.uniform(-0.5, 0.5),
dish_position[1] + rng.uniform(-0.5, 0.5),
dish_position[2] + 1.5 + rng.uniform(-0.5, 0.5),
)
rep.functional.modify.pose(
camera_prim, position_value=camera_position, look_at_value=dish_prim, look_at_up_axis=(0, 0, 1)
)
render_product.hydra_texture.set_updates_enabled(True)
rep.orchestrator.step(delta_time=0.0, rt_subframes=16)
render_product.hydra_texture.set_updates_enabled(False)
print(f"[SDG] Removing temp variation layer...")
variation_layer.Clear()
root_layer.subLayerPaths.remove(variation_layer.identifier)
def run_simready_randomizations(num_scenarios: int, output_dir: str | None = None) -> None:
"""Run multiple SimReady randomization scenarios and capture the results.
Args:
num_scenarios: Number of independently randomized scenes to capture.
output_dir: Directory for captured RGB images, or None to use the example output directory.
"""
print(f"[SDG] Initializing scene...")
stage_utils.create_new_stage()
stage = stage_utils.get_current_stage(backend="usd")
rng = np.random.default_rng(SEED)
rep.set_global_seed(SEED)
rep.orchestrator.set_capture_on_play(False)
SimulationManager.set_physics_dt(1.0 / 60.0)
carb.settings.get_settings().set("rtx/post/dlss/execMode", 2)
print(f"[SDG] Setting up lighting...")
rep.functional.create.xform(name="World")
rep.functional.create.dome_light(intensity=500, parent="/World", name="DomeLight")
rep.functional.create.distant_light(intensity=2500, parent="/World", name="DistantLight", rotation=(-75, 0, 0))
enable_simready_explorer()
tables, dishes, items = search_assets()
if output_dir is None:
output_dir = os.path.join(os.getcwd(), "_out_simready_assets")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=output_dir)
writer = rep.writers.get("BasicWriter")
print(f"[SDG] Initializing writer, output directory: {output_dir}...")
writer.initialize(backend=backend, rgb=True)
print(f"[SDG] Creating camera and render product...")
camera_prim = rep.functional.create.camera(position=(5, 5, 5), look_at=(0, 0, 0), parent="/World", name="Camera")
rp = rep.create.render_product(camera_prim, (512, 512))
rp.hydra_texture.set_updates_enabled(False)
writer.attach(rp)
for i in range(num_scenarios):
print(f"[SDG] Scenario {i + 1}/{num_scenarios}")
run_simready_randomization(
stage=stage, camera_prim=camera_prim, render_product=rp, tables=tables, dishes=dishes, items=items, rng=rng
)
print("[SDG] Wait for the data to be written and cleanup render products...")
rep.orchestrator.wait_until_complete()
writer.detach()
rp.destroy()
print(f"[SDG] Starting SDG pipeline with {num_scenarios} scenarios...")
run_simready_randomizations(num_scenarios)
Object Reconstruction Assets SDG Example#
This example uses a mesh created with NVIDIA 3D Object Reconstruction, a workflow that turns stereo video of a real object into a textured, simulation-ready USD mesh. The example batch-creates every instance of the reconstructed mesh in a single call, then drops them onto a floor with colliders in sequential waves: each wave is released only after the previous wave has settled onto the growing pile, and a frame is captured after each wave settles.
Each capture randomizes the floor color and writes RGB, colorized depth, and semantic segmentation data with BasicWriter. A dome light with an intensity of 500 lights the scene.
The standalone application can be run directly (on Windows use python.bat instead of python.sh):
./python.sh standalone_examples/api/isaacsim.replicator.examples/object_reconstruction_assets_sdg.py
Optional flags include --num-assets, --num-waves, and --num-scenarios. The Script Editor snippet exposes the same values through NUM_ASSETS, NUM_WAVES, and NUM_SCENARIOS constants.
Object Reconstruction Assets SDG Example
import asyncio
import os
import random
import carb.settings
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import numpy as np
import omni.replicator.core as rep
from isaacsim.core.experimental.prims import RigidPrim
from isaacsim.storage.native import get_assets_root_path_async
# Asset produced by the NVIDIA 3D Object Reconstruction workflow (https://github.com/NVIDIA/3DObjectReconstruction)
TOASTER_PASTRY_USD = "/Isaac/Samples/Replicator/3DObjectReconstruction/toaster_pastry.usd"
NUM_ASSETS = 24 # Total instances spawned per scenario; any remainder is released in the final wave
NUM_WAVES = 2 # Number of sequential waves per scenario, each settled wave is captured before the next is released
NUM_SCENARIOS = 2 # Number of times the wave sequence is reset and repeated; total captures = NUM_SCENARIOS * NUM_WAVES
BASE_DROP_HEIGHT = 1.0 # Spawn height for the first wave of a scenario (m)
HEIGHT_STEP_PER_WAVE = 0.3 # Extra spawn height per wave, giving clearance above the growing pile (m)
WITHIN_WAVE_HEIGHT_JITTER = 0.05 # Per-asset height variance within a wave to avoid exact overlaps (m)
SPAWN_XY_JITTER = 0.35 # Random horizontal offset +/- (m)
FALL_SPEED_THRESHOLD = 0.3 # Linear speed above which a wave is considered actively falling (m/s)
SETTLE_SPEED_THRESHOLD = 0.02 # Linear speed below which a wave is considered settled (m/s)
RNG_SEED = 17 # Reproducible randomization seed
MAX_STEPS_PER_WAVE = 50 # Maximum simulation steps to wait for a wave to settle
async def run_example_async(num_assets: int, num_waves: int, num_scenarios: int) -> None:
assets_per_wave, remainder = divmod(num_assets, num_waves)
if remainder:
print(
f"[SDG] Asset count is not divisible by wave count; last wave takes the remainder of {remainder} "
f"(size {assets_per_wave + remainder})."
)
await stage_utils.create_new_stage_async()
assets_root_path = await get_assets_root_path_async()
rng = random.Random(RNG_SEED)
# Disable capture on play, frames will be captured manually
rep.orchestrator.set_capture_on_play(False)
# Set DLSS to Quality mode (2) for best SDG results (Options: 0 (Performance), 1 (Balanced), 2 (Quality), 3 (Auto))
carb.settings.get_settings().set("rtx/post/dlss/execMode", 2)
rep.functional.create.xform(name="World")
rep.functional.create.dome_light(intensity=500, parent="/World", name="DomeLight")
rep.functional.create.scope(name="Looks", parent="/World")
# Using a cube to avoid collision tunneling: a zero-thickness plane can let fast-falling assets pass through.
floor = rep.functional.create.cube(parent="/World", name="Floor", scale=(5, 5, 0.05))
rep.functional.physics.apply_collider(floor)
floor_material = rep.functional.create.material(
mdl="OmniPBR.mdl",
diffuse_color_constant=(0.6, 0.6, 0.6),
bind_prims=floor,
parent="/World/Looks",
name="FloorMaterial",
)
# Batch-create every instance in a single call instead of looping over individual create.reference calls.
all_prims = rep.functional.create_batch.reference(
usd_path=assets_root_path + TOASTER_PASTRY_USD,
semantics={"class": "toaster_pastry"},
parent="/World",
name="ToasterPastry",
count=num_assets,
)
# Every asset starts out kinematic: it ignores gravity and can be freely repositioned until its wave releases
# it by disabling kinematic mode. A single persistent view covers every asset across all waves and scenarios.
rep.functional.physics.apply_rigid_body(all_prims, approximation="convexHull", kinematicEnabled=True)
all_rigid_prims = RigidPrim([prim.GetPath().pathString for prim in all_prims])
# Replicator setup, render product is disabled by default and enabled only at capture time
camera = rep.functional.create.camera(position=(1.5, 1.5, 1.5), look_at=(0, 0, 0), parent="/World")
render_product = rep.create.render_product(camera, (640, 480))
# Enable render product updates only at capture time
render_product.hydra_texture.set_updates_enabled(False)
output_dir = os.path.join(os.getcwd(), "_out_object_reconstruction_assets_drop")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=output_dir)
writer = rep.writers.get("BasicWriter")
writer.initialize(
backend=backend,
rgb=True,
distance_to_camera=True,
colorize_depth=True,
semantic_segmentation=True,
colorize_semantic_segmentation=True,
)
writer.attach(render_product)
# Start the simulation
print("[SDG] Starting simulation")
app_utils.play()
for scenario_idx in range(num_scenarios):
# Reset every asset for a new scenario: freeze it back to kinematic (no-op the first time) and stagger
# each wave's spawn height for clearance above the pile the previous scenario left behind.
rep.functional.modify.attribute(all_prims, "physics:kinematicEnabled", True)
positions, rotations = [], []
for i in range(num_assets):
wave_idx = min(i // assets_per_wave, num_waves - 1)
x = rng.uniform(-SPAWN_XY_JITTER, SPAWN_XY_JITTER)
y = rng.uniform(-SPAWN_XY_JITTER, SPAWN_XY_JITTER)
z = BASE_DROP_HEIGHT + wave_idx * HEIGHT_STEP_PER_WAVE + rng.uniform(0, WITHIN_WAVE_HEIGHT_JITTER)
positions.append((x, y, z))
rotations.append((rng.uniform(0, 360), rng.uniform(0, 360), rng.uniform(0, 360)))
rep.functional.modify.pose(all_prims, position_value=positions, rotation_value=rotations)
# Let the kinematic targets sync to their new poses before releasing any wave, otherwise the released
# wave inherits a large implied velocity from the pose jump instead of starting at rest.
await app_utils.update_app_async()
# Drop one wave at a time: release it, wait for it to settle onto the (growing) pile, then capture
for wave_idx in range(num_waves):
wave_end = num_assets if wave_idx == num_waves - 1 else (wave_idx + 1) * assets_per_wave
wave_slice = slice(wave_idx * assets_per_wave, wave_end)
wave_prims = all_prims[wave_slice]
wave_size = wave_slice.stop - wave_slice.start
# Release this wave from kinematic mode so gravity takes over
rep.functional.modify.attribute(wave_prims, "physics:kinematicEnabled", False)
# Zero out the implied teleport velocity only after disabling kinematic mode, since PhysX
# rejects velocity writes on kinematic bodies.
all_rigid_prims.set_velocities(
linear_velocities=np.zeros((wave_size, 3)),
angular_velocities=np.zeros((wave_size, 3)),
indices=list(range(wave_slice.start, wave_slice.stop)),
)
# Step the simulation until the wave has fallen and settled
has_fallen = False
for _ in range(MAX_STEPS_PER_WAVE):
await app_utils.update_app_async()
linear_velocities, _ = all_rigid_prims.get_velocities()
max_speed = float(np.linalg.norm(linear_velocities.numpy()[wave_slice], axis=1).max())
if not has_fallen:
# Wait until the wave has started moving before checking for a settled state
if max_speed > FALL_SPEED_THRESHOLD:
has_fallen = True
continue
if max_speed < SETTLE_SPEED_THRESHOLD:
break
# Randomize the floor color and capture a frame of the settled wave
color = (rng.random(), rng.random(), rng.random())
rep.functional.modify.attribute(floor_material, "inputs:diffuse_color_constant", color)
print(
f"[SDG] Scenario {scenario_idx + 1}/{num_scenarios} - Wave {wave_idx + 1}/{num_waves} settled, "
f"floor color -> {tuple(round(c, 2) for c in color)}"
)
render_product.hydra_texture.set_updates_enabled(True)
print(f"[SDG] Capturing frame {scenario_idx * num_waves + wave_idx + 1}/{num_scenarios * num_waves}")
await rep.orchestrator.step_async(delta_time=0.0, pause_timeline=False, rt_subframes=16)
render_product.hydra_texture.set_updates_enabled(False)
# Pause the simulation and clean up resources
total_captures = num_scenarios * num_waves
print(f"[SDG] Simulation complete. {total_captures} frames saved to {output_dir}")
app_utils.pause()
await rep.orchestrator.wait_until_complete_async()
writer.detach()
render_product.destroy()
asyncio.ensure_future(run_example_async(NUM_ASSETS, NUM_WAVES, NUM_SCENARIOS))
Object Reconstruction Assets SDG Example
"""Demonstrate synthetic data generation with a 3D object-reconstructed real-world asset."""
from isaacsim import SimulationApp
simulation_app = SimulationApp(launch_config={"headless": False})
import argparse
import os
import random
import carb
import carb.settings
import isaacsim.core.experimental.utils.app as app_utils
import isaacsim.core.experimental.utils.stage as stage_utils
import numpy as np
import omni.replicator.core as rep
from isaacsim.core.experimental.prims import RigidPrim
from isaacsim.storage.native import get_assets_root_path
# Asset produced by the NVIDIA 3D Object Reconstruction workflow (https://github.com/NVIDIA/3DObjectReconstruction)
TOASTER_PASTRY_USD = "/Isaac/Samples/Replicator/3DObjectReconstruction/toaster_pastry.usd"
NUM_ASSETS = 24 # Total instances spawned per scenario; any remainder is released in the final wave
NUM_WAVES = 2 # Number of sequential waves per scenario, each settled wave is captured before the next is released
NUM_SCENARIOS = 2 # Number of times the wave sequence is reset and repeated; total captures = NUM_SCENARIOS * NUM_WAVES
BASE_DROP_HEIGHT = 1.0 # Spawn height for the first wave of a scenario (m)
HEIGHT_STEP_PER_WAVE = 0.3 # Extra spawn height per wave, giving clearance above the growing pile (m)
WITHIN_WAVE_HEIGHT_JITTER = 0.05 # Per-asset height variance within a wave to avoid exact overlaps (m)
SPAWN_XY_JITTER = 0.35 # Random horizontal offset +/- (m)
FALL_SPEED_THRESHOLD = 0.3 # Linear speed above which a wave is considered actively falling (m/s)
SETTLE_SPEED_THRESHOLD = 0.02 # Linear speed below which a wave is considered settled (m/s)
RNG_SEED = 17 # Reproducible randomization seed
MAX_STEPS_PER_WAVE = 50 # Maximum simulation steps to wait for a wave to settle
parser = argparse.ArgumentParser()
parser.add_argument(
"--num-assets", type=int, default=NUM_ASSETS, help="Total number of asset instances to spawn per scenario."
)
parser.add_argument("--num-waves", type=int, default=NUM_WAVES, help="Number of sequential waves per scenario.")
parser.add_argument(
"--num-scenarios", type=int, default=NUM_SCENARIOS, help="Number of times the wave sequence is reset and repeated."
)
args, _ = parser.parse_known_args()
def run_example(num_assets: int, num_waves: int, num_scenarios: int) -> None:
"""Drop batches of a 3D-reconstructed asset onto a growing pile and capture a frame per settled wave.
Args:
num_assets: Total number of asset instances to spawn per scenario; any remainder is released in the final wave.
num_waves: Number of sequential waves per scenario. Each wave is released only after the previous one settled.
num_scenarios: Number of times the wave sequence is reset (assets re-spawned) and repeated.
"""
if num_waves <= 0:
raise ValueError("num_waves must be positive.")
if num_assets < num_waves:
raise ValueError("num_assets must be at least num_waves.")
assets_per_wave, remainder = divmod(num_assets, num_waves)
if remainder:
print(
f"[SDG] Asset count is not divisible by wave count; last wave takes the remainder of {remainder} "
f"(size {assets_per_wave + remainder})."
)
stage_utils.create_new_stage()
assets_root_path = get_assets_root_path()
rng = random.Random(RNG_SEED)
# Disable capture on play, frames will be captured manually
rep.orchestrator.set_capture_on_play(False)
# Set DLSS to Quality mode (2) for best SDG results (Options: 0 (Performance), 1 (Balanced), 2 (Quality), 3 (Auto))
carb.settings.get_settings().set("rtx/post/dlss/execMode", 2)
rep.functional.create.xform(name="World")
rep.functional.create.dome_light(intensity=500, parent="/World", name="DomeLight")
rep.functional.create.scope(name="Looks", parent="/World")
# Using a cube to avoid collision tunneling: a zero-thickness plane can let fast-falling assets pass through.
floor = rep.functional.create.cube(parent="/World", name="Floor", scale=(5, 5, 0.05))
rep.functional.physics.apply_collider(floor)
floor_material = rep.functional.create.material(
mdl="OmniPBR.mdl",
diffuse_color_constant=(0.6, 0.6, 0.6),
bind_prims=floor,
parent="/World/Looks",
name="FloorMaterial",
)
# Batch-create every instance in a single call instead of looping over individual create.reference calls.
all_prims = rep.functional.create_batch.reference(
usd_path=assets_root_path + TOASTER_PASTRY_USD,
semantics={"class": "toaster_pastry"},
parent="/World",
name="ToasterPastry",
count=num_assets,
)
# Every asset starts out kinematic: it ignores gravity and can be freely repositioned until its wave releases
# it by disabling kinematic mode. A single persistent view covers every asset across all waves and scenarios.
rep.functional.physics.apply_rigid_body(all_prims, approximation="convexHull", kinematicEnabled=True)
all_rigid_prims = RigidPrim([prim.GetPath().pathString for prim in all_prims])
# Replicator setup, render product is disabled by default and enabled only at capture time
camera = rep.functional.create.camera(position=(1.5, 1.5, 1.5), look_at=(0, 0, 0), parent="/World")
render_product = rep.create.render_product(camera, (640, 480))
# Enable render product updates only at capture time
render_product.hydra_texture.set_updates_enabled(False)
output_dir = os.path.join(os.getcwd(), "_out_object_reconstruction_assets_drop")
backend = rep.backends.get("DiskBackend")
backend.initialize(output_dir=output_dir)
writer = rep.writers.get("BasicWriter")
writer.initialize(
backend=backend,
rgb=True,
distance_to_camera=True,
colorize_depth=True,
semantic_segmentation=True,
colorize_semantic_segmentation=True,
)
writer.attach(render_product)
# Start the simulation
print("[SDG] Starting simulation")
app_utils.play()
for scenario_idx in range(num_scenarios):
# Reset every asset for a new scenario: freeze it back to kinematic (no-op the first time) and stagger
# each wave's spawn height for clearance above the pile the previous scenario left behind.
rep.functional.modify.attribute(all_prims, "physics:kinematicEnabled", True)
positions, rotations = [], []
for i in range(num_assets):
wave_idx = min(i // assets_per_wave, num_waves - 1)
x = rng.uniform(-SPAWN_XY_JITTER, SPAWN_XY_JITTER)
y = rng.uniform(-SPAWN_XY_JITTER, SPAWN_XY_JITTER)
z = BASE_DROP_HEIGHT + wave_idx * HEIGHT_STEP_PER_WAVE + rng.uniform(0, WITHIN_WAVE_HEIGHT_JITTER)
positions.append((x, y, z))
rotations.append((rng.uniform(0, 360), rng.uniform(0, 360), rng.uniform(0, 360)))
rep.functional.modify.pose(all_prims, position_value=positions, rotation_value=rotations)
# Let the kinematic targets sync to their new poses before releasing any wave, otherwise the released
# wave inherits a large implied velocity from the pose jump instead of starting at rest.
app_utils.update_app()
# Drop one wave at a time: release it, wait for it to settle onto the (growing) pile, then capture
for wave_idx in range(num_waves):
wave_end = num_assets if wave_idx == num_waves - 1 else (wave_idx + 1) * assets_per_wave
wave_slice = slice(wave_idx * assets_per_wave, wave_end)
wave_prims = all_prims[wave_slice]
wave_size = wave_slice.stop - wave_slice.start
# Release this wave from kinematic mode so gravity takes over
rep.functional.modify.attribute(wave_prims, "physics:kinematicEnabled", False)
# Zero out the implied teleport velocity only after disabling kinematic mode, since PhysX
# rejects velocity writes on kinematic bodies.
all_rigid_prims.set_velocities(
linear_velocities=np.zeros((wave_size, 3)),
angular_velocities=np.zeros((wave_size, 3)),
indices=list(range(wave_slice.start, wave_slice.stop)),
)
# Step the simulation until the wave has fallen and settled
has_fallen = False
for _ in range(MAX_STEPS_PER_WAVE):
app_utils.update_app()
linear_velocities, _ = all_rigid_prims.get_velocities()
max_speed = float(np.linalg.norm(linear_velocities.numpy()[wave_slice], axis=1).max())
if not has_fallen:
# Wait until the wave has started moving before checking for a settled state
if max_speed > FALL_SPEED_THRESHOLD:
has_fallen = True
continue
if max_speed < SETTLE_SPEED_THRESHOLD:
break
# Randomize the floor color and capture a frame of the settled wave
color = (rng.random(), rng.random(), rng.random())
rep.functional.modify.attribute(floor_material, "inputs:diffuse_color_constant", color)
print(
f"[SDG] Scenario {scenario_idx + 1}/{num_scenarios} - Wave {wave_idx + 1}/{num_waves} settled, "
f"floor color -> {tuple(round(c, 2) for c in color)}"
)
render_product.hydra_texture.set_updates_enabled(True)
rep.orchestrator.step(delta_time=0.0, pause_timeline=False, rt_subframes=16)
render_product.hydra_texture.set_updates_enabled(False)
# Pause the simulation and clean up resources
total_captures = num_scenarios * num_waves
print(f"[SDG] Simulation complete. {total_captures} frames saved to {output_dir}")
app_utils.pause()
rep.orchestrator.wait_until_complete()
writer.detach()
render_product.destroy()
run_example(args.num_assets, args.num_waves, args.num_scenarios)