👨‍💻 dev vrm

development
fukurou

fukurou

the supreme coder
ADMIN
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)"
 
fukurou

fukurou

the supreme coder
ADMIN
Python: 
import asyncio
import threading
import json
import websockets
from flask import Flask, send_from_directory
import os
import time

# ───────────────────────────────────────────────
# HARDWARE SKILL
# ───────────────────────────────────────────────

class DiVrmController:
    def __init__(self):
        # hardware skill: no output, no speech
        self._clients = set()
        self._ws_server = None
        self._flask_thread = None
        self._loop = None
        self._running = False

        # root folder (where vrm_viewer.html + 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
    # ───────────────────────────────────────────────
    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
    # ───────────────────────────────────────────────
    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)

    # ───────────────────────────────────────────────
    # LIFECYCLE
    # ───────────────────────────────────────────────
    def manifest(self):
        if self._running:
            return

        self._running = True

        # start flask
        self._flask_thread = threading.Thread(
            target=self._run_flask,
            daemon=True
        )
        self._flask_thread.start()

        # start asyncio loop
        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):
        self._running = False
        if self._loop:
            self._loop.call_soon_threadsafe(self._loop.stop)

    # ───────────────────────────────────────────────
    # INPUT (HARDWARE DRIVER)
    # ───────────────────────────────────────────────
    def input(self, ear: str):
        if not ear:
            return

        ear = ear.strip()

        # stop
        if ear == "stop":
            if self._loop:
                asyncio.run_coroutine_threadsafe(
                    self._send_command("stop"),
                    self._loop
                )
            return

        # quit
        if ear in ("quit", "exit"):
            self.ghost()
            return

        # full command: play mixamo/Talking.fbx
        if ear.startswith("play "):
            path = ear[5:].strip()  # remove "play "
            if self._loop:
                asyncio.run_coroutine_threadsafe(
                    self._send_command("play", path),
                    self._loop
                )
            return

        # filename only: Talking.fbx
        mixamo_path = os.path.join(self._ROOT, "mixamo", ear)
        if os.path.isfile(mixamo_path):
            if self._loop:
                asyncio.run_coroutine_threadsafe(
                    self._send_command("play", f"mixamo/{ear}"),
                    self._loop
                )
            return

        # ignore anything else silently (hardware skill)
        return


# ───────────────────────────────────────────────
# SIMPLE MAIN TEST
# ───────────────────────────────────────────────

if __name__ == "__main__":
    ctrl = DiVrmController()

    print("Starting VRM controller…")
    ctrl.manifest()

    # give servers time to start
    time.sleep(2)

    # EXACT TEST COMMAND YOU REQUESTED
    print("TEST: play mixamo/Talking.fbx")
    ctrl.input("play mixamo/Talking.fbx")

    time.sleep(5)

    print("Stopping…")
    ctrl.input("stop")

    time.sleep(1)

    print("Shutting down…")
    ctrl.ghost()

    print("Done.")
 
You must log in or register to reply here.
Top