#!/usr/bin/env python3 """ Text User Interface (TUI) for Voice Assistant with Text Input AI Now Inc - Del Mar Demo Unit Laboratory Assistant: Claw 🏭 Features: - Voice activation via "Hey Osiris" - Text input alternative (type commands) - Rich terminal formatting - Conversation history """ import os import sys import json import logging import signal import time import threading from pathlib import Path from datetime import datetime # Try to import rich for beautiful TUI try: from rich.console import Console from rich.panel import Panel from rich import box from rich.prompt import Prompt RICH_AVAILABLE = True except ImportError: RICH_AVAILABLE = False print("Installing rich for TUI...") import subprocess subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'rich']) from rich.console import Console from rich.panel import Panel from rich.prompt import Prompt from rich import box RICH_AVAILABLE = True print("Rich installed!") from assistant import VoiceAssistant from tts_engine import TTSEngine from hotword_detector import HotwordDetector # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class TUIAssistant: """TUI-based voice assistant with text input support.""" def __init__(self, config_path: str = "config.json"): self.config_path = Path(config_path) self.config = self._load_config() # Initialize components logger.info("Initializing TUI voice assistant...") self.assistant = VoiceAssistant(str(self.config_path)) self.tts = TTSEngine(str(self.config_path)) self.hotword_detector = HotwordDetector( str(self.config_path).replace("config.json", "hotword_config.json") ) # State self.is_running = False self.voice_mode = False # True when voice activation is active self.conversation_history = [] self.max_history = 20 # Rich console self.console = Console() # Setup signal handlers signal.signal(signal.SIGINT, self._signal_handler) signal.signal(signal.SIGTERM, self._signal_handler) # Set callback for hotword detection self.hotword_detector.set_callback(self._on_hotword) logger.info("TUI Voice assistant initialized") def _load_config(self) -> dict: """Load configuration.""" try: with open(self.config_path, 'r') as f: return json.load(f) except FileNotFoundError: logger.warning("Config not found, using defaults") return {} def _signal_handler(self, sig, frame): """Handle shutdown signals.""" logger.info("Shutdown signal received") self.is_running = False def _on_hotword(self): """Callback when hotword is detected.""" self.console.print("\n[bold green]✨ Hotword detected! ✨[/bold green]") self.voice_mode = True self.tts.speak("Yes?", "en") def _add_to_history(self, role: str, text: str, method: str = "text"): """Add message to conversation history.""" timestamp = datetime.now().strftime("%H:%M:%S") self.conversation_history.append({ 'timestamp': timestamp, 'role': role, 'text': text, 'method': method # 'text', 'voice', or 'assistant' }) if len(self.conversation_history) > self.max_history: self.conversation_history.pop(0) def _display_history(self): """Display recent conversation history.""" if len(self.conversation_history) == 0: return self.console.print("\n[bold]📜 Recent Conversation:[/bold]") for msg in self.conversation_history[-10:]: # Show last 10 messages if msg['role'] == 'user': icon = "👤" if msg['method'] == 'text' else "🎤" color = "green" if msg['method'] == 'text' else "yellow" self.console.print(f"[{color}]{icon} [{msg['timestamp']}]: {msg['text']}") else: self.console.print(f"[cyan]🤖 [{msg['timestamp']}]: {msg['text']}") self.console.print() def _process_command(self, command: str, method: str = "text"): """Process a command from user (text or voice).""" if not command or command.strip() == "": return self._add_to_history('user', command, method) # Check for special commands cmd_lower = command.lower().strip() if cmd_lower in ['quit', 'exit', 'bye']: self.console.print("\n[bold]Goodbye! 👋[/bold]") self.tts.speak("Goodbye!", "en") self.is_running = False return elif cmd_lower == 'history': self._display_history() return elif cmd_lower == 'help': self._show_help() return # Process with assistant self.console.print("\n[bold blue]🤔 Processing...[/bold blue]") response = self.assistant.process(command) self._add_to_history('assistant', response, 'assistant') self.console.print(f"[bold cyan]🤖:[/bold cyan] {response}") # Speak response self.console.print("[bold blue]🔊 Speaking...[/bold blue]") self.tts.speak(response, "en") def _show_help(self): """Show help message.""" help_text = """ [bold]Voice Assistant Commands:[/bold] • Type your command or question • Say "Hey Osiris" for voice activation • Type 'history' to see recent conversation • Type 'quit' or 'exit' to stop [bold]Examples:[/bold] • "What time is it?" • "Play some music" • "Tell me a joke" • "What's the weather?" """ self.console.print(Panel(help_text, box=box.ROUNDED, border_style="green")) def run(self): """Run the TUI voice assistant.""" logger.info("Starting TUI voice assistant...") # Display welcome screen self.console.print(Panel.fit( "[bold]🎤 Voice Assistant - AI Now Inc[/bold]\n" "[cyan]Laboratory Assistant: Claw 🏭[/cyan]\n\n" "Modes:\n" " • [bold]Type commands[/bold] in the prompt\n" " • Say '[bold]Hey Osiris[/bold]' for voice activation\n\n" "Commands: [bold]help[/bold] | [bold]history[/bold] | [bold]quit[/bold]", box=box.ROUNDED, border_style="blue" )) self._show_help() self.is_running = True # Start voice listener in background voice_thread = threading.Thread(target=self._voice_listener, daemon=True) voice_thread.start() # Main loop - text input while self.is_running: try: # Get text input from user prompt_text = "\n[bold green]👤 You:[/bold green] " if not self.voice_mode else "\n[bold yellow]🎤 Voice Mode:[/bold yellow] " user_input = Prompt.ask(prompt_text) if user_input: self._process_command(user_input, method='text') except KeyboardInterrupt: logger.info("Interrupted by user") break except Exception as e: logger.error(f"Error: {e}") time.sleep(1) self.shutdown() def _voice_listener(self): """Background thread to listen for voice hotword.""" logger.info("Voice listener started") while self.is_running: try: if self.voice_mode: # Voice mode active - wait for speech # In a full implementation, this would use speech recognition time.sleep(0.5) self.voice_mode = False # Reset after use else: # Wait for hotword detected = self.hotword_detector.detect(timeout=5) if detected: self._on_hotword() except Exception as e: logger.error(f"Voice listener error: {e}") time.sleep(1) logger.info("Voice listener stopped") def shutdown(self): """Clean shutdown.""" logger.info("Shutting down...") self.is_running = False if hasattr(self, 'hotword_detector'): self.hotword_detector.stop() self.console.print("\n[bold red]Voice Assistant stopped.[/bold red]") logger.info("Shutdown complete") def main(): """Main entry point.""" import argparse parser = argparse.ArgumentParser(description='TUI Voice Assistant') parser.add_argument('--config', default='config.json', help='Config file path') args = parser.parse_args() app = TUIAssistant(config_path=args.config) app.run() if __name__ == '__main__': main()