Process Guide

Cross-platform DCC process monitoring, lifecycle management, and crash recovery.

Overview

Provides:

• Process Monitoring — Live resource usage via PyProcessMonitor (CPU, memory, status)
• DCC Launching — Async spawn/terminate/kill via PyDccLauncher
• Crash Recovery — Restart policy with exponential/fixed backoff via PyCrashRecoveryPolicy
• Background Watching — Event polling via PyProcessWatcher

PyProcessMonitor

Track and query process resource usage using sysinfo.

Basic Usage

import os
from dcc_mcp_core import PyProcessMonitor

monitor = PyProcessMonitor()

# Track current process
monitor.track(os.getpid(), "self")

# Refresh before querying
monitor.refresh()

# Query specific PID
info = monitor.query(os.getpid())
if info:
    print(f"Status: {info['status']}")
    print(f"CPU: {info['cpu_usage_percent']:.1f}%")
    print(f"Memory: {info['memory_bytes'] / 1024 / 1024:.1f} MB")

Track/Untrack

monitor = PyProcessMonitor()

# Track by PID
monitor.track(pid=1234, name="maya")

# Stop tracking
monitor.untrack(pid=1234)

Query Methods

monitor.refresh()

# Query single process
info = monitor.query(pid=1234)

# Query all tracked processes
all_info = monitor.list_all()
for info in all_info:
    print(f"{info['name']}: {info['cpu_usage_percent']}% CPU")

# Check if alive
if monitor.is_alive(pid=1234):
    print("Process is running")

# Count tracked
print(f"Tracking {monitor.tracked_count()} processes")

Returned Dict Keys

Key	Type	Description
pid	int	Process ID
name	str	User-defined name
status	str	OS status string
cpu_usage_percent	float	CPU usage (0-100)
memory_bytes	int	Memory usage in bytes
restart_count	int	Restart count

PyDccLauncher

Launch and manage DCC processes asynchronously.

Basic Launch

from dcc_mcp_core import PyDccLauncher

launcher = PyDccLauncher()

# Launch a DCC
info = launcher.launch(
    name="maya",
    executable="/usr/autodesk/maya/bin/maya",
    args=["-prompt", "-batch"],
    launch_timeout_ms=30000,
)

print(f"Launched PID: {info['pid']}")

Launch with Environment

info = launcher.launch(
    name="nuke-mcp",
    executable="/opt/Nuke15.2/Nuke15.2",
    args=["--disable-nuke-frameserver", "project.nk"],
    launch_timeout_ms=60000,
    environment={
        "NUKE_DISABLE_FRAMESERVER": "1",
        "DCC_MCP_NUKE_PORT": "0",
    },
    working_directory="/projects/solar-system",
)

Environment values and the working directory apply only to the launched child. The parent process environment is unchanged, so multiple isolated DCC sessions can use different runtime policies safely.

Process Lifecycle

# Terminate gracefully
launcher.terminate("maya", timeout_ms=5000)

# Kill forcefully
launcher.kill("maya")

# Get PID by name
pid = launcher.pid_of("maya")
if pid:
    print(f"Maya running as PID {pid}")

# Check running count
print(f"Running: {launcher.running_count()} processes")

# Check restart count
print(f"Restart count: {launcher.restart_count('maya')}")

Maya Example

launcher = PyDccLauncher()

maya_info = launcher.launch(
    name="maya-2025",
    executable="/usr/autodesk/maya/bin/maya",
    args=["-prompt", "-batch"],
    launch_timeout_ms=60000,
)

print(f"Maya running as PID {maya_info['pid']}")

# ... do work ...

launcher.terminate("maya-2025")

PyCrashRecoveryPolicy

Automatic restart policy with backoff strategies.

Basic Policy

from dcc_mcp_core import PyCrashRecoveryPolicy

policy = PyCrashRecoveryPolicy(max_restarts=3)
policy.use_exponential_backoff(initial_ms=1000, max_delay_ms=30000)

# Check if should restart
if policy.should_restart("crashed"):
    delay = policy.next_delay_ms("maya", attempt=0)
    print(f"Restarting in {delay}ms...")

Fixed Backoff

policy = PyCrashRecoveryPolicy(max_restarts=5)
policy.use_fixed_backoff(delay_ms=2000)

if policy.should_restart("unresponsive"):
    delay = policy.next_delay_ms("maya", attempt=0)
    print(f"Retrying in {delay}ms...")

Exponential Backoff

policy = PyCrashRecoveryPolicy(max_restarts=3)
policy.use_exponential_backoff(initial_ms=1000, max_delay_ms=30000)

# Attempt 0 -> 1000ms, Attempt 1 -> 2000ms, Attempt 2 -> 4000ms
for attempt in range(3):
    if policy.should_restart("crashed"):
        delay = policy.next_delay_ms("maya", attempt=attempt)
        print(f"Attempt {attempt}: waiting {delay}ms")

Managing Policy State

policy = PyCrashRecoveryPolicy(max_restarts=3)

# Check max_restarts limit
print(f"Max restarts: {policy.max_restarts}")

# Check restart eligibility
if policy.should_restart("crashed"):
    # Attempt restart
    pass

PyProcessWatcher

Async background process watcher with event polling.

Basic Watch

import os
import time
from dcc_mcp_core import PyProcessWatcher

watcher = PyProcessWatcher(poll_interval_ms=200)
watcher.track(os.getpid(), "self")
watcher.start()

time.sleep(0.5)

# Poll for events
for event in watcher.poll_events():
    print(f"Event: {event['type']} - {event['name']}")

watcher.stop()

Event Types

Event dicts contain: type, pid, name

Event Type	Additional Fields
heartbeat	new_status, cpu_usage_percent, memory_bytes
status_changed	old_status, new_status
exited	—

Polling Pattern

watcher = PyProcessWatcher(poll_interval_ms=500)
watcher.track(pid=1234, name="maya")
watcher.start()

try:
    while True:
        events = watcher.poll_events()
        for event in events:
            if event["type"] == "exited":
                print(f"{event['name']} exited")
            elif event["type"] == "heartbeat":
                print(f"CPU: {event['cpu_usage_percent']}%")
        time.sleep(0.1)
finally:
    watcher.stop()

Start/Stop

watcher = PyProcessWatcher()

watcher.track(pid=1234, name="maya")
watcher.start()

# ... do work ...

watcher.stop()

# Check status
print(f"Watcher running: {watcher.is_running()}")
print(f"Tracked: {watcher.tracked_count()}")

Complete Example

Auto-Restart DCC

import time
from dcc_mcp_core import PyDccLauncher, PyProcessWatcher, PyCrashRecoveryPolicy

launcher = PyDccLauncher()
watcher = PyProcessWatcher(poll_interval_ms=500)
policy = PyCrashRecoveryPolicy(max_restarts=5)
policy.use_exponential_backoff(initial_ms=1000, max_delay_ms=30000)

# Launch Maya
info = launcher.launch(
    name="maya",
    executable="/usr/autodesk/maya/bin/maya",
    args=["-prompt"],
)
print(f"Launched Maya PID {info['pid']}")

watcher.track(info["pid"], "maya")
watcher.start()

attempt = 0
while True:
    events = watcher.poll_events()
    for event in events:
        if event["type"] == "exited":
            print("Maya exited")
            if policy.should_restart("crashed") and attempt < 5:
                delay = policy.next_delay_ms("maya", attempt=attempt)
                print(f"Restarting in {delay}ms...")
                time.sleep(delay / 1000)
                info = launcher.launch(
                    name="maya",
                    executable="/usr/autodesk/maya/bin/maya",
                    args=["-prompt"],
                )
                watcher.track(info["pid"], "maya")
                attempt += 1
            else:
                print("Max restarts exceeded")
                watcher.stop()
                exit(1)

    time.sleep(0.1)

Best Practices

1. Always Refresh Before Query

monitor.refresh()
info = monitor.query(pid=1234)  # Now has fresh data

2. Handle Missing Processes Gracefully

info = monitor.query(pid=1234)
if info is None:
    print("Process not found")
else:
    print(f"CPU: {info['cpu_usage_percent']}%")

3. Use Appropriate Timeouts

# Short timeout for quick operations
launcher.terminate("quick_proc", timeout_ms=2000)

# Longer timeout for DCC apps
launcher.terminate("maya", timeout_ms=10000)

4. Monitor Resource Usage

def check_resources():
    monitor.refresh()
    for info in monitor.list_all():
        if info["cpu_usage_percent"] > 90:
            print(f"High CPU: {info['name']}")
        if info["memory_bytes"] > 10 * 1024 * 1024 * 1024:
            print(f"High memory: {info['name']}")