Thorsten/input_helper.py

#!/usr/bin/env python3
"""
Input-Helper: Entwicklungstool für den Sprachassistenten.
Nimmt Text-Eingabe im Terminal entgegen und leitet sie als JSON an llm_module.py weiter.

Aufruf:
    python input_helper.py                  # interaktiver Modus (Endlosschleife)
    python input_helper.py "Deine Frage"    # einmaliger Aufruf mit Argument
    python input_helper.py --emotion drunk  # mit Emotion

Optionen:
    --emotion NAME    Emotion für TTS (amused|angry|disgusted|drunk|neutral|sleepy|surprised|whisper)
    --no-play         Nur JSON erzeugen, kein Audio abspielen
    --json-only       Gibt nur das JSON aus, ruft LLM nicht auf
    --history         Aktiviert Gesprächsverlauf (Kontext merken)
    --help            Diese Hilfe
"""

import sys
import json
import os
import subprocess
import argparse
from pathlib import Path

SCRIPT_DIR     = Path(__file__).parent
LLM_INPUT_FILE = SCRIPT_DIR / "data" / "llm_input.json"

VALID_EMOTIONS = ["amused", "angry", "disgusted", "drunk", "neutral", "sleepy", "surprised", "whisper"]

def parse_args():
    parser = argparse.ArgumentParser(description="Voice Assistant Input Helper", add_help=True)
    parser.add_argument("text", nargs="?", help="Eingabetext (optional, sonst interaktiv)")
    parser.add_argument("--emotion", default="neutral", choices=VALID_EMOTIONS,
                        help="Emotion für die TTS-Ausgabe (Standard: neutral)")
    parser.add_argument("--no-play", action="store_true",
                        help="Kein Audio abspielen, nur Datei erzeugen")
    parser.add_argument("--json-only", action="store_true",
                        help="Nur JSON ausgeben, LLM nicht aufrufen")
    parser.add_argument("--history", action="store_true",
                        help="Gesprächsverlauf aktivieren")
    return parser.parse_args()

def build_payload(text: str, emotion: str, autoplay: bool, history: list) -> dict:
    """Erstellt das JSON-Payload für das LLM-Modul."""
    return {
        "request": text,
        "type": "query",
        "emotion": emotion,
        "autoplay": autoplay,
        "history": history,
    }

def save_and_run(payload: dict, json_only: bool):
    """Speichert JSON und ruft LLM-Modul auf."""
    LLM_INPUT_FILE.parent.mkdir(parents=True, exist_ok=True)
    with open(LLM_INPUT_FILE, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)

    if json_only:
        print(json.dumps(payload, ensure_ascii=False, indent=2))
        return None

    llm_script  = SCRIPT_DIR / "llm_module.py"
    venv_python = Path.home() / ".venv" / "bin" / "python"
    python_bin  = str(venv_python) if venv_python.exists() else sys.executable

    result = subprocess.run(
        [python_bin, str(llm_script), str(LLM_INPUT_FILE)],
        cwd=SCRIPT_DIR,
    )
    return result.returncode

def get_last_answer() -> str | None:
    """Liest die letzte LLM-Antwort aus der TTS-Input-Datei (für History)."""
    tts_file = SCRIPT_DIR / "data" / "tts_input.json"
    if tts_file.exists():
        try:
            with open(tts_file) as f:
                data = json.load(f)
            return data.get("text")
        except Exception:
            pass
    return None

def interactive_loop(args):
    """Interaktiver Modus: Eingabe in Schleife."""
    history = []
    print("=" * 60)
    print("  Thorsten Voice Assistant – Entwicklungs-Helper")
    print("=" * 60)
    print(f"  Emotion: {args.emotion}  |  History: {'an' if args.history else 'aus'}")
    print("  Tippe deine Anfrage oder 'exit' zum Beenden.")
    print("  Befehle: /emotion <name>  /history  /clear  /exit")
    print("=" * 60)

    emotion = args.emotion

    while True:
        try:
            user_input = input("\n>> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n[Helper] Beendet.")
            break

        if not user_input:
            continue

        # Meta-Befehle
        if user_input.lower() in ("exit", "quit", "/exit"):
            print("[Helper] Beendet.")
            break
        elif user_input.startswith("/emotion "):
            new_emotion = user_input.split(" ", 1)[1].strip().lower()
            if new_emotion in VALID_EMOTIONS:
                emotion = new_emotion
                print(f"[Helper] Emotion geändert zu: {emotion}")
            else:
                print(f"[Helper] Ungültige Emotion. Verfügbar: {', '.join(VALID_EMOTIONS)}")
            continue
        elif user_input == "/history":
            print(f"[Helper] Verlauf ({len(history)} Einträge):")
            for i, h in enumerate(history):
                print(f"  [{i}] User: {h.get('user', '')} → Assistant: {h.get('assistant', '')[:50]}...")
            continue
        elif user_input == "/clear":
            history.clear()
            print("[Helper] Verlauf gelöscht.")
            continue

        last_request = user_input
        payload = build_payload(user_input, emotion, not args.no_play, history if args.history else [])
        save_and_run(payload, args.json_only)

        # History aktualisieren
        if args.history:
            answer = get_last_answer()
            if answer:
                history.append({"user": last_request, "assistant": answer})
                # Max. 10 Einträge behalten
                if len(history) > 10:
                    history = history[-10:]

def main():
    args = parse_args()

    if args.text:
        # Einmaliger Aufruf
        payload = build_payload(args.text, args.emotion, not args.no_play, [])
        save_and_run(payload, args.json_only)
    else:
        # Interaktiver Modus
        interactive_loop(args)

if __name__ == "__main__":
    main()