135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
"""Entry-Point für whisper-local."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
|
|
from whisper_local.config import Config, load_config
|
|
from whisper_local.hotkey import create_listener
|
|
from whisper_local.inserter import create_inserter
|
|
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,
|
|
)
|
|
if sys.platform == "win32":
|
|
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(),
|
|
)
|
|
else:
|
|
_preloaded_model = None
|
|
|
|
self.transcriber = Transcriber(
|
|
model_name=config.whisper_model,
|
|
compute_type=config.compute_type,
|
|
language=config.language,
|
|
model=_preloaded_model,
|
|
)
|
|
self.inserter = create_inserter()
|
|
self.hotkey = create_listener(key_name=config.hotkey)
|
|
self.hotkey.on_press = self.on_press
|
|
self.hotkey.on_release = self.on_release
|
|
self.tray = create_tray(on_settings=self._open_settings, on_quit=self._quit)
|
|
|
|
async def on_press(self) -> None:
|
|
"""Callback: Hotkey gedrückt — Aufnahme starten."""
|
|
logger.info("Aufnahme startet...")
|
|
self.tray.set_state(AppState.RECORDING)
|
|
self.recorder.start()
|
|
|
|
async def on_release(self) -> None:
|
|
"""Callback: Hotkey losgelassen — Aufnahme stoppen, transkribieren, einfügen."""
|
|
audio = self.recorder.stop()
|
|
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()
|
|
|
|
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,
|
|
)
|
|
if self._loop is not None:
|
|
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())
|
|
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()
|