"""Entry-Point für whisper-local.""" import asyncio import logging import sys from whisper_local.config import Config, load_config from whisper_local.microphone import create_monitor from whisper_local.hotkey import create_listener from whisper_local.inserter import create_inserter from whisper_local.media import create_media_controller from whisper_local.recorder import Recorder from whisper_local.transcriber import Transcriber from whisper_local.tray import AppState, create_tray logger = logging.getLogger(__name__) class App: def __init__(self, config: Config | None = None): if config is None: config = load_config() self._config = config self._loop: asyncio.AbstractEventLoop | None = None self._hotkey_task: asyncio.Task | None = None self._quit_event: asyncio.Event | None = None self.recorder = Recorder( sample_rate=config.sample_rate, channels=config.channels, min_duration=config.min_duration, device=config.microphone or None, ) from whisper_local.tray._download_progress import load_model_with_progress from whisper_local.transcriber import _model_cache_dir _preloaded_model = load_model_with_progress( model_name=config.whisper_model, compute_type=config.compute_type, download_root=_model_cache_dir(), ) self.transcriber = Transcriber( model_name=config.whisper_model, compute_type=config.compute_type, language=config.language, model=_preloaded_model, ) self.inserter = create_inserter() self.media = create_media_controller( enabled=config.pause_media_during_recording ) self.hotkey = create_listener(key_name=config.hotkey) self.hotkey.on_press = self.on_press self.hotkey.on_release = self.on_release self.monitor = create_monitor(config.microphone or None) self.monitor.on_device_added = self._on_microphone_added self.monitor.on_device_removed = self._on_microphone_removed self.monitor.on_configured_missing = self._on_configured_microphone_missing self.tray = create_tray(on_settings=self._open_settings, on_quit=self._quit) async def on_press(self) -> None: """Callback: Hotkey gedrückt — Medien pausieren + Aufnahme starten.""" logger.info("Aufnahme startet...") self.tray.set_state(AppState.RECORDING) await self.media.pause() self.recorder.start() async def on_release(self) -> None: """Callback: Hotkey losgelassen — Aufnahme stoppen, Medien fortsetzen, transkribieren, einfügen.""" try: audio = self.recorder.stop() finally: try: await self.media.resume() except Exception as e: logger.warning("Fehler beim Fortsetzen der Medienwiedergabe: %s", e) if audio is None: logger.info("Keine Audio-Daten, übersprungen") self.tray.set_state(AppState.WAITING) return logger.info("Transkribiere...") self.tray.set_state(AppState.TRANSCRIBING) text = self.transcriber.transcribe(audio) if text: await self.inserter.insert(text) self.tray.set_state(AppState.WAITING) def _quit(self) -> None: """Beendet die Anwendung sauber.""" if self._loop is not None and self._quit_event is not None: self._loop.call_soon_threadsafe(self._quit_event.set) def _open_settings(self) -> None: """Öffnet den Einstellungs-Dialog.""" from whisper_local.tray._settings import SettingsDialog SettingsDialog(config=self._config, on_save=self._on_config_reload).open() async def _on_configured_microphone_missing(self) -> None: """Konfiguriertes Mikrofon nicht gefunden — auf Standard wechseln.""" from whisper_local.tray._notification import notify device_name = self._config.microphone or "Mikrofon" logger.warning("Konfiguriertes Mikrofon '%s' nicht gefunden, nutze Standard", device_name) if self.recorder.is_recording: self.recorder.stop() self.recorder = Recorder( sample_rate=self._config.sample_rate, channels=self._config.channels, min_duration=self._config.min_duration, device=None, ) notify( "Mikrofon nicht gefunden", f'„{device_name}“ ist nicht verfügbar. Standard-Mikrofon wird verwendet.', ) self.tray.set_warning("Mikrofon nicht gefunden") async def _on_microphone_added(self, device_name: str) -> None: """Neues Mikrofon erkannt — konfiguriertes Gerät ggf. wiederherstellen.""" if device_name != self._config.microphone: return from whisper_local.tray._notification import notify logger.info("Konfiguriertes Mikrofon '%s' wieder verfügbar", device_name) self.recorder = Recorder( sample_rate=self._config.sample_rate, channels=self._config.channels, min_duration=self._config.min_duration, device=self._config.microphone or None, ) notify("Mikrofon verbunden", f'„{device_name}" ist wieder verfügbar.') self.tray.set_warning(None) async def _on_microphone_removed(self, device_name: str) -> None: """Mikrofon entfernt — konfiguriertes Gerät → Fallback auslösen.""" logger.info("Mikrofon entfernt: %s", device_name) if device_name == self._config.microphone: await self._on_configured_microphone_missing() def _on_config_reload(self, new_config: Config) -> None: """Übernimmt neue Konfiguration ohne App-Neustart.""" self._config = new_config self.recorder = Recorder( sample_rate=new_config.sample_rate, channels=new_config.channels, min_duration=new_config.min_duration, device=new_config.microphone or None, ) self.monitor.stop() self.monitor = create_monitor(new_config.microphone or None) self.monitor.on_device_added = self._on_microphone_added self.monitor.on_device_removed = self._on_microphone_removed self.monitor.on_configured_missing = self._on_configured_microphone_missing if self._loop is not None: asyncio.run_coroutine_threadsafe(self.monitor.start(), self._loop) self.tray.set_warning(None) old_media = self.media self.media = create_media_controller( enabled=new_config.pause_media_during_recording ) if self._loop is not None: asyncio.run_coroutine_threadsafe(old_media.resume(), self._loop) asyncio.run_coroutine_threadsafe( self._restart_hotkey(new_config.hotkey), self._loop ) async def _restart_hotkey(self, key_name: str) -> None: """Stoppt den alten Hotkey-Listener und startet einen neuen.""" self.hotkey.stop() await asyncio.sleep(0.1) self.hotkey = create_listener(key_name=key_name) self.hotkey.on_press = self.on_press self.hotkey.on_release = self.on_release self._hotkey_task = asyncio.create_task(self.hotkey.listen()) async def run(self) -> None: """Startet den Hauptloop.""" self._loop = asyncio.get_running_loop() self._quit_event = asyncio.Event() logger.info("whisper-local gestartet, warte auf Hotkey...") self.tray.start() self._hotkey_task = asyncio.create_task(self.hotkey.listen()) asyncio.create_task(self.monitor.start()) await self._quit_event.wait() def main(): logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) app = App() try: asyncio.run(app.run()) except KeyboardInterrupt: logger.info("Beendet") sys.exit(0) if __name__ == "__main__": main()