From 921ffc4145c18fda9a83e96534847a3da31d0b0d Mon Sep 17 00:00:00 2001 From: Storm Dragon Date: Fri, 2 Jan 2026 00:37:52 -0500 Subject: [PATCH] Steam improvements. I can't say it's perfect by any means, but much much improved. --- src/cthulhu/event_manager.py | 49 ++- src/cthulhu/script_manager.py | 6 + src/cthulhu/scripts/apps/__init__.py | 1 + src/cthulhu/scripts/apps/meson.build | 1 + .../scripts/apps/steamwebhelper/__init__.py | 1 + .../scripts/apps/steamwebhelper/meson.build | 9 + .../scripts/apps/steamwebhelper/script.py | 305 ++++++++++++++++++ 7 files changed, 369 insertions(+), 3 deletions(-) create mode 100644 src/cthulhu/scripts/apps/steamwebhelper/__init__.py create mode 100644 src/cthulhu/scripts/apps/steamwebhelper/meson.build create mode 100644 src/cthulhu/scripts/apps/steamwebhelper/script.py diff --git a/src/cthulhu/event_manager.py b/src/cthulhu/event_manager.py index 35fc16a..f9a6624 100644 --- a/src/cthulhu/event_manager.py +++ b/src/cthulhu/event_manager.py @@ -60,6 +60,7 @@ class EventManager: self._active = False self._enqueueCount = 0 self._dequeueCount = 0 + self._cmdlineCache = {} self._eventQueue = queue.Queue(0) self._gidleId = 0 self._gidleLock = threading.Lock() @@ -175,6 +176,43 @@ class EventManager: return False + def _getAppCmdline(self, app): + pid = AXObject.get_process_id(app) + if pid == -1: + return "" + + cmdline = self._cmdlineCache.get(pid) + if cmdline is None: + cmdline = debug.getCmdline(pid) + self._cmdlineCache[pid] = cmdline + return cmdline + + def _isSteamApp(self, app): + name = AXObject.get_name(app) + if not name: + nameLower = "" + else: + nameLower = name.lower() + + if "steamwebhelper" in nameLower or nameLower in ("steam", "steam web helper"): + return True + + cmdline = self._getAppCmdline(app) + return "steamwebhelper" in cmdline + + def _isSteamNotificationEvent(self, event): + for obj in (event.any_data, event.source): + if not isinstance(obj, Atspi.Accessible): + continue + if AXUtilities.is_notification(obj) or AXUtilities.is_alert(obj): + return True + liveAttr = AXObject.get_attribute(obj, 'live') + containerLive = AXObject.get_attribute(obj, 'container-live') + if liveAttr in ('assertive', 'polite') or containerLive in ('assertive', 'polite'): + return True + + return False + def _ignore(self, event): """Returns True if this event should be ignored.""" @@ -241,9 +279,14 @@ class EventManager: debug.printMessage(debug.LEVEL_INFO, msg, True) return True if script.app != app: - msg = 'EVENT MANAGER: Ignoring because event is not from active app' - debug.printMessage(debug.LEVEL_INFO, msg, True) - return True + # Allow Steam notifications from inactive apps. + if self._isSteamApp(app) and self._isSteamNotificationEvent(event): + msg = 'EVENT MANAGER: Allowing Steam notification from inactive app' + debug.printMessage(debug.LEVEL_INFO, msg, True) + else: + msg = 'EVENT MANAGER: Ignoring because event is not from active app' + debug.printMessage(debug.LEVEL_INFO, msg, True) + return True if event.type.startswith('object:text-changed') \ and self.EMBEDDED_OBJECT_CHARACTER in event.any_data \ diff --git a/src/cthulhu/script_manager.py b/src/cthulhu/script_manager.py index f21abd3..47e9602 100644 --- a/src/cthulhu/script_manager.py +++ b/src/cthulhu/script_manager.py @@ -113,6 +113,12 @@ class ScriptManager: debug.printMessage(debug.LEVEL_INFO, msg, True) return None + pid = AXObject.get_process_id(app) + if pid != -1: + cmdline = debug.getCmdline(pid) + if "steamwebhelper" in cmdline: + return "steamwebhelper" + altNames = list(self._appNames.keys()) if name.endswith(".py") or name.endswith(".bin"): name = name.split('.')[0] diff --git a/src/cthulhu/scripts/apps/__init__.py b/src/cthulhu/scripts/apps/__init__.py index 1d73da5..8333088 100644 --- a/src/cthulhu/scripts/apps/__init__.py +++ b/src/cthulhu/scripts/apps/__init__.py @@ -40,5 +40,6 @@ __all__ = ['Banshee', 'soffice', 'SeaMonkey', 'smuxi-frontend-gnome', + 'steamwebhelper', 'Thunderbird', 'xfwm4'] diff --git a/src/cthulhu/scripts/apps/meson.build b/src/cthulhu/scripts/apps/meson.build index a9ecfe6..91991cd 100644 --- a/src/cthulhu/scripts/apps/meson.build +++ b/src/cthulhu/scripts/apps/meson.build @@ -25,4 +25,5 @@ subdir('notify-osd') subdir('pidgin') subdir('soffice') subdir('smuxi-frontend-gnome') +subdir('steamwebhelper') subdir('xfwm4') \ No newline at end of file diff --git a/src/cthulhu/scripts/apps/steamwebhelper/__init__.py b/src/cthulhu/scripts/apps/steamwebhelper/__init__.py new file mode 100644 index 0000000..585c964 --- /dev/null +++ b/src/cthulhu/scripts/apps/steamwebhelper/__init__.py @@ -0,0 +1 @@ +from .script import Script diff --git a/src/cthulhu/scripts/apps/steamwebhelper/meson.build b/src/cthulhu/scripts/apps/steamwebhelper/meson.build new file mode 100644 index 0000000..2ae9c2d --- /dev/null +++ b/src/cthulhu/scripts/apps/steamwebhelper/meson.build @@ -0,0 +1,9 @@ +steamwebhelper_python_sources = files([ + '__init__.py', + 'script.py', +]) + +python3.install_sources( + steamwebhelper_python_sources, + subdir: 'cthulhu/scripts/apps/steamwebhelper' +) diff --git a/src/cthulhu/scripts/apps/steamwebhelper/script.py b/src/cthulhu/scripts/apps/steamwebhelper/script.py new file mode 100644 index 0000000..b5314d8 --- /dev/null +++ b/src/cthulhu/scripts/apps/steamwebhelper/script.py @@ -0,0 +1,305 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2024 Stormux +# +# 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. +# +# Cthulhu project: https://git.stormux.org/storm/cthulhu + +"""Custom script for Steam Web Helper (Steam client UI).""" + +__id__ = "$Id$" +__version__ = "$Revision$" +__date__ = "$Date$" +__copyright__ = "Copyright (c) 2024 Stormux" +__license__ = "LGPL" + +import time + +from cthulhu import debug +from cthulhu import settings +from cthulhu import cthulhu_state +from cthulhu import settings_manager +from cthulhu import structural_navigation +from cthulhu.ax_object import AXObject +from cthulhu.ax_utilities import AXUtilities +from cthulhu.ax_utilities_relation import AXUtilitiesRelation +from cthulhu.ax_utilities_role import AXUtilitiesRole +from cthulhu.scripts.toolkits import Chromium + +settingsManager = settings_manager.getManager() + + +class SteamStructuralNavigation(structural_navigation.StructuralNavigation): + def _headingGetter(self, document, arg=None): + def isUsefulHeading(candidate): + if not self._script.utilities.inDocumentContent(candidate): + return False + + text = self._script.utilities.displayedText(candidate) or AXObject.get_name(candidate) + if not text: + return False + + return not text.strip().isdigit() + + if arg is None: + return super()._headingGetter(document, arg) + + headings = super()._headingGetter(document, None) + headings = [heading for heading in headings if isUsefulHeading(heading)] + return [heading for heading in headings if self._script.utilities.headingLevel(heading) == arg] + + +class Script(Chromium.Script): + """Script for Steam Web Helper with background notification support.""" + + def __init__(self, app): + """Creates a new script for the Steam Web Helper application. + + Arguments: + - app: the application to create a script for. + """ + super().__init__(app) + # CRITICAL: Enable background event processing for notifications + # This allows Steam notifications to be spoken even when Steam + # is not the focused application. + self.presentIfInactive = True + self._lastSteamNotification = ("", 0.0) + + def onShowingChanged(self, event): + """Callback for object:state-changed:showing accessibility events.""" + + # Check if this is a Steam notification/alert becoming visible + if event.detail1 and self._isSteamNotification(event.source): + self._presentSteamNotification(event.source) + return + + # Fall through to Chromium/web handling + super().onShowingChanged(event) + + def onChildrenAdded(self, event): + """Callback for object:children-changed:add accessibility events.""" + + # Check if a notification element was added + if self._isSteamNotification(event.any_data): + self._presentSteamNotification(event.any_data) + return + + super().onChildrenAdded(event) + + def getStructuralNavigation(self): + types = self.getEnabledStructuralNavigationTypes() + enable = settingsManager.getSetting('structuralNavigationEnabled') + return SteamStructuralNavigation(self, types, enable) + + def onFocusedChanged(self, event): + """Callback for object:state-changed:focused accessibility events.""" + + if event.detail1 and AXUtilities.is_document_web(event.source): + prevDocument = self.utilities.getDocumentForObject(cthulhu_state.locusOfFocus) + if prevDocument and AXUtilities.is_embedded(prevDocument) \ + and self.utilities.inDocumentContent(cthulhu_state.locusOfFocus): + msg = "STEAM: Ignoring focus event for top-level document while in embedded content" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return True + + if event.detail1 \ + and AXUtilitiesRole.is_button(event.source) \ + and cthulhu_state.locusOfFocus in AXUtilitiesRelation.get_is_labelled_by(event.source): + msg = "STEAM: Ignoring focus event for labelled button" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return True + + return super().onFocusedChanged(event) + + def onTextInserted(self, event): + """Callback for object:text-changed:insert accessibility events.""" + + if self._presentSteamLiveRegionText(event): + return + + super().onTextInserted(event) + + def _isSteamNotification(self, obj): + """Detect if object is a Steam notification. + + Arguments: + - obj: the accessible object to check + + Returns True if obj appears to be a notification element. + """ + if obj is None: + return False + + # Check for NOTIFICATION or ALERT roles + if AXUtilities.is_notification(obj) or AXUtilities.is_alert(obj): + msg = "STEAM: Found notification/alert role" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return True + + # Check for live region with assertive/polite politeness + # (Steam may use ARIA live regions for toasts) + liveAttr = AXObject.get_attribute(obj, 'live') + containerLive = AXObject.get_attribute(obj, 'container-live') + if liveAttr in ['assertive', 'polite'] or containerLive in ['assertive', 'polite']: + politeness = liveAttr or containerLive + msg = f"STEAM: Found live region with politeness: {politeness}" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return True + + def isNotificationCandidate(candidate): + if AXUtilities.is_notification(candidate) or AXUtilities.is_alert(candidate): + return True + candidateLive = AXObject.get_attribute(candidate, 'live') + candidateContainerLive = AXObject.get_attribute(candidate, 'container-live') + return candidateLive in ['assertive', 'polite'] or candidateContainerLive in ['assertive', 'polite'] + + ancestor = AXObject.find_ancestor(obj, isNotificationCandidate) + if ancestor: + msg = "STEAM: Found notification ancestor" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return True + + return False + + def _presentSteamNotification(self, obj): + """Speak and save the notification. + + Arguments: + - obj: the notification accessible object + """ + text = self._getNotificationText(obj) + if not text: + msg = "STEAM: Notification had no text content" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return + + self._presentSteamNotificationText(text, obj) + + def _presentSteamLiveRegionText(self, event): + if not isinstance(event.any_data, str): + return False + + text = event.any_data.strip() + if not text: + return False + + if not self._isSteamLiveRegion(event.source): + return False + + msg = f"STEAM: Live region text inserted: {text}" + debug.printMessage(debug.LEVEL_INFO, msg, True) + self._presentSteamNotificationText(text, event.source) + return True + + def _isSteamLiveRegion(self, obj): + if not obj: + return False + + liveAttr = AXObject.get_attribute(obj, 'live') + containerLive = AXObject.get_attribute(obj, 'container-live') + if liveAttr in ['assertive', 'polite'] or containerLive in ['assertive', 'polite']: + return True + + def isLiveRegion(candidate): + candidateLive = AXObject.get_attribute(candidate, 'live') + candidateContainerLive = AXObject.get_attribute(candidate, 'container-live') + return candidateLive in ['assertive', 'polite'] \ + or candidateContainerLive in ['assertive', 'polite'] + + return AXObject.find_ancestor(obj, isLiveRegion) is not None + + def _presentSteamNotificationText(self, text, obj): + if self._isDuplicateSteamNotification(text): + msg = "STEAM: Suppressing duplicate notification" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return + + notificationMsg = f"Steam: {text}" + + # Speak the notification + voice = self.speechGenerator.voice(obj=obj, string=notificationMsg) + self.speakMessage(notificationMsg, voice=voice) + + # Display on braille + self.displayBrailleMessage(notificationMsg, flashTime=settings.brailleFlashTime) + + # Save to notification history + self.notificationPresenter.save_notification(notificationMsg) + + msg = f"STEAM: Presented notification: {text}" + debug.printMessage(debug.LEVEL_INFO, msg, True) + + def _isDuplicateSteamNotification(self, text): + lastText, lastTime = self._lastSteamNotification + now = time.monotonic() + if text == lastText and (now - lastTime) < 1.0: + return True + + self._lastSteamNotification = (text, now) + return False + + def _getNotificationText(self, obj): + """Extract text from notification element. + + Arguments: + - obj: the notification accessible object + + Returns the text content of the notification, or empty string if none found. + """ + def getText(candidate): + name = AXObject.get_name(candidate) + if name: + return name + return self.utilities.displayedText(candidate) + + # Try getting name directly + text = getText(obj) + if text: + return text + + # Try getting all text from labels/text elements + texts = [] + for label in AXUtilities.find_all_labels(obj): + text = getText(label) + if text: + texts.append(text) + + if texts: + return ' '.join(texts) + + def isNamedButton(candidate): + if not AXUtilitiesRole.is_button(candidate): + return False + return bool(getText(candidate)) + + descendantButton = AXObject.find_descendant(obj, isNamedButton) + if descendantButton: + return getText(descendantButton) + + ancestorButton = AXObject.find_ancestor(obj, isNamedButton) + if ancestorButton: + return getText(ancestorButton) + + def hasReadableText(candidate): + return bool(getText(candidate)) + + descendantText = AXObject.find_descendant(obj, hasReadableText) + if descendantText: + return getText(descendantText) + + # Fall back to displayed text of the object itself + return ""