From 03b33060b4394420acb4f4c83a0ac9afdab01f82 Mon Sep 17 00:00:00 2001 From: Claw - AI Now Inc Date: Sun, 1 Mar 2026 12:19:10 -0800 Subject: [PATCH] Add TUI version with rich library for better text display --- requirements.txt | 3 + tui_app.py | 262 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 265 insertions(+) create mode 100644 tui_app.py diff --git a/requirements.txt b/requirements.txt index 82edb87..b75bb0f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -39,6 +39,9 @@ colorlog>=6.0.0 fuzzywuzzy>=0.18.0 # Fuzzy string matching for music search python-Levenshtein>=0.19.0 # Fast string matching +# TUI (Text User Interface) +rich>=13.0.0 + # OpenClaw Client openai>=1.0.0 elevenlabs>=0.2.0 diff --git a/tui_app.py b/tui_app.py new file mode 100644 index 0000000..e8011a3 --- /dev/null +++ b/tui_app.py @@ -0,0 +1,262 @@ +#!/usr/bin/env python3 +""" +Text User Interface (TUI) for Voice Assistant +AI Now Inc - Del Mar Demo Unit +Laboratory Assistant: Claw 🏭 +""" + +import os +import sys +import json +import logging +import signal +import time +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.live import Live + from rich.layout import Layout + from rich.text import Text + from rich.table import Table + from rich import box + RICH_AVAILABLE = True +except ImportError: + RICH_AVAILABLE = False + +from assistant import VoiceAssistant +from tts_engine import TTSEngine +from speech_recognizer import BilingualSpeechRecognizer +from music_player import MusicPlayer +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 rich text display.""" + + 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.is_awake = False + self.current_status = "🎤 Listening for hotword..." + self.last_transcript = "" + self.last_response = "" + self.conversation_history = [] + self.max_history = 10 + + # Rich console + if RICH_AVAILABLE: + self.console = Console() + else: + self.console = None + + # Setup signal handlers + signal.signal(signal.SIGINT, self._signal_handler) + signal.signal(signal.SIGTERM, self._signal_handler) + + 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 _update_status(self, status: str): + """Update the current status.""" + self.current_status = status + if self.console: + self.console.print(f"\n[bold blue]→[/bold blue] {status}") + + def _add_to_history(self, role: str, text: str): + """Add message to conversation history.""" + timestamp = datetime.now().strftime("%H:%M:%S") + self.conversation_history.append({ + 'timestamp': timestamp, + 'role': role, + 'text': text + }) + # Keep only last N entries + if len(self.conversation_history) > self.max_history: + self.conversation_history.pop(0) + + def _display_conversation(self): + """Display conversation history.""" + if not self.console or not RICH_AVAILABLE: + return + + self.console.print("\n[bold]Recent Conversation:[/bold]") + for msg in self.conversation_history[-5:]: # Show last 5 messages + role_icon = "👤" if msg['role'] == 'user' else "🤖" + role_color = "green" if msg['role'] == 'user' else "cyan" + self.console.print(f"[{role_color}]{role_icon} [{msg['timestamp']}]:[/bold] {msg['text']}") + self.console.print() + + def _on_hotword_detected(self): + """Callback when hotword is detected.""" + self.is_awake = True + self._update_status("🔔 Hotword detected! Listening...") + + if self.console: + self.console.print("\n[bold green]✨ Voice Assistant Activated! ✨[/bold green]\n") + + # Speak activation tone + self.tts.speak("Yes?", "en") + + def listen_and_respond(self): + """Main loop: listen and respond.""" + self.is_running = True + + # Display welcome screen + if self.console and RICH_AVAILABLE: + self.console.print(Panel.fit( + "[bold]🎤 Voice Assistant - AI Now Inc[/bold]\n" + "[cyan]Laboratory Assistant: Claw 🏭[/cyan]\n\n" + "Say '[bold]Hey Osiris[/bold]' or '[bold]你好 Osiris[/bold]' to activate\n" + "Press [bold]Ctrl+C[/bold] to stop", + box=box.ROUNDED, + border_style="blue" + )) + else: + print("\n" + "="*60) + print(" 🎤 Voice Assistant - AI Now Inc") + print(" Laboratory Assistant: Claw 🏭") + print("="*60) + print("\n Say 'Hey Osiris' or '你好 Osiris' to activate") + print(" Press Ctrl+C to stop\n") + + # Set hotword callback + self.hotword_detector.set_callback(self._on_hotword_detected) + + # Main loop + self._update_status("🎤 Listening for hotword...") + + while self.is_running: + try: + if self.is_awake: + # Already activated, listen for command + self._update_status("👂 Listening for command...") + transcript = self.hotword_detector.listen_once() + + if transcript: + self._add_to_history('user', transcript) + self._update_status(f"📝 Heard: '{transcript}'") + + # Process with assistant + self._update_status("🤔 Processing...") + response = self.assistant.process(transcript) + + self._add_to_history('assistant', response) + self._update_status(f"💬 Response: '{response}'") + + # Speak response + self._update_status("🔊 Speaking...") + self.tts.speak(response, "en") + + self.is_awake = False + self._update_status("🎤 Listening for hotword...") + else: + self.is_awake = False + self._update_status("🎤 Listening for hotword...") + else: + # Wait for hotword + time.sleep(0.5) + + except Exception as e: + logger.error(f"Error in main loop: {e}") + self.is_awake = False + time.sleep(1) + + def run(self): + """Run the TUI voice assistant.""" + logger.info("Starting TUI voice assistant...") + + try: + # Initialize hotword detector + self.hotword_detector.start() + logger.info("Hotword detector started") + + # Run main loop + self.listen_and_respond() + + except KeyboardInterrupt: + logger.info("Interrupted by user") + except Exception as e: + logger.error(f"Fatal error: {e}") + raise + finally: + self.shutdown() + + def shutdown(self): + """Clean shutdown.""" + logger.info("Shutting down...") + self.is_running = False + + if hasattr(self, 'hotword_detector'): + self.hotword_detector.stop() + + if self.console and RICH_AVAILABLE: + self.console.print("\n[bold red]Voice Assistant stopped.[/bold red]") + else: + print("\nVoice Assistant stopped.") + + 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() + + # Check for rich + if not RICH_AVAILABLE: + 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.live import Live + from rich.layout import Layout + from rich.text import Text + from rich.table import Table + from rich import box + print("Rich installed successfully!") + + app = TUIAssistant(config_path=args.config) + app.run() + + +if __name__ == '__main__': + main()