whisper-local/whisper_local/__main__.py

"""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()