aoeu

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."""