# 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.

## Implementation

This tutorial will showcase how to create custom scene randomization Python scripts. These scripts will create prims in a new stage and randomize their rotation and locations: in a sphere, on a sphere, and between two spheres.

The following image shows the result after running the randomization in the Script Editor:

The following functions take as input the radius (or radii) of the sphere(s) and generate a random 3D point on the surface of a sphere, within a sphere, and between two spheres. These points will determine the prim locations.

Randomization Functions
```# Generate a random 3D point on the surface of a sphere of a given radius.
[..]
return x, y, z

# Generate a random 3D point within a sphere of a given radius, ensuring a uniform distribution throughout the volume.
[..]
return x, y, z

# Generate a random 3D point between two spheres, ensuring a uniform distribution throughout the volume.
[..]
return x, y, z
```

The following snippet creates prims in a new stage and randomizes their rotation and locations using the previously defined functions.

Spawning and Randomizing Prims
```# Create the default prims
on_sphere_prims = [stage.DefinePrim(f"/World/sphere_{i}", "Sphere") for i in range(prim_count)]
in_sphere_prims = [stage.DefinePrim(f"/World/cube_{i}", "Cube") for i in range(prim_count)]
between_spheres_prims = [stage.DefinePrim(f"/World/cylinder_{i}", "Cylinder") for i in range(prim_count)]

[..]

# Randomize the prims
for _ in range(10):
for in_sphere_prim in in_sphere_prims:
rand_rot = (random.uniform(0, 360), random.uniform(0, 360), random.uniform(0, 360))
in_sphere_prim.GetAttribute("xformOp:rotateXYZ").Set(rand_rot)
in_sphere_prim.GetAttribute("xformOp:translate").Set(rand_loc)

for on_sphere_prim in on_sphere_prims:
rand_rot = (random.uniform(0, 360), random.uniform(0, 360), random.uniform(0, 360))
on_sphere_prim.GetAttribute("xformOp:rotateXYZ").Set(rand_rot)
on_sphere_prim.GetAttribute("xformOp:translate").Set(rand_loc)

for between_spheres_prim in between_spheres_prims:
rand_rot = (random.uniform(0, 360), random.uniform(0, 360), random.uniform(0, 360))
between_spheres_prim.GetAttribute("xformOp:rotateXYZ").Set(rand_rot)
between_spheres_prim.GetAttribute("xformOp:translate").Set(rand_loc)
```

Snippet to run in the Script Editor:

Script Editor
```import math
import random
from itertools import chain

import omni.replicator.core as rep
import omni.usd
from pxr import UsdGeom

# Generate a random 3D point on the surface of a sphere of a given radius.
# Generate a random direction by spherical coordinates (phi, theta)
phi = random.uniform(0, 2 * math.pi)
# Sample costheta to ensure uniform distribution of points on the sphere (surface is proportional to sin(theta))
costheta = random.uniform(-1, 1)
theta = math.acos(costheta)

# Convert from spherical to Cartesian coordinates
x = radius * math.sin(theta) * math.cos(phi)
y = radius * math.sin(theta) * math.sin(phi)

return x, y, z

# Generate a random 3D point within a sphere of a given radius, ensuring a uniform distribution throughout the volume.
# Generate a random direction by spherical coordinates (phi, theta)
phi = random.uniform(0, 2 * math.pi)
# Sample costheta to ensure uniform distribution of points on the sphere (surface is proportional to sin(theta))
costheta = random.uniform(-1, 1)
theta = math.acos(costheta)

# Scale the radius uniformly within the sphere, applying the cube root to a random value
# to account for volume's cubic growth with radius (r^3), ensuring spatial uniformity.
r = radius * (random.random() ** (1 / 3))

# Convert from spherical to Cartesian coordinates
x = r * math.sin(theta) * math.cos(phi)
y = r * math.sin(theta) * math.sin(phi)
z = r * math.cos(theta)

return x, y, z

# Generate a random 3D point between two spheres, ensuring a uniform distribution throughout the volume.

# Generate a random direction by spherical coordinates (phi, theta)
phi = random.uniform(0, 2 * math.pi)
# Sample costheta to ensure uniform distribution of points on the sphere (surface is proportional to sin(theta))
costheta = random.uniform(-1, 1)
theta = math.acos(costheta)

# Uniformly distribute points between two spheres by weighting the radius to match volume growth (r^3),
# ensuring spatial uniformity by taking the cube root of a value between the radii cubed.

# Convert from spherical to Cartesian coordinates
x = r * math.sin(theta) * math.cos(phi)
y = r * math.sin(theta) * math.sin(phi)
z = r * math.cos(theta)

return x, y, z

stage = omni.usd.get_context().get_stage()
prim_count = 500
prim_scale = 0.1

# Create the default prims
on_sphere_prims = [stage.DefinePrim(f"/World/sphere_{i}", "Sphere") for i in range(prim_count)]
in_sphere_prims = [stage.DefinePrim(f"/World/cube_{i}", "Cube") for i in range(prim_count)]
between_spheres_prims = [stage.DefinePrim(f"/World/cylinder_{i}", "Cylinder") for i in range(prim_count)]

# Add xformOps and scale to the prims
for prim in chain(on_sphere_prims, in_sphere_prims, between_spheres_prims):
if not prim.HasAttribute("xformOp:translate"):
if not prim.HasAttribute("xformOp:scale"):
if not prim.HasAttribute("xformOp:rotateXYZ"):
prim.GetAttribute("xformOp:scale").Set((prim_scale, prim_scale, prim_scale))

# Randomize the prims
for _ in range(10):
for in_sphere_prim in in_sphere_prims:
rand_rot = (random.uniform(0, 360), random.uniform(0, 360), random.uniform(0, 360))
in_sphere_prim.GetAttribute("xformOp:rotateXYZ").Set(rand_rot)
in_sphere_prim.GetAttribute("xformOp:translate").Set(rand_loc)

for on_sphere_prim in on_sphere_prims:
rand_rot = (random.uniform(0, 360), random.uniform(0, 360), random.uniform(0, 360))
on_sphere_prim.GetAttribute("xformOp:rotateXYZ").Set(rand_rot)
on_sphere_prim.GetAttribute("xformOp:translate").Set(rand_loc)

for between_spheres_prim in between_spheres_prims:
rand_rot = (random.uniform(0, 360), random.uniform(0, 360), random.uniform(0, 360))
between_spheres_prim.GetAttribute("xformOp:rotateXYZ").Set(rand_rot)
between_spheres_prim.GetAttribute("xformOp:translate").Set(rand_loc)
```

As a next step, the Custom Node Tutorial is used to create custom OmniGraph nodes for the randomization functions. The node descriptions and implementations can be found in the following code snippets:

OgnSampleInSphere.ogn
```{
"OgnSampleInSphere": {
"version": 1,
"description": "Assignes uniformly sampled location in a sphere.",
"language": "Python",
"uiName": "Sample In Sphere"
},
"inputs": {
"prims": {
"type": "target",
"description": "prims to randomize",
"default": []
},
"execIn": {
"type": "execution",
"description": "exec",
"default": 0
},
"type": "float",
"default": 1.0
}
},
"outputs": {
"execOut": {
"type": "execution",
"description": "exec"
}
}
}
}
```
OgnSampleOnSphere.ogn
```{
"OgnSampleOnSphere": {
"version": 1,
"description": "Assignes uniformly sampled location on a sphere.",
"language": "Python",
"uiName": "Sample On Sphere"
},
"inputs": {
"prims": {
"type": "target",
"description": "prims to randomize",
"default": []
},
"execIn": {
"type": "execution",
"description": "exec",
"default": 0
},
"type": "float",
"default": 1.0
}
},
"outputs": {
"execOut": {
"type": "execution",
"description": "exec"
}
}
}
}
```
OgnSampleBetweenSpheres.ogn
```{
"OgnSampleBetweenSpheres": {
"version": 1,
"description": "Assignes uniformly sampled between two spheres",
"language": "Python",
"uiName": "Sample Between Spheres"
},
"inputs": {
"prims": {
"type": "target",
"description": "prims to randomize",
"default": []
},
"execIn": {
"type": "execution",
"description": "exec",
"default": 0
},
"type": "float",
"default": 0.5
},
"type": "float",
"default": 1.0
}
},
"outputs": {
"execOut": {
"type": "execution",
"description": "exec"
}
}
}
}
```
OgnSampleInSphere.py
```import numpy as np
import omni.graph.core as og
import omni.usd
from pxr import Sdf, UsdGeom

class OgnSampleInSphere:
@staticmethod
def compute(db) -> bool:
prim_paths = db.inputs.prims
if len(prim_paths) == 0:
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 UsdGeom.Xformable(prim):
prim_type = prim.GetTypeName()
raise ValueError(
f"Expected prim at {prim.GetPath()} to be an Xformable prim but got type {prim_type}"
)
if not prim.HasAttribute("xformOp:translate"):

except Exception as error:
db.log_error(str(error))
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False

samples = []
for _ in range(len(prims)):
# Generate a random direction by spherical coordinates (phi, theta)
phi = np.random.uniform(0, 2 * np.pi)
# Sample costheta to ensure uniform distribution of points on the sphere (surface is proportional to sin(theta))
costheta = np.random.uniform(-1, 1)
theta = np.arccos(costheta)

# Scale the radius uniformly within the sphere, applying the cube root to a random value
# to account for volume's cubic growth with radius (r^3), ensuring spatial uniformity.
r = radius * (np.random.random() ** (1 / 3))

# Convert from spherical to Cartesian coordinates
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)

samples.append((x, y, z))

with Sdf.ChangeBlock():
for prim, sample in zip(prims, samples):
prim.GetAttribute("xformOp:translate").Set(sample)

db.outputs.execOut = og.ExecutionAttributeState.ENABLED
return True
```
OgnSampleOnSphere.py
```import numpy as np
import omni.graph.core as og
import omni.usd
from pxr import Sdf, UsdGeom

class OgnSampleOnSphere:
@staticmethod
def compute(db) -> bool:
prim_paths = db.inputs.prims
if len(prim_paths) == 0:
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 UsdGeom.Xformable(prim):
prim_type = prim.GetTypeName()
raise ValueError(
f"Expected prim at {prim.GetPath()} to be an Xformable prim but got type {prim_type}"
)
if not prim.HasAttribute("xformOp:translate"):

except Exception as error:
db.log_error(str(error))
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False

samples = []
for _ in range(len(prims)):
# Generate a random direction by spherical coordinates (phi, theta)
phi = np.random.uniform(0, 2 * np.pi)
# Sample costheta to ensure uniform distribution of points on the sphere (surface is proportional to sin(theta))
costheta = np.random.uniform(-1, 1)
theta = np.arccos(costheta)

# Convert from spherical to Cartesian coordinates
x = radius * np.sin(theta) * np.cos(phi)
y = radius * np.sin(theta) * np.sin(phi)

samples.append((x, y, z))

with Sdf.ChangeBlock():
for prim, sample in zip(prims, samples):
prim.GetAttribute("xformOp:translate").Set(sample)

db.outputs.execOut = og.ExecutionAttributeState.ENABLED
return True
```
OgnSampleBetweenSpheres.py
```import numpy as np
import omni.graph.core as og
import omni.usd
from pxr import Sdf, UsdGeom

class OgnSampleBetweenSpheres:
@staticmethod
def compute(db) -> bool:
prim_paths = db.inputs.prims
if len(prim_paths) == 0:
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 UsdGeom.Xformable(prim):
prim_type = prim.GetTypeName()
raise ValueError(
f"Expected prim at {prim.GetPath()} to be an Xformable prim but got type {prim_type}"
)
if not prim.HasAttribute("xformOp:translate"):
raise ValueError(
)

except Exception as error:
db.log_error(str(error))
db.outputs.execOut = og.ExecutionAttributeState.DISABLED
return False

samples = []
for _ in range(len(prims)):
# Generate a random direction by spherical coordinates (phi, theta)
phi = np.random.uniform(0, 2 * np.pi)
# Sample costheta to ensure uniform distribution of points on the sphere (surface is proportional to sin(theta))
costheta = np.random.uniform(-1, 1)
theta = np.arccos(costheta)

# Uniformly distribute points between two spheres by weighting the radius to match volume growth (r^3),
# ensuring spatial uniformity by taking the cube root of a value between the radii cubed.

# Convert from spherical to Cartesian coordinates
x = r * np.sin(theta) * np.cos(phi)
y = r * np.sin(theta) * np.sin(phi)
z = r * np.cos(theta)

samples.append((x, y, z))

with Sdf.ChangeBlock():
for prim, sample in zip(prims, samples):
prim.GetAttribute("xformOp:translate").Set(sample)

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 already added to the built-in `omni.replicator.isaac` extension and are available by default. Other custom nodes created through the OmniGraph tutorial will be accessible via 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 newly created extension needs to be enabled. This can be done by navigating to Window –> Extensions –> THIRD PARTY –> `omni.new.extension` –> ENABLED:

Once the OmniGraph randomization nodes are created, they can be manually added to a pre-existing SDG pipeline graph. To create a simple 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

cube = rep.create.cube(count=50, scale=0.1)
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 `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 code snippet demonstrates how ReplicatorItems can be created for the custom nodes:

ReplicatorWrapper
```import omni.replicator.core as rep
from omni.replicator.core.scripts.utils import (
ReplicatorItem,
ReplicatorWrapper,
create_node,
set_target_prims,
)

@ReplicatorWrapper
def on_sphere(
input_prims: ReplicatorItem | list[str] | None = None,
) -> ReplicatorItem:

if input_prims:
set_target_prims(node, "inputs:prims", input_prims)
return node

@ReplicatorWrapper
def in_sphere(
input_prims: ReplicatorItem | list[str] | None = None,
) -> ReplicatorItem:

if input_prims:
set_target_prims(node, "inputs:prims", input_prims)
return node

@ReplicatorWrapper
def between_spheres(
input_prims: ReplicatorItem | list[str] | None = None,
) -> ReplicatorItem:

if input_prims:
set_target_prims(node, "inputs:prims", input_prims)
return node

prim_count = 50
prim_scale = 0.1

# Create the default prims
sphere = rep.create.sphere(count=prim_count, scale=prim_scale)
cube = rep.create.cube(count=prim_count, scale=prim_scale)
cylinder = rep.create.cylinder(count=prim_count, scale=prim_scale)

# Create the randomization graph
with rep.trigger.on_frame():
with sphere:
rep.randomizer.rotation()

with cube:
rep.randomizer.rotation()
For this tutorial the `create_node` function uses `"omni.replicator.isaac.OgnSampleInSphere"` as the node path, this path needs to be replaced in case the custom nodes are not part of the built-in `omni.replicator.isaac` 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, `Replicator` –> `Preview` (or `Step`) can be called from the UI. The following image shows the generated graph and the resulting randomization: