file_command_handler.pyā¢5.09 kB
#!/usr/bin/env python
"""
File Command Handler for After Effects MCP
This module provides a reliable file-based command mechanism for After Effects MCP.
"""
import os
import json
import time
import logging
from pathlib import Path
import shutil
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
# Configuration
TEMP_DIR = "C:/ae_temp"
COMMAND_FILE = os.path.join(TEMP_DIR, "command.json")
RESULT_FILE = os.path.join(TEMP_DIR, "result.json")
FILE_TIMEOUT = 30 # seconds
def ensure_temp_dir():
"""Ensure the temporary directory exists"""
try:
Path(TEMP_DIR).mkdir(parents=True, exist_ok=True)
logger.info(f"Temp directory ready: {TEMP_DIR}")
return True
except Exception as e:
logger.error(f"Failed to create temp directory: {e}")
return False
def clean_temp_files():
"""Clean up any existing command or result files"""
try:
if os.path.exists(COMMAND_FILE):
os.remove(COMMAND_FILE)
logger.info(f"Removed existing command file")
if os.path.exists(RESULT_FILE):
os.remove(RESULT_FILE)
logger.info(f"Removed existing result file")
return True
except Exception as e:
logger.error(f"Error cleaning temp files: {e}")
return False
def send_command(command_data):
"""Send a command to After Effects via file"""
if not ensure_temp_dir():
return {
"status": "error",
"message": "Failed to create temp directory"
}
# Clean up any existing files
clean_temp_files()
# Write the command to a temporary file first, then move it to the final location
# This ensures atomic write operation
temp_command_file = os.path.join(TEMP_DIR, "temp_command.json")
try:
# Write to temp file
with open(temp_command_file, 'w') as f:
json.dump(command_data, f, indent=2)
# Move to final location (atomic operation)
shutil.move(temp_command_file, COMMAND_FILE)
logger.info(f"Command file written successfully: {json.dumps(command_data)[:100]}...")
logger.info(f"Waiting for After Effects to process command...")
# Wait for the result file
start_time = time.time()
while not os.path.exists(RESULT_FILE):
time.sleep(0.5)
elapsed = time.time() - start_time
if elapsed > FILE_TIMEOUT:
logger.error(f"Timeout waiting for After Effects response")
return {
"status": "error",
"message": "Timeout waiting for After Effects response. Please make sure the MCP Bridge script is running and click 'Check for Command'."
}
# Check if command file still exists (After Effects should remove it)
if not os.path.exists(COMMAND_FILE) and elapsed > 2:
logger.info("Command file was processed but no result file yet")
# Give the file system a moment to complete writing
time.sleep(0.2)
# Read the result
with open(RESULT_FILE, 'r') as f:
result_data = json.load(f)
logger.info(f"Received response: {json.dumps(result_data)[:100]}...")
# Clean up
clean_temp_files()
return result_data
except Exception as e:
logger.error(f"Error in file command handler: {e}")
# Clean up any partial files
clean_temp_files()
return {
"status": "error",
"message": f"Error in file command handler: {str(e)}"
}
def create_text_layer(text, font_size=50, color="#FFFFFF"):
"""Create a text layer in After Effects"""
command = {
"action": "create_text_layer",
"params": {
"text": text,
"fontSize": font_size,
"color": color
},
"confirmation_message": "Creating text layer..."
}
return send_command(command)
def run_custom_code(code):
"""Run custom ExtendScript code in After Effects"""
command = {
"action": "run_custom_code",
"params": {
"code": code
},
"confirmation_message": "Running custom code..."
}
return send_command(command)
def test_connection():
"""Test the connection to After Effects"""
logger.info("Testing connection to After Effects...")
# Simple test command to create a text layer
result = create_text_layer("Connection Test", 50, "#00AAFF")
if result.get("status") == "success":
logger.info("Connection test successful!")
return True
else:
logger.error(f"Connection test failed: {result.get('message', 'Unknown error')}")
return False
if __name__ == "__main__":
# Run a simple test when executed directly
ensure_temp_dir()
test_connection()