import bpy
import sys
import os
import time
import json
import socket
import threading
import traceback
import queue
from pathlib import Path
# Add the repo root to sys.path
current_dir = Path(__file__).parent.resolve()
repo_root = current_dir.parent.parent
sys.path.insert(0, str(repo_root))
print(f"Repo root added to sys.path: {repo_root}")
try:
import addon
# Register the addon classes
addon.register()
# Configuration
PORT = 9876
HOST = 'localhost'
print(f"Starting manual test server on {HOST}:{PORT}")
# Queue for main thread execution
# Items are (command_dict, result_queue)
execution_queue = queue.Queue()
class TestServer:
def __init__(self, host, port):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind((host, port))
self.sock.listen(1)
self.running = True
# Instantiate the real server class to use its logic
self.real_server = addon.BlenderMCPServer(host, port)
def start(self):
t = threading.Thread(target=self.accept_loop)
t.daemon = True
t.start()
def accept_loop(self):
print("Accept loop started")
while self.running:
try:
client, addr = self.sock.accept()
print(f"Connection from {addr}")
t = threading.Thread(target=self.handle_client, args=(client,))
t.daemon = True
t.start()
except Exception as e:
if self.running:
print(f"Accept error: {e}")
def handle_client(self, client):
print("Handling client")
try:
while self.running:
# Read data
data = client.recv(32768)
if not data:
print("Client disconnected (received 0 bytes)")
break
print(f"Received {len(data)} bytes")
try:
# Parse JSON
msg = json.loads(data.decode('utf-8'))
print(f"Command type: {msg.get('type')}")
# Put in queue for main thread
res_q = queue.Queue()
execution_queue.put((msg, res_q))
# Wait for result
result = res_q.get()
# Send response
response_data = json.dumps(result).encode('utf-8')
client.sendall(response_data)
print(f"Sent response: {len(response_data)} bytes")
except json.JSONDecodeError as e:
print(f"JSON Decode Error: {e}")
# Keep reading? Or disconnect?
# For tests, we assume valid JSON per packet usually
pass
except Exception as e:
print(f"Error processing message: {e}")
traceback.print_exc()
break
except Exception as e:
print(f"Client handler error: {e}")
finally:
client.close()
print("Client socket closed")
# Start our test server
server = TestServer(HOST, PORT)
server.start()
# Mock the global server instance so handlers can find it if they look (unlikely but safe)
if not hasattr(bpy.types, "blendermcp_server"):
bpy.types.blendermcp_server = server.real_server
print("BLENDER_MCP_SERVER_READY")
sys.stdout.flush()
# Main thread loop to process commands
print("Entering main loop...")
while True:
try:
# Poll queue with timeout to allow other events (if any)
try:
cmd, res_q = execution_queue.get(timeout=0.05)
# Execute command
# print(f"Executing {cmd.get('type')}...")
try:
# Use the real server's execute_command method
# This calls the handlers which call bpy
result = server.real_server.execute_command(cmd)
res_q.put(result)
except Exception as e:
print(f"Execution error: {e}")
traceback.print_exc()
res_q.put({"status": "error", "message": str(e)})
except queue.Empty:
pass
# Optional: Keep Blender UI responsive-ish (not strictly needed for headless)
# time.sleep(0.01)
except KeyboardInterrupt:
print("Stopping server...")
break
except Exception as e:
print(f"Main loop error: {e}")
traceback.print_exc()
except Exception as e:
print(f"Fatal error in start_blender_server.py: {e}")
traceback.print_exc()
sys.exit(1)
