Initial VRC OSC commit

src/osc/gateway.py

@@ -0,0 +1,85 @@
+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", "終了しました")