blacklight/platypush
1
0
Fork 0
forked from platypush/platypush
Code Issues Pull requests Projects Releases Packages Wiki Activity

Refactored D-Bus integration

Browse source 
- Added ability to listen for signals
- Improved introspection output
- `dbus` plugin and backend have now been merged
- Migrated from `dbus` to `pydbus`
This commit is contained in:
Fabio Manganiello 2022-02-07 15:45:19 +01:00
parent 1914322fda
commit 786286eac6
Signed by: blacklight
GPG key ID: D90FBA7F76362774
13 changed files with 471 additions and 166 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -19,3 +19,6 @@ platypush/requests
/http-client.env.json
/platypush/backend/http/static/css/dist
/tests/etc/dashboards
.coverage
coverage.xml
.vimsessions

1
docs/source/backends.rst
View file

@ -19,7 +19,6 @@ Backends
platypush/backend/chat.telegram.rst
platypush/backend/clipboard.rst
platypush/backend/covid19.rst
platypush/backend/dbus.rst
platypush/backend/file.monitor.rst
platypush/backend/foursquare.rst
platypush/backend/github.rst

3
docs/source/conf.py
View file

@ -52,6 +52,7 @@ extensions = [
'sphinx.ext.githubpages',
'sphinx_rtd_theme',
'sphinx_marshmallow',
'defusedxml',
]
# Add any paths that contain templates here, relative to this directory.
@ -216,7 +217,7 @@ autodoc_mock_imports = ['googlesamples.assistant.grpc.audio_helpers',
'gevent.wsgi',
'Adafruit_IO',
'pyperclip',
'dbus',
'pydbus',
'inputs',
'inotify',
'omxplayer',

5
docs/source/platypush/backend/dbus.rst
View file

@ -1,5 +0,0 @@
``dbus``
==========================
.. automodule:: platypush.backend.dbus
:members:

16
platypush/backend/__init__.py
View file

@ -99,6 +99,7 @@ class Backend(Thread, EventGenerator, ExtensionWithManifest):
if self._is_expected_response(msg):
# Expected response, trigger the response handler
clear_timeout()
# pylint: disable=unsubscriptable-object
self._request_context['on_response'](msg)
self.stop()
return
@ -110,12 +111,13 @@ class Backend(Thread, EventGenerator, ExtensionWithManifest):
""" Internal only - returns true if we are expecting for a response
and msg is that response """
# pylint: disable=unsubscriptable-object
return self._request_context \
and isinstance(msg, Response) \
and msg.id == self._request_context['request'].id
def _get_backend_config(self):
config_name = 'backend.' + self.__class__.__name__.split('Backend')[0].lower()
config_name = 'backend.' + self.__class__.__name__.split('Backend', maxsplit=1)[0].lower()
return Config.get(config_name)
def _setup_response_handler(self, request, on_response, response_timeout):
@ -196,7 +198,7 @@ class Backend(Thread, EventGenerator, ExtensionWithManifest):
self.send_message(response, **kwargs)
def send_message(self, msg, queue_name=None, **kwargs):
def send_message(self, msg, queue_name=None, **_):
"""
Sends a platypush.message.Message to a node.
To be implemented in the derived classes. By default, if the Redis
@ -213,8 +215,10 @@ class Backend(Thread, EventGenerator, ExtensionWithManifest):
if not redis:
raise KeyError()
except KeyError:
self.logger.warning("Backend {} does not implement send_message " +
"and the fallback Redis backend isn't configured")
self.logger.warning((
"Backend {} does not implement send_message " +
"and the fallback Redis backend isn't configured"
).format(self.__class__.__name__))
return
redis.send_message(msg, queue_name=queue_name)
@ -233,6 +237,7 @@ class Backend(Thread, EventGenerator, ExtensionWithManifest):
while not self.should_stop() and not has_error:
try:
# pylint: disable=not-callable
self.loop()
except Exception as e:
has_error = True
@ -259,7 +264,6 @@ class Backend(Thread, EventGenerator, ExtensionWithManifest):
def on_stop(self):
""" Callback invoked when the process stops """
pass
def stop(self):
""" Stops the backend thread by sending a STOP event on its bus """
@ -281,7 +285,7 @@ class Backend(Thread, EventGenerator, ExtensionWithManifest):
redis_backend = get_backend('redis')
if not redis_backend:
self.logger.warning('Redis backend not configured - some ' +
self.logger.warning('Redis backend not configured - some '
'web server features may not be working properly')
redis_args = {}
else:

86
platypush/backend/dbus/__init__.py
View file

@ -1,86 +0,0 @@
from typing import Union
# noinspection PyPackageRequirements,PyUnresolvedReferences
from gi.repository import GLib
import dbus
import dbus.service
import dbus.mainloop.glib
from platypush.backend import Backend
from platypush.context import get_bus
from platypush.message import Message
from platypush.message.event import Event
from platypush.message.request import Request
from platypush.utils import run
# noinspection PyPep8Naming
class DBusService(dbus.service.Object):
@classmethod
def _parse_msg(cls, msg: Union[dict, list]):
import json
return Message.build(json.loads(json.dumps(msg)))
@dbus.service.method('org.platypush.MessageBusInterface', in_signature='a{sv}', out_signature='v')
def Post(self, msg: dict):
"""
This method accepts a message as a dictionary (either representing a valid request or an event) and either
executes it (request) or forwards it to the application bus (event).
:param msg: Request or event, as a dictionary.
:return: The return value of the request, or 0 if the message is an event.
"""
msg = self._parse_msg(msg)
if isinstance(msg, Request):
ret = run(msg.action, **msg.args)
if ret is None:
ret = '' # DBus doesn't like None return types
return ret
elif isinstance(msg, Event):
get_bus().post(msg)
return 0
class DbusBackend(Backend):
"""
This backend acts as a proxy that receives messages (requests or events) on the DBus and forwards them to the
application bus.
The name of the messaging interface exposed by Platypush is ``org.platypush.MessageBusInterface`` and it exposes
``Post`` method, which accepts a dictionary representing a valid Platypush message (either a request or an event)
and either executes it or forwards it to the application bus.
Requires:
* **dbus-python** (``pip install dbus-python``)
"""
def __init__(self, bus_name='org.platypush.Bus', service_path='/MessageService', *args, **kwargs):
"""
:param bus_name: Name of the bus where the application will listen for incoming messages (default:
``org.platypush.Bus``).
:param service_path: Path to the service exposed by the app (default: ``/MessageService``).
"""
super().__init__(*args, **kwargs)
self.bus_name = bus_name
self.service_path = service_path
def run(self):
super().run()
# noinspection PyUnresolvedReferences
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
name = dbus.service.BusName(self.bus_name, bus)
srv = DBusService(bus, self.service_path)
loop = GLib.MainLoop()
# noinspection PyProtectedMember
self.logger.info('Starting DBus main loop - bus name: {}, service: {}'.format(name._name, srv._object_path))
loop.run()
# vim:sw=4:ts=4:et:

7
platypush/backend/dbus/manifest.yaml
View file

@ -1,7 +0,0 @@
manifest:
events: {}
install:
pip:
- dbus-python
package: platypush.backend.dbus
type: backend

18
platypush/message/event/__init__.py
View file

@ -1,8 +1,10 @@
import copy
import hashlib
import json
import random
import re
import time
import uuid
from datetime import date
@ -18,6 +20,7 @@ class Event(Message):
# will be disabled. Logging is usually disabled for events with a very
# high frequency that would otherwise pollute the logs e.g. camera capture
# events
# pylint: disable=redefined-builtin
def __init__(self, target=None, origin=None, id=None, timestamp=None,
disable_logging=False, disable_web_clients_notification=False, **kwargs):
"""
@ -63,10 +66,7 @@ class Event(Message):
@staticmethod
def _generate_id():
""" Generate a unique event ID """
id = ''
for i in range(0, 16):
id += '%.2x' % random.randint(0, 255)
return id
return hashlib.md5(str(uuid.uuid1()).encode()).hexdigest()
def matches_condition(self, condition):
"""
@ -205,13 +205,13 @@ def flatten(args):
for (key, value) in args.items():
if isinstance(value, date):
args[key] = value.isoformat()
elif isinstance(value, dict) or isinstance(value, list):
elif isinstance(value, (dict, list)):
flatten(args[key])
elif isinstance(args, list):
for i in range(0, len(args)):
if isinstance(args[i], date):
args[i] = args[i].isoformat()
elif isinstance(args[i], dict) or isinstance(args[i], list):
for i, arg in enumerate(args):
if isinstance(arg, date):
args[i] = arg.isoformat()
elif isinstance(arg, (dict, list)):
flatten(args[i])
# vim:sw=4:ts=4:et:

23
platypush/message/event/dbus.py Normal file
View file

@ -0,0 +1,23 @@
from typing import Optional, Iterable, Any
from platypush.message.event import Event
class DbusSignalEvent(Event):
"""
Event triggered when a signal is received on the D-Bus.
"""
def __init__(
self, bus: str, interface: str, sender: str, path: str, signal: str,
params: Optional[Iterable[Any]] = None, **kwargs
):
"""
:param bus: Bus type (``session`` or ``system``).
:param interface: Name of the interface associated to the signal.
:param sender: D-Bus name of the sender of the signal.
:param path: Path of the object associated to the signal.
:param signal: Signal name.
:param params: Signal payload.
"""
super().__init__(bus=bus, interface=interface, sender=sender,
path=path, signal=signal, params=params, **kwargs)

441
platypush/plugins/dbus/__init__.py
View file

@ -1,11 +1,25 @@
import enum
import json
from typing import Set, Dict, Optional
from xml.etree import ElementTree
from typing import Set, Dict, Optional, Iterable, Callable, Union
import dbus
from gi.repository import GLib # type: ignore
from pydbus import SessionBus, SystemBus
from pydbus.bus import Bus
from defusedxml import ElementTree
from platypush.plugins import Plugin, action
from platypush.context import get_bus
from platypush.message import Message
from platypush.message.event import Event
from platypush.message.event.dbus import DbusSignalEvent
from platypush.message.request import Request
from platypush.plugins import RunnablePlugin, action
from platypush.schemas.dbus import DbusSignalSchema
from platypush.utils import run
_default_service_name = 'org.platypush.Bus'
_default_service_path = '/'
_default_interface_name = 'org.platypush.Bus'
class BusType(enum.Enum):
@ -13,31 +27,197 @@ class BusType(enum.Enum):
SESSION = 'session'
class DbusPlugin(Plugin):
class DBusService():
"""
<node>
<interface name="org.platypush.Bus">
<method name="Post">
<arg type="s" name="msg" direction="in"/>
<arg type="s" name="response" direction="out"/>
</method>
</interface>
</node>
"""
@classmethod
def _parse_msg(cls, msg: Union[str, dict]) -> dict:
return Message.build(json.loads(json.dumps(msg)))
def Post(self, msg: dict):
"""
This method accepts a message as a JSON object
(either representing a valid request or an event) and either
executes it (request) or forwards it to the application bus (event).
:param msg: Request or event, as a dictionary.
:return: The return value of the request, or 0 if the message is an event.
"""
msg = self._parse_msg(msg)
if isinstance(msg, Request):
ret = run(msg.action, **msg.args)
if ret is None:
ret = '' # DBus doesn't like None return types
return ret
if isinstance(msg, Event):
get_bus().post(msg)
return 0
class DbusPlugin(RunnablePlugin):
"""
Plugin to interact with DBus.
This plugin can be used for the following tasks:
* It can expose a D-Bus interface that other applications can use to push messages
to Platypush (either action requests or events) serialized in JSON format.
You can disable this listener by setting ``service_name`` to ``null`` in your
configuration. If the D-Bus Platypush interface is enabled then you can push
Platypush events and requests in JSON format from another application or script
by specifying:
* The D-Bus service (default: ``org.platypush.Bus``)
* The D-Bus interface (default: ``org.platypush.Bus``)
* The D-Bus path (default: ``/``)
* The D-Bus method (``Post``)
* The Platypush JSON payload (first argument of the request). Format:
``{"type": "request", "action": "module.action", "args": {...}}``
* It can subscribe to multiple D-Bus signals, and it triggers a ``DbusSignalEvent``
when an event is received (signal filters should be specified in the ``signals``
configuration).
* It can be used to query and inspect D-Bus objects through the :meth:`.query` method.
* It can be used to execute methods exponsed by D-Bus objects through the
:meth:`.execute` method.
Requires:
* **dbus-python** (``pip install dbus-python``)
* **pydbus** (``pip install pydbus``)
* **defusedxml** (``pip install defusedxml``)
Triggers:
* :class:`platypush.message.event.dbus.DbusSignalEvent` when a signal is received.
"""
def __init__(self, **kwargs):
def __init__(
self, signals: Optional[Iterable[dict]] = None,
service_name: Optional[str] = _default_service_name,
service_path: Optional[str] = _default_service_path, **kwargs
):
"""
:param signals: Specify this if you want to subscribe to specific DBus
signals. Structure:
.. schema:: dbus.DbusSignalSchema(many=True)
For example, to subscribe to all the messages on the session bus:
.. code-block:: yaml
dbus:
signals:
- bus: session
:param service_name: Name of the D-Bus service where Platypush will listen
for new messages (requests and events). Set to null if you want to disable
message execution over D-Bus for Platypush (default: ``org.platypush.Bus``).
:param service_path: The path of the D-Bus message listener. Set to null
if you want to disable message execution over D-Bus for Platypush
(default: ``/``).
"""
super().__init__(**kwargs)
self._system_bus = SystemBus()
self._session_bus = SessionBus()
self._loop = None
self._signals = DbusSignalSchema().load(signals or [], many=True)
self._signal_handlers = [
self._get_signal_handler(**signal)
for signal in self._signals
]
self.service_name = service_name
self.service_path = service_path
@staticmethod
def _get_bus_names(bus: dbus.Bus) -> Set[str]:
return set([str(name) for name in bus.list_names() if not name.startswith(':')])
def _get_signal_handler(bus: str, **_) -> Callable:
def handler(sender, path, interface, signal, params):
get_bus().post(
DbusSignalEvent(
bus=bus, signal=signal, path=path,
interface=interface, sender=sender, params=params
)
)
@classmethod
def path_names(cls, bus: dbus.Bus, service: str, object_path='/', paths=None, service_dict=None):
if not paths:
return handler
def _get_bus(self, bus_type: Union[str, BusType]) -> Bus:
if isinstance(bus_type, str):
bus_type = BusType(bus_type.lower())
return self._system_bus if bus_type == BusType.SYSTEM else self._session_bus
def _init_signal_listeners(self):
for i, signal in enumerate(self._signals):
handler = self._signal_handlers[i]
bus = self._get_bus(signal['bus'])
bus.subscribe(
signal_fired=handler,
signal=signal.get('signal'),
sender=signal.get('sender'),
object=signal.get('path'),
iface=signal.get('interface'),
)
def _init_service(self):
if not (self.service_name and self.service_path):
return
self._session_bus.publish(
self.service_name,
('/', DBusService()),
)
def main(self):
self._init_signal_listeners()
self._init_service()
self._loop = GLib.MainLoop()
self._loop.run()
def stop(self):
self._should_stop.set()
if self._loop:
self._loop.quit()
self._loop = None
self.logger.info('Stopped D-Bus main loop')
@staticmethod
def _get_bus_names(bus: Bus) -> Set[str]:
return {str(name) for name in bus.dbus.ListNames() if not name.startswith(':')}
def path_names(self, bus: Bus, service: str, object_path='/', paths=None, service_dict=None):
if paths is None:
paths = {}
if service_dict is None:
service_dict = {}
paths[object_path] = {}
obj = bus.get_object(service, object_path)
interface = dbus.Interface(obj, 'org.freedesktop.DBus.Introspectable')
try:
obj = bus.get(service, object_path)
interface = obj['org.freedesktop.DBus.Introspectable']
except GLib.GError as e:
self.logger.warning(f'Could not inspect D-Bus object {service}, path={object_path}: {e}')
return {}
except KeyError as e:
self.logger.warning(f'Could not get interfaces on the D-Bus object {service}, path={object_path}: {e}')
return {}
xml_string = interface.Introspect()
root = ElementTree.fromstring(xml_string)
@ -46,44 +226,190 @@ class DbusPlugin(Plugin):
if object_path == '/':
object_path = ''
new_path = '/'.join((object_path, child.attrib['name']))
cls.path_names(bus, service, new_path, paths)
self.path_names(bus, service, new_path, paths, service_dict=service_dict)
else:
if not object_path:
object_path = '/'
function_dict = {}
functions_dict = {}
for func in list(child):
if func.tag not in function_dict.keys():
function_dict[func.tag] = []
function_dict[func.tag].append(func.attrib['name'])
function_dict = {'name': func.attrib['name']}
for arg in list(func):
if arg.tag != 'arg':
continue
if function_dict:
paths[object_path][child.attrib['name']] = function_dict
function_dict['args'] = function_dict.get('args', [])
function_dict['args'].append(arg.attrib)
if func.tag not in functions_dict:
functions_dict[func.tag] = []
functions_dict[func.tag].append(function_dict)
if functions_dict:
paths[object_path][child.attrib['name']] = functions_dict
if not service_dict:
service_dict = {}
if paths:
service_dict[service] = paths
return service_dict
@action
def query(self, service: Optional[str] = None, system_bus: bool = True, session_bus: bool = True) \
def query(self, service: Optional[str] = None, bus=tuple(t.value for t in BusType)) \
-> Dict[str, dict]:
"""
Query DBus for a specific service or for the full list of services.
:param service: Service name (default: None, query all services).
:param system_bus: Query the system bus (default: True).
:param session_bus: Query the session bus (default: True).
:return: A ``{service_name -> {properties}}`` mapping.
:param bus: Which bus(ses) should be queried (default: both ``system`` and ``session``).
:return: A ``{service_name -> {properties}}`` mapping. Example:
.. code-block:: json
"session": {
"org.platypush.Bus": {
"/": {
"org.freedesktop.DBus.Properties": {
"method": [
{
"name": "Get",
"args": [
{
"type": "s",
"name": "interface_name",
"direction": "in"
},
{
"type": "s",
"name": "property_name",
"direction": "in"
},
{
"type": "v",
"name": "value",
"direction": "out"
}
]
},
{
"name": "GetAll",
"args": [
{
"type": "s",
"name": "interface_name",
"direction": "in"
},
{
"type": "a{sv}",
"name": "properties",
"direction": "out"
}
]
},
{
"name": "Set",
"args": [
{
"type": "s",
"name": "interface_name",
"direction": "in"
},
{
"type": "s",
"name": "property_name",
"direction": "in"
},
{
"type": "v",
"name": "value",
"direction": "in"
}
]
}
],
"signal": [
{
"name": "PropertiesChanged",
"args": [
{
"type": "s",
"name": "interface_name"
},
{
"type": "a{sv}",
"name": "changed_properties"
},
{
"type": "as",
"name": "invalidated_properties"
}
]
}
]
},
"org.freedesktop.DBus.Introspectable": {
"method": [
{
"name": "Introspect",
"args": [
{
"type": "s",
"name": "xml_data",
"direction": "out"
}
]
}
]
},
"org.freedesktop.DBus.Peer": {
"method": [
{
"name": "Ping"
},
{
"name": "GetMachineId",
"args": [
{
"type": "s",
"name": "machine_uuid",
"direction": "out"
}
]
}
]
},
"org.platypush.Bus": {
"method": [
{
"name": "Post",
"args": [
{
"type": "s",
"name": "msg",
"direction": "in"
},
{
"type": "s",
"name": "response",
"direction": "out"
}
]
}
]
}
}
}
}
"""
busses = {}
response = {}
if system_bus:
busses['system'] = dbus.SystemBus()
if session_bus:
busses['session'] = dbus.SessionBus()
if isinstance(bus, str):
bus = (bus,)
if BusType.SYSTEM.value in bus:
busses['system'] = self._system_bus
if BusType.SESSION.value in bus:
busses['session'] = self._session_bus
for bus_name, bus in busses.items():
services = {}
@ -91,40 +417,59 @@ class DbusPlugin(Plugin):
if not service:
for srv in service_names:
services[srv] = self.path_names(bus, srv)
services.update(self.path_names(bus, srv))
elif service in service_names:
services[service] = self.path_names(bus, service)
services.update(self.path_names(bus, service))
response[bus_name] = services
return response
@action
def execute(self, service: str, path: str, method_name: str, args: Optional[list] = None,
interface: Optional[str] = None, bus_type: str = BusType.SESSION.value):
def execute(
self,
service: str,
interface: str,
method_name: str,
bus: str = BusType.SESSION.value,
path: str = '/',
args: Optional[list] = None
):
"""
Execute a method exposed on DBus.
:param service: Service/bus name (e.g. ``org.platypush.Bus``).
:param path: Object path (e.g. ``/MessageService``).
:param method_name: Method name (e.g. ``Post``).
:param service: D-Bus service name.
:param interface: D-Bus nterface name.
:param method_name: Method name.
:param bus: Bus type. Supported: ``system`` and ``session`` (default: ``session``).
:param path: Object path.
:param args: Arguments to be passed to the method, depending on the method signature.
:param interface: Interface name (e.g. ``org.platypush.MessageBusInterface``).
:param bus_type: Bus type (supported: ``system`` and ``session`` - default: ``session``).
:return: Return value of the executed method.
"""
if not args:
args = []
kwargs = {}
if interface:
kwargs['dbus_interface'] = interface
bus = self._get_bus(bus)
obj = bus.get(service, path)[interface]
method = getattr(obj, method_name, None)
assert method, (
f'No such method exposed by service={service}, '
f'interface={interface}: {method_name}'
)
bus_type = BusType(bus_type)
bus = dbus.SessionBus() if bus_type == BusType.SESSION else dbus.SystemBus()
obj = bus.get_object(bus_name=service, object_path=path)
ret = getattr(obj, method_name)(*args, **kwargs)
return json.loads(json.dumps(ret))
# Normalize any lists/dictionaries to JSON strings
for i, arg in enumerate(args):
if isinstance(arg, (list, tuple, dict)):
args[i] = json.dumps(arg)
ret = method(*args)
try:
ret = json.loads(json.dumps(ret))
except Exception as e:
self.logger.debug(e)
return ret
# vim:sw=4:ts=4:et:

6
platypush/plugins/dbus/manifest.yaml
View file

@ -1,7 +1,9 @@
manifest:
events: {}
events:
platypush.message.event.dbus.DbusSignalEvent: When a signal is received
install:
pip:
- dbus-python
- pydbus
- defusedxml
package: platypush.plugins.dbus
type: plugin

26
platypush/schemas/dbus.py Normal file
View file

@ -0,0 +1,26 @@
from marshmallow import fields
from marshmallow.schema import Schema
from marshmallow.validate import OneOf
class DbusSignalSchema(Schema):
bus = fields.String(
required=True,
validate=OneOf(['system', 'session'])
)
interface = fields.String(allow_none=True, metadata={
'description': 'The DBus interface that should be monitored (default: all)'
})
path = fields.String(allow_none=True, metadata={
'description': 'Path of the resource to be monitored (default: all)'
})
signal = fields.String(allow_none=True, metadata={
'description': 'Signal name filter (default: all signals)'
})
sender = fields.String(allow_none=True, metadata={
'description': 'Signal sender filter (default: all senders)'
})

2
setup.py
View file

@ -227,7 +227,7 @@ setup(
# Support for luma.oled display drivers
'luma-oled': ['luma.oled @ git+https://github.com/rm-hull/luma.oled'],
# Support for DBus integration
'dbus': ['dbus-python'],
'dbus': ['pydbus', 'defusedxml'],
# Support for Twilio integration
'twilio': ['twilio'],
# Support for DHT11/DHT22/AM2302 temperature/humidity sensors