Action Graph Code Samples - Python

This file contains a collection of examples for implementing OGN nodes that work in Action Graphs. The features that are particular to Action Graph are accessed with <omni.graph.action.IActionGraph>. All the concepts of OmniGraph apply equally to Action Graph nodes. See OGN User Guide.

`{note} The API used in these samples are usable in kit-sdk version 105.1 or greater. `

Password Branch Node

This example demonstrates branching the incoming control flow based on input data. The node activates the opened output if the password is correct.


from omni.graph.action import get_interface  # import the ActionGraph API


class OgnPassword:
    @staticmethod
    def compute(db) -> bool:
        password = db.inputs.password

        # enable the output execution if authorized
        if password == "Mellon":
            get_interface().set_execution_enabled("outputs:opened")
        else:
            get_interface().set_execution_enabled("outputs:denied")

        return True

[C++ Version]

OnSelect Node

This example demonstrates an event node that activates the selected output when the kit selection changes. Note that this is simpler than the C++ implementation because we use one subscription per node instance instead of sharing the subscription between instances.


import carb
import omni.kit.app
import omni.usd
from com.myextension.ogn.OgnOnSelectDatabase import OgnOnSelectDatabase  # The generated database class
from omni.graph.action import get_interface


class OgnOnSelectInternalState:
    """Convenience class for maintaining per-node state information"""

    def __init__(self):
        self.sub = None  # The stage-event subscription holder
        self.selection_changed = False  # Set to True when a selection change has happened

    def first_time_subscribe(self):
        """Set up the stage event subscription"""
        usd_context = omni.usd.get_context()
        events = usd_context.get_stage_event_stream()
        self.sub = events.create_subscription_to_pop(self._on_stage_event)

    def _on_stage_event(self, e: carb.events.IEvent):
        """The event callback"""
        if e is None:
            return
        if e.type == int(omni.usd.StageEventType.SELECTION_CHANGED):
            self.selection_changed = True


# -----------------------------------------------------------------------------


class OgnOnSelect:
    @staticmethod
    def internal_state():
        """Returns an object that will contain per-node state information"""
        return OgnOnSelectInternalState()

    @staticmethod
    def release(node):
        """Clean up the subscription when node is removed"""
        state = OgnOnSelectDatabase.per_node_internal_state(node)
        if state.sub:
            state.sub.unsubscribe()
            state.sub = None

    @staticmethod
    def compute(db) -> bool:
        state = db.state.internal_state

        if state.sub is None:
            # The initial compute call, set up our subscription
            state.first_time_subscribe()

        if state.selection_changed:
            state.selection_changed = False
            get_interface().set_execution_enabled("outputs:selected")

        return True

[C++ Version]

While Node

This example demonstrates activating an output several times in one update. The node activates the loopBody output while the condition is true, and finally calling the finished output.


from omni.graph.action import get_interface


class OgnWhile:
    @staticmethod
    def compute(db) -> bool:
        keep_going = db.inputs.keepGoing

        if keep_going:
            get_interface().set_execution_enabled_and_pushed("outputs:loopBody")
        else:
            get_interface().set_execution_enabled("outputs:finished")

        return True

[C++ Version]

DoN Node

This example demonstrates a node that enters a latent state for N ticks, before triggering the finished output. While counting down the evaluation will be “paused”. The node logic is if state:count is at the initial value (0), then start the latent state. If the count has reached n, end the latent state and trigger the output.

This is done with {py:func}`omni.graph.action.IActionGraph.start_latent_state` and {py:func}`omni.graph.action.IActionGraph.end_latent_state`.


from omni.graph.action import get_interface


class OgnDoN:
    @staticmethod
    def compute(db) -> bool:
        count = db.state.count
        n = db.inputs.n

        if count == 0:
            get_interface().start_latent_state()
            db.state.count += 1
        elif count >= n:
            db.state.count = 0
            get_interface().end_latent_state()
            get_interface().set_execution_enabled("outputs:finished")
        else:
            get_interface().set_execution_enabled("outputs:tick")
            db.state.count += 1

        return True

[C++ Version]