63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
# whisper_local/microphone/_poll.py
|
|
"""Polling-basierter Mikrofon-Monitor (cross-platform)."""
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
import sounddevice as sd
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PollMonitor:
|
|
def __init__(self, configured_device: str | None, interval: float = 2.5):
|
|
self.configured_device = configured_device
|
|
self.interval = interval
|
|
self.on_device_added: Callable[[str], Awaitable[None]] | None = None
|
|
self.on_device_removed: Callable[[str], Awaitable[None]] | None = None
|
|
self.on_configured_missing: Callable[[], Awaitable[None]] | None = None
|
|
self._task: asyncio.Task | None = None
|
|
self._known_devices: set[str] = set()
|
|
|
|
def _current_devices(self) -> set[str]:
|
|
try:
|
|
return {
|
|
dev["name"]
|
|
for dev in sd.query_devices()
|
|
if dev["max_input_channels"] > 0
|
|
}
|
|
except Exception:
|
|
logger.exception("Fehler beim Abfragen der Audiogeräte")
|
|
return self._known_devices.copy()
|
|
|
|
async def start(self) -> None:
|
|
self._known_devices = self._current_devices()
|
|
if (
|
|
self.configured_device
|
|
and self.configured_device not in self._known_devices
|
|
and self.on_configured_missing
|
|
):
|
|
await self.on_configured_missing()
|
|
self._task = asyncio.create_task(self._loop())
|
|
|
|
def stop(self) -> None:
|
|
if self._task is not None:
|
|
self._task.cancel()
|
|
self._task = None
|
|
|
|
async def _loop(self) -> None:
|
|
while True:
|
|
await asyncio.sleep(self.interval)
|
|
current = self._current_devices()
|
|
added = current - self._known_devices
|
|
removed = self._known_devices - current
|
|
self._known_devices = current
|
|
|
|
for name in added:
|
|
if self.on_device_added:
|
|
await self.on_device_added(name)
|
|
|
|
for name in removed:
|
|
if self.on_device_removed:
|
|
await self.on_device_removed(name)
|