voice-agent/windows-dictate/dictate.py

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