Cthulhu support updated to use d-bus.
This commit is contained in:
Storm Dragon
+155
-31
@@ -180,6 +180,137 @@ class OrcaRemoteController:
|
||||
return False
|
||||
|
||||
|
||||
class CthulhuRemoteController:
|
||||
"""D-Bus interface for Cthulhu 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.stormux.Cthulhu.Service"
|
||||
self.main_path = "/org/stormux/Cthulhu/Service"
|
||||
self.proxy = None
|
||||
self.speech_proxy = None
|
||||
self.available = self._test_availability()
|
||||
CthulhuRemoteController._is_available = self.available
|
||||
CthulhuRemoteController._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_cthulhu_dbus_address(self):
|
||||
"""Try to find Cthulhu's D-Bus session address from its process"""
|
||||
try:
|
||||
# Find Cthulhu process
|
||||
result = subprocess.run(
|
||||
["pgrep", "-x", "cthulhu"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=1
|
||||
)
|
||||
if result.returncode == 0:
|
||||
pid = result.stdout.strip().split('\n')[0]
|
||||
# Read Cthulhu'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 Cthulhu 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 Cthulhu's D-Bus session
|
||||
cthulhu_bus_address = self._get_cthulhu_dbus_address()
|
||||
if cthulhu_bus_address:
|
||||
# Temporarily set the environment variable
|
||||
old_address = os.environ.get('DBUS_SESSION_BUS_ADDRESS')
|
||||
try:
|
||||
os.environ['DBUS_SESSION_BUS_ADDRESS'] = cthulhu_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:
|
||||
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 Cthulhu speech/braille output"""
|
||||
if not self.available or not self.proxy:
|
||||
return False
|
||||
|
||||
try:
|
||||
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 or not self.speech_proxy:
|
||||
return False
|
||||
|
||||
try:
|
||||
return self._call_with_timeout(lambda: self.speech_proxy.ExecuteCommand("InterruptSpeech", False), timeout_seconds=2) or False
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# Initialize speech provider based on platform
|
||||
if platform.system() == "Windows":
|
||||
# Set up DLL paths for Windows
|
||||
@@ -213,11 +344,12 @@ else:
|
||||
speechProvider = "orca_remote"
|
||||
orca = orca_remote
|
||||
else:
|
||||
# Fall back to Cthulhu
|
||||
try:
|
||||
output = subprocess.check_output(["pgrep", "cthulhu"])
|
||||
speechProvider = "cthulhu"
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
# Fall back to Cthulhu remote controller
|
||||
cthulhu_remote = CthulhuRemoteController()
|
||||
if cthulhu_remote.available:
|
||||
speechProvider = "cthulhu_remote"
|
||||
cthulhu = cthulhu_remote
|
||||
else:
|
||||
try:
|
||||
import accessible_output2.outputs.auto
|
||||
s = accessible_output2.outputs.auto.Auto()
|
||||
@@ -348,23 +480,16 @@ class SpeechHandler:
|
||||
orca.interrupt_speech()
|
||||
if not orca.present_message(text):
|
||||
print(f"Orca remote error - message not delivered", file=sys.stderr)
|
||||
elif speechProvider == "cthulhu_remote":
|
||||
# Try Cthulhu, interrupt current speech then present message
|
||||
cthulhu.interrupt_speech()
|
||||
if not cthulhu.present_message(text):
|
||||
print(f"Cthulhu remote error - message not delivered", file=sys.stderr)
|
||||
elif speechProvider == "speechd":
|
||||
spd.cancel()
|
||||
spd.speak(text)
|
||||
elif speechProvider == "accessible_output2":
|
||||
s.speak(text, interrupt=True)
|
||||
else: # Cthulhu
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
["socat", "-", "UNIX-CLIENT:/tmp/cthulhu.sock"],
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True
|
||||
)
|
||||
process.communicate(input=text)
|
||||
except Exception as e:
|
||||
print(f"Cthulhu error: {e}", file=sys.stderr)
|
||||
|
||||
def process_line(self, line: str) -> Optional[str]:
|
||||
"""Process a line of game output for speech"""
|
||||
@@ -2152,12 +2277,10 @@ class DoomLauncher(QMainWindow):
|
||||
if orca_remote.available:
|
||||
available_providers.append("orca_remote")
|
||||
|
||||
# Check Cthulhu
|
||||
try:
|
||||
subprocess.check_output(["pgrep", "cthulhu"])
|
||||
available_providers.append("cthulhu")
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
pass
|
||||
# Check Cthulhu remote controller
|
||||
cthulhu_remote = CthulhuRemoteController()
|
||||
if cthulhu_remote.available:
|
||||
available_providers.append("cthulhu_remote")
|
||||
|
||||
# Check accessible_output2
|
||||
try:
|
||||
@@ -2176,7 +2299,7 @@ class DoomLauncher(QMainWindow):
|
||||
provider_names = {
|
||||
"accessible_output2": "Accessible Output2",
|
||||
"orca_remote": "Orca Remote",
|
||||
"cthulhu": "Cthulhu",
|
||||
"cthulhu_remote": "Cthulhu Remote",
|
||||
"speechd": "Speech Dispatcher"
|
||||
}
|
||||
|
||||
@@ -2211,7 +2334,7 @@ class DoomLauncher(QMainWindow):
|
||||
|
||||
def set_speech_provider(self, provider: str):
|
||||
"""Update the global speech provider"""
|
||||
global speechProvider, s, spd, orca
|
||||
global speechProvider, s, spd, orca, cthulhu
|
||||
|
||||
try:
|
||||
if provider == "accessible_output2":
|
||||
@@ -2226,12 +2349,13 @@ class DoomLauncher(QMainWindow):
|
||||
else:
|
||||
QMessageBox.warning(self, "Error", "Orca Remote is not available")
|
||||
return
|
||||
elif provider == "cthulhu":
|
||||
try:
|
||||
subprocess.check_output(["pgrep", "cthulhu"])
|
||||
speechProvider = "cthulhu"
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
QMessageBox.warning(self, "Error", "Cthulhu is not running")
|
||||
elif provider == "cthulhu_remote":
|
||||
cthulhu_remote = CthulhuRemoteController()
|
||||
if cthulhu_remote.available:
|
||||
cthulhu = cthulhu_remote
|
||||
speechProvider = "cthulhu_remote"
|
||||
else:
|
||||
QMessageBox.warning(self, "Error", "Cthulhu Remote is not available")
|
||||
return
|
||||
elif provider == "speechd":
|
||||
import speechd
|
||||
|
||||
Reference in New Issue
Block a user