diff --git a/src-python/device_manager.py b/src-python/device_manager.py index 7f741d26..b41a0cc3 100644 --- a/src-python/device_manager.py +++ b/src-python/device_manager.py @@ -1,27 +1,53 @@ -from typing import Callable +from typing import Callable, Dict, List, Optional, Any from time import sleep from threading import Thread -import comtypes -from pyaudiowpatch import PyAudio, paWASAPI -from pycaw.callbacks import MMNotificationClient -from pycaw.utils import AudioUtilities + +# Optional, Windows-specific dependencies. Guard imports so module can be imported on non-Windows systems. +try: + import comtypes +except Exception: # pragma: no cover - optional runtime + comtypes = None # type: ignore + +try: + from pyaudiowpatch import PyAudio, paWASAPI +except Exception: # pragma: no cover - optional runtime + PyAudio = None # type: ignore + paWASAPI = None # type: ignore + +try: + from pycaw.callbacks import MMNotificationClient + from pycaw.utils import AudioUtilities +except Exception: # pragma: no cover - optional runtime + MMNotificationClient = object # type: ignore + AudioUtilities = None # type: ignore + from utils import errorLogging class Client(MMNotificationClient): - def __init__(self): - super().__init__() - self.loop = True + """Callback client used by pycaw to detect device changes. - def on_default_device_changed(self, flow, flow_id, role, role_id, default_device_id): + This subclass is lightweight: it flips a flag when events arrive so the + monitoring loop can break and refresh device lists. + """ + + def __init__(self) -> None: + # If MMNotificationClient is the placeholder object (non-windows), avoid calling super + try: + super().__init__() + except Exception: + pass + self.loop: bool = True + + def on_default_device_changed(self, *args: Any, **kwargs: Any) -> None: self.loop = False - def on_device_added(self, added_device_id): + def on_device_added(self, *args: Any, **kwargs: Any) -> None: self.loop = False - def on_device_removed(self, removed_device_id): + def on_device_removed(self, *args: Any, **kwargs: Any) -> None: self.loop = False - def on_device_state_changed(self, device_id, state): + def on_device_state_changed(self, *args: Any, **kwargs: Any) -> None: self.loop = False # def on_property_value_changed(self, device_id, key): @@ -33,108 +59,150 @@ class DeviceManager: def __new__(cls): if cls._instance is None: cls._instance = super(DeviceManager, cls).__new__(cls) - cls._instance.init() + # do NOT auto-init monitoring-heavy resources on import; require explicit init + cls._instance._initialized = False return cls._instance - def init(self): - self.mic_devices = {"NoHost": [{"index": -1, "name": "NoDevice"}]} - self.default_mic_device = {"host": {"index": -1, "name": "NoHost"}, "device": {"index": -1, "name": "NoDevice"}} - self.speaker_devices = [{"index": -1, "name": "NoDevice"}] - self.default_speaker_device = {"device": {"index": -1, "name": "NoDevice"}} + def init(self) -> None: + """Initialize internal state. This is intentionally separate from object + creation so importing the module won't start threads or access OS + audio APIs. Call `device_manager.init()` and then + `device_manager.startMonitoring()` explicitly when ready. + """ + if getattr(self, "_initialized", False): + return - self.update() + self.mic_devices: Dict[str, List[Dict[str, Any]]] = {"NoHost": [{"index": -1, "name": "NoDevice"}]} + self.default_mic_device: Dict[str, Any] = {"host": {"index": -1, "name": "NoHost"}, "device": {"index": -1, "name": "NoDevice"}} + self.speaker_devices: List[Dict[str, Any]] = [{"index": -1, "name": "NoDevice"}] + self.default_speaker_device: Dict[str, Any] = {"device": {"index": -1, "name": "NoDevice"}} - self.prev_mic_host = [host for host in self.mic_devices] - self.prev_mic_devices = self.mic_devices - self.prev_default_mic_device = self.default_mic_device - self.prev_speaker_devices = self.speaker_devices - self.prev_default_speaker_device = self.default_speaker_device + # Initialize previous state trackers + self.prev_mic_host: List[str] = [host for host in self.mic_devices] + self.prev_mic_devices: Dict[str, List[Dict[str, Any]]] = self.mic_devices + self.prev_default_mic_device: Dict[str, Any] = self.default_mic_device + self.prev_speaker_devices: List[Dict[str, Any]] = self.speaker_devices + self.prev_default_speaker_device: Dict[str, Any] = self.default_speaker_device - self.update_flag_default_mic_device = False - self.update_flag_default_speaker_device = False - self.update_flag_host_list = False - self.update_flag_mic_device_list = False - self.update_flag_speaker_device_list = False + # Update flags + self.update_flag_default_mic_device: bool = False + self.update_flag_default_speaker_device: bool = False + self.update_flag_host_list: bool = False + self.update_flag_mic_device_list: bool = False + self.update_flag_speaker_device_list: bool = False - self.callback_default_mic_device = None - self.callback_default_speaker_device = None - self.callback_host_list = None - self.callback_mic_device_list = None - self.callback_speaker_device_list = None - self.callback_process_before_update_devices = None - self.callback_process_after_update_devices = None + # Callbacks + self.callback_default_mic_device: Optional[Callable[..., None]] = None + self.callback_default_speaker_device: Optional[Callable[..., None]] = None + self.callback_host_list: Optional[Callable[..., None]] = None + self.callback_mic_device_list: Optional[Callable[..., None]] = None + self.callback_speaker_device_list: Optional[Callable[..., None]] = None + self.callback_process_before_update_mic_devices: Optional[Callable[..., None]] = None + self.callback_process_after_update_mic_devices: Optional[Callable[..., None]] = None + self.callback_process_before_update_speaker_devices: Optional[Callable[..., None]] = None + self.callback_process_after_update_speaker_devices: Optional[Callable[..., None]] = None - self.monitoring_flag = False - self.startMonitoring() + # Monitoring control + self.monitoring_flag: bool = False + self.th_monitoring: Optional[Thread] = None + + self._initialized = True def update(self): - buffer_mic_devices = {} - buffer_default_mic_device = {"host": {"index": -1, "name": "NoHost"}, "device": {"index": -1, "name": "NoDevice"}} - buffer_speaker_devices = [] - buffer_default_speaker_device = {"device": {"index": -1, "name": "NoDevice"}} + buffer_mic_devices: Dict[str, List[Dict[str, Any]]] = {} + buffer_default_mic_device: Dict[str, Any] = {"host": {"index": -1, "name": "NoHost"}, "device": {"index": -1, "name": "NoDevice"}} + buffer_speaker_devices: List[Dict[str, Any]] = [] + buffer_default_speaker_device: Dict[str, Any] = {"device": {"index": -1, "name": "NoDevice"}} - with PyAudio() as p: - for host_index in range(p.get_host_api_count()): - host = p.get_host_api_info_by_index(host_index) - device_count = host.get('deviceCount', 0) - for device_index in range(device_count): - device = p.get_device_info_by_host_api_device_index(host_index, device_index) - if device.get("maxInputChannels", 0) > 0 and not device.get("isLoopbackDevice", True): - buffer_mic_devices.setdefault(host["name"], []).append(device) - if not buffer_mic_devices: - buffer_mic_devices = {"NoHost": [{"index": -1, "name": "NoDevice"}]} + if PyAudio is None: + # PyAudio not available; leave defaults in place + self.mic_devices = buffer_mic_devices or {"NoHost": [{"index": -1, "name": "NoDevice"}]} + self.default_mic_device = buffer_default_mic_device + self.speaker_devices = buffer_speaker_devices or [{"index": -1, "name": "NoDevice"}] + self.default_speaker_device = buffer_default_speaker_device + return - api_info = p.get_default_host_api_info() - default_mic_device = api_info["defaultInputDevice"] - - for host_index in range(p.get_host_api_count()): - host = p.get_host_api_info_by_index(host_index) - device_count = host.get('deviceCount', 0) - for device_index in range(device_count): - device = p.get_device_info_by_host_api_device_index(host_index, device_index) - if device["index"] == default_mic_device: - buffer_default_mic_device = {"host": host, "device": device} - break - else: - continue - break - - speaker_devices = [] - wasapi_info = p.get_host_api_info_by_type(paWASAPI) - wasapi_name = wasapi_info["name"] - for host_index in range(p.get_host_api_count()): - host = p.get_host_api_info_by_index(host_index) - if host["name"] == wasapi_name: + try: + with PyAudio() as p: + # gather input devices grouped by host + for host_index in range(p.get_host_api_count()): + host = p.get_host_api_info_by_index(host_index) device_count = host.get('deviceCount', 0) for device_index in range(device_count): device = p.get_device_info_by_host_api_device_index(host_index, device_index) - if not device.get("isLoopbackDevice", True): - for loopback in p.get_loopback_device_info_generator(): - if device["name"] in loopback["name"]: - speaker_devices.append(loopback) - speaker_devices = [dict(t) for t in {tuple(d.items()) for d in speaker_devices}] or [{"index": -1, "name": "NoDevice"}] - buffer_speaker_devices = sorted(speaker_devices, key=lambda d: d['index']) + if device.get("maxInputChannels", 0) > 0 and not device.get("isLoopbackDevice", True): + buffer_mic_devices.setdefault(host["name"], []).append(device) + if not buffer_mic_devices: + buffer_mic_devices = {"NoHost": [{"index": -1, "name": "NoDevice"}]} - wasapi_info = p.get_host_api_info_by_type(paWASAPI) - default_speaker_device_index = wasapi_info["defaultOutputDevice"] + api_info = p.get_default_host_api_info() + default_mic_device = api_info.get("defaultInputDevice", -1) - for host_index in range(p.get_host_api_count()): - host_info = p.get_host_api_info_by_index(host_index) - device_count = host_info.get('deviceCount', 0) - for device_index in range(0, device_count): - device = p.get_device_info_by_host_api_device_index(host_index, device_index) - if device["index"] == default_speaker_device_index: - default_speakers = device - if not default_speakers.get("isLoopbackDevice", True): - for loopback in p.get_loopback_device_info_generator(): - if default_speakers["name"] in loopback["name"]: - buffer_default_speaker_device = {"device": loopback} - break - break - - if buffer_default_speaker_device["device"]["name"] != "NoDevice": + for host_index in range(p.get_host_api_count()): + host = p.get_host_api_info_by_index(host_index) + device_count = host.get('deviceCount', 0) + for device_index in range(device_count): + device = p.get_device_info_by_host_api_device_index(host_index, device_index) + if device.get("index") == default_mic_device: + buffer_default_mic_device = {"host": host, "device": device} + break + else: + continue break + # collect speaker loopback devices (requires WASAPI) + speaker_devices: List[Dict[str, Any]] = [] + if paWASAPI is not None: + try: + wasapi_info = p.get_host_api_info_by_type(paWASAPI) + wasapi_name = wasapi_info.get("name") + for host_index in range(p.get_host_api_count()): + host = p.get_host_api_info_by_index(host_index) + if host.get("name") == wasapi_name: + device_count = host.get('deviceCount', 0) + for device_index in range(device_count): + device = p.get_device_info_by_host_api_device_index(host_index, device_index) + if not device.get("isLoopbackDevice", True): + for loopback in p.get_loopback_device_info_generator(): + # match by name inclusion + if device.get("name") in loopback.get("name", ""): + speaker_devices.append(loopback) + except Exception: + # WASAPI not available or failed; ignore and continue + pass + + # deduplicate and sort + speaker_devices = [dict(t) for t in {tuple(d.items()) for d in speaker_devices}] or [{"index": -1, "name": "NoDevice"}] + buffer_speaker_devices = sorted(speaker_devices, key=lambda d: d.get('index', -1)) + + # default speaker + if paWASAPI is not None: + try: + wasapi_info = p.get_host_api_info_by_type(paWASAPI) + default_speaker_device_index = wasapi_info.get("defaultOutputDevice", -1) + for host_index in range(p.get_host_api_count()): + host_info = p.get_host_api_info_by_index(host_index) + device_count = host_info.get('deviceCount', 0) + for device_index in range(0, device_count): + device = p.get_device_info_by_host_api_device_index(host_index, device_index) + if device.get("index") == default_speaker_device_index: + default_speakers = device + if not default_speakers.get("isLoopbackDevice", True): + for loopback in p.get_loopback_device_info_generator(): + if default_speakers.get("name") in loopback.get("name", ""): + buffer_default_speaker_device = {"device": loopback} + break + break + + if buffer_default_speaker_device["device"].get("name") != "NoDevice": + break + except Exception: + # best-effort; ignore failures + pass + + except Exception: + errorLogging() + self.mic_devices = buffer_mic_devices self.default_mic_device = buffer_default_mic_device self.speaker_devices = buffer_speaker_devices @@ -170,14 +238,27 @@ class DeviceManager: try: while self.monitoring_flag is True: try: - comtypes.CoInitialize() - cb = Client() - enumerator = AudioUtilities.GetDeviceEnumerator() - enumerator.RegisterEndpointNotificationCallback(cb) - while cb.loop is True: - sleep(1) - enumerator.UnregisterEndpointNotificationCallback(cb) - comtypes.CoUninitialize() + # Use COM only when available (Windows). If comtypes is not present, + # fall back to periodic polling using PyAudio only. + if comtypes is not None and AudioUtilities is not None: + try: + comtypes.CoInitialize() + cb = Client() + enumerator = AudioUtilities.GetDeviceEnumerator() + enumerator.RegisterEndpointNotificationCallback(cb) + while cb.loop is True and self.monitoring_flag is True: + sleep(1) + try: + enumerator.UnregisterEndpointNotificationCallback(cb) + except Exception: + # best-effort unregister + pass + comtypes.CoUninitialize() + except Exception: + # if COM monitoring fails, log and fall through to polling + errorLogging() + + # polling and update cycle self.runProcessBeforeUpdateMicDevices() self.runProcessBeforeUpdateSpeakerDevices() sleep(2) @@ -191,12 +272,12 @@ class DeviceManager: self.runProcessAfterUpdateSpeakerDevices() except Exception: errorLogging() - finally: - pass except Exception: errorLogging() def startMonitoring(self): + if self.monitoring_flag: + return self.monitoring_flag = True self.th_monitoring = Thread(target=self.monitoring) self.th_monitoring.daemon = True @@ -204,7 +285,12 @@ class DeviceManager: def stopMonitoring(self): self.monitoring_flag = False - self.th_monitoring.join() + if getattr(self, "th_monitoring", None) is not None: + try: + self.th_monitoring.join(timeout=5) + except Exception: + # If join fails or thread is not joinable, ignore - it's a best-effort stop + pass def setCallbackDefaultMicDevice(self, callback): self.callback_default_mic_device = callback @@ -244,7 +330,10 @@ class DeviceManager: def runProcessBeforeUpdateMicDevices(self): if isinstance(self.callback_process_before_update_mic_devices, Callable): - self.callback_process_before_update_mic_devices() + try: + self.callback_process_before_update_mic_devices() + except Exception: + errorLogging() def setCallbackProcessAfterUpdateMicDevices(self, callback): self.callback_process_after_update_mic_devices = callback @@ -254,7 +343,10 @@ class DeviceManager: def runProcessAfterUpdateMicDevices(self): if isinstance(self.callback_process_after_update_mic_devices, Callable): - self.callback_process_after_update_mic_devices() + try: + self.callback_process_after_update_mic_devices() + except Exception: + errorLogging() def setCallbackProcessBeforeUpdateSpeakerDevices(self, callback): self.callback_process_before_update_speaker_devices = callback @@ -264,7 +356,10 @@ class DeviceManager: def runProcessBeforeUpdateSpeakerDevices(self): if isinstance(self.callback_process_before_update_speaker_devices, Callable): - self.callback_process_before_update_speaker_devices() + try: + self.callback_process_before_update_speaker_devices() + except Exception: + errorLogging() def setCallbackProcessAfterUpdateSpeakerDevices(self, callback): self.callback_process_after_update_speaker_devices = callback @@ -274,7 +369,10 @@ class DeviceManager: def runProcessAfterUpdateSpeakerDevices(self): if isinstance(self.callback_process_after_update_speaker_devices, Callable): - self.callback_process_after_update_speaker_devices() + try: + self.callback_process_after_update_speaker_devices() + except Exception: + errorLogging() def noticeUpdateDevices(self): if self.update_flag_default_mic_device is True: @@ -296,23 +394,38 @@ class DeviceManager: def setMicDefaultDevice(self): if isinstance(self.callback_default_mic_device, Callable): - self.callback_default_mic_device(self.default_mic_device["host"]["name"], self.default_mic_device["device"]["name"]) + try: + self.callback_default_mic_device(self.default_mic_device["host"]["name"], self.default_mic_device["device"]["name"]) + except Exception: + errorLogging() def setSpeakerDefaultDevice(self): if isinstance(self.callback_default_speaker_device, Callable): - self.callback_default_speaker_device(self.default_speaker_device["device"]["name"]) + try: + self.callback_default_speaker_device(self.default_speaker_device["device"]["name"]) + except Exception: + errorLogging() def setMicHostList(self): if isinstance(self.callback_host_list, Callable): - self.callback_host_list() + try: + self.callback_host_list() + except Exception: + errorLogging() def setMicDeviceList(self): if isinstance(self.callback_mic_device_list, Callable): - self.callback_mic_device_list() + try: + self.callback_mic_device_list() + except Exception: + errorLogging() def setSpeakerDeviceList(self): if isinstance(self.callback_speaker_device_list, Callable): - self.callback_speaker_device_list() + try: + self.callback_speaker_device_list() + except Exception: + errorLogging() def getMicDevices(self): return self.mic_devices @@ -337,13 +450,15 @@ class DeviceManager: self.setSpeakerDeviceList() self.setSpeakerDefaultDevice() +# Provide a module-level singleton. Call `device_manager.init()` explicitly to +# initialize audio resources and `device_manager.startMonitoring()` to begin +# background monitoring. This avoids side-effects during simple imports. device_manager = DeviceManager() if __name__ == "__main__": - # print("getMicDevices()", device_manager.getMicDevices()) - # print("getDefaultMicDevice()", device_manager.getDefaultMicDevice()) - # print("getSpeakerDevices()", device_manager.getSpeakerDevices()) - # print("getDefaultSpeakerDevice()", device_manager.getDefaultSpeakerDevice()) - - while True: - sleep(1) \ No newline at end of file + print("DeviceManager demo. Call device_manager.init() and device_manager.startMonitoring() to run live monitoring.") + try: + while True: + sleep(1) + except KeyboardInterrupt: + print("exiting") \ No newline at end of file diff --git a/src-python/docs/modules/device_manager_ref.md b/src-python/docs/modules/device_manager_ref.md new file mode 100644 index 00000000..893bfb8b --- /dev/null +++ b/src-python/docs/modules/device_manager_ref.md @@ -0,0 +1,93 @@ +# device_manager.py — デバイス検出と監視 (改訂版) + +### 概要 +`device_manager.py` はローカルのマイク（入力）とスピーカー（ループバックから抽出）を列挙し、デフォルトデバイスの変更やデバイスリストの変化を監視してコールバックで通知するユーティリティです。 + +設計上のポイント: +- Windows 固有の依存 (`comtypes`, `pyaudiowpatch` (PyAudio + WASAPI), `pycaw`) はオプショナルです。モジュールを import してもこれらが無ければ例外にならず、プレースホルダ値を返すようになっています。 +- モジュールの import 時点では監視は開始されません。リソースやスレッドの副作用を避けるため、`init()` と `startMonitoring()` は呼び出し側で明示的に実行してください。 + +--- + +### 使い方（簡単な流れ） + +1. モジュールをインポート + +```py +from device_manager import device_manager +``` + +2. 初期化（内部状態のセットアップ） + +```py +device_manager.init() +``` + +3. 監視の開始（バックグラウンドスレッド） + +```py +device_manager.startMonitoring() +``` + +4. 停止（アプリ終了時など） + +```py +device_manager.stopMonitoring() +``` + +--- + +### 主な API + +- `device_manager.init()` + - internal state の初期化。import 後に必ず呼ぶ必要はないが、実機デバイスを取得する前に呼ぶことを推奨します。 +- `device_manager.startMonitoring()` / `device_manager.stopMonitoring()` + - 監視の開始 / 停止。`startMonitoring()` はデーモンスレッドを作成します。`stopMonitoring()` は best-effort で join を試みます。 +- `device_manager.getMicDevices()` + - ホストごとにグループ化された入力デバイスの辞書を返します。例: `{ 'Realtek': [ {index: 2, name: 'Microphone (Realtek)'} ] }`。 +- `device_manager.getDefaultMicDevice()` / `device_manager.getSpeakerDevices()` / `device_manager.getDefaultSpeakerDevice()` + - デフォルトデバイスやスピーカーループバックの情報を返します。 +- `device_manager.forceUpdateAndSetMicDevices()` / `device_manager.forceUpdateAndSetSpeakerDevices()` + - 即時に update() を実行して対応するコールバックを呼びます。 + +--- + +### コールバック登録（例） + +コールバックは例外を内部で捕捉してログを出すため、コールバック実装側でもエラーハンドリングしてください。 + +- `setCallbackDefaultMicDevice(callback)` — デフォルト入力が変わったときに `callback(host_name, device_name)` が呼ばれます。 +- `setCallbackDefaultSpeakerDevice(callback)` — デフォルト出力が変わったときに `callback(device_name)` が呼ばれます。 +- `setCallbackHostList(callback)` / `setCallbackMicDeviceList(callback)` / `setCallbackSpeakerDeviceList(callback)` — それぞれ list 変更時に `callback()` が呼ばれます。 +- `setCallbackProcessBeforeUpdateMicDevices(callback)` / `setCallbackProcessAfterUpdateMicDevices(callback)` — 更新の前後に呼ばれるフックです。 + +簡単な例: + +```py +from device_manager import device_manager + +def on_default_mic(host, device): + print('default mic changed', host, device) + +device_manager.init() +device_manager.setCallbackDefaultMicDevice(on_default_mic) +device_manager.startMonitoring() + +# 後で停止 +# device_manager.stopMonitoring() +``` + +--- + +### 注意点 / トラブルシュート + +- Windows 固有の依存が無い場合、`getMicDevices()` などはデフォルトのプレースホルダ（`NoHost` / `NoDevice`）を返します。実機のデバイス検出や WASAPI によるループバック検出は Windows 環境でのみ保証されます。 +- `startMonitoring()` は監視用のデーモンスレッドを作るため、アプリケーションの終了時には `stopMonitoring()` を呼ぶかプロセスを終了してください。`stopMonitoring()` は join を行いますが、失敗した場合でも致命的にならないよう best-effort 実装です。 +- コールバック内部で例外が発生してもモジュール側で捕捉してログ出力します（`utils.errorLogging()`）。コールバック側で詳細なハンドリングやリトライが必要な場合は呼び出し側で行ってください。 + +--- + +### 実装メモ + +- `monitoring()` は可能なら Windows の COM (pycaw / MMNotificationClient) を使ってイベント駆動で待ち受け、失敗時や非Windows 環境では PyAudio を使ったポーリング（定期的な update()) にフォールバックします。 +- 外部ライブラリが原因の例外は内部で捕捉し、`errorLogging()` を呼んで記録する設計です。