PyVtT/voice_to_text_tray.py

#!/usr/bin/env python3
import sys
import subprocess
import os
import threading
import socket
import json
import requests
from PyQt5.QtWidgets import QApplication, QSystemTrayIcon, QMenu, QAction
from PyQt5.QtGui import QIcon
from PyQt5.QtCore import QThread, pyqtSignal

# === Config ===
def read_configurations():
    script_dir = os.path.dirname(os.path.abspath(__file__))
    settings_path = os.path.join(script_dir, "pyvtt.settings.json")
    with open(settings_path) as f:
        return json.load(f)

CONFIGURATION = read_configurations()
CURRENT_PRESET = CONFIGURATION["presets"][0]  # Default to first preset

def notify(title: str, message: str) -> None:
    try:
        subprocess.run(["notify-send", "-a", "Voice to Text", "-i", "audio-input-microphone", title, message], check=True)
    except subprocess.CalledProcessError as e:
        print("Fehler beim Benachrichtigen mit 'notify-send'.")
        print(e)

# === Worker Thread for Whisper + Ollama ===
class WhisperWorker(QThread):
    finished = pyqtSignal(str)

    def run(self):
        try:
            # Whisper ausführen
            whisper_cmd = [
                CONFIGURATION["whisper_path"],
                "-m", CURRENT_PRESET["whisper_model"],
                "-f", CONFIGURATION["audio_file"],
                "-l", CURRENT_PRESET["language"],
                "-otxt",
                "-of", CONFIGURATION["output_file"].replace(".txt", "")
            ]
            try:
                subprocess.run(whisper_cmd, check=True)
            except subprocess.CalledProcessError as e:
                print(f"Whisper Fehler: {e}")
                notify("Fehler", "Ein Fehler mit 'Whisper' ist aufgetreten!")
                return
            
            try:
                with open(CONFIGURATION["output_file"], "r") as f:
                    raw_result = f.read().strip().replace("\n", " ")
            except Exception as e:
                print(f"Datei Fehler: {e}")
                notify("Fehler", "Ein Fehler beim Lesen der Whisper-Ausgabe ist aufgetreten!")
                return
            
            print("Whisper Transkript erhalten.")

            # --- An Ollama schicken ---
            payload = {
                "model": CURRENT_PRESET["ollama_model"],
                "prompt": CURRENT_PRESET["ollama_prompt"] + raw_result,
                "stream": False
            }
            ollama_endpoint = f"{CONFIGURATION['ollama_url']}:{CONFIGURATION['ollama_port']}/api/generate"
            response = requests.post(ollama_endpoint, json=payload)

            try:
                response.raise_for_status()
            except requests.exceptions.HTTPError as e:
                print(f"HTTP Fehler: {e}")
                notify("Fehler", "Ein Fehler bei der Kommunikation mit 'Ollama' ist aufgetreten!")
                return

            formatted_result = response.json().get("response", "").strip()
            formatted_result = "\n".join(line.strip() for line in formatted_result.splitlines())
            print("Ollama Antwort erhalten.")

            # Ergebnis ins Clipboard kopieren
            try:
                subprocess.run(["wl-copy"], input=formatted_result.encode(), check=True)
            except subprocess.CalledProcessError as e:
                print(f"Clipboard Fehler: {e}")
                notify("Fehler", "Ein Fehler beim Kopieren des Ergebnisses ist aufgetreten!")
                return
            
            notify("Spracherkennung", "Transkription abgeschlossen!")
            self.finished.emit(formatted_result)

        except Exception as e:
            print(f"Fehler: {e}")
            notify("Fehler", "Ein Fehler ist aufgetreten!")
            return

# === Socket Listener Thread ===
class SocketListener(threading.Thread):
    def __init__(self, tray_app):
        super().__init__(daemon=True)
        self.tray_app = tray_app
        if os.path.exists(CONFIGURATION["socket_path"]):
            os.remove(CONFIGURATION["socket_path"])
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.bind(CONFIGURATION["socket_path"])
        os.chmod(CONFIGURATION["socket_path"], 0o666)
        self.sock.listen(1)

    def run(self):
        while True:
            conn, _ = self.sock.accept()
            with conn:
                data = conn.recv(1024).decode().strip()
                if data == "toggle":
                    self.tray_app.toggle_recording()
                elif data == "start":
                    self.tray_app.start_recording()
                elif data == "stop":
                    self.tray_app.stop_recording_if_possible()

# === Tray Application ===
class TrayApp:
    def __init__(self):
        self.app = QApplication(sys.argv)
        self.tray = QSystemTrayIcon(QIcon.fromTheme("audio-input-microphone"))
        self.menu = QMenu()

        self.app.aboutToQuit.connect(self.cleanup)

        # Preset Menü
        self.preset_actions = []
        self.preset_group = QMenu("Presets")
        for i, preset in enumerate(CONFIGURATION["presets"]):
            action = QAction(preset["name"], self.menu)
            action.setCheckable(True)
            if i == 0:
                action.setChecked(True)
            action.triggered.connect(lambda checked, index=i: self.set_preset(index))
            self.preset_group.addAction(action)
            self.preset_actions.append(action)
        self.menu.addMenu(self.preset_group)

        # Quit
        self.quit_action = QAction("Beenden")
        self.quit_action.triggered.connect(self.app.quit)
        self.menu.addAction(self.quit_action)

        self.tray.setContextMenu(self.menu)
        self.tray.setToolTip("Voice to Text")
        self.tray.show()

        self.recording_process = None

        self.socket_listener = SocketListener(self)
        self.socket_listener.start()

    def set_preset(self, index):
        global CURRENT_PRESET
        print(f"Preset gewechselt: {CONFIGURATION['presets'][index]['name']}")
        CURRENT_PRESET = CONFIGURATION["presets"][index]
        # Nur einer darf gecheckt sein
        for i, action in enumerate(self.preset_actions):
            action.setChecked(i == index)

    def start_recording(self):
        if self.recording_process is None:
            print("Starte Aufnahme...")
            self.recording_process = subprocess.Popen([
                "ffmpeg", "-f", "pulse", "-i", "default", "-ar", "16000",
                "-ac", "1", CONFIGURATION["audio_file"], "-y", "-loglevel", "quiet"
            ])
            notify("Aufnahme", "Aufnahme gestartet!")

    def stop_recording_if_possible(self):
        if self.recording_process:
            print("Stoppe Aufnahme...")
            self.recording_process.terminate()
            self.recording_process.wait()
            self.recording_process = None
            notify("Aufnahme", "Aufnahme beendet, verarbeite...")
            self.start_whisper_worker()

    def toggle_recording(self):
        if self.recording_process:
            self.stop_recording_if_possible()
        else:
            self.start_recording()

    def start_whisper_worker(self):
        self.worker = WhisperWorker()
        self.worker.finished.connect(self.show_result)
        self.worker.start()

    def show_result(self, text):
        print(f"Fertig:\n{text}")

    def cleanup(self):
        if os.path.exists(CONFIGURATION["socket_path"]):
            os.remove(CONFIGURATION["socket_path"])
        print("Socket sauber entfernt.")

    def run(self):
        sys.exit(self.app.exec_())

if __name__ == "__main__":
    TrayApp().run()