Mercurial > libervia-backend
view libervia/backend/bridge/bridge_constructor/constructors/dbus/dbus_core_template.py @ 4192:1d24ff583794
plugin forums: parsing fix + formatting:
- reformatted with black
- fix parsing of forums
- minor improvements
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 12 Dec 2023 12:17:15 +0100 |
parents | 4b842c1fb686 |
children | 0d7bb4df2343 |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia communication bridge # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from types import MethodType from functools import partialmethod from twisted.internet import defer, reactor from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.core.exceptions import BridgeInitError from libervia.backend.tools import config from txdbus import client, objects, error from txdbus.interface import DBusInterface, Method, Signal log = getLogger(__name__) # Interface prefix const_INT_PREFIX = config.config_get( config.parse_main_conf(), "", "bridge_dbus_int_prefix", "org.libervia.Libervia") const_ERROR_PREFIX = const_INT_PREFIX + ".error" const_OBJ_PATH = "/org/libervia/Libervia/bridge" const_CORE_SUFFIX = ".core" const_PLUGIN_SUFFIX = ".plugin" class ParseError(Exception): pass class DBusException(Exception): pass class MethodNotRegistered(DBusException): dbusErrorName = const_ERROR_PREFIX + ".MethodNotRegistered" class GenericException(DBusException): def __init__(self, twisted_error): """ @param twisted_error (Failure): instance of twisted Failure error message is used to store a repr of message and condition in a tuple, so it can be evaluated by the frontend bridge. """ try: # twisted_error.value is a class class_ = twisted_error.value().__class__ except TypeError: # twisted_error.value is an instance class_ = twisted_error.value.__class__ data = twisted_error.getErrorMessage() try: data = (data, twisted_error.value.condition) except AttributeError: data = (data,) else: data = (str(twisted_error),) self.dbusErrorName = ".".join( (const_ERROR_PREFIX, class_.__module__, class_.__name__) ) super(GenericException, self).__init__(repr(data)) @classmethod def create_and_raise(cls, exc): raise cls(exc) class DBusObject(objects.DBusObject): core_iface = DBusInterface( const_INT_PREFIX + const_CORE_SUFFIX, ##METHODS_DECLARATIONS_PART## ##SIGNALS_DECLARATIONS_PART## ) plugin_iface = DBusInterface( const_INT_PREFIX + const_PLUGIN_SUFFIX ) dbusInterfaces = [core_iface, plugin_iface] def __init__(self, path): super().__init__(path) log.debug("Init DBusObject...") self.cb = {} def register_method(self, name, cb): self.cb[name] = cb def _callback(self, name, *args, **kwargs): """Call the callback if it exists, raise an exception else""" try: cb = self.cb[name] except KeyError: raise MethodNotRegistered else: d = defer.maybeDeferred(cb, *args, **kwargs) d.addErrback(GenericException.create_and_raise) return d ##METHODS_PART## class bridge: def __init__(self): log.info("Init DBus...") self._obj = DBusObject(const_OBJ_PATH) async def post_init(self): try: conn = await client.connect(reactor) except error.DBusException as e: if e.errName == "org.freedesktop.DBus.Error.NotSupported": log.error( _( "D-Bus is not launched, please see README to see instructions on " "how to launch it" ) ) raise BridgeInitError(str(e)) conn.exportObject(self._obj) await conn.requestBusName(const_INT_PREFIX) ##SIGNALS_PART## def register_method(self, name, callback): log.debug(f"registering DBus bridge method [{name}]") self._obj.register_method(name, callback) def emit_signal(self, name, *args): self._obj.emitSignal(name, *args) def add_method( self, name, int_suffix, in_sign, out_sign, method, async_=False, doc={} ): """Dynamically add a method to D-Bus bridge""" # FIXME: doc parameter is kept only temporary, the time to remove it from calls log.debug(f"Adding method {name!r} to D-Bus bridge") self._obj.plugin_iface.addMethod( Method(name, arguments=in_sign, returns=out_sign) ) # we have to create a method here instead of using partialmethod, because txdbus # uses __func__ which doesn't work with partialmethod def caller(self_, *args, **kwargs): return self_._callback(name, *args, **kwargs) setattr(self._obj, f"dbus_{name}", MethodType(caller, self._obj)) self.register_method(name, method) def add_signal(self, name, int_suffix, signature, doc={}): """Dynamically add a signal to D-Bus bridge""" log.debug(f"Adding signal {name!r} to D-Bus bridge") self._obj.plugin_iface.addSignal(Signal(name, signature)) setattr(bridge, name, partialmethod(bridge.emit_signal, name))