# -*- 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)