Modular Behavior Scripting#

Overview#

This tutorial introduces the isaacsim.replicator.behavior extension, providing multiple examples of modular behavior scripts in Isaac Sim Replicator for synthetic data generation (SDG). By utilizing Behavior Scripts, reusable, shareable, and easily modifiable behaviors can be developed and attached to prims in a USD stage, acting as randomizers or custom smart-asset behaviors.

The behavior script examples can be found under:

/exts/isaacsim.replicator.behavior/isaacsim/replicator/behavior/behaviors/*

The example section provides a demonstration of how to use the behavior scripts to create a custom synthetic data generation pipeline:

Behavior Scripts#

Behavior Scripts are modular Python scripts attached to prims in a USD stage. By default, they include template code that responds to timeline events such as start, pause, stop, and update. These scripts define specific behaviors or randomizations applied to prims during simulation or data generation. Attaching scripts directly to prims integrates the behaviors into the USD, making them modular because scripts can be easily attached, detached, or swapped on prims without altering core logic. They are sharable since behaviors can be embedded within assets and shared across different projects or stages. They are configurable as variables can be exposed through USD attributes for customization without modifying the script code. Additionally, they are persistent; since scripts reside on the prims, they persist with the USD stage and can be versioned and managed accordingly.

The advantages of behavior scripts include reusability, allowing them to be written once and reused across multiple prims or projects. They offer encapsulation by containing behavior logic within the prims, reducing external dependencies. They provide interactivity because parameters can be adjusted through the UI, enabling modifications without programming. Finally, they ensure integration by becoming an integral part of the asset, which maintains consistency across different environments.

Exposing Variables Through USD Attributes#

To enhance flexibility and accessibility, the input parameters in the provided behavior scripts examples can be exposed as USD attributes on prims. This approach allows users to modify behavior parameters directly from the UI without altering the script code.

The benefits of exposing variables include customization, interactivity, and consistency. Parameters such as target locations, ranges, or other settings can be adjusted per prim instance, using the UI to tweak behaviors and observe immediate effects, while maintaining a uniform interface for modifying behaviors across different scripts.

The exposed variables are implemented using the USD API to create custom attributes with appropriate namespaces on the prim. These attributes are then read by the behavior scripts during execution to adjust their logic accordingly.

The UI implementation for exposing the variables is done in isaacsim.replicator.behavior.ui. It extends the Property panel of the selected prims in the stage with a custom section for the exposed variables. The UI is automatically generated based on the exposed variables defined in the behavior script, displaying them as editable fields in the generated widget.

Example of Exposed Variables Definition:

VARIABLES_TO_EXPOSE = [
    {
        "attr_name": "targetLocation",
        "attr_type": Sdf.ValueTypeNames.Vector3d,
        "default_value": Gf.Vec3d(0.0, 0.0, 0.0),
        "doc": "The 3D vector specifying the location to look at.",
    },
    {
        "attr_name": "targetPrimPath",
        "attr_type": Sdf.ValueTypeNames.String,
        "default_value": "",
        "doc": "The path of the target prim to look at. If specified, it has priority over the target location.",
    },
    # Additional variables...
]

Custom Event-Based Behavior Scripts#

While behavior scripts are timeline-based by default, some behaviors need to operate independently of the simulation timeline. Event-based scripting allows behaviors to be triggered by custom events, providing greater control over when and how they execute. This is achieved by skipping the default behavior functions and instead listening to and publishing custom events.

Custom events are defined and managed within Omniverse using an event bus system, enabling scripts to publish or subscribe to these events and facilitating communication between different components or behaviors.

Event-based scripting offers flexibility by allowing customization of when behaviors are executed, independent of the simulation timeline. It enhances modularity by decoupling behaviors from the core simulation loop, making them more modular. Additionally, it improves scalability by managing complex workflows through orchestrating multiple behaviors via events.

For example, the volume_stack_randomizer.py script randomizes the stacking of objects by simulating physics before the simulation starts. By using custom events, behaviors can be triggered before the simulation, execution flow can be controlled by starting, stopping, or resetting behaviors based on specific events rather than timeline updates, and performance can be enhanced by avoiding unnecessary computations during each simulation frame through decoupling certain behaviors.

Script Examples#

In this section, various behavior scripts available in the isaacsim.replicator.behavior extension are explored. Each script provides specific functionality that can enhance synthetic data generation workflows. The scripts are designed to be modular, reusable, and customizable through exposed variables.

The folder path for the behavior scripts is:

/exts/isaacsim.replicator.behavior/isaacsim/replicator/behavior/behaviors/*

Location Randomizer#

The location_randomizer.py script randomizes the location of prims within specified bounds during runtime. This behavior can be applied to multiple prims simultaneously and is customizable through exposed variables.

Key Features:

Position Range: Randomizes prim positions within defined minimum and maximum bounds.
Relative Positioning: Supports positioning relative to a target prim.
Child Inclusion: Option to include child prims in the randomization.
Update Interval: Controls the frequency of randomization (e.g., every frame, every N frames).

Exposed Variables:

The script exposes several variables for user customization:

range:minPosition (Vector3d): Minimum position for randomization.
range:maxPosition (Vector3d): Maximum position for randomization.
frame:useRelativeFrame (Bool): Whether to use a relative frame for randomization.
frame:targetPrimPath (String): Path to a target prim for relative randomization.
includeChildren (Bool): Whether to include child prims in the behavior.
interval (UInt): Interval for updating the behavior; 0 means every frame.

Including Children Prims:

The inclusion of child prims is controlled by the includeChildren variable. When set to True, the script recursively includes all child prims of the assigned prim. The relevant code snippet is:

def _setup(self):
    # Fetch the exposed variables
    include_children = self._get_exposed_variable("includeChildren")

    # Get the prims to apply the behavior to
    if include_children:
        self._valid_prims = [
            prim for prim in Usd.PrimRange(self.prim) if prim.IsA(UsdGeom.Xformable)
        ]
    elif self.prim.IsA(UsdGeom.Xformable):
        self._valid_prims = [self.prim]
    else:
        self._valid_prims = []
        carb.log_warn(f"[{self.prim_path}] No valid prims found.")

In this snippet:

If includeChildren is True, the script uses Usd.PrimRange to iterate over all descendant prims of the assigned prim and selects those that are transformable (UsdGeom.Xformable).
If includeChildren is False, it checks if the assigned prim itself is transformable and includes it if so.
If no valid prims are found, it logs a warning.

Randomizing Prim Locations:

The core functionality of randomizing locations is implemented in the _randomize_location method:

def _randomize_location(self, prim):
    # Generate a random offset within the bounds
    random_offset = Gf.Vec3d(
        random.uniform(self._min_position[0], self._max_position[0]),
        random.uniform(self._min_position[1], self._max_position[1]),
        random.uniform(self._min_position[2], self._max_position[2]),
    )

    # Initialize location
    loc = random_offset

    # Handle the target prim if specified
    if self._target_prim:
        target_loc = get_world_location(self._target_prim)

        if self._use_relative_frame:
            # Maintain the offset from the target prim
            loc = target_loc + self._target_offsets[prim] + random_offset
        else:
            # Move the prim to the randomized location relative to the target prim
            loc = target_loc + random_offset
    else:
        if self._use_relative_frame:
            # Add the initial location if using the relative frame
            loc += self._initial_locations[prim]

    # Set the randomized location to the prim
    self._set_location(prim, loc)

In this snippet:

A random offset within the specified bounds is generated.
If a target prim is specified, the new location is calculated relative to it.
If using a relative frame, the initial location is considered in the calculation.
The new location is then applied to the prim using _set_location.

Usage Summary:

Initialization: Exposed variables are created as USD attributes on the prim during on_init, allowing users to adjust them via the UI.
Setup: In on_play, the script fetches the exposed variables and prepares the list of prims to randomize.
Randomization: The _apply_behavior method randomizes the positions of the prims based on the specified parameters.
Reset: In on_stop, the script resets the prims to their initial positions.

Example Usage:

To randomize the positions of objects within a certain area every 5 frames:

Attach location_randomizer.py to the parent prim.
Set range:minPosition and range:maxPosition to define the area bounds.
Enable includeChildren to include all child prims.
Set interval to 5 for updates every 5 frames.

By incorporating the location_randomizer.py script, you can introduce positional variability to objects in your scene, enhancing the diversity of your synthetic datasets.

Rotation Randomizer#

The rotation_randomizer.py script applies random rotations to prims during runtime. It can be applied to multiple prims simultaneously and is customizable through exposed variables.

Key Features:

Rotation Range: Randomizes prim rotations within specified minimum and maximum Euler angle bounds.
Child Inclusion: Option to include child prims in the randomization.
Update Interval: Controls how frequently the randomization occurs.

Exposed Variables:

range:minRotation (Vector3d): Minimum rotation angles (in degrees) for randomization.
range:maxRotation (Vector3d): Maximum rotation angles (in degrees) for randomization.
includeChildren (Bool): Whether to include child prims in the behavior.
interval (UInt): Interval for updating the behavior; 0 means every frame.

Including Children Prims:

The inclusion of child prims is managed by the includeChildren variable. When set to True, the script includes all descendant prims that are transformable. The relevant code snippet is:

def _setup(self):
    include_children = self._get_exposed_variable("includeChildren")

    if include_children:
        self._valid_prims = [
            prim for prim in Usd.PrimRange(self.prim) if prim.IsA(UsdGeom.Xformable)
        ]
    elif self.prim.IsA(UsdGeom.Xformable):
        self._valid_prims = [self.prim]
    else:
        self._valid_prims = []
        carb.log_warn(f"[{self.prim_path}] No valid prims found.")

In this snippet:

If includeChildren is True, all transformable descendant prims are included.
If False, only the assigned prim is considered if it is transformable.
If no valid prims are found, a warning is logged.

Randomizing Prim Rotations:

The core functionality of applying random rotations is implemented in the _randomize_rotation method:

def _randomize_rotation(self, prim):
    rotation = (
        Gf.Rotation(Gf.Vec3d.XAxis(), random.uniform(self._min_rotation[0], self._max_rotation[0]))
        * Gf.Rotation(Gf.Vec3d.YAxis(), random.uniform(self._min_rotation[1], self._max_rotation[1]))
        * Gf.Rotation(Gf.Vec3d.ZAxis(), random.uniform(self._min_rotation[2], self._max_rotation[2]))
    )
    set_rotation_with_ops(prim, rotation)

In this snippet:

Random Euler angles are generated within the specified bounds for each axis.
A composite rotation is created by multiplying rotations around the X, Y, and Z axes.
The new rotation is applied to the prim using set_rotation_with_ops, which handles various transformation operations.

Usage Summary:

Initialization: Exposed variables are created as USD attributes on the prim during on_init, allowing users to adjust them via the UI.
Setup: In on_play, the script fetches the exposed variables and prepares the list of prims to randomize.
Randomization: The _apply_behavior method randomizes the rotations of the prims based on the specified parameters.
Reset: In on_stop, the script resets the prims to their initial rotations.

Example Usage:

To apply random rotations to objects every 10 frames:

Attach rotation_randomizer.py to the parent prim.
Set range:minRotation and range:maxRotation to define the rotation bounds.
Enable includeChildren to include all child prims.
Set interval to 10 for updates every 10 frames.

Look At Behavior#

The look_at_behavior.py script orients prims to continuously face a specified target location or another prim. This is particularly useful for making cameras or sensors track moving objects.

Key Features:

Target Specification: Can look at a fixed location or dynamically track another prim.
Up Axis Control: Allows setting the up axis to maintain orientation consistency.
Child Inclusion: Option to include child prims in the behavior.
Update Interval: Controls how frequently the orientation is updated.

Exposed Variables:

targetLocation (Vector3d): The 3D coordinates to look at.
targetPrimPath (String): Path to the target prim; takes precedence over targetLocation if specified.
upAxis (Vector3d): The up axis for orientation (e.g., (0, 0, 1) for +Z axis).
includeChildren (Bool): Whether to include child prims in the behavior.
interval (UInt): Interval for updating the behavior; 0 means every frame.

Target Prim Usage:

The script can target another prim for the look-at behavior. When targetPrimPath is provided, it overrides targetLocation. Here’s how the target prim is handled:

def _setup(self):
    target_prim_path = self._get_exposed_variable("targetPrimPath")
    if target_prim_path:
        self._target_prim = self.stage.GetPrimAtPath(target_prim_path)
        if not self._target_prim or not self._target_prim.IsValid() or not self._target_prim.IsA(UsdGeom.Xformable):
            self._target_prim = None
            carb.log_warn(f"[{self.prim_path}] Invalid target prim path: {target_prim_path}")

Orientation Calculation:

The core functionality calculates the rotation needed for the prim to look at the target:

def _apply_behavior(self):
    target_location = self._get_target_location()

    for prim in self._valid_prims:
        eye = get_world_location(prim)
        look_at_rotation = calculate_look_at_rotation(eye, target_location, self._up_axis)
        set_rotation_with_ops(prim, look_at_rotation)

In this snippet:

get_world_location(prim) retrieves the current position of the prim.
calculate_look_at_rotation computes the necessary rotation to face the target.
set_rotation_with_ops applies the rotation using existing transformation operations.

Usage Summary:

Initialization: Exposed variables are created during on_init.
Setup: In on_play, the script prepares the list of prims and determines the target.
Orientation Update: In on_update, the prims are oriented towards the target based on the specified interval.
Reset: In on_stop, the script resets the prims to their initial rotations.

Example Usage:

To make a camera look at a moving object:

Attach look_at_behavior.py to the camera prim.
Set targetPrimPath to the path of the moving object.
Adjust upAxis if necessary to maintain the desired orientation.
Set interval to control the update frequency.

Light Randomizer#

The light_randomizer.py script randomizes properties of light prims, such as color and intensity, enhancing scene variability.

Key Features:

Color Randomization: Varies the RGB values of light sources within specified ranges.
Intensity Randomization: Adjusts light intensity between defined minimum and maximum values.
Child Inclusion: Option to include child light prims in the behavior.
Update Interval: Controls how frequently the properties are randomized.

Exposed Variables:

includeChildren (Bool): Whether to include child light prims.
interval (UInt): Interval for updating the behavior; 0 means every frame.
range:minColor (Color3f): Minimum RGB values for color randomization.
range:maxColor (Color3f): Maximum RGB values for color randomization.
range:intensity (Float2): Range for intensity randomization as (min, max).

Randomizing Light Properties:

The script applies random values to the color and intensity of light prims:

def _apply_behavior(self):
    for prim in self._valid_prims:
        rand_color = (
            random.uniform(self._min_color[0], self._max_color[0]),
            random.uniform(self._min_color[1], self._max_color[1]),
            random.uniform(self._min_color[2], self._max_color[2]),
        )
        prim.GetAttribute("inputs:color").Set(rand_color)

        rand_intensity = random.uniform(self._intensity_range[0], self._intensity_range[1])
        prim.GetAttribute("inputs:intensity").Set(rand_intensity)

In this snippet:

Random RGB values are generated within the specified color range.
Random intensity values are generated within the specified intensity range.
The new values are applied to the corresponding attributes of the light prims.

Including Child Lights:

Child light prims can be included based on the includeChildren variable:

def _setup(self):
    include_children = self._get_exposed_variable("includeChildren")
    if include_children:
        self._valid_prims = [prim for prim in Usd.PrimRange(self.prim) if prim.HasAPI(UsdLux.LightAPI)]
    elif self.prim.HasAPI(UsdLux.LightAPI):
        self._valid_prims = [self.prim]
    else:
        self._valid_prims = []
        carb.log_warn(f"[{self.prim_path}] No valid light prims found.")

Usage Summary:

Initialization: Exposed variables are set up in on_init.
Setup: In on_play, the script identifies valid light prims and caches original attributes.
Randomization: In on_update, color and intensity are randomized based on the specified interval.
Reset: In on_stop, the script restores original light properties.

Example Usage:

To randomize light properties every frame:

Attach light_randomizer.py to a light prim or a parent prim containing lights.
Set range:minColor and range:maxColor to define the color range.
Set range:intensity to define the intensity range.
Enable includeChildren if you want to include child lights.

Texture Randomizer#

The texture_randomizer.py script randomly applies textures to visual prims, enhancing material diversity.

Key Features:

Texture Selection: Randomly selects textures from provided assets or CSV list.
Material Creation: Generates materials with randomized parameters like scale and rotation.
Child Inclusion: Option to include child prims in the behavior.
Update Interval: Controls how often textures are randomized.

Exposed Variables:

includeChildren (Bool): Whether to include child prims.
interval (UInt): Interval for updating the behavior; 0 means every frame.
textures:assets (AssetArray): List of texture assets to use.
textures:csv (String): CSV string of texture URLs.
projectUvwProbability (Float): Probability of setting project_uvw to True.
textureScaleRange (Float2): Texture scale range as (min, max).
textureRotateRange (Float2): Texture rotation range in degrees as (min, max).

Randomizing Textures:

The script applies random textures and parameters to materials bound to prims:

def _apply_behavior(self):
    for mat in self._texture_materials:
        shader = UsdShade.Shader(omni.usd.get_shader_from_material(mat.GetPrim(), get_prim=True))
        diffuse_texture = random.choice(self._texture_urls)
        shader.GetInput("diffuse_texture").Set(diffuse_texture)
        project_uvw = random.choices(
            [True, False],
            weights=[self._project_uvw_probability, 1 - self._project_uvw_probability],
            k=1,
        )[0]
        shader.GetInput("project_uvw").Set(bool(project_uvw))
        texture_scale = random.uniform(self._texture_scale_range[0], self._texture_scale_range[1])
        shader.GetInput("texture_scale").Set((texture_scale, texture_scale))
        texture_rotate = random.uniform(self._texture_rotate_range[0], self._texture_rotate_range[1])
        shader.GetInput("texture_rotate").Set(texture_rotate)

In this snippet:

A random texture is selected from the provided list.
Material parameters like project_uvw, texture_scale, and texture_rotate are randomized.
The shader inputs are updated with the new values.

Including Child Prims:

The script includes child prims based on the includeChildren variable:

def _setup(self):
    include_children = self._get_exposed_variable("includeChildren")
    if include_children:
        self._valid_prims = [prim for prim in Usd.PrimRange(self.prim) if prim.IsA(UsdGeom.Gprim)]
    elif self.prim.IsA(UsdGeom.Gprim):
        self._valid_prims = [self.prim]
    else:
        self._valid_prims = []
        carb.log_warn(f"[{self.prim_path}] No valid prims found.")

Usage Summary:

Initialization: Exposed variables are set up in on_init.
Setup: In on_play, the script prepares the list of textures and creates materials.
Randomization: In on_update, textures and material parameters are randomized.
Reset: In on_stop, the script restores original materials and cleans up.

Example Usage:

To randomize textures on objects:

Attach texture_randomizer.py to a prim.
Provide texture assets via textures:assets or textures:csv.
Adjust parameters like projectUvwProbability, textureScaleRange, and textureRotateRange.
Enable includeChildren to apply to child prims.

Volume Stack Randomizer#

The volume_stack_randomizer.py script randomly drops and stacks assets within the area of specified prims. It leverages physics simulation to create realistic stacking of objects, enhancing scene variability for synthetic data generation.

Key Features:

Asset Randomization: Selects and spawns assets from provided lists or CSV paths.
Physics Simulation: Utilizes physics to naturally stack objects.
Event-Based Execution: Operates independently of the simulation timeline using custom events.
Customizable Parameters: Exposed variables allow customization of behavior, such as number of assets, drop height, and simulation rendering.

Exposed Variables:

includeChildren (Bool): Whether to include child prims in the behavior.
event:input (String): Event name to subscribe to for controlling the behavior.
event:output (String): Event name to publish to after executing behavior functions.
assets:assets (AssetArray): List of asset references to spawn.
assets:csv (String): CSV string of asset URLs to spawn.
assets:numRange (Int2): Range for the number of assets to spawn (min, max).
dropHeight (Float): Height from which to drop the assets.
renderSimulation (Bool): Whether to render the simulation steps when stacking.
removeRigidBodyDynamics (Bool): Whether to remove rigid body dynamics after simulation.
preserveSimulationState (Bool): Whether to keep the final simulation state after simulation.

Core Structure:

The script defines the VolumeStackRandomizer class inheriting from BehaviorScript:

class VolumeStackRandomizer(BehaviorScript):
    BEHAVIOR_NS = "volumeStackRandomizer"
    EVENT_NAME_IN = f"{EXTENSION_NAME}.{BEHAVIOR_NS}.in"
    EVENT_NAME_OUT = f"{EXTENSION_NAME}.{BEHAVIOR_NS}.out"
    ACTION_FUNCTION_MAP = {
        "setup": "_setup_async",
        "run": "_run_behavior_async",
        "reset": "_reset_async",
    }
    VARIABLES_TO_EXPOSE = [
        # Exposed variables as defined above...
    ]

    def on_init(self):
        # Initialization logic...
        pass

    async def _setup_async(self):
        # Asynchronous setup logic...
        pass

    async def _run_behavior_async(self):
        # Asynchronous behavior execution...
        pass

    async def _reset_async(self):
        # Asynchronous reset logic...
        pass

    def _on_event(self, event: carb.events.IEvent):
        # Event handling logic...
        pass

    # Additional helper methods...

Including Child Prims:

The inclusion of child prims is controlled by the includeChildren variable. When set to True, all descendant prims that are geometrically valid are included in the behavior.

def _setup_async(self):
    include_children = self._get_exposed_variable("includeChildren")
    if include_children:
        self._valid_prims = [prim for prim in Usd.PrimRange(self.prim) if prim.IsA(UsdGeom.Gprim)]
    elif self.prim.IsA(UsdGeom.Gprim):
        self._valid_prims = [self.prim]
    else:
        self._valid_prims = []
        carb.log_warn(f"[{self.prim_path}] No valid prims found.")

Control with Custom Events

Below is an example of how to start the Volume Stack Randomizer using custom events. The events can control all the behavior scripts at once or individually for each prim. The script can be run from the Script Editor:

Volume Stack Randomizer control using custom events

import asyncio
import inspect

import omni.kit.commands
import omni.usd
from isaacsim.replicator.behavior.behaviors import VolumeStackRandomizer
from isaacsim.replicator.behavior.global_variables import EXPOSED_ATTR_NS
from isaacsim.replicator.behavior.utils.behavior_utils import (
    add_behavior_script_with_parameters_async,
    publish_event_and_wait_for_completion_async,
)
from pxr import Gf, Sdf, UsdGeom

NESTED_PLANES_PATH = "/World/NestedPlanes"
SINGLE_PLANE_PATH = "/World/my_plane"


async def setup_scenario_async():
    # Create a new stage with a distant light
    omni.usd.get_context().new_stage()
    stage = omni.usd.get_context().get_stage()
    distant_light = stage.DefinePrim("/Lights/DistantLight", "DistantLight")
    if not distant_light.GetAttribute("xformOp:rotateXYZ"):
        UsdGeom.Xformable(distant_light).AddRotateXYZOp()
    distant_light.GetAttribute("xformOp:rotateXYZ").Set((0, 45, 90))
    distant_light.CreateAttribute("inputs:intensity", Sdf.ValueTypeNames.Float).Set(4000)

    # Create single and nested planes
    omni.kit.commands.execute("CreateMeshPrimWithDefaultXform", prim_path=SINGLE_PLANE_PATH, prim_type="Plane")
    single_plane_prim = stage.GetPrimAtPath(SINGLE_PLANE_PATH)
    single_plane_prim.GetAttribute("xformOp:scale").Set(Gf.Vec3d(1, 2, 1))

    nested_plane_xform = stage.DefinePrim(NESTED_PLANES_PATH, "Xform")
    for i in range(3):
        nested_plane_path = f"{NESTED_PLANES_PATH}/my_nested_plane_{i}"
        omni.kit.commands.execute("CreateMeshPrimWithDefaultXform", prim_path=nested_plane_path, prim_type="Plane")
        nested_plane_prim = stage.GetPrimAtPath(nested_plane_path)
        nested_plane_prim.GetAttribute("xformOp:translate").Set(Gf.Vec3d(2, 1.5 * i, 0))

    # Add behavior scripts
    script_path = inspect.getfile(VolumeStackRandomizer)
    parameters = {
        f"{EXPOSED_ATTR_NS}:{VolumeStackRandomizer.BEHAVIOR_NS}:assets:csv": (
            "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxC_01.usd,"
            "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxD_01.usd,"
            "/Isaac/Props/KLT_Bin/small_KLT_visual.usd,"
        ),
        f"{EXPOSED_ATTR_NS}:{VolumeStackRandomizer.BEHAVIOR_NS}:assets:numRange": Gf.Vec2i(2, 15),
    }
    await add_behavior_script_with_parameters_async(single_plane_prim, script_path, parameters)
    await add_behavior_script_with_parameters_async(nested_plane_xform, script_path, parameters)


async def run_stacking_simulation_async(prim_path=None):
    # Helper function to handle publishing and waiting for events
    async def wait_for_event_with_action(action, expected_state, max_wait, prim_path=None):
        publish_payload = {"prim_path": prim_path, "action": action} if prim_path else {"action": action}
        expected_payload = (
            {"prim_path": prim_path, "state_name": expected_state} if prim_path else {"state_name": expected_state}
        )
        return await publish_event_and_wait_for_completion_async(
            publish_payload=publish_payload,
            expected_payload=expected_payload,
            publish_event_name=VolumeStackRandomizer.EVENT_NAME_IN,
            subscribe_event_name=VolumeStackRandomizer.EVENT_NAME_OUT,
            max_wait_updates=max_wait,
        )

    # Define and execute the stacking simulation steps
    actions = [("reset", "RESET", 10), ("setup", "SETUP", 500), ("run", "FINISHED", 1500)]
    for action, state, wait in actions:
        print(f"Executing '{action}' and waiting for state '{state}'...")
        if not await wait_for_event_with_action(action, state, wait, prim_path=prim_path):
            print(f"Failed to complete '{action}' with state '{state}'.")
            return


async def run_example_async():
    print(f"Running example in series, stacking '{SINGLE_PLANE_PATH}' first, then '{NESTED_PLANES_PATH}'..")
    await setup_scenario_async()
    await run_stacking_simulation_async(SINGLE_PLANE_PATH)
    await run_stacking_simulation_async(NESTED_PLANES_PATH)

    print(f"Running examples at once, stacking '{SINGLE_PLANE_PATH}' and '{NESTED_PLANES_PATH}' concurrently..")
    await setup_scenario_async()
    await run_stacking_simulation_async()


asyncio.ensure_future(run_example_async())

Summary:

Initialization: The script initializes by exposing variables and subscribing to custom events.
Setup Phase: Prepares the simulation environment by spawning assets and setting up physics materials.
Run Phase: Executes the simulation, dropping and stacking assets with physics-based interactions.
Reset Phase: Cleans up the simulation environment, removing assets and resetting states.

Key Takeaways:

Event-Based Control: The script uses custom events to manage its lifecycle, allowing external control without querying the scene.
Physics Integration: Utilizes physics simulation for realistic asset stacking, enhancing the diversity of generated scenes.
Customization: Exposed variables provide flexibility to tailor the behavior to specific needs, such as asset selection and simulation parameters.

Templates#

This section provides template scripts that serve as starting points for creating new behaviors. These examples illustrate how to structure behavior scripts, expose variables, and handle events.

Available Templates:

example_behavior.py: A basic template providing boilerplate code for creating new behaviors.
base_behavior.py and example_base_behavior.py: Demonstrate how to build upon a base behavior class for structured development.
example_custom_event_behavior.py: Shows how to implement behaviors using custom events.

Key Aspects of the Templates:

Variable Exposure: All templates demonstrate how to expose variables as USD attributes for customization via the UI.
Behavior Structure: Provide necessary methods (on_init, on_play, on_update, on_stop, on_destroy) to integrate with the simulation timeline.
Extensibility: Base behavior classes allow for easy extension and reuse in new behaviors.

Example#

Below is an example demonstrating the use of behavior scripts to set up and run synthetic data generation in Isaac Sim. It showcases how to utilize behavior scripts for stacking simulations, texture randomization, light behavior, and camera tracking, ultimately capturing synthetic data with randomized scene configurations.

Key Highlights of the Example:

Volume Stacking Simulation: Randomly stack assets using physics simulation to create realistic arrangements.
Texture Randomization: Apply randomized textures to assets for scene diversity.
Light and Camera Behaviors: Add randomization to light properties and make the camera track a specific target.
Synthetic Data Capture: Generate and save synthetic images with the configured behaviors.

Example Script:

The demo script can be run directly from the Script Editor:

Behavior script-based SDG

import asyncio
import inspect
import os
import random

import omni.kit.app
import omni.replicator.core as rep
import omni.timeline
import omni.usd
from isaacsim.core.utils.semantics import add_update_semantics, remove_all_semantics
from isaacsim.replicator.behavior.behaviors import (
    LightRandomizer,
    LocationRandomizer,
    LookAtBehavior,
    RotationRandomizer,
    TextureRandomizer,
    VolumeStackRandomizer,
)
from isaacsim.replicator.behavior.global_variables import EXPOSED_ATTR_NS
from isaacsim.replicator.behavior.utils.behavior_utils import (
    add_behavior_script_with_parameters_async,
    publish_event_and_wait_for_completion_async,
)
from isaacsim.storage.native import get_assets_root_path_async
from pxr import Gf, UsdGeom


async def setup_and_run_stacking_simulation_async(prim):
    STACK_ASSETS_CSV = (
        "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxC_01.usd,"
        "/Isaac/Environments/Simple_Warehouse/Props/SM_CardBoxD_01.usd,"
        "/Isaac/Props/KLT_Bin/small_KLT_visual.usd,"
    )

    # Add the behavior script with custom parameters
    script_path = inspect.getfile(VolumeStackRandomizer)
    parameters = {
        f"{EXPOSED_ATTR_NS}:{VolumeStackRandomizer.BEHAVIOR_NS}:assets:csv": STACK_ASSETS_CSV,
        f"{EXPOSED_ATTR_NS}:{VolumeStackRandomizer.BEHAVIOR_NS}:assets:numRange": Gf.Vec2i(2, 15),
    }
    await add_behavior_script_with_parameters_async(prim, script_path, parameters)

    # Helper function to handle publishing and waiting for events
    async def handle_event(action, expected_state, max_wait):
        return await publish_event_and_wait_for_completion_async(
            publish_payload={"prim_path": prim.GetPath(), "action": action},
            expected_payload={"prim_path": prim.GetPath(), "state_name": expected_state},
            publish_event_name=VolumeStackRandomizer.EVENT_NAME_IN,
            subscribe_event_name=VolumeStackRandomizer.EVENT_NAME_OUT,
            max_wait_updates=max_wait,
        )

    # Define and execute the stacking simulation steps
    actions = [("reset", "RESET", 10), ("setup", "SETUP", 500), ("run", "FINISHED", 1500)]
    for action, state, wait in actions:
        print(f"Executing '{action}' and waiting for state '{state}'...")
        if not await handle_event(action, state, wait):
            print(f"Failed to complete '{action}' with state '{state}'.")
            return

    print("Stacking simulation finished.")


async def setup_texture_randomizer_async(prim):
    TEXTURE_ASSETS_CSV = (
        "/Isaac/Materials/Textures/Patterns/nv_bamboo_desktop.jpg,"
        "/Isaac/Materials/Textures/Patterns/nv_wood_boards_brown.jpg,"
        "/Isaac/Materials/Textures/Patterns/nv_wooden_wall.jpg,"
    )

    script_path = inspect.getfile(TextureRandomizer)
    parameters = {
        f"{EXPOSED_ATTR_NS}:{TextureRandomizer.BEHAVIOR_NS}:interval": 5,
        f"{EXPOSED_ATTR_NS}:{TextureRandomizer.BEHAVIOR_NS}:textures:csv": TEXTURE_ASSETS_CSV,
    }
    await add_behavior_script_with_parameters_async(prim, script_path, parameters)


async def setup_light_behaviors_async(prim):
    # Light randomization
    light_script_path = inspect.getfile(LightRandomizer)
    light_parameters = {
        f"{EXPOSED_ATTR_NS}:{LightRandomizer.BEHAVIOR_NS}:interval": 4,
        f"{EXPOSED_ATTR_NS}:{LightRandomizer.BEHAVIOR_NS}:range:intensity": Gf.Vec2f(20000, 120000),
    }
    await add_behavior_script_with_parameters_async(prim, light_script_path, light_parameters)

    # Location randomization
    location_script_path = inspect.getfile(LocationRandomizer)
    location_parameters = {
        f"{EXPOSED_ATTR_NS}:{LocationRandomizer.BEHAVIOR_NS}:interval": 2,
        f"{EXPOSED_ATTR_NS}:{LocationRandomizer.BEHAVIOR_NS}:range:minPosition": Gf.Vec3d(-1.25, -1.25, 0.0),
        f"{EXPOSED_ATTR_NS}:{LocationRandomizer.BEHAVIOR_NS}:range:maxPosition": Gf.Vec3d(1.25, 1.25, 0.0),
    }
    await add_behavior_script_with_parameters_async(prim, location_script_path, location_parameters)


async def setup_target_asset_behaviors_async(prim):
    # Rotation randomization with default parameters
    rotation_script_path = inspect.getfile(RotationRandomizer)
    await add_behavior_script_with_parameters_async(prim, rotation_script_path, {})

    # Location randomization
    location_script_path = inspect.getfile(LocationRandomizer)
    location_parameters = {
        f"{EXPOSED_ATTR_NS}:{LocationRandomizer.BEHAVIOR_NS}:interval": 3,
        f"{EXPOSED_ATTR_NS}:{LocationRandomizer.BEHAVIOR_NS}:range:minPosition": Gf.Vec3d(-0.2, -0.2, -0.2),
        f"{EXPOSED_ATTR_NS}:{LocationRandomizer.BEHAVIOR_NS}:range:maxPosition": Gf.Vec3d(0.2, 0.2, 0.2),
    }
    await add_behavior_script_with_parameters_async(prim, location_script_path, location_parameters)


async def setup_camera_behaviors_async(prim, target_prim_path):
    # Look at behavior following the target asset
    script_path = inspect.getfile(LookAtBehavior)
    parameters = {
        f"{EXPOSED_ATTR_NS}:{LookAtBehavior.BEHAVIOR_NS}:targetPrimPath": target_prim_path,
    }
    await add_behavior_script_with_parameters_async(prim, script_path, parameters)


async def setup_writer_and_capture_data_async(camera_path, num_captures):
    # Create the writer and the render product
    rp = rep.create.render_product(camera_path, (512, 512))
    writer = rep.writers.get("BasicWriter")
    output_directory = os.getcwd() + "/out_behaviors_sdg"
    print(f"output_directory: {output_directory}")
    writer.initialize(output_dir=output_directory, rgb=True)
    writer.attach(rp)

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

    # Start the timeline for the behavior scripts to run
    timeline = omni.timeline.get_timeline_interface()
    timeline.play()
    await omni.kit.app.get_app().next_update_async()

    # Capture frames
    for i in range(num_captures):
        # Advance the app (including the timeline)
        await omni.kit.app.get_app().next_update_async()

        # Capture and write frame
        print(f"Capturing frame {i} at time {timeline.get_current_time():.4f}")
        await rep.orchestrator.step_async(rt_subframes=32, delta_time=0.0, pause_timeline=False)

    # Stop the timeline (and the behavior scripts triggering)
    timeline.stop()

    # Free the renderer resources
    writer.detach()
    rp.destroy()

    # Make sure all the frames are written from the backend queue
    await rep.orchestrator.wait_until_complete_async()


async def run_example_async():
    STAGE_URL = "/Isaac/Samples/Replicator/Stage/warehouse_pallets_behavior_scripts.usd"
    PALLETS_ROOT_PATH = "/Root/Pallets"
    LIGHTS_ROOT_PATH = "/Root/Lights"
    CAMERA_PATH = "/Root/Camera_01"
    TARGET_ASSET_URL = "/Isaac/Props/YCB/Axis_Aligned/035_power_drill.usd"
    TARGET_ASSET_PATH = "/Root/Target"
    TARGET_ASSET_LABEL = "power_drill"
    TARGET_ASSET_LOCATION = (-1.5, 5.5, 1.5)

    # Open stage
    assets_root_path = await get_assets_root_path_async()
    print(f"Opening stage from {assets_root_path + STAGE_URL}")
    await omni.usd.get_context().open_stage_async(assets_root_path + STAGE_URL)
    stage = omni.usd.get_context().get_stage()

    # Check if all required prims exist in the stage
    pallets_root_prim = stage.GetPrimAtPath(PALLETS_ROOT_PATH)
    lights_root_prim = stage.GetPrimAtPath(LIGHTS_ROOT_PATH)
    camera_prim = stage.GetPrimAtPath(CAMERA_PATH)
    if not all([pallets_root_prim.IsValid(), lights_root_prim.IsValid(), camera_prim.IsValid()]):
        print(f"Not all required prims exist in the stage.")
        return

    # Spawn the target asset at the requested location, label it with the target asset label
    target_prim = stage.DefinePrim(TARGET_ASSET_PATH, "Xform")
    target_prim.GetReferences().AddReference(assets_root_path + TARGET_ASSET_URL)
    if not target_prim.HasAttribute("xformOp:translate"):
        UsdGeom.Xformable(target_prim).AddTranslateOp()
    target_prim.GetAttribute("xformOp:translate").Set(TARGET_ASSET_LOCATION)
    remove_all_semantics(target_prim, recursive=True)
    add_update_semantics(target_prim, TARGET_ASSET_LABEL)

    # Setup and run the stacking simulation before capturing the data
    await setup_and_run_stacking_simulation_async(pallets_root_prim)

    # Setup texture randomizer
    await setup_texture_randomizer_async(pallets_root_prim)

    # Setup the light behaviors
    await setup_light_behaviors_async(lights_root_prim)

    # Setup the target asset behaviors
    await setup_target_asset_behaviors_async(target_prim)

    # Setup the camera behaviors
    await setup_camera_behaviors_async(camera_prim, str(target_prim.GetPath()))

    # Setup the writer and capture the data, behavior scripts are triggered by running the timeline
    await setup_writer_and_capture_data_async(camera_path=camera_prim.GetPath(), num_captures=6)


random.seed(10)
rep.set_global_seed(10)

asyncio.ensure_future(run_example_async())