4 Commits

13 changed files with 2794 additions and 15 deletions

View File

@ -1,6 +1,6 @@
Welcome to Cthulhu
We are excited to have you here and welcome your contributions to the Cthulhu screen reader project! This project is a fork of Cthulhu, with a focus on creating an open and collaborative community where contributions are encouraged.
We are excited to have you here and welcome your contributions to the Cthulhu screen reader project! This project is a fork of Orca, with a focus on creating an open and collaborative community where contributions are encouraged.
How to Contribute

View File

@ -1,6 +1,10 @@
ACLOCAL_AMFLAGS = -I m4 ${ACLOCAL_FLAGS}
if BUILD_HELP
SUBDIRS = docs icons po src help
else
SUBDIRS = docs icons po src
endif
DISTCHECK_CONFIGURE_FLAGS = \
--disable-scrollkeeper

View File

@ -32,10 +32,14 @@ autoreconf --verbose --force --install -Wno-portability || {
exit 1
}
which yelp-build > /dev/null || {
echo "Try installing the 'yelp-tools' package."
exit 1
}
# Only check for yelp-build if help documentation will be built
# Skip check if SKIP_YELP environment variable is set
if [ "${SKIP_YELP}" != "1" ]; then
which yelp-build > /dev/null || {
echo "Try installing the 'yelp-tools' package, or set SKIP_YELP=1 to skip documentation."
exit 1
}
fi
cd "$olddir"

View File

@ -24,8 +24,16 @@ GETTEXT_PACKAGE=AC_PACKAGE_TARNAME
AC_SUBST(GETTEXT_PACKAGE)
AC_DEFINE_UNQUOTED(GETTEXT_PACKAGE,"$GETTEXT_PACKAGE", [gettext package])
# User Documentation
YELP_HELP_INIT
# User Documentation (optional)
AC_ARG_ENABLE([help],
AS_HELP_STRING([--disable-help], [Disable building help documentation]))
AS_IF([test "x$enable_help" != "xno"], [
YELP_HELP_INIT
BUILD_HELP=yes
], [
BUILD_HELP=no
])
AM_CONDITIONAL(BUILD_HELP, test "x$BUILD_HELP" = "xyes")
PKG_CHECK_MODULES([PYGOBJECT], [pygobject-3.0 >= pygobject_required_version])
PKG_CHECK_MODULES([ATSPI2], [atspi-2 >= atspi_required_version])
@ -129,6 +137,7 @@ src/cthulhu/plugins/ByeCthulhu/Makefile
src/cthulhu/plugins/HelloCthulhu/Makefile
src/cthulhu/plugins/Clipboard/Makefile
src/cthulhu/plugins/DisplayVersion/Makefile
src/cthulhu/plugins/IndentationAudio/Makefile
src/cthulhu/plugins/hello_world/Makefile
src/cthulhu/plugins/self_voice/Makefile
src/cthulhu/plugins/SimplePluginSystem/Makefile

View File

@ -1,7 +1,7 @@
# Maintainer: Storm Dragon <storm_dragon@stormux.org>
pkgname=cthulhu
pkgver=0.4
pkgver=2025.07.01
pkgrel=1
pkgdesc="Screen reader for individuals who are blind or visually impaired forked from Orca"
url="https://git.stormux.org/storm/cthulhu"
@ -25,6 +25,7 @@ depends=(
python-atspi
python-cairo
python-gobject
python-pluggy
python-setproctitle
socat
speech-dispatcher
@ -33,15 +34,14 @@ depends=(
)
makedepends=(
git
itstool
yelp-tools
)
source=("git+https://git.stormux.org/storm/cthulhu.git")
b2sums=('SKIP')
prepare() {
cd cthulhu
NOCONFIGURE=1 ./autogen.sh
git checkout testing
NOCONFIGURE=1 SKIP_YELP=1 ./autogen.sh
}
pkgver() {
@ -51,7 +51,7 @@ pkgver() {
build() {
cd cthulhu
./configure --prefix=/usr --sysconfdir=/etc --localstatedir=/var
./configure --prefix=/usr --sysconfdir=/etc --localstatedir=/var --disable-help
make
}

View File

@ -23,5 +23,5 @@
# Fork of Orca Screen Reader (GNOME)
# Original source: https://gitlab.gnome.org/GNOME/orca
version = "2025.06.05"
codeName = "plugins"
version = "2025.07.01"
codeName = "testing"

659
src/cthulhu/dbus_service.py Normal file
View File

@ -0,0 +1,659 @@
# Cthulhu
#
# Copyright 2025 Stormux <storm_dragon@stormux.org>
# Copyright 2025 Valve Corporation
# Author: Joanmarie Diggs <jdiggs@igalia.com>
#
# 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.
"""Provides a D-Bus interface for remotely controlling Cthulhu."""
__id__ = "$Id$"
__version__ = "$Revision$"
__date__ = "$Date$"
__copyright__ = "Copyright (c) 2025 Stormux <storm_dragon@stormux.org>"
__license__ = "LGPL"
import enum
from typing import Callable
try:
from dasbus.connection import SessionMessageBus
from dasbus.error import DBusError
from dasbus.loop import EventLoop
from dasbus.server.interface import dbus_interface
from dasbus.server.publishable import Publishable
_dasbus_available = True
except ImportError:
_dasbus_available = False
from gi.repository import GLib
from . import debug
from . import cthulhu_platform # pylint: disable=no-name-in-module
from . import script_manager
from . import cthulhu_state
# Lazy import to avoid circular dependency
def _get_input_event():
from . import input_event
return input_event
class HandlerType(enum.Enum):
"""Enumeration of handler types for D-Bus methods."""
COMMAND = enum.auto()
GETTER = enum.auto()
SETTER = enum.auto()
def command(func):
"""Decorator to mark a method as a D-Bus command using its docstring.
Usage:
@command
def toggle_speech(self, script=None, event=None):
'''Toggles speech on and off.'''
# method implementation
"""
description = func.__doc__ or f"D-Bus command: {func.__name__}"
func.dbus_command_description = description
return func
def getter(func):
"""Decorator to mark a method as a D-Bus getter using its docstring.
Usage:
@getter
def get_rate(self):
'''Returns the current speech rate.'''
# method implementation
"""
description = func.__doc__ or f"D-Bus getter: {func.__name__}"
func.dbus_getter_description = description
return func
def setter(func):
"""Decorator to mark a method as a D-Bus setter using its docstring.
Usage:
@setter
def set_rate(self, value):
'''Sets the current speech rate.'''
# method implementation
"""
description = func.__doc__ or f"D-Bus setter: {func.__name__}"
func.dbus_setter_description = description
return func
class _HandlerInfo:
"""Stores processed information about a function exposed via D-Bus."""
def __init__(
self,
python_function_name: str,
description: str,
action: Callable[..., bool],
handler_type: 'HandlerType' = HandlerType.COMMAND
):
self.python_function_name: str = python_function_name
self.description: str = description
self.action: Callable[..., bool] = action
self.handler_type: HandlerType = handler_type
if _dasbus_available:
@dbus_interface("org.stormux.Cthulhu.Module")
class CthulhuModuleDBusInterface(Publishable):
"""A D-Bus interface representing a specific Cthulhu module (e.g., a manager)."""
def __init__(self,
module_name: str,
handlers_info: list[_HandlerInfo]):
super().__init__()
self._module_name = module_name
self._commands: dict[str, _HandlerInfo] = {}
self._getters: dict[str, _HandlerInfo] = {}
self._setters: dict[str, _HandlerInfo] = {}
for info in handlers_info:
handler_type = getattr(info, "handler_type", HandlerType.COMMAND)
normalized_name = self._normalize_handler_name(info.python_function_name)
if handler_type == HandlerType.GETTER:
self._getters[normalized_name] = info
elif handler_type == HandlerType.SETTER:
self._setters[normalized_name] = info
else:
self._commands[normalized_name] = info
msg = (
f"DBUS SERVICE: CthulhuModuleDBusInterface for {module_name} initialized "
f"with {len(self._commands)} commands, {len(self._getters)} getters, "
f"{len(self._setters)} setters."
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
def ExecuteRuntimeGetter(self, getter_name: str) -> GLib.Variant: # pylint: disable=invalid-name
"""Executes the named getter returning the value as a GLib.Variant for D-Bus marshalling."""
handler_info = self._getters.get(getter_name)
if not handler_info:
msg = f"DBUS SERVICE: Unknown getter '{getter_name}' for '{self._module_name}'."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return GLib.Variant("v", GLib.Variant("s", ""))
result = handler_info.action()
msg = f"DBUS SERVICE: Getter '{getter_name}' returned: {result}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return self._to_variant(result)
def ExecuteRuntimeSetter(self, setter_name: str, value: GLib.Variant) -> bool: # pylint: disable=invalid-name
"""Executes the named setter, returning True if succeeded."""
handler_info = self._setters.get(setter_name)
if handler_info is None:
msg = f"DBUS SERVICE: Unknown setter '{setter_name}' for '{self._module_name}'."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return False
unpacked = value.unpack()
result = handler_info.action(unpacked)
msg = f"DBUS SERVICE: Setter '{setter_name}' with value '{unpacked}' returned: {result}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return result
def ListCommands(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name
"""Returns a list of (command_name, description) for this module (commands only)."""
command_list = []
for camel_case_name, info in self._commands.items():
command_list.append((camel_case_name, info.description))
return command_list
def ListRuntimeGetters(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name
"""Returns a list of (getter_name, description) for this module."""
getter_list = []
for camel_case_name, info in self._getters.items():
getter_list.append((camel_case_name, info.description))
return getter_list
def ListRuntimeSetters(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name
"""Returns a list of (setter_name, description) for this module."""
setter_list = []
for camel_case_name, info in self._setters.items():
setter_list.append((camel_case_name, info.description))
return setter_list
def ExecuteCommand(self, command_name: str, notify_user: bool) -> bool: # pylint: disable=invalid-name
"""Executes the named command and returns True if the command succeeded."""
if command_name not in self._commands:
msg = f"DBUS SERVICE: Unknown command '{command_name}' for '{self._module_name}'."
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
handler_info = self._commands[command_name]
result = handler_info.action(notify_user)
msg = (
f"DBUS SERVICE: '{command_name}' in '{self._module_name}' executed. "
f"Result: {result}, notify_user: {notify_user}"
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
return result
def for_publication(self):
"""Returns the D-Bus interface XML for publication."""
return self.__dbus_xml__ # pylint: disable=no-member
@staticmethod
def _normalize_handler_name(function_name: str) -> str:
"""Normalizes a Python function name for D-Bus exposure (getter/setter/command)."""
if function_name.startswith("get_") or function_name.startswith("set_"):
function_name = function_name[4:]
return "".join(word.capitalize() for word in function_name.split("_"))
@staticmethod
def _to_variant(result):
"""Converts a Python value to a correctly-typed GLib.Variant for D-Bus marshalling."""
if isinstance(result, bool):
return GLib.Variant("b", result)
elif isinstance(result, int):
return GLib.Variant("i", result)
elif isinstance(result, float):
return GLib.Variant("d", result)
elif isinstance(result, str):
return GLib.Variant("s", result)
elif isinstance(result, dict):
return GLib.Variant(
"a{sv}", {str(k): GLib.Variant("v", v) for k, v in result.items()})
elif isinstance(result, list) or isinstance(result, tuple):
if all(isinstance(x, str) for x in result):
return GLib.Variant("as", list(result))
elif all(isinstance(x, int) for x in result):
return GLib.Variant("ax", list(result))
elif all(isinstance(x, bool) for x in result):
return GLib.Variant("ab", list(result))
else:
return GLib.Variant("av", [GLib.Variant("v", x) for x in result])
elif result is None:
return GLib.Variant("v", GLib.Variant("s", ""))
else:
return GLib.Variant("s", str(result))
@dbus_interface("org.stormux.Cthulhu.Service")
class CthulhuDBusServiceInterface(Publishable):
"""Internal D-Bus service object that handles D-Bus specifics."""
def __init__(self) -> None:
super().__init__()
self._registered_modules: set[str] = set()
msg = "DBUS SERVICE: CthulhuDBusServiceInterface initialized."
debug.printMessage(debug.LEVEL_INFO, msg, True)
def for_publication(self):
"""Returns the D-Bus interface XML for publication."""
return self.__dbus_xml__ # pylint: disable=no-member
def add_module_interface(
self,
module_name: str,
handlers_info: list[_HandlerInfo],
bus: SessionMessageBus,
object_path_base: str
) -> None:
"""Creates and prepares a D-Bus interface for a Cthulhu module."""
object_path = f"{object_path_base}/{module_name}"
if module_name in self._registered_modules:
msg = f"DBUS SERVICE: Interface {module_name} already registered. Replacing."
debug.printMessage(debug.LEVEL_INFO, msg, True)
try:
bus.unpublish_object(object_path)
except DBusError as e:
msg = f"DBUS SERVICE: Error unpublishing old interface for {module_name}: {e}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._registered_modules.discard(module_name)
try:
module_iface = CthulhuModuleDBusInterface(module_name, handlers_info)
bus.publish_object(object_path, module_iface)
self._registered_modules.add(module_name)
msg = f"DBUS SERVICE: Successfully published {module_name} at {object_path}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
except DBusError as e:
msg = (
f"DBUS SERVICE: Failed to create or publish D-Bus interface for "
f"module {module_name} at {object_path}: {e}"
)
debug.printMessage(debug.LEVEL_SEVERE, msg, True)
def remove_module_interface(
self,
module_name: str,
bus: SessionMessageBus,
object_path_base: str
) -> bool:
"""Removes and unpublishes a D-Bus interface for a Cthulhu module."""
if module_name not in self._registered_modules:
msg = f"DBUS SERVICE: Module {module_name} is not registered."
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
object_path = f"{object_path_base}/{module_name}"
try:
bus.unpublish_object(object_path)
self._registered_modules.discard(module_name)
msg = f"DBUS SERVICE: Successfully removed {module_name} from {object_path}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
except DBusError as e:
msg = f"DBUS SERVICE: Error removing interface for {module_name}: {e}"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
def ListModules(self) -> list[str]: # pylint: disable=invalid-name
"""Returns a list of registered module names."""
return list(self._registered_modules)
def ListCommands(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name
"""Returns available commands on the main service interface."""
commands = []
for attr_name in dir(self):
if not attr_name.startswith('_') and attr_name[0].isupper():
attr = getattr(self, attr_name)
if callable(attr) and hasattr(attr, '__doc__'):
description = (attr.__doc__.strip() if attr.__doc__
else f"Service command: {attr_name}")
commands.append((attr_name, description))
return sorted(commands)
def PresentMessage(self, message: str) -> bool: # pylint: disable=invalid-name
"""Presents message to the user."""
msg = f"DBUS SERVICE: PresentMessage called with: '{message}'"
debug.printMessage(debug.LEVEL_INFO, msg, True)
manager = script_manager.getManager()
script = cthulhu_state.activeScript or script_manager.getManager().getDefaultScript()
if script is None:
msg = "DBUS SERVICE: No script available"
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
script.presentMessage(message)
return True
def GetVersion(self) -> str: # pylint: disable=invalid-name
"""Returns Cthulhu's version and revision if available."""
result = cthulhu_platform.version
if cthulhu_platform.revision:
result += f" (rev {cthulhu_platform.revision})"
msg = f"DBUS SERVICE: GetVersion called, returning: {result}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return result
def shutdown_service(self, bus: SessionMessageBus, object_path_base: str) -> None:
"""Releases D-Bus resources held by this service and its modules."""
msg = "DBUS SERVICE: Releasing D-Bus resources for service."
debug.printMessage(debug.LEVEL_INFO, msg, True)
for module_name in list(self._registered_modules):
module_object_path = f"{object_path_base}/{module_name}"
msg = (
f"DBUS SERVICE: Shutting down and unpublishing module {module_name} "
f"from main service."
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
try:
bus.unpublish_object(module_object_path)
except DBusError as e:
msg = f"DBUS SERVICE: Error unpublishing interface for {module_name}: {e}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._registered_modules.clear()
else:
# Fallback classes when dasbus is not available
class CthulhuModuleDBusInterface:
"""Fallback class when dasbus is not available."""
def __init__(self, *args, **kwargs):
pass
class CthulhuDBusServiceInterface:
"""Fallback class when dasbus is not available."""
def __init__(self, *args, **kwargs):
pass
class CthulhuRemoteController:
"""Manages Cthulhu's D-Bus service for remote control."""
SERVICE_NAME = "org.stormux.Cthulhu.Service"
OBJECT_PATH = "/org/stormux/Cthulhu/Service"
def __init__(self) -> None:
self._dbus_service_interface: CthulhuDBusServiceInterface | None = None
self._is_running: bool = False
self._bus: SessionMessageBus | None = None
self._event_loop: EventLoop | None = None
self._pending_registrations: dict[str, object] = {}
self._dasbus_available = _dasbus_available
def start(self) -> bool:
"""Starts the D-Bus service."""
if not self._dasbus_available:
msg = "REMOTE CONTROLLER: dasbus library not available, D-Bus service disabled."
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
if self._is_running:
msg = "REMOTE CONTROLLER: Start called but service is already running."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
msg = "REMOTE CONTROLLER: Attempting to start D-Bus service."
debug.printMessage(debug.LEVEL_INFO, msg, True)
try:
self._bus = SessionMessageBus()
msg = (
f"REMOTE CONTROLLER: SessionMessageBus acquired: "
f"{self._bus.connection.get_unique_name()}"
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
except DBusError as e:
self._bus = None
msg = f"REMOTE CONTROLLER: Failed to acquire D-Bus session bus: {e}"
debug.printMessage(debug.LEVEL_SEVERE, msg, True)
return False
self._dbus_service_interface = CthulhuDBusServiceInterface()
try:
self._bus.publish_object(self.OBJECT_PATH, self._dbus_service_interface)
self._bus.register_service(self.SERVICE_NAME)
except DBusError as e:
msg = f"REMOTE CONTROLLER: Failed to publish service or request name: {e}"
debug.printMessage(debug.LEVEL_SEVERE, msg, True)
if self._dbus_service_interface and self._bus:
try:
self._bus.unpublish_object(self.OBJECT_PATH)
except DBusError:
pass
self._dbus_service_interface = None
self._bus = None
return False
self._is_running = True
msg = (
f"REMOTE CONTROLLER: Service started name={self.SERVICE_NAME} "
f"path={self.OBJECT_PATH}."
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._process_pending_registrations()
return True
def _process_pending_registrations(self) -> None:
"""Processes any module registrations that were queued before the service was ready."""
if not self._pending_registrations:
return
msg = (
f"REMOTE CONTROLLER: Processing {len(self._pending_registrations)} "
f"pending module registrations."
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
for module_name, module_instance in self._pending_registrations.items():
msg = f"REMOTE CONTROLLER: Processing pending registration for {module_name}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._register_decorated_commands_internal(module_name, module_instance)
self._pending_registrations.clear()
def register_decorated_module(self, module_name: str, module_instance) -> None:
"""Registers a module's decorated D-Bus commands."""
if not self._dasbus_available:
return
if not self._is_running or not self._dbus_service_interface or not self._bus:
msg = (
f"REMOTE CONTROLLER: Service not ready; queuing decorated registration "
f"for {module_name}."
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._pending_registrations[module_name] = module_instance
return
self._register_decorated_commands_internal(module_name, module_instance)
def _register_decorated_commands_internal(self, module_name: str, module_instance) -> None:
"""Internal method that registers decorated commands from a module instance."""
if not self._is_running or not self._dbus_service_interface or not self._bus:
msg = (
f"REMOTE CONTROLLER: Internal error - _register_decorated_commands_internal "
f"called for {module_name} but service is not ready."
)
debug.printMessage(debug.LEVEL_SEVERE, msg, True)
return
handlers_info = []
for attr_name in dir(module_instance):
attr = getattr(module_instance, attr_name)
# Command
if callable(attr) and hasattr(attr, "dbus_command_description"):
description = attr.dbus_command_description
def _create_wrapper(method=attr):
def _wrapper(notify_user):
event = _get_input_event().RemoteControllerEvent()
script = cthulhu_state.activeScript
return method(script=script, event=event, notify_user=notify_user)
return _wrapper
handler_info = _HandlerInfo(
python_function_name=attr_name,
description=description,
action=_create_wrapper(),
handler_type=HandlerType.COMMAND
)
handlers_info.append(handler_info)
msg = f"REMOTE CONTROLLER: Found decorated command '{attr_name}': {description}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
# Getter
elif callable(attr) and hasattr(attr, "dbus_getter_description"):
description = attr.dbus_getter_description
def _create_getter_wrapper(method=attr):
def _wrapper(_notify_user=None):
return method()
return _wrapper
handler_info = _HandlerInfo(
python_function_name=attr_name,
description=description,
action=_create_getter_wrapper(),
handler_type=HandlerType.GETTER
)
handlers_info.append(handler_info)
msg = f"REMOTE CONTROLLER: Found decorated getter '{attr_name}': {description}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
# Setter
elif callable(attr) and hasattr(attr, "dbus_setter_description"):
description = attr.dbus_setter_description
def _create_setter_wrapper(method=attr):
def _wrapper(value):
return method(value)
return _wrapper
handler_info = _HandlerInfo(
python_function_name=attr_name,
description=description,
action=_create_setter_wrapper(),
handler_type=HandlerType.SETTER
)
handlers_info.append(handler_info)
msg = f"REMOTE CONTROLLER: Found decorated setter '{attr_name}': {description}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
if not handlers_info:
return
self._dbus_service_interface.add_module_interface(
module_name, handlers_info, self._bus, self.OBJECT_PATH)
msg = (
f"REMOTE CONTROLLER: Successfully registered {len(handlers_info)} "
f"decorated commands/getters/setters for module {module_name}."
)
debug.printMessage(debug.LEVEL_INFO, msg, True)
def deregister_module_commands(self, module_name: str) -> bool:
"""Deregisters D-Bus commands for a Cthulhu module."""
if not self._dasbus_available:
return False
if module_name in self._pending_registrations:
msg = f"REMOTE CONTROLLER: Removing pending registration for {module_name}."
debug.printMessage(debug.LEVEL_INFO, msg, True)
del self._pending_registrations[module_name]
return True
if not self._is_running or not self._dbus_service_interface or not self._bus:
msg = (
f"REMOTE CONTROLLER: Cannot deregister commands for {module_name}; "
"service not running or bus not available."
)
debug.printMessage(debug.LEVEL_WARNING, msg, True)
return False
return self._dbus_service_interface.remove_module_interface(
module_name, self._bus, self.OBJECT_PATH)
def shutdown(self) -> None:
"""Shuts down the D-Bus service."""
if not self._dasbus_available:
return
if not self._is_running:
msg = "REMOTE CONTROLLER: Shutdown called but service is not running."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return
msg = "REMOTE CONTROLLER: Attempting to shut down D-Bus service."
debug.printMessage(debug.LEVEL_INFO, msg, True)
if self._dbus_service_interface and self._bus:
self._dbus_service_interface.shutdown_service(self._bus, self.OBJECT_PATH)
try:
self._bus.unpublish_object(self.OBJECT_PATH)
except DBusError as e:
msg = f"REMOTE CONTROLLER: Error unpublishing main service object: {e}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._dbus_service_interface = None
if self._bus:
try:
self._bus.unregister_service(self.SERVICE_NAME)
except DBusError as e:
msg = f"REMOTE CONTROLLER: Error releasing bus name: {e}"
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._bus.disconnect()
self._bus = None
self._is_running = False
msg = "REMOTE CONTROLLER: D-Bus service shut down."
debug.printMessage(debug.LEVEL_INFO, msg, True)
self._pending_registrations.clear()
def is_running(self) -> bool:
"""Checks if the D-Bus service is currently running."""
return self._is_running
_remote_controller: CthulhuRemoteController = CthulhuRemoteController()
def get_remote_controller() -> CthulhuRemoteController:
"""Returns the CthulhuRemoteController singleton."""
return _remote_controller

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,10 @@
indentationaudio_PYTHON = \
__init__.py \
plugin.py
indentationaudiodir = $(pkgdatadir)/cthulhu/plugins/IndentationAudio
indentationaudio_DATA = \
plugin.info
EXTRA_DIST = $(indentationaudio_DATA)

View File

@ -0,0 +1,14 @@
#!/usr/bin/env python3
#
# Copyright (c) 2025 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.
"""IndentationAudio plugin package."""
from .plugin import IndentationAudio
__all__ = ['IndentationAudio']

View File

@ -0,0 +1,9 @@
[Core]
Name = IndentationAudio
Module = IndentationAudio
[Documentation]
Description = Provides audio feedback for indentation level changes when navigating code or text
Author = Stormux
Version = 1.0.0
Website = https://git.stormux.org/storm/cthulhu

View File

@ -0,0 +1,334 @@
#!/usr/bin/env python3
#
# Copyright (c) 2025 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.
"""IndentationAudio plugin for Cthulhu - Provides audio feedback for indentation level changes."""
import logging
import re
from gi.repository import GLib
from cthulhu.plugin import Plugin, cthulhu_hookimpl
from cthulhu import cmdnames
from cthulhu import debug
from cthulhu import input_event
from cthulhu import keybindings
from cthulhu import messages
from cthulhu import script_utilities
from cthulhu import settings_manager
from cthulhu import sound
from cthulhu.sound_generator import Tone
logger = logging.getLogger(__name__)
_settingsManager = settings_manager.getManager()
class IndentationAudio(Plugin):
"""Plugin that provides audio cues for indentation level changes."""
def __init__(self, *args, **kwargs):
"""Initialize the IndentationAudio plugin."""
super().__init__(*args, **kwargs)
self._enabled = True # Start enabled by default
self._last_indentation_level = {} # Track per-object indentation
self._event_listener_id = None
self._kb_binding = None
self._player = None
# Audio settings
self._base_frequency = 200 # Base frequency in Hz
self._frequency_step = 80 # Hz per indentation level
self._tone_duration = 0.15 # Seconds
self._max_frequency = 1200 # Cap frequency to avoid harsh sounds
logger.info("IndentationAudio plugin initialized")
@cthulhu_hookimpl
def activate(self, plugin=None):
"""Activate the IndentationAudio plugin."""
if plugin is not None and plugin is not self:
return
try:
logger.info("=== IndentationAudio plugin activation starting ===")
# Initialize sound player
self._player = sound.getPlayer()
# Register keybinding for toggle (Cthulhu+I)
self._register_keybinding()
# Connect to text caret movement events
self._connect_to_events()
logger.info("IndentationAudio plugin activated successfully")
return True
except Exception as e:
logger.error(f"Failed to activate IndentationAudio plugin: {e}")
return False
@cthulhu_hookimpl
def deactivate(self, plugin=None):
"""Deactivate the IndentationAudio plugin."""
if plugin is not None and plugin is not self:
return
try:
logger.info("=== IndentationAudio plugin deactivation starting ===")
# Disconnect from events
self._disconnect_from_events()
# Clear tracking data
self._last_indentation_level.clear()
logger.info("IndentationAudio plugin deactivated successfully")
return True
except Exception as e:
logger.error(f"Failed to deactivate IndentationAudio plugin: {e}")
return False
def _register_keybinding(self):
"""Register the Cthulhu+I keybinding for toggling the plugin."""
try:
if not self.app:
logger.error("IndentationAudio: No app reference available for keybinding")
return
api_helper = self.app.getAPIHelper()
if not api_helper:
logger.error("IndentationAudio: No API helper available")
return
# Register Cthulhu+I keybinding
gesture_string = "Cthulhu+i"
description = "Toggle indentation audio feedback"
handler = self._toggle_indentation_audio
self._kb_binding = api_helper.registerGestureByString(
gesture_string, handler, description
)
if self._kb_binding:
logger.info(f"IndentationAudio: Registered keybinding {gesture_string}")
else:
logger.error(f"IndentationAudio: Failed to register keybinding {gesture_string}")
except Exception as e:
logger.error(f"IndentationAudio: Error registering keybinding: {e}")
def _connect_to_events(self):
"""Connect to text navigation events."""
try:
# Hook into the dynamic API to make ourselves available to scripts
if self.app:
api_manager = self.app.getDynamicApiManager()
api_manager.registerAPI('IndentationAudioPlugin', self)
logger.info("IndentationAudio: Registered with dynamic API manager")
# We'll also monkey-patch the default script's sayLine method
self._monkey_patch_script_methods()
except Exception as e:
logger.error(f"IndentationAudio: Error connecting to events: {e}")
def _disconnect_from_events(self):
"""Disconnect from text navigation events."""
try:
# Unregister from dynamic API
if self.app:
api_manager = self.app.getDynamicApiManager()
api_manager.unregisterAPI('IndentationAudioPlugin')
# Restore original script methods
self._restore_script_methods()
except Exception as e:
logger.error(f"IndentationAudio: Error disconnecting from events: {e}")
def _monkey_patch_script_methods(self):
"""Monkey-patch the default script's line navigation methods."""
try:
# Get the current active script
if self.app:
state = self.app.getDynamicApiManager().getAPI('CthulhuState')
if state and hasattr(state, 'activeScript') and state.activeScript:
script = state.activeScript
# Store original method
if hasattr(script, 'sayLine'):
self._original_sayLine = script.sayLine
# Create wrapped version
def wrapped_sayLine(obj):
# Call original method first
result = self._original_sayLine(obj)
# Add our indentation audio check
try:
line, caretOffset, startOffset = script.getTextLineAtCaret(obj)
self.check_indentation_change(obj, line)
except Exception as e:
logger.error(f"IndentationAudio: Error in wrapped_sayLine: {e}")
return result
# Replace the method
script.sayLine = wrapped_sayLine
logger.info("IndentationAudio: Successfully monkey-patched sayLine method")
except Exception as e:
logger.error(f"IndentationAudio: Error monkey-patching script methods: {e}")
def _restore_script_methods(self):
"""Restore original script methods."""
try:
if self.app and hasattr(self, '_original_sayLine'):
state = self.app.getDynamicApiManager().getAPI('CthulhuState')
if state and hasattr(state, 'activeScript') and state.activeScript:
script = state.activeScript
if hasattr(script, 'sayLine'):
script.sayLine = self._original_sayLine
logger.info("IndentationAudio: Restored original sayLine method")
except Exception as e:
logger.error(f"IndentationAudio: Error restoring script methods: {e}")
def _toggle_indentation_audio(self, script, inputEvent=None):
"""Toggle the indentation audio feedback on/off."""
try:
self._enabled = not self._enabled
state = "enabled" if self._enabled else "disabled"
# Announce the state change
message = f"Indentation audio {state}"
if hasattr(script, 'speakMessage'):
script.speakMessage(message)
logger.info(f"IndentationAudio: Toggled to {state}")
return True
except Exception as e:
logger.error(f"IndentationAudio: Error toggling state: {e}")
return False
def _calculate_indentation_level(self, line_text):
"""Calculate the indentation level of a line."""
if not line_text:
return 0
# Remove non-breaking spaces and convert to regular spaces
line = line_text.replace("\u00a0", " ")
# Find the first non-whitespace character
match = re.search(r"[^ \t]", line)
if not match:
return 0 # Empty or whitespace-only line
indent_text = line[:match.start()]
# Calculate indentation level (4 spaces = 1 level, 1 tab = 1 level)
level = 0
for char in indent_text:
if char == '\t':
level += 1
elif char == ' ':
level += 0.25 # 4 spaces = 1 level
return int(level)
def _generate_indentation_tone(self, new_level, old_level):
"""Generate an audio tone for indentation level change."""
if not self._enabled or not self._player:
return
# Calculate frequency based on new indentation level
frequency = min(
self._base_frequency + (new_level * self._frequency_step),
self._max_frequency
)
# Determine stereo panning based on change direction
# Left channel for indent increase, right for decrease
volume_multiplier = 0.7
try:
# Create tone
tone = Tone(
duration=self._tone_duration,
frequency=frequency,
volumeMultiplier=volume_multiplier,
wave=Tone.SINE_WAVE
)
# Play the tone
self._player.play(tone, interrupt=False)
debug_msg = f"IndentationAudio: Played tone - Level: {new_level}, Freq: {frequency}Hz"
logger.debug(debug_msg)
except Exception as e:
logger.error(f"IndentationAudio: Error generating tone: {e}")
def check_indentation_change(self, obj, line_text):
"""Check if indentation has changed and play audio cue if needed.
This method is intended to be called by scripts during line navigation.
"""
if not self._enabled or not line_text:
return
try:
# Get object identifier for tracking
obj_id = str(obj) if obj else "unknown"
# Calculate current indentation level
current_level = self._calculate_indentation_level(line_text)
# Get previous level for this object
previous_level = self._last_indentation_level.get(obj_id, current_level)
# Update tracking
self._last_indentation_level[obj_id] = current_level
# Play audio cue if indentation changed
if current_level != previous_level:
self._generate_indentation_tone(current_level, previous_level)
debug_msg = f"IndentationAudio: Indentation changed from {previous_level} to {current_level}"
logger.debug(debug_msg)
except Exception as e:
logger.error(f"IndentationAudio: Error checking indentation change: {e}")
def is_enabled(self):
"""Return whether the plugin is currently enabled."""
return self._enabled
def set_enabled(self, enabled):
"""Set the enabled state of the plugin."""
self._enabled = enabled
def on_script_change(self, new_script):
"""Handle when the active script changes."""
try:
# Restore previous script if it was patched
self._restore_script_methods()
# Re-apply patches to new script
self._monkey_patch_script_methods()
# Clear tracking data for new context
self._last_indentation_level.clear()
logger.info("IndentationAudio: Handled script change")
except Exception as e:
logger.error(f"IndentationAudio: Error handling script change: {e}")

View File

@ -1,4 +1,4 @@
SUBDIRS = Clipboard DisplayVersion hello_world self_voice ByeCthulhu HelloCthulhu SimplePluginSystem
SUBDIRS = Clipboard DisplayVersion IndentationAudio hello_world self_voice ByeCthulhu HelloCthulhu SimplePluginSystem
cthulhu_pythondir=$(pkgpythondir)/plugins