Add keyboard shortcuts, configuration management, and command processing

This commit is contained in:
jatinkrmalik
2025-04-07 17:05:15 -05:00
parent 40072f2e61
commit e86e22c5f7
4 changed files with 611 additions and 18 deletions
+173
View File
@@ -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
+157 -18
View File
@@ -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
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))
+146
View File
@@ -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
+135
View File
@@ -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