Usage Examples

Creating and Managing Activity Events

import omni.activity.core as act

# Register a callback function to process activity events
def activity_callback(node: act.INode):
    print(f"Activity Node: {node.name}")
    
    # Process child nodes
    for i in range(node.child_count):
        child = node.get_child(i)
        print(f"  Child Node: {child.name}")
        
        # Process events in each child node
        for j in range(child.event_count):
            event = child.get_event(j)
            event_type = "BEGAN" if event.event_type == act.EventType.BEGAN else \
                         "UPDATED" if event.event_type == act.EventType.UPDATED else "ENDED"
            print(f"    Event: {event_type}, Progress: {event.payload.get('progress', 'N/A')}")

# Get activity system instance and register callback
activity = act.get_instance()
callback_id = activity.create_callback_to_pop(activity_callback)

# Enable activity system and generate events
act.enable()
act.began("MainTask|SubTask", progress=0.0)
act.updated("MainTask|SubTask", progress=0.5)
act.ended("MainTask|SubTask", progress=1.0)

# Process all pending callbacks
activity.pump()

# Clean up
activity.remove_callback(callback_id)
act.disable()

---

Working with Nested Activity Hierarchies

import omni.activity.core as act

# Register a callback function to handle different levels of hierarchy
def hierarchy_callback(node: act.INode):
    process_node(node, depth=0)

def process_node(node: act.INode, depth=0):
    indent = "  " * depth
    print(f"{indent}Node: {node.name}")
    
    # Process all events in this node
    for i in range(node.event_count):
        event = node.get_event(i)
        print(f"{indent}  Event Type: {event.event_type.name}")
        if "message" in event.payload:
            print(f"{indent}  Message: {event.payload['message']}")
    
    # Process all child nodes recursively
    for i in range(node.child_count):
        child = node.get_child(i)
        process_node(child, depth + 1)

# Get activity instance and register callback
activity = act.get_instance()
callback_id = activity.create_callback_to_pop(hierarchy_callback)

# Enable activity tracking
act.enable()

# Create a more complex hierarchy of events
act.began("Project", message="Starting project")
act.began("Project|Loading", message="Loading assets")
act.ended("Project|Loading", message="Assets loaded")
act.began("Project|Processing", message="Processing data")
act.updated("Project|Processing", message="Processing in progress")
act.ended("Project|Processing", message="Processing complete")
act.ended("Project", message="Project complete")

# Process all events
activity.pump()

# Clean up
activity.remove_callback(callback_id)
act.disable()

---

Tracking Progress with Activity Events

import omni.activity.core as act
import time

# Callback to monitor progress of activities
def progress_callback(node: act.INode):
    # Only process leaf nodes with events
    if node.child_count == 0 and node.event_count > 0:
        print(f"Activity: {node.name}")
        
        # Look for the latest event with progress information
        latest_progress = 0.0
        latest_type = None
        
        for i in range(node.event_count):
            event = node.get_event(i)
            if 'progress' in event.payload:
                latest_progress = event.payload['progress']
                latest_type = event.event_type
        
        # Report progress based on event type
        if latest_type == act.EventType.BEGAN:
            print(f"  Started: {latest_progress * 100:.1f}% complete")
        elif latest_type == act.EventType.UPDATED:
            print(f"  Progress: {latest_progress * 100:.1f}% complete")
        elif latest_type == act.EventType.ENDED:
            print(f"  Completed: {latest_progress * 100:.1f}% complete")

# Get activity instance and register callback
activity = act.get_instance()
callback_id = activity.create_callback_to_pop(progress_callback)

# Enable activity tracking
act.enable()

# Simulate a task with progress updates
act.began("ImportTask", progress=0.0)

# Simulate work with progress updates
for i in range(1, 10):
    progress = i / 10.0
    act.updated("ImportTask", progress=progress)
    time.sleep(0.1)  # Simulate time passing
    
    # Process events after each update
    activity.pump()

act.ended("ImportTask", progress=1.0)

# Process final event
activity.pump()

# Clean up
activity.remove_callback(callback_id)
act.disable()

---

Working with Event Timestamps and Custom Payload Data

import omni.activity.core as act
import time

# Callback function to analyze event data including timestamps
def event_data_callback(node: act.INode):
    for i in range(node.event_count):
        event = node.get_event(i)
        
        # Get event information
        event_name = node.name
        event_type = event.event_type.name
        timestamp = event.event_timestamp
        
        print(f"Event: {event_name}, Type: {event_type}, Timestamp: {timestamp}")
        
        # Extract and process custom payload data
        payload = event.payload
        print("  Payload data:")
        
        # Standard payload field - progress
        if "progress" in payload:
            print(f"    Progress: {payload['progress']}")
        
        # Custom payload fields
        if "user" in payload:
            print(f"    User: {payload['user']}")
        
        if "message" in payload:
            print(f"    Message: {payload['message']}")
            
        if "status" in payload:
            print(f"    Status: {payload['status']}")

# Get activity instance and register callback
activity = act.get_instance()
callback_id = activity.create_callback_to_pop(event_data_callback)

# Enable activity tracking
act.enable()

# Send events with custom payload data
act.began("DocumentOperation", 
          progress=0.0, 
          user="johndoe", 
          message="Starting document operation")

time.sleep(0.5)  # Simulate time passing

act.updated("DocumentOperation", 
            progress=0.5, 
            status="half-way done")

time.sleep(0.5)  # Simulate time passing

act.ended("DocumentOperation", 
          progress=1.0, 
          message="Operation completed successfully", 
          status="complete")

# Process all events
activity.pump()

# Clean up
activity.remove_callback(callback_id)
act.disable()