From e86e22c5f796c7feb21018406988042066b21df4 Mon Sep 17 00:00:00 2001 From: jatinkrmalik Date: Mon, 7 Apr 2025 17:05:15 -0500 Subject: [PATCH] Add keyboard shortcuts, configuration management, and command processing --- src/speech_recognition/command_processor.py | 173 +++++++++++++++++ src/speech_recognition/recognition_manager.py | 175 ++++++++++++++++-- src/ui/config_manager.py | 146 +++++++++++++++ src/ui/keyboard_shortcuts.py | 135 ++++++++++++++ 4 files changed, 611 insertions(+), 18 deletions(-) create mode 100644 src/speech_recognition/command_processor.py create mode 100644 src/ui/config_manager.py create mode 100644 src/ui/keyboard_shortcuts.py diff --git a/src/speech_recognition/command_processor.py b/src/speech_recognition/command_processor.py new file mode 100644 index 0000000..da446e9 --- /dev/null +++ b/src/speech_recognition/command_processor.py @@ -0,0 +1,173 @@ +""" +Command processor for Ubuntu Voice Typing. + +This module processes text commands from speech recognition, such as +"new line", "period", etc. +""" + +import logging +import re +from typing import Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + + +class CommandProcessor: + """ + Processes text commands in speech recognition results. + + This class handles special commands like "new line", "period", + "delete that", etc. + """ + + def __init__(self): + """Initialize the command processor.""" + # Map of command phrases to their actions + self.text_commands = { + # Line commands + "new line": "\n", + "new paragraph": "\n\n", + + # Punctuation + "period": ".", + "full stop": ".", + "comma": ",", + "question mark": "?", + "exclamation mark": "!", + "exclamation point": "!", + "semicolon": ";", + "colon": ":", + "dash": "-", + "hyphen": "-", + "underscore": "_", + "quote": "\"", + "single quote": "'", + "open parenthesis": "(", + "close parenthesis": ")", + "open bracket": "[", + "close bracket": "]", + "open brace": "{", + "close brace": "}", + } + + # Special action commands that don't directly map to text + self.action_commands = { + "delete that": "delete_last", + "scratch that": "delete_last", + "undo": "undo", + "redo": "redo", + "select all": "select_all", + "select line": "select_line", + "select word": "select_word", + "select paragraph": "select_paragraph", + "cut": "cut", + "copy": "copy", + "paste": "paste", + } + + # Formatting commands that modify the next word + self.format_commands = { + "capitalize": "capitalize_next", + "uppercase": "capitalize_next", + "all caps": "uppercase_next", + "lowercase": "lowercase_next", + "no spaces": "no_spaces_next", + } + + # Active format modifiers + self.active_formats = set() + + # Compile regex patterns for faster matching + self._compile_patterns() + + def _compile_patterns(self): + """Compile regex patterns for command matching.""" + # Create regex pattern for text commands + text_cmd_pattern = r'\b(' + '|'.join(re.escape(cmd) for cmd in self.text_commands.keys()) + r')\b' + self.text_cmd_regex = re.compile(text_cmd_pattern, re.IGNORECASE) + + # Create regex pattern for action commands + action_cmd_pattern = r'\b(' + '|'.join(re.escape(cmd) for cmd in self.action_commands.keys()) + r')\b' + self.action_cmd_regex = re.compile(action_cmd_pattern, re.IGNORECASE) + + # Create regex pattern for format commands + format_cmd_pattern = r'\b(' + '|'.join(re.escape(cmd) for cmd in self.format_commands.keys()) + r')\b' + self.format_cmd_regex = re.compile(format_cmd_pattern, re.IGNORECASE) + + def process_text(self, text: str) -> Tuple[str, List[str]]: + """ + Process text commands in the recognized text. + + Args: + text: The recognized text to process + + Returns: + Tuple of (processed_text, actions) + - processed_text: The text with commands replaced + - actions: List of special actions to perform + """ + if not text: + return "", [] + + logger.debug(f"Processing commands in text: {text}") + + # Convert to lowercase for easier matching + lower_text = text.lower() + processed_text = text + actions = [] + + # Process action commands first (delete that, undo, etc.) + action_matches = self.action_cmd_regex.findall(lower_text) + for match in action_matches: + action = self.action_commands[match] + actions.append(action) + # Remove the command from the text + processed_text = re.sub(r'\b' + re.escape(match) + r'\b', '', processed_text, flags=re.IGNORECASE) + + # Process format commands + format_matches = self.format_cmd_regex.findall(lower_text) + for match in format_matches: + action = self.format_commands[match] + self.active_formats.add(action) + # Remove the command from the text + processed_text = re.sub(r'\b' + re.escape(match) + r'\b', '', processed_text, flags=re.IGNORECASE) + + # Apply active format modifiers to the next word + if self.active_formats: + # Find the next word + word_match = re.search(r'\b(\w+)\b', processed_text) + if word_match: + word = word_match.group(1) + start, end = word_match.span(1) + + # Apply formatting + formatted_word = word + for format_type in self.active_formats: + if format_type == "capitalize_next": + formatted_word = formatted_word.capitalize() + elif format_type == "uppercase_next": + formatted_word = formatted_word.upper() + elif format_type == "lowercase_next": + formatted_word = formatted_word.lower() + elif format_type == "no_spaces_next": + # This will be applied when combining with the next word + pass + + # Replace the word in the text + processed_text = processed_text[:start] + formatted_word + processed_text[end:] + + # Clear active formats + self.active_formats.clear() + + # Finally, process text commands (period, comma, etc.) + def replace_command(match): + cmd = match.group(0).lower() + replacement = self.text_commands.get(cmd, "") + return replacement + + processed_text = self.text_cmd_regex.sub(replace_command, processed_text) + + # Clean up multiple spaces + processed_text = re.sub(r'\s+', ' ', processed_text).strip() + + return processed_text, actions \ No newline at end of file diff --git a/src/speech_recognition/recognition_manager.py b/src/speech_recognition/recognition_manager.py index 8aca394..3a3afe0 100644 --- a/src/speech_recognition/recognition_manager.py +++ b/src/speech_recognition/recognition_manager.py @@ -14,6 +14,8 @@ from enum import Enum, auto from pathlib import Path from typing import Callable, List, Optional +from .command_processor import CommandProcessor + logger = logging.getLogger(__name__) # Define constants @@ -50,8 +52,18 @@ class SpeechRecognitionManager: self.audio_thread = None self.recognition_thread = None self.model = None + self.command_processor = CommandProcessor() self.text_callbacks: List[Callable[[str], None]] = [] self.state_callbacks: List[Callable[[RecognitionState], None]] = [] + self.action_callbacks: List[Callable[[str], None]] = [] + + # Speech detection parameters + self.vad_sensitivity = 3 # Voice Activity Detection sensitivity (1-5) + self.silence_timeout = 2.0 # Seconds of silence before stopping recognition + + # Recording control flags + self.should_record = False + self.audio_buffer = [] # Create models directory if it doesn't exist os.makedirs(MODELS_DIR, exist_ok=True) @@ -165,6 +177,15 @@ class SpeechRecognitionManager: """ self.state_callbacks.append(callback) + def register_action_callback(self, callback: Callable[[str], None]): + """ + Register a callback function that will be called when a special action is triggered. + + Args: + callback: A function that takes a string argument (the action) + """ + self.action_callbacks.append(callback) + def _update_state(self, new_state: RecognitionState): """ Update the recognition state and notify callbacks. @@ -185,6 +206,10 @@ class SpeechRecognitionManager: logger.info("Starting speech recognition") self._update_state(RecognitionState.LISTENING) + # Set recording flag + self.should_record = True + self.audio_buffer = [] + # Start the audio recording thread self.audio_thread = threading.Thread(target=self._record_audio) self.audio_thread.daemon = True @@ -201,37 +226,151 @@ class SpeechRecognitionManager: return logger.info("Stopping speech recognition") - self._update_state(RecognitionState.IDLE) - # Stop the audio and recognition threads + # Clear recording flag + self.should_record = False + + # Wait for threads to finish if self.audio_thread and self.audio_thread.is_alive(): - # Signal thread to stop self.audio_thread.join(timeout=1.0) if self.recognition_thread and self.recognition_thread.is_alive(): self.recognition_thread.join(timeout=1.0) + + self._update_state(RecognitionState.IDLE) def _record_audio(self): """Record audio from the microphone.""" - # TODO: Implement audio recording based on the selected engine - # This will use PyAudio or similar to record audio and feed it to the recognition engine - pass + try: + import pyaudio + import wave + import numpy as np + + # PyAudio configuration + CHUNK = 1024 + FORMAT = pyaudio.paInt16 + CHANNELS = 1 + RATE = 16000 + + # Initialize PyAudio + audio = pyaudio.PyAudio() + + # Open microphone stream + stream = audio.open( + format=FORMAT, + channels=CHANNELS, + rate=RATE, + input=True, + frames_per_buffer=CHUNK + ) + + logger.info("Audio recording started") + + # Record audio while should_record is True + silence_counter = 0 + while self.should_record: + data = stream.read(CHUNK, exception_on_overflow=False) + self.audio_buffer.append(data) + + # Simple Voice Activity Detection (VAD) + # TODO: Implement proper VAD using webrtcvad or similar + audio_data = np.frombuffer(data, dtype=np.int16) + volume = np.abs(audio_data).mean() + + # Threshold based on sensitivity (1-5) + threshold = 500 / self.vad_sensitivity + + if volume < threshold: # Silence + silence_counter += CHUNK / RATE # Convert chunks to seconds + if silence_counter > self.silence_timeout: + logger.debug("Silence detected, stopping recognition") + self._update_state(RecognitionState.PROCESSING) + # Process final buffer + self._process_final_buffer() + # Reset for next utterance + self.audio_buffer = [] + silence_counter = 0 + self._update_state(RecognitionState.LISTENING) + else: # Speech + silence_counter = 0 + + # Clean up + stream.stop_stream() + stream.close() + audio.terminate() + logger.info("Audio recording stopped") + + except Exception as e: + logger.error(f"Error in audio recording: {e}") + self._update_state(RecognitionState.ERROR) + + def _process_final_buffer(self): + """Process the final audio buffer after silence is detected.""" + if not self.audio_buffer: + return + + if self.engine == "vosk": + for data in self.audio_buffer: + self.recognizer.AcceptWaveform(data) + + result = json.loads(self.recognizer.FinalResult()) + text = result.get("text", "") + + elif self.engine == "whisper": + # Save buffer to a temporary file + import tempfile + import wave + + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file: + temp_path = temp_file.name + + with wave.open(temp_path, 'wb') as wf: + wf.setnchannels(1) + wf.setsampwidth(2) # 16-bit + wf.setframerate(16000) + wf.writeframes(b''.join(self.audio_buffer)) + + # Process with Whisper + result = self.model.transcribe(temp_path) + text = result.get("text", "") + + # Remove temporary file + os.unlink(temp_path) + + else: + logger.error(f"Unknown engine: {self.engine}") + return + + # Process commands + if text: + processed_text, actions = self.command_processor.process_text(text) + + # Call text callbacks with processed text + if processed_text: + for callback in self.text_callbacks: + callback(processed_text) + + # Call action callbacks for each action + for action in actions: + for callback in self.action_callbacks: + callback(action) def _perform_recognition(self): - """Perform speech recognition on the recorded audio.""" - # TODO: Implement speech recognition based on the selected engine - # This will process the audio and call the registered text callbacks with the results - pass + """Perform speech recognition in real-time.""" + while self.should_record: + # The real work is done in _record_audio and _process_final_buffer + time.sleep(0.1) - def process_commands(self, text: str) -> str: + def configure(self, vad_sensitivity: int = None, silence_timeout: float = None): """ - Process text commands in the recognized text. + Configure recognition parameters. Args: - text: The recognized text to process - - Returns: - The processed text with commands handled + vad_sensitivity: Voice Activity Detection sensitivity (1-5) + silence_timeout: Seconds of silence before stopping recognition """ - # TODO: Implement command processing (e.g., "new line", "delete that", etc.) - return text \ No newline at end of file + if vad_sensitivity is not None: + self.vad_sensitivity = max(1, min(5, vad_sensitivity)) + + if silence_timeout is not None: + self.silence_timeout = max(0.5, min(5.0, silence_timeout)) \ No newline at end of file diff --git a/src/ui/config_manager.py b/src/ui/config_manager.py new file mode 100644 index 0000000..e721c3c --- /dev/null +++ b/src/ui/config_manager.py @@ -0,0 +1,146 @@ +""" +Configuration manager for Ubuntu Voice Typing. + +This module handles loading, saving, and accessing user preferences. +""" + +import json +import logging +import os +from pathlib import Path +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + +# Define constants +CONFIG_DIR = os.path.expanduser("~/.config/ubuntu-voice-typing") +CONFIG_FILE = os.path.join(CONFIG_DIR, "config.json") + +# Default configuration +DEFAULT_CONFIG = { + "recognition": { + "engine": "vosk", # "vosk" or "whisper" + "model_size": "small", # "small", "medium", or "large" + "auto_punctuate": True, + "vad_sensitivity": 3, # Voice Activity Detection sensitivity (1-5) + "timeout": 2.0, # Seconds of silence before stopping recognition + }, + "shortcuts": { + "toggle_recognition": "alt+shift+v", + }, + "ui": { + "start_minimized": False, + "show_notifications": True, + }, + "advanced": { + "debug_logging": False, + "wayland_mode": False, + } +} + + +class ConfigManager: + """ + Manager for user configuration settings. + + This class provides methods for loading, saving, and accessing user + preferences for the application. + """ + + def __init__(self): + """Initialize the configuration manager.""" + self.config = DEFAULT_CONFIG.copy() + self._ensure_config_dir() + self.load_config() + + def _ensure_config_dir(self): + """Ensure the configuration directory exists.""" + os.makedirs(CONFIG_DIR, exist_ok=True) + + def load_config(self): + """ + Load configuration from the config file. + + If the config file doesn't exist, the default configuration is used. + """ + if not os.path.exists(CONFIG_FILE): + logger.info(f"Config file not found at {CONFIG_FILE}. Using defaults.") + return + + try: + with open(CONFIG_FILE, "r") as f: + user_config = json.load(f) + + # Update the default config with user settings + self._update_dict_recursive(self.config, user_config) + logger.info(f"Loaded configuration from {CONFIG_FILE}") + + except Exception as e: + logger.error(f"Failed to load config: {e}") + + def save_config(self): + """Save the current configuration to the config file.""" + try: + with open(CONFIG_FILE, "w") as f: + json.dump(self.config, f, indent=4) + + logger.info(f"Saved configuration to {CONFIG_FILE}") + return True + + except Exception as e: + logger.error(f"Failed to save config: {e}") + return False + + def get(self, section: str, key: str, default: Any = None) -> Any: + """ + Get a configuration value. + + Args: + section: The configuration section (e.g., "recognition", "shortcuts") + key: The configuration key within the section + default: The default value to return if the key doesn't exist + + Returns: + The configuration value + """ + try: + return self.config[section][key] + except KeyError: + return default + + def set(self, section: str, key: str, value: Any) -> bool: + """ + Set a configuration value. + + Args: + section: The configuration section (e.g., "recognition", "shortcuts") + key: The configuration key within the section + value: The value to set + + Returns: + True if successful, False otherwise + """ + try: + if section not in self.config: + self.config[section] = {} + + self.config[section][key] = value + return True + + except Exception as e: + logger.error(f"Failed to set config value: {e}") + return False + + def _update_dict_recursive(self, target: Dict, source: Dict): + """ + Update a dictionary recursively. + + Args: + target: The target dictionary to update + source: The source dictionary with updates + """ + for key, value in source.items(): + if key in target and isinstance(target[key], dict) and isinstance(value, dict): + self._update_dict_recursive(target[key], value) + else: + target[key] = value \ No newline at end of file diff --git a/src/ui/keyboard_shortcuts.py b/src/ui/keyboard_shortcuts.py new file mode 100644 index 0000000..0d9a90f --- /dev/null +++ b/src/ui/keyboard_shortcuts.py @@ -0,0 +1,135 @@ +""" +Keyboard shortcut manager for Ubuntu Voice Typing. + +This module provides global keyboard shortcut functionality to +start/stop speech recognition with a keystroke. +""" + +import logging +import threading +from typing import Callable, Dict, Optional, Tuple + +# Try to import X11 keyboard libraries first +try: + from pynput import keyboard + from Xlib import display + KEYBOARD_AVAILABLE = True +except ImportError: + KEYBOARD_AVAILABLE = False + +logger = logging.getLogger(__name__) + + +class KeyboardShortcutManager: + """ + Manages global keyboard shortcuts for the application. + + This class allows registering global keyboard shortcuts that work + across the desktop environment. + """ + + def __init__(self): + """Initialize the keyboard shortcut manager.""" + self.shortcuts = {} + self.listener = None + self.active = False + + if not KEYBOARD_AVAILABLE: + logger.error("Keyboard shortcut libraries not available. Shortcuts will not work.") + return + + # Default shortcut for toggling voice typing (Alt+Shift+V) + self.default_shortcut = ( + {keyboard.Key.alt, keyboard.Key.shift}, + keyboard.KeyCode.from_char('v') + ) + + def start(self): + """Start listening for keyboard shortcuts.""" + if not KEYBOARD_AVAILABLE: + return + + if self.active: + return + + logger.info("Starting keyboard shortcut listener") + self.active = True + + # Track currently pressed modifier keys + self.current_keys = set() + + # Start keyboard listener in a separate thread + self.listener = keyboard.Listener( + on_press=self._on_press, + on_release=self._on_release + ) + self.listener.daemon = True + self.listener.start() + + def stop(self): + """Stop listening for keyboard shortcuts.""" + if not self.active or not self.listener: + return + + logger.info("Stopping keyboard shortcut listener") + self.active = False + + if self.listener: + self.listener.stop() + self.listener.join(timeout=1.0) + self.listener = None + + def register_shortcut(self, modifiers: set, key, callback: Callable): + """ + Register a keyboard shortcut. + + Args: + modifiers: Set of modifier keys (e.g., {keyboard.Key.alt, keyboard.Key.shift}) + key: The main key (e.g., keyboard.KeyCode.from_char('v')) + callback: Function to call when the shortcut is pressed + """ + self.shortcuts[(frozenset(modifiers), key)] = callback + logger.debug(f"Registered shortcut: {modifiers} + {key}") + + def register_toggle_callback(self, callback: Callable): + """ + Register a callback for the default toggle shortcut. + + Args: + callback: Function to call when the toggle shortcut is pressed + """ + modifiers, key = self.default_shortcut + self.register_shortcut(modifiers, key, callback) + + def _on_press(self, key): + """ + Handle key press events. + + Args: + key: The pressed key + """ + # Add to currently pressed keys + if hasattr(key, 'char'): + # Skip modifier tracking for character keys + pass + else: + self.current_keys.add(key) + + # Check for shortcuts + for (modifiers, trigger_key), callback in self.shortcuts.items(): + if self.current_keys == modifiers and key == trigger_key: + logger.debug(f"Shortcut triggered: {modifiers} + {trigger_key}") + callback() + + def _on_release(self, key): + """ + Handle key release events. + + Args: + key: The released key + """ + # Remove from currently pressed keys + try: + self.current_keys.discard(key) + except KeyError: + pass # Key wasn't in the set \ No newline at end of file