fix_web_connection.pyā¢10 kB
#!/usr/bin/env python
"""
Fix Web UI to After Effects Connection
This script fixes the connection between the web UI and After Effects by ensuring commands
are properly sent to the correct location.
"""
import os
import sys
import json
import time
import logging
import shutil
import socket
import asyncio
from pathlib import Path
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler("mcp_connection_fix.log")
]
)
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")
MONITOR_INTERVAL = 0.5 # seconds
TCP_HOST = "127.0.0.1"
TCP_PORT = 8250
class WebToAEBridge:
"""Bridge between Web UI and After Effects"""
def __init__(self):
self.ensure_temp_dir()
self.command_queue = []
self.use_tcp = False
self.tcp_socket = None
def ensure_temp_dir(self):
"""Ensure the temporary directory exists"""
try:
Path(TEMP_DIR).mkdir(parents=True, exist_ok=True)
logger.info(f"Temp directory ready: {TEMP_DIR}")
# Test write access
test_file = os.path.join(TEMP_DIR, "test_write.txt")
with open(test_file, 'w') as f:
f.write("Testing write access")
os.remove(test_file)
logger.info("Temp directory is writable")
return True
except Exception as e:
logger.error(f"Failed to create or access temp directory: {e}")
return False
def clean_temp_files(self):
"""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 test_tcp_connection(self):
"""Test if TCP connection to After Effects is available"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
result = sock.connect_ex((TCP_HOST, TCP_PORT))
sock.close()
if result == 0:
logger.info("TCP connection to After Effects is available")
return True
else:
logger.info("TCP connection to After Effects is not available")
return False
except Exception as e:
logger.error(f"Error testing TCP connection: {e}")
return False
def send_command_via_file(self, command_data):
"""Send command to After Effects via file"""
try:
# Clean up any existing files
self.clean_temp_files()
# Create a temporary file first
temp_file = f"{COMMAND_FILE}.tmp"
with open(temp_file, 'w') as f:
json.dump(command_data, f, indent=2)
# Move to final location (atomic operation)
shutil.move(temp_file, COMMAND_FILE)
logger.info(f"Command file created at: {COMMAND_FILE}")
logger.info(f"Command data: {json.dumps(command_data)[:100]}...")
return True
except Exception as e:
logger.error(f"Error sending command via file: {e}")
return False
async def wait_for_result(self, timeout=30):
"""Wait for result file from After Effects"""
start_time = time.time()
while not os.path.exists(RESULT_FILE):
await asyncio.sleep(MONITOR_INTERVAL)
elapsed = time.time() - start_time
if elapsed > timeout:
logger.error(f"Timeout waiting for result file after {timeout} seconds")
return None
try:
# Wait a moment to ensure file is fully written
await asyncio.sleep(0.2)
with open(RESULT_FILE, 'r') as f:
result_data = json.load(f)
logger.info(f"Result received: {json.dumps(result_data)[:100]}...")
# Clean up result file
os.remove(RESULT_FILE)
return result_data
except Exception as e:
logger.error(f"Error reading result file: {e}")
return None
def intercept_websocket_command(self, command_data):
"""Intercept command from web UI and ensure it's properly sent to After Effects"""
logger.info(f"Intercepted command from web UI: {json.dumps(command_data)[:100]}...")
# Add to queue for processing
self.command_queue.append(command_data)
# Return a placeholder response to the web UI
return {
"status": "queued",
"message": "Command queued for processing",
"queue_position": len(self.command_queue)
}
async def process_command_queue(self):
"""Process commands in the queue"""
while self.command_queue:
command = self.command_queue.pop(0)
logger.info(f"Processing command from queue: {json.dumps(command)[:100]}...")
# Send command via file
success = self.send_command_via_file(command)
if success:
# Wait for result
result = await self.wait_for_result()
if result:
logger.info(f"Command processed successfully: {json.dumps(result)[:100]}...")
else:
logger.error("No result received from After Effects")
else:
logger.error("Failed to send command to After Effects")
async def monitor_command_file(self):
"""Monitor command file to detect when it's processed"""
logger.info("Starting command file monitor")
while True:
if os.path.exists(COMMAND_FILE):
file_size = os.path.getsize(COMMAND_FILE)
if file_size > 0:
logger.info(f"Command file found: {COMMAND_FILE} ({file_size} bytes)")
# Wait a moment to see if After Effects processes it
await asyncio.sleep(2)
# If file still exists, it wasn't processed
if os.path.exists(COMMAND_FILE):
logger.warning("Command file not processed by After Effects")
# Try to read the command
try:
with open(COMMAND_FILE, 'r') as f:
command_data = json.load(f)
logger.info(f"Command data: {json.dumps(command_data)[:100]}...")
# Remove the file to avoid repeated processing
os.remove(COMMAND_FILE)
# Add back to queue
self.command_queue.append(command_data)
logger.info("Command added back to queue for retry")
except Exception as e:
logger.error(f"Error reading command file: {e}")
# Remove corrupted file
if os.path.exists(COMMAND_FILE):
os.remove(COMMAND_FILE)
# Process any commands in the queue
if self.command_queue:
await self.process_command_queue()
# Wait before checking again
await asyncio.sleep(MONITOR_INTERVAL)
async def test_connection():
"""Test the connection between web UI and After Effects"""
bridge = WebToAEBridge()
# Create a test command
test_command = {
"action": "create_text_layer",
"params": {
"text": "Connection Test",
"fontSize": 72,
"color": "#FF0000"
}
}
logger.info("Sending test command...")
# Send command via file
success = bridge.send_command_via_file(test_command)
if success:
logger.info("Waiting for result...")
result = await bridge.wait_for_result()
if result:
logger.info(f"Test successful! Result: {json.dumps(result)}")
return True
else:
logger.error("No result received from After Effects")
return False
else:
logger.error("Failed to send test command")
return False
async def main():
"""Main function"""
logger.info("=" * 60)
logger.info("Fix Web UI to After Effects Connection")
logger.info("=" * 60)
bridge = WebToAEBridge()
# Clean up any existing files
bridge.clean_temp_files()
# Test connection
test_result = await test_connection()
if test_result:
logger.info("Connection test successful!")
# Start monitoring command file
await bridge.monitor_command_file()
else:
logger.error("Connection test failed!")
logger.info("Please check that After Effects is running with the MCP Bridge script loaded")
logger.info("and click 'Check for Command' in the MCP Bridge panel.")
if __name__ == "__main__":
asyncio.run(main())