Initial commit
This commit is contained in:
Zhongwei Li
485
skills/addon/websocket_server.py
Normal file
485
skills/addon/websocket_server.py
Normal file
@@ -0,0 +1,485 @@
|
||||
"""
|
||||
WebSocket Server for Blender Toolkit
|
||||
Claude Code와 통신하기 위한 WebSocket 서버
|
||||
|
||||
이 모듈은 Blender 내부에서 WebSocket 서버를 실행하여
|
||||
외부 클라이언트(Claude Code)와 JSON-RPC 스타일 통신을 제공합니다.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Any, Dict, Union
|
||||
|
||||
import bpy
|
||||
from aiohttp import web
|
||||
from aiohttp.web import Request, WebSocketResponse
|
||||
|
||||
from .utils.logger import get_logger
|
||||
from .utils.security import validate_port
|
||||
|
||||
# 모듈 로거 초기화
|
||||
logger = get_logger('websocket_server')
|
||||
|
||||
# 보안 상수
|
||||
MAX_CONNECTIONS = 5 # 최대 동시 연결 수 (로컬 환경)
|
||||
|
||||
|
||||
class BlenderWebSocketServer:
|
||||
"""WebSocket 서버 메인 클래스"""
|
||||
|
||||
def __init__(self, port: int = 9400):
|
||||
self.port = validate_port(port)
|
||||
self.app = None
|
||||
self.runner = None
|
||||
self.site = None
|
||||
self.clients = []
|
||||
|
||||
async def handle_command(
|
||||
self, request: Request
|
||||
) -> Union[WebSocketResponse, web.Response]:
|
||||
"""WebSocket 연결 핸들러"""
|
||||
# 로컬호스트만 허용 (보안)
|
||||
peername = request.transport.get_extra_info('peername')
|
||||
if peername:
|
||||
host = peername[0]
|
||||
if host not in ('127.0.0.1', '::1', 'localhost'):
|
||||
logger.warning(
|
||||
"Rejected connection from non-localhost: %s", host
|
||||
)
|
||||
return web.Response(
|
||||
status=403, text="Only localhost connections allowed"
|
||||
)
|
||||
|
||||
# 최대 연결 수 제한 (DoS 방지)
|
||||
if len(self.clients) >= MAX_CONNECTIONS:
|
||||
logger.warning(
|
||||
"Connection limit reached (%d)", MAX_CONNECTIONS
|
||||
)
|
||||
return web.Response(status=503, text="Too many connections")
|
||||
|
||||
ws = web.WebSocketResponse()
|
||||
await ws.prepare(request)
|
||||
|
||||
self.clients.append(ws)
|
||||
logger.info("Client connected (total: %d)", len(self.clients))
|
||||
print(f"✅ Client connected (total: {len(self.clients)})")
|
||||
|
||||
async for msg in ws:
|
||||
if msg.type == web.WSMsgType.TEXT:
|
||||
try:
|
||||
data = json.loads(msg.data)
|
||||
logger.debug("Received message: %s", data)
|
||||
response = await self.process_command(data)
|
||||
await ws.send_json(response)
|
||||
except (json.JSONDecodeError, KeyError, ValueError) as e:
|
||||
logger.error("Error handling message: %s", e, exc_info=True)
|
||||
await ws.send_json({
|
||||
"id": data.get("id") if 'data' in locals() else None,
|
||||
"error": {
|
||||
"code": -1,
|
||||
"message": str(e)
|
||||
}
|
||||
})
|
||||
elif msg.type == web.WSMsgType.ERROR:
|
||||
logger.error('WebSocket error: %s', ws.exception())
|
||||
print(f'❌ WebSocket error: {ws.exception()}')
|
||||
|
||||
self.clients.remove(ws)
|
||||
logger.info("Client disconnected (total: %d)", len(self.clients))
|
||||
print(f"🔌 Client disconnected (total: {len(self.clients)})")
|
||||
return ws
|
||||
|
||||
async def process_command(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""명령 처리"""
|
||||
method = data.get("method")
|
||||
params = data.get("params", {})
|
||||
msg_id = data.get("id")
|
||||
|
||||
logger.info("Processing command: %s", method)
|
||||
logger.debug("Command params: %s", params)
|
||||
print(f"📨 Received command: {method}")
|
||||
|
||||
try:
|
||||
# 메서드 라우팅
|
||||
if method.startswith("Armature."):
|
||||
result = await self.handle_armature_command(method, params)
|
||||
elif method.startswith("Retargeting."):
|
||||
result = await self.handle_retargeting_command(method, params)
|
||||
elif method.startswith("BoneMapping."):
|
||||
result = await self.handle_bonemapping_command(method, params)
|
||||
elif method.startswith("Animation."):
|
||||
result = await self.handle_animation_command(method, params)
|
||||
elif method.startswith("Import."):
|
||||
result = await self.handle_import_command(method, params)
|
||||
elif method.startswith("Geometry."):
|
||||
result = await self.handle_geometry_command(method, params)
|
||||
elif method.startswith("Object."):
|
||||
result = await self.handle_object_command(method, params)
|
||||
elif method.startswith("Modifier."):
|
||||
result = await self.handle_modifier_command(method, params)
|
||||
elif method.startswith("Material."):
|
||||
result = await self.handle_material_command(method, params)
|
||||
elif method.startswith("Collection."):
|
||||
result = await self.handle_collection_command(method, params)
|
||||
else:
|
||||
raise ValueError(f"Unknown method: {method}")
|
||||
|
||||
logger.info("Command %s completed successfully", method)
|
||||
return {"id": msg_id, "result": result}
|
||||
except (ValueError, KeyError, AttributeError, RuntimeError) as e:
|
||||
logger.error("Error processing %s: %s", method, str(e), exc_info=True)
|
||||
print(f"❌ Error processing {method}: {str(e)}")
|
||||
return {
|
||||
"id": msg_id,
|
||||
"error": {"code": -1, "message": str(e)}
|
||||
}
|
||||
|
||||
async def handle_armature_command(self, method: str, params: Dict) -> Any:
|
||||
"""아마추어 관련 명령 처리"""
|
||||
from .retargeting import get_bones, list_armatures
|
||||
|
||||
if method == "Armature.getBones":
|
||||
armature_name = params.get("armatureName")
|
||||
return get_bones(armature_name)
|
||||
elif method == "Armature.list":
|
||||
return list_armatures()
|
||||
else:
|
||||
raise ValueError(f"Unknown armature method: {method}")
|
||||
|
||||
async def handle_retargeting_command(self, method: str, params: Dict) -> Any:
|
||||
"""리타게팅 명령 처리"""
|
||||
from .retargeting import auto_map_bones, retarget_animation, get_preset_bone_mapping
|
||||
|
||||
if method == "Retargeting.autoMapBones":
|
||||
return auto_map_bones(
|
||||
params.get("sourceArmature"),
|
||||
params.get("targetArmature")
|
||||
)
|
||||
elif method == "Retargeting.retargetAnimation":
|
||||
return retarget_animation(
|
||||
params.get("sourceArmature"),
|
||||
params.get("targetArmature"),
|
||||
params.get("boneMap"),
|
||||
params.get("preserveRotation", True),
|
||||
params.get("preserveLocation", False)
|
||||
)
|
||||
elif method == "Retargeting.getPresetMapping":
|
||||
preset = params.get("preset")
|
||||
return get_preset_bone_mapping(preset)
|
||||
else:
|
||||
raise ValueError(f"Unknown retargeting method: {method}")
|
||||
|
||||
async def handle_bonemapping_command(self, method: str, params: Dict) -> Any:
|
||||
"""본 매핑 명령 처리"""
|
||||
from .retargeting import store_bone_mapping, load_bone_mapping
|
||||
|
||||
if method == "BoneMapping.show":
|
||||
return store_bone_mapping(
|
||||
params.get("sourceArmature"),
|
||||
params.get("targetArmature"),
|
||||
params.get("boneMapping")
|
||||
)
|
||||
elif method == "BoneMapping.get":
|
||||
return load_bone_mapping(
|
||||
params.get("sourceArmature"),
|
||||
params.get("targetArmature")
|
||||
)
|
||||
else:
|
||||
raise ValueError(f"Unknown bone mapping method: {method}")
|
||||
|
||||
async def handle_animation_command(self, method: str, params: Dict) -> Any:
|
||||
"""애니메이션 명령 처리"""
|
||||
from .retargeting import list_animations, play_animation, stop_animation, add_to_nla
|
||||
|
||||
if method == "Animation.list":
|
||||
armature_name = params.get("armatureName")
|
||||
return list_animations(armature_name)
|
||||
elif method == "Animation.play":
|
||||
return play_animation(
|
||||
params.get("armatureName"),
|
||||
params.get("actionName"),
|
||||
params.get("loop", True)
|
||||
)
|
||||
elif method == "Animation.stop":
|
||||
return stop_animation()
|
||||
elif method == "Animation.addToNLA":
|
||||
return add_to_nla(
|
||||
params.get("armatureName"),
|
||||
params.get("actionName"),
|
||||
params.get("trackName")
|
||||
)
|
||||
else:
|
||||
raise ValueError(f"Unknown animation method: {method}")
|
||||
|
||||
async def handle_import_command(self, method: str, params: Dict) -> Any:
|
||||
"""임포트 명령 처리"""
|
||||
from .retargeting import import_fbx, import_dae
|
||||
|
||||
if method == "Import.fbx":
|
||||
return import_fbx(params.get("filepath"))
|
||||
elif method == "Import.dae":
|
||||
return import_dae(params.get("filepath"))
|
||||
else:
|
||||
raise ValueError(f"Unknown import method: {method}")
|
||||
|
||||
async def handle_geometry_command(self, method: str, params: Dict) -> Any:
|
||||
"""도형 생성 명령 처리"""
|
||||
from .commands.geometry import (
|
||||
create_cube, create_sphere, create_cylinder,
|
||||
create_plane, create_cone, create_torus,
|
||||
get_vertices, move_vertex, subdivide_mesh, extrude_face
|
||||
)
|
||||
|
||||
if method == "Geometry.createCube":
|
||||
return create_cube(
|
||||
location=tuple(params.get("location", [0, 0, 0])),
|
||||
size=params.get("size", 2.0),
|
||||
name=params.get("name")
|
||||
)
|
||||
elif method == "Geometry.createSphere":
|
||||
return create_sphere(
|
||||
location=tuple(params.get("location", [0, 0, 0])),
|
||||
radius=params.get("radius", 1.0),
|
||||
segments=params.get("segments", 32),
|
||||
ring_count=params.get("ringCount", 16),
|
||||
name=params.get("name")
|
||||
)
|
||||
elif method == "Geometry.createCylinder":
|
||||
return create_cylinder(
|
||||
location=tuple(params.get("location", [0, 0, 0])),
|
||||
radius=params.get("radius", 1.0),
|
||||
depth=params.get("depth", 2.0),
|
||||
vertices=params.get("vertices", 32),
|
||||
name=params.get("name")
|
||||
)
|
||||
elif method == "Geometry.createPlane":
|
||||
return create_plane(
|
||||
location=tuple(params.get("location", [0, 0, 0])),
|
||||
size=params.get("size", 2.0),
|
||||
name=params.get("name")
|
||||
)
|
||||
elif method == "Geometry.createCone":
|
||||
return create_cone(
|
||||
location=tuple(params.get("location", [0, 0, 0])),
|
||||
radius1=params.get("radius1", 1.0),
|
||||
depth=params.get("depth", 2.0),
|
||||
vertices=params.get("vertices", 32),
|
||||
name=params.get("name")
|
||||
)
|
||||
elif method == "Geometry.createTorus":
|
||||
return create_torus(
|
||||
location=tuple(params.get("location", [0, 0, 0])),
|
||||
major_radius=params.get("majorRadius", 1.0),
|
||||
minor_radius=params.get("minorRadius", 0.25),
|
||||
major_segments=params.get("majorSegments", 48),
|
||||
minor_segments=params.get("minorSegments", 12),
|
||||
name=params.get("name")
|
||||
)
|
||||
elif method == "Geometry.getVertices":
|
||||
return get_vertices(params.get("name"))
|
||||
elif method == "Geometry.moveVertex":
|
||||
return move_vertex(
|
||||
object_name=params.get("objectName"),
|
||||
vertex_index=params.get("vertexIndex"),
|
||||
new_position=tuple(params.get("newPosition"))
|
||||
)
|
||||
elif method == "Geometry.subdivideMesh":
|
||||
return subdivide_mesh(
|
||||
name=params.get("name"),
|
||||
cuts=params.get("cuts", 1)
|
||||
)
|
||||
elif method == "Geometry.extrudeFace":
|
||||
return extrude_face(
|
||||
object_name=params.get("objectName"),
|
||||
face_index=params.get("faceIndex"),
|
||||
offset=params.get("offset", 1.0)
|
||||
)
|
||||
else:
|
||||
raise ValueError(f"Unknown geometry method: {method}")
|
||||
|
||||
async def handle_object_command(self, method: str, params: Dict) -> Any:
|
||||
"""오브젝트 명령 처리"""
|
||||
from .commands.geometry import (
|
||||
delete_object, transform_object, duplicate_object, list_objects
|
||||
)
|
||||
|
||||
if method == "Object.delete":
|
||||
return delete_object(params.get("name"))
|
||||
elif method == "Object.transform":
|
||||
location = params.get("location")
|
||||
rotation = params.get("rotation")
|
||||
scale = params.get("scale")
|
||||
return transform_object(
|
||||
name=params.get("name"),
|
||||
location=tuple(location) if location else None,
|
||||
rotation=tuple(rotation) if rotation else None,
|
||||
scale=tuple(scale) if scale else None
|
||||
)
|
||||
elif method == "Object.duplicate":
|
||||
location = params.get("location")
|
||||
return duplicate_object(
|
||||
name=params.get("name"),
|
||||
new_name=params.get("newName"),
|
||||
location=tuple(location) if location else None
|
||||
)
|
||||
elif method == "Object.list":
|
||||
return list_objects(params.get("type"))
|
||||
else:
|
||||
raise ValueError(f"Unknown object method: {method}")
|
||||
|
||||
async def handle_modifier_command(self, method: str, params: Dict) -> Any:
|
||||
"""모디파이어 명령 처리"""
|
||||
from .commands.modifier import (
|
||||
add_modifier, apply_modifier, list_modifiers, remove_modifier,
|
||||
toggle_modifier, modify_modifier_properties, get_modifier_info, reorder_modifier
|
||||
)
|
||||
|
||||
if method == "Modifier.add":
|
||||
properties = params.get("properties", {})
|
||||
return add_modifier(
|
||||
object_name=params.get("objectName"),
|
||||
modifier_type=params.get("modifierType"),
|
||||
name=params.get("name")
|
||||
)
|
||||
elif method == "Modifier.apply":
|
||||
return apply_modifier(
|
||||
object_name=params.get("objectName"),
|
||||
modifier_name=params.get("modifierName")
|
||||
)
|
||||
elif method == "Modifier.list":
|
||||
return list_modifiers(
|
||||
object_name=params.get("objectName")
|
||||
)
|
||||
elif method == "Modifier.remove":
|
||||
return remove_modifier(
|
||||
object_name=params.get("objectName"),
|
||||
modifier_name=params.get("modifierName")
|
||||
)
|
||||
elif method == "Modifier.toggle":
|
||||
return toggle_modifier(
|
||||
object_name=params.get("objectName"),
|
||||
modifier_name=params.get("modifierName"),
|
||||
viewport=params.get("viewport"),
|
||||
render=params.get("render")
|
||||
)
|
||||
elif method == "Modifier.modify":
|
||||
properties = params.get("properties", {})
|
||||
return modify_modifier_properties(
|
||||
object_name=params.get("objectName"),
|
||||
modifier_name=params.get("modifierName"),
|
||||
**properties
|
||||
)
|
||||
elif method == "Modifier.getInfo":
|
||||
return get_modifier_info(
|
||||
object_name=params.get("objectName"),
|
||||
modifier_name=params.get("modifierName")
|
||||
)
|
||||
elif method == "Modifier.reorder":
|
||||
return reorder_modifier(
|
||||
object_name=params.get("objectName"),
|
||||
modifier_name=params.get("modifierName"),
|
||||
direction=params.get("direction")
|
||||
)
|
||||
else:
|
||||
raise ValueError(f"Unknown modifier method: {method}")
|
||||
|
||||
async def handle_material_command(self, method: str, params: Dict) -> Any:
|
||||
"""머티리얼 명령 처리"""
|
||||
from .commands.material import (
|
||||
create_material, list_materials, delete_material,
|
||||
assign_material, list_object_materials,
|
||||
set_material_base_color, set_material_metallic, set_material_roughness,
|
||||
set_material_emission, get_material_properties
|
||||
)
|
||||
|
||||
if method == "Material.create":
|
||||
return create_material(
|
||||
name=params.get("name"),
|
||||
use_nodes=params.get("useNodes", True)
|
||||
)
|
||||
elif method == "Material.list":
|
||||
return list_materials()
|
||||
elif method == "Material.delete":
|
||||
return delete_material(name=params.get("name"))
|
||||
elif method == "Material.assign":
|
||||
return assign_material(
|
||||
object_name=params.get("objectName"),
|
||||
material_name=params.get("materialName"),
|
||||
slot_index=params.get("slotIndex", 0)
|
||||
)
|
||||
elif method == "Material.listObjectMaterials":
|
||||
return list_object_materials(object_name=params.get("objectName"))
|
||||
elif method == "Material.setBaseColor":
|
||||
color = params.get("color")
|
||||
return set_material_base_color(
|
||||
material_name=params.get("materialName"),
|
||||
color=tuple(color) if isinstance(color, list) else color
|
||||
)
|
||||
elif method == "Material.setMetallic":
|
||||
return set_material_metallic(
|
||||
material_name=params.get("materialName"),
|
||||
metallic=params.get("metallic")
|
||||
)
|
||||
elif method == "Material.setRoughness":
|
||||
return set_material_roughness(
|
||||
material_name=params.get("materialName"),
|
||||
roughness=params.get("roughness")
|
||||
)
|
||||
elif method == "Material.setEmission":
|
||||
color = params.get("color")
|
||||
return set_material_emission(
|
||||
material_name=params.get("materialName"),
|
||||
color=tuple(color) if isinstance(color, list) else color,
|
||||
strength=params.get("strength", 1.0)
|
||||
)
|
||||
elif method == "Material.getProperties":
|
||||
return get_material_properties(material_name=params.get("materialName"))
|
||||
else:
|
||||
raise ValueError(f"Unknown material method: {method}")
|
||||
|
||||
async def handle_collection_command(self, method: str, params: Dict) -> Any:
|
||||
"""컬렉션 명령 처리"""
|
||||
from .commands.collection import (
|
||||
create_collection, list_collections, add_to_collection,
|
||||
remove_from_collection, delete_collection
|
||||
)
|
||||
|
||||
if method == "Collection.create":
|
||||
return create_collection(name=params.get("name"))
|
||||
elif method == "Collection.list":
|
||||
return list_collections()
|
||||
elif method == "Collection.addObject":
|
||||
return add_to_collection(
|
||||
object_name=params.get("objectName"),
|
||||
collection_name=params.get("collectionName")
|
||||
)
|
||||
elif method == "Collection.removeObject":
|
||||
return remove_from_collection(
|
||||
object_name=params.get("objectName"),
|
||||
collection_name=params.get("collectionName")
|
||||
)
|
||||
elif method == "Collection.delete":
|
||||
return delete_collection(name=params.get("name"))
|
||||
else:
|
||||
raise ValueError(f"Unknown collection method: {method}")
|
||||
|
||||
async def start(self):
|
||||
"""서버 시작"""
|
||||
self.app = web.Application()
|
||||
self.app.router.add_get('/ws', self.handle_command)
|
||||
|
||||
self.runner = web.AppRunner(self.app)
|
||||
await self.runner.setup()
|
||||
|
||||
self.site = web.TCPSite(self.runner, '127.0.0.1', self.port)
|
||||
await self.site.start()
|
||||
|
||||
print(f"✅ Blender WebSocket Server started on port {self.port}")
|
||||
|
||||
async def stop(self):
|
||||
"""서버 중지"""
|
||||
if self.site:
|
||||
await self.site.stop()
|
||||
if self.runner:
|
||||
await self.runner.cleanup()
|
||||
print("🛑 Blender WebSocket Server stopped")
|
||||
Reference in New Issue
Block a user