From 265feb8188876dedcb95b66a90c134bf846029a1 Mon Sep 17 00:00:00 2001 From: Storm Dragon Date: Sun, 26 Apr 2026 13:18:23 -0400 Subject: [PATCH] Backport Orca's d-bus remote fixes. --- README-DEVELOPMENT.md | 8 +- README-REMOTE-CONTROLLER.md | 263 ++-- README.md | 25 +- REMOTE-CONTROLLER-COMMANDS.md | 76 +- src/cthulhu/dbus_service.py | 1405 ++++++++++-------- src/cthulhu/plugin_system_manager.py | 4 +- tests/test_dbus_service_native_interfaces.py | 124 ++ 7 files changed, 1017 insertions(+), 888 deletions(-) create mode 100644 tests/test_dbus_service_native_interfaces.py diff --git a/README-DEVELOPMENT.md b/README-DEVELOPMENT.md index 6571e6c..c6ce53f 100644 --- a/README-DEVELOPMENT.md +++ b/README-DEVELOPMENT.md @@ -59,8 +59,8 @@ cthulhu Cthulhu now includes a D-Bus service for remote control: -- **Service**: `org.stormux.Cthulhu.Service` -- **Path**: `/org/stormux/Cthulhu/Service` +- **Service**: `org.stormux.Cthulhu1.Service` +- **Path**: `/org/stormux/Cthulhu1/Service` - **Requires**: `dasbus` library (should be installed) ### Testing D-Bus Service @@ -70,10 +70,10 @@ Cthulhu now includes a D-Bus service for remote control: ~/.local/bin/cthulhu # In another terminal, test the service -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service org.stormux.Cthulhu.Service GetVersion +busctl --user call org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service org.stormux.Cthulhu1.Service GetVersion # Present a message via D-Bus -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service org.stormux.Cthulhu.Service PresentMessage s "Hello from D-Bus" +busctl --user call org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service org.stormux.Cthulhu1.Service PresentMessage s "Hello from D-Bus" ``` ## Development Workflow diff --git a/README-REMOTE-CONTROLLER.md b/README-REMOTE-CONTROLLER.md index 320d4be..bc7d403 100644 --- a/README-REMOTE-CONTROLLER.md +++ b/README-REMOTE-CONTROLLER.md @@ -22,10 +22,11 @@ on any Linux desktop environment or window manager. Cthulhu exposes a D-Bus service at: -- **Service Name**: `org.stormux.Cthulhu.Service` -- **Main Object Path**: `/org/stormux/Cthulhu/Service` -- **Module Object Paths**: `/org/stormux.Cthulhu/Service/ModuleName` - (e.g., `/org/stormux/Cthulhu/Service/SpeechAndVerbosityManager`) +- **Service Name**: `org.stormux.Cthulhu1.Service` +- **Main Object Path**: `/org/stormux/Cthulhu1/Service` +- **Module Object Paths**: `/org/stormux/Cthulhu1/Service/ModuleName` + (e.g., `/org/stormux/Cthulhu1/Service/SpeechManager`) +- **Module Interfaces**: `org.stormux.Cthulhu1.ModuleName` See [REMOTE-CONTROLLER-COMMANDS.md](REMOTE-CONTROLLER-COMMANDS.md) for a complete list of available commands. @@ -43,34 +44,34 @@ While this documentation primarily uses `gdbus` for examples, you can use any D- ### Using `busctl` (systemd D-Bus tool) ```bash -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service GetVersion +busctl --user call org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service \ + org.stormux.Cthulhu1.Service GetVersion ``` ### Using Python with `dasbus` ```python from dasbus.connection import SessionMessageBus bus = SessionMessageBus() -proxy = bus.get_proxy("org.stormux.Cthulhu.Service", "/org/stormux/Cthulhu/Service") +proxy = bus.get_proxy("org.stormux.Cthulhu1.Service", "/org/stormux/Cthulhu1/Service") version = proxy.GetVersion() ``` ### Using `qdbus` (Qt D-Bus tool - available on KDE) ```bash -qdbus org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service.GetVersion +qdbus org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service \ + org.stormux.Cthulhu1.Service.GetVersion ``` ## Service-Level Commands -Commands available directly on the main service (`/org/stormux/Cthulhu/Service`): +Commands available directly on the main service (`/org/stormux/Cthulhu1/Service`): ### Get Cthulhu's Version ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service \ - --method org.stormux.Cthulhu.Service.GetVersion +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service \ + --method org.stormux.Cthulhu1.Service.GetVersion ``` **Returns:** String containing the version (and revision if available) @@ -78,9 +79,9 @@ gdbus call --session --dest org.stormux.Cthulhu.Service \ ### Present a Custom Message in Speech and/or Braille ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service \ - --method org.stormux.Cthulhu.Service.PresentMessage "Your message here" +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service \ + --method org.stormux.Cthulhu1.Service.PresentMessage "Your message here" ``` **Parameters:** @@ -92,9 +93,9 @@ gdbus call --session --dest org.stormux.Cthulhu.Service \ ### Show Cthulhu's Preferences GUI ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service \ - --method org.stormux.Cthulhu.Service.ShowPreferences +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service \ + --method org.stormux.Cthulhu1.Service.ShowPreferences ``` **Returns:** Boolean indicating success @@ -102,44 +103,39 @@ gdbus call --session --dest org.stormux.Cthulhu.Service \ ### Quit Cthulhu ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service \ - --method org.stormux.Cthulhu.Service.Quit +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service \ + --method org.stormux.Cthulhu1.Service.Quit ``` **Returns:** Boolean indicating if the quit request was accepted -### List Available Service Commands +## Discovering Modules and Their Capabilities + +Use the standard DBus introspection interface to discover registered modules: ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service \ - --method org.stormux.Cthulhu.Service.ListCommands +gdbus introspect --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service --recurse ``` -**Returns:** List of (command_name, description) tuples - -### List Registered Modules +The child `` entries beneath `/org/stormux/Cthulhu1/Service` are the +registered modules. To inspect the methods and properties for one module: ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service \ - --method org.stormux.Cthulhu.Service.ListModules +gdbus introspect --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/SpeechManager ``` -**Returns:** List of module names - ## Interacting with Modules -Each registered module exposes its own set of operations. Based on the underlying Cthulhu code, these -are categorized as **Commands**, **Runtime Getters**, and **Runtime Setters**: +Each registered module exposes its own native DBus interface. Based on the underlying Cthulhu code, +these are categorized as **Commands** and **Properties**: - **Commands**: Actions that perform a task. These typically correspond to Cthulhu commands bound to a keystroke (e.g., `IncreaseRate`). -- **Runtime Getters**: Operations that retrieve the current value of an item, often a setting - (e.g., `GetRate`). -- **Runtime Setters**: Operations that set the current value of an item, often a setting - (e.g., `SetRate`). Note that setting a value does NOT cause it to become permanently saved. +- **Properties**: Runtime values, often settings (e.g., `Rate`). Setting a property does not cause + it to become permanently saved. You can discover and execute these for each module. @@ -147,41 +143,41 @@ You can discover and execute these for each module. Plugins that expose D-Bus decorators are automatically registered as modules using the naming convention `Plugin_` (e.g., `Plugin_GameMode`, `Plugin_WindowTitleReader`). Use -`ListModules` to discover available plugin modules at runtime. +standard DBus introspection to discover available plugin modules at runtime. #### Plugin_WindowTitleReader Controls for the Window Title Reader plugin: -- Parameterized command: `SetEnabled` (`enabled`: bool) -- Runtime getter: `Enabled` +- Method: `SetEnabled` (`enabled`: bool, `notify_user`: bool) +- Property: `Enabled` Example: ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/Plugin_WindowTitleReader \ - --method org.stormux.Cthulhu.Module.ExecuteParameterizedCommand \ - 'SetEnabled' '{"enabled": }' false +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader \ + --method org.stormux.Cthulhu1.Plugin_WindowTitleReader.SetEnabled true false # Check current state -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/Plugin_WindowTitleReader \ - --method org.stormux.Cthulhu.Module.ExecuteRuntimeGetter 'Enabled' +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader \ + --method org.freedesktop.DBus.Properties.Get \ + org.stormux.Cthulhu1.Plugin_WindowTitleReader Enabled ``` Busctl example: ```bash -busctl --user call org.stormux.Cthulhu.Service \ - /org/stormux/Cthulhu/Service/Plugin_WindowTitleReader \ - org.stormux.Cthulhu.Module ExecuteParameterizedCommand \ - s a{sv} b 'SetEnabled' 1 enabled b true false +busctl --user call org.stormux.Cthulhu1.Service \ + /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader \ + org.stormux.Cthulhu1.Plugin_WindowTitleReader SetEnabled bb true false # Check current state -busctl --user call org.stormux.Cthulhu.Service \ - /org/stormux/Cthulhu/Service/Plugin_WindowTitleReader \ - org.stormux.Cthulhu.Module ExecuteRuntimeGetter s 'Enabled' +busctl --user call org.stormux.Cthulhu1.Service \ + /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader \ + org.freedesktop.DBus.Properties Get ss \ + org.stormux.Cthulhu1.Plugin_WindowTitleReader Enabled ``` ### PluginSystemManager Module @@ -190,166 +186,85 @@ The `PluginSystemManager` module provides session-only plugin control: - `ListPlugins` - `ListActivePlugins` -- `IsPluginActive` (parameterized) -- `SetPluginActive` (parameterized) +- `IsPluginActive` +- `SetPluginActive` - `RescanPlugins` These calls do **not** persist changes to user preferences. -### Discovering Module Capabilities - -#### List Commands for a Module - -```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ListCommands -``` - -Replace `ModuleName` with an actual module name from `ListModules`. - -**Returns:** List of (command_name, description) tuples. - -#### List Parameterized Commands for a Module - -```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ListParameterizedCommands -``` - -Replace `ModuleName` with an actual module name from `ListModules`. - -**Returns:** List of (command_name, description, parameters) tuples, where `parameters` is a -list of (parameter_name, parameter_type) tuples. - -**Example output:** - -```bash -([('GetVoicesForLanguage', - 'Returns a list of available voices for the specified language.', - [('language', 'str'), ('variant', 'str'), ('notify_user', 'bool')])],) -``` - -#### List Runtime Getters for a Module - -```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ListRuntimeGetters -``` - -Replace `ModuleName` with an actual module name from `ListModules`. - -**Returns:** List of (getter_name, description) tuples. - -#### List Runtime Setters for a Module - -```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ListRuntimeSetters -``` - -Replace `ModuleName` with an actual module name from `ListModules`. - -**Returns:** List of (setter_name, description) tuples. - ### Executing Module Operations -#### Execute a Runtime Getter +#### Get a Property ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ExecuteRuntimeGetter 'PropertyName' +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/ModuleName \ + --method org.freedesktop.DBus.Properties.Get \ + org.stormux.Cthulhu1.ModuleName PropertyName ``` -**Parameters:** - -- `PropertyName` (string): The name of the runtime getter to execute. - -**Returns:** The value returned by the getter as a GLib variant (type depends on the getter). - ##### Example: Get the current speech rate ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/SpeechAndVerbosityManager \ - --method org.stormux.Cthulhu.Module.ExecuteRuntimeGetter 'Rate' - +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/SpeechManager \ + --method org.freedesktop.DBus.Properties.Get \ + org.stormux.Cthulhu1.SpeechManager Rate ``` -This will return the rate as a GLib Variant. - -#### Execute a Runtime Setter +#### Set a Property ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ExecuteRuntimeSetter 'PropertyName' +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/ModuleName \ + --method org.freedesktop.DBus.Properties.Set \ + org.stormux.Cthulhu1.ModuleName PropertyName '' ``` -**Parameters:** - -- `PropertyName` (string): The name of the runtime setter to execute. -- ``: The value to set, as a GLib variant (type depends on the setter). - -**Returns:** Boolean indicating success. - ##### Example: Set the current speech rate ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/SpeechAndVerbosityManager \ - --method org.stormux.Cthulhu.Module.ExecuteRuntimeSetter 'Rate' '<90>' +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/SpeechManager \ + --method org.freedesktop.DBus.Properties.Set \ + org.stormux.Cthulhu1.SpeechManager Rate '<90>' ``` #### Execute a Module Command ```bash # With user notification -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ExecuteCommand 'CommandName' true +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/ModuleName \ + --method org.stormux.Cthulhu1.ModuleName.CommandName true # Without user notification (silent) -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ExecuteCommand 'CommandName' false +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/ModuleName \ + --method org.stormux.Cthulhu1.ModuleName.CommandName false ``` -**Parameters (both required):** - -- `CommandName` (string): The name of the command to execute - `notify_user` (boolean): Whether to notify the user of the action (see section below) -**Returns:** Boolean indicating success - #### Execute a Parameterized Command ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ExecuteParameterizedCommand 'CommandName' \ - '{"param1": <"value1">, "param2": <"value2">}' false +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/ModuleName \ + --method org.stormux.Cthulhu1.ModuleName.CommandName \ + "value1" "value2" false ``` **Parameters:** -- `CommandName` (string): The name of the parameterized command to execute -- `parameters` (dict): Dictionary of parameter names and values as GLib variants - `notify_user` (boolean): Whether to notify the user of the action -**Returns:** The result returned by the command as a GLib variant (type depends on the command) - ##### Example: Get voices for a specific language ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/SpeechAndVerbosityManager \ - --method org.stormux.Cthulhu.Module.ExecuteParameterizedCommand 'GetVoicesForLanguage' \ - '{"language": <"en-us">, "variant": <"">}' false +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/SpeechManager \ + --method org.stormux.Cthulhu1.SpeechManager.GetVoicesForLanguage "en-us" "" false ``` This will return a list of available voices for US English. @@ -362,9 +277,9 @@ Some commands inherently don't make sense to announce. For example: ```bash # This command should simply stop speech, not announce that it is stopping speech. -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/SpeechAndVerbosityManager \ - --method org.stormux.Cthulhu.Module.ExecuteCommand 'InterruptSpeech' true +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/SpeechManager \ + --method org.stormux.Cthulhu1.SpeechManager.InterruptSpeech true ``` In those cases Cthulhu will ignore the value of `notify_user`. diff --git a/README.md b/README.md index 54b5b8c..10ad95e 100644 --- a/README.md +++ b/README.md @@ -66,22 +66,21 @@ toolkit, OpenOffice/LibreOffice, Gecko, WebKitGtk, and KDE Qt toolkit. Cthulhu exposes a D-Bus service for external automation and integrations. ### Service Details -- **Service Name**: `org.stormux.Cthulhu.Service` -- **Main Object Path**: `/org/stormux/Cthulhu/Service` -- **Module Object Paths**: `/org/stormux/Cthulhu/Service/` +- **Service Name**: `org.stormux.Cthulhu1.Service` +- **Main Object Path**: `/org/stormux/Cthulhu1/Service` +- **Module Object Paths**: `/org/stormux/Cthulhu1/Service/` +- **Module Interfaces**: `org.stormux.Cthulhu1.` ### Discovering Capabilities ```bash -# List registered modules -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service \ - --method org.stormux.Cthulhu.Service.ListModules +# List registered module object paths and introspect their methods/properties +gdbus introspect --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service --recurse -# List commands on a module -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/ModuleName \ - --method org.stormux.Cthulhu.Module.ListCommands +# Inspect one module +gdbus introspect --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/ModuleName ``` ### Plugin Modules @@ -95,8 +94,8 @@ The `PluginSystemManager` module provides **session-only** plugin control (no pr - `ListPlugins` - `ListActivePlugins` -- `IsPluginActive` (parameterized) -- `SetPluginActive` (parameterized) +- `IsPluginActive` +- `SetPluginActive` - `RescanPlugins` ### Plugin Preferences Pages diff --git a/REMOTE-CONTROLLER-COMMANDS.md b/REMOTE-CONTROLLER-COMMANDS.md index 4e481b6..79cb094 100644 --- a/REMOTE-CONTROLLER-COMMANDS.md +++ b/REMOTE-CONTROLLER-COMMANDS.md @@ -11,7 +11,7 @@ This document lists the currently available D-Bus commands in Cthulhu's Remote C ## Service-Level Commands -Available on the main service object `/org/stormux/Cthulhu/Service`: +Available on the main service object `/org/stormux/Cthulhu1/Service`: ### Service Commands @@ -21,40 +21,37 @@ Available on the main service object `/org/stormux/Cthulhu/Service`: | `PresentMessage` | Present a message via speech/braille | `message` (string) | Boolean (success) | | `ShowPreferences` | Opens Cthulhu's preferences GUI | None | Boolean (success) | | `Quit` | Exits Cthulhu | None | Boolean (accepted) | -| `ListCommands` | Lists available service commands | None | List of (name, description) tuples | -| `ListModules` | Lists registered D-Bus modules | None | List of module names | ### Example Usage ```bash # Get Cthulhu version -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service GetVersion +busctl --user call org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service \ + org.stormux.Cthulhu1.Service GetVersion # Present a custom message -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service PresentMessage s "Hello from D-Bus" - -# List available commands -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service ListCommands - -# List registered modules -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service ListModules +busctl --user call org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service \ + org.stormux.Cthulhu1.Service PresentMessage s "Hello from D-Bus" # Open preferences -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service ShowPreferences +busctl --user call org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service \ + org.stormux.Cthulhu1.Service ShowPreferences # Quit Cthulhu -busctl --user call org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service \ - org.stormux.Cthulhu.Service Quit +busctl --user call org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service \ + org.stormux.Cthulhu1.Service Quit ``` ## Module-Level Commands -Module-level commands are available and can be discovered via `ListModules`. Two key additions are: +Module-level commands and properties are discovered with standard DBus introspection: + +```bash +gdbus introspect --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service --recurse +``` + +Two key additions are: ### PluginSystemManager @@ -62,8 +59,8 @@ Session-only plugin control (does not persist preferences): - `ListPlugins` - `ListActivePlugins` -- `IsPluginActive` (parameterized) -- `SetPluginActive` (parameterized) +- `IsPluginActive` +- `SetPluginActive` - `RescanPlugins` ### Plugin Modules @@ -73,31 +70,30 @@ convention `Plugin_` (e.g., `Plugin_GameMode`, `Plugin_WindowTitleRe #### WindowTitleReader (Plugin_WindowTitleReader) -- `SetEnabled` (parameterized) -> enabled (bool) -- `Enabled` (runtime getter) +- `SetEnabled` -> enabled (bool), notify_user (bool) +- `Enabled` (property) Example: ```bash -gdbus call --session --dest org.stormux.Cthulhu.Service \ - --object-path /org/stormux/Cthulhu/Service/Plugin_WindowTitleReader \ - --method org.stormux.Cthulhu.Module.ExecuteParameterizedCommand \ - 'SetEnabled' '{"enabled": }' false +gdbus call --session --dest org.stormux.Cthulhu1.Service \ + --object-path /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader \ + --method org.stormux.Cthulhu1.Plugin_WindowTitleReader.SetEnabled true false ``` Busctl example: ```bash -busctl --user call org.stormux.Cthulhu.Service \ - /org/stormux/Cthulhu/Service/Plugin_WindowTitleReader \ - org.stormux.Cthulhu.Module ExecuteParameterizedCommand \ - s a{sv} b 'SetEnabled' 1 enabled b true false +busctl --user call org.stormux.Cthulhu1.Service \ + /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader \ + org.stormux.Cthulhu1.Plugin_WindowTitleReader SetEnabled bb true false ``` # Check current state -busctl --user call org.stormux.Cthulhu.Service \ - /org/stormux/Cthulhu/Service/Plugin_WindowTitleReader \ - org.stormux.Cthulhu.Module ExecuteRuntimeGetter s 'Enabled' +busctl --user call org.stormux.Cthulhu1.Service \ + /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader \ + org.freedesktop.DBus.Properties Get ss \ + org.stormux.Cthulhu1.Plugin_WindowTitleReader Enabled See [README-REMOTE-CONTROLLER.md](README-REMOTE-CONTROLLER.md) for comprehensive D-Bus API documentation and usage examples. @@ -107,11 +103,15 @@ See [README-REMOTE-CONTROLLER.md](README-REMOTE-CONTROLLER.md) for comprehensive # Check if Cthulhu's D-Bus service is running busctl --user list | grep Cthulhu -# Introspect the service to see all available methods -busctl --user introspect org.stormux.Cthulhu.Service /org/stormux/Cthulhu/Service +# Introspect the service to see available module nodes +busctl --user introspect org.stormux.Cthulhu1.Service /org/stormux/Cthulhu1/Service + +# Introspect a module to see its methods and properties +busctl --user introspect org.stormux.Cthulhu1.Service \ + /org/stormux/Cthulhu1/Service/Plugin_WindowTitleReader # Get detailed service information -busctl --user status org.stormux.Cthulhu.Service +busctl --user status org.stormux.Cthulhu1.Service ``` ## Contributing diff --git a/src/cthulhu/dbus_service.py b/src/cthulhu/dbus_service.py index 45b2057..01c771d 100644 --- a/src/cthulhu/dbus_service.py +++ b/src/cthulhu/dbus_service.py @@ -1,6 +1,5 @@ # Cthulhu # -# Copyright 2025 Stormux # Copyright 2025 Valve Corporation # Author: Joanmarie Diggs # @@ -21,54 +20,67 @@ """Provides a D-Bus interface for remotely controlling Cthulhu.""" -__id__ = "$Id$" -__version__ = "$Revision$" -__date__ = "$Date$" -__copyright__ = "Copyright (c) 2025 Stormux " -__license__ = "LGPL" - +import ast +import contextlib import enum import inspect import os -from typing import Any, Callable, Optional +import types +import typing +import xml.etree.ElementTree as ET +from collections.abc import Callable +from typing import Optional + +from gi.repository import GLib 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 + from dasbus.typing import UInt32 as UInt32 # noqa: PLC0414 (re-export) _dasbus_available = True except ImportError: + SessionMessageBus = None + DBusError = Exception + Publishable = object + UInt32 = int _dasbus_available = False -from gi.repository import GLib - import gi gi.require_version("Atspi", "2.0") from gi.repository import Atspi -from . import debug -from . import cthulhu_platform # pylint: disable=no-name-in-module -from . import script_manager -from . import cthulhu_state +from . import ( # pylint: disable=no-name-in-module + cthulhu_state, + debug, + script_manager, + cthulhu_platform, +) + + +if not _dasbus_available: + def dbus_interface(_interface_name): + """Returns a no-op interface decorator when dasbus is unavailable.""" + + def _decorator(cls): + cls.__dbus_xml__ = "" + return cls + + return _decorator + -# Lazy imports to avoid circular dependency def _get_input_event(): - from . import input_event + from . import input_event # pylint: disable=import-outside-toplevel + return input_event + def _get_input_event_manager(): - from . import input_event_manager + from . import input_event_manager # pylint: disable=import-outside-toplevel + return input_event_manager -class HandlerType(enum.Enum): - """Enumeration of handler types for D-Bus methods.""" - - COMMAND = enum.auto() - PARAMETERIZED_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. @@ -83,18 +95,6 @@ def command(func): 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 parameterized_command(func): """Decorator to mark a method as a D-Bus parameterized command using its docstring. @@ -116,6 +116,21 @@ def parameterized_command(func): func.dbus_parameterized_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. @@ -129,29 +144,8 @@ def setter(func): func.dbus_setter_description = description return func -def _extract_function_parameters(func: Callable) -> list[tuple[str, str]]: - """Extract parameter names and types from a function signature.""" - sig = inspect.signature(func) - parameters = [] - - skip_params = {"self", "script", "event"} - for param_name, param in sig.parameters.items(): - if param_name in skip_params: - continue - - if param.annotation != inspect.Parameter.empty: - if hasattr(param.annotation, "__name__"): - type_str = param.annotation.__name__ - else: - type_str = str(param.annotation).replace("typing.", "") - else: - type_str = "Any" - parameters.append((param_name, type_str)) - - return parameters - -def _filter_kwargs_for_callable(method: Callable, kwargs: dict[str, Any]) -> dict[str, Any]: +def _filter_kwargs_for_callable(method: Callable, kwargs: dict[str, object]) -> dict[str, object]: """Filters kwargs down to what the callable accepts.""" try: @@ -168,442 +162,588 @@ def _filter_kwargs_for_callable(method: Callable, kwargs: dict[str, Any]) -> dic return {key: value for key, value in kwargs.items() if key in allowed} -class _HandlerInfo: - """Stores processed information about a function exposed via D-Bus.""" +class _Kind(enum.Enum): + """Decorated-method kinds detected during module registration.""" - def __init__( - self, - python_function_name: str, - description: str, - action: Callable[..., bool], - handler_type: 'HandlerType' = HandlerType.COMMAND, - parameters: Optional[list[tuple[str, str]]] = None - ): - self.python_function_name: str = python_function_name - self.description: str = description - self.action: Callable[..., bool] = action - self.handler_type: HandlerType = handler_type - self.parameters: list[tuple[str, str]] = parameters or [] + COMMAND = enum.auto() + PARAMETERIZED = enum.auto() + GETTER = enum.auto() + SETTER = enum.auto() -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).""" +class _ModuleRegistration: # pylint: disable=too-many-instance-attributes + """Tracks a module's decorated methods and its published D-Bus interface.""" - def __init__(self, - module_name: str, - handlers_info: list[_HandlerInfo]): - super().__init__() - self._module_name = module_name - self._commands: dict[str, _HandlerInfo] = {} - self._parameterized_commands: dict[str, _HandlerInfo] = {} - self._getters: dict[str, _HandlerInfo] = {} - self._setters: dict[str, _HandlerInfo] = {} + def __init__(self, module_name: str) -> None: + self._module_name: str = module_name + self._commands: dict[str, Callable] = {} + self._parameterized_commands: dict[str, Callable] = {} + self._getters: dict[str, Callable] = {} + self._setters: dict[str, Callable] = {} + self._descriptions: dict[str, str] = {} + self._dbus_object: Optional[object] = None + self._object_path: str = "" - for info in handlers_info: - handler_type = getattr(info, "handler_type", HandlerType.COMMAND) - normalized_name = self._normalize_handler_name(info.python_function_name, handler_type) - if handler_type == HandlerType.GETTER: - self._getters[normalized_name] = info - elif handler_type == HandlerType.SETTER: - self._setters[normalized_name] = info - elif handler_type == HandlerType.PARAMETERIZED_COMMAND: - self._parameterized_commands[normalized_name] = info - else: - self._commands[normalized_name] = info + def get_module_name(self) -> str: + """Returns the module name.""" - msg = ( - f"DBUS SERVICE: CthulhuModuleDBusInterface for {module_name} initialized " - f"with {len(self._commands)} command(s), " - f"{len(self._parameterized_commands)} parameterized command(s), " - f"{len(self._getters)} getter(s), {len(self._setters)} setter(s)." - ) - debug.printMessage(debug.LEVEL_INFO, msg, True) + return self._module_name - 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.""" + def get_commands(self) -> dict[str, Callable]: + """Returns the simple (non-parameterized) commands.""" - 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", "")) + return self._commands - 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 get_parameterized_commands(self) -> dict[str, Callable]: + """Returns the parameterized commands.""" - def ExecuteRuntimeSetter(self, setter_name: str, value: GLib.Variant) -> bool: # pylint: disable=invalid-name - """Executes the named setter, returning True if succeeded.""" + return self._parameterized_commands - 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 + def get_getters(self) -> dict[str, Callable]: + """Returns the property getters.""" - 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 + return self._getters - def ListCommands(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name - """Returns a list of (command_name, description) for this module (commands only).""" + def get_setters(self) -> dict[str, Callable]: + """Returns the property setters.""" - command_list = [] - for camel_case_name, info in self._commands.items(): - command_list.append((camel_case_name, info.description)) - return command_list + return self._setters - def ListParameterizedCommands( # pylint: disable=invalid-name - self, - ) -> list[tuple[str, str, list[tuple[str, str]]]]: - """Returns a list of (command_name, description, parameters) for this module.""" + def get_descriptions(self) -> dict[str, str]: + """Returns the description for each exposed CamelCase member name.""" - command_list = [] - for camel_case_name, info in self._parameterized_commands.items(): - command_list.append((camel_case_name, info.description, info.parameters)) - return command_list + return self._descriptions - def ListRuntimeGetters(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name - """Returns a list of (getter_name, description) for this module.""" + def get_object_path(self) -> str: + """Returns the D-Bus object path under which the module is published.""" - getter_list = [] - for camel_case_name, info in self._getters.items(): - getter_list.append((camel_case_name, info.description)) - return getter_list + return self._object_path - def ListRuntimeSetters(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name - """Returns a list of (setter_name, description) for this module.""" + def set_object_path(self, path: str) -> None: + """Stores the D-Bus object path under which the module was published.""" - setter_list = [] - for camel_case_name, info in self._setters.items(): - setter_list.append((camel_case_name, info.description)) - return setter_list + self._object_path = path - 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.""" + def get_dbus_object(self) -> Optional[object]: + """Returns the published D-Bus interface instance, or None if not published.""" - 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 + return self._dbus_object - 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 set_dbus_object(self, obj: object) -> None: + """Stores the published D-Bus interface instance.""" - def ExecuteParameterizedCommand( # pylint: disable=invalid-name - self, - command_name: str, - parameters: dict[str, GLib.Variant], - notify_user: bool - ) -> GLib.Variant: - """Executes the named command with parameters and returns the result.""" + self._dbus_object = obj - handler_info = self._parameterized_commands.get(command_name) - if not handler_info: - msg = ( - f"DBUS SERVICE: Unknown parameterized command '{command_name}' for " - f"'{self._module_name}'." - ) - debug.printMessage(debug.LEVEL_WARNING, msg, True) - return GLib.Variant("b", False) + def is_empty(self) -> bool: + """Returns True if the registration has no decorated members.""" - kwargs = {name: variant.unpack() for name, variant in parameters.items()} - kwargs["notify_user"] = notify_user - result = handler_info.action(**kwargs) - msg = f"DBUS SERVICE: Parameterized '{command_name}' in '{self._module_name}' executed." - debug.printMessage(debug.LEVEL_INFO, msg, True) - return self._to_variant(result) + return not ( + self._commands or self._parameterized_commands or self._getters or self._setters + ) - def for_publication(self): - """Returns the D-Bus interface XML for publication.""" + def total_member_count(self) -> int: + """Returns the total number of exposed members (commands + properties).""" - return self.__dbus_xml__ # pylint: disable=no-member + return ( + len(self._commands) + + len(self._parameterized_commands) + + len(self._getters) + + len(self._setters) + ) + + def find_command(self, command_name: str) -> Optional[Callable]: + """Returns the original command method for the given CamelCase name, or None.""" + + return self._commands.get(command_name) or self._parameterized_commands.get(command_name) + + def find_getter(self, property_name: str) -> Optional[Callable]: + """Returns the original getter for the given CamelCase property name, or None.""" + + return self._getters.get(property_name) + + def find_setter(self, property_name: str) -> Optional[Callable]: + """Returns the original setter for the given CamelCase property name, or None.""" + + return self._setters.get(property_name) + + def add_decorated_member( + self, kind: _Kind, attr_name: str, method: Callable, description: str + ) -> None: + """Records a decorated method under its DBus member name.""" + + dbus_name = self._dbus_name_for(attr_name, kind) + if kind is _Kind.COMMAND: + self._commands[dbus_name] = method + self._descriptions[dbus_name] = description + elif kind is _Kind.PARAMETERIZED: + self._parameterized_commands[dbus_name] = method + self._descriptions[dbus_name] = description + elif kind is _Kind.GETTER: + self._getters[dbus_name] = method + if dbus_name not in self._descriptions: + self._descriptions[dbus_name] = description + elif kind is _Kind.SETTER: + self._setters[dbus_name] = method + self._descriptions[dbus_name] = description + + @staticmethod + def _dbus_name_for(attr_name: str, kind: _Kind) -> str: + """Returns the CamelCase D-Bus name for a Python attribute name and decorator kind.""" + + if kind in (_Kind.GETTER, _Kind.SETTER) and attr_name.startswith(("get_", "set_")): + attr_name = attr_name[4:] + return "".join(word.capitalize() for word in attr_name.split("_")) + + @classmethod + def from_module_instance( + cls, module_name: str, module_instance: object + ) -> "_ModuleRegistration": + """Walks module_instance and groups its decorated members by kind.""" + + registration = cls(module_name) + for attr_name in dir(module_instance): + method = getattr(module_instance, attr_name, None) + if not callable(method): + continue + kind, description = cls._classify_method(method) + if kind is None: + continue + registration.add_decorated_member(kind, attr_name, method, description) + return registration + + @staticmethod + def _classify_method(method: Callable) -> tuple[Optional[_Kind], str]: + """Returns (kind, description) for a decorated method, or (None, '') if undecorated.""" + + description = getattr(method, "dbus_command_description", None) + if description is not None: + return _Kind.COMMAND, description + + description = getattr(method, "dbus_parameterized_command_description", None) + if description is not None: + return _Kind.PARAMETERIZED, description + + description = getattr(method, "dbus_getter_description", None) + if description is not None: + return _Kind.GETTER, description + + description = getattr(method, "dbus_setter_description", None) + if description is not None: + return _Kind.SETTER, description + + return None, "" - @staticmethod - def _normalize_handler_name( - function_name: str, - handler_type: HandlerType = HandlerType.COMMAND - ) -> str: - """Normalizes a Python function name for D-Bus exposure (getter/setter/command).""" +class _InterfaceBuilder: + """Builds an introspectable D-Bus interface class from a _ModuleRegistration.""" - # Only strip prefixes for getters and setters, not for commands - if handler_type in (HandlerType.GETTER, HandlerType.SETTER): - 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("_")) + _RESERVED_PARAMS = frozenset({"self", "script", "event", "notify_user"}) + _BUILTIN_TYPES: typing.ClassVar[dict[str, object]] = { + "bool": bool, + "int": int, + "UInt32": UInt32, + "float": float, + "str": str, + "list": list, + "tuple": tuple, + "dict": dict, + "None": type(None), + } - @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) - if isinstance(result, int): - return GLib.Variant("i", result) - if isinstance(result, float): - return GLib.Variant("d", result) - if isinstance(result, str): - return GLib.Variant("s", result) - if isinstance(result, dict): - return GLib.Variant( - "a{sv}", {str(k): GLib.Variant("v", v) for k, v in result.items()}) - if isinstance(result, (list, tuple)): - if all(isinstance(x, str) for x in result): - return GLib.Variant("as", list(result)) - if all(isinstance(x, bool) for x in result): - return GLib.Variant("ab", list(result)) - if all(isinstance(x, int) for x in result): - return GLib.Variant("ax", list(result)) - if all(isinstance(x, (list, tuple)) for x in result): - if not result: - return GLib.Variant("av", []) - first_len = len(result[0]) - converted = [tuple(str(item or "") for item in x) for x in result] - signature = "(" + "s" * first_len + ")" - return GLib.Variant(f"a{signature}", converted) - return GLib.Variant("av", [GLib.Variant("v", x) for x in result]) - if result is None: - return GLib.Variant("v", GLib.Variant("s", "")) - 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) + @classmethod + def build(cls, registration: _ModuleRegistration) -> type: + """Dynamically constructs a D-Bus interface class for the registered module.""" 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.""" + namespace: dict[str, object] = {"for_publication": for_publication} + for cname, method in registration.get_commands().items(): + namespace[cname] = cls._make_command_method(method) + for cname, method in registration.get_parameterized_commands().items(): + namespace[cname] = cls._make_parameterized_command_method(method) + getters = registration.get_getters() + setters = registration.get_setters() + for cname in set(getters) | set(setters): + namespace[cname] = cls._make_property(getters.get(cname), setters.get(cname)) - 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}" + module_name = registration.get_module_name() + new_cls = type(f"{module_name}DBusInterface", (Publishable,), namespace) + interface_name = f"org.stormux.Cthulhu1.{module_name}" + new_cls = dbus_interface(interface_name)(new_cls) + new_cls.__dbus_xml__ = cls._inject_docstrings( + new_cls.__dbus_xml__, registration.get_descriptions() + ) + return new_cls + + @staticmethod + def _inject_docstrings(xml_text: str, descriptions: dict[str, str]) -> str: + """Adds org.gtk.GDBus.DocString annotations to methods and properties in the XML.""" + + if not descriptions: + return xml_text + # XML is generated in-process by dasbus, not untrusted input. + root = ET.fromstring(xml_text) # noqa: S314 + for iface in root.findall("interface"): + for element in list(iface.findall("method")) + list(iface.findall("property")): + name = element.get("name") + description = descriptions.get(name) if name else None + if not description: + continue + annotation = ET.Element( + "annotation", + {"name": "org.gtk.GDBus.DocString", "value": description}, ) - debug.printMessage(debug.LEVEL_SEVERE, msg, True) + element.insert(0, annotation) + return ET.tostring(root, encoding="unicode") - 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.""" + @staticmethod + def _strip_optional(annotation): + """Returns the non-None branch of Optional[T] / T | None, else the annotation unchanged.""" - 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 + origin = typing.get_origin(annotation) + union_type = getattr(types, "UnionType", None) + union_origins = (typing.Union,) if union_type is None else (typing.Union, union_type) + if origin in union_origins: + non_none = tuple(arg for arg in typing.get_args(annotation) if arg is not type(None)) + if len(non_none) == 1: + return non_none[0] + return annotation - 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 + @classmethod + def _resolve_annotation(cls, annotation): + """Resolves annotation to a real type, or returns the original string.""" - def ListModules(self) -> list[str]: # pylint: disable=invalid-name - """Returns a list of registered module names.""" + # Resolve each annotation independently. typing.get_type_hints is all-or-nothing: + # a single TYPE_CHECKING-only name on a sibling parameter (e.g. `script: default.Script`) + # makes it raise NameError and lose every annotation in the function. Walking the AST + # per-annotation lets resolvable names, such as `language: str`, survive. + if not isinstance(annotation, str): + return annotation + try: + tree = ast.parse(annotation, mode="eval") + except SyntaxError: + return annotation + try: + return cls._type_from_node(tree.body) + except (KeyError, AttributeError, TypeError): + return annotation - return list(self._registered_modules) + @classmethod + def _type_from_node(cls, node: ast.AST): + """Reconstructs a type from an AST node using a fixed builtin whitelist.""" - def ListCommands(self) -> list[tuple[str, str]]: # pylint: disable=invalid-name - """Returns available commands on the main service interface.""" + if isinstance(node, ast.Name): + return cls._BUILTIN_TYPES[node.id] + if isinstance(node, ast.Constant): + if node.value is None: + return type(None) + raise KeyError(node.value) + if isinstance(node, ast.Subscript): + base = cls._type_from_node(node.value) + slice_node = node.slice + if isinstance(slice_node, ast.Tuple): + args = tuple(cls._type_from_node(elt) for elt in slice_node.elts) + return base[args] + return base[cls._type_from_node(slice_node)] + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): + return typing.Union[cls._type_from_node(node.left), cls._type_from_node(node.right)] + raise KeyError(ast.dump(node)) - 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)) + @classmethod + def _resolved_signature(cls, method: Callable) -> inspect.Signature: + """Returns method's signature with string annotations resolved per-parameter.""" - return sorted(commands) + sig = inspect.signature(method) + new_params = [ + param.replace(annotation=cls._resolve_annotation(param.annotation)) + for param in sig.parameters.values() + ] + return sig.replace( + parameters=new_params, + return_annotation=cls._resolve_annotation(sig.return_annotation), + ) - def ShowPreferences(self) -> bool: # pylint: disable=invalid-name - """Shows Cthulhu's preferences GUI.""" + @classmethod + def _make_command_method(cls, method: Callable) -> Callable: + """Builds a D-Bus method (notify_user) -> bool wrapping an @command method.""" - msg = "DBUS SERVICE: ShowPreferences called." - debug.printMessage(debug.LEVEL_INFO, msg, True) + return_annotation = cls._resolved_signature(method).return_annotation + coerce_to_bool = return_annotation in (inspect.Signature.empty, type(None), bool) + if coerce_to_bool: + return_annotation = bool + def Method(_self, notify_user: bool = True) -> bool: # pylint: disable=invalid-name + # Local imports break a circular import: dbus_service is imported by speech_manager, + # and script_manager (transitively) imports speech_manager. + event = _get_input_event().RemoteControllerEvent() manager = script_manager.get_manager() - script = cthulhu_state.activeScript or manager.get_default_script() - if script is None: - msg = "DBUS SERVICE: No script available" - debug.printMessage(debug.LEVEL_WARNING, msg, True) - return False - - script.showPreferencesGUI() - return True - - 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.get_manager() - script = cthulhu_state.activeScript or script_manager.get_manager().get_default_script() - 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, AT-SPI version, and session information.""" - - result = cthulhu_platform.version - if cthulhu_platform.revision: - result += f" (rev {cthulhu_platform.revision})" - - atspi_version = Atspi.get_version() - result += f", AT-SPI2 version: {atspi_version[0]}.{atspi_version[1]}.{atspi_version[2]}" - - session_type = os.environ.get("XDG_SESSION_TYPE") or "" - session_desktop = os.environ.get("XDG_SESSION_DESKTOP") or "" - session = f"{session_type} {session_desktop}".strip() - if session: - result += f", Session: {session}" - - msg = f"DBUS SERVICE: GetVersion called, returning: {result}" - debug.printMessage(debug.LEVEL_INFO, msg, True) + script = manager.get_active_script() or manager.get_default_script() + kwargs = {"script": script, "event": event, "notify_user": notify_user} + result = method(**_filter_kwargs_for_callable(method, kwargs)) + _get_input_event_manager().get_manager().process_remote_controller_event(event) + if coerce_to_bool: + return bool(result) return result - def Quit(self) -> bool: # pylint: disable=invalid-name - """Quits Cthulhu. Returns True if the quit request was accepted.""" + Method.__signature__ = inspect.Signature( # type: ignore[attr-defined] + [ + inspect.Parameter("self", inspect.Parameter.POSITIONAL_OR_KEYWORD), + inspect.Parameter( + "notify_user", + inspect.Parameter.POSITIONAL_OR_KEYWORD, + default=True, + annotation=bool, + ), + ], + return_annotation=return_annotation, + ) + Method.__annotations__ = {"notify_user": bool, "return": return_annotation} + return Method - msg = "DBUS SERVICE: Quit called." - debug.printMessage(debug.LEVEL_INFO, msg, True) + @classmethod + def _make_parameterized_command_method(cls, method: Callable) -> Callable: + """Builds a D-Bus method mirroring a @parameterized_command's user-facing signature.""" - from . import cthulhu # pylint: disable=import-outside-toplevel + original_sig = cls._resolved_signature(method) + user_params = [ + (name, param) + for name, param in original_sig.parameters.items() + if name not in cls._RESERVED_PARAMS + ] - # cthulhu.shutdown() shuts down the dbus service, so send the response immediately and then - # do the actual shutdown after a brief delay. - def _delayed_shutdown(): - cthulhu.shutdown() - return False - - GLib.timeout_add(100, _delayed_shutdown) - return True - - 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." + new_params = [inspect.Parameter("self", inspect.Parameter.POSITIONAL_OR_KEYWORD)] + annotations: dict[str, object] = {} + for name, param in user_params: + annotation = cls._strip_optional(param.annotation) + if annotation is inspect.Signature.empty: + annotation = cls._type_for_default(param.default) + new_params.append( + inspect.Parameter( + name, + inspect.Parameter.POSITIONAL_OR_KEYWORD, + default=param.default, + annotation=annotation, ) - 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() + ) + annotations[name] = annotation -else: - # Fallback classes when dasbus is not available - class CthulhuModuleDBusInterface: - """Fallback class when dasbus is not available.""" - def __init__(self, *args, **kwargs): - pass + new_params.append( + inspect.Parameter( + "notify_user", + inspect.Parameter.POSITIONAL_OR_KEYWORD, + default=True, + annotation=bool, + ) + ) + annotations["notify_user"] = bool + + return_annotation = original_sig.return_annotation + if return_annotation is inspect.Signature.empty: + return_annotation = bool + annotations["return"] = return_annotation + + new_sig = inspect.Signature(new_params, return_annotation=return_annotation) + + def Method(_self, *args, **kwargs): # pylint: disable=invalid-name + bound = new_sig.bind(_self, *args, **kwargs) + bound.apply_defaults() + bound.arguments.pop("self", None) + notify_user = bound.arguments.pop("notify_user") + + event = _get_input_event().RemoteControllerEvent() + manager = script_manager.get_manager() + script = manager.get_active_script() or manager.get_default_script() + kwargs = {"script": script, "event": event, "notify_user": notify_user} + kwargs.update(bound.arguments) + result = method(**_filter_kwargs_for_callable(method, kwargs)) + _get_input_event_manager().get_manager().process_remote_controller_event(event) + return result + + Method.__signature__ = new_sig # type: ignore[attr-defined] + Method.__annotations__ = annotations + return Method + + @staticmethod + def _type_for_default(default: object) -> object: + """Returns a DBus-compatible type hint for an unannotated parameter.""" + + if default is inspect.Signature.empty: + return str + if isinstance(default, bool): + return bool + if isinstance(default, int): + return int + if isinstance(default, float): + return float + if isinstance(default, str): + return str + return str + + @classmethod + def _make_property( + cls, get_method: Optional[Callable], set_method: Optional[Callable] + ) -> property: + """Builds a D-Bus property from a getter and/or setter pair.""" + + read = cls._make_property_getter(get_method) if get_method is not None else None + write = cls._make_property_setter(set_method) if set_method is not None else None + return property(read, write) + + @classmethod + def _make_property_getter(cls, get_method: Callable) -> Callable: + """Builds the read accessor for a D-Bus property, wrapping the original @getter method.""" + + return_annotation = cls._resolved_signature(get_method).return_annotation + if return_annotation is inspect.Signature.empty: + return_annotation = bool + + def read(_self, _original=get_method): + return _original() + + read.__signature__ = inspect.Signature( # type: ignore[attr-defined] + [inspect.Parameter("self", inspect.Parameter.POSITIONAL_OR_KEYWORD)], + return_annotation=return_annotation, + ) + read.__annotations__ = {"return": return_annotation} + return read + + @classmethod + def _make_property_setter(cls, set_method: Callable) -> Callable: + """Builds the write accessor for a D-Bus property, wrapping the original @setter method.""" + + set_sig = cls._resolved_signature(set_method) + value_param = next(param for name, param in set_sig.parameters.items() if name != "self") + value_type = cls._strip_optional(value_param.annotation) + if value_type is inspect.Signature.empty: + value_type = bool + + def write(_self, value, _original=set_method): + _original(value) + + write.__signature__ = inspect.Signature( # type: ignore[attr-defined] + [ + inspect.Parameter("self", inspect.Parameter.POSITIONAL_OR_KEYWORD), + inspect.Parameter( + "value", + inspect.Parameter.POSITIONAL_OR_KEYWORD, + annotation=value_type, + ), + ] + ) + write.__annotations__ = {"value": value_type} + return write + + +@dbus_interface("org.stormux.Cthulhu1.Service") +class CthulhuDBusServiceInterface(Publishable): + """Internal D-Bus service object that handles D-Bus specifics.""" + + def for_publication(self): + """Returns the D-Bus interface XML for publication.""" + + return self.__dbus_xml__ # pylint: disable=no-member + + def ShowPreferences(self) -> bool: # pylint: disable=invalid-name + """Shows Cthulhu's preferences GUI.""" + + msg = "DBUS SERVICE: ShowPreferences called." + debug.print_message(debug.LEVEL_INFO, msg, True) + + manager = script_manager.get_manager() + script = manager.get_active_script() or manager.get_default_script() + if script is None: + msg = "DBUS SERVICE: No script available" + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + + script.showPreferencesGUI() + return True + + def PresentMessage(self, message: str) -> bool: # pylint: disable=invalid-name + """Presents message to the user.""" + + msg = f"DBUS SERVICE: PresentMessage called with: '{message}'" + debug.print_message(debug.LEVEL_INFO, msg, True) + + manager = script_manager.get_manager() + script = cthulhu_state.activeScript or manager.get_default_script() + if script is None: + msg = "DBUS SERVICE: No script available" + debug.print_message(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})" + + atspi_version = Atspi.get_version() + result += f", AT-SPI2 version: {atspi_version[0]}.{atspi_version[1]}.{atspi_version[2]}" + + session_type = os.environ.get("XDG_SESSION_TYPE") or "" + session_desktop = os.environ.get("XDG_SESSION_DESKTOP") or "" + session = f"{session_type} {session_desktop}".strip() + if session: + result += f", Session: {session}" + + msg = f"DBUS SERVICE: GetVersion called, returning: {result}" + debug.print_message(debug.LEVEL_INFO, msg, True) + return result + + def Quit(self) -> bool: # pylint: disable=invalid-name + """Quits Cthulhu.""" + + msg = "DBUS SERVICE: Quit called." + debug.print_message(debug.LEVEL_INFO, msg, True) + + from . import cthulhu # pylint: disable=import-outside-toplevel + + # cthulhu.shutdown() shuts down the dbus service, so send the response immediately and then + # do the actual shutdown after a brief delay. + def _delayed_shutdown(): + cthulhu.shutdown() + return False + + GLib.timeout_add(100, _delayed_shutdown) + return True - 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" + SERVICE_NAME = "org.stormux.Cthulhu1.Service" + OBJECT_PATH = "/org/stormux/Cthulhu1/Service" def __init__(self) -> None: self._dbus_service_interface: Optional[CthulhuDBusServiceInterface] = None self._is_running: bool = False self._bus: Optional[SessionMessageBus] = None - self._event_loop: Optional[EventLoop] = None + self._registered: dict[str, _ModuleRegistration] = {} self._pending_registrations: dict[str, object] = {} - self._total_commands: int = 0 - self._total_getters: int = 0 - self._total_setters: int = 0 - self._total_modules: int = 0 - self._dasbus_available = _dasbus_available def start(self) -> bool: """Starts the D-Bus service.""" - if not self._dasbus_available: + if not _dasbus_available: msg = "REMOTE CONTROLLER: dasbus library not available, D-Bus service disabled." - debug.printMessage(debug.LEVEL_WARNING, msg, True) + debug.print_message(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) + debug.print_message(debug.LEVEL_INFO, msg, True) return True msg = "REMOTE CONTROLLER: Attempting to start D-Bus service." - debug.printMessage(debug.LEVEL_INFO, msg, True) + debug.print_message(debug.LEVEL_INFO, msg, True) try: self._bus = SessionMessageBus() @@ -611,11 +751,11 @@ class CthulhuRemoteController: f"REMOTE CONTROLLER: SessionMessageBus acquired: " f"{self._bus.connection.get_unique_name()}" ) - debug.printMessage(debug.LEVEL_INFO, msg, True) + debug.print_message(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) + debug.print_message(debug.LEVEL_SEVERE, msg, True) return False self._dbus_service_interface = CthulhuDBusServiceInterface() @@ -624,291 +764,242 @@ class CthulhuRemoteController: 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) + debug.print_message(debug.LEVEL_SEVERE, msg, True) if self._dbus_service_interface and self._bus: - try: + with contextlib.suppress(DBusError): 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}." + f"REMOTE CONTROLLER: Service started name={self.SERVICE_NAME} path={self.OBJECT_PATH}." ) - debug.printMessage(debug.LEVEL_INFO, msg, True) + debug.print_message(debug.LEVEL_INFO, msg, True) self._process_pending_registrations() self._print_registration_summary() 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 = [] - commands_count = 0 - getters_count = 0 - setters_count = 0 - - 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 - if script is None: - manager = script_manager.get_manager() - script = manager.get_default_script() - kwargs = { - "script": script, - "event": event, - "notify_user": notify_user - } - call_kwargs = _filter_kwargs_for_callable(method, kwargs) - rv = method(**call_kwargs) - _get_input_event_manager().get_manager().process_remote_controller_event(event) - return rv - return _wrapper - handler_info = _HandlerInfo( - python_function_name=attr_name, - description=description, - action=_create_wrapper(), - handler_type=HandlerType.COMMAND - ) - handlers_info.append(handler_info) - commands_count += 1 - msg = f"REMOTE CONTROLLER: Found decorated command '{attr_name}': {description}" - debug.printMessage(debug.LEVEL_INFO, msg, True) - # Parameterized Command - elif callable(attr) and hasattr(attr, "dbus_parameterized_command_description"): - description = attr.dbus_parameterized_command_description - def _create_parameterized_wrapper(method=attr): - def _wrapper(**kwargs): - event = _get_input_event().RemoteControllerEvent() - script = cthulhu_state.activeScript - if script is None: - manager = script_manager.get_manager() - script = manager.get_default_script() - merged_kwargs = { - "script": script, - "event": event - } - merged_kwargs.update(kwargs) - call_kwargs = _filter_kwargs_for_callable(method, merged_kwargs) - rv = method(**call_kwargs) - _get_input_event_manager().get_manager().process_remote_controller_event(event) - return rv - return _wrapper - handler_info = _HandlerInfo( - python_function_name=attr_name, - description=description, - action=_create_parameterized_wrapper(), - handler_type=HandlerType.PARAMETERIZED_COMMAND, - parameters=_extract_function_parameters(attr) - ) - handlers_info.append(handler_info) - commands_count += 1 - msg = f"REMOTE CONTROLLER: Found decorated parameterized 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) - getters_count += 1 - 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) - setters_count += 1 - msg = f"REMOTE CONTROLLER: Found decorated setter '{attr_name}': {description}" - debug.printMessage(debug.LEVEL_INFO, msg, True) - - if not handlers_info: - return - - self._total_commands += commands_count - self._total_getters += getters_count - self._total_setters += setters_count - self._total_modules += 1 - - 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"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) + debug.print_message(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: + debug.print_message(debug.LEVEL_INFO, msg, True) + + for module_name in list(self._registered): + self._unpublish_module(module_name) + + if self._dbus_service_interface is not None and self._bus is not None: + with contextlib.suppress(DBusError): 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: + if self._bus is not None: + with contextlib.suppress(DBusError): 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) + debug.print_message(debug.LEVEL_INFO, msg, True) self._pending_registrations.clear() - self._total_commands = 0 - self._total_getters = 0 - self._total_setters = 0 - self._total_modules = 0 - def is_running(self) -> bool: - """Checks if the D-Bus service is currently running.""" + def register_decorated_module(self, module_name: str, module_instance: object) -> None: + """Registers a module's decorated D-Bus methods, getters, and setters.""" - return self._is_running + if not self._is_running or self._bus is None: + msg = f"REMOTE CONTROLLER: Service not ready; queuing registration for {module_name}." + debug.print_message(debug.LEVEL_INFO, msg, True) + self._pending_registrations[module_name] = module_instance + return + self._publish_module(module_name, module_instance) - def _count_system_commands(self) -> int: - """Counts the system-wide D-Bus commands available on the main service interface.""" + def deregister_module_commands(self, module_name: str) -> bool: + """Deregisters a previously-registered module.""" - if not self._dbus_service_interface: - return 0 + if module_name in self._pending_registrations: + del self._pending_registrations[module_name] + msg = f"REMOTE CONTROLLER: Removed pending registration for {module_name}." + debug.print_message(debug.LEVEL_INFO, msg, True) + return True - system_commands = 0 - for attr_name in dir(self._dbus_service_interface): - if not attr_name.startswith("_") and attr_name[0].isupper(): - attr = getattr(self._dbus_service_interface, attr_name) - if callable(attr) and hasattr(attr, "__doc__"): - system_commands += 1 - return system_commands + if module_name not in self._registered: + msg = f"REMOTE CONTROLLER: Module '{module_name}' is not registered." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + + return self._unpublish_module(module_name) + + def present_message_internal(self, message: str) -> bool: + """Presents a message via speech and/or braille without a D-Bus round-trip.""" + + if self._dbus_service_interface is None: + msg = "REMOTE CONTROLLER: Cannot present message; service not started." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + + return self._dbus_service_interface.PresentMessage(message) + + def execute_command_internal( + self, + module_name: str, + command_name: str, + notify_user: bool = True, + ) -> bool: + """Executes a module command without a D-Bus round-trip.""" + + registration = self._registered.get(module_name) + if registration is None: + msg = f"REMOTE CONTROLLER: Module '{module_name}' not found." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + + method = registration.find_command(command_name) + if method is None: + msg = f"REMOTE CONTROLLER: Unknown command '{command_name}' in '{module_name}'." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + + event = _get_input_event().RemoteControllerEvent() + manager = script_manager.get_manager() + script = manager.get_active_script() or manager.get_default_script() + kwargs = {"script": script, "event": event, "notify_user": notify_user} + result = method(**_filter_kwargs_for_callable(method, kwargs)) + _get_input_event_manager().get_manager().process_remote_controller_event(event) + return bool(result) + + def get_value_internal(self, module_name: str, property_name: str) -> object: + """Gets a runtime value from a module without a D-Bus round-trip.""" + + registration = self._registered.get(module_name) + if registration is None: + msg = f"REMOTE CONTROLLER: Module '{module_name}' not found." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return None + + method = registration.find_getter(property_name) + if method is None: + msg = f"REMOTE CONTROLLER: Unknown getter '{property_name}' in '{module_name}'." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return None + + return method() + + def set_value_internal(self, module_name: str, property_name: str, value: object) -> bool: + """Sets a runtime value on a module without a D-Bus round-trip.""" + + registration = self._registered.get(module_name) + if registration is None: + msg = f"REMOTE CONTROLLER: Module '{module_name}' not found." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + + method = registration.find_setter(property_name) + if method is None: + msg = f"REMOTE CONTROLLER: Unknown setter '{property_name}' in '{module_name}'." + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + + result = method(value) + if result is None: + return True + return bool(result) + + def _publish_module(self, module_name: str, module_instance: object) -> None: + """Builds the per-module D-Bus interface and publishes it on the bus.""" + + if self._bus is None: + return + + if module_name in self._registered: + msg = f"REMOTE CONTROLLER: Module {module_name} already registered. Replacing." + debug.print_message(debug.LEVEL_INFO, msg, True) + self._unpublish_module(module_name) + + registration = _ModuleRegistration.from_module_instance(module_name, module_instance) + if registration.is_empty(): + return + + try: + interface_class = _InterfaceBuilder.build(registration) + # for_publication is supplied to the namespace, so the class isn't actually + # abstract; pylint can't statically infer that. + dbus_object = interface_class() # pylint: disable=abstract-class-instantiated + object_path = f"{self.OBJECT_PATH}/{module_name}" + self._bus.publish_object(object_path, dbus_object) + except DBusError as e: + msg = f"REMOTE CONTROLLER: Failed to publish module {module_name}: {e}" + debug.print_message(debug.LEVEL_SEVERE, msg, True) + return + + registration.set_dbus_object(dbus_object) + registration.set_object_path(object_path) + self._registered[module_name] = registration + + msg = ( + f"REMOTE CONTROLLER: Registered {registration.total_member_count()} member(s) " + f"for {module_name} at {object_path}." + ) + debug.print_message(debug.LEVEL_INFO, msg, True) + + def _unpublish_module(self, module_name: str) -> bool: + """Removes a module's D-Bus interface from the bus.""" + + registration = self._registered.get(module_name) + if registration is None or self._bus is None: + return False + try: + self._bus.unpublish_object(registration.get_object_path()) + except DBusError as e: + msg = f"REMOTE CONTROLLER: Error unpublishing {module_name}: {e}" + debug.print_message(debug.LEVEL_WARNING, msg, True) + return False + del self._registered[module_name] + msg = f"REMOTE CONTROLLER: Unpublished {module_name}." + debug.print_message(debug.LEVEL_INFO, msg, True) + return True + + def _process_pending_registrations(self) -> None: + """Publishes any modules that registered before the service was ready.""" + + if not self._pending_registrations: + return + msg = ( + f"REMOTE CONTROLLER: Processing {len(self._pending_registrations)} pending " + f"module registrations." + ) + debug.print_message(debug.LEVEL_INFO, msg, True) + for module_name, module_instance in list(self._pending_registrations.items()): + self._publish_module(module_name, module_instance) + self._pending_registrations.clear() def _print_registration_summary(self) -> None: - """Prints a summary of all registered D-Bus handlers.""" + """Logs a summary of the registered modules and their member counts.""" - system_commands_count = self._count_system_commands() - total_handlers = self._total_commands + self._total_getters + self._total_setters - msg = ( - f"REMOTE CONTROLLER: Registration complete. Summary: " - f"{self._total_modules} modules, " - f"{self._total_commands} module commands, " - f"{self._total_getters} module getters, " - f"{self._total_setters} module setters, " - f"{system_commands_count} system commands. " - f"Total handlers: {total_handlers + system_commands_count}." + modules = len(self._registered) + commands = sum( + len(reg.get_commands()) + len(reg.get_parameterized_commands()) + for reg in self._registered.values() ) - debug.printMessage(debug.LEVEL_INFO, msg, True) + getters = sum(len(reg.get_getters()) for reg in self._registered.values()) + setters = sum(len(reg.get_setters()) for reg in self._registered.values()) + msg = ( + f"REMOTE CONTROLLER: Registration summary: {modules} modules, " + f"{commands} commands, {getters} getters, {setters} setters." + ) + debug.print_message(debug.LEVEL_INFO, msg, True) + _remote_controller: CthulhuRemoteController = CthulhuRemoteController() + def get_remote_controller() -> CthulhuRemoteController: """Returns the CthulhuRemoteController singleton.""" diff --git a/src/cthulhu/plugin_system_manager.py b/src/cthulhu/plugin_system_manager.py index 0b46c88..3a1b625 100644 --- a/src/cthulhu/plugin_system_manager.py +++ b/src/cthulhu/plugin_system_manager.py @@ -669,14 +669,14 @@ class PluginSystemManager: logger.error(f"Failed to deregister D-Bus module for plugin {plugin_info.get_module_name()}: {error}") @dbus_service.command - def list_plugins(self): + def list_plugins(self) -> list[str]: """Returns a list of available plugin module names.""" if not self._plugins: self.rescanPlugins() return [info.get_module_name() for info in self.plugins] @dbus_service.command - def list_active_plugins(self): + def list_active_plugins(self) -> list[str]: """Returns a list of currently active plugin module names.""" return [info.get_module_name() for info in self.plugins if info.loaded] diff --git a/tests/test_dbus_service_native_interfaces.py b/tests/test_dbus_service_native_interfaces.py new file mode 100644 index 0000000..5c26b14 --- /dev/null +++ b/tests/test_dbus_service_native_interfaces.py @@ -0,0 +1,124 @@ +import unittest +from unittest import mock + +from cthulhu import dbus_service + + +class SampleModule: + def __init__(self): + self.enabled = False + self.calls = [] + + @dbus_service.command + def toggle_feature(self, script=None, event=None, notify_user=True): + """Toggles the test feature.""" + self.calls.append((script, event, notify_user)) + return True + + @dbus_service.command + def command_without_notify(self, script=None, event=None): + """Runs a command that does not accept notify_user.""" + self.calls.append((script, event)) + return True + + @dbus_service.command + def list_items(self) -> "list[str]": + """Returns a list from a command.""" + return ["alpha", "beta"] + + @dbus_service.parameterized_command + def speak_text( + self, + text: "str", + count: "int" = 1, + script=None, + event=None, + notify_user=True, + ) -> "str": + """Speaks text a number of times.""" + self.calls.append((text, count, script, event, notify_user)) + return text * count + + @dbus_service.getter + def get_enabled(self) -> "bool": + """Returns whether the feature is enabled.""" + return self.enabled + + @dbus_service.setter + def set_enabled(self, value: "bool"): + """Sets whether the feature is enabled.""" + self.enabled = value + + +class NativeInterfaceBuilderTest(unittest.TestCase): + def test_registration_groups_decorated_members_by_native_names(self): + registration = dbus_service._ModuleRegistration.from_module_instance( + "TestModule", SampleModule() + ) + + self.assertIn("ToggleFeature", registration.get_commands()) + self.assertIn("SpeakText", registration.get_parameterized_commands()) + self.assertIn("Enabled", registration.get_getters()) + self.assertIn("Enabled", registration.get_setters()) + self.assertEqual(registration.total_member_count(), 6) + + def test_interface_builder_exposes_methods_and_properties(self): + registration = dbus_service._ModuleRegistration.from_module_instance( + "TestModule", SampleModule() + ) + interface_class = dbus_service._InterfaceBuilder.build(registration) + + self.assertTrue(hasattr(interface_class, "ToggleFeature")) + self.assertTrue(hasattr(interface_class, "CommandWithoutNotify")) + self.assertTrue(hasattr(interface_class, "ListItems")) + self.assertTrue(hasattr(interface_class, "SpeakText")) + self.assertIsInstance(interface_class.Enabled, property) + self.assertFalse(hasattr(interface_class, "ExecuteCommand")) + self.assertIn("org.stormux.Cthulhu1.TestModule", interface_class.__dbus_xml__) + + def test_generated_command_method_uses_active_or_default_script(self): + module = SampleModule() + registration = dbus_service._ModuleRegistration.from_module_instance("TestModule", module) + interface = dbus_service._InterfaceBuilder.build(registration)() + script = mock.Mock() + manager = mock.Mock() + manager.get_active_script.return_value = None + manager.get_default_script.return_value = script + event_manager = mock.Mock() + event_manager_module = mock.Mock() + event_manager_module.get_manager.return_value = event_manager + + with ( + mock.patch.object(dbus_service.script_manager, "get_manager", return_value=manager), + mock.patch.object( + dbus_service, "_get_input_event_manager", return_value=event_manager_module + ), + ): + self.assertTrue(interface.ToggleFeature(False)) + self.assertTrue(interface.CommandWithoutNotify(False)) + self.assertEqual(interface.ListItems(False), ["alpha", "beta"]) + + self.assertEqual(module.calls[0][0], script) + self.assertFalse(module.calls[0][2]) + self.assertEqual(module.calls[1][0], script) + self.assertEqual(event_manager.process_remote_controller_event.call_count, 3) + + def test_generated_properties_call_getters_and_setters(self): + module = SampleModule() + registration = dbus_service._ModuleRegistration.from_module_instance("TestModule", module) + interface = dbus_service._InterfaceBuilder.build(registration)() + + self.assertFalse(interface.Enabled) + interface.Enabled = True + self.assertTrue(module.enabled) + self.assertTrue(interface.Enabled) + + +class NativeRemoteControllerTest(unittest.TestCase): + def test_controller_uses_versioned_service_name_and_object_path(self): + self.assertEqual(dbus_service.CthulhuRemoteController.SERVICE_NAME, "org.stormux.Cthulhu1.Service") + self.assertEqual(dbus_service.CthulhuRemoteController.OBJECT_PATH, "/org/stormux/Cthulhu1/Service") + + +if __name__ == "__main__": + unittest.main()