10 Commits

Author SHA1 Message Date
Storm Dragon e2f9a7c2e2 A few touch ups before release tag. 2026-05-14 18:42:06 -04:00
Storm Dragon 922ba60445 Make new cthulhu-git arch package for testing latest code. 2026-05-09 22:08:13 -04:00
Storm Dragon a5f7c9a8f3 Work on sound positioning for objects. Enable in sounds in preferences. 2026-05-08 07:41:44 -04:00
Storm Dragon 6f33caade1 Sound should now stop when speech stops. This means no more interrupting speech just to have the link sounds on a page continue to play. 2026-05-08 01:02:57 -04:00
Storm Dragon e54600ff4d Reworked the xterm hand off so it's faster and more reliable. 2026-05-07 18:05:16 -04:00
Storm Dragon e6b6b1051e A few minor fixes for windows that aren't reported by at-spi. 2026-05-07 12:31:25 -04:00
Storm Dragon e377f39fe3 version update. 2026-05-06 19:27:11 -04:00
Storm Dragon 42006f4725 Merge branch 'testing' 2026-05-06 19:23:44 -04:00
Storm Dragon 3ae49f48ac A bit of pre-release polish. 2026-05-06 19:23:07 -04:00
Storm Dragon 4c0c0013ca Version bump. 2026-05-02 18:53:06 -04:00
38 changed files with 1409 additions and 13670 deletions
+1
View File
@@ -75,6 +75,7 @@ debug.log
# Package artifacts
*.pkg.tar.zst
distro-packages/*/cthulhu/
distro-packages/*/cthulhu-git/
distro-packages/*/pkg/
# Generated makefiles (should not be committed)
-13049
View File
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,126 @@
# Maintainer: Storm Dragon <storm_dragon@stormux.org>
pkgname=cthulhu-git
_pkgname=cthulhu
pkgver=2026.05.06.r394.ga5f7c9a
pkgrel=1
pkgdesc="Desktop-agnostic screen reader with plugin system, forked from Orca"
url="https://git.stormux.org/storm/cthulhu"
arch=(any)
license=(LGPL)
provides=("${_pkgname}")
conflicts=("${_pkgname}")
depends=(
# Core AT-SPI accessibility
at-spi2-core
gobject-introspection-runtime
python-gobject
python-cairo
gtk3
python-pywayland
# Audio and speech
speech-dispatcher
gstreamer
gst-plugins-base # playbin, audiotestsrc, basic decoders
gst-plugins-good # pulsesink, more decoders
# Braille support
brltty
liblouis
# Plugin system and D-Bus remote control
python-pluggy
python-tomlkit
python-dasbus
# AI Assistant dependencies (for screenshots, HTTP requests, and actions)
python-requests
python-pyautogui
# Desktop integration
hicolor-icon-theme
libwnck3
pango
# System utilities
python
python-setproctitle
socat # for self-voicing feature
xorg-xkbcomp
xorg-xmodmap
)
optdepends=(
'espeak-ng: Alternative TTS engine'
'festival: Alternative TTS engine'
'flite: Lightweight TTS engine'
'espeak: Legacy TTS engine'
# AI Assistant providers (optional)
'claude-code: Claude AI provider support'
'openai-codex: ChatGPT AI provider support'
'gemini-cli: Gemini AI provider support'
'ollama: Local AI model support'
# OCR plugin dependencies (optional)
'python-pillow: Image processing for OCR and AI Assistant'
'python-pytesseract: Python wrapper for Tesseract OCR engine'
'python-pdf2image: PDF to image conversion for OCR'
'python-scipy: Scientific computing for OCR color analysis'
'python-webcolors: Color name lookup for OCR text decoration'
'tesseract: OCR engine for text recognition'
'tesseract-data-eng: English language data for Tesseract'
# nvda2cthulhu plugin (optional)
'python-msgpack: Msgpack decoding for nvda2cthulhu'
'python-tornado: WebSocket server for nvda2cthulhu'
# Window Title Reader plugin (optional)
'python-xlib: X11 access for Wine window title plugin'
)
makedepends=(
git
meson
ninja
python-build
python-installer
python-wheel
)
install=cthulhu.install
source=(
"${_pkgname}::git+https://git.stormux.org/storm/${_pkgname}.git#branch=master"
"cthulhu.install"
)
b2sums=(
'SKIP'
'SKIP'
)
pkgver() {
cd "${_pkgname}"
local projectVersion revisionCount commitHash
projectVersion=$(sed -n "s/^[[:space:]]*version: '\([^']*\)'.*/\1/p" meson.build)
projectVersion=${projectVersion%-master}
projectVersion=${projectVersion//-/.}
[[ -n "${projectVersion}" ]] || return 1
revisionCount=$(git rev-list --count HEAD)
commitHash=$(git rev-parse --short=7 HEAD)
printf "%s.r%s.g%s\n" "${projectVersion}" "${revisionCount}" "${commitHash}"
}
build() {
cd "${_pkgname}"
arch-meson _build
meson compile -C _build
}
package() {
cd "${_pkgname}"
meson install -C _build --destdir "$pkgdir"
# Remove icon cache - it will be generated by post-install hooks
rm -f "$pkgdir/usr/share/icons/hicolor/icon-theme.cache"
}
# vim:set sw=2 sts=-1 et:
@@ -1,7 +1,7 @@
# Maintainer: Storm Dragon <storm_dragon@stormux.org>
pkgname=cthulhu
pkgver=2026.03.02
pkgver=2026.05.14
pkgrel=1
pkgdesc="Desktop-agnostic screen reader with plugin system, forked from Orca"
url="https://git.stormux.org/storm/cthulhu"
@@ -0,0 +1,11 @@
post_install() {
gtk-update-icon-cache -q -t -f usr/share/icons/hicolor
}
post_upgrade() {
post_install
}
post_remove() {
gtk-update-icon-cache -q -t -f usr/share/icons/hicolor
}
+1 -1
View File
@@ -82,7 +82,7 @@ find $PKG -print0 | xargs -0 file | grep -e "executable" -e "shared object" | gr
| cut -f 1 -d : | xargs strip --strip-unneeded 2> /dev/null || true
mkdir -p $PKG/usr/doc/$PRGNAM-$VERSION
cp -a AUTHORS COPYING ChangeLog NEWS README.md \
cp -a AUTHORS COPYING ChangeLog README.md \
$PKG/usr/doc/$PRGNAM-$VERSION
cat $CWD/$PRGNAM.SlackBuild > $PKG/usr/doc/$PRGNAM-$VERSION/$PRGNAM.SlackBuild
-1
View File
@@ -17,7 +17,6 @@ This package requires the following packages, all available from SlackBuilds.org
- at-spi2-core
- brltty
- gobject-introspection
- gsettings-desktop-schemas
- gstreamer
- gst-plugins-base
- gst-plugins-good
+1 -1
View File
@@ -5,6 +5,6 @@ DOWNLOAD="https://git.stormux.org/storm/cthulhu.git"
MD5SUM="SKIP"
DOWNLOAD_x86_64=""
MD5SUM_x86_64=""
REQUIRES="at-spi2-core brltty gobject-introspection gsettings-desktop-schemas gstreamer gst-plugins-base gst-plugins-good gtk3 liblouis libwnck3 python3-atspi python3-cairo python3-gobject python3-setproctitle speech-dispatcher"
REQUIRES="at-spi2-core brltty gobject-introspection gstreamer gst-plugins-base gst-plugins-good gtk3 liblouis libwnck3 python3-atspi python3-cairo python3-gobject python3-setproctitle speech-dispatcher"
MAINTAINER="Storm Dragon"
EMAIL="storm_dragon@stormux.org"
+1 -1
View File
@@ -82,7 +82,7 @@ find $PKG -print0 | xargs -0 file | grep -e "executable" -e "shared object" | gr
| cut -f 1 -d : | xargs strip --strip-unneeded 2> /dev/null || true
mkdir -p $PKG/usr/doc/$PRGNAM-$VERSION
cp -a AUTHORS COPYING ChangeLog NEWS README.md \
cp -a AUTHORS COPYING ChangeLog README.md \
$PKG/usr/doc/$PRGNAM-$VERSION
cat $CWD/$PRGNAM.SlackBuild > $PKG/usr/doc/$PRGNAM-$VERSION/$PRGNAM.SlackBuild
+9 -11
View File
@@ -322,17 +322,15 @@ originated as a community effort led by the Sun Microsystems Inc.
Accessibility Program Office and with contributions from many community members.
.SH SEE ALSO
For more information please visit
.B cthulhu
wiki at
.UR http://live.gnome.org/Cthulhu
<http://live.gnome.org/Cthulhu>
.B Cthulhu
at
.UR https://git.stormux.org/storm/cthulhu
<https://git.stormux.org/storm/cthulhu>
.UE
.P
The
.B cthulhu
mailing list
.UR http://mail.gnome.org/mailman/listinfo/cthulhu-list
<http://mail.gnome.org/mailman/listinfo/cthulhu-list>
To post a message to all
.B cthulhu
list, send a email to https://groups.io/g/stormux
.B Stormux
community list is available at
.UR https://groups.io/g/stormux
<https://groups.io/g/stormux>
.UE
+1 -1
View File
@@ -1,5 +1,5 @@
project('cthulhu',
version: '2026.03.02-master',
version: '2026.05.14-master',
meson_version: '>= 1.0.0',
)
+1 -1
View File
@@ -58,7 +58,7 @@ fi
cthulhuVersionFile="${scriptDir}/src/cthulhu/cthulhuVersion.py"
mesonFile="${scriptDir}/meson.build"
pkgbuildFile="${scriptDir}/distro-packages/Arch-Linux/PKGBUILD"
pkgbuildFile="${scriptDir}/distro-packages/Arch-Linux/cthulhu/PKGBUILD"
for path in "$cthulhuVersionFile" "$mesonFile" "$pkgbuildFile"; do
if [[ ! -f "$path" ]]; then
+107 -1
View File
@@ -109,6 +109,111 @@ class Backend:
if key not in targetTable:
targetTable[key] = newValue
def _legacyProfileValue(self, profileTable, sectionName, legacyKey):
section = profileTable.get(sectionName)
if not isinstance(section, dict):
return None
return section.get(legacyKey)
def _normalizeLegacyValue(self, currentKey, legacyValue):
if currentKey == 'keyboardLayout' and isinstance(legacyValue, str):
layout = legacyValue.strip().lower()
if layout == 'desktop':
return settings.GENERAL_KEYBOARD_LAYOUT_DESKTOP
if layout == 'laptop':
return settings.GENERAL_KEYBOARD_LAYOUT_LAPTOP
return legacyValue
def _normalizeLegacyProfile(self, profileName, profileTable):
if not isinstance(profileTable, dict):
return profileTable
profileSettings = {}
legacySections = {
'metadata',
'plugins',
'ai-assistant',
'ocr',
}
for key, value in profileTable.items():
if key in legacySections:
continue
if key == 'keybindings':
if isinstance(value, dict):
keybindings = dict(value)
keybindings.pop('keyboard-layout', None)
keybindings.pop('desktop-modifier-keys', None)
keybindings.pop('laptop-modifier-keys', None)
if keybindings:
profileSettings[key] = keybindings
continue
profileSettings[key] = value
displayName = self._legacyProfileValue(profileTable, 'metadata', 'display-name')
internalName = self._legacyProfileValue(profileTable, 'metadata', 'internal-name')
if 'profile' not in profileSettings and (displayName or internalName):
profileSettings['profile'] = [
displayName or str(profileName).title(),
internalName or profileName,
]
legacyKeyMap = {
('keybindings', 'keyboard-layout'): 'keyboardLayout',
('keybindings', 'desktop-modifier-keys'): 'cthulhuModifierKeys',
('keybindings', 'laptop-modifier-keys'): 'cthulhuModifierKeys',
('plugins', 'active-plugins'): 'activePlugins',
('plugins', 'plugin-sources'): 'pluginSources',
('ai-assistant', 'enabled'): 'aiAssistantEnabled',
('ai-assistant', 'provider'): 'aiProvider',
('ai-assistant', 'api-key-file'): 'aiApiKeyFile',
('ai-assistant', 'ollama-model'): 'aiOllamaModel',
('ai-assistant', 'ollama-endpoint'): 'aiOllamaEndpoint',
('ai-assistant', 'confirmation-required'): 'aiConfirmationRequired',
('ai-assistant', 'action-timeout'): 'aiActionTimeout',
('ai-assistant', 'screenshot-quality'): 'aiScreenshotQuality',
('ai-assistant', 'max-context-length'): 'aiMaxContextLength',
('ocr', 'language-code'): 'ocrLanguageCode',
('ocr', 'scale-factor'): 'ocrScaleFactor',
('ocr', 'grayscale-image'): 'ocrGrayscaleImg',
('ocr', 'invert-image'): 'ocrInvertImg',
('ocr', 'black-white-image'): 'ocrBlackWhiteImg',
('ocr', 'black-white-threshold'): 'ocrBlackWhiteImgValue',
('ocr', 'color-calculation'): 'ocrColorCalculation',
('ocr', 'color-calculation-max'): 'ocrColorCalculationMax',
('ocr', 'copy-to-clipboard'): 'ocrCopyToClipboard',
}
for (sectionName, legacyKey), currentKey in legacyKeyMap.items():
if currentKey in profileSettings:
continue
legacyValue = self._legacyProfileValue(profileTable, sectionName, legacyKey)
if legacyValue is not None:
profileSettings[currentKey] = self._normalizeLegacyValue(currentKey, legacyValue)
return profileSettings
def _normalizeProfiles(self, profiles):
if not isinstance(profiles, dict):
return {}
return {
profileName: self._normalizeLegacyProfile(profileName, profileTable)
for profileName, profileTable in profiles.items()
}
def _normalizeProfilesDocument(self, prefsDoc):
profiles = prefsDoc.get('profiles')
if not isinstance(profiles, dict):
return
for profileName in list(profiles.keys()):
profileTable = profiles[profileName]
normalizedProfile = self._normalizeLegacyProfile(profileName, profileTable)
profiles[profileName] = self._stripNone(normalizedProfile)
if 'format-version' in prefsDoc:
del prefsDoc['format-version']
def saveDefaultSettings(self, general, pronunciations, keybindings):
""" Save default settings for all the properties from
cthulhu.settings. """
@@ -167,6 +272,7 @@ class Backend:
general = self._stripNone(general)
prefsDoc = self._readDocument(self.settingsFile)
self._normalizeProfilesDocument(prefsDoc)
profiles = prefsDoc.get('profiles')
if profiles is None or not isinstance(profiles, dict):
prefsDoc['profiles'] = {}
@@ -192,7 +298,7 @@ class Backend:
self.general = dict(prefsDoc.get('general', {}))
self.pronunciations = dict(prefsDoc.get('pronunciations', {}))
self.keybindings = dict(prefsDoc.get('keybindings', {}))
self.profiles = dict(prefsDoc.get('profiles', {}))
self.profiles = self._normalizeProfiles(dict(prefsDoc.get('profiles', {})))
except Exception:
return
+1 -1
View File
@@ -178,7 +178,7 @@ class BrailleGenerator(generator.Generator):
Atspi.Role.EXTENDED,
Atspi.Role.LINK]
# egg-list-box, e.g. privacy panel in gnome-control-center
# egg-list-box-style containers can expose selected panels as list items.
if AXUtilities.is_list_box(AXObject.get_parent(obj)):
doNotPresent.append(AXObject.get_role(obj))
+17
View File
@@ -1359,6 +1359,23 @@
<property name="width">2</property>
</packing>
</child>
<child>
<object class="GtkCheckButton" id="spatializeObjectSoundsCheckButton">
<property name="label" translatable="yes">Use stereo _position for object sounds</property>
<property name="visible">True</property>
<property name="can_focus">True</property>
<property name="receives_default">False</property>
<property name="use_underline">True</property>
<property name="xalign">0</property>
<property name="draw_indicator">True</property>
<signal name="toggled" handler="checkButtonToggled" swapped="no"/>
</object>
<packing>
<property name="left_attach">0</property>
<property name="top_attach">6</property>
<property name="width">2</property>
</packing>
</child>
</object>
</child>
</object>
-22
View File
@@ -42,7 +42,6 @@ from . import dbus_service
if TYPE_CHECKING:
from types import FrameType
from gi.repository.Gio import Settings as GSettings
from gi.repository import Gtk
from .settings_manager import SettingsManager
@@ -250,12 +249,6 @@ from gi.repository import Atspi
from gi.repository import Gdk
from gi.repository import GObject
try:
from gi.repository.Gio import Settings
a11yAppSettings: Optional[GSettings] = Settings(schema_id='org.gnome.desktop.a11y.applications')
except Exception:
a11yAppSettings = None
from . import braille
from . import debug
from . import event_manager
@@ -299,15 +292,6 @@ from . import resource_manager
# Old global variables removed - now using cthulhuApp.* instead
def onEnabledChanged(gsetting: GSettings, key: str) -> None:
try:
enabled: bool = gsetting.get_boolean(key)
except Exception:
return
if key == 'screen-reader-enabled' and not enabled:
shutdown()
EXIT_CODE_HANG: int = 50
# The user-settings module (see loadUserSettings).
@@ -651,12 +635,6 @@ def init() -> bool:
signal.alarm(0)
_initialized = True
# In theory, we can do this through dbus. In practice, it fails to
# work sometimes. Until we know why, we need to leave this as-is
# so that we respond when gnome-control-center is used to stop Cthulhu.
if a11yAppSettings:
a11yAppSettings.connect('changed', onEnabledChanged)
debug.printMessage(debug.LEVEL_INFO, 'CTHULHU: Initialized', True)
return True
+1 -1
View File
@@ -23,5 +23,5 @@
# Forked from Orca screen reader.
# Cthulhu project: https://git.stormux.org/storm/cthulhu
version = "2026.03.02"
version = "2026.05.14"
codeName = "master"
+11 -11
View File
@@ -3136,6 +3136,9 @@ print(json.dumps(result))
self.get_widget("playSoundForValueCheckButton").set_active(
prefs.get("playSoundForValue", settings.playSoundForValue)
)
self.get_widget("spatializeObjectSoundsCheckButton").set_active(
prefs.get("spatializeObjectSounds", settings.spatializeObjectSounds)
)
self.get_widget("beepProgressBarUpdatesCheckButton").set_active(
prefs.get("beepProgressBarUpdates", settings.beepProgressBarUpdates)
)
@@ -4984,12 +4987,10 @@ print(json.dumps(result))
def applyButtonClicked(self, widget):
"""Signal handler for the "clicked" signal for the applyButton
GtkButton widget. The user has clicked the Apply button.
Write out the users preferences. If GNOME accessibility hadn't
previously been enabled, warn the user that they will need to
log out. Shut down any active speech servers that were started.
Reload the users preferences to get the new speech, braille and
key echo value to take effect. Do not dismiss the configuration
window.
Write out the users preferences. Shut down any active speech servers
that were started. Reload the users preferences to get the new
speech, braille and key echo value to take effect. Do not dismiss
the configuration window.
Arguments:
- widget: the component that generated the signal.
@@ -5042,11 +5043,10 @@ print(json.dumps(result))
def okButtonClicked(self, widget=None):
"""Signal handler for the "clicked" signal for the okButton
GtkButton widget. The user has clicked the OK button.
Write out the users preferences. If GNOME accessibility hadn't
previously been enabled, warn the user that they will need to
log out. Shut down any active speech servers that were started.
Reload the users preferences to get the new speech, braille and
key echo value to take effect. Hide the configuration window.
Write out the users preferences. Shut down any active speech servers
that were started. Reload the users preferences to get the new
speech, braille and key echo value to take effect. Hide the
configuration window.
Arguments:
- widget: the component that generated the signal.
+283 -2
View File
@@ -39,7 +39,8 @@ __copyright__ = "Copyright (c) 2024 Igalia, S.L." \
"Copyright (c) 2024 GNOME Foundation Inc."
__license__ = "LGPL"
from typing import TYPE_CHECKING, Optional, Union, Tuple, List, Dict
import os
from typing import TYPE_CHECKING, Optional, Union, Tuple, List, Dict, Any
import gi
gi.require_version("Atspi", "2.0")
@@ -52,8 +53,10 @@ from . import input_event
from . import script_manager
from . import settings
from . import cthulhu_state
from .wnck_support import load_wnck
from .ax_object import AXObject
from .ax_utilities import AXUtilities
from .ax_utilities_application import AXUtilitiesApplication
if TYPE_CHECKING:
from . import keybindings
@@ -70,6 +73,9 @@ class InputEventManager:
self._mapped_keysyms: List[int] = []
self._grabbed_bindings: Dict[int, keybindings.KeyBinding] = {}
self._paused: bool = False
self._wnck = None
self._did_attempt_wnck_load: bool = False
self._scriptWithSuspendedGrabsForXterm = None
def activate_device(self) -> Atspi.Device:
"""Creates and returns the AT-SPI device used by this manager."""
@@ -367,6 +373,263 @@ class InputEventManager:
or AXUtilities.is_dialog_or_alert(x),
)
def _get_wnck(self):
"""Returns Wnck when available for X11 active-window checks."""
if not self._did_attempt_wnck_load:
self._did_attempt_wnck_load = True
try:
self._wnck = load_wnck()
except Exception as error:
msg = f"INPUT EVENT MANAGER: Wnck unavailable for active-window check: {error}"
debug.print_message(debug.LEVEL_INFO, msg, True)
self._wnck = None
return self._wnck
def _get_active_x11_window(self) -> Optional[Any]:
"""Returns the active X11 window if Wnck can provide it."""
wnck = self._get_wnck()
if wnck is None:
return None
try:
screen = wnck.Screen.get_default()
if screen is None:
return None
screen.force_update()
return screen.get_active_window()
except Exception as error:
msg = f"INPUT EVENT MANAGER: Could not obtain active X11 window: {error}"
debug.print_message(debug.LEVEL_INFO, msg, True)
return None
def _get_active_x11_window_pid(self) -> int:
"""Returns the PID of the active X11 window if Wnck can provide it."""
window = self._get_active_x11_window()
if window is None:
return -1
try:
return int(window.get_pid())
except Exception as error:
msg = f"INPUT EVENT MANAGER: Could not obtain active X11 window PID: {error}"
debug.print_message(debug.LEVEL_INFO, msg, True)
return -1
@staticmethod
def _identifier_is_xterm(value: Any) -> bool:
"""Returns True if value identifies XTerm."""
if not isinstance(value, str):
return False
identifier = os.path.basename(value.strip().lower())
return identifier == "xterm"
def _active_x11_window_is_xterm(self) -> bool:
"""Returns True when the active X11 window appears to be XTerm."""
window = self._get_active_x11_window()
if window is None:
return False
for attrName in ("get_class_group_name", "get_class_instance_name", "get_name"):
attr = getattr(window, attrName, None)
if not callable(attr):
continue
try:
if self._identifier_is_xterm(attr()):
return True
except Exception:
continue
getClassGroup = getattr(window, "get_class_group", None)
if callable(getClassGroup):
try:
classGroup = getClassGroup()
except Exception:
classGroup = None
for attrName in ("get_name", "get_res_class"):
attr = getattr(classGroup, attrName, None)
if not callable(attr):
continue
try:
if self._identifier_is_xterm(attr()):
return True
except Exception:
continue
pid = self._get_active_x11_window_pid()
if pid < 1:
return False
try:
with open(f"/proc/{pid}/cmdline", "rb") as cmdlineFile:
executable = cmdlineFile.read().split(b"\0", 1)[0].decode(errors="ignore")
except OSError:
return False
return self._identifier_is_xterm(executable)
def _find_active_x11_atspi_window(self) -> Optional[Atspi.Accessible]:
"""Returns the focused AT-SPI window for the active X11 PID, if possible."""
x11Pid = self._get_active_x11_window_pid()
if x11Pid < 1:
return None
app = AXUtilitiesApplication.get_application_with_pid(x11Pid)
if app is None:
return None
candidates = [
child for child in AXObject.iter_children(app)
if AXUtilities.is_frame(child)
or AXUtilities.is_window(child)
or AXUtilities.is_dialog_or_alert(child)
]
if not candidates:
tokens = ["INPUT EVENT MANAGER: No AT-SPI windows found for active X11 pid", x11Pid]
debug.print_tokens(debug.LEVEL_INFO, tokens, True)
return None
focusedCandidates = []
for window in candidates:
if AXUtilities.is_focused(window) or AXUtilities.get_focused_object(window) is not None:
focusedCandidates.append(window)
if len(focusedCandidates) == 1:
tokens = [
"INPUT EVENT MANAGER: Recovered active AT-SPI window from X11 pid",
x11Pid,
focusedCandidates[0],
]
debug.print_tokens(debug.LEVEL_INFO, tokens, True)
return focusedCandidates[0]
if focusedCandidates:
tokens = [
"INPUT EVENT MANAGER: Multiple focused AT-SPI windows for active X11 pid",
x11Pid,
focusedCandidates,
]
debug.print_tokens(debug.LEVEL_INFO, tokens, True)
return focusedCandidates[0]
tokens = [
"INPUT EVENT MANAGER: Active X11 pid",
x11Pid,
"has AT-SPI windows but none report focus",
candidates,
]
debug.print_tokens(debug.LEVEL_INFO, tokens, True)
return None
def _active_x11_window_differs_from(self, obj: Optional[Atspi.Accessible]) -> bool:
"""Returns True when X11 focus is known to be outside the AT-SPI context."""
if obj is None:
return False
x11Pid = self._get_active_x11_window_pid()
if x11Pid < 1:
return False
axPid = AXObject.get_process_id(obj)
if axPid < 1:
return False
result = x11Pid != axPid
if result:
tokens = [
"INPUT EVENT MANAGER: Active X11 window PID",
x11Pid,
"differs from AT-SPI active window PID",
axPid,
"for",
obj,
]
debug.print_tokens(debug.LEVEL_INFO, tokens, True)
return result
@staticmethod
def _get_active_script_app() -> Optional[Atspi.Accessible]:
"""Returns the app for the active script, if one is available."""
script = script_manager.get_manager().get_active_script()
if script is None:
return None
return getattr(script, "app", None)
@staticmethod
def _lacks_atspi_context(
window: Optional[Atspi.Accessible],
focus: Optional[Atspi.Accessible],
pendingFocus: Optional[Atspi.Accessible],
) -> bool:
"""Returns True when Cthulhu has no meaningful AT-SPI input context."""
if window is not None or focus is not None or pendingFocus is not None:
return False
return True
def _should_pass_through_for_active_xterm(
self,
window: Optional[Atspi.Accessible],
focus: Optional[Atspi.Accessible],
pendingFocus: Optional[Atspi.Accessible],
) -> bool:
"""Returns True when XTerm is active and Cthulhu lacks matching AT-SPI context."""
if pendingFocus is not None:
return False
return self._active_x11_window_is_xterm()
def _suspend_key_grabs_for_xterm(self) -> None:
"""Suspends active-script key grabs so XTerm/Fenrir can receive them."""
script = script_manager.get_manager().get_active_script()
if script is None or script is self._scriptWithSuspendedGrabsForXterm:
return
if self._scriptWithSuspendedGrabsForXterm is not None:
self._restore_key_grabs_after_xterm()
removeGrabs = getattr(script, "removeKeyGrabs", None)
if not callable(removeGrabs):
return
msg = "INPUT EVENT MANAGER: Removing active script key grabs while XTerm is focused."
debug.print_message(debug.LEVEL_INFO, msg, True)
removeGrabs()
self._scriptWithSuspendedGrabsForXterm = script
def _restore_key_grabs_after_xterm(self) -> None:
"""Restores key grabs after leaving XTerm."""
script = self._scriptWithSuspendedGrabsForXterm
self._scriptWithSuspendedGrabsForXterm = None
if script is None:
return
if script is not script_manager.get_manager().get_active_script():
return
addGrabs = getattr(script, "addKeyGrabs", None)
if not callable(addGrabs):
return
msg = "INPUT EVENT MANAGER: Restoring active script key grabs after leaving XTerm."
debug.print_message(debug.LEVEL_INFO, msg, True)
addGrabs()
# pylint: disable=too-many-arguments
# pylint: disable=too-many-positional-arguments
def process_keyboard_event(
@@ -400,10 +663,19 @@ class InputEventManager:
if pendingFocus is not None:
tokens = ["INPUT EVENT MANAGER: Using pending self-hosted focus for keyboard event:", pendingFocus]
debug.print_tokens(debug.LEVEL_INFO, tokens, True)
window = manager.get_active_window()
focus = manager.get_locus_of_focus()
if self._should_pass_through_for_active_xterm(window, focus, pendingFocus):
msg = "INPUT EVENT MANAGER: Passing through keyboard event; active XTerm is focused."
debug.print_message(debug.LEVEL_INFO, msg, True)
self._suspend_key_grabs_for_xterm()
return False
self._restore_key_grabs_after_xterm()
if pressed:
window = manager.get_active_window()
if not AXUtilities.can_be_active_window(window, clear_cache=True):
new_window = AXUtilities.find_active_window()
if new_window is None:
new_window = self._find_active_x11_atspi_window()
if new_window is not None:
window = new_window
tokens = ["INPUT EVENT MANAGER: Updating window and active window to", window]
@@ -423,6 +695,15 @@ class InputEventManager:
# One example: Brave's popup menus live in frames which lack the active
# state. Failing to revalidate the window on a key press is inconclusive;
# do not wipe out the last known window and focus state.
focus = pendingFocus or manager.get_locus_of_focus()
staleContext = window or self._get_active_script_app()
if self._active_x11_window_differs_from(staleContext):
msg = (
"INPUT EVENT MANAGER: X11 focus moved to an untracked window; "
"preserving Cthulhu key handling."
)
debug.print_message(debug.LEVEL_INFO, msg, True)
tokens = [
"WARNING:",
window,
+64
View File
@@ -66,6 +66,7 @@ from . import pronunciation_dict
from . import settings
from . import settings_manager
from . import text_attribute_names
from .ax_component import AXComponent
from .ax_document_selection import AXDocumentSelection
from .ax_object import AXObject
from .ax_selection import AXSelection
@@ -3259,6 +3260,10 @@ class Utilities:
icon = manager.getLinkSoundIcon(visited=AXUtilities.is_visited(link))
if icon:
rect = None
if cthulhu.cthulhuApp.settingsManager.getSetting('spatializeObjectSounds'):
rect = AXText.get_range_rect(obj, start_index, end_index)
self.spatializeSoundIcon(icon, link, rect)
icons.append(icon)
else:
missingIcon = True
@@ -3271,6 +3276,65 @@ class Utilities:
return spoken, icons
def spatializeSoundIcon(
self,
icon: Any,
obj: Optional[Atspi.Accessible],
rect: Optional[Any] = None,
) -> Any:
"""Attach an optional stereo pan value to an icon based on object position."""
if not cthulhu.cthulhuApp.settingsManager.getSetting('spatializeObjectSounds'):
return icon
pan = self.getSoundPanForObject(obj, rect)
if pan is None:
return icon
try:
icon.pan = pan
except Exception:
pass
return icon
def getSoundPanForObject(
self,
obj: Optional[Atspi.Accessible],
rect: Optional[Any] = None,
) -> Optional[float]:
"""Return a conservative left/right pan value for obj, or None if unknown."""
if obj is None and rect is None:
return None
rect = rect or AXComponent.get_rect(obj)
width = getattr(rect, "width", 0)
if width <= 0:
return None
frame, dialog = self.frameAndDialog(obj)
reference = dialog or frame or self.activeWindow()
if reference is None:
return None
referenceRect = AXComponent.get_rect(reference)
referenceWidth = getattr(referenceRect, "width", 0)
if referenceWidth <= 0:
return None
centerX = getattr(rect, "x", 0) + width / 2
left = getattr(referenceRect, "x", 0)
position = (centerX - left) / referenceWidth
position = max(0.0, min(1.0, position))
edge = 0.05
if position <= edge:
return -1.0
if position >= 1.0 - edge:
return 1.0
return ((position - edge) / (1.0 - (2 * edge)) * 2.0) - 1.0
def _processMultiCaseString(self, string: Any) -> Any:
return re.sub(r'(?<=[a-z])(?=[A-Z])', ' ', string)
+100
View File
@@ -1525,6 +1525,106 @@ class Script(script.Script):
for character in itemString:
self.speakCharacter(character)
def _diagnostic_object_summary(self, obj):
if obj is None:
return "None"
try:
if AXObject.is_dead(obj):
return "dead"
except Exception:
pass
try:
name = AXObject.get_name(obj) or ""
except Exception:
name = "<error>"
try:
role = AXObject.get_role_name(obj) or ""
except Exception:
role = "<error>"
try:
pid = AXObject.get_process_id(obj)
except Exception:
pid = "<error>"
return f"name={name!r}, role={role!r}, pid={pid}"
def _diagnostic_callable_value(self, obj, methodName):
method = getattr(obj, methodName, None)
if not callable(method):
return "<unavailable>"
try:
return method()
except Exception as error:
return f"<error: {error}>"
@dbus_service.command
def get_diagnostic_state(self, script=None, event=None, notify_user=True) -> str:
"""Dumps runtime state useful for diagnosing sluggish web-app behavior."""
app = cthulhu.cthulhuApp
activeScript = cthulhu_state.activeScript
eventManager = getattr(app, "eventManager", None)
compositor = getattr(app, "compositorStateAdapter", None)
inputManager = getattr(eventManager, "_inputEventManager", None)
lines = [
f"timestamp={time.strftime('%Y-%m-%d %H:%M:%S')}",
f"default-script={self.__class__.__module__}.{self.__class__.__name__}",
f"active-script={activeScript.__class__.__module__}.{activeScript.__class__.__name__}" if activeScript else "active-script=None",
f"active-window={self._diagnostic_object_summary(cthulhu_state.activeWindow)}",
f"locus-of-focus={self._diagnostic_object_summary(cthulhu_state.locusOfFocus)}",
f"pending-self-hosted-focus={self._diagnostic_object_summary(getattr(cthulhu_state, 'pendingSelfHostedFocus', None))}",
]
if eventManager is not None:
eventQueue = getattr(eventManager, "_eventQueue", None)
try:
queueSize = eventQueue.qsize() if eventQueue is not None else "<unavailable>"
except Exception as error:
queueSize = f"<error: {error}>"
prioritizedEvent = getattr(eventManager, "_prioritizedEvent", None)
lines.extend([
f"event-manager-active={getattr(eventManager, '_active', '<unavailable>')}",
f"event-queue-size={queueSize}",
f"events-suspended={getattr(eventManager, '_eventsSuspended', '<unavailable>')}",
f"churn-suppressed={getattr(eventManager, '_churnSuppressed', '<unavailable>')}",
f"state-pause-atspi-churn={cthulhu_state.pauseAtspiChurn}",
f"prioritized-context-token={getattr(eventManager, '_prioritizedContextToken', None)}",
f"state-prioritized-context-token={cthulhu_state.prioritizedDesktopContextToken}",
f"prioritized-event-type={getattr(prioritizedEvent, 'type', None)}",
f"gidle-id={getattr(eventManager, '_gidleId', '<unavailable>')}",
f"prioritized-idle-id={getattr(eventManager, '_prioritizedIdleId', '<unavailable>')}",
])
if inputManager is not None:
lines.extend([
f"key-handling-active={getattr(eventManager, '_keyHandlingActive', '<unavailable>')}",
f"input-manager-device={getattr(inputManager, '_device', None)}",
f"input-manager-watcher={getattr(inputManager, '_keyWatcher', None)}",
])
if compositor is not None:
snapshot = self._diagnostic_callable_value(compositor, "get_snapshot")
lines.append(f"compositor-snapshot={snapshot}")
if activeScript is not None:
lines.extend([
f"active-script-focus-mode={self._diagnostic_callable_value(activeScript, 'inFocusMode')}",
f"active-script-focus-sticky={self._diagnostic_callable_value(activeScript, 'focusModeIsSticky')}",
f"active-script-browse-sticky={self._diagnostic_callable_value(activeScript, 'browseModeIsSticky')}",
f"active-script-structural-navigation={getattr(getattr(activeScript, 'structuralNavigation', None), 'enabled', '<unavailable>')}",
])
report = "\n".join(lines)
debug.printMessage(debug.LEVEL_INFO, "CTHULHU DIAGNOSTIC STATE:\n" + report, True)
if notify_user and script is not None:
script.presentMessage("Cthulhu diagnostic state written to debug log.")
return report
@dbus_service.command
def sayAll(self, inputEvent, obj=None, offset=None, notify_user=True):
"""Speaks the entire document or text, starting from the current position."""
+2
View File
@@ -90,6 +90,7 @@ userCustomizableSettings = [
"playSoundForState",
"playSoundForPositionInSet",
"playSoundForValue",
"spatializeObjectSounds",
"roleSoundPresentation",
"soundTheme",
"verbalizePunctuationStyle",
@@ -353,6 +354,7 @@ playSoundForRole = False
playSoundForState = False
playSoundForPositionInSet = False
playSoundForValue = False
spatializeObjectSounds = False
roleSoundPresentation = ROLE_SOUND_PRESENTATION_SOUND_AND_SPEECH
soundTheme = "default"
+1 -2
View File
@@ -462,8 +462,7 @@ class SettingsManager(object):
debug.printMessage(debug.LEVEL_INFO, msg, True)
def _enableAccessibility(self) -> bool:
"""Enables the GNOME accessibility flag. Users need to log out and
then back in for this to take effect.
"""Enables the desktop accessibility bus flag when available.
Returns True if an action was taken (i.e., accessibility was not
set prior to this call).
+8
View File
@@ -128,6 +128,10 @@ class Player:
def stop(self, _element: Any = None) -> None:
"""Stops current sound playback."""
with self._workerLock:
if self._workerProcess is None or self._workerProcess.poll() is not None:
return
self._sendWorkerCommand({"action": "stop"}, waitForResponse=False)
def shutdown(self) -> None:
@@ -150,6 +154,7 @@ class Player:
"path": icon.path,
"volume": self._get_configured_volume(),
"interrupt": interrupt,
"pan": getattr(icon, "pan", 0.0),
},
waitForResponse=False,
)
@@ -172,6 +177,7 @@ class Player:
"path": icon.path,
"volume": self._get_configured_volume(),
"interrupt": interrupt,
"pan": getattr(icon, "pan", 0.0),
},
waitForResponse=True,
timeout=timeout,
@@ -211,6 +217,7 @@ class Player:
"frequency": tone.frequency,
"volume": tone.volume,
"wave": tone.wave,
"pan": getattr(tone, "pan", 0.0),
"interrupt": interrupt,
}
@@ -578,6 +585,7 @@ def playIconSafely(icon: Icon, timeoutSeconds: int = 10) -> Tuple[bool, Optional
"path": icon.path,
"volume": Player._get_configured_volume(),
"interrupt": True,
"pan": getattr(icon, "pan", 0.0),
},
waitForResponse=True,
timeout=max(0.1, float(timeoutSeconds)),
+21 -2
View File
@@ -52,8 +52,9 @@ METHOD_PREFIX = "_generate"
class Icon:
"""Sound file representing a particular aspect of an object."""
def __init__(self, location, filename):
def __init__(self, location, filename, pan=0.0):
self.path = os.path.join(location, filename)
self.pan = max(-1.0, min(1.0, float(pan)))
def __str__(self):
return f'Icon(path: {self.path}, isValid: {self.isValid()})'
@@ -83,6 +84,7 @@ class Tone:
self.frequency = min(max(0, frequency), 20000)
self.volume = cthulhu.cthulhuApp.settingsManager.getSetting('soundVolume') * volumeMultiplier
self.wave = wave
self.pan = 0.0
def __str__(self):
return 'Tone(duration: %s, frequency: %s, volume: %s, wave: %s)' \
@@ -105,7 +107,24 @@ class SoundGenerator(generator.Generator):
def generateSound(self, obj, **args):
"""Returns an array of sounds for the complete presentation of obj."""
return self.generate(obj, **args)
result = self.generate(obj, **args)
return self._spatializeSounds(result, obj)
def _spatializeSounds(self, sounds, obj):
if not cthulhu.cthulhuApp.settingsManager.getSetting('spatializeObjectSounds'):
return sounds
spatialize = getattr(self._script.utilities, "spatializeSoundIcon", None)
if spatialize is None:
return sounds
result = []
for sound in sounds or []:
if isinstance(sound, list):
result.append(self._spatializeSounds(sound, obj))
else:
result.append(spatialize(sound, obj))
return result
#####################################################################
# #
+35 -13
View File
@@ -40,6 +40,13 @@ def _clamp_volume(volume: Any) -> float:
return 1.0
def _clamp_pan(pan: Any) -> float:
try:
return max(-1.0, min(1.0, float(pan)))
except Exception:
return 0.0
def _write_json_line(payload: dict[str, Any]) -> None:
sys.stdout.write(json.dumps(payload) + "\n")
sys.stdout.flush()
@@ -52,24 +59,28 @@ def _report_recovery_required(message: str) -> None:
def _create_file_player(
playerId: str,
soundSink: Optional[str] = None,
) -> tuple[Optional[Any], Optional[str], Optional[str]]:
) -> tuple[Optional[Any], Optional[Any], Optional[str], Optional[str]]:
player = Gst.ElementFactory.make("playbin", playerId)
if player is None:
return None, None, "Failed to create playbin for file playback"
return None, None, None, "Failed to create playbin for file playback"
panorama = Gst.ElementFactory.make("audiopanorama", f"{playerId}-panorama")
if panorama is not None:
player.set_property("audio-filter", panorama)
configuredSink = sound_sink.normalize_sound_sink_choice(soundSink)
if configuredSink == settings.SOUND_SINK_AUTO:
return player, "playbin-default", None
return player, panorama, "playbin-default", None
audioSink, sinkName, sinkError = sound_sink.create_audio_sink(
f"{playerId}-output",
configuredSink,
)
if audioSink is None:
return None, sinkName, sinkError or f"Failed to create audio sink for {configuredSink}"
return None, panorama, sinkName, sinkError or f"Failed to create audio sink for {configuredSink}"
player.set_property("audio-sink", audioSink)
return player, sinkName or configuredSink, None
return player, panorama, sinkName or configuredSink, None
class SoundWorker:
@@ -85,7 +96,7 @@ class SoundWorker:
self._currentCommand: Optional[dict[str, Any]] = None
self._toneTimeoutId = 0
self._filePlayer, fileSinkName, fileSinkError = _create_file_player(
self._filePlayer, self._filePanorama, fileSinkName, fileSinkError = _create_file_player(
"cthulhu-sound-worker-file",
soundSink,
)
@@ -104,6 +115,7 @@ class SoundWorker:
self._toneSource = Gst.ElementFactory.make('audiotestsrc', 'cthulhu-sound-worker-source')
self._toneVolume = Gst.ElementFactory.make('volume', 'cthulhu-sound-worker-volume')
self._tonePanorama = Gst.ElementFactory.make('audiopanorama', 'cthulhu-sound-worker-panorama')
toneSink, toneSinkName, toneSinkError = sound_sink.create_audio_sink(
'cthulhu-sound-worker-tone-output',
soundSink
@@ -115,14 +127,18 @@ class SoundWorker:
self._tonePipeline.add(self._toneSource)
if self._toneVolume is not None:
self._tonePipeline.add(self._toneVolume)
if self._tonePanorama is not None:
self._tonePipeline.add(self._tonePanorama)
self._tonePipeline.add(self._toneSink)
toneElements = [self._toneSource]
if self._toneVolume is not None:
if not self._toneSource.link(self._toneVolume):
raise RuntimeError("Failed to link tone source to volume")
if not self._toneVolume.link(self._toneSink):
raise RuntimeError("Failed to link tone volume to sink")
elif not self._toneSource.link(self._toneSink):
raise RuntimeError("Failed to link tone source to sink")
toneElements.append(self._toneVolume)
if self._tonePanorama is not None:
toneElements.append(self._tonePanorama)
toneElements.append(self._toneSink)
for source, target in zip(toneElements, toneElements[1:]):
if not source.link(target):
raise RuntimeError("Failed to link tone playback pipeline")
toneBus = self._tonePipeline.get_bus()
if toneBus is None:
@@ -238,6 +254,8 @@ class SoundWorker:
self._currentCommand = command
self._filePlayer.set_property("uri", soundPath.resolve().as_uri())
self._filePlayer.set_property("volume", _clamp_volume(command.get("volume", 1.0)))
if self._filePanorama is not None:
self._filePanorama.set_property("panorama", _clamp_pan(command.get("pan", 0.0)))
stateChange = self._filePlayer.set_state(Gst.State.PLAYING)
if stateChange == Gst.StateChangeReturn.FAILURE:
self._currentCommand = None
@@ -266,6 +284,8 @@ class SoundWorker:
self._toneSource.set_property('volume', 1.0)
else:
self._toneSource.set_property('volume', _clamp_volume(command.get("volume", 1.0)))
if self._tonePanorama is not None:
self._tonePanorama.set_property("panorama", _clamp_pan(command.get("pan", 0.0)))
self._toneSource.set_property('freq', frequency)
self._toneSource.set_property('wave', wave)
@@ -377,7 +397,7 @@ def play_file_once(
print("GStreamer is not available", file=sys.stderr)
return 1
player, sinkName, sinkError = _create_file_player(
player, panorama, sinkName, sinkError = _create_file_player(
"cthulhu-sound-helper-once",
soundSink,
)
@@ -387,6 +407,8 @@ def play_file_once(
player.set_property("uri", pathlib.Path(soundPath).resolve().as_uri())
player.set_property("volume", _clamp_volume(volume))
if panorama is not None:
panorama.set_property("panorama", 0.0)
player.set_state(Gst.State.PLAYING)
bus = player.get_bus()
+2
View File
@@ -694,6 +694,8 @@ def stop() -> None:
_speechserver.stop() # type: ignore
if _echoSpeechserver and _echoSpeechserver != _speechserver:
_echoSpeechserver.stop() # type: ignore
player = sound.getPlayer()
player.stop()
def shutdown() -> None:
debug.printMessage(debug.LEVEL_INFO, 'SPEECH: Shutting down', True)
-532
View File
@@ -1,532 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2025 Stormux
# Copyright (c) 2025 Igalia, S.L.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the
# Free Software Foundation, Inc., Franklin Street, Fifth Floor,
# Boston MA 02110-1301 USA.
"""Enhanced speech settings management for D-Bus remote controller."""
__id__ = "$Id$"
__version__ = "$Revision$"
__date__ = "$Date$"
__copyright__ = "Copyright (c) 2025 Stormux"
__license__ = "LGPL"
from . import cthulhu_state
from . import debug
from . import dbus_service
from . import messages
from . import settings
from . import settings_manager
class SpeechDBusManager:
"""Enhanced speech settings for D-Bus remote control."""
def __init__(self):
"""Initialize the speech D-Bus manager."""
self._settings_manager = settings_manager.getManager()
@dbus_service.getter
def get_verbosity_level(self) -> str:
"""Returns the current speech verbosity level."""
level = self._settings_manager.getSetting("speechVerbosityLevel")
if level == settings.VERBOSITY_LEVEL_BRIEF:
return "brief"
else:
return "verbose"
@dbus_service.setter
def set_verbosity_level(self, value: str) -> bool:
"""Sets the speech verbosity level."""
if value.lower() == "brief":
setting_value = settings.VERBOSITY_LEVEL_BRIEF
elif value.lower() == "verbose":
setting_value = settings.VERBOSITY_LEVEL_VERBOSE
else:
msg = f"SPEECH DBUS MANAGER: Invalid verbosity level: {value}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
msg = f"SPEECH DBUS MANAGER: Setting verbosity level to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("speechVerbosityLevel", setting_value)
return True
@dbus_service.getter
def get_capitalization_style(self) -> str:
"""Returns the current capitalization style."""
style = self._settings_manager.getSetting("capitalizationStyle")
if style == settings.CAPITALIZATION_STYLE_NONE:
return "none"
elif style == settings.CAPITALIZATION_STYLE_SPELL:
return "spell"
elif style == settings.CAPITALIZATION_STYLE_ICON:
return "icon"
else:
return "none"
@dbus_service.setter
def set_capitalization_style(self, value: str) -> bool:
"""Sets the capitalization style."""
value_lower = value.lower()
if value_lower == "none":
setting_value = settings.CAPITALIZATION_STYLE_NONE
elif value_lower == "spell":
setting_value = settings.CAPITALIZATION_STYLE_SPELL
elif value_lower == "icon":
setting_value = settings.CAPITALIZATION_STYLE_ICON
else:
msg = f"SPEECH DBUS MANAGER: Invalid capitalization style: {value}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
msg = f"SPEECH DBUS MANAGER: Setting capitalization style to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("capitalizationStyle", setting_value)
return True
@dbus_service.getter
def get_punctuation_level(self) -> str:
"""Returns the current punctuation level."""
level = self._settings_manager.getSetting("verbalizePunctuationStyle")
if level == settings.PUNCTUATION_STYLE_NONE:
return "none"
elif level == settings.PUNCTUATION_STYLE_SOME:
return "some"
elif level == settings.PUNCTUATION_STYLE_MOST:
return "most"
elif level == settings.PUNCTUATION_STYLE_ALL:
return "all"
else:
return "some"
@dbus_service.setter
def set_punctuation_level(self, value: str) -> bool:
"""Sets the punctuation level."""
value_lower = value.lower()
if value_lower == "none":
setting_value = settings.PUNCTUATION_STYLE_NONE
elif value_lower == "some":
setting_value = settings.PUNCTUATION_STYLE_SOME
elif value_lower == "most":
setting_value = settings.PUNCTUATION_STYLE_MOST
elif value_lower == "all":
setting_value = settings.PUNCTUATION_STYLE_ALL
else:
msg = f"SPEECH DBUS MANAGER: Invalid punctuation level: {value}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
msg = f"SPEECH DBUS MANAGER: Setting punctuation level to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("verbalizePunctuationStyle", setting_value)
return True
@dbus_service.getter
def get_speak_numbers_as_digits(self) -> bool:
"""Returns whether numbers are spoken as digits."""
return self._settings_manager.getSetting("speakNumbersAsDigits")
@dbus_service.setter
def set_speak_numbers_as_digits(self, value: bool) -> bool:
"""Sets whether numbers are spoken as digits."""
msg = f"SPEECH DBUS MANAGER: Setting speak numbers as digits to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("speakNumbersAsDigits", value)
return True
@dbus_service.getter
def get_speech_is_muted(self) -> bool:
"""Returns whether speech output is temporarily muted."""
return self._settings_manager.getSetting("silenceSpeech")
@dbus_service.setter
def set_speech_is_muted(self, value: bool) -> bool:
"""Sets whether speech output is temporarily muted."""
msg = f"SPEECH DBUS MANAGER: Setting speech muted to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("silenceSpeech", value)
return True
@dbus_service.getter
def get_only_speak_displayed_text(self) -> bool:
"""Returns whether only displayed text should be spoken."""
return self._settings_manager.getSetting("onlySpeakDisplayedText")
@dbus_service.setter
def set_only_speak_displayed_text(self, value: bool) -> bool:
"""Sets whether only displayed text should be spoken."""
msg = f"SPEECH DBUS MANAGER: Setting only speak displayed text to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("onlySpeakDisplayedText", value)
return True
@dbus_service.getter
def get_speak_indentation_and_justification(self) -> bool:
"""Returns whether speaking of indentation and justification is enabled."""
return self._settings_manager.getSetting("enableSpeechIndentation")
def _sync_indentation_presentation_mode(self, enable_speech):
mode = self._settings_manager.getSetting("indentationPresentationMode") \
or settings.indentationPresentationMode
if enable_speech:
if mode == settings.INDENTATION_PRESENTATION_OFF:
mode = settings.INDENTATION_PRESENTATION_SPEECH
elif mode == settings.INDENTATION_PRESENTATION_BEEPS:
mode = settings.INDENTATION_PRESENTATION_SPEECH_AND_BEEPS
else:
if mode == settings.INDENTATION_PRESENTATION_SPEECH:
mode = settings.INDENTATION_PRESENTATION_OFF
elif mode == settings.INDENTATION_PRESENTATION_SPEECH_AND_BEEPS:
mode = settings.INDENTATION_PRESENTATION_BEEPS
self._settings_manager.setSetting("indentationPresentationMode", mode)
@dbus_service.setter
def set_speak_indentation_and_justification(self, value: bool) -> bool:
"""Sets whether speaking of indentation and justification is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting speak indentation and justification to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableSpeechIndentation", value)
self._sync_indentation_presentation_mode(value)
return True
@dbus_service.command
def toggle_speech(self, script=None, event=None):
"""Toggles speech on and off."""
tokens = ["SPEECH DBUS MANAGER: toggle_speech. Script:", script, "Event:", event]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
if script is not None:
script.presentationInterrupt()
if self.get_speech_is_muted():
self.set_speech_is_muted(False)
if script is not None:
script.presentMessage(messages.SPEECH_ENABLED)
elif not self._settings_manager.getSetting("enableSpeech"):
self._settings_manager.setSetting("enableSpeech", True)
if script is not None:
script.presentMessage(messages.SPEECH_ENABLED)
else:
if script is not None:
script.presentMessage(messages.SPEECH_DISABLED)
self.set_speech_is_muted(True)
@dbus_service.command
def toggle_verbosity(self, script=None, event=None):
"""Toggles speech verbosity level between verbose and brief."""
tokens = ["SPEECH DBUS MANAGER: toggle_verbosity. Script:", script, "Event:", event]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
current_level = self._settings_manager.getSetting("speechVerbosityLevel")
if current_level == settings.VERBOSITY_LEVEL_BRIEF:
if script is not None:
script.presentMessage(messages.SPEECH_VERBOSITY_VERBOSE)
self._settings_manager.setSetting("speechVerbosityLevel", settings.VERBOSITY_LEVEL_VERBOSE)
else:
if script is not None:
script.presentMessage(messages.SPEECH_VERBOSITY_BRIEF)
self._settings_manager.setSetting("speechVerbosityLevel", settings.VERBOSITY_LEVEL_BRIEF)
@dbus_service.command
def change_number_style(self, script=None, event=None):
"""Changes spoken number style between digits and words."""
tokens = ["SPEECH DBUS MANAGER: change_number_style. Script:", script, "Event:", event]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
speak_digits = self.get_speak_numbers_as_digits()
if speak_digits:
brief = messages.NUMBER_STYLE_WORDS_BRIEF
full = messages.NUMBER_STYLE_WORDS_FULL
else:
brief = messages.NUMBER_STYLE_DIGITS_BRIEF
full = messages.NUMBER_STYLE_DIGITS_FULL
self.set_speak_numbers_as_digits(not speak_digits)
if script is not None:
script.presentMessage(full, brief)
@dbus_service.command
def say_all(self, script=None, event=None):
"""Speaks the entire document or text, starting from the current position."""
tokens = ["SPEECH DBUS MANAGER: say_all. Script:", script, "Event:", event]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
# Use the current active script if not provided
if script is None:
script = cthulhu_state.activeScript
if script is None:
msg = "SPEECH DBUS MANAGER: No active script available for Say All"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
# Call the script's Say All method
try:
script.sayAll(event, notify_user=False)
return True
except Exception as e:
msg = f"SPEECH DBUS MANAGER: Error during Say All: {e}"
debug.printMessage(debug.LEVEL_SEVERE, msg, True)
return False
# Key Echo Controls
@dbus_service.getter
def get_key_echo_enabled(self) -> bool:
"""Returns whether echo of key presses is enabled."""
return self._settings_manager.getSetting("enableKeyEcho")
@dbus_service.setter
def set_key_echo_enabled(self, value: bool) -> bool:
"""Sets whether echo of key presses is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable key echo to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableKeyEcho", value)
return True
@dbus_service.getter
def get_character_echo_enabled(self) -> bool:
"""Returns whether echo of inserted characters is enabled."""
return self._settings_manager.getSetting("enableEchoByCharacter")
@dbus_service.setter
def set_character_echo_enabled(self, value: bool) -> bool:
"""Sets whether echo of inserted characters is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable character echo to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableEchoByCharacter", value)
return True
@dbus_service.getter
def get_word_echo_enabled(self) -> bool:
"""Returns whether word echo is enabled."""
return self._settings_manager.getSetting("enableEchoByWord")
@dbus_service.setter
def set_word_echo_enabled(self, value: bool) -> bool:
"""Sets whether word echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable word echo to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableEchoByWord", value)
return True
@dbus_service.getter
def get_sentence_echo_enabled(self) -> bool:
"""Returns whether sentence echo is enabled."""
return self._settings_manager.getSetting("enableEchoBySentence")
@dbus_service.setter
def set_sentence_echo_enabled(self, value: bool) -> bool:
"""Sets whether sentence echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable sentence echo to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableEchoBySentence", value)
return True
@dbus_service.getter
def get_alphabetic_keys_enabled(self) -> bool:
"""Returns whether alphabetic keys will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enableAlphabeticKeys")
@dbus_service.setter
def set_alphabetic_keys_enabled(self, value: bool) -> bool:
"""Sets whether alphabetic keys will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable alphabetic keys to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableAlphabeticKeys", value)
return True
@dbus_service.getter
def get_numeric_keys_enabled(self) -> bool:
"""Returns whether numeric keys will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enableNumericKeys")
@dbus_service.setter
def set_numeric_keys_enabled(self, value: bool) -> bool:
"""Sets whether numeric keys will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable numeric keys to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableNumericKeys", value)
return True
@dbus_service.getter
def get_punctuation_keys_enabled(self) -> bool:
"""Returns whether punctuation keys will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enablePunctuationKeys")
@dbus_service.setter
def set_punctuation_keys_enabled(self, value: bool) -> bool:
"""Sets whether punctuation keys will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable punctuation keys to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enablePunctuationKeys", value)
return True
@dbus_service.getter
def get_space_enabled(self) -> bool:
"""Returns whether space key will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enableSpace")
@dbus_service.setter
def set_space_enabled(self, value: bool) -> bool:
"""Sets whether space key will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable space to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableSpace", value)
return True
@dbus_service.getter
def get_modifier_keys_enabled(self) -> bool:
"""Returns whether modifier keys will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enableModifierKeys")
@dbus_service.setter
def set_modifier_keys_enabled(self, value: bool) -> bool:
"""Sets whether modifier keys will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable modifier keys to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableModifierKeys", value)
return True
@dbus_service.getter
def get_function_keys_enabled(self) -> bool:
"""Returns whether function keys will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enableFunctionKeys")
@dbus_service.setter
def set_function_keys_enabled(self, value: bool) -> bool:
"""Sets whether function keys will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable function keys to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableFunctionKeys", value)
return True
@dbus_service.getter
def get_action_keys_enabled(self) -> bool:
"""Returns whether action keys will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enableActionKeys")
@dbus_service.setter
def set_action_keys_enabled(self, value: bool) -> bool:
"""Sets whether action keys will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable action keys to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableActionKeys", value)
return True
@dbus_service.getter
def get_navigation_keys_enabled(self) -> bool:
"""Returns whether navigation keys will be echoed when key echo is enabled."""
return self._settings_manager.getSetting("enableNavigationKeys")
@dbus_service.setter
def set_navigation_keys_enabled(self, value: bool) -> bool:
"""Sets whether navigation keys will be echoed when key echo is enabled."""
msg = f"SPEECH DBUS MANAGER: Setting enable navigation keys to {value}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._settings_manager.setSetting("enableNavigationKeys", value)
return True
@dbus_service.command
def cycle_key_echo(self, script=None, event=None):
"""Cycle through the key echo levels."""
tokens = ["SPEECH DBUS MANAGER: cycle_key_echo. Script:", script, "Event:", event]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
# Get current settings
key = self._settings_manager.getSetting("enableKeyEcho")
word = self._settings_manager.getSetting("enableEchoByWord")
sentence = self._settings_manager.getSetting("enableEchoBySentence")
# Cycle through the combinations: none -> key -> word -> sentence -> all -> none
if not key and not word and not sentence:
# None -> Key only
new_key, new_word, new_sentence = True, False, False
brief = messages.KEY_ECHO_KEY_BRIEF
full = messages.KEY_ECHO_KEY_FULL
elif key and not word and not sentence:
# Key -> Word
new_key, new_word, new_sentence = False, True, False
brief = messages.KEY_ECHO_WORD_BRIEF
full = messages.KEY_ECHO_WORD_FULL
elif not key and word and not sentence:
# Word -> Sentence
new_key, new_word, new_sentence = False, False, True
brief = messages.KEY_ECHO_SENTENCE_BRIEF
full = messages.KEY_ECHO_SENTENCE_FULL
elif not key and not word and sentence:
# Sentence -> All
new_key, new_word, new_sentence = True, True, True
brief = messages.KEY_ECHO_KEY_AND_WORD_BRIEF
full = messages.KEY_ECHO_KEY_AND_WORD_FULL
else:
# All -> None
new_key, new_word, new_sentence = False, False, False
brief = messages.KEY_ECHO_NONE_BRIEF
full = messages.KEY_ECHO_NONE_FULL
# Apply new settings
self._settings_manager.setSetting("enableKeyEcho", new_key)
self._settings_manager.setSetting("enableEchoByWord", new_word)
self._settings_manager.setSetting("enableEchoBySentence", new_sentence)
if script is not None:
script.presentMessage(full, brief)
+30 -9
View File
@@ -122,11 +122,30 @@ class SpeechGenerator(generator.Generator):
"""Return the themed sound icon for obj's role, if any."""
if role == Atspi.Role.LINK or AXUtilities.is_link(obj):
return sound_theme_manager.getManager().getLinkSoundIcon(
icon = sound_theme_manager.getManager().getLinkSoundIcon(
visited=AXUtilities.is_visited(obj)
)
return self._spatializeRoleSoundIcon(icon, obj)
return sound_theme_manager.getManager().getRoleSoundIcon(role)
icon = sound_theme_manager.getManager().getRoleSoundIcon(role)
return self._spatializeRoleSoundIcon(icon, obj)
def _spatializeRoleSoundIcon(self, icon, obj):
"""Spatialize icon if the experimental object sound setting is enabled."""
try:
enabled = cthulhu.cthulhuApp.settingsManager.getSetting('spatializeObjectSounds')
except Exception:
enabled = False
if not enabled:
return icon
spatialize = getattr(self._script.utilities, "spatializeSoundIcon", None)
if spatialize is None:
return icon
return spatialize(icon, obj)
def _addGlobals(self, globalsDict):
"""Other things to make available from the formatting string.
@@ -717,7 +736,7 @@ class SpeechGenerator(generator.Generator):
and AXUtilities.is_selected(obj):
return []
# egg-list-box, e.g. privacy panel in gnome-control-center
# egg-list-box-style containers can expose selected panels as list items.
if AXUtilities.is_list_box(parent):
doNotPresent.append(AXObject.get_role(obj))
@@ -827,7 +846,7 @@ class SpeechGenerator(generator.Generator):
# #
#####################################################################
def _applyStateSound(self, result, role, stateKey):
def _applyStateSound(self, result, role, stateKey, obj=None):
if not result:
return result
@@ -841,6 +860,7 @@ class SpeechGenerator(generator.Generator):
icon = sound_theme_manager.getManager().getRoleStateSoundIcon(role, stateKey)
if not icon:
return result
icon = self._spatializeRoleSoundIcon(icon, obj)
if roleSoundPresentation == settings.ROLE_SOUND_PRESENTATION_SOUND_ONLY:
return [icon]
@@ -868,7 +888,7 @@ class SpeechGenerator(generator.Generator):
stateKey = "checked"
else:
stateKey = "unchecked"
result = self._applyStateSound(result, role, stateKey)
result = self._applyStateSound(result, role, stateKey, obj)
return result
def _generateExpandableState(self, obj, **args):
@@ -909,7 +929,8 @@ class SpeechGenerator(generator.Generator):
result = self._applyStateSound(
result,
Atspi.Role.CHECK_MENU_ITEM,
"checked"
"checked",
obj,
)
return result
@@ -941,7 +962,7 @@ class SpeechGenerator(generator.Generator):
result.extend(self.voice(STATE, obj=obj, **args))
stateKey = "checked" if AXUtilities.is_checked(obj) else "unchecked"
role = args.get('role', AXObject.get_role(obj))
result = self._applyStateSound(result, role, stateKey)
result = self._applyStateSound(result, role, stateKey, obj)
return result
def _generateSwitchState(self, obj, **args):
@@ -955,7 +976,7 @@ class SpeechGenerator(generator.Generator):
stateKey = "checked" if (AXUtilities.is_checked(obj) or AXUtilities.is_pressed(obj)) \
else "unchecked"
role = args.get('role', AXObject.get_role(obj))
result = self._applyStateSound(result, role, stateKey)
result = self._applyStateSound(result, role, stateKey, obj)
return result
def _generateToggleState(self, obj, **args):
@@ -973,7 +994,7 @@ class SpeechGenerator(generator.Generator):
stateKey = "checked" if (AXUtilities.is_checked(obj) or AXUtilities.is_pressed(obj)) \
else "unchecked"
role = args.get('role', AXObject.get_role(obj))
result = self._applyStateSound(result, role, stateKey)
result = self._applyStateSound(result, role, stateKey, obj)
return result
#####################################################################
@@ -0,0 +1,328 @@
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu import input_event_manager
class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
def test_active_x11_window_differs_from_cached_atspi_window_by_pid(self):
manager = input_event_manager.InputEventManager()
cachedWindow = object()
with (
mock.patch.object(manager, "_get_active_x11_window_pid", return_value=1002),
mock.patch.object(input_event_manager.AXObject, "get_process_id", return_value=1001),
mock.patch.object(input_event_manager.debug, "print_tokens"),
):
self.assertTrue(manager._active_x11_window_differs_from(cachedWindow))
def test_keyboard_event_preserves_key_handling_when_unknown_x11_window_is_not_xterm(self):
manager = input_event_manager.InputEventManager()
staleWindow = object()
staleFocus = object()
focusManager = mock.Mock()
focusManager.get_active_window.return_value = staleWindow
focusManager.get_locus_of_focus.return_value = staleFocus
scriptManager = mock.Mock()
keyboardEvent = mock.Mock()
keyboardEvent.is_modifier_key.return_value = False
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None),
mock.patch.object(manager, "_get_top_level_window", return_value=None),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False),
mock.patch.object(manager, "_active_x11_window_differs_from", return_value=True),
mock.patch.object(manager, "last_event_was_keyboard", return_value=False),
mock.patch.object(input_event_manager.debug, "print_message"),
):
result = manager.process_keyboard_event(
mock.Mock(),
True,
36,
65293,
0,
"Return",
)
self.assertTrue(result)
scriptManager.set_active_script.assert_not_called()
focusManager.clear_state.assert_not_called()
keyboardEvent.process.assert_called_once_with()
def test_keyboard_event_uses_active_script_app_when_cached_window_is_missing(self):
manager = input_event_manager.InputEventManager()
staleApp = object()
staleScript = mock.Mock(app=staleApp)
focusManager = mock.Mock()
focusManager.get_active_window.return_value = None
focusManager.get_locus_of_focus.return_value = None
scriptManager = mock.Mock()
scriptManager.get_active_script.return_value = staleScript
keyboardEvent = mock.Mock()
keyboardEvent.is_modifier_key.return_value = False
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None),
mock.patch.object(manager, "_get_top_level_window", return_value=None),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False),
mock.patch.object(manager, "_active_x11_window_differs_from", return_value=True) as differs,
mock.patch.object(manager, "last_event_was_keyboard", return_value=False),
mock.patch.object(input_event_manager.debug, "print_message"),
):
result = manager.process_keyboard_event(
mock.Mock(),
True,
81,
65434,
0,
"KP_Page_Up",
)
self.assertTrue(result)
differs.assert_called_once_with(staleApp)
scriptManager.set_active_script.assert_not_called()
focusManager.clear_state.assert_not_called()
keyboardEvent.process.assert_called_once_with()
def test_keyboard_press_keeps_cthulhu_commands_when_unknown_window_is_not_xterm(self):
manager = input_event_manager.InputEventManager()
focusManager = mock.Mock()
focusManager.get_active_window.return_value = None
focusManager.get_locus_of_focus.return_value = None
scriptManager = mock.Mock()
activeScript = mock.Mock(app=None)
scriptManager.get_active_script.return_value = activeScript
keyboardEvent = mock.Mock()
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.cthulhu_state, "pendingSelfHostedFocus", None),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None),
mock.patch.object(manager, "_get_top_level_window", return_value=None),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=False),
mock.patch.object(manager, "_active_x11_window_differs_from", return_value=False),
mock.patch.object(manager, "last_event_was_keyboard", return_value=False),
mock.patch.object(input_event_manager.debug, "print_message"),
):
result = manager.process_keyboard_event(
mock.Mock(),
True,
79,
65429,
0,
"KP_Home",
)
self.assertTrue(result)
activeScript.removeKeyGrabs.assert_not_called()
keyboardEvent.process.assert_called_once_with()
def test_keyboard_press_suspends_grabs_and_passes_through_when_active_xterm_is_focused(self):
manager = input_event_manager.InputEventManager()
staleWindow = object()
staleFocus = object()
focusManager = mock.Mock()
focusManager.get_active_window.return_value = staleWindow
focusManager.get_locus_of_focus.return_value = staleFocus
scriptManager = mock.Mock()
activeScript = mock.Mock(app=None)
scriptManager.get_active_script.return_value = activeScript
keyboardEvent = mock.Mock()
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.cthulhu_state, "pendingSelfHostedFocus", None),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=True),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window") as canBeActiveWindow,
mock.patch.object(input_event_manager.AXUtilities, "find_active_window") as findActiveWindow,
mock.patch.object(input_event_manager.debug, "print_message"),
):
result = manager.process_keyboard_event(
mock.Mock(),
True,
79,
65429,
0,
"KP_Home",
)
self.assertFalse(result)
activeScript.removeKeyGrabs.assert_called_once_with()
canBeActiveWindow.assert_not_called()
findActiveWindow.assert_not_called()
keyboardEvent.process.assert_not_called()
def test_finds_focused_atspi_window_for_active_x11_pid(self):
manager = input_event_manager.InputEventManager()
app = object()
unfocusedWindow = object()
focusedWindow = object()
with (
mock.patch.object(manager, "_get_active_x11_window_pid", return_value=23875),
mock.patch.object(
input_event_manager.AXUtilitiesApplication,
"get_application_with_pid",
return_value=app,
),
mock.patch.object(
input_event_manager.AXObject,
"iter_children",
return_value=iter([unfocusedWindow, focusedWindow]),
),
mock.patch.object(input_event_manager.AXUtilities, "is_frame", return_value=True),
mock.patch.object(input_event_manager.AXUtilities, "is_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "is_dialog_or_alert", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "is_focused", return_value=False),
mock.patch.object(
input_event_manager.AXUtilities,
"get_focused_object",
side_effect=[None, object()],
),
mock.patch.object(input_event_manager.debug, "print_tokens"),
):
result = manager._find_active_x11_atspi_window()
self.assertIs(result, focusedWindow)
def test_keyboard_event_recovers_context_from_active_x11_pid_window(self):
manager = input_event_manager.InputEventManager()
recoveredWindow = object()
focusManager = mock.Mock()
focusManager.get_active_window.return_value = None
focusManager.get_locus_of_focus.return_value = None
scriptManager = mock.Mock()
activeScript = mock.Mock(app=None)
scriptManager.get_active_script.return_value = activeScript
keyboardEvent = mock.Mock()
keyboardEvent.is_modifier_key.return_value = False
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.cthulhu_state, "pendingSelfHostedFocus", None),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(input_event_manager.AXUtilities, "can_be_active_window", return_value=False),
mock.patch.object(input_event_manager.AXUtilities, "find_active_window", return_value=None),
mock.patch.object(manager, "_find_active_x11_atspi_window", return_value=recoveredWindow),
mock.patch.object(manager, "last_event_was_keyboard", return_value=False),
mock.patch.object(input_event_manager.debug, "print_message"),
mock.patch.object(input_event_manager.debug, "print_tokens"),
):
result = manager.process_keyboard_event(
mock.Mock(),
True,
79,
65429,
0,
"KP_Home",
)
self.assertTrue(result)
focusManager.set_active_window.assert_called_once_with(recoveredWindow)
activeScript.removeKeyGrabs.assert_not_called()
keyboardEvent.set_window.assert_called_once_with(recoveredWindow)
keyboardEvent.process.assert_called_once_with()
def test_keyboard_release_suspends_grabs_and_passes_through_when_active_xterm_is_focused(self):
manager = input_event_manager.InputEventManager()
focusManager = mock.Mock()
focusManager.get_active_window.return_value = None
focusManager.get_locus_of_focus.return_value = None
scriptManager = mock.Mock()
activeScript = mock.Mock(app=None)
scriptManager.get_active_script.return_value = activeScript
keyboardEvent = mock.Mock()
with (
mock.patch.object(input_event_manager.cthulhu_state, "capturingKeys", False),
mock.patch.object(input_event_manager.cthulhu_state, "pendingSelfHostedFocus", None),
mock.patch.object(input_event_manager.focus_manager, "get_manager", return_value=focusManager),
mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager),
mock.patch.object(input_event_manager.input_event, "KeyboardEvent", return_value=keyboardEvent),
mock.patch.object(manager, "_should_pass_through_for_active_xterm", return_value=True),
mock.patch.object(input_event_manager.debug, "print_message"),
):
result = manager.process_keyboard_event(
mock.Mock(),
False,
79,
65429,
0,
"KP_Home",
)
self.assertFalse(result)
activeScript.removeKeyGrabs.assert_called_once_with()
keyboardEvent.process.assert_not_called()
def test_xterm_grab_suspension_is_idempotent(self):
manager = input_event_manager.InputEventManager()
activeScript = mock.Mock()
scriptManager = mock.Mock()
scriptManager.get_active_script.return_value = activeScript
with mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager):
manager._suspend_key_grabs_for_xterm()
manager._suspend_key_grabs_for_xterm()
activeScript.removeKeyGrabs.assert_called_once_with()
def test_xterm_grab_restore_readds_grabs_for_same_active_script(self):
manager = input_event_manager.InputEventManager()
activeScript = mock.Mock()
scriptManager = mock.Mock()
scriptManager.get_active_script.return_value = activeScript
with mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager):
manager._suspend_key_grabs_for_xterm()
manager._restore_key_grabs_after_xterm()
activeScript.removeKeyGrabs.assert_called_once_with()
activeScript.addKeyGrabs.assert_called_once_with()
def test_xterm_grab_restore_does_not_readd_grabs_for_inactive_script(self):
manager = input_event_manager.InputEventManager()
oldScript = mock.Mock()
newScript = mock.Mock()
scriptManager = mock.Mock()
scriptManager.get_active_script.side_effect = [oldScript, newScript]
with mock.patch.object(input_event_manager.script_manager, "get_manager", return_value=scriptManager):
manager._suspend_key_grabs_for_xterm()
manager._restore_key_grabs_after_xterm()
oldScript.removeKeyGrabs.assert_called_once_with()
oldScript.addKeyGrabs.assert_not_called()
def test_identifier_is_xterm_matches_exact_xterm_only(self):
self.assertTrue(input_event_manager.InputEventManager._identifier_is_xterm("xterm"))
self.assertTrue(input_event_manager.InputEventManager._identifier_is_xterm("/usr/bin/xterm"))
self.assertFalse(input_event_manager.InputEventManager._identifier_is_xterm("uxterm"))
self.assertFalse(input_event_manager.InputEventManager._identifier_is_xterm("gnome-terminal"))
if __name__ == "__main__":
unittest.main()
+69
View File
@@ -16,6 +16,7 @@ speech_stub.speak = mock.Mock()
sys.modules.setdefault("cthulhu.speech", speech_stub)
from cthulhu import settings
from cthulhu.sound_generator import SoundGenerator
from cthulhu.sound_theme_manager import SoundThemeManager
from cthulhu.speech_generator import SpeechGenerator
from cthulhu.script_utilities import Utilities
@@ -146,6 +147,74 @@ class LinkIndicatorPresentationRegressionTests(unittest.TestCase):
self.assertEqual(spoken, "Docs link")
self.assertEqual(icons, [icon])
def test_sound_pan_uses_object_position_inside_active_frame(self):
utility = object.__new__(Utilities)
obj = mock.Mock()
frame = mock.Mock()
icon = SimpleNamespace(path="/tmp/link.wav")
fakeApp = mock.Mock()
fakeApp.settingsManager.getSetting.return_value = True
def get_rect(target):
if target is frame:
return SimpleNamespace(x=0, y=0, width=1000, height=800)
if target is obj:
return SimpleNamespace(x=495, y=0, width=10, height=20)
return SimpleNamespace(x=0, y=0, width=0, height=0)
utility.frameAndDialog = mock.Mock(return_value=[frame, None])
with (
mock.patch("cthulhu.script_utilities.cthulhu.cthulhuApp", fakeApp),
mock.patch("cthulhu.script_utilities.AXComponent.get_rect", side_effect=get_rect),
):
utility.spatializeSoundIcon(icon, obj)
self.assertEqual(icon.pan, 0.0)
def test_sound_pan_hard_pans_only_at_extreme_edges(self):
utility = object.__new__(Utilities)
obj = mock.Mock()
frame = mock.Mock()
fakeApp = mock.Mock()
fakeApp.settingsManager.getSetting.return_value = True
utility.frameAndDialog = mock.Mock(return_value=[frame, None])
with (
mock.patch("cthulhu.script_utilities.cthulhu.cthulhuApp", fakeApp),
mock.patch(
"cthulhu.script_utilities.AXComponent.get_rect",
side_effect=[
SimpleNamespace(x=0, y=0, width=10, height=20),
SimpleNamespace(x=0, y=0, width=1000, height=800),
],
),
):
self.assertEqual(utility.getSoundPanForObject(obj), -1.0)
class SoundGeneratorSpatializationRegressionTests(unittest.TestCase):
def test_generated_icons_and_tones_are_spatialized(self):
generator = object.__new__(SoundGenerator)
icon = SimpleNamespace(path="/tmp/link.wav")
tone = SimpleNamespace(duration=0.1, frequency=440, volume=0.5, wave=0)
generator._script = mock.Mock()
generator._script.utilities.spatializeSoundIcon.side_effect = lambda sound, obj: sound
fakeApp = mock.Mock()
fakeApp.settingsManager.getSetting.return_value = True
obj = mock.Mock()
with mock.patch("cthulhu.sound_generator.cthulhu.cthulhuApp", fakeApp):
result = generator._spatializeSounds([icon, [tone]], obj)
self.assertEqual(result, [icon, [tone]])
generator._script.utilities.spatializeSoundIcon.assert_has_calls(
[mock.call(icon, obj), mock.call(tone, obj)]
)
if __name__ == "__main__":
unittest.main()
+47
View File
@@ -0,0 +1,47 @@
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cthulhu.plugins.nvda2cthulhu import plugin as nvda_plugin
class Nvda2CthulhuRegressionTests(unittest.TestCase):
def _make_plugin(self):
plugin = nvda_plugin.Nvda2Cthulhu.__new__(nvda_plugin.Nvda2Cthulhu)
plugin.interruptEnabled = True
plugin._dependencies_available = mock.Mock(return_value=True)
return plugin
def test_text_cancel_speech_frame_is_cancel_request(self):
plugin = self._make_plugin()
self.assertEqual(("CancelSpeech", None), plugin._parse_request("CancelSpeech"))
def test_websocket_speech_is_queued_instead_of_handled_synchronously(self):
plugin = self._make_plugin()
plugin._translation_enabled = mock.Mock(return_value=False)
with mock.patch.object(nvda_plugin.speech, "speak") as speak:
plugin.handle_message("hello")
speak.assert_not_called()
def test_queued_websocket_speech_runs_on_main_loop(self):
if not hasattr(nvda_plugin, "GLib"):
self.fail("nvda2cthulhu does not expose a GLib main-loop dispatcher")
plugin = self._make_plugin()
plugin._translation_enabled = mock.Mock(return_value=False)
with mock.patch.object(nvda_plugin.GLib, "idle_add", side_effect=lambda callback, *args: callback(*args)), \
mock.patch.object(nvda_plugin.speech, "speak") as speak:
plugin.handle_message("hello")
speak.assert_called_once_with("hello", interrupt=True)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,32 @@
import sys
import unittest
import importlib
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
sys.modules.pop("cthulhu.speech", None)
speech = importlib.import_module("cthulhu.speech")
class PresentationInterruptSoundRegressionTests(unittest.TestCase):
def test_speech_stop_also_stops_sound_playback(self):
speechServer = mock.Mock()
echoServer = mock.Mock()
soundPlayer = mock.Mock()
with (
mock.patch.object(speech, "_speechserver", speechServer),
mock.patch.object(speech, "_echoSpeechserver", echoServer),
mock.patch.object(speech.sound, "getPlayer", return_value=soundPlayer),
):
speech.stop()
speechServer.stop.assert_called_once_with()
echoServer.stop.assert_called_once_with()
soundPlayer.stop.assert_called_once_with()
if __name__ == "__main__":
unittest.main()
+28 -6
View File
@@ -20,36 +20,48 @@ class _FakePlaybin:
class SoundHelperBackendTests(unittest.TestCase):
def test_create_file_player_uses_playbin_default_sink_for_auto(self):
fakePlaybin = _FakePlaybin()
fakePanorama = object()
with (
mock.patch.object(
sound_helper.Gst.ElementFactory,
"make",
return_value=fakePlaybin,
side_effect=[fakePlaybin, fakePanorama],
) as makeElement,
mock.patch.object(
sound_helper.sound_sink,
"create_audio_sink",
) as createAudioSink,
):
player, sinkName, error = sound_helper._create_file_player("worker-file", settings.SOUND_SINK_AUTO)
player, panorama, sinkName, error = sound_helper._create_file_player(
"worker-file",
settings.SOUND_SINK_AUTO,
)
self.assertIs(player, fakePlaybin)
self.assertIs(panorama, fakePanorama)
self.assertEqual(sinkName, "playbin-default")
self.assertIsNone(error)
createAudioSink.assert_not_called()
makeElement.assert_called_once_with("playbin", "worker-file")
makeElement.assert_has_calls(
[
mock.call("playbin", "worker-file"),
mock.call("audiopanorama", "worker-file-panorama"),
]
)
self.assertNotIn("audio-sink", fakePlaybin.properties)
self.assertIs(fakePlaybin.properties["audio-filter"], fakePanorama)
def test_create_file_player_sets_explicit_sink_when_requested(self):
fakePlaybin = _FakePlaybin()
fakePanorama = object()
fakeSink = object()
with (
mock.patch.object(
sound_helper.Gst.ElementFactory,
"make",
return_value=fakePlaybin,
side_effect=[fakePlaybin, fakePanorama],
) as makeElement,
mock.patch.object(
sound_helper.sound_sink,
@@ -57,14 +69,24 @@ class SoundHelperBackendTests(unittest.TestCase):
return_value=(fakeSink, "pulsesink", None),
) as createAudioSink,
):
player, sinkName, error = sound_helper._create_file_player("worker-file", settings.SOUND_SINK_PULSE)
player, panorama, sinkName, error = sound_helper._create_file_player(
"worker-file",
settings.SOUND_SINK_PULSE,
)
self.assertIs(player, fakePlaybin)
self.assertIs(panorama, fakePanorama)
self.assertEqual(sinkName, "pulsesink")
self.assertIsNone(error)
makeElement.assert_called_once_with("playbin", "worker-file")
makeElement.assert_has_calls(
[
mock.call("playbin", "worker-file"),
mock.call("audiopanorama", "worker-file-panorama"),
]
)
createAudioSink.assert_called_once()
self.assertIs(fakePlaybin.properties["audio-sink"], fakeSink)
self.assertIs(fakePlaybin.properties["audio-filter"], fakePanorama)
if __name__ == "__main__":
@@ -41,6 +41,7 @@ class SoundPreferencesBuilderTests(unittest.TestCase):
self.assertIn("soundsTabLabel", objectIds)
self.assertIn("enableSoundCheckButton", objectIds)
self.assertIn("soundVolumeScale", objectIds)
self.assertIn("spatializeObjectSoundsCheckButton", objectIds)
self.assertIn("progressBarBeepIntervalSpinButton", objectIds)
def test_notebook_tab_positions_keep_sounds_page_and_label_aligned(self):
@@ -90,6 +91,7 @@ class SoundPreferencesControllerTests(unittest.TestCase):
"playSoundForState": True,
"playSoundForPositionInSet": False,
"playSoundForValue": False,
"spatializeObjectSounds": True,
}
widgets = {
"soundSinkCombo": mock.Mock(),
@@ -102,6 +104,7 @@ class SoundPreferencesControllerTests(unittest.TestCase):
"playSoundForStateCheckButton": mock.Mock(),
"playSoundForPositionInSetCheckButton": mock.Mock(),
"playSoundForValueCheckButton": mock.Mock(),
"spatializeObjectSoundsCheckButton": mock.Mock(),
"beepProgressBarUpdatesCheckButton": mock.Mock(),
}
gui.get_widget = widgets.__getitem__
@@ -128,6 +131,7 @@ class SoundPreferencesControllerTests(unittest.TestCase):
widgets["playSoundForStateCheckButton"].set_active.assert_called_once_with(True)
widgets["playSoundForPositionInSetCheckButton"].set_active.assert_called_once_with(False)
widgets["playSoundForValueCheckButton"].set_active.assert_called_once_with(False)
widgets["spatializeObjectSoundsCheckButton"].set_active.assert_called_once_with(True)
widgets["beepProgressBarUpdatesCheckButton"].set_active.assert_called_once_with(True)
widgets["progressBarBeepIntervalSpinButton"].set_value.assert_called_once_with(0)
+46 -1
View File
@@ -38,8 +38,11 @@ from cthulhu import sound_sink
class _FakeProcess:
def __init__(self, returnCode=None):
self.returnCode = returnCode
def poll(self):
return None
return self.returnCode
class SoundSinkTests(unittest.TestCase):
@@ -107,6 +110,48 @@ class PlayerRecoveryTests(unittest.TestCase):
self.assertEqual(stopReasons, ["lost audio sink"])
self.assertEqual(startedSinks, [settings.SOUND_SINK_AUTO])
def test_stop_does_not_start_worker_when_worker_is_not_running(self):
player = sound.Player()
with mock.patch.object(player, "_sendWorkerCommand") as sendCommand:
player.stop()
sendCommand.assert_not_called()
def test_stop_sends_command_to_running_worker(self):
player = sound.Player()
player._workerProcess = _FakeProcess()
with mock.patch.object(player, "_sendWorkerCommand") as sendCommand:
player.stop()
sendCommand.assert_called_once_with({"action": "stop"}, waitForResponse=False)
def test_icon_pan_is_sent_to_worker(self):
player = sound.Player()
icon = sound.Icon("/tmp", "link.wav")
icon.pan = -0.5
with (
mock.patch.object(icon, "isValid", return_value=True),
mock.patch.object(player, "_sendWorkerCommand", return_value=(True, None)) as sendCommand,
):
player.play(icon)
command = sendCommand.call_args.args[0]
self.assertEqual(command["pan"], -0.5)
def test_tone_pan_is_sent_to_worker(self):
player = sound.Player()
tone = sound.Tone(0.1, 440)
tone.pan = 0.5
with mock.patch.object(player, "_sendWorkerCommand", return_value=(True, None)) as sendCommand:
player.play(tone)
command = sendCommand.call_args.args[0]
self.assertEqual(command["pan"], 0.5)
if __name__ == "__main__":
unittest.main()
@@ -86,6 +86,25 @@ class LegacyTomlSchemaMigrationTests(unittest.TestCase):
self.assertNotIn("format-version = 2", savedSettings)
self.assertNotIn("[profiles.default.metadata]", savedSettings)
def test_legacy_profile_keybindings_are_preserved(self):
legacySettings = LEGACY_SETTINGS.replace(
'desktop-modifier-keys = ["Insert", "KP_Insert"]',
'desktop-modifier-keys = ["Insert", "KP_Insert"]\ncustom-binding = "kb:cthulhu+x"',
)
with tempfile.TemporaryDirectory() as tempDir:
Path(tempDir, "user-settings.toml").write_text(
legacySettings,
encoding="utf-8",
)
backend = Backend(tempDir)
self.assertEqual(
backend.getKeybindings("default"),
{"custom-binding": "kb:cthulhu+x"},
)
if __name__ == "__main__":
unittest.main()