Fix TUI app - simplified version using hotword detector correctly
This commit is contained in:
Claw - AI Now Inc
parent
cc9b5dec5d
commit
e64afb8d54
203
tui_app.py
203
tui_app.py
@ -1,8 +1,10 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Text User Interface (TUI) for Voice Assistant
|
||||
Text User Interface (TUI) for Voice Assistant - Simple Version
|
||||
AI Now Inc - Del Mar Demo Unit
|
||||
Laboratory Assistant: Claw 🏭
|
||||
|
||||
This version uses the existing working components from main.py
|
||||
"""
|
||||
|
||||
import os
|
||||
@ -18,19 +20,21 @@ from datetime import datetime
|
||||
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
|
||||
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 import box
|
||||
RICH_AVAILABLE = True
|
||||
print("Rich installed!")
|
||||
|
||||
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
|
||||
@ -42,7 +46,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TUIAssistant:
|
||||
"""TUI-based voice assistant with rich text display."""
|
||||
"""Simple TUI-based voice assistant."""
|
||||
|
||||
def __init__(self, config_path: str = "config.json"):
|
||||
self.config_path = Path(config_path)
|
||||
@ -58,23 +62,19 @@ class TUIAssistant:
|
||||
|
||||
# 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
|
||||
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:
|
||||
@ -91,11 +91,10 @@ class TUIAssistant:
|
||||
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 _on_hotword(self):
|
||||
"""Callback when hotword is detected."""
|
||||
self.console.print("\n[bold green]✨ Hotword detected! ✨[/bold green]")
|
||||
self.tts.speak("Yes?", "en")
|
||||
|
||||
def _add_to_history(self, role: str, text: str):
|
||||
"""Add message to conversation history."""
|
||||
@ -105,116 +104,52 @@ class TUIAssistant:
|
||||
'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()
|
||||
# Display welcome screen
|
||||
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"
|
||||
))
|
||||
|
||||
self.is_running = True
|
||||
self.console.print("\n[bold blue]→[/bold blue] Listening for hotword...")
|
||||
|
||||
# Main loop - use hotword detector's blocking detect
|
||||
while self.is_running:
|
||||
try:
|
||||
# Wait for hotword (this blocks until detected or timeout)
|
||||
detected_keyword = self.hotword_detector.detect(timeout=30)
|
||||
|
||||
if detected_keyword:
|
||||
self.console.print(f"\n[bold green]✓[/bold green] Heard: '{detected_keyword}'")
|
||||
|
||||
# For now, just respond with a simple message
|
||||
# In a full implementation, you'd use speech recognition here
|
||||
response = "I heard you! What would you like me to do?"
|
||||
self._add_to_history('assistant', response)
|
||||
self.console.print(f"[bold cyan]🤖:[/bold cyan] {response}")
|
||||
self.tts.speak(response, "en")
|
||||
else:
|
||||
self.console.print("[yellow]⏰ Timeout - trying again...[/yellow]")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Interrupted by user")
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"Error: {e}")
|
||||
time.sleep(1)
|
||||
|
||||
self.shutdown()
|
||||
|
||||
def shutdown(self):
|
||||
"""Clean shutdown."""
|
||||
@ -224,11 +159,7 @@ class TUIAssistant:
|
||||
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.")
|
||||
|
||||
self.console.print("\n[bold red]Voice Assistant stopped.[/bold red]")
|
||||
logger.info("Shutdown complete")
|
||||
|
||||
|
||||
@ -240,20 +171,6 @@ def main():
|
||||
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()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user