xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Test Structure Reorganization and CI Improvements (#5)

Browse Source 
* Add GitHub Actions CI/CD workflows and status badges

* Fix CI workflow dependency installation

* Fix CI workflow: remove pip caching and add explicit requirements.txt

* Fix CI workflow: add system dependencies for PyGObject and pycairo

* Refactor code for improved readability and consistency

- Cleaned up whitespace and formatting in audio_feedback.py, config_manager.py, keyboard_shortcuts.py, tray_indicator.py, run_basic_tests.py, and test_basic.py.
- Enhanced the structure of the code by aligning comments and ensuring consistent indentation.
- Updated function definitions and calls to maintain a uniform style across the codebase.
- Improved logging messages for better clarity during debugging.
- Ensured all files end with a newline for better compatibility with various tools.

* Refactor imports for consistency and clarity across multiple modules

* Fix test imports by setting PYTHONPATH and adding conftest.py

* Update CI workflow to use Python 3.12 and ensure proper import path in tests

* Fix CI: Add GObject Introspection development dependencies

* Update CI: Use virtual environment like local setup

* Fix CI

* Update README and setup.py: Change Python version to 3.8 and correct author email

* Restructure tests for better organization and maintainability

* Fix formatting

* Update CI workflow to remove redundant test step

* Fix CI tests by properly mocking VOSK model loading

* Update CI and Tests

* Simplify
This commit is contained in:
Jatin K Malik
2025-04-16 20:23:40 -05:00
committed by GitHub
parent 7590332598
commit 3645c692fd
20 changed files with 996 additions and 682 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.py
+21 -19
View File
@@ -26,20 +26,22 @@ logger = logging.getLogger(__name__)
def parse_arguments():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(description="Vocalinux")
parser.add_argument("--debug", action="store_true", help="Enable debug logging")
parser.add_argument(
"--debug", action="store_true", help="Enable debug logging"
"--model",
type=str,
default="small",
help="Speech recognition model size (small, medium, large)",
)
parser.add_argument(
"--model", type=str, default="small",
help="Speech recognition model size (small, medium, large)"
"--engine",
type=str,
default="vosk",
choices=["vosk", "whisper"],
help="Speech recognition engine to use",
)
parser.add_argument(
"--engine", type=str, default="vosk", choices=["vosk", "whisper"],
help="Speech recognition engine to use"
)
parser.add_argument(
"--wayland", action="store_true",
help="Force Wayland compatibility mode"
"--wayland", action="store_true", help="Force Wayland compatibility mode"
)
return parser.parse_args()
@@ -47,36 +49,36 @@ def parse_arguments():
def main():
"""Main entry point for the application."""
args = parse_arguments()
# Configure debug logging if requested
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
logger.debug("Debug logging enabled")
# Initialize main components
logger.info("Initializing Vocalinux...")
# Initialize speech recognition engine
speech_engine = recognition_manager.SpeechRecognitionManager(
engine=args.engine,
model_size=args.model,
)
# Initialize text injection system
text_system = text_injector.TextInjector(wayland_mode=args.wayland)
# Connect speech recognition to text injection
speech_engine.register_text_callback(text_system.inject_text)
# Initialize and start the system tray indicator
indicator = tray_indicator.TrayIndicator(
speech_engine=speech_engine,
text_injector=text_system,
)
# Start the GTK main loop
indicator.run()
if __name__ == "__main__":
main()
main()
src/speech_recognition/command_processor.py
+58 -35
View File
@@ -15,11 +15,11 @@ 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
@@ -27,7 +27,6 @@ class CommandProcessor:
# Line commands
"new line": "\n",
"new paragraph": "\n\n",
# Punctuation
"period": ".",
"full stop": ".",
@@ -40,7 +39,7 @@ class CommandProcessor:
"dash": "-",
"hyphen": "-",
"underscore": "_",
"quote": "\"",
"quote": '"',
"single quote": "'",
"open parenthesis": "(",
"close parenthesis": ")",
@@ -49,7 +48,7 @@ class CommandProcessor:
"open brace": "{",
"close brace": "}",
}
# Special action commands that don't directly map to text
self.action_commands = {
"delete that": "delete_last",
@@ -64,7 +63,7 @@ class CommandProcessor:
"copy": "copy",
"paste": "paste",
}
# Formatting commands that modify the next word
self.format_commands = {
"capitalize": "capitalize_next",
@@ -73,34 +72,46 @@ class CommandProcessor:
"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'
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'
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'
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
@@ -108,38 +119,48 @@ class CommandProcessor:
"""
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)
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)
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)
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:
@@ -152,22 +173,24 @@ class CommandProcessor:
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:]
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
processed_text = re.sub(r"\s+", " ", processed_text).strip()
return processed_text, actions
src/speech_recognition/recognition_manager.py
+118 -90
View File
@@ -19,24 +19,36 @@ from .command_processor import CommandProcessor
# Fix imports to use absolute imports instead of relative imports
try:
# Try absolute import first (when installed as a package)
from src.ui.audio_feedback import play_start_sound, play_stop_sound, play_error_sound
from src.ui.audio_feedback import (
play_error_sound,
play_start_sound,
play_stop_sound,
)
except ImportError:
try:
# Fallback for development environment
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
from src.ui.audio_feedback import play_start_sound, play_stop_sound, play_error_sound
sys.path.insert(
0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
)
from src.ui.audio_feedback import (
play_error_sound,
play_start_sound,
play_stop_sound,
)
except ImportError:
# Last resort fallback - define stub functions
def play_start_sound():
logging.warning("Audio feedback not available")
def play_stop_sound():
logging.warning("Audio feedback not available")
def play_error_sound():
logging.warning("Audio feedback not available")
logger = logging.getLogger(__name__)
# Define constants
@@ -45,6 +57,7 @@ MODELS_DIR = os.path.expanduser("~/.local/share/vocalinux/models")
class RecognitionState(Enum):
"""Enum representing the state of the speech recognition system."""
IDLE = auto()
LISTENING = auto()
PROCESSING = auto()
@@ -54,15 +67,15 @@ class RecognitionState(Enum):
class SpeechRecognitionManager:
"""
Manager class for speech recognition engines.
This class provides a unified interface for working with different
speech recognition engines (VOSK and Whisper).
"""
def __init__(self, engine: str = "vosk", model_size: str = "small"):
"""
Initialize the speech recognition manager.
Args:
engine: The speech recognition engine to use ("vosk" or "whisper")
model_size: The size of the model to use ("small", "medium", "large")
@@ -77,20 +90,22 @@ class SpeechRecognitionManager:
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)
logger.info(f"Initializing speech recognition with {engine} engine and {model_size} model")
logger.info(
f"Initializing speech recognition with {engine} engine and {model_size} model"
)
# Initialize the selected speech recognition engine
if engine == "vosk":
self._init_vosk()
@@ -98,39 +113,46 @@ class SpeechRecognitionManager:
self._init_whisper()
else:
raise ValueError(f"Unsupported speech recognition engine: {engine}")
def _init_vosk(self):
"""Initialize the VOSK speech recognition engine."""
try:
from vosk import Model, KaldiRecognizer
from vosk import KaldiRecognizer, Model
self.vosk_model_path = self._get_vosk_model_path()
if not os.path.exists(self.vosk_model_path):
logger.info(f"VOSK model not found at {self.vosk_model_path}. Downloading...")
logger.info(
f"VOSK model not found at {self.vosk_model_path}. Downloading..."
)
self._download_vosk_model()
logger.info(f"Loading VOSK model from {self.vosk_model_path}")
self.model = Model(self.vosk_model_path)
self.recognizer = KaldiRecognizer(self.model, 16000)
except ImportError:
logger.error("Failed to import VOSK. Please install it with 'pip install vosk'")
logger.error(
"Failed to import VOSK. Please install it with 'pip install vosk'"
)
self.state = RecognitionState.ERROR
raise
def _init_whisper(self):
"""Initialize the Whisper speech recognition engine."""
try:
import whisper
logger.info(f"Loading Whisper {self.model_size} model")
self.model = whisper.load_model(self.model_size)
except ImportError:
logger.error("Failed to import Whisper. Please install it with 'pip install whisper torch'")
logger.error(
"Failed to import Whisper. Please install it with 'pip install whisper torch'"
)
self.state = RecognitionState.ERROR
raise
def _get_vosk_model_path(self) -> str:
"""Get the path to the VOSK model based on the selected size."""
model_map = {
@@ -138,176 +160,182 @@ class SpeechRecognitionManager:
"medium": "vosk-model-en-us-0.22",
"large": "vosk-model-en-us-0.42",
}
model_name = model_map.get(self.model_size, model_map["small"])
return os.path.join(MODELS_DIR, model_name)
def _download_vosk_model(self):
"""Download the VOSK model if it doesn't exist."""
import zipfile
import requests
import tqdm
import zipfile
model_urls = {
"small": "https://alphacephei.com/vosk/models/vosk-model-small-en-us-0.15.zip",
"medium": "https://alphacephei.com/vosk/models/vosk-model-en-us-0.22.zip",
"large": "https://alphacephei.com/vosk/models/vosk-model-en-us-0.42.zip",
}
url = model_urls.get(self.model_size)
if not url:
raise ValueError(f"Unknown model size: {self.model_size}")
model_name = os.path.basename(url).replace(".zip", "")
model_path = os.path.join(MODELS_DIR, model_name)
zip_path = os.path.join(MODELS_DIR, os.path.basename(url))
# Download the model
logger.info(f"Downloading VOSK model from {url}")
response = requests.get(url, stream=True)
total_size = int(response.headers.get("content-length", 0))
with open(zip_path, "wb") as f:
for data in tqdm.tqdm(response.iter_content(chunk_size=1024), total=total_size//1024, unit="KB"):
for data in tqdm.tqdm(
response.iter_content(chunk_size=1024),
total=total_size // 1024,
unit="KB",
):
f.write(data)
# Extract the model
logger.info(f"Extracting VOSK model to {model_path}")
with zipfile.ZipFile(zip_path, "r") as zip_ref:
zip_ref.extractall(MODELS_DIR)
# Remove the zip file
os.remove(zip_path)
logger.info("VOSK model downloaded and extracted successfully")
def register_text_callback(self, callback: Callable[[str], None]):
"""
Register a callback function that will be called when text is recognized.
Args:
callback: A function that takes a string argument (the recognized text)
"""
self.text_callbacks.append(callback)
def register_state_callback(self, callback: Callable[[RecognitionState], None]):
"""
Register a callback function that will be called when the recognition state changes.
Args:
callback: A function that takes a RecognitionState argument
"""
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.
Args:
new_state: The new recognition state
"""
self.state = new_state
for callback in self.state_callbacks:
callback(new_state)
def start_recognition(self):
"""Start the speech recognition process."""
if self.state != RecognitionState.IDLE:
logger.warning(f"Cannot start recognition in current state: {self.state}")
return
logger.info("Starting speech recognition")
self._update_state(RecognitionState.LISTENING)
# Play the start sound
play_start_sound()
# 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
self.audio_thread.start()
# Start the recognition thread
self.recognition_thread = threading.Thread(target=self._perform_recognition)
self.recognition_thread.daemon = True
self.recognition_thread.start()
def stop_recognition(self):
"""Stop the speech recognition process."""
if self.state == RecognitionState.IDLE:
return
logger.info("Stopping speech recognition")
# Play the stop sound
play_stop_sound()
# Clear recording flag
self.should_record = False
# Wait for threads to finish
if self.audio_thread and self.audio_thread.is_alive():
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."""
try:
import pyaudio
import wave
import numpy as np
import pyaudio
# 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
frames_per_buffer=CHUNK,
)
logger.info("Audio recording started")
# Record audio while should_record is True
silence_counter = 0
while self.should_record:
try:
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:
@@ -324,85 +352,85 @@ class SpeechRecognitionManager:
except Exception as e:
logger.error(f"Error reading audio data: {e}")
break
# 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}")
play_error_sound()
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:
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))
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 in real-time."""
while self.should_record:
# The real work is done in _record_audio and _process_final_buffer
time.sleep(0.1)
def configure(self, vad_sensitivity: int = None, silence_timeout: float = None):
"""
Configure recognition parameters.
Args:
vad_sensitivity: Voice Activity Detection sensitivity (1-5)
silence_timeout: Seconds of silence before stopping recognition
"""
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))
self.silence_timeout = max(0.5, min(5.0, silence_timeout))
src/text_injection/text_injector.py
+119 -81
View File
@@ -18,6 +18,7 @@ logger = logging.getLogger(__name__)
class DesktopEnvironment(Enum):
"""Enum representing the desktop environment."""
X11 = "x11"
WAYLAND = "wayland"
WAYLAND_XDOTOOL = "wayland-xdotool" # Wayland with XWayland fallback
@@ -27,46 +28,59 @@ class DesktopEnvironment(Enum):
class TextInjector:
"""
Class for injecting text into the active application.
This class handles the injection of text into the currently focused
application window, supporting both X11 and Wayland environments.
"""
def __init__(self, wayland_mode: bool = False):
"""
Initialize the text injector.
Args:
wayland_mode: Force Wayland compatibility mode
"""
self.environment = self._detect_environment()
# Force Wayland mode if requested
if wayland_mode and self.environment == DesktopEnvironment.X11:
logger.info("Forcing Wayland compatibility mode")
self.environment = DesktopEnvironment.WAYLAND
logger.info(f"Using text injection for {self.environment.value} environment")
# Check for required tools
self._check_dependencies()
# Test if wtype actually works in this environment
if self.environment == DesktopEnvironment.WAYLAND and hasattr(self, 'wayland_tool') and self.wayland_tool == "wtype":
if (
self.environment == DesktopEnvironment.WAYLAND
and hasattr(self, "wayland_tool")
and self.wayland_tool == "wtype"
):
try:
# Try a test with wtype
result = subprocess.run(["wtype", "test"], stderr=subprocess.PIPE, text=True, check=False)
result = subprocess.run(
["wtype", "test"], stderr=subprocess.PIPE, text=True, check=False
)
error_output = result.stderr.lower()
if "compositor does not support" in error_output or result.returncode != 0:
logger.warning(f"Wayland compositor does not support virtual keyboard protocol: {error_output}")
if (
"compositor does not support" in error_output
or result.returncode != 0
):
logger.warning(
f"Wayland compositor does not support virtual keyboard protocol: {error_output}"
)
if shutil.which("xdotool"):
logger.info("Automatically switching to XWayland fallback with xdotool")
logger.info(
"Automatically switching to XWayland fallback with xdotool"
)
self.environment = DesktopEnvironment.WAYLAND_XDOTOOL
else:
logger.error("No fallback text injection method available")
except Exception as e:
logger.warning(f"Error testing wtype: {e}, will try to use it anyway")
# Verify XWayland fallback works - perform a test injection
if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL:
logger.info("Testing XWayland text injection fallback")
@@ -77,11 +91,11 @@ class TextInjector:
self._test_xdotool_fallback()
except Exception as e:
logger.error(f"XWayland fallback test failed: {e}")
def _detect_environment(self) -> DesktopEnvironment:
"""
Detect the current desktop environment (X11 or Wayland).
Returns:
The detected desktop environment
"""
@@ -97,22 +111,26 @@ class TextInjector:
elif "DISPLAY" in os.environ:
return DesktopEnvironment.X11
else:
logger.warning("Could not detect desktop environment, defaulting to X11")
logger.warning(
"Could not detect desktop environment, defaulting to X11"
)
return DesktopEnvironment.X11
def _check_dependencies(self):
"""Check for the required tools for text injection."""
if self.environment == DesktopEnvironment.X11:
# Check for xdotool
if not shutil.which("xdotool"):
logger.error("xdotool not found. Please install it with: sudo apt install xdotool")
logger.error(
"xdotool not found. Please install it with: sudo apt install xdotool"
)
raise RuntimeError("Missing required dependency: xdotool")
else:
# Check for wtype or ydotool for Wayland
wtype_available = shutil.which("wtype") is not None
ydotool_available = shutil.which("ydotool") is not None
xdotool_available = shutil.which("xdotool") is not None
if wtype_available:
self.wayland_tool = "wtype"
logger.info(f"Using {self.wayland_tool} for Wayland text injection")
@@ -122,35 +140,39 @@ class TextInjector:
elif xdotool_available:
# Fallback to xdotool with XWayland
self.environment = DesktopEnvironment.WAYLAND_XDOTOOL
logger.info("No native Wayland tools found. Using xdotool with XWayland as fallback")
logger.info(
"No native Wayland tools found. Using xdotool with XWayland as fallback"
)
else:
logger.error("No text injection tools found. Please install one of:\n"
"- wtype: sudo apt install wtype\n"
"- ydotool: sudo apt install ydotool\n"
"- xdotool: sudo apt install xdotool")
logger.error(
"No text injection tools found. Please install one of:\n"
"- wtype: sudo apt install wtype\n"
"- ydotool: sudo apt install ydotool\n"
"- xdotool: sudo apt install xdotool"
)
raise RuntimeError("Missing required dependencies for text injection")
def _test_xdotool_fallback(self):
"""Test if xdotool is working correctly with XWayland."""
try:
# Get the DISPLAY environment variable for XWayland
xwayland_display = os.environ.get("DISPLAY", ":0")
logger.debug(f"Using DISPLAY={xwayland_display} for XWayland")
# Try using xdotool with explicit DISPLAY setting
test_env = os.environ.copy()
test_env["DISPLAY"] = xwayland_display
# Check if we can get active window (less intrusive test)
window_id = subprocess.run(
["xdotool", "getwindowfocus"],
env=test_env,
stdout=subprocess.PIPE,
["xdotool", "getwindowfocus"],
env=test_env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False
check=False,
)
if window_id.returncode != 0 or "failed" in window_id.stderr.lower():
logger.warning(f"XWayland detection test failed: {window_id.stderr}")
# Try to force XWayland environment more explicitly
@@ -159,32 +181,39 @@ class TextInjector:
logger.debug("XWayland test successful")
except Exception as e:
logger.error(f"Failed to test XWayland fallback: {e}")
def inject_text(self, text: str):
"""
Inject text into the currently focused application.
Args:
text: The text to inject
"""
if not text or not text.strip():
return
logger.debug(f"Injecting text: {text}")
# Escape special characters for shell
escaped_text = self._escape_text(text)
try:
if self.environment == DesktopEnvironment.X11 or self.environment == DesktopEnvironment.WAYLAND_XDOTOOL:
if (
self.environment == DesktopEnvironment.X11
or self.environment == DesktopEnvironment.WAYLAND_XDOTOOL
):
self._inject_with_xdotool(escaped_text)
else:
try:
self._inject_with_wayland_tool(escaped_text)
except subprocess.CalledProcessError as e:
logger.warning(f"Wayland tool failed: {e}. Falling back to xdotool")
if "compositor does not support" in str(e).lower() and shutil.which("xdotool"):
logger.info("Automatically switching to XWayland fallback permanently")
if "compositor does not support" in str(e).lower() and shutil.which(
"xdotool"
):
logger.info(
"Automatically switching to XWayland fallback permanently"
)
self.environment = DesktopEnvironment.WAYLAND_XDOTOOL
self._inject_with_xdotool(escaped_text)
else:
@@ -192,43 +221,44 @@ class TextInjector:
except Exception as e:
logger.error(f"Failed to inject text: {e}")
from ..ui.audio_feedback import play_error_sound
play_error_sound() # Play error sound when text injection fails
def _escape_text(self, text: str) -> str:
"""
Escape special characters in the text for shell commands.
Args:
text: The original text
Returns:
The escaped text
"""
# Replace common special characters
replacements = {
"'": r"\'",
'"': r'\"',
'`': r'\`',
'\\': r'\\',
'$': r'\$',
'"': r"\"",
"`": r"\`",
"\\": r"\\",
"$": r"\$",
}
result = text
for char, replacement in replacements.items():
result = result.replace(char, replacement)
return result
def _inject_with_xdotool(self, text: str):
"""
Inject text using xdotool for X11 environments.
Args:
text: The text to inject (already escaped)
"""
# Create environment with explicit X11 settings for Wayland compatibility
env = os.environ.copy()
if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL:
# Force X11 backend for XWayland
env["GDK_BACKEND"] = "x11"
@@ -236,39 +266,39 @@ class TextInjector:
# Ensure DISPLAY is set correctly for XWayland
if "DISPLAY" not in env or not env["DISPLAY"]:
env["DISPLAY"] = ":0"
logger.debug(f"Using XWayland with DISPLAY={env['DISPLAY']}")
# Add a small delay to ensure text is injected properly
time.sleep(0.3) # Increased delay for better reliability
# Try to ensure the window has focus using more robust approach
try:
# Get current active window
active_window = subprocess.run(
["xdotool", "getactivewindow"],
env=env,
stdout=subprocess.PIPE,
["xdotool", "getactivewindow"],
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False
check=False,
)
if active_window.returncode == 0 and active_window.stdout.strip():
window_id = active_window.stdout.strip()
# Focus explicitly on that window
subprocess.run(
["xdotool", "windowactivate", "--sync", window_id],
env=env,
stdout=subprocess.DEVNULL,
["xdotool", "windowactivate", "--sync", window_id],
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False
check=False,
)
# Wait a moment for the focus to take effect
time.sleep(0.2)
except Exception as e:
logger.debug(f"Window focus command failed: {e}")
# Inject text using xdotool
try:
max_retries = 2
@@ -277,44 +307,50 @@ class TextInjector:
# Inject in smaller chunks to avoid issues with very long text
chunk_size = 20 # Reduced chunk size for better reliability
for i in range(0, len(text), chunk_size):
chunk = text[i:i+chunk_size]
chunk = text[i : i + chunk_size]
# First try with clearmodifiers
cmd = ["xdotool", "type", "--clearmodifiers", chunk]
result = subprocess.run(cmd, env=env, check=True, stderr=subprocess.PIPE, text=True)
result = subprocess.run(
cmd, env=env, check=True, stderr=subprocess.PIPE, text=True
)
# Add a larger delay between chunks
if i + chunk_size < len(text):
time.sleep(0.1)
logger.info(f"Text injected using xdotool: '{text[:20]}...' ({len(text)} chars)")
logger.info(
f"Text injected using xdotool: '{text[:20]}...' ({len(text)} chars)"
)
break # Successfully injected
except subprocess.CalledProcessError as chunk_error:
if retry < max_retries:
logger.warning(f"Retrying text injection (attempt {retry+1}/{max_retries}): {chunk_error.stderr}")
logger.warning(
f"Retrying text injection (attempt {retry+1}/{max_retries}): {chunk_error.stderr}"
)
time.sleep(0.5) # Wait before retry
else:
raise # Re-raise on final attempt
# Try to reset any stuck modifiers
try:
subprocess.run(
["xdotool", "key", "--clearmodifiers", "Escape"],
env=env,
stdout=subprocess.DEVNULL,
["xdotool", "key", "--clearmodifiers", "Escape"],
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False
check=False,
)
except Exception:
pass # Ignore any errors from this command
except subprocess.CalledProcessError as e:
logger.error(f"xdotool error: {e.stderr}")
raise
def _inject_with_wayland_tool(self, text: str):
"""
Inject text using a Wayland-compatible tool (wtype or ydotool).
Args:
text: The text to inject (already escaped)
"""
@@ -322,6 +358,8 @@ class TextInjector:
cmd = ["wtype", text]
else: # ydotool
cmd = ["ydotool", "type", text]
result = subprocess.run(cmd, check=True, stderr=subprocess.PIPE, text=True)
logger.info(f"Text injected using {self.wayland_tool}: '{text[:20]}...' ({len(text)} chars)")
logger.info(
f"Text injected using {self.wayland_tool}: '{text[:20]}...' ({len(text)} chars)"
)
src/ui/audio_feedback.py
+31 -23
View File
@@ -6,8 +6,8 @@ This module provides audio feedback for various recognition states.
import logging
import os
import subprocess
import shutil
import subprocess
from pathlib import Path
logger = logging.getLogger(__name__)
@@ -27,22 +27,22 @@ ERROR_SOUND = os.path.join(SOUNDS_DIR, "error.wav")
def _get_audio_player():
"""
Determine the best available audio player on the system.
Returns:
tuple: (player_command, format_supported)
"""
# Check for PulseAudio paplay (preferred)
if shutil.which("paplay"):
return "paplay", ["wav"]
# Check for ALSA aplay
if shutil.which("aplay"):
return "aplay", ["wav"]
# Check for play (from SoX)
if shutil.which("play"):
return "play", ["wav"]
# Check for mplayer
if shutil.which("mplayer"):
return "mplayer", ["wav"]
@@ -55,40 +55,48 @@ def _get_audio_player():
def _play_sound_file(sound_path):
"""
Play a sound file using the best available player.
Args:
sound_path: Path to the sound file
"""
if not os.path.exists(sound_path):
logger.warning(f"Sound file not found: {sound_path}")
return False
player, formats = _get_audio_player()
if not player:
return False
file_ext = os.path.splitext(sound_path)[1].lower().lstrip('.')
file_ext = os.path.splitext(sound_path)[1].lower().lstrip(".")
if file_ext not in formats:
logger.warning(f"Format {file_ext} not supported by {player}")
return False
try:
if player == "paplay":
subprocess.Popen([player, sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
subprocess.Popen(
[player, sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
elif player == "aplay":
subprocess.Popen([player, "-q", sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
subprocess.Popen(
[player, "-q", sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
elif player == "mplayer":
subprocess.Popen([player, "-really-quiet", sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
subprocess.Popen(
[player, "-really-quiet", sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
elif player == "play":
subprocess.Popen([player, "-q", sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
subprocess.Popen(
[player, "-q", sound_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return True
except Exception as e:
logger.error(f"Failed to play sound {sound_path}: {e}")
@@ -107,4 +115,4 @@ def play_stop_sound():
def play_error_sound():
"""Play the sound for error notifications."""
_play_sound_file(ERROR_SOUND)
_play_sound_file(ERROR_SOUND)
src/ui/config_manager.py
+31 -27
View File
@@ -19,11 +19,11 @@ 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"
"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
"timeout": 2.0, # Seconds of silence before stopping recognition
},
"shortcuts": {
"toggle_recognition": "alt+shift+v",
@@ -35,71 +35,71 @@ DEFAULT_CONFIG = {
"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
"""
@@ -107,40 +107,44 @@ class ConfigManager:
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):
if (
key in target
and isinstance(target[key], dict)
and isinstance(value, dict)
):
self._update_dict_recursive(target[key], value)
else:
target[key] = value
target[key] = value
src/ui/keyboard_shortcuts.py
+57 -43
View File
@@ -13,6 +13,7 @@ from typing import Callable, Dict, Optional, Tuple
# Try to import X11 keyboard libraries first
try:
from pynput import keyboard
KEYBOARD_AVAILABLE = True
except ImportError:
KEYBOARD_AVAILABLE = False
@@ -23,51 +24,52 @@ 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
self.last_trigger_time = 0 # Track last trigger time to prevent double triggers
if not KEYBOARD_AVAILABLE:
logger.error("Keyboard shortcut libraries not available. Shortcuts will not work.")
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')
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()
try:
# Start keyboard listener in a separate thread
self.listener = keyboard.Listener(
on_press=self._on_press,
on_release=self._on_release
on_press=self._on_press, on_release=self._on_release
)
self.listener.daemon = True
self.listener.start()
# Verify the listener started successfully
if not self.listener.is_alive():
logger.error("Failed to start keyboard listener")
@@ -77,15 +79,15 @@ class KeyboardShortcutManager:
except Exception as e:
logger.error(f"Error starting keyboard listener: {e}")
self.active = False
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:
try:
self.listener.stop()
@@ -94,11 +96,11 @@ class KeyboardShortcutManager:
logger.error(f"Error stopping keyboard listener: {e}")
finally:
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'))
@@ -106,79 +108,91 @@ class KeyboardShortcutManager:
"""
shortcut_key = (frozenset(modifiers), key)
self.shortcuts[shortcut_key] = callback
# Create readable description of the shortcut for logging
mod_names = [self._get_key_name(mod) for mod in modifiers]
key_name = self._get_key_name(key)
shortcut_desc = "+".join(mod_names + [key_name])
logger.info(f"Registered shortcut: {shortcut_desc}")
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 _get_key_name(self, key):
"""Get a readable name for a key object."""
if hasattr(key, 'char') and key.char:
if hasattr(key, "char") and key.char:
return key.char.upper()
elif hasattr(key, 'name'):
elif hasattr(key, "name"):
return key.name
else:
return str(key)
def _on_press(self, key):
"""
Handle key press events.
Args:
key: The pressed key
"""
try:
# Add to currently pressed keys (only for modifier keys)
if key in {
keyboard.Key.alt, keyboard.Key.alt_l, keyboard.Key.alt_r,
keyboard.Key.shift, keyboard.Key.shift_l, keyboard.Key.shift_r,
keyboard.Key.ctrl, keyboard.Key.ctrl_l, keyboard.Key.ctrl_r,
keyboard.Key.cmd, keyboard.Key.cmd_l, keyboard.Key.cmd_r
keyboard.Key.alt,
keyboard.Key.alt_l,
keyboard.Key.alt_r,
keyboard.Key.shift,
keyboard.Key.shift_l,
keyboard.Key.shift_r,
keyboard.Key.ctrl,
keyboard.Key.ctrl_l,
keyboard.Key.ctrl_r,
keyboard.Key.cmd,
keyboard.Key.cmd_l,
keyboard.Key.cmd_r,
}:
# Normalize left/right variants
normalized_key = self._normalize_modifier_key(key)
self.current_keys.add(normalized_key)
# Check for shortcuts (only once per 500ms to prevent double triggers)
current_time = time.time()
if current_time - self.last_trigger_time > 0.5: # 500ms debounce
for (modifiers, trigger_key), callback in self.shortcuts.items():
# Normalize the trigger key for character keys
if isinstance(trigger_key, keyboard.KeyCode) and hasattr(trigger_key, 'char'):
if isinstance(key, keyboard.KeyCode) and hasattr(key, 'char'):
if isinstance(trigger_key, keyboard.KeyCode) and hasattr(
trigger_key, "char"
):
if isinstance(key, keyboard.KeyCode) and hasattr(key, "char"):
key_matches = key.char.lower() == trigger_key.char.lower()
else:
key_matches = False
else:
key_matches = key == trigger_key
# Check if modifiers match and key matches
normalized_modifiers = {self._normalize_modifier_key(m) for m in modifiers}
normalized_modifiers = {
self._normalize_modifier_key(m) for m in modifiers
}
if self.current_keys == normalized_modifiers and key_matches:
logger.debug(f"Shortcut triggered: {modifiers} + {trigger_key}")
self.last_trigger_time = current_time
# Run callback in a separate thread to avoid blocking
threading.Thread(target=callback, daemon=True).start()
except Exception as e:
logger.error(f"Error in keyboard shortcut handling: {e}")
def _on_release(self, key):
"""
Handle key release events.
Args:
key: The released key
"""
@@ -189,7 +203,7 @@ class KeyboardShortcutManager:
self.current_keys.discard(normalized_key)
except Exception as e:
logger.error(f"Error in keyboard release handling: {e}")
def _normalize_modifier_key(self, key):
"""Normalize left/right variants of modifier keys to their base form."""
# Map left/right variants to their base key
@@ -201,7 +215,7 @@ class KeyboardShortcutManager:
keyboard.Key.ctrl_l: keyboard.Key.ctrl,
keyboard.Key.ctrl_r: keyboard.Key.ctrl,
keyboard.Key.cmd_l: keyboard.Key.cmd,
keyboard.Key.cmd_r: keyboard.Key.cmd
keyboard.Key.cmd_r: keyboard.Key.cmd,
}
return key_mapping.get(key, key)
return key_mapping.get(key, key)
src/ui/tray_indicator.py
+59 -55
View File
@@ -5,7 +5,6 @@ This module provides a system tray indicator for controlling the speech
recognition process and displaying its status.
"""
import gi
import logging
import os
import signal
@@ -13,13 +12,18 @@ import sys
import threading
from typing import Callable, Dict, Optional
import gi
# Import GTK
gi.require_version("Gtk", "3.0")
gi.require_version("AppIndicator3", "0.1")
from gi.repository import Gtk, AppIndicator3, GObject, GLib
from gi.repository import AppIndicator3, GLib, GObject, Gtk
# Import local modules
from speech_recognition.recognition_manager import SpeechRecognitionManager, RecognitionState
from speech_recognition.recognition_manager import (
RecognitionState,
SpeechRecognitionManager,
)
from text_injection.text_injector import TextInjector
from ui.keyboard_shortcuts import KeyboardShortcutManager
@@ -36,66 +40,66 @@ PROCESSING_ICON = "microphone-process"
class TrayIndicator:
"""
System tray indicator for Vocalinux.
This class provides a system tray icon with a menu for controlling
the speech recognition process.
"""
def __init__(self, speech_engine: SpeechRecognitionManager, text_injector: TextInjector):
def __init__(
self, speech_engine: SpeechRecognitionManager, text_injector: TextInjector
):
"""
Initialize the system tray indicator.
Args:
speech_engine: The speech recognition manager instance
text_injector: The text injector instance
"""
self.speech_engine = speech_engine
self.text_injector = text_injector
# Initialize keyboard shortcut manager
self.shortcut_manager = KeyboardShortcutManager()
# Ensure icon directory exists
os.makedirs(ICON_DIR, exist_ok=True)
# Register for speech recognition state changes
self.speech_engine.register_state_callback(self._on_recognition_state_changed)
# Initialize the icon files
self._init_icons()
# Initialize the indicator (in the GTK main thread)
GLib.idle_add(self._init_indicator)
# Set up keyboard shortcuts
self.shortcut_manager.register_toggle_callback(self._toggle_recognition)
self.shortcut_manager.start()
def _init_icons(self):
"""Initialize the icon files for the tray indicator."""
# TODO: Copy default icons to the icon directory if they don't exist
pass
def _init_indicator(self):
"""Initialize the system tray indicator."""
logger.info("Initializing system tray indicator")
# Create the indicator
self.indicator = AppIndicator3.Indicator.new(
APP_ID,
DEFAULT_ICON,
AppIndicator3.IndicatorCategory.APPLICATION_STATUS
APP_ID, DEFAULT_ICON, AppIndicator3.IndicatorCategory.APPLICATION_STATUS
)
# Set the icon path
self.indicator.set_icon_theme_path(ICON_DIR)
# Set the indicator status
self.indicator.set_status(AppIndicator3.IndicatorStatus.ACTIVE)
# Create the menu
self.menu = Gtk.Menu()
# Add menu items
self._add_menu_item("Start Voice Typing", self._on_start_clicked)
self._add_menu_item("Stop Voice Typing", self._on_stop_clicked)
@@ -104,29 +108,29 @@ class TrayIndicator:
self._add_menu_separator()
self._add_menu_item("About", self._on_about_clicked)
self._add_menu_item("Quit", self._on_quit_clicked)
# Set the indicator menu
self.indicator.set_menu(self.menu)
# Show the menu
self.menu.show_all()
# Update the UI based on the initial state
self._update_ui(RecognitionState.IDLE)
return False # Remove idle callback
def _toggle_recognition(self):
"""Toggle the recognition state between IDLE and LISTENING."""
if self.speech_engine.state == RecognitionState.IDLE:
self.speech_engine.start_recognition()
else:
self.speech_engine.stop_recognition()
def _add_menu_item(self, label: str, callback: Callable):
"""
Add a menu item to the indicator menu.
Args:
label: The label for the menu item
callback: The callback function to call when the item is clicked
@@ -135,26 +139,26 @@ class TrayIndicator:
item.connect("activate", callback)
self.menu.append(item)
return item
def _add_menu_separator(self):
"""Add a separator to the indicator menu."""
separator = Gtk.SeparatorMenuItem()
self.menu.append(separator)
def _on_recognition_state_changed(self, state: RecognitionState):
"""
Handle changes in the speech recognition state.
Args:
state: The new recognition state
"""
# Update the UI in the GTK main thread
GLib.idle_add(self._update_ui, state)
def _update_ui(self, state: RecognitionState):
"""
Update the UI based on the recognition state.
Args:
state: The current recognition state
"""
@@ -174,13 +178,13 @@ class TrayIndicator:
self.indicator.set_icon(DEFAULT_ICON)
self._set_menu_item_enabled("Start Voice Typing", True)
self._set_menu_item_enabled("Stop Voice Typing", False)
return False # Remove idle callback
def _set_menu_item_enabled(self, label: str, enabled: bool):
"""
Set the enabled state of a menu item by its label.
Args:
label: The label of the menu item
enabled: Whether the item should be enabled
@@ -189,74 +193,74 @@ class TrayIndicator:
if isinstance(item, Gtk.MenuItem) and item.get_label() == label:
item.set_sensitive(enabled)
break
def _on_start_clicked(self, widget):
"""Handle click on the Start Voice Typing menu item."""
logger.debug("Start Voice Typing clicked")
self.speech_engine.start_recognition()
def _on_stop_clicked(self, widget):
"""Handle click on the Stop Voice Typing menu item."""
logger.debug("Stop Voice Typing clicked")
self.speech_engine.stop_recognition()
def _on_settings_clicked(self, widget):
"""Handle click on the Settings menu item."""
logger.debug("Settings clicked")
# TODO: Implement settings dialog
def _on_about_clicked(self, widget):
"""Handle click on the About menu item."""
logger.debug("About clicked")
about_dialog = Gtk.AboutDialog()
about_dialog.set_program_name("Vocalinux")
about_dialog.set_version("0.1.0")
about_dialog.set_copyright("© 2025 Vocalinux Team")
about_dialog.set_comments("A seamless voice dictation system for Ubuntu")
about_dialog.set_website("https://github.com/vocalinux")
about_dialog.set_website("https://github.com/jatinkrmalik/vocalinux")
about_dialog.set_website_label("GitHub Repository")
about_dialog.set_license_type(Gtk.License.GPL_3_0)
# Run the dialog
about_dialog.run()
about_dialog.destroy()
def _on_quit_clicked(self, widget):
"""Handle click on the Quit menu item."""
logger.debug("Quit clicked")
self._quit()
def _quit(self):
"""Quit the application."""
logger.info("Quitting application")
# Stop the keyboard shortcut manager
self.shortcut_manager.stop()
Gtk.main_quit()
def run(self):
"""Run the application main loop."""
logger.info("Starting GTK main loop")
# Set up signal handlers for graceful termination
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
# Start the GTK main loop
try:
Gtk.main()
except KeyboardInterrupt:
self._quit()
def _signal_handler(self, sig, frame):
"""
Handle signals (e.g., SIGINT, SIGTERM).
Args:
sig: The signal number
frame: The current stack frame
"""
logger.info(f"Received signal {sig}, shutting down...")
GLib.idle_add(self._quit)
GLib.idle_add(self._quit)