"""Voice-Agent Windows-Diktat — Push-to-Talk-Client. Hotkey halten = aufnehmen, loslassen = an Mac-Worker schicken, Ergebnis am Cursor einfuegen (Zwischenablage + Strg+V). Konfig: config.toml neben dem Script (siehe config.example.toml). """ from __future__ import annotations import io import logging import sys import threading import time import tomllib import wave from pathlib import Path import keyboard import numpy as np import pyperclip import requests import sounddevice as sd from PIL import Image, ImageDraw from pystray import Icon, Menu, MenuItem SAMPLE_RATE = 16000 CHANNELS = 1 DTYPE = "int16" MIN_SECONDS = 0.2 MAX_SECONDS = 120 HERE = Path(__file__).resolve().parent CONFIG_PATH = HERE / "config.toml" EXAMPLE_PATH = HERE / "config.example.toml" log = logging.getLogger("voice-agent.dictate") def load_config() -> dict: path = CONFIG_PATH if CONFIG_PATH.exists() else EXAMPLE_PATH if not path.exists(): raise SystemExit( f"Weder {CONFIG_PATH.name} noch {EXAMPLE_PATH.name} gefunden in {HERE}." ) with path.open("rb") as f: cfg = tomllib.load(f) if path == EXAMPLE_PATH: log.warning( "Nutze config.example.toml als Fallback — bitte nach config.toml kopieren." ) return cfg class Recorder: """Sammelt 16-kHz-Mono-Audio waehrend die Hotkey-Taste gehalten wird.""" def __init__(self, sample_rate: int = SAMPLE_RATE): self.sample_rate = sample_rate self._frames: list[np.ndarray] = [] self._stream: sd.InputStream | None = None self._lock = threading.Lock() self.recording = False def _callback(self, indata, _frames, _time_info, status) -> None: if status: log.debug("audio status: %s", status) with self._lock: self._frames.append(indata.copy()) if len(self._frames) * indata.shape[0] >= self.sample_rate * MAX_SECONDS: log.warning("MAX_SECONDS erreicht — Aufnahme wird automatisch beendet") self._stop_stream() def _stop_stream(self) -> None: if self._stream is not None: try: self._stream.stop() self._stream.close() except Exception: # noqa: BLE001 pass self._stream = None self.recording = False def start(self) -> None: if self.recording: return with self._lock: self._frames = [] self._stream = sd.InputStream( samplerate=self.sample_rate, channels=CHANNELS, dtype=DTYPE, callback=self._callback, ) self._stream.start() self.recording = True def stop(self) -> bytes | None: if self._stream is None and not self._frames: return None self._stop_stream() with self._lock: frames = list(self._frames) self._frames = [] if not frames: return None audio = np.concatenate(frames, axis=0) if len(audio) < self.sample_rate * MIN_SECONDS: return None return _to_wav_bytes(audio, self.sample_rate) def _to_wav_bytes(audio: np.ndarray, sample_rate: int) -> bytes: buf = io.BytesIO() with wave.open(buf, "wb") as w: w.setnchannels(CHANNELS) w.setsampwidth(2) w.setframerate(sample_rate) w.writeframes(audio.tobytes()) return buf.getvalue() def paste_at_cursor(text: str) -> None: """Text in Zwischenablage legen + Strg+V senden. Alter Inhalt wird nach 500 ms wiederhergestellt.""" if not text: return try: previous = pyperclip.paste() except Exception: # noqa: BLE001 previous = None pyperclip.copy(text) time.sleep(0.05) keyboard.send("ctrl+v") def _restore() -> None: time.sleep(0.5) if previous is not None: try: pyperclip.copy(previous) except Exception: # noqa: BLE001 pass threading.Thread(target=_restore, daemon=True).start() def transcribe(wav: bytes, mac_url: str, language: str, timeout: int) -> str: files = {"audio": ("dictate.wav", wav, "audio/wav")} data = {"language": language} r = requests.post( f"{mac_url.rstrip('/')}/api/transcribe", files=files, data=data, timeout=timeout, ) r.raise_for_status() return (r.json().get("text") or "").strip() _COLORS = { "idle": "#3b82f6", "rec": "#dc2626", "send": "#f59e0b", "err": "#737373", } def _make_icon(color: str) -> Image.Image: img = Image.new("RGBA", (64, 64), (0, 0, 0, 0)) d = ImageDraw.Draw(img) d.ellipse((6, 6, 58, 58), fill=color) return img def main() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) cfg = load_config() mac_url = cfg["mac_url"] hotkey = str(cfg.get("hotkey", "f8")).lower() language = cfg.get("language", "de") timeout = int(cfg.get("timeout", 120)) suppress = bool(cfg.get("suppress_hotkey", False)) recorder = Recorder() icon = Icon("voice-agent-dictate", _make_icon(_COLORS["idle"]), "Voice-Agent Diktat") state_lock = threading.Lock() def set_state(state: str, tip: str) -> None: icon.icon = _make_icon(_COLORS[state]) icon.title = f"Voice-Agent Diktat — {tip}" def on_press(_evt) -> None: with state_lock: if recorder.recording: return try: recorder.start() except Exception as e: # noqa: BLE001 log.error("Audio-Start fehlgeschlagen: %s", e) set_state("err", f"Audio-Fehler: {e}") return set_state("rec", "Aufnahme...") log.info("recording start") def on_release(_evt) -> None: with state_lock: if not recorder.recording and not recorder._frames: return wav = recorder.stop() if not wav: log.info("zu kurz / leer — verworfen") set_state("idle", "Idle") return set_state("send", "Sende an Mac...") log.info("recording stop (%d KB)", len(wav) // 1024) try: text = transcribe(wav, mac_url, language, timeout) except requests.RequestException as e: log.error("transcribe failed: %s", e) set_state("err", f"Mac-Fehler: {e}") return if not text: log.info("leeres Transkript") set_state("idle", "Idle") return log.info("text: %s", text[:120]) paste_at_cursor(text) set_state("idle", "Idle") keyboard.on_press_key(hotkey, on_press, suppress=suppress) keyboard.on_release_key(hotkey, on_release, suppress=suppress) def quit_app(_icon=None, _item=None) -> None: icon.stop() sys.exit(0) icon.menu = Menu(MenuItem("Beenden", quit_app)) log.info("Bereit. Hotkey halten: %s | Mac: %s | Sprache: %s", hotkey, mac_url, language) icon.run() if __name__ == "__main__": main()