Custom Replicator Randomization Nodes#
This tutorial provides an example of how to create custom randomization nodes for the omni.replicator extension.
Learning Objectives#
The goal of this tutorial is to demonstrate how to create custom OmniGraph randomization nodes. These nodes can then be further integrated into the Synthetic Data Generation (SDG) pipeline graph of Replicator.
This tutorial will showcase how to:
Create custom scene randomization Python scripts.
Wrap the scripts as OmniGraph nodes and manually add them to an existing SDG pipeline graph.
Encapsulate the OmniGraph nodes as ReplicatorItems to be automatically added to the SDG pipeline graph using Replicator’s API.
Prerequisites#
Familiarity with Replicator functional APIs and Isaac Sim stage utilities for creating custom scene randomizers. See Randomization Snippets for more details.
Familiarity with omni.replicator and its randomization API replicator randomizers.
Basic knowledge of OmniGraph and how to create OmniGraph Nodes.
Experience running simulations via the Script Editor.
Implementation#
This tutorial will showcase how to create custom scene randomization Python scripts. These scripts create prims in a new stage and randomize their rotation and locations: in a sphere, on a sphere, and between two spheres. Positions are sampled with NumPy and applied in batch through rep.functional.modify.pose.
The following image shows the result after running the randomization in the Script Editor:
The following functions take sphere radius values and return batched position samples for on a sphere, within a sphere, and between two spheres. Rotations use uniform Euler angles in degrees.
Randomization Functions
def sample_points_on_sphere(radius: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points on the surface of a sphere."""
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
x = radius * np.sin(theta) * np.cos(phi)
y = radius * np.sin(theta) * np.sin(phi)
z = radius * np.cos(theta)
return np.stack([x, y, z], axis=1)
def sample_points_in_sphere(radius: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points uniformly within a sphere volume."""
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
r = radius * (rng.random(count) ** (1 / 3))
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)
return np.stack([x, y, z], axis=1)
def sample_points_between_spheres(radius1: float, radius2: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points uniformly between two concentric spheres."""
if radius1 > radius2:
radius1, radius2 = radius2, radius1
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
r = rng.uniform(radius1**3, radius2**3, count) ** (1 / 3)
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)
return np.stack([x, y, z], axis=1)
The following snippet creates prims in a new stage and randomizes their rotation and locations using rep.functional.create_batch and rep.functional.modify.pose. Spheres are placed on a sphere, cubes in a sphere, and cylinders between two spheres so each shape maps to a distinct randomizer in the viewport.
Spawning and Randomizing Prims
async def run_example_async():
await stage_utils.create_new_stage_async()
await app_utils.update_app_async()
rep.functional.create.xform(name="World")
rep.functional.create.dome_light(parent="/World", intensity=500)
rng = rep.rng.ReplicatorRNG(seed=42).generator
on_sphere_prims = rep.functional.create_batch.sphere(
count=PRIM_COUNT, parent="/World", name="sphere", scale=PRIM_SCALE
)
in_sphere_prims = rep.functional.create_batch.cube(count=PRIM_COUNT, parent="/World", name="cube", scale=PRIM_SCALE)
between_spheres_prims = rep.functional.create_batch.cylinder(
count=PRIM_COUNT, parent="/World", name="cylinder", scale=PRIM_SCALE
)
for _ in range(NUM_ITERATIONS):
await app_utils.update_app_async()
rep.functional.modify.pose(
in_sphere_prims,
position_value=sample_points_in_sphere(RAD_IN, PRIM_COUNT, rng),
rotation_value=rng.uniform(0, 360, size=(PRIM_COUNT, 3)),
)
rep.functional.modify.pose(
on_sphere_prims,
position_value=sample_points_on_sphere(RAD_ON, PRIM_COUNT, rng),
rotation_value=rng.uniform(0, 360, size=(PRIM_COUNT, 3)),
)
rep.functional.modify.pose(
between_spheres_prims,
position_value=sample_points_between_spheres(RAD_BET1, RAD_BET2, PRIM_COUNT, rng),
rotation_value=rng.uniform(0, 360, size=(PRIM_COUNT, 3)),
)
Snippet to run in the Script Editor:
Full Script Editor Script
import asyncio
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
PRIM_COUNT = 300
PRIM_SCALE = 0.1
RAD_IN = 0.5
RAD_ON = 1.5
RAD_BET1 = 2.5
RAD_BET2 = 3.5
NUM_ITERATIONS = 10
def sample_points_on_sphere(radius: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points on the surface of a sphere."""
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
x = radius * np.sin(theta) * np.cos(phi)
y = radius * np.sin(theta) * np.sin(phi)
z = radius * np.cos(theta)
return np.stack([x, y, z], axis=1)
def sample_points_in_sphere(radius: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points uniformly within a sphere volume."""
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
r = radius * (rng.random(count) ** (1 / 3))
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)
return np.stack([x, y, z], axis=1)
def sample_points_between_spheres(radius1: float, radius2: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points uniformly between two concentric spheres."""
if radius1 > radius2:
radius1, radius2 = radius2, radius1
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
r = rng.uniform(radius1**3, radius2**3, count) ** (1 / 3)
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)
return np.stack([x, y, z], axis=1)
async def run_example_async():
await stage_utils.create_new_stage_async()
await app_utils.update_app_async()
rep.functional.create.xform(name="World")
rep.functional.create.dome_light(parent="/World", intensity=500)
rng = rep.rng.ReplicatorRNG(seed=42).generator
on_sphere_prims = rep.functional.create_batch.sphere(
count=PRIM_COUNT, parent="/World", name="sphere", scale=PRIM_SCALE
)
in_sphere_prims = rep.functional.create_batch.cube(count=PRIM_COUNT, parent="/World", name="cube", scale=PRIM_SCALE)
between_spheres_prims = rep.functional.create_batch.cylinder(
count=PRIM_COUNT, parent="/World", name="cylinder", scale=PRIM_SCALE
)
for _ in range(NUM_ITERATIONS):
await app_utils.update_app_async()
rep.functional.modify.pose(
in_sphere_prims,
position_value=sample_points_in_sphere(RAD_IN, PRIM_COUNT, rng),
rotation_value=rng.uniform(0, 360, size=(PRIM_COUNT, 3)),
)
rep.functional.modify.pose(
on_sphere_prims,
position_value=sample_points_on_sphere(RAD_ON, PRIM_COUNT, rng),
rotation_value=rng.uniform(0, 360, size=(PRIM_COUNT, 3)),
)
rep.functional.modify.pose(
between_spheres_prims,
position_value=sample_points_between_spheres(RAD_BET1, RAD_BET2, PRIM_COUNT, rng),
rotation_value=rng.uniform(0, 360, size=(PRIM_COUNT, 3)),
)
asyncio.ensure_future(run_example_async())
As a next step, custom OmniGraph Nodes are created for the randomization functions. Each node implementation inlines the same NumPy sampling logic shown above and writes xformOp:translate on stage prims resolved from inputs:prims. The node descriptions and implementations can be found in the following code snippets:
OgnSampleInSphere.ogn
{
"OgnSampleInSphere": {
"version": 2,
"description": "Assigns uniformly sampled location in a sphere.",
"language": "Python",
"categoryDefinitions": "config/CategoryDefinition.json",
"categories": ["isaacReplicatorExamples"],
"icon": "icons/isaac-sim.svg",
"metadata": {
"uiName": "Sample In Sphere"
},
"inputs": {
"prims": {
"type": "target",
"description": "prims to randomize",
"default": []
},
"execIn": {
"type": "execution",
"description": "exec",
"default": 0
},
"radius": {
"type": "float",
"description": "sphere radius",
"default": 1.0
},
"seed": {
"type": "int",
"description": "Random number generator seed. A negative value uses the Replicator global seed.",
"default": -1
}
},
"outputs": {
"execOut": {
"type": "execution",
"description": "exec"
}
}
}
}
OgnSampleOnSphere.ogn
{
"OgnSampleOnSphere": {
"version": 2,
"description": "Assigns uniformly sampled location on a sphere.",
"language": "Python",
"categoryDefinitions": "config/CategoryDefinition.json",
"categories": ["isaacReplicatorExamples"],
"icon": "icons/isaac-sim.svg",
"metadata": {
"uiName": "Sample On Sphere"
},
"inputs": {
"prims": {
"type": "target",
"description": "prims to randomize",
"default": []
},
"execIn": {
"type": "execution",
"description": "exec",
"default": 0
},
"radius": {
"type": "float",
"description": "sphere radius",
"default": 1.0
},
"seed": {
"type": "int",
"description": "Random number generator seed. A negative value uses the Replicator global seed.",
"default": -1
}
},
"outputs": {
"execOut": {
"type": "execution",
"description": "exec"
}
}
}
}
OgnSampleBetweenSpheres.ogn
{
"OgnSampleBetweenSpheres": {
"version": 2,
"description": "Assigns uniformly sampled between two spheres",
"language": "Python",
"categoryDefinitions": "config/CategoryDefinition.json",
"categories": ["isaacReplicatorExamples"],
"icon": "icons/isaac-sim.svg",
"metadata": {
"uiName": "Sample Between Spheres"
},
"inputs": {
"prims": {
"type": "target",
"description": "prims to randomize",
"default": []
},
"execIn": {
"type": "execution",
"description": "exec",
"default": 0
},
"radius1": {
"type": "float",
"description": "inner sphere radius",
"default": 0.5
},
"radius2": {
"type": "float",
"description": "outer sphere radius",
"default": 1.0
},
"seed": {
"type": "int",
"description": "Random number generator seed. A negative value uses the Replicator global seed.",
"default": -1
}
},
"outputs": {
"execOut": {
"type": "execution",
"description": "exec"
}
}
}
}
OgnSampleInSphere.py
"""Sample target prim positions uniformly inside a sphere."""
from typing import Any
import numpy as np
import omni.graph.core as og
import omni.replicator.core as rep
import omni.usd
from pxr import Sdf, UsdGeom
def sample_points_in_sphere(radius: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points uniformly within a sphere volume.
Args:
radius: Positive sphere radius. The node validates this precondition before calling the helper.
count: Nonnegative number of points to sample.
rng: Random number generator used for sampling.
Returns:
Cartesian point coordinates with shape ``(count, 3)``.
"""
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
r = radius * (rng.random(count) ** (1 / 3))
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)
return np.stack([x, y, z], axis=1)
class OgnSampleInSphereInternalState:
"""Store a Replicator-aware random number generator across node evaluations."""
def __init__(self) -> None:
self.rng = rep.rng.ReplicatorRNG()
class OgnSampleInSphere:
"""Replicator OmniGraph node that writes random positions within one radius."""
@staticmethod
def internal_state() -> OgnSampleInSphereInternalState:
"""Create the node's persistent internal state.
Returns:
Persistent state for the node.
"""
return OgnSampleInSphereInternalState()
@staticmethod
def release(node: og.Node) -> None:
"""Release subscriptions owned by the node's random number generator.
Args:
node: OmniGraph node whose subscriptions are released.
"""
rep.rng.release(node.get_prim_path())
@staticmethod
def compute(db: Any) -> bool:
"""Move each input prim to a uniformly sampled point inside a sphere.
The node reads target prim paths from ``inputs:prims`` and the radius from ``inputs:radius``.
It samples direction uniformly and scales radius by the cube root of a random value so points
are distributed through volume rather than clustered near the center. Positions are written
to ``xformOp:translate`` on each target prim. Empty prim inputs, invalid prim paths, or a
non-positive radius disable ``outputs:execOut`` and return ``False``.
Args:
db: OmniGraph database object containing node inputs and outputs.
Returns:
True when all target prims are sampled and updated, False otherwise.
"""
prim_paths = db.inputs.prims
if len(prim_paths) == 0:
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
radius = db.inputs.radius
if radius <= 0:
db.log_error(f"Radius must be positive, got {radius}")
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
stage = omni.usd.get_context().get_stage()
prims = [stage.GetPrimAtPath(str(path)) for path in prim_paths]
try:
for prim in prims:
if not prim.IsValid():
raise ValueError(f"Invalid prim path: {prim.GetPath()}")
if not UsdGeom.Xformable(prim):
raise ValueError(
f"Expected prim at {prim.GetPath()} to be an Xformable prim but got type "
f"{prim.GetTypeName()}"
)
if not prim.HasAttribute("xformOp:translate"):
UsdGeom.Xformable(prim).AddTranslateOp()
except Exception as error:
db.log_error(str(error))
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
state = db.shared_state
if state.rng.seed != db.inputs.seed:
node_id = (
db.node.get_attribute("inputs:nodeId").get() if db.node.get_attribute_exists("inputs:nodeId") else 0
)
state.rng.initialize(db.inputs.seed, db.node, node_id)
positions = sample_points_in_sphere(radius, len(prims), state.rng.generator)
with Sdf.ChangeBlock():
for prim, position in zip(prims, positions):
prim.GetAttribute("xformOp:translate").Set(tuple(position))
db.outputs.execOut = og.ExecutionAttributeState.ENABLED
return True
OgnSampleOnSphere.py
"""Sample target prim positions uniformly on a sphere surface."""
from typing import Any
import numpy as np
import omni.graph.core as og
import omni.replicator.core as rep
import omni.usd
from pxr import Sdf, UsdGeom
def sample_points_on_sphere(radius: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points on the surface of a sphere.
Args:
radius: Positive sphere radius. The node validates this precondition before calling the helper.
count: Nonnegative number of points to sample.
rng: Random number generator used for sampling.
Returns:
Cartesian point coordinates with shape ``(count, 3)``.
"""
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
x = radius * np.sin(theta) * np.cos(phi)
y = radius * np.sin(theta) * np.sin(phi)
z = radius * np.cos(theta)
return np.stack([x, y, z], axis=1)
class OgnSampleOnSphereInternalState:
"""Store a Replicator-aware random number generator across node evaluations."""
def __init__(self) -> None:
self.rng = rep.rng.ReplicatorRNG()
class OgnSampleOnSphere:
"""Replicator OmniGraph node that writes random positions at a fixed radius."""
@staticmethod
def internal_state() -> OgnSampleOnSphereInternalState:
"""Create the node's persistent internal state.
Returns:
Persistent state for the node.
"""
return OgnSampleOnSphereInternalState()
@staticmethod
def release(node: og.Node) -> None:
"""Release subscriptions owned by the node's random number generator.
Args:
node: OmniGraph node whose subscriptions are released.
"""
rep.rng.release(node.get_prim_path())
@staticmethod
def compute(db: Any) -> bool:
"""Move each input prim to a uniformly sampled point on a sphere surface.
The node reads target prim paths from ``inputs:prims`` and the fixed radius from
``inputs:radius``. It samples spherical directions by uniform azimuth and uniform
``cos(theta)`` so the surface distribution is not biased toward the poles. Positions
are written to ``xformOp:translate`` on each target prim. Empty prim inputs, invalid
prim paths, or a non-positive radius disable ``outputs:execOut`` and return ``False``.
Args:
db: OmniGraph database object containing node inputs and outputs.
Returns:
True when all target prims are sampled and updated, False otherwise.
"""
prim_paths = db.inputs.prims
if len(prim_paths) == 0:
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
radius = db.inputs.radius
if radius <= 0:
db.log_error(f"Radius must be positive, got {radius}")
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
stage = omni.usd.get_context().get_stage()
prims = [stage.GetPrimAtPath(str(path)) for path in prim_paths]
try:
for prim in prims:
if not prim.IsValid():
raise ValueError(f"Invalid prim path: {prim.GetPath()}")
if not UsdGeom.Xformable(prim):
raise ValueError(
f"Expected prim at {prim.GetPath()} to be an Xformable prim but got type "
f"{prim.GetTypeName()}"
)
if not prim.HasAttribute("xformOp:translate"):
UsdGeom.Xformable(prim).AddTranslateOp()
except Exception as error:
db.log_error(str(error))
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
state = db.shared_state
if state.rng.seed != db.inputs.seed:
node_id = (
db.node.get_attribute("inputs:nodeId").get() if db.node.get_attribute_exists("inputs:nodeId") else 0
)
state.rng.initialize(db.inputs.seed, db.node, node_id)
positions = sample_points_on_sphere(radius, len(prims), state.rng.generator)
with Sdf.ChangeBlock():
for prim, position in zip(prims, positions):
prim.GetAttribute("xformOp:translate").Set(tuple(position))
db.outputs.execOut = og.ExecutionAttributeState.ENABLED
return True
OgnSampleBetweenSpheres.py
"""Sample target prim positions uniformly inside a spherical shell."""
from typing import Any
import numpy as np
import omni.graph.core as og
import omni.replicator.core as rep
import omni.usd
from pxr import Sdf, UsdGeom
def sample_points_between_spheres(radius1: float, radius2: float, count: int, rng: np.random.Generator) -> np.ndarray:
"""Sample random 3D points uniformly between two concentric spheres.
Args:
radius1: First shell boundary, expected to be nonnegative after node validation.
radius2: Second shell boundary, expected to be positive after node validation. The helper orders the
validated boundaries before sampling.
count: Nonnegative number of points to sample.
rng: Random number generator used for sampling.
Returns:
Cartesian point coordinates with shape ``(count, 3)``.
"""
if radius1 > radius2:
radius1, radius2 = radius2, radius1
phi = rng.uniform(0, 2 * np.pi, count)
costheta = rng.uniform(-1, 1, count)
theta = np.arccos(costheta)
r = rng.uniform(radius1**3, radius2**3, count) ** (1 / 3)
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)
return np.stack([x, y, z], axis=1)
class OgnSampleBetweenSpheresInternalState:
"""Store a Replicator-aware random number generator across node evaluations."""
def __init__(self) -> None:
self.rng = rep.rng.ReplicatorRNG()
class OgnSampleBetweenSpheres:
"""Replicator OmniGraph node that writes random positions between two radii."""
@staticmethod
def internal_state() -> OgnSampleBetweenSpheresInternalState:
"""Create the node's persistent internal state.
Returns:
Persistent state for the node.
"""
return OgnSampleBetweenSpheresInternalState()
@staticmethod
def release(node: og.Node) -> None:
"""Release subscriptions owned by the node's random number generator.
Args:
node: OmniGraph node whose subscriptions are released.
"""
rep.rng.release(node.get_prim_path())
@staticmethod
def compute(db: Any) -> bool:
"""Move each input prim to a uniformly sampled point between two concentric spheres.
The node accepts the target prim paths from ``inputs:prims`` and two shell radii from
``inputs:radius1`` and ``inputs:radius2``. Empty prim input disables ``outputs:execOut`` and
returns ``False`` without logging. Otherwise, the node first requires ``radius1`` to be
nonnegative and ``radius2`` to be positive, then orders valid radii for sampling. Positions
are written to ``xformOp:translate`` on each target prim. Invalid prims or invalid radii log
an error, disable ``outputs:execOut``, and return ``False``.
Args:
db: OmniGraph database object containing node inputs and outputs.
Returns:
True when all target prims are sampled and updated, False otherwise.
"""
prim_paths = db.inputs.prims
if len(prim_paths) == 0:
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
radius1 = db.inputs.radius1
radius2 = db.inputs.radius2
if radius1 < 0 or radius2 <= 0:
db.log_error(f"Radius must be positive and larger radius larger than 0, got {radius1} and {radius2}")
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
stage = omni.usd.get_context().get_stage()
prims = [stage.GetPrimAtPath(str(path)) for path in prim_paths]
try:
for prim in prims:
if not prim.IsValid():
raise ValueError(f"Invalid prim path: {prim.GetPath()}")
if not UsdGeom.Xformable(prim):
raise ValueError(
f"Expected prim at {prim.GetPath()} to be an Xformable prim but got type "
f"{prim.GetTypeName()}"
)
if not prim.HasAttribute("xformOp:translate"):
UsdGeom.Xformable(prim).AddTranslateOp()
except Exception as error:
db.log_error(str(error))
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False
state = db.shared_state
if state.rng.seed != db.inputs.seed:
node_id = (
db.node.get_attribute("inputs:nodeId").get() if db.node.get_attribute_exists("inputs:nodeId") else 0
)
state.rng.initialize(db.inputs.seed, db.node, node_id)
positions = sample_points_between_spheres(radius1, radius2, len(prims), state.rng.generator)
with Sdf.ChangeBlock():
for prim, position in zip(prims, positions):
prim.GetAttribute("xformOp:translate").Set(tuple(position))
db.outputs.execOut = og.ExecutionAttributeState.ENABLED
return True
After this step, the randomizers will be available as nodes in the graph editor. For this tutorial the nodes are part of the built-in isaacsim.replicator.examples extension. Other custom nodes created through the OmniGraph tutorial will be accessible through the omni.new.extension extension (if the default tutorial-provided extension name was used). An example of accessing the nodes in an action graph is depicted below:
Note
If the custom nodes are not available, the corresponding extension needs to be enabled under Window > Extensions. For the tutorial nodes, enable isaacsim.replicator.examples (for example Window > Extensions > ``isaacsim.replicator.examples`` > ENABLED). If you followed the generic OmniGraph tutorial instead, enable omni.new.extension under THIRD PARTY.
After the OmniGraph randomization nodes are created, they can be manually added to a pre-existing SDG pipeline graph. To create a basic SDG graph, the following snippet can be used in the Script Editor to randomize the rotations of the created cubes every frame.
Basic SDG Pipeline
import omni.replicator.core as rep
rep.functional.create.xform(name="World")
rep.functional.create_batch.cube(count=50, parent="/World", name="cube", scale=0.1)
cube = rep.get.prims(path_pattern="/World/cube_")
with rep.trigger.on_frame():
with cube:
rep.randomizer.rotation()
After the snippet is executed in the Script Editor, the generated graph can be opened at /Replicator/SDGPipeline and the custom nodes can be added to the graph. The following image shows the result after the custom nodes are added to the SDG pipeline graph together with the resulting randomization (from the UI using Tools > Replicator > Preview or Step):
To avoid manually adding the custom nodes to the SDG pipeline graph, the Replicator API can be used to automatically insert the nodes into the graph. For this purpose, the nodes need to be encapsulated as ReplicatorItems using the @ReplicatorWrapper decorator. The following snippet wires the same shape-to-randomizer mapping as the script above (spheres on a sphere, cubes in a sphere, cylinders between spheres):
ReplicatorWrapper
import asyncio
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 omni.replicator.core.scripts.utils import (
ReplicatorItem,
ReplicatorWrapper,
create_node,
set_target_prims,
)
PRIM_COUNT = 50
PRIM_SCALE = 0.1
RAD_IN = 0.5
RAD_ON = 1.5
RAD_BET1 = 2.5
RAD_BET2 = 3.5
NUM_STEPS = 10
@ReplicatorWrapper
def on_sphere(
radius: float = 1.0,
input_prims: ReplicatorItem | list[str] | None = None,
) -> ReplicatorItem:
node = create_node("isaacsim.replicator.examples.OgnSampleOnSphere", radius=radius)
if input_prims:
set_target_prims(node, "inputs:prims", input_prims)
return node
@ReplicatorWrapper
def in_sphere(
radius: float = 1.0,
input_prims: ReplicatorItem | list[str] | None = None,
) -> ReplicatorItem:
node = create_node("isaacsim.replicator.examples.OgnSampleInSphere", radius=radius)
if input_prims:
set_target_prims(node, "inputs:prims", input_prims)
return node
@ReplicatorWrapper
def between_spheres(
radius1: float = 0.5,
radius2: float = 1.0,
input_prims: ReplicatorItem | list[str] | None = None,
) -> ReplicatorItem:
node = create_node("isaacsim.replicator.examples.OgnSampleBetweenSpheres", radius1=radius1, radius2=radius2)
if input_prims:
set_target_prims(node, "inputs:prims", input_prims)
return node
async def run_example_async():
await stage_utils.create_new_stage_async()
app_utils.enable_extension("isaacsim.replicator.examples")
await app_utils.update_app_async()
rep.functional.create.xform(name="World")
rep.functional.create.dome_light(name="Light", parent="/World", intensity=500.0)
rep.functional.create_batch.sphere(count=PRIM_COUNT, parent="/World", name="sphere", scale=PRIM_SCALE)
rep.functional.create_batch.cube(count=PRIM_COUNT, parent="/World", name="cube", scale=PRIM_SCALE)
rep.functional.create_batch.cylinder(count=PRIM_COUNT, parent="/World", name="cylinder", scale=PRIM_SCALE)
sphere = rep.get.prims(path_pattern="/World/sphere_")
cube = rep.get.prims(path_pattern="/World/cube_")
cylinder = rep.get.prims(path_pattern="/World/cylinder_")
with rep.trigger.on_frame():
with sphere:
rep.randomizer.rotation()
on_sphere(RAD_ON, input_prims=sphere)
with cube:
rep.randomizer.rotation()
in_sphere(RAD_IN, input_prims=cube)
with cylinder:
rep.randomizer.rotation()
between_spheres(RAD_BET1, RAD_BET2, input_prims=cylinder)
rep.orchestrator.set_capture_on_play(False)
await rep.orchestrator.preview_async()
for _ in range(NUM_STEPS):
await rep.orchestrator.step_async()
asyncio.ensure_future(run_example_async())
Note
For this tutorial the create_node function uses "isaacsim.replicator.examples.OgnSampleInSphere" as the node path. Ensure the isaacsim.replicator.examples extension is enabled before running the snippet. Replace the node path if your custom nodes live in a different extension.
After the snippet is executed in the Script Editor, the custom nodes will be automatically added to the SDG pipeline graph. To trigger the randomization, Tools > Replicator > Preview (or Step) can be called from the UI. The following image shows the generated graph and the resulting randomization:
