Usage Examples

Monitor System and GPU Memory

import omni.resourcemonitor as rm

# Acquire the resource monitor interface
resource_monitor = rm.acquire_resource_monitor_interface()

# Get host (system) memory information
total_host_memory = resource_monitor.get_total_host_memory()
available_host_memory = resource_monitor.get_available_host_memory()
used_host_memory = total_host_memory - available_host_memory

print(f"Host Memory: {available_host_memory / (1024**3):.2f} GB available out of {total_host_memory / (1024**3):.2f} GB total")
print(f"Host Memory Usage: {used_host_memory / (1024**3):.2f} GB ({used_host_memory / total_host_memory * 100:.2f}%)")

# Get device (GPU) memory information for device 0
device_index = 0
try:
    total_device_memory = resource_monitor.get_total_device_memory(device_index)
    available_device_memory = resource_monitor.get_available_device_memory(device_index)
    used_device_memory = total_device_memory - available_device_memory

    print(f"Device {device_index} Memory: {available_device_memory / (1024**3):.2f} GB available out of {total_device_memory / (1024**3):.2f} GB total")
    print(f"Device {device_index} Memory Usage: {used_device_memory / (1024**3):.2f} GB ({used_device_memory / total_device_memory * 100:.2f}%)")
except Exception as e:
    print(f"Could not query device {device_index} memory: {e}")

# Release the resource monitor interface when done
rm.release_resource_monitor_interface(resource_monitor)

---

Subscribe to Memory Events

import omni.resourcemonitor as rm
import carb.settings
import time

# Acquire the resource monitor interface
resource_monitor = rm.acquire_resource_monitor_interface()

# Define event handler
def on_resource_event(event):
    # Convert event type to ResourceMonitorEventType enum
    event_type = rm.ResourceMonitorEventType(event.type)
    
    if event_type == rm.ResourceMonitorEventType.HOST_MEMORY:
        print(f"Host memory update: {resource_monitor.get_available_host_memory() / (1024**2):.2f} MB available")
    elif event_type == rm.ResourceMonitorEventType.DEVICE_MEMORY:
        print(f"Device memory update: {resource_monitor.get_available_device_memory(0) / (1024**2):.2f} MB available")
    elif event_type == rm.ResourceMonitorEventType.LOW_HOST_MEMORY:
        print(f"WARNING: Low host memory: {resource_monitor.get_available_host_memory() / (1024**2):.2f} MB available")
    elif event_type == rm.ResourceMonitorEventType.LOW_DEVICE_MEMORY:
        print(f"WARNING: Low device memory: {resource_monitor.get_available_device_memory(0) / (1024**2):.2f} MB available")

# Subscribe to resource monitor events
event_stream = resource_monitor.get_event_stream()
subscription = event_stream.create_subscription_to_pop(on_resource_event, name="resource_monitor_subscription")

print("Monitoring resources for 10 seconds...")
time.sleep(10)  # Monitor for 10 seconds

# Clean up
subscription = None  # Remove subscription
rm.release_resource_monitor_interface(resource_monitor)

---

Configure Memory Warning Thresholds

import omni.resourcemonitor as rm
import carb.settings

# Acquire the resource monitor interface
resource_monitor = rm.acquire_resource_monitor_interface()

# Configure resource monitor settings
settings = carb.settings.get_settings()

# Set query interval (in seconds)
settings.set_float(rm.timeBetweenQueriesSettingName, 1.0)  # Check memory every second

# Enable memory warnings
settings.set_bool(rm.sendHostMemoryWarningSettingName, True)
settings.set_bool(rm.sendDeviceMemoryWarningSettingName, True)

# Set warning thresholds in MB (warn when less than 2GB available)
settings.set_int(rm.hostMemoryWarnMBSettingName, 2000)
settings.set_int(rm.deviceMemoryWarnMBSettingName, 2000)

# Or set warning thresholds as a fraction of total memory (warn when less than 20% available)
settings.set_float(rm.hostMemoryWarnFractionSettingName, 0.2)
settings.set_float(rm.deviceMemoryWarnFractionSettingName, 0.2)

print("Resource monitor warning thresholds configured")

# Release the resource monitor interface when done
rm.release_resource_monitor_interface(resource_monitor)

---

Detect Low Memory Conditions

import omni.resourcemonitor as rm
import carb.settings
import asyncio
import numpy as np

async def monitor_memory_during_allocation():
    # Acquire the resource monitor interface
    resource_monitor = rm.acquire_resource_monitor_interface()
    
    # Get initial memory readings
    initial_host_memory = resource_monitor.get_available_host_memory()
    print(f"Initial available host memory: {initial_host_memory / (1024**3):.2f} GB")
    
    # Set up memory warning detection
    memory_warning_received = False
    
    def on_memory_event(event):
        nonlocal memory_warning_received
        if event.type == int(rm.ResourceMonitorEventType.LOW_HOST_MEMORY):
            memory_warning_received = True
            print("LOW MEMORY WARNING RECEIVED!")
    
    # Configure resource monitor for frequent updates and warnings
    settings = carb.settings.get_settings()
    settings.set_float(rm.timeBetweenQueriesSettingName, 0.1)  # Query every 0.1 seconds
    settings.set_bool(rm.sendHostMemoryWarningSettingName, True)
    
    # Set warning threshold to trigger when memory drops below current - 1GB
    warning_threshold_mb = (initial_host_memory - 1 * 1024 * 1024 * 1024) // (1024 * 1024)
    settings.set_int(rm.hostMemoryWarnMBSettingName, warning_threshold_mb)
    
    # Subscribe to memory events
    subscription = resource_monitor.get_event_stream().create_subscription_to_pop(
        on_memory_event, name='memory_monitor'
    )
    
    try:
        # Perform memory-intensive operation (allocate ~1.5 GB)
        print("Allocating a large array...")
        memory_to_allocate = 1.5 * 1024 * 1024 * 1024
        elements = int(memory_to_allocate // np.dtype(np.int64).itemsize)
        large_array = np.zeros(elements, dtype=np.int64)
        large_array[:] = 1  # Fill array to ensure memory is actually allocated
        
        # Check memory after allocation
        post_allocation_memory = resource_monitor.get_available_host_memory()
        print(f"Available host memory after allocation: {post_allocation_memory / (1024**3):.2f} GB")
        print(f"Memory used by allocation: {(initial_host_memory - post_allocation_memory) / (1024**3):.2f} GB")
        
        # Wait for warnings to be processed
        for _ in range(5):
            await asyncio.sleep(0.2)
            if memory_warning_received:
                break
        
        print(f"Memory warning received: {memory_warning_received}")
    finally:
        # Clean up
        subscription = None
        rm.release_resource_monitor_interface(resource_monitor)

# Run the async function
asyncio.ensure_future(monitor_memory_during_allocation())