Python:
import asyncio
import threading
import json
import websockets
from flask import Flask, send_from_directory
import os
from Skill import Skill
class DiVrmController(Skill):
def __init__(self):
super().__init__()
self.set_skill_type(1) # regular skill
self.set_skill_lobe(2) # hardware skill
# runtime state
self._clients = set()
self._ws_server = None
self._flask_thread = None
self._loop = None
self._running = False
# file root (folder where vrm_viewer.html and mixamo/ live)
self._ROOT = os.path.dirname(os.path.abspath(__file__))
# flask app
self._app = Flask(__name__)
self._app.add_url_rule("/", "index", self._serve_index)
self._app.add_url_rule("/<path:filename>", "file", self._serve_file)
# ───────────────────────────────────────────────
# FLASK HANDLERS
# ───────────────────────────────────────────────
def _serve_index(self):
return send_from_directory(self._ROOT, "vrm_viewer.html")
def _serve_file(self, filename):
return send_from_directory(self._ROOT, filename)
def _run_flask(self):
self._app.run(port=5500, debug=False, use_reloader=False)
# ───────────────────────────────────────────────
# WEBSOCKET HANDLER
# ───────────────────────────────────────────────
async def _ws_handler(self, websocket):
self._clients.add(websocket)
try:
await websocket.wait_closed()
finally:
self._clients.discard(websocket)
async def _send_command(self, action, path=None):
if not self._clients:
return
msg = {"action": action}
if path:
msg["path"] = path.replace("\\", "/")
data = json.dumps(msg)
dead = []
for ws in list(self._clients):
try:
await ws.send(data)
except:
dead.append(ws)
for ws in dead:
self._clients.discard(ws)
# ───────────────────────────────────────────────
# SKILL LIFECYCLE
# ───────────────────────────────────────────────
def manifest(self):
"""Start Flask + WebSocket servers."""
if self._running:
return
self._running = True
# Start Flask in background thread
self._flask_thread = threading.Thread(
target=self._run_flask,
daemon=True
)
self._flask_thread.start()
# Start asyncio loop in background thread
def start_loop():
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._loop.run_until_complete(self._start_ws_server())
self._loop.run_forever()
threading.Thread(target=start_loop, daemon=True).start()
async def _start_ws_server(self):
self._ws_server = await websockets.serve(
self._ws_handler, "localhost", 8765
)
def ghost(self):
"""Shutdown servers cleanly."""
self._running = False
if self._loop:
self._loop.call_soon_threadsafe(self._loop.stop)
# ───────────────────────────────────────────────
# INPUT: HARDWARE DRIVER BEHAVIOR
# ───────────────────────────────────────────────
def input(self, ear: str, skin: str, eye: str):
# ear is either:
# - "stop"
# - "quit"/"exit"
# - "<file>.fbx" (inside mixamo/)
if not ear:
return
ear = ear.strip()
# stop command
if ear == "stop":
if self._loop:
asyncio.run_coroutine_threadsafe(
self._send_command("stop"),
self._loop
)
return
# quit command
if ear in ("quit", "exit"):
self.ghost()
return
# otherwise treat ear as a filename under mixamo/
# e.g. "Talking.fbx" → mixamo/Talking.fbx
mixamo_path = os.path.join(self._ROOT, "mixamo", ear)
if os.path.isfile(mixamo_path):
# send play command with relative path as browser expects
if self._loop:
asyncio.run_coroutine_threadsafe(
self._send_command("play", f"mixamo/{ear}"),
self._loop
)
else:
# hardware skill: silently ignore invalid input
pass
# ───────────────────────────────────────────────
def skillNotes(self, param: str) -> str:
return "VRM animation controller hardware skill (plays mixamo/*.fbx via WebSocket)"