602 lines
18 KiB
Markdown
602 lines
18 KiB
Markdown
# osc.py - OSC通信・OSCQueryプロトコル管理
|
||
|
||
## 概要
|
||
|
||
VRChatとの高度なOSC（Open Sound Control）通信を管理する包括的なシステムです。基本的なOSCメッセージ送信に加え、OSCQueryプロトコルによる双方向通信、パラメータ監視、自動サービス発見機能を提供します。
|
||
|
||
## 主要機能
|
||
|
||
### OSC通信機能
|
||
- VRChatチャットボックスへのメッセージ送信
|
||
- タイピング状態の制御
|
||
- パラメータ値の動的取得
|
||
|
||
### OSCQuery対応
|
||
- 自動サービス発見・接続
|
||
- リアルタイムパラメータ監視
|
||
- 双方向エンドポイント公開
|
||
|
||
### 堅牢性機能
|
||
- 防御的プログラミング設計
|
||
- 欠損ライブラリの優雅な処理
|
||
- 自動エラー復旧機構
|
||
|
||
## クラス構造
|
||
|
||
### OSCHandler クラス
|
||
|
||
```python
|
||
class OSCHandler:
|
||
def __init__(self, ip_address: str = "127.0.0.1", port: int = 9000) -> None:
|
||
self.is_osc_query_enabled: bool
|
||
self.osc_ip_address: str
|
||
self.osc_port: int
|
||
self.udp_client: udp_client.SimpleUDPClient
|
||
self.osc_server: Optional[osc_server.ThreadingOSCUDPServer]
|
||
self.osc_query_service: Optional[OSCQueryService]
|
||
self.browser: Optional[OSCQueryBrowser]
|
||
```
|
||
|
||
OSC通信の中核管理クラス
|
||
|
||
#### 属性
|
||
- **is_osc_query_enabled**: OSCQuery機能の有効性フラグ
|
||
- **osc_ip_address**: 送信先IPアドレス
|
||
- **osc_port**: UDP通信ポート
|
||
- **udp_client**: OSC送信クライアント
|
||
- **osc_server**: ローカルOSCサーバー
|
||
- **osc_query_service**: OSCQueryサービスインスタンス
|
||
- **browser**: OSCQueryブラウザー
|
||
|
||
## 主要メソッド
|
||
|
||
### メッセージ送信
|
||
|
||
```python
|
||
def sendMessage(self, message: str = "", notification: bool = True) -> None
|
||
```
|
||
|
||
VRChatチャットボックスにメッセージを送信
|
||
|
||
#### パラメータ
|
||
- **message**: 送信するテキストメッセージ
|
||
- **notification**: 通知フラグ（音・表示の有無）
|
||
|
||
```python
|
||
def sendTyping(self, flag: bool = False) -> None
|
||
```
|
||
|
||
タイピング状態をVRChatに送信
|
||
|
||
#### パラメータ
|
||
- **flag**: タイピング中フラグ
|
||
|
||
### パラメータ監視
|
||
|
||
```python
|
||
def getOSCParameterMuteSelf() -> Optional[bool]
|
||
```
|
||
|
||
VRChatのMuteSelfパラメータ値を取得
|
||
|
||
#### 戻り値
|
||
- **Optional[bool]**: ミュート状態（取得失敗時はNone）
|
||
|
||
```python
|
||
def getOSCParameterValue(self, address: str) -> Any
|
||
```
|
||
|
||
任意のOSCパラメータ値を取得
|
||
|
||
#### パラメータ
|
||
- **address**: OSCアドレス（例："/avatar/parameters/MuteSelf"）
|
||
|
||
#### 戻り値
|
||
- **Any**: パラメータ値（取得失敗時はNone）
|
||
|
||
### 設定変更
|
||
|
||
```python
|
||
def setOscIpAddress(self, ip_address: str) -> None
|
||
```
|
||
|
||
送信先IPアドレスを変更し、サービスを再初期化
|
||
|
||
#### パラメータ
|
||
- **ip_address**: 新しいIPアドレス
|
||
|
||
```python
|
||
def setOscPort(self, port: int) -> None
|
||
```
|
||
|
||
送信ポートを変更し、サービスを再初期化
|
||
|
||
#### パラメータ
|
||
- **port**: 新しいUDPポート番号
|
||
|
||
## 使用方法
|
||
|
||
### 基本的なメッセージ送信
|
||
|
||
```python
|
||
from models.osc.osc import OSCHandler
|
||
|
||
# OSCハンドラーの初期化
|
||
osc = OSCHandler(ip_address="127.0.0.1", port=9000)
|
||
|
||
# チャットボックスにメッセージを送信
|
||
osc.sendMessage("こんにちは、VRChat！", notification=True)
|
||
|
||
# タイピング状態の制御
|
||
osc.sendTyping(True) # タイピング開始
|
||
# ... 実際のタイピング処理 ...
|
||
osc.sendTyping(False) # タイピング終了
|
||
|
||
# 再度メッセージ送信
|
||
osc.sendMessage("翻訳完了しました", notification=False)
|
||
```
|
||
|
||
### リモートVRChatへの接続
|
||
|
||
```python
|
||
# リモートVRChatインスタンスへの接続
|
||
remote_osc = OSCHandler(ip_address="192.168.1.100", port=9000)
|
||
|
||
# OSCQuery機能は自動的に無効化される
|
||
print(f"OSCQuery有効: {remote_osc.getIsOscQueryEnabled()}") # False
|
||
|
||
# 基本的なメッセージ送信は利用可能
|
||
remote_osc.sendMessage("リモートからの翻訳結果", notification=True)
|
||
```
|
||
|
||
### パラメータ監視（ローカル接続時のみ）
|
||
|
||
```python
|
||
# ローカル接続でのパラメータ監視
|
||
local_osc = OSCHandler(ip_address="127.0.0.1", port=9000)
|
||
|
||
if local_osc.getIsOscQueryEnabled():
|
||
# MuteSelfパラメータの監視
|
||
mute_status = local_osc.getOSCParameterMuteSelf()
|
||
|
||
if mute_status is not None:
|
||
if mute_status:
|
||
print("ユーザーはミュート中です")
|
||
else:
|
||
print("ユーザーはミュート解除中です")
|
||
else:
|
||
print("MuteSelfパラメータの取得に失敗しました")
|
||
|
||
# カスタムパラメータの監視
|
||
custom_value = local_osc.getOSCParameterValue("/avatar/parameters/CustomParam")
|
||
if custom_value is not None:
|
||
print(f"カスタムパラメータ値: {custom_value}")
|
||
```
|
||
|
||
### 双方向OSC通信の設定
|
||
|
||
```python
|
||
def handle_mute_change(address, *args):
|
||
"""ミュート状態変更のハンドラー"""
|
||
print(f"ミュート状態が変更されました: {args}")
|
||
|
||
def handle_typing_change(address, *args):
|
||
"""タイピング状態変更のハンドラー"""
|
||
print(f"タイピング状態: {args}")
|
||
|
||
def handle_chatbox_input(address, *args):
|
||
"""チャットボックス入力のハンドラー"""
|
||
print(f"チャットボックス入力: {args}")
|
||
|
||
# OSCパラメータハンドラーの設定
|
||
osc_handlers = {
|
||
"/avatar/parameters/MuteSelf": handle_mute_change,
|
||
"/chatbox/typing": handle_typing_change,
|
||
"/chatbox/input": handle_chatbox_input
|
||
}
|
||
|
||
osc = OSCHandler()
|
||
osc.setDictFilterAndTarget(osc_handlers)
|
||
|
||
# OSCサーバー開始（OSCQuery自動公開）
|
||
osc.receiveOscParameters()
|
||
|
||
print("OSC受信サーバーが開始されました")
|
||
print("VRChatからのパラメータ変更を監視中...")
|
||
|
||
# メッセージ送信テスト
|
||
import time
|
||
time.sleep(2)
|
||
osc.sendMessage("双方向通信テスト", notification=True)
|
||
|
||
# 長時間実行
|
||
time.sleep(30)
|
||
|
||
# クリーンアップ
|
||
osc.oscServerStop()
|
||
```
|
||
|
||
### 動的設定変更
|
||
|
||
```python
|
||
# 実行時のIP・ポート変更
|
||
osc = OSCHandler(ip_address="127.0.0.1", port=9000)
|
||
|
||
# 初期設定でローカル接続
|
||
osc.sendMessage("ローカル接続テスト")
|
||
|
||
print("リモート接続に切り替え中...")
|
||
osc.setOscIpAddress("192.168.1.150") # 自動的にOSCQueryが無効化
|
||
osc.sendMessage("リモート接続テスト")
|
||
|
||
print("ポート変更...")
|
||
osc.setOscPort(9001)
|
||
osc.sendMessage("新しいポートでのテスト")
|
||
|
||
print("ローカル接続に戻る...")
|
||
osc.setOscIpAddress("127.0.0.1") # OSCQueryが再度有効化
|
||
osc.sendMessage("ローカル接続復帰テスト")
|
||
```
|
||
|
||
## OSCQuery詳細機能
|
||
|
||
### 自動サービス発見
|
||
|
||
```python
|
||
class VRChatMonitor:
|
||
"""VRChatサービス監視クラス"""
|
||
|
||
def __init__(self):
|
||
self.osc = OSCHandler()
|
||
self.monitoring = False
|
||
|
||
def start_monitoring(self):
|
||
"""VRChatパラメータの継続監視開始"""
|
||
|
||
if not self.osc.getIsOscQueryEnabled():
|
||
print("OSCQuery機能が無効です（ローカル接続のみサポート）")
|
||
return
|
||
|
||
# OSCハンドラー設定
|
||
handlers = {
|
||
"/avatar/parameters/MuteSelf": self.on_mute_change,
|
||
"/avatar/parameters/Voice": self.on_voice_change,
|
||
"/avatar/parameters/Viseme": self.on_viseme_change,
|
||
"/avatar/parameters/GestureLeft": self.on_gesture_left,
|
||
"/avatar/parameters/GestureRight": self.on_gesture_right
|
||
}
|
||
|
||
self.osc.setDictFilterAndTarget(handlers)
|
||
self.osc.receiveOscParameters()
|
||
|
||
self.monitoring = True
|
||
print("VRChatパラメータ監視を開始しました")
|
||
|
||
def on_mute_change(self, address, *args):
|
||
print(f"ミュート状態変更: {args[0] if args else 'Unknown'}")
|
||
|
||
def on_voice_change(self, address, *args):
|
||
print(f"音声レベル: {args[0] if args else 'Unknown'}")
|
||
|
||
def on_viseme_change(self, address, *args):
|
||
print(f"口形変化: {args[0] if args else 'Unknown'}")
|
||
|
||
def on_gesture_left(self, address, *args):
|
||
print(f"左手ジェスチャー: {args[0] if args else 'Unknown'}")
|
||
|
||
def on_gesture_right(self, address, *args):
|
||
print(f"右手ジェスチャー: {args[0] if args else 'Unknown'}")
|
||
|
||
def stop_monitoring(self):
|
||
"""監視停止"""
|
||
self.osc.oscServerStop()
|
||
self.monitoring = False
|
||
print("VRChatパラメータ監視を停止しました")
|
||
|
||
# 使用例
|
||
monitor = VRChatMonitor()
|
||
monitor.start_monitoring()
|
||
|
||
# 監視中に他の処理を実行
|
||
time.sleep(60) # 1分間監視
|
||
|
||
monitor.stop_monitoring()
|
||
```
|
||
|
||
### リアルタイムパラメータ追跡
|
||
|
||
```python
|
||
class ParameterTracker:
|
||
"""パラメータ値の追跡・履歴管理"""
|
||
|
||
def __init__(self, osc_handler):
|
||
self.osc = osc_handler
|
||
self.parameter_history = {}
|
||
self.tracking_active = False
|
||
|
||
def track_parameter(self, address, interval=0.1):
|
||
"""指定されたパラメータを定期監視"""
|
||
|
||
import threading
|
||
|
||
def monitoring_loop():
|
||
while self.tracking_active:
|
||
try:
|
||
value = self.osc.getOSCParameterValue(address)
|
||
if value is not None:
|
||
timestamp = time.time()
|
||
|
||
if address not in self.parameter_history:
|
||
self.parameter_history[address] = []
|
||
|
||
# 値が変更された場合のみ記録
|
||
if (not self.parameter_history[address] or
|
||
self.parameter_history[address][-1][1] != value):
|
||
|
||
self.parameter_history[address].append((timestamp, value))
|
||
print(f"{address}: {value} (時刻: {timestamp:.2f})")
|
||
|
||
# 履歴サイズ制限（最新100件まで）
|
||
if len(self.parameter_history[address]) > 100:
|
||
self.parameter_history[address] = self.parameter_history[address][-100:]
|
||
|
||
time.sleep(interval)
|
||
|
||
except Exception as e:
|
||
print(f"パラメータ追跡エラー: {e}")
|
||
time.sleep(interval)
|
||
|
||
self.tracking_active = True
|
||
thread = threading.Thread(target=monitoring_loop, daemon=True)
|
||
thread.start()
|
||
|
||
def stop_tracking(self):
|
||
"""追跡停止"""
|
||
self.tracking_active = False
|
||
|
||
def get_parameter_history(self, address):
|
||
"""パラメータの履歴取得"""
|
||
return self.parameter_history.get(address, [])
|
||
|
||
def get_latest_value(self, address):
|
||
"""最新パラメータ値取得"""
|
||
history = self.get_parameter_history(address)
|
||
return history[-1][1] if history else None
|
||
|
||
# 使用例
|
||
osc = OSCHandler()
|
||
tracker = ParameterTracker(osc)
|
||
|
||
# MuteSelfパラメータの追跡開始
|
||
tracker.track_parameter("/avatar/parameters/MuteSelf", interval=0.5)
|
||
|
||
# しばらく監視
|
||
time.sleep(30)
|
||
|
||
# 結果確認
|
||
mute_history = tracker.get_parameter_history("/avatar/parameters/MuteSelf")
|
||
print(f"MuteSelf変更履歴: {len(mute_history)}件")
|
||
|
||
for timestamp, value in mute_history[-5:]: # 最新5件表示
|
||
print(f" {time.ctime(timestamp)}: {value}")
|
||
|
||
tracker.stop_tracking()
|
||
```
|
||
|
||
## エラーハンドリング・復旧機構
|
||
|
||
### 堅牢な接続管理
|
||
|
||
```python
|
||
class RobustOSCHandler:
|
||
"""堅牢性を高めたOSCハンドラー"""
|
||
|
||
def __init__(self, ip_address="127.0.0.1", port=9000):
|
||
self.osc = OSCHandler(ip_address, port)
|
||
self.connection_retries = 3
|
||
self.retry_delay = 1.0
|
||
|
||
def safe_send_message(self, message, notification=True, max_retries=None):
|
||
"""安全なメッセージ送信（リトライ機構付き）"""
|
||
|
||
retries = max_retries or self.connection_retries
|
||
|
||
for attempt in range(retries):
|
||
try:
|
||
self.osc.sendMessage(message, notification)
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"送信試行 {attempt + 1}/{retries} 失敗: {e}")
|
||
|
||
if attempt < retries - 1:
|
||
time.sleep(self.retry_delay * (attempt + 1)) # 指数バックオフ
|
||
|
||
# 接続再初期化を試行
|
||
try:
|
||
self.osc.udp_client = udp_client.SimpleUDPClient(
|
||
self.osc.osc_ip_address,
|
||
self.osc.osc_port
|
||
)
|
||
except Exception as reconnect_error:
|
||
print(f"再接続失敗: {reconnect_error}")
|
||
|
||
print(f"メッセージ送信に失敗しました: '{message}'")
|
||
return False
|
||
|
||
def safe_get_parameter(self, address, timeout=5.0):
|
||
"""安全なパラメータ取得（タイムアウト付き）"""
|
||
|
||
if not self.osc.getIsOscQueryEnabled():
|
||
return None
|
||
|
||
import threading
|
||
import queue
|
||
|
||
result_queue = queue.Queue()
|
||
|
||
def parameter_getter():
|
||
try:
|
||
value = self.osc.getOSCParameterValue(address)
|
||
result_queue.put(value)
|
||
except Exception as e:
|
||
result_queue.put(e)
|
||
|
||
# タイムアウト付きでパラメータ取得
|
||
thread = threading.Thread(target=parameter_getter, daemon=True)
|
||
thread.start()
|
||
|
||
try:
|
||
result = result_queue.get(timeout=timeout)
|
||
if isinstance(result, Exception):
|
||
raise result
|
||
return result
|
||
|
||
except queue.Empty:
|
||
print(f"パラメータ取得タイムアウト: {address}")
|
||
return None
|
||
|
||
# 使用例
|
||
robust_osc = RobustOSCHandler()
|
||
|
||
# 堅牢な送信
|
||
success = robust_osc.safe_send_message("堅牢性テスト", notification=True)
|
||
print(f"送信成功: {success}")
|
||
|
||
# 安全なパラメータ取得
|
||
mute_value = robust_osc.safe_get_parameter("/avatar/parameters/MuteSelf", timeout=3.0)
|
||
print(f"MuteSelf値: {mute_value}")
|
||
```
|
||
|
||
## パフォーマンス最適化
|
||
|
||
### 効率的な通信管理
|
||
|
||
```python
|
||
class OptimizedOSCHandler:
|
||
"""パフォーマンス最適化OSCハンドラー"""
|
||
|
||
def __init__(self, ip_address="127.0.0.1", port=9000):
|
||
self.osc = OSCHandler(ip_address, port)
|
||
self.message_queue = []
|
||
self.batch_size = 10
|
||
self.batch_interval = 0.1
|
||
self.last_batch_time = 0
|
||
|
||
def queue_message(self, message, notification=True):
|
||
"""メッセージをキューに追加（バッチ送信用）"""
|
||
|
||
self.message_queue.append((message, notification))
|
||
|
||
# バッチサイズまたは時間間隔でフラッシュ
|
||
current_time = time.time()
|
||
|
||
if (len(self.message_queue) >= self.batch_size or
|
||
current_time - self.last_batch_time >= self.batch_interval):
|
||
self.flush_messages()
|
||
|
||
def flush_messages(self):
|
||
"""キューされたメッセージを一括送信"""
|
||
|
||
if not self.message_queue:
|
||
return
|
||
|
||
# 最新のメッセージのみ送信（重複排除）
|
||
if len(self.message_queue) > 1:
|
||
# 最後のメッセージを優先
|
||
last_message, last_notification = self.message_queue[-1]
|
||
self.osc.sendMessage(last_message, last_notification)
|
||
else:
|
||
message, notification = self.message_queue[0]
|
||
self.osc.sendMessage(message, notification)
|
||
|
||
# キューをクリア
|
||
self.message_queue.clear()
|
||
self.last_batch_time = time.time()
|
||
|
||
def send_immediate(self, message, notification=True):
|
||
"""即座にメッセージ送信（キューをバイパス）"""
|
||
self.flush_messages() # 既存キューを先にフラッシュ
|
||
self.osc.sendMessage(message, notification)
|
||
|
||
# 使用例
|
||
optimized_osc = OptimizedOSCHandler()
|
||
|
||
# 複数のメッセージを効率的に送信
|
||
for i in range(20):
|
||
optimized_osc.queue_message(f"バッチメッセージ {i}")
|
||
time.sleep(0.05) # 短い間隔
|
||
|
||
# 残りのメッセージをフラッシュ
|
||
optimized_osc.flush_messages()
|
||
|
||
# 即座に送信が必要な重要メッセージ
|
||
optimized_osc.send_immediate("緊急メッセージ", notification=True)
|
||
```
|
||
|
||
## 依存関係・要件
|
||
|
||
### 必須依存関係
|
||
- `pythonosc`: 基本OSC通信ライブラリ
|
||
- `threading`: 並行処理制御
|
||
- `time`: 時間管理機能
|
||
|
||
### オプション依存関係
|
||
- `tinyoscquery`: OSCQuery機能（ローカル接続時のみ）
|
||
- `utils`: エラーログ機能（フォールバック処理あり）
|
||
|
||
### システム要件
|
||
```python
|
||
# 最小システム要件
|
||
requirements = {
|
||
"python_version": "3.7+",
|
||
"network": "UDP通信対応",
|
||
"vrchat_version": "OSCサポート版（2022年8月以降）",
|
||
"local_ports": "空きUDP/TCPポート（OSCQuery使用時）"
|
||
}
|
||
|
||
# 推奨環境
|
||
recommended = {
|
||
"network_latency": "< 10ms（ローカル接続）",
|
||
"cpu_usage": "OSCQuery使用時は追加CPU負荷",
|
||
"memory": "tinyoscquery使用時は追加メモリ"
|
||
}
|
||
```
|
||
|
||
## 注意事項・制限
|
||
|
||
### OSCQuery制限
|
||
- ローカルホスト（127.0.0.1/localhost）接続時のみ利用可能
|
||
- tinyoscqueryライブラリが必要
|
||
- ファイアウォール設定によっては動作しない可能性
|
||
|
||
### 通信制限
|
||
- UDPプロトコルのため送達保証なし
|
||
- VRChatのOSC受信制限（レート制限あり）
|
||
- ネットワーク環境による遅延・パケット loss
|
||
|
||
### プラットフォーム依存
|
||
```python
|
||
# 既知の制限事項
|
||
limitations = {
|
||
"windows": "Windowsファイアウォールの設定が必要な場合あり",
|
||
"macos": "セキュリティ設定によるポート制限の可能性",
|
||
"linux": "一部のLinuxディストリビューションでの互換性問題",
|
||
"vrchat_platform": "PC版VRChatのみOSCサポート"
|
||
}
|
||
```
|
||
|
||
## 関連モジュール
|
||
|
||
- `config.py`: OSC設定管理
|
||
- `controller.py`: OSC機能制御インターフェース
|
||
- `model.py`: OSC機能統合
|
||
- `utils.py`: エラーログ・ネットワークユーティリティ
|
||
|
||
## 将来の改善点
|
||
|
||
- より高度なOSCQueryパラメータ監視
|
||
- カスタムOSCプロトコル拡張
|
||
- パフォーマンス監視・分析機能
|
||
- 自動再接続・復旧機構の改善
|
||
- VRChatアバター固有パラメータ対応 |