Add native Wine accessibility bridge

This commit is contained in:
Storm Dragon
2026-07-17 18:17:23 -04:00
parent 0576b6f79f
commit c9f56cb7ed
24 changed files with 1188 additions and 8 deletions
+76
View File
@@ -678,6 +678,82 @@ class CthulhuDBusServiceInterface(Publishable):
script.presentMessage(message)
return True
def SpeakText(self, text: str, interrupt: bool) -> bool: # pylint: disable=invalid-name
"""Speaks plain text without also presenting it in braille."""
if not text:
return False
from . import speech # pylint: disable=import-outside-toplevel
speech.speak(text, interrupt=interrupt)
return True
def BrailleMessage(self, text: str) -> bool: # pylint: disable=invalid-name
"""Displays a message on the active braille display."""
if not text:
return False
from . import braille # pylint: disable=import-outside-toplevel
braille.displayMessage(text, flashTime=-1)
return True
def CancelSpeech(self) -> bool: # pylint: disable=invalid-name
"""Stops current speech and sound output."""
from . import speech # pylint: disable=import-outside-toplevel
speech.stop()
return True
def SpeakSsml(self, requestId: str, ssml: str) -> int: # pylint: disable=invalid-name
"""Rejects SSML when its timing and mark semantics cannot be preserved."""
del requestId
try:
root = ET.fromstring(ssml)
except ET.ParseError:
debug.print_message(debug.LEVEL_WARNING, "DBUS SERVICE: Invalid SSML rejected", True)
return 87 # ERROR_INVALID_PARAMETER
if root.tag.rsplit("}", 1)[-1].lower() != "speak":
debug.print_message(debug.LEVEL_WARNING, "DBUS SERVICE: Non-speak SSML rejected", True)
return 87
debug.print_message(
debug.LEVEL_INFO,
"DBUS SERVICE: SSML rejected because mark and completion semantics are unavailable",
True,
)
return 50 # ERROR_NOT_SUPPORTED
def PresentAccessibleEvent( # pylint: disable=invalid-name,too-many-arguments
self,
sourceId: str,
eventType: str,
name: str,
role: str,
value: str,
states: str,
description: str,
position: int,
count: int,
windowTitle: str,
) -> bool:
"""Presents a normalized accessibility event received from Wine."""
del sourceId, eventType, states
parts = [part for part in (name, role, value, description) if part]
if position > 0 and count > 0:
parts.append(f"{position} of {count}")
if windowTitle and windowTitle not in parts:
parts.append(windowTitle)
if not parts:
return False
return self.PresentMessage(", ".join(parts))
def GetVersion(self) -> str: # pylint: disable=invalid-name
"""Returns Cthulhu's version and revision if available."""
+1
View File
@@ -116,6 +116,7 @@ cthulhu_python_sources = files([
'typing_echo_presenter.py',
'wnck_support.py',
'where_am_i_presenter.py',
'wine_access_manager.py',
])
# Note: Main executable (cthulhu) is installed from src/cthulhu.py
@@ -0,0 +1,5 @@
"""Wine accessibility integration plugin."""
from .plugin import WineAccessibility
__all__ = ["WineAccessibility"]
@@ -0,0 +1,9 @@
python3.install_sources(
files('__init__.py', 'plugin.py'),
subdir: 'cthulhu/plugins/WineAccessibility'
)
install_data(
'plugin.info',
install_dir: python3.get_install_dir() / 'cthulhu' / 'plugins' / 'WineAccessibility'
)
@@ -0,0 +1,8 @@
name = Wine Accessibility
version = 1.0.0
description = Provides screen reader access to standard Wine and Proton applications
authors = Stormux
website = https://git.stormux.org/storm/cthulhu
copyright = Copyright 2026
builtin = false
hidden = false
@@ -0,0 +1,104 @@
"""Automatically manages Cthulhu's Wine accessibility helper."""
from gi.repository import GLib, Gtk
from cthulhu import settings
from cthulhu import settings_manager
from cthulhu.plugin import Plugin, cthulhu_hookimpl
from cthulhu.wine_access_manager import WineAccessManager
class WineAccessibility(Plugin):
"""Starts one helper for each active same-user Wine or Proton prefix."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.settingsManager = settings_manager.getManager()
self.manager = WineAccessManager()
self.pollSourceId = 0
self.prefsGrid = None
self.prefsWidgets = {}
@cthulhu_hookimpl
def activate(self, plugin=None):
if plugin is not None and plugin is not self:
return
self._refresh_manager_settings()
if self._setting("wineAccessibilityEnabled") and self._setting(
"wineAccessibilityAutoManage"
):
self.manager.poll()
self.pollSourceId = GLib.timeout_add(self.manager.POLL_INTERVAL_MS, self.manager.poll)
@cthulhu_hookimpl
def deactivate(self, plugin=None):
if plugin is not None and plugin is not self:
return
if self.pollSourceId:
GLib.source_remove(self.pollSourceId)
self.pollSourceId = 0
self.manager.shutdown()
def _setting(self, name):
value = self.settingsManager.getSetting(name)
return getattr(settings, name) if value is None else value
def _refresh_manager_settings(self):
self.manager.dialogReaderEnabled = bool(self._setting("wineAccessibilityDialogReader"))
self.manager.controllerEnabled = bool(self._setting("wineAccessibilityController"))
def getPreferencesGUI(self):
if self.prefsGrid is None:
self.prefsGrid = Gtk.Grid(
row_spacing=6,
column_spacing=12,
margin_left=12,
margin_right=12,
margin_top=12,
margin_bottom=12,
)
labels = (
("enabled", "Enable Wine accessibility"),
("auto", "Automatically manage Wine and Proton prefixes"),
("dialogs", "Read standard Wine controls"),
("controller", "Enable NVDA Controller and Tolk compatibility"),
)
for row, (name, label) in enumerate(labels):
widget = Gtk.CheckButton(label=label)
if name == "enabled":
widget.connect("toggled", self._update_preferences_sensitivity)
self.prefsGrid.attach(widget, 0, row, 1, 1)
self.prefsWidgets[name] = widget
values = {
"enabled": "wineAccessibilityEnabled",
"auto": "wineAccessibilityAutoManage",
"dialogs": "wineAccessibilityDialogReader",
"controller": "wineAccessibilityController",
}
for name, settingName in values.items():
self.prefsWidgets[name].set_active(bool(self._setting(settingName)))
self._update_preferences_sensitivity()
return self.prefsGrid, "Wine Accessibility"
def getPreferencesFromGUI(self):
if not self.prefsWidgets:
return {}
return {
"wineAccessibilityEnabled": self.prefsWidgets["enabled"].get_active(),
"wineAccessibilityAutoManage": self.prefsWidgets["auto"].get_active(),
"wineAccessibilityDialogReader": self.prefsWidgets["dialogs"].get_active(),
"wineAccessibilityController": self.prefsWidgets["controller"].get_active(),
}
def _update_preferences_sensitivity(self, _widget=None):
if not self.prefsWidgets:
return
enabled = self.prefsWidgets["enabled"].get_active()
for name in ("auto", "dialogs", "controller"):
self.prefsWidgets[name].set_sensitive(enabled)
def refresh_settings(self):
"""Applies settings saved from the plugin preferences page."""
self.deactivate()
self.activate()
+1
View File
@@ -14,3 +14,4 @@ subdir('hello_world')
subdir('self_voice')
subdir('SSIPProxy')
subdir('WindowTitleReader')
subdir('WineAccessibility')
+10 -1
View File
@@ -65,6 +65,10 @@ userCustomizableSettings = [
"useCustomEchoForSentence",
"gameMode",
"nvda2cthulhuTranslateEnabled",
"wineAccessibilityEnabled",
"wineAccessibilityAutoManage",
"wineAccessibilityDialogReader",
"wineAccessibilityController",
"enableAlphabeticKeys",
"enableNumericKeys",
"enablePunctuationKeys",
@@ -304,6 +308,10 @@ messagesAreDetailed = True
enablePauseBreaks = True
gameMode = False
nvda2cthulhuTranslateEnabled = False
wineAccessibilityEnabled = True
wineAccessibilityAutoManage = True
wineAccessibilityDialogReader = True
wineAccessibilityController = True
speakDescription = True
speakContextBlockquote = True
speakContextPanel = True
@@ -506,7 +514,8 @@ activePlugins = [
'OCR',
'SpeechHistory',
'SSIPProxy',
'WindowTitleReader'
'WindowTitleReader',
'WineAccessibility'
]
pluginSources = []
+223
View File
@@ -0,0 +1,223 @@
"""Manages one Cthulhu Wine accessibility helper per active prefix."""
from __future__ import annotations
import os
from pathlib import Path
import shutil
import subprocess
import time
from typing import Optional
from . import debug
from . import cthulhu_platform
class PrefixProcess:
"""Tracks a helper process and restart state for one Wine prefix."""
def __init__(self) -> None:
self.process: Optional[subprocess.Popen] = None
self.lastSeen = 0.0
self.startedAt = 0.0
self.failures = 0
self.nextStart = 0.0
class WineAccessManager:
"""Discovers Wine processes and supervises prefix-scoped helpers."""
POLL_INTERVAL_MS = 2000
IDLE_GRACE_SECONDS = 10.0
MAX_BACKOFF_SECONDS = 60.0
MANAGED_ENVIRONMENT_KEY = "CTHULHU_WINE_ACCESS_MANAGED"
def __init__(self, helperPath: Optional[str] = None) -> None:
self.helperPath = helperPath or self._find_helper()
self.prefixes: dict[str, PrefixProcess] = {}
self.dialogReaderEnabled = True
self.controllerEnabled = True
@staticmethod
def _find_helper() -> Optional[str]:
configured = os.environ.get("CTHULHU_WINE_ACCESS_HELPER")
if configured and os.access(configured, os.X_OK):
return configured
installed = os.path.join(
cthulhu_platform.prefix,
"libexec",
"cthulhu",
"wine",
"cthulhu-wine-access.exe",
)
if os.access(installed, os.X_OK):
return installed
return shutil.which("cthulhu-wine-access.exe")
@staticmethod
def _read_environ(pid: int) -> dict[str, str]:
try:
raw = Path(f"/proc/{pid}/environ").read_bytes()
except (OSError, PermissionError):
return {}
result = {}
for entry in raw.split(b"\0"):
if b"=" not in entry:
continue
key, value = entry.split(b"=", 1)
result[key.decode(errors="replace")] = value.decode(errors="replace")
return result
@staticmethod
def _prefix_from_environ(environment: dict[str, str]) -> Optional[str]:
prefix = environment.get("WINEPREFIX")
if not prefix:
compatPath = environment.get("STEAM_COMPAT_DATA_PATH")
if compatPath:
prefix = os.path.join(compatPath, "pfx")
if not prefix:
return None
return os.path.realpath(os.path.expanduser(prefix))
@staticmethod
def _loader_for_process(pid: int, environment: dict[str, str]) -> Optional[str]:
configured = environment.get("WINELOADER")
if configured and os.access(configured, os.X_OK):
return configured
try:
executable = os.readlink(f"/proc/{pid}/exe")
except OSError:
return None
name = os.path.basename(executable).lower()
if name in {"wine", "wine64"}:
return executable
if name in {"wine-preloader", "wine64-preloader"}:
loaderName = name.removesuffix("-preloader")
loader = os.path.join(os.path.dirname(executable), loaderName)
if os.access(loader, os.X_OK):
return loader
return None
def discover_prefixes(self) -> dict[str, tuple[dict[str, str], Optional[str]]]:
"""Returns active prefixes with the environment and loader that created them."""
discovered = {}
ownUid = os.getuid()
helperPids = {
state.process.pid
for state in self.prefixes.values()
if state.process is not None and state.process.poll() is None
}
for entry in Path("/proc").iterdir():
if not entry.name.isdigit():
continue
try:
if entry.stat().st_uid != ownUid:
continue
except OSError:
continue
pid = int(entry.name)
if pid in helperPids:
continue
environment = self._read_environ(pid)
if environment.get(self.MANAGED_ENVIRONMENT_KEY) == "1":
continue
prefix = self._prefix_from_environ(environment)
if prefix is None:
continue
loader = self._loader_for_process(pid, environment)
if loader is None:
continue
previous = discovered.get(prefix)
if previous is None:
discovered[prefix] = (environment, loader)
return discovered
def poll(self) -> bool:
"""Reconciles helper processes with currently active Wine prefixes."""
if not self.helperPath:
return True
now = time.monotonic()
discovered = self.discover_prefixes()
for prefix, (environment, loader) in discovered.items():
state = self.prefixes.setdefault(prefix, PrefixProcess())
state.lastSeen = now
self._ensure_running(prefix, state, environment, loader, now)
for prefix, state in list(self.prefixes.items()):
self._record_exit(state, now)
if prefix not in discovered and now - state.lastSeen >= self.IDLE_GRACE_SECONDS:
self._stop(state)
del self.prefixes[prefix]
return True
def _ensure_running(
self,
prefix: str,
state: PrefixProcess,
sourceEnvironment: dict[str, str],
loader: Optional[str],
now: float,
) -> None:
self._record_exit(state, now)
if state.process is not None or now < state.nextStart:
return
environment = os.environ.copy()
for key in ("DISPLAY", "WAYLAND_DISPLAY", "XDG_RUNTIME_DIR", "DBUS_SESSION_BUS_ADDRESS"):
if sourceEnvironment.get(key):
environment[key] = sourceEnvironment[key]
environment["WINEPREFIX"] = prefix
environment[self.MANAGED_ENVIRONMENT_KEY] = "1"
if loader:
environment["WINELOADER"] = loader
command = [self.helperPath]
if not self.dialogReaderEnabled:
command.append("--no-dialog-reader")
if not self.controllerEnabled:
command.append("--no-controller")
try:
state.process = subprocess.Popen(
command,
env=environment,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
state.startedAt = now
debug.print_message(
debug.LEVEL_INFO, f"WINE ACCESS: Started helper for {prefix}", True
)
except OSError as error:
state.failures += 1
state.nextStart = now + min(2 ** state.failures, self.MAX_BACKOFF_SECONDS)
debug.print_message(
debug.LEVEL_WARNING, f"WINE ACCESS: Failed to start helper: {error}", True
)
def _record_exit(self, state: PrefixProcess, now: float) -> None:
if state.process is None or state.process.poll() is None:
return
runtime = now - state.startedAt
state.process = None
state.failures = state.failures + 1 if runtime < 10.0 else 0
state.nextStart = now + min(2 ** state.failures, self.MAX_BACKOFF_SECONDS)
@staticmethod
def _stop(state: PrefixProcess) -> None:
if state.process is None or state.process.poll() is not None:
state.process = None
return
state.process.terminate()
try:
state.process.wait(timeout=2.0)
except subprocess.TimeoutExpired:
state.process.kill()
state.process = None
def shutdown(self) -> None:
"""Stops all managed helper processes."""
for state in self.prefixes.values():
self._stop(state)
self.prefixes.clear()