Trixy7/trixy_core/input/keyboard_input.py

# -*- coding: utf-8 -*-
"""
Keyboard Input Service.

Ermöglicht Text-Eingabe über die Konsole als Alternative zur Spracheingabe.
Cross-platform: Windows, Linux, macOS, Raspberry Pi OS.

Features:
- Ctrl+I zum Öffnen des Text-Inputs
- Automatische Wiederöffnung bei Follow-Up-Fragen
- Integration mit Conversation-System
- Sendet TextInput Command an Server

Benötigt: pip install prompt_toolkit
"""

import asyncio
import sys
import uuid
from typing import TYPE_CHECKING, Callable, Awaitable

from trixy_core.network.cmd import (
    TextInput, FollowUpRequest, TextInputAck,
    MusicPlayPause, MusicNext, MusicStop, MediaStopAll,
)
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

if TYPE_CHECKING:
    from trixy_core.application import IApplication
    from trixy_core.network.client_connection import ClientConnection

# Lazy import für prompt_toolkit
_prompt_toolkit_available = None


def _check_prompt_toolkit() -> bool:
    """Prüft ob prompt_toolkit verfügbar ist."""
    global _prompt_toolkit_available
    if _prompt_toolkit_available is None:
        try:
            import prompt_toolkit
            _prompt_toolkit_available = True
        except ImportError:
            _prompt_toolkit_available = False
    return _prompt_toolkit_available


class KeyboardInputService:
    """
    Service für Keyboard-Eingabe mit Ctrl+I Trigger.

    Ermöglicht Text-Eingabe als Alternative zu Sprache.
    Integriert sich mit dem Conversation-System.

    Features:
    - Ctrl+I öffnet Text-Input-Prompt
    - Enter sendet Text an Server
    - Automatische Wiederöffnung bei Follow-Ups
    - Conversation-Tracking

    Usage:
        service = KeyboardInputService(app, connection)
        await service.start()
        # Drücke Ctrl+I für Eingabe
    """

    def __init__(
        self,
        application: "IApplication",
        connection: "ClientConnection | None" = None,
        prompt: str = ">>> ",
    ) -> None:
        """
        Initialisiert den Keyboard-Input-Service.

        Args:
            application: Die Trixy-Anwendung
            connection: Client-Verbindung zum Server (für Client-Modus)
            prompt: Eingabe-Prompt
        """
        self._app = application
        self._connection = connection
        self._prompt = prompt
        self._running = False
        self._input_active = False

        # Tasks
        self._key_listener_task: asyncio.Task | None = None
        self._input_task: asyncio.Task | None = None

        # Conversation State
        self._current_session_id: str = ""
        self._awaiting_follow_up = False
        self._follow_up_number = 0
        self._follow_up_question = ""

        # Event für Input-Trigger
        self._input_requested = asyncio.Event()

    @property
    def is_running(self) -> bool:
        """Läuft der Service?"""
        return self._running

    @property
    def has_active_session(self) -> bool:
        """Gibt es eine aktive Session?"""
        return bool(self._current_session_id)

    def set_connection(self, connection: "ClientConnection") -> None:
        """Setzt die Server-Verbindung."""
        self._connection = connection

    async def start(self) -> bool:
        """
        Startet den Keyboard-Input-Service.

        Returns:
            True wenn erfolgreich gestartet
        """
        if not _check_prompt_toolkit():
            pwarn(
                "Keyboard-Input (Ctrl+I) deaktiviert: prompt_toolkit nicht installiert. "
                "Installieren mit: pip install prompt_toolkit"
            )
            return False

        if self._running:
            return True

        self._running = True

        # Key-Listener starten
        self._key_listener_task = asyncio.create_task(self._key_listener_loop())

        # Input-Handler starten
        self._input_task = asyncio.create_task(self._input_handler_loop())

        pinfo("Keyboard-Shortcuts: Ctrl+I=Eingabe, Ctrl+P=Play/Pause, Ctrl+N=Next, Ctrl+S=Stop")
        return True

    async def stop(self) -> None:
        """Stoppt den Keyboard-Input-Service."""
        self._running = False
        self._input_requested.set()  # Unblock waiting

        for task in [self._key_listener_task, self._input_task]:
            if task and not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass

        pdebug("Keyboard-Input gestoppt")

    async def request_input(self, follow_up_question: str = "", session_id: str = "") -> None:
        """
        Fordert eine Texteingabe an (z.B. für Follow-Up).

        Args:
            follow_up_question: Die Follow-Up-Frage (falls vorhanden)
            session_id: Session-ID für die Fortsetzung
        """
        if follow_up_question:
            self._awaiting_follow_up = True
            self._follow_up_question = follow_up_question
            self._follow_up_number += 1
            if session_id:
                self._current_session_id = session_id
            pdebug(f"Follow-Up Input angefordert: {follow_up_question[:50]}...")
        else:
            self._awaiting_follow_up = False
            self._follow_up_question = ""
            self._follow_up_number = 0
            self._current_session_id = ""

        self._input_requested.set()

    def handle_follow_up_request(self, request: FollowUpRequest) -> None:
        """
        Behandelt eine Follow-Up-Anfrage vom Server.

        Args:
            request: Die Follow-Up-Anfrage
        """
        asyncio.create_task(self.request_input(
            follow_up_question=request.question,
            session_id=request.session_id,
        ))

    def handle_conversation_end(self, session_id: str) -> None:
        """
        Behandelt das Ende einer Conversation.

        Args:
            session_id: Die Session-ID
        """
        if session_id == self._current_session_id:
            self._current_session_id = ""
            self._awaiting_follow_up = False
            self._follow_up_question = ""
            self._follow_up_number = 0
            pdebug("Conversation beendet")

    async def _key_listener_loop(self) -> None:
        """
        Lauscht auf Tastenkombinationen.

        Unterstützte Shortcuts:
        - Ctrl+I / Tab: Text-Eingabe öffnen
        - Ctrl+P: Musik Play/Pause
        - Ctrl+N: Nächstes Lied
        - Ctrl+S: Alles stoppen (Musik + TTS)
        - Ctrl+C: Anwendung beenden
        """
        from prompt_toolkit.input import create_input
        from prompt_toolkit.keys import Keys

        try:
            # Input erstellen
            inp = create_input()

            with inp.raw_mode():
                while self._running:
                    # Auf Tastendruck warten
                    keys = await asyncio.get_event_loop().run_in_executor(
                        None, inp.read_keys
                    )

                    for key_press in keys:
                        # Ctrl+I = Tab im Terminal -> Text-Eingabe
                        if key_press.key == Keys.ControlI or key_press.key == Keys.Tab:
                            if not self._input_active:
                                self._input_requested.set()

                        # Ctrl+P -> Musik Play/Pause
                        elif key_press.key == Keys.ControlP:
                            await self._send_music_play_pause()

                        # Ctrl+N -> Nächstes Lied
                        elif key_press.key == Keys.ControlN:
                            await self._send_music_next()

                        # Ctrl+S -> Alles stoppen (Musik + TTS)
                        elif key_press.key == Keys.ControlS:
                            await self._send_media_stop_all()

                        # Ctrl+C -> Anwendung beenden
                        elif key_press.key == Keys.ControlC:
                            pinfo("Ctrl+C empfangen, beende Anwendung...")
                            self._app.shutdown()
                            return

        except asyncio.CancelledError:
            pass
        except Exception as e:
            # Fallback: Einfacher Modus ohne Shortcuts
            pdebug(f"Key-Listener-Fehler: {e}, wechsle zu einfachem Modus")
            await self._simple_key_listener()

    async def _simple_key_listener(self) -> None:
        """Einfacher Key-Listener als Fallback."""
        # Im einfachen Modus: Drücke Enter für Eingabe
        pinfo("Einfacher Modus: Drücke Enter für Texteingabe")

        loop = asyncio.get_running_loop()

        while self._running:
            try:
                # Warte auf Enter
                await loop.run_in_executor(None, sys.stdin.readline)
                if not self._input_active:
                    self._input_requested.set()
            except asyncio.CancelledError:
                break
            except Exception:
                await asyncio.sleep(0.1)

    async def _input_handler_loop(self) -> None:
        """Hauptschleife für Input-Handling."""
        from prompt_toolkit import PromptSession
        from prompt_toolkit.history import InMemoryHistory
        from prompt_toolkit.patch_stdout import patch_stdout

        history = InMemoryHistory()
        session: PromptSession = PromptSession(history=history)

        while self._running:
            try:
                # Warte auf Input-Request
                await self._input_requested.wait()
                self._input_requested.clear()

                if not self._running:
                    break

                self._input_active = True

                # Prompt vorbereiten
                if self._awaiting_follow_up and self._follow_up_question:
                    # Follow-Up: Frage anzeigen
                    print(f"\n[Rückfrage] {self._follow_up_question}")
                    prompt = f"Antwort> "
                else:
                    # Neue Eingabe
                    print()  # Neue Zeile
                    prompt = self._prompt

                # Text einlesen
                with patch_stdout():
                    text = await session.prompt_async(prompt)

                self._input_active = False

                if not text or not text.strip():
                    continue

                text = text.strip()
                pdebug(f"Keyboard-Input: {text}")

                # Text verarbeiten
                await self._process_input(text)

            except EOFError:
                # Ctrl+D - Eingabe abbrechen
                self._input_active = False
                continue
            except KeyboardInterrupt:
                # Ctrl+C - Ignorieren
                self._input_active = False
                continue
            except asyncio.CancelledError:
                break
            except Exception as e:
                perror(f"Fehler in Input-Handler: {e}")
                self._input_active = False
                await asyncio.sleep(0.1)

    async def _process_input(self, text: str) -> None:
        """
        Verarbeitet die Texteingabe.

        Args:
            text: Der eingegebene Text
        """
        # Session-ID generieren falls neue Conversation
        if not self._current_session_id:
            self._current_session_id = f"kb-{uuid.uuid4().hex[:8]}"

        # Room und Alias ermitteln
        room = getattr(self._app, "room", "local")
        alias = getattr(self._app, "alias", "keyboard")

        if self._connection and self._connection.is_connected:
            # Client-Modus: TextInput Command an Server senden
            cmd = TextInput(
                satellite_id=alias,
                text=text,
                session_id=self._current_session_id,
                room_id=room,
                language="de",
                is_follow_up=self._awaiting_follow_up,
                follow_up_number=self._follow_up_number,
            )

            success = await self._connection.send_command(cmd)
            if success:
                pdebug(f"TextInput gesendet: session={self._current_session_id}")
            else:
                perror("Fehler beim Senden des TextInput Commands")

            # Follow-Up zurücksetzen (wird bei neuer Anfrage wieder gesetzt)
            self._awaiting_follow_up = False
            self._follow_up_question = ""

        else:
            # Standalone-Modus: Direkt Event auslösen
            from trixy_core.events.event_data.basic import SpeechRecognized

            event_data = SpeechRecognized(
                text=text,
                room_id=room,
                satellite_id=alias,
                confidence=1.0,
                language="de",
                is_final=True,
                source="keyboard",
            )

            await self._app.events.trigger("speech_recognized", event_data)

            # Follow-Up zurücksetzen
            self._awaiting_follow_up = False
            self._follow_up_question = ""

    async def _send_music_play_pause(self) -> None:
        """Toggle Musik Play/Pause."""
        alias = getattr(self._app, "alias", "keyboard")

        if self._connection and self._connection.is_connected:
            # Client-Modus: Command an Server senden
            cmd = MusicPlayPause(satellite_id=alias)
            success = await self._connection.send_command(cmd)
            if success:
                pdebug("MusicPlayPause Command an Server gesendet")
            else:
                perror("Fehler beim Senden des MusicPlayPause Commands")
        else:
            # Standalone-Modus: Lokale Wiedergabe steuern
            audio_manager = getattr(self._app, "audio_manager", None)
            if audio_manager:
                is_paused = audio_manager.toggle_music_pause()
                status = "pausiert" if is_paused else "fortgesetzt"
                pinfo(f"Musik {status}")
                event_name = "music_paused" if is_paused else "music_resumed"
                await self._app.events.emit(event_name, {"satellite_id": alias})
            else:
                await self._app.events.emit("music_play_pause", {"satellite_id": alias})
                pdebug("MusicPlayPause Event ausgelöst")

    async def _send_music_next(self) -> None:
        """Nächstes Lied - mit Crossfade wenn aktiviert."""
        alias = getattr(self._app, "alias", "keyboard")

        if self._connection and self._connection.is_connected:
            # Client-Modus: Command an Server senden
            cmd = MusicNext(satellite_id=alias, use_crossfade=True)
            success = await self._connection.send_command(cmd)
            if success:
                pdebug("MusicNext Command an Server gesendet")
            else:
                perror("Fehler beim Senden des MusicNext Commands")
        else:
            # Standalone-Modus: Lokales Event auslösen
            await self._app.events.emit(
                "music_next", {"satellite_id": alias, "use_crossfade": True}
            )
            pdebug("MusicNext Event ausgelöst")

    async def _send_media_stop_all(self) -> None:
        """Stoppt alle Medien-Wiedergabe (Musik + TTS)."""
        alias = getattr(self._app, "alias", "keyboard")

        # Lokale Wiedergabe sofort stoppen (behebt ALSA underrun)
        audio_manager = getattr(self._app, "audio_manager", None)
        if audio_manager:
            audio_manager.stop_all()
            pinfo("Lokale Wiedergabe gestoppt")

        if self._connection and self._connection.is_connected:
            # Client-Modus: Command an Server senden
            cmd = MediaStopAll(satellite_id=alias)
            success = await self._connection.send_command(cmd)
            if success:
                pdebug("MediaStopAll Command an Server gesendet")
            else:
                perror("Fehler beim Senden des MediaStopAll Commands")
        else:
            # Standalone-Modus: Lokale Wiedergabe stoppen
            audio_manager = getattr(self._app, "audio_manager", None)
            if audio_manager:
                audio_manager.stop_all()
                pinfo("Wiedergabe gestoppt")
                await self._app.events.emit("media_stopped", {"satellite_id": alias})
            else:
                await self._app.events.emit("music_stop", {"satellite_id": alias})
                await self._app.events.emit("tts_stop", {"satellite_id": alias})
                pinfo("Wiedergabe gestoppt")


class SimpleKeyboardInput:
    """
    Einfache Keyboard-Input-Klasse ohne prompt_toolkit.

    Fallback wenn prompt_toolkit nicht installiert ist.
    Kein Ctrl+I Support - nur Enter für Eingabe.
    """

    def __init__(
        self,
        application: "IApplication",
        connection: "ClientConnection | None" = None,
        prompt: str = ">>> ",
    ) -> None:
        """Initialisiert SimpleKeyboardInput."""
        self._app = application
        self._connection = connection
        self._prompt = prompt
        self._running = False
        self._task: asyncio.Task | None = None

        # Conversation State
        self._current_session_id: str = ""
        self._awaiting_follow_up = False
        self._follow_up_number = 0
        self._follow_up_question = ""

    def set_connection(self, connection: "ClientConnection") -> None:
        """Setzt die Server-Verbindung."""
        self._connection = connection

    async def start(self) -> bool:
        """Startet den Input-Service."""
        if self._running:
            return True

        self._running = True
        self._task = asyncio.create_task(self._input_loop())
        pinfo("Keyboard-Input aktiviert (Enter für Texteingabe)")
        return True

    async def stop(self) -> None:
        """Stoppt den Input-Service."""
        self._running = False
        if self._task and not self._task.done():
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

    async def request_input(self, follow_up_question: str = "", session_id: str = "") -> None:
        """Fordert eine Texteingabe an."""
        if follow_up_question:
            self._awaiting_follow_up = True
            self._follow_up_question = follow_up_question
            self._follow_up_number += 1
            if session_id:
                self._current_session_id = session_id
            print(f"\n[Rückfrage] {follow_up_question}")

    def handle_follow_up_request(self, request: FollowUpRequest) -> None:
        """Behandelt eine Follow-Up-Anfrage."""
        asyncio.create_task(self.request_input(
            follow_up_question=request.question,
            session_id=request.session_id,
        ))

    def handle_conversation_end(self, session_id: str) -> None:
        """Behandelt das Ende einer Conversation."""
        if session_id == self._current_session_id:
            self._current_session_id = ""
            self._awaiting_follow_up = False
            self._follow_up_question = ""
            self._follow_up_number = 0

    async def _input_loop(self) -> None:
        """Hauptschleife für einfachen Input."""
        loop = asyncio.get_running_loop()

        while self._running:
            try:
                # Prompt
                if self._awaiting_follow_up:
                    prompt = "Antwort> "
                else:
                    prompt = self._prompt

                # Input in Thread ausführen
                text = await loop.run_in_executor(None, lambda: input(prompt))

                if not text or not text.strip():
                    continue

                text = text.strip()
                await self._process_input(text)

            except EOFError:
                break
            except KeyboardInterrupt:
                continue
            except asyncio.CancelledError:
                break
            except Exception as e:
                perror(f"Fehler in Input-Loop: {e}")
                await asyncio.sleep(0.1)

    async def _process_input(self, text: str) -> None:
        """Verarbeitet die Texteingabe."""
        if not self._current_session_id:
            self._current_session_id = f"kb-{uuid.uuid4().hex[:8]}"

        room = getattr(self._app, "room", "local")
        alias = getattr(self._app, "alias", "keyboard")

        if self._connection and self._connection.is_connected:
            cmd = TextInput(
                satellite_id=alias,
                text=text,
                session_id=self._current_session_id,
                room_id=room,
                language="de",
                is_follow_up=self._awaiting_follow_up,
                follow_up_number=self._follow_up_number,
            )
            await self._connection.send_command(cmd)
        else:
            from trixy_core.events.event_data.basic import SpeechRecognized

            event_data = SpeechRecognized(
                text=text,
                room_id=room,
                satellite_id=alias,
                confidence=1.0,
                language="de",
                is_final=True,
                source="keyboard",
            )
            await self._app.events.trigger("speech_recognized", event_data)

        self._awaiting_follow_up = False
        self._follow_up_question = ""


def create_keyboard_input(
    application: "IApplication",
    connection: "ClientConnection | None" = None,
    prompt: str = ">>> ",
) -> KeyboardInputService | SimpleKeyboardInput:
    """
    Factory-Funktion für Keyboard-Input.

    Wählt automatisch die beste verfügbare Implementierung.

    Args:
        application: Die Trixy-Anwendung
        connection: Server-Verbindung (für Client-Modus)
        prompt: Eingabe-Prompt

    Returns:
        KeyboardInputService oder SimpleKeyboardInput
    """
    if _check_prompt_toolkit():
        return KeyboardInputService(application, connection, prompt)
    else:
        pwarn("prompt_toolkit nicht gefunden, nutze einfachen Input-Modus")
        return SimpleKeyboardInput(application, connection, prompt)