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
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_instance_internal_state(node)
if state.sub:
state.sub.unsubscribe()
state.sub = None
@staticmethod
def compute(db) -> bool:
state = db.state.per_instance_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
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
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