Add TUI version with rich library for better text display
This commit is contained in:
parent
b76d96fa23
commit
03b33060b4
@ -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
|
||||
|
||||
262
tui_app.py
Normal file
262
tui_app.py
Normal file
@ -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()
|
||||
Loading…
x
Reference in New Issue
Block a user