From 6901d05301af145c69f2f914195c3a13ca8b82a3 Mon Sep 17 00:00:00 2001 From: Neui Date: Sat, 10 Oct 2020 01:08:52 +0200 Subject: [PATCH] Add class to easily implement DBus Interfaces In GLib, there is an Client Proxy you can use to access signals and properties from an DBus interface, but nothing similar for the Server-Side without generating code. So this lays out an class implementers subclass, specify some metadata and implement the methods that the interfaces uses. --- GTG/core/dbus/__init__.py | 0 GTG/core/dbus/dbus.py | 176 ++++++++++++++++++++++++++++++++++++++ GTG/core/meson.build | 6 ++ 3 files changed, 182 insertions(+) create mode 100644 GTG/core/dbus/__init__.py create mode 100644 GTG/core/dbus/dbus.py diff --git a/GTG/core/dbus/__init__.py b/GTG/core/dbus/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/GTG/core/dbus/dbus.py b/GTG/core/dbus/dbus.py new file mode 100644 index 0000000000..9832178c1f --- /dev/null +++ b/GTG/core/dbus/dbus.py @@ -0,0 +1,176 @@ +from gi.repository import GLib, Gio +import logging +import os +import traceback + + +log = logging.getLogger(__name__) + + +def _get_installed_xml_from_interface_name(interface_name: str): + """Return an DBus interface XML from system installed folders""" + raw_paths = GLib.get_system_data_dirs() + raw_paths.append(GLib.get_user_data_dir()) + for raw_path in raw_paths: + full_path = os.path.join(raw_path, 'dbus-1', 'interfaces', + interface_name + ".xml") + try: + with open(full_path, 'rt') as f: + log.debug(f"Found file {full_path}") + return f.read() + except OSError: + pass + return None + + +def _get_internal_xml_from_interface_name(interface_name: str): + """Return an DBus interface XML from internal data folders""" + log.debug("TODO: Implement this later") + pass + + +def get_xml_from_interface_name(interface_name: str, + use_internal: bool = True, + use_system: bool = True): + """ + Return an DBus interface XML from either provided by the system (distro) + or internal. + """ + if use_internal: + xml = _get_internal_xml_from_interface_name(interface_name) + if xml is not None: + return xml + if use_system: + xml = _get_installed_xml_from_interface_name(interface_name) + if xml is not None: + return xml + return None + + +class DBusReturnError(Exception): + """ + An error that an DBus interface implementation can throw to indicate + that it should return with that error. + """ + def __init__(self, name, message): + super().__init__(message) + self.name = str(name) + self.message = str(message) + + +class DBusInterfaceImplService(): + INTERFACE_NAME = None + NODE_INFO = None + INTERFACE_INFO = None + _object_id = None + _dbus_connection = None + + def __init__(self, wrapped_object=None): + if wrapped_object is None: + wrapped_object = self + self._wrapped_object = wrapped_object + + def __get_interface_info_from_node_info(self): + for interface in self.NODE_INFO.interfaces: + if interface.name == self.INTERFACE_NAME: + return interface + return None + + def __get_node_info(self): + xml = get_xml_from_interface_name(self.INTERFACE_NAME) + if xml is not None: + return Gio.DBusNodeInfo.new_for_xml(xml) + return None + + def _get_info(self): + if self.NODE_INFO is None and self.INTERFACE_NAME is not None: + self.NODE_INFO = self.__get_node_info() + log.debug(f"Got node info for %r: %r", + self.INTERFACE_NAME, self.NODE_INFO) + if self.NODE_INFO is not None and self.INTERFACE_INFO is None: + self.INTERFACE_INFO = self.__get_interface_info_from_node_info() + log.debug(f"Got interface info for %r: %r", + self.INTERFACE_NAME, self.INTERFACE_INFO) + + def dbus_register(self, dbus_connection, object_path): + """ + Register this implementation on the specified connection and object. + """ + self._get_info() + self._object_id = dbus_connection.register_object( + object_path, + self.INTERFACE_INFO, + self._handle_method_call, + self._handle_get_property, self._handle_set_property) + self._dbus_connection = dbus_connection + return self._object_id + + def dbus_unregister(self): + """Unregister this implementation.""" + if self._dbus_connection is not None and self._object_id is not None: + log.debug("Unregister %r on %r via %r", + self.INTERFACE_NAME, + self._dbus_connection, + self._object_id) + ret = self._dbus_connection.unregister_object(self._object_id) + self._dbus_connection = None + self._object_id = None + if not ret: + log.warn("Unregister for %r failed!", self.INTERFACE_NAME) + return ret + else: + log.warn("Trying to unregister not-registered %r", + self.INTERFACE_NAME) + return False + + def _handle_method_call(self, connection, + sender, object_path, + interface_name, method_name, + parameters, invocation): + dbg = "%s.%s%s on %s by %s" + dbg_args = (interface_name, method_name, parameters, + object_path, sender) + log.debug(f"Called {dbg}", *dbg_args) + + try: + method = self._wrapped_object.__getattribute__(method_name) + except AttributeError as e: + log.debug(f"{dbg} → Internal python exception: %s", *dbg_args, e) + invocation.return_dbus_error("python." + type(e).__name__, str(e)) + return + + try: + ret = method(*parameters.unpack()) + except GLib.Error as e: + log.debug(f"{dbg} → GLib Error: %s", *dbg_args, e) + invocation.return_gerror(e) + except DBusReturnError as e: + log.debug(f"{dbg} → Custom Error: %s: %s", + *dbg_args, e.name, e.message) + invocation.return_dbus_error(e.name, e.message) + except Exception as e: + log.warn(f"{dbg} → Python exception: %s", *dbg_args, e) + traceback.print_exc() + invocation.return_dbus_error("python." + type(e).__name__, str(e)) + else: + if type(ret) != GLib.Variant: + if len(invocation.get_method_info().out_args) == 1: + ret = (ret,) # Make one-value return work as expected + ret = GLib.Variant(self.__get_return_variant(invocation), ret) + log.debug(f"{dbg} → Returning %r", *dbg_args, ret) + invocation.return_value(ret) + + def _handle_get_property(self, *args): + """Get a DBus-available property""" + log.debug(f"TODO: Handle get property: {args}") # Not used currently + return None + + def _handle_set_property(self, *args): + """Set a DBus-available property""" + log.debug(f"TODO: Handle set property: {args}") # Not used currently + return None + + def __get_return_variant(self, invocation): + """Get Variant Type string that should be returned for a DBus method""" + info = invocation.get_method_info() + return '(' + "".join([arg.signature for arg in info.out_args]) + ')' diff --git a/GTG/core/meson.build b/GTG/core/meson.build index 3552e9e6f5..935de05695 100644 --- a/GTG/core/meson.build +++ b/GTG/core/meson.build @@ -52,5 +52,11 @@ gtg_core_plugin_sources = [ 'plugins/engine.py', ] +gtg_core_dbus_sources = [ + 'dbus/__init__.py', + 'dbus/dbus.py', +] + python3.install_sources(gtg_core_sources, subdir: 'GTG' / 'core', pure: true) python3.install_sources(gtg_core_plugin_sources, subdir: 'GTG' / 'core' / 'plugins', pure: true) +python3.install_sources(gtg_core_dbus_sources, subdir: 'GTG' / 'core' / 'dbus', pure: true)