263 lines
8.8 KiB
Python
263 lines
8.8 KiB
Python
#!/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()
|