storm/cthulhu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Steam improvements. I can't say it's perfect by any means, but much much improved.

Browse Source
This commit is contained in:
Storm Dragon
2026-01-02 00:37:52 -05:00
parent afdd812f2f
commit 921ffc4145
7 changed files with 369 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
src/cthulhu/event_manager.py
View File

@@ -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 \

6
src/cthulhu/script_manager.py
View File

@@ -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]

1
src/cthulhu/scripts/apps/__init__.py
View File

@@ -40,5 +40,6 @@ __all__ = ['Banshee',
'soffice',
'SeaMonkey',
'smuxi-frontend-gnome',
'steamwebhelper',
'Thunderbird',
'xfwm4']

1
src/cthulhu/scripts/apps/meson.build
View File

@@ -25,4 +25,5 @@ subdir('notify-osd')
subdir('pidgin')
subdir('soffice')
subdir('smuxi-frontend-gnome')
subdir('steamwebhelper')
subdir('xfwm4')

1
src/cthulhu/scripts/apps/steamwebhelper/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .script import Script

9
src/cthulhu/scripts/apps/steamwebhelper/meson.build Normal file
View File

@@ -0,0 +1,9 @@
steamwebhelper_python_sources = files([
'__init__.py',
'script.py',
])
python3.install_sources(
steamwebhelper_python_sources,
subdir: 'cthulhu/scripts/apps/steamwebhelper'
)

305
src/cthulhu/scripts/apps/steamwebhelper/script.py Normal file
View File

@@ -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 ""