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