diff --git a/src/tooling/stt_cli.py b/src/tooling/stt_cli.py index def935a..bf7972c 100644 --- a/src/tooling/stt_cli.py +++ b/src/tooling/stt_cli.py @@ -22,6 +22,12 @@ from rich.live import Live from rich.text import Text from rich.table import Table +try: + import rumps + RUMPS_AVAILABLE = True +except ImportError: + RUMPS_AVAILABLE = False + # Create STT app that can be imported as a subcommand stt_app = typer.Typer( name="stt", @@ -84,6 +90,308 @@ class TranscriptionDisplay: self.final_text += f"[{timestamp}] {text}\n" +if RUMPS_AVAILABLE: + class STTStatusBarApp(rumps.App): + """macOS Status Bar App for controlling STT functionality.""" + + def __init__(self, name="STT", **kwargs): + super().__init__(name, **kwargs) + + # Initialize state + self.recorder = None + self.is_running = False + self.is_paused = False + self.output_file = None + self.transcription_thread = None + self.stop_event = threading.Event() + + # Configuration + self.wake_word = "jarvis" + self.model = "base" + self.language = "" + self.realtime = True + self.sensitivity = 0.6 + self.device = "auto" + self.save_to_file = None + + # Menu setup + self.menu = [ + rumps.MenuItem("Start STT", callback=self.start_stt), + rumps.MenuItem("Stop STT", callback=self.stop_stt), + rumps.separator, + rumps.MenuItem("Pause", callback=self.pause_stt), + rumps.MenuItem("Resume", callback=self.resume_stt), + rumps.separator, + { + "Settings": { + "Wake Word": { + "jarvis": rumps.MenuItem("jarvis", callback=self.set_wake_word), + "alexa": rumps.MenuItem("alexa", callback=self.set_wake_word), + "hey google": rumps.MenuItem("hey google", callback=self.set_wake_word), + "hey siri": rumps.MenuItem("hey siri", callback=self.set_wake_word), + "computer": rumps.MenuItem("computer", callback=self.set_wake_word), + }, + "Model": { + "tiny": rumps.MenuItem("tiny", callback=self.set_model), + "base": rumps.MenuItem("base", callback=self.set_model), + "small": rumps.MenuItem("small", callback=self.set_model), + "medium": rumps.MenuItem("medium", callback=self.set_model), + } + } + }, + rumps.separator, + rumps.MenuItem("Show Recent Transcriptions", callback=self.show_transcriptions), + rumps.MenuItem("Save to File...", callback=self.select_output_file), + rumps.separator, + rumps.MenuItem("Quit", callback=rumps.quit_application), + ] + + # Set initial menu states + self.update_menu_states() + self.set_current_wake_word() + self.set_current_model() + + def update_menu_states(self): + """Update menu item states based on current status.""" + if self.is_running: + if self.is_paused: + self.title = "πŸŽ™οΈβΈοΈ" + self.menu["Start STT"].set_callback(None) + self.menu["Stop STT"].set_callback(self.stop_stt) + self.menu["Pause"].set_callback(None) + self.menu["Resume"].set_callback(self.resume_stt) + else: + self.title = "πŸŽ™οΈπŸ”΄" + self.menu["Start STT"].set_callback(None) + self.menu["Stop STT"].set_callback(self.stop_stt) + self.menu["Pause"].set_callback(self.pause_stt) + self.menu["Resume"].set_callback(None) + else: + self.title = "πŸŽ™οΈβš«" + self.menu["Start STT"].set_callback(self.start_stt) + self.menu["Stop STT"].set_callback(None) + self.menu["Pause"].set_callback(None) + self.menu["Resume"].set_callback(None) + + def set_current_wake_word(self): + """Set checkmark for current wake word.""" + for word in ["jarvis", "alexa", "hey google", "hey siri", "computer"]: + self.menu["Settings"]["Wake Word"][word].state = (word == self.wake_word) + + def set_current_model(self): + """Set checkmark for current model.""" + for model in ["tiny", "base", "small", "medium"]: + self.menu["Settings"]["Model"][model].state = (model == self.model) + + def set_wake_word(self, sender): + """Set the wake word from menu selection.""" + self.wake_word = sender.title + self.set_current_wake_word() + if self.is_running: + rumps.notification("STT Settings", "Wake Word Changed", f"Restart STT to use '{self.wake_word}'") + + def set_model(self, sender): + """Set the model from menu selection.""" + self.model = sender.title + self.set_current_model() + if self.is_running: + rumps.notification("STT Settings", "Model Changed", f"Restart STT to use '{self.model}' model") + + def start_stt(self, _): + """Start STT functionality.""" + if self.is_running: + return + + try: + from RealtimeSTT import AudioToTextRecorder + except ImportError: + rumps.alert("RealtimeSTT not installed", "Please install with: pip install RealtimeSTT") + return + + try: + # Determine device + if self.device == "auto": + try: + import torch + self.device = "cuda" if torch.cuda.is_available() else "cpu" + except ImportError: + self.device = "cpu" + + # Setup output file if specified + if self.save_to_file: + self.save_to_file.parent.mkdir(parents=True, exist_ok=True) + self.output_file = open(self.save_to_file, 'a', encoding='utf-8') + timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + self.output_file.write(f"\n=== STT Session Started: {timestamp} ===\n") + self.output_file.flush() + + # Configure recorder + recorder_config = { + "model": self.model, + "wake_words": self.wake_word, + "wake_words_sensitivity": self.sensitivity, + "device": self.device, + "on_recording_start": self.on_recording_start, + "on_recording_stop": self.on_recording_stop, + "on_wakeword_detected": self.on_wakeword_detected, + "on_wakeword_timeout": self.on_wakeword_timeout, + "on_wakeword_detection_start": self.on_wakeword_detection_start, + } + + if self.language: + recorder_config["language"] = self.language + + if self.realtime: + recorder_config.update({ + "enable_realtime_transcription": True, + "on_realtime_transcription_update": self.on_realtime_transcription, + }) + + # Initialize recorder + self.recorder = AudioToTextRecorder(**recorder_config) + + # Start transcription thread + self.is_running = True + self.is_paused = False + self.stop_event.clear() + self.transcription_thread = threading.Thread(target=self.transcription_loop, daemon=True) + self.transcription_thread.start() + + self.update_menu_states() + rumps.notification("STT Started", f"Wake word: {self.wake_word}", f"Model: {self.model} | Device: {self.device}") + + except Exception as e: + rumps.alert("STT Error", f"Failed to start STT: {str(e)}") + + def stop_stt(self, _): + """Stop STT functionality.""" + if not self.is_running: + return + + self.is_running = False + self.is_paused = False + self.stop_event.set() + + if self.recorder: + try: + self.recorder.shutdown() + except: + pass + self.recorder = None + + if self.output_file: + timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + self.output_file.write(f"=== STT Session Ended: {timestamp} ===\n\n") + self.output_file.close() + self.output_file = None + + self.update_menu_states() + rumps.notification("STT Stopped", "Speech-to-text has been stopped", "") + + def pause_stt(self, _): + """Pause STT functionality.""" + if not self.is_running or self.is_paused: + return + + self.is_paused = True + self.update_menu_states() + rumps.notification("STT Paused", "Speech recognition paused", "Resume from menu") + + def resume_stt(self, _): + """Resume STT functionality.""" + if not self.is_running or not self.is_paused: + return + + self.is_paused = False + self.update_menu_states() + rumps.notification("STT Resumed", "Speech recognition resumed", f"Listening for '{self.wake_word}'") + + def transcription_loop(self): + """Main transcription loop running in background thread.""" + while self.is_running and not self.stop_event.is_set(): + try: + if not self.is_paused and self.recorder: + text = self.recorder.text() + if text and text.strip(): + self.on_transcription_complete(text) + else: + time.sleep(0.1) # Small delay when paused + except Exception as e: + print(f"Transcription error: {e}") + time.sleep(1) # Longer delay on error + + def on_realtime_transcription(self, text: str): + """Handle real-time transcription updates.""" + # Could update a notification or log + pass + + def on_transcription_complete(self, text: str): + """Handle completed transcriptions.""" + if text.strip(): + # Show notification with transcription + rumps.notification("Transcription", "Speech detected:", text[:100] + ("..." if len(text) > 100 else "")) + + # Save to file if specified + if self.output_file: + timestamp = datetime.datetime.now().strftime("%H:%M:%S") + self.output_file.write(f"[{timestamp}] {text}\n") + self.output_file.flush() + + def on_recording_start(self): + """Called when recording starts.""" + pass + + def on_recording_stop(self): + """Called when recording stops.""" + pass + + def on_wakeword_detected(self): + """Called when wake word is detected.""" + pass + + def on_wakeword_timeout(self): + """Called when wake word times out.""" + pass + + def on_wakeword_detection_start(self): + """Called when starting to listen for wake words.""" + pass + + def show_transcriptions(self, _): + """Show recent transcriptions in an alert.""" + if self.output_file and self.save_to_file and self.save_to_file.exists(): + try: + with open(self.save_to_file, 'r', encoding='utf-8') as f: + lines = f.readlines() + recent = ''.join(lines[-10:]) # Last 10 lines + if recent.strip(): + rumps.alert("Recent Transcriptions", recent) + else: + rumps.alert("No Transcriptions", "No transcriptions found yet.") + except Exception as e: + rumps.alert("Error", f"Could not read transcriptions: {e}") + else: + rumps.alert("No File", "No output file configured. Use 'Save to File...' to set one.") + + def select_output_file(self, _): + """Select output file for saving transcriptions.""" + try: + # Simple file selection using input dialog + response = rumps.Window( + message="Enter filename for transcriptions:", + title="Save Transcriptions", + default_text="stt_transcriptions.txt", + ok="Save", + cancel="Cancel" + ).run() + + if response.clicked and response.text: + self.save_to_file = Path.home() / "Documents" / response.text + rumps.notification("File Set", "Transcriptions will be saved to:", str(self.save_to_file)) + except Exception as e: + rumps.alert("Error", f"Could not set output file: {e}") + + @stt_app.command("listen") def listen_cmd( wake_word: str = typer.Option( @@ -433,13 +741,53 @@ def info_cmd(): "tooling stt listen --wake-word alexa # Use alexa wake word", "tooling stt listen --model tiny # Use faster tiny model", "tooling stt test --duration 5 # Test for 5 seconds", - "tooling stt listen --save-to-file transcripts.txt # Save to file" + "tooling stt listen --save-to-file transcripts.txt # Save to file", + "tooling stt statusbar # Launch status bar app" ] for example in examples: console.print(f" [dim]${example}[/dim]") +@stt_app.command("statusbar") +def statusbar_cmd(): + """Launch macOS status bar app for STT control.""" + + if not RUMPS_AVAILABLE: + console.print("[bold red]❌ rumps not available.[/bold red]") + console.print("Install with: [bold]pip install rumps[/bold]") + console.print("Note: rumps requires macOS and PyObjC") + raise typer.Exit(1) + + # Check RealtimeSTT installation + try: + from RealtimeSTT import AudioToTextRecorder + except ImportError: + console.print("[bold red]❌ RealtimeSTT not installed.[/bold red]") + console.print("Install with: [bold]pip install RealtimeSTT[/bold]") + raise typer.Exit(1) + + console.print(Panel( + "[bold]Starting STT Status Bar App[/bold]\n\n" + "β€’ Look for the πŸŽ™οΈ icon in your menu bar\n" + "β€’ Click to access STT controls\n" + "β€’ Use Start/Stop/Pause/Resume options\n" + "β€’ Configure wake words and models in Settings\n" + "β€’ Press [bold red]Ctrl+C[/bold red] here to quit", + title="πŸ“± Status Bar App", + border_style="green" + )) + + try: + app = STTStatusBarApp("STT") + app.run() + except KeyboardInterrupt: + console.print("\n[bold yellow]⚠️ Status bar app stopped.[/bold yellow]") + except Exception as e: + console.print(f"\n[bold red]❌ Status bar app error: {e}[/bold red]") + raise typer.Exit(1) + + # For backward compatibility when run directly def cli_main(): """Entry point for the STT CLI script when run directly."""