diff --git a/Requirements.txt b/Requirements.txt index 8b077ee..c70a74c 100644 --- a/Requirements.txt +++ b/Requirements.txt @@ -2,3 +2,4 @@ accessible_output2 PySide6>=6.0.0 python-vlc setproctitle>=1.2.0 +dasbus; sys_platform == 'linux' diff --git a/Toby Doom Launcher.py b/Toby Doom Launcher.py index 1b5db0a..9dcbeb9 100755 --- a/Toby Doom Launcher.py +++ b/Toby Doom Launcher.py @@ -50,58 +50,133 @@ from PySide6.QtCore import Qt, QTimer import webbrowser class OrcaRemoteController: - """D-Bus interface for Orca screen reader remote control""" - + """D-Bus interface for Orca screen reader remote control using dasbus library""" + _instance = None + _availability_checked = False + _is_available = False + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + def __init__(self): + if self._availability_checked: + return self.service_name = "org.gnome.Orca.Service" self.main_path = "/org/gnome/Orca/Service" + self.proxy = None + self.speech_proxy = None self.available = self._test_availability() - + OrcaRemoteController._is_available = self.available + OrcaRemoteController._availability_checked = True + + def _call_with_timeout(self, func, timeout_seconds=2): + """Execute a function with timeout using threading""" + result = [None] + exception = [None] + + def wrapper(): + try: + result[0] = func() + except Exception as e: + exception[0] = e + + thread = threading.Thread(target=wrapper, daemon=True) + thread.start() + thread.join(timeout=timeout_seconds) + + if thread.is_alive(): + # Timeout occurred + return None + if exception[0]: + raise exception[0] + return result[0] + + def _get_orca_dbus_address(self): + """Try to find Orca's D-Bus session address from its process""" + try: + # Find Orca process + result = subprocess.run( + ["pgrep", "-x", "orca"], + capture_output=True, + text=True, + timeout=1 + ) + if result.returncode == 0: + pid = result.stdout.strip().split('\n')[0] + # Read Orca's environment + with open(f"/proc/{pid}/environ", 'r') as f: + environ = f.read() + for var in environ.split('\0'): + if var.startswith('DBUS_SESSION_BUS_ADDRESS='): + return var.split('=', 1)[1] + except Exception: + pass + return None + def _test_availability(self): """Test if Orca remote controller is available""" + def test_connection(): + from dasbus.connection import SessionMessageBus + + # First try with current session bus + try: + bus = SessionMessageBus() + self.proxy = bus.get_proxy(self.service_name, self.main_path) + self.proxy.ListCommands() + self.speech_proxy = bus.get_proxy( + self.service_name, + f"{self.main_path}/SpeechAndVerbosityManager" + ) + return True + except Exception: + # If that fails, try to find Orca's D-Bus session + orca_bus_address = self._get_orca_dbus_address() + if orca_bus_address: + # Temporarily set the environment variable + old_address = os.environ.get('DBUS_SESSION_BUS_ADDRESS') + try: + os.environ['DBUS_SESSION_BUS_ADDRESS'] = orca_bus_address + bus = SessionMessageBus() + self.proxy = bus.get_proxy(self.service_name, self.main_path) + self.proxy.ListCommands() + self.speech_proxy = bus.get_proxy( + self.service_name, + f"{self.main_path}/SpeechAndVerbosityManager" + ) + return True + finally: + # Restore original address + if old_address: + os.environ['DBUS_SESSION_BUS_ADDRESS'] = old_address + elif 'DBUS_SESSION_BUS_ADDRESS' in os.environ: + del os.environ['DBUS_SESSION_BUS_ADDRESS'] + raise + try: - result = subprocess.run([ - "gdbus", "call", "--session", - "--dest", self.service_name, - "--object-path", self.main_path, - "--method", "org.gnome.Orca.Service.ListCommands" - ], check=True, capture_output=True, text=True, timeout=2) - return True - except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError): + return self._call_with_timeout(test_connection, timeout_seconds=2) or False + except Exception: return False - + def present_message(self, message): """Present a message via Orca speech/braille output""" - if not self.available: + if not self.available or not self.proxy: return False - + try: - result = subprocess.run([ - "gdbus", "call", "--session", - "--dest", self.service_name, - "--object-path", self.main_path, - "--method", "org.gnome.Orca.Service.PresentMessage", - message - ], check=True, capture_output=True, text=True, timeout=2) - return "true" in result.stdout.lower() - except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError): + return self._call_with_timeout(lambda: self.proxy.PresentMessage(message), timeout_seconds=2) or False + except Exception: return False - + def interrupt_speech(self): """Interrupt current speech via SpeechAndVerbosityManager""" - if not self.available: + if not self.available or not self.speech_proxy: return False - + try: - result = subprocess.run([ - "gdbus", "call", "--session", - "--dest", self.service_name, - "--object-path", f"{self.main_path}/SpeechAndVerbosityManager", - "--method", "org.gnome.Orca.Module.ExecuteCommand", - "InterruptSpeech", "false" - ], check=True, capture_output=True, text=True, timeout=2) - return "true" in result.stdout.lower() - except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError): + return self._call_with_timeout(lambda: self.speech_proxy.ExecuteCommand("InterruptSpeech", False), timeout_seconds=2) or False + except Exception: return False