feat(dbus): expose dictation state signal and property
Alongside Toggle/Start/Stop, the renamed interface now publishes dictation state (idle/listening/processing) as a StateChanged signal and a readable State property, so status bars and scripts can follow dictation without scraping the tray. Signals fire only on change, so double start/stop emits no duplicates; command-line triggers behave exactly as before. Refs #21
This commit is contained in:
@@ -15,9 +15,34 @@ from typing import Callable, Optional
|
||||
from gi.repository import Gio, GLib
|
||||
|
||||
from .app_identity import BUS_NAME, INTERFACE_NAME, OBJECT_PATH
|
||||
from .common_types import RecognitionState
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: Dictation states exposed on the bus. ``ERROR`` is not dictation activity:
|
||||
#: the engine is idle again once it has failed, so it maps to ``idle``.
|
||||
DICTATION_STATE_IDLE = "idle"
|
||||
DICTATION_STATE_LISTENING = "listening"
|
||||
DICTATION_STATE_PROCESSING = "processing"
|
||||
DICTATION_STATES: tuple[str, ...] = (
|
||||
DICTATION_STATE_IDLE,
|
||||
DICTATION_STATE_LISTENING,
|
||||
DICTATION_STATE_PROCESSING,
|
||||
)
|
||||
STATE_CHANGED_SIGNAL = "StateChanged"
|
||||
STATE_PROPERTY = "State"
|
||||
PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"
|
||||
|
||||
|
||||
def state_from_recognition(state: RecognitionState) -> str:
|
||||
"""Map an engine state to the bus-visible dictation state."""
|
||||
if state == RecognitionState.LISTENING:
|
||||
return DICTATION_STATE_LISTENING
|
||||
if state == RecognitionState.PROCESSING:
|
||||
return DICTATION_STATE_PROCESSING
|
||||
return DICTATION_STATE_IDLE
|
||||
|
||||
|
||||
# Maps CLI trigger names to the D-Bus method names exposed below.
|
||||
COMMAND_METHODS = {
|
||||
"toggle": "Toggle",
|
||||
@@ -31,6 +56,10 @@ _INTROSPECTION_XML = f"""
|
||||
<method name="Toggle"/>
|
||||
<method name="Start"/>
|
||||
<method name="Stop"/>
|
||||
<signal name="{STATE_CHANGED_SIGNAL}">
|
||||
<arg name="state" type="s" direction="out"/>
|
||||
</signal>
|
||||
<property name="{STATE_PROPERTY}" type="s" access="read"/>
|
||||
</interface>
|
||||
</node>
|
||||
"""
|
||||
@@ -70,9 +99,37 @@ class VocalinuxDBusService:
|
||||
self._owner_id = 0
|
||||
self._registration_id = 0
|
||||
self._connection: Optional[Gio.DBusConnection] = None
|
||||
self._dictation_state = DICTATION_STATE_IDLE
|
||||
self._node_info = Gio.DBusNodeInfo.new_for_xml(_INTROSPECTION_XML)
|
||||
self._register()
|
||||
|
||||
def get_state(self) -> str:
|
||||
"""Return the current bus-visible dictation state (``idle`` at start)."""
|
||||
return self._dictation_state
|
||||
|
||||
def set_state(self, state: str) -> bool:
|
||||
"""Publish a dictation state, emitting the signal only on change.
|
||||
|
||||
Unknown values are ignored. Returns True when subscribers were
|
||||
notified, False when the state was unchanged (or unknown), which is
|
||||
what keeps double start/stop from emitting duplicate transitions.
|
||||
"""
|
||||
if state not in DICTATION_STATES or state == self._dictation_state:
|
||||
return False
|
||||
self._dictation_state = state
|
||||
if self._connection is not None:
|
||||
try:
|
||||
self._connection.emit_signal(
|
||||
None,
|
||||
OBJECT_PATH,
|
||||
INTERFACE_NAME,
|
||||
STATE_CHANGED_SIGNAL,
|
||||
GLib.Variant("(s)", (state,)),
|
||||
)
|
||||
except GLib.Error:
|
||||
logger.debug("Failed to emit dictation state signal", exc_info=True)
|
||||
return True
|
||||
|
||||
def _register(self) -> None:
|
||||
"""Request ownership of the bus name on the session bus."""
|
||||
try:
|
||||
@@ -131,6 +188,9 @@ class VocalinuxDBusService:
|
||||
invocation: Gio.DBusMethodInvocation,
|
||||
) -> None:
|
||||
"""Dispatch an incoming method call to the matching callback."""
|
||||
if interface_name == PROPERTIES_INTERFACE:
|
||||
self._handle_properties_call(method_name, parameters, invocation)
|
||||
return
|
||||
callback = self._callbacks.get(method_name)
|
||||
if callback is None:
|
||||
invocation.return_error_literal(
|
||||
@@ -145,6 +205,33 @@ class VocalinuxDBusService:
|
||||
GLib.idle_add(self._invoke, callback)
|
||||
invocation.return_value(None)
|
||||
|
||||
def _handle_properties_call(
|
||||
self,
|
||||
method_name: str,
|
||||
parameters: GLib.Variant,
|
||||
invocation: Gio.DBusMethodInvocation,
|
||||
) -> None:
|
||||
"""Serve reads of the ``State`` property."""
|
||||
if method_name == "Get":
|
||||
interface, name = parameters.unpack()
|
||||
if interface == INTERFACE_NAME and name == STATE_PROPERTY:
|
||||
invocation.return_value(
|
||||
GLib.Variant("(v)", (GLib.Variant("s", self._dictation_state),))
|
||||
)
|
||||
return
|
||||
elif method_name == "GetAll":
|
||||
(interface,) = parameters.unpack()
|
||||
if interface == INTERFACE_NAME:
|
||||
invocation.return_value(
|
||||
GLib.Variant("(a{sv})", ({"State": GLib.Variant("s", self._dictation_state)},))
|
||||
)
|
||||
return
|
||||
invocation.return_error_literal(
|
||||
Gio.dbus_error_quark(),
|
||||
Gio.DBusError.UNKNOWN_PROPERTY,
|
||||
f"Unknown property call: {method_name}",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _invoke(callback: Callable[[], None]) -> bool:
|
||||
try:
|
||||
|
||||
@@ -921,6 +921,18 @@ class TrayIndicator:
|
||||
else:
|
||||
keepalive.cancel()
|
||||
|
||||
# Publish dictation state on D-Bus for status bars and scripts
|
||||
# (ticket #21). The service may not exist yet when the first state
|
||||
# callback fires during __init__, hence the guarded lookup.
|
||||
from ..dbus_service import state_from_recognition
|
||||
|
||||
dbus_service = getattr(self, "_dbus_service", None)
|
||||
if dbus_service is not None:
|
||||
try:
|
||||
dbus_service.set_state(state_from_recognition(state))
|
||||
except Exception:
|
||||
logger.debug("Failed to publish dictation state", exc_info=True)
|
||||
|
||||
# Update the UI in the GTK main thread
|
||||
GLib.idle_add(self._update_ui, state)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user