86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
from pythonosc.dispatcher import Dispatcher
|
|
from pythonosc.osc_server import BlockingOSCUDPServer
|
|
from datetime import datetime
|
|
import sys
|
|
|
|
class VrcOscGateway:
|
|
def __init__(self, host="127.0.0.1", port=9001):
|
|
self.host = host
|
|
self.port = port
|
|
self.dispatcher = Dispatcher()
|
|
self.last_values = {}
|
|
|
|
def log(self, level, message):
|
|
now = datetime.now().strftime("%H:%M:%S")
|
|
print(f"[{level}] {now} {message}", flush=True)
|
|
|
|
def to_bool(self, value):
|
|
if isinstance(value, bool):
|
|
return value
|
|
|
|
if isinstance(value, int) or isinstance(value, float):
|
|
return value != 0
|
|
|
|
if isinstance(value, str):
|
|
lowered = value.strip().lower()
|
|
if lowered in ("true", "1", "on"):
|
|
return True
|
|
if lowered in ("false", "0", "off"):
|
|
return False
|
|
|
|
raise ValueError(f"unsupported OSC value: {value}")
|
|
|
|
def is_rising_edge(self, address, value):
|
|
current = self.to_bool(value)
|
|
previous = self.last_values.get(address, False)
|
|
self.last_values[address] = current
|
|
return current and not previous
|
|
|
|
def is_changed(self, address, value):
|
|
current = self.to_bool(value)
|
|
previous = self.last_values.get(address, None)
|
|
|
|
if previous == current:
|
|
return False
|
|
|
|
self.last_values[address] = current
|
|
return True
|
|
|
|
def map(self, parameter_name, handler):
|
|
address = f"/avatar/parameters/{parameter_name}"
|
|
self.dispatcher.map(address, handler)
|
|
self.log("INFO", f"mapped {address}")
|
|
|
|
def set_default_debug(self, enabled=False):
|
|
if not enabled:
|
|
return
|
|
|
|
def on_any(address, *args):
|
|
if address.startswith("/avatar/parameters/"):
|
|
self.log("INFO", f"other address={address} args={args}")
|
|
|
|
self.dispatcher.set_default_handler(on_any)
|
|
|
|
def serve_forever(self):
|
|
self.log("CHECK", "STEP=1 状態確認")
|
|
self.log("INFO", f"listen={self.host}:{self.port}")
|
|
|
|
self.log("CHECK", "STEP=2 想定状態判定")
|
|
self.log("INFO", "VRChat OSC Enabled")
|
|
self.log("INFO", "Python は Windows PowerShell 側で実行")
|
|
self.log("INFO", "Unity側 MA Parameters と OSC名が一致していること")
|
|
|
|
|
|
try:
|
|
server = BlockingOSCUDPServer((self.host, self.port), self.dispatcher)
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
self.log("INFO", "Ctrl+C detected")
|
|
except Exception as e:
|
|
self.log("FATAL", f"OSC server error STEP=3 detail={e}")
|
|
input("Press Enter to exit...")
|
|
sys.exit(1)
|
|
|
|
self.log("CHECK", "STEP=4 結果確認")
|
|
self.log("INFO", "終了しました")
|