Bundle pydbus library
This commit is contained in:
wheaney
parent
3cf21b1ff6
commit
cfdbc82b40
|
|
@ -37,6 +37,7 @@ mkdir -p $PACKAGE_APPS_DIR
|
|||
mkdir -p $PACKAGE_SCHEMAS_DIR
|
||||
|
||||
cp src/*.py $PACKAGE_BREEZY_SRC_DIR
|
||||
cp -r lib $PACKAGE_BREEZY_SRC_DIR
|
||||
cp -L modules/PyXRLinuxDriverIPC/xrdriveripc.py $PACKAGE_BREEZY_SRC_DIR
|
||||
cp $UI_BUILD_PATH/src/breezydesktop.gresource $PACKAGE_BREEZY_DIR
|
||||
cp -r po/mo/* $PACKAGE_LOCALE_DIR
|
||||
|
|
|
|||
|
|
@ -0,0 +1,115 @@
|
|||
pydbus
|
||||
======
|
||||
.. image:: https://travis-ci.org/LEW21/pydbus.svg?branch=master
|
||||
:target: https://travis-ci.org/LEW21/pydbus
|
||||
.. image:: https://badge.fury.io/py/pydbus.svg
|
||||
:target: https://badge.fury.io/py/pydbus
|
||||
|
||||
Pythonic DBus library.
|
||||
|
||||
Changelog: https://github.com/LEW21/pydbus/releases
|
||||
|
||||
Requirements
|
||||
------------
|
||||
* Python 2.7+ - but works best on 3.4+ (help system is nicer there)
|
||||
* PyGI_ (not packaged on pypi, you need to install it from your distribution's repository - it's usually called python-gi, python-gobject or pygobject)
|
||||
* GLib_ 2.46+ and girepository_ 1.46+ (Ubuntu 16.04+) - for object publication support
|
||||
|
||||
.. _PyGI: https://wiki.gnome.org/Projects/PyGObject
|
||||
.. _GLib: https://developer.gnome.org/glib/
|
||||
.. _girepository: https://wiki.gnome.org/Projects/GObjectIntrospection
|
||||
|
||||
Examples
|
||||
--------
|
||||
|
||||
Send a desktop notification
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SessionBus
|
||||
|
||||
bus = SessionBus()
|
||||
notifications = bus.get('.Notifications')
|
||||
|
||||
notifications.Notify('test', 0, 'dialog-information', "Hello World!", "pydbus works :)", [], {}, 5000)
|
||||
|
||||
List systemd units
|
||||
~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SystemBus
|
||||
|
||||
bus = SystemBus()
|
||||
systemd = bus.get(".systemd1")
|
||||
|
||||
for unit in systemd.ListUnits():
|
||||
print(unit)
|
||||
|
||||
Start or stop systemd unit
|
||||
~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SystemBus
|
||||
|
||||
bus = SystemBus()
|
||||
systemd = bus.get(".systemd1")
|
||||
|
||||
job1 = systemd.StopUnit("ssh.service", "fail")
|
||||
job2 = systemd.StartUnit("ssh.service", "fail")
|
||||
|
||||
Watch for new systemd jobs
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SystemBus
|
||||
from gi.repository import GLib
|
||||
|
||||
bus = SystemBus()
|
||||
systemd = bus.get(".systemd1")
|
||||
|
||||
systemd.JobNew.connect(print)
|
||||
GLib.MainLoop().run()
|
||||
|
||||
# or
|
||||
|
||||
systemd.onJobNew = print
|
||||
GLib.MainLoop().run()
|
||||
|
||||
View object's API
|
||||
~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SessionBus
|
||||
|
||||
bus = SessionBus()
|
||||
notifications = bus.get('.Notifications')
|
||||
|
||||
help(notifications)
|
||||
|
||||
More examples & documentation
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
|
||||
The Tutorial_ contains more examples and docs.
|
||||
|
||||
.. _Tutorial: https://github.com/LEW21/pydbus/blob/master/doc/tutorial.rst
|
||||
|
||||
Copyright Information
|
||||
---------------------
|
||||
|
||||
Copyright (C) 2014, 2015, 2016 Linus Lewandowski <linus@lew21.net>
|
||||
|
||||
This library is free software; you can redistribute it and/or
|
||||
modify it under the terms of the GNU Lesser General Public
|
||||
License as published by the Free Software Foundation; either
|
||||
version 2.1 of the License, or (at your option) any later version.
|
||||
|
||||
This library 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
|
||||
Lesser General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU Lesser General Public
|
||||
License along with this library; if not, write to the Free Software
|
||||
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
|
||||
|
||||
|
||||
|
|
@ -0,0 +1 @@
|
|||
pip
|
||||
|
|
@ -0,0 +1,137 @@
|
|||
Metadata-Version: 2.0
|
||||
Name: pydbus
|
||||
Version: 0.6.0
|
||||
Summary: Pythonic DBus library
|
||||
Home-page: https://github.com/LEW21/pydbus
|
||||
Author: Linus Lewandowski
|
||||
Author-email: linus@lew21.net
|
||||
License: LGPLv2+
|
||||
Keywords: dbus
|
||||
Platform: UNKNOWN
|
||||
Classifier: Development Status :: 5 - Production/Stable
|
||||
Classifier: Intended Audience :: Developers
|
||||
Classifier: Natural Language :: English
|
||||
Classifier: License :: OSI Approved :: GNU Lesser General Public License v2 or later (LGPLv2+)
|
||||
Classifier: Programming Language :: Python
|
||||
Classifier: Programming Language :: Python :: 3
|
||||
Classifier: Programming Language :: Python :: 3.3
|
||||
Classifier: Programming Language :: Python :: 3.4
|
||||
Classifier: Programming Language :: Python :: 3.5
|
||||
Classifier: Programming Language :: Python :: 2
|
||||
Classifier: Programming Language :: Python :: 2.7
|
||||
|
||||
pydbus
|
||||
======
|
||||
.. image:: https://travis-ci.org/LEW21/pydbus.svg?branch=master
|
||||
:target: https://travis-ci.org/LEW21/pydbus
|
||||
.. image:: https://badge.fury.io/py/pydbus.svg
|
||||
:target: https://badge.fury.io/py/pydbus
|
||||
|
||||
Pythonic DBus library.
|
||||
|
||||
Changelog: https://github.com/LEW21/pydbus/releases
|
||||
|
||||
Requirements
|
||||
------------
|
||||
* Python 2.7+ - but works best on 3.4+ (help system is nicer there)
|
||||
* PyGI_ (not packaged on pypi, you need to install it from your distribution's repository - it's usually called python-gi, python-gobject or pygobject)
|
||||
* GLib_ 2.46+ and girepository_ 1.46+ (Ubuntu 16.04+) - for object publication support
|
||||
|
||||
.. _PyGI: https://wiki.gnome.org/Projects/PyGObject
|
||||
.. _GLib: https://developer.gnome.org/glib/
|
||||
.. _girepository: https://wiki.gnome.org/Projects/GObjectIntrospection
|
||||
|
||||
Examples
|
||||
--------
|
||||
|
||||
Send a desktop notification
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SessionBus
|
||||
|
||||
bus = SessionBus()
|
||||
notifications = bus.get('.Notifications')
|
||||
|
||||
notifications.Notify('test', 0, 'dialog-information', "Hello World!", "pydbus works :)", [], {}, 5000)
|
||||
|
||||
List systemd units
|
||||
~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SystemBus
|
||||
|
||||
bus = SystemBus()
|
||||
systemd = bus.get(".systemd1")
|
||||
|
||||
for unit in systemd.ListUnits():
|
||||
print(unit)
|
||||
|
||||
Start or stop systemd unit
|
||||
~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SystemBus
|
||||
|
||||
bus = SystemBus()
|
||||
systemd = bus.get(".systemd1")
|
||||
|
||||
job1 = systemd.StopUnit("ssh.service", "fail")
|
||||
job2 = systemd.StartUnit("ssh.service", "fail")
|
||||
|
||||
Watch for new systemd jobs
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SystemBus
|
||||
from gi.repository import GLib
|
||||
|
||||
bus = SystemBus()
|
||||
systemd = bus.get(".systemd1")
|
||||
|
||||
systemd.JobNew.connect(print)
|
||||
GLib.MainLoop().run()
|
||||
|
||||
# or
|
||||
|
||||
systemd.onJobNew = print
|
||||
GLib.MainLoop().run()
|
||||
|
||||
View object's API
|
||||
~~~~~~~~~~~~~~~~~
|
||||
.. code-block:: python
|
||||
|
||||
from pydbus import SessionBus
|
||||
|
||||
bus = SessionBus()
|
||||
notifications = bus.get('.Notifications')
|
||||
|
||||
help(notifications)
|
||||
|
||||
More examples & documentation
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
|
||||
The Tutorial_ contains more examples and docs.
|
||||
|
||||
.. _Tutorial: https://github.com/LEW21/pydbus/blob/master/doc/tutorial.rst
|
||||
|
||||
Copyright Information
|
||||
---------------------
|
||||
|
||||
Copyright (C) 2014, 2015, 2016 Linus Lewandowski <linus@lew21.net>
|
||||
|
||||
This library is free software; you can redistribute it and/or
|
||||
modify it under the terms of the GNU Lesser General Public
|
||||
License as published by the Free Software Foundation; either
|
||||
version 2.1 of the License, or (at your option) any later version.
|
||||
|
||||
This library 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
|
||||
Lesser General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU Lesser General Public
|
||||
License along with this library; if not, write to the Free Software
|
||||
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
pydbus-0.6.0.dist-info/DESCRIPTION.rst,sha256=pKEYrpPtLiD4ksnqzDs6ZepA9c3uoy5Mjvuxoi1_pEU,3033
|
||||
pydbus-0.6.0.dist-info/INSTALLER,sha256=zuuue4knoyJ-UwPPXg8fezS7VCrXJQrAP7zeNuwvFQg,4
|
||||
pydbus-0.6.0.dist-info/METADATA,sha256=JKSoIIJKdCLkuWiXdKtv4bSyMt9rA7BM3_RJbxqk5y4,3838
|
||||
pydbus-0.6.0.dist-info/RECORD,,
|
||||
pydbus-0.6.0.dist-info/REQUESTED,sha256=47DEQpj8HBSa-_TImW-5JCeuQeRkm5NMpJWZG3hSuFU,0
|
||||
pydbus-0.6.0.dist-info/WHEEL,sha256=o2k-Qa-RMNIJmUdIc7KU6VWR_ErNRbWNlxDIpl7lm34,110
|
||||
pydbus-0.6.0.dist-info/metadata.json,sha256=fouQjd0oTDJrVRvQO8dwqvMZbNib8Ui4TgArvHoxzmQ,916
|
||||
pydbus-0.6.0.dist-info/top_level.txt,sha256=X1ybDik1ZA7yilKlt-MUQCPJBNlAW9Jp6bHoAYgAgN0,7
|
||||
pydbus-0.6.0.dist-info/zip-safe,sha256=AbpHGcgLb-kRsJGnwFEktk7uzpZOCcBY74-YBdrKVGs,1
|
||||
pydbus/__init__.py,sha256=5ouUtYT15bJ-LaEFea3ZccXkhlZHyf1ykwKLNnFMlIs,148
|
||||
pydbus/__pycache__/__init__.cpython-312.pyc,,
|
||||
pydbus/__pycache__/_inspect3.cpython-312.pyc,,
|
||||
pydbus/__pycache__/auto_names.cpython-312.pyc,,
|
||||
pydbus/__pycache__/bus.cpython-312.pyc,,
|
||||
pydbus/__pycache__/bus_names.cpython-312.pyc,,
|
||||
pydbus/__pycache__/exitable.cpython-312.pyc,,
|
||||
pydbus/__pycache__/generic.cpython-312.pyc,,
|
||||
pydbus/__pycache__/identifier.cpython-312.pyc,,
|
||||
pydbus/__pycache__/method_call_context.cpython-312.pyc,,
|
||||
pydbus/__pycache__/proxy.cpython-312.pyc,,
|
||||
pydbus/__pycache__/proxy_method.cpython-312.pyc,,
|
||||
pydbus/__pycache__/proxy_property.cpython-312.pyc,,
|
||||
pydbus/__pycache__/proxy_signal.cpython-312.pyc,,
|
||||
pydbus/__pycache__/publication.cpython-312.pyc,,
|
||||
pydbus/__pycache__/registration.cpython-312.pyc,,
|
||||
pydbus/__pycache__/request_name.cpython-312.pyc,,
|
||||
pydbus/__pycache__/subscription.cpython-312.pyc,,
|
||||
pydbus/__pycache__/timeout.cpython-312.pyc,,
|
||||
pydbus/_inspect3.py,sha256=lDzzSThwvbMQ3PwY-SVpWByu___QPhbSvPY0xwQScc8,708
|
||||
pydbus/auto_names.py,sha256=elJqIT2dIhioBYUMvIqtH6Pt_tFCf29QSzp0QXbndy8,519
|
||||
pydbus/bus.py,sha256=dMJAqRq85TJAz8iDZSg8COhjy8sabrofZ2jjSe6Pxds,1565
|
||||
pydbus/bus_names.py,sha256=vdkfHdQy_9FzMWkJaXJZpDpvktV9N2TFyRzLE8X1cDI,3533
|
||||
pydbus/exitable.py,sha256=FblBVEUjz6DyqgJpOVlE8-TMeVaC1ZPf9wyi7ejQ-gI,982
|
||||
pydbus/generic.py,sha256=2K9VNwu-TkMWws818H2BiyLsoX99fZuJgbmoNtVLzjE,2589
|
||||
pydbus/identifier.py,sha256=iVrw6rDMyNiQouNUAMpvEhGCz1Bog-R0HwzmvsHs0LY,383
|
||||
pydbus/method_call_context.py,sha256=ilIh0jJmmdoJX26UF1uV9eYvdQXXgYoyxzNHKvIJcJA,1039
|
||||
pydbus/proxy.py,sha256=Og9VbKJ4sZvr4dI7cBwVKsklWAqfnEmAlSpoaPJIs-g,4203
|
||||
pydbus/proxy_method.py,sha256=I741zD_vBixhpeKArTVLaUDTQsZGJ98xaX2SGORJkb0,3089
|
||||
pydbus/proxy_property.py,sha256=WUkOV4V8DuvmN1-yAD54IJklZqgJex2eTV2KZ20PJ1k,1066
|
||||
pydbus/proxy_signal.py,sha256=D60OxO635erbIknziTzX3QDChhImmjzgUpwjVe8gy6A,2112
|
||||
pydbus/publication.py,sha256=oJKPraVVv0YPLWN2-y0KjjjW1Lu4NNTg9lNoVneYLlY,1391
|
||||
pydbus/registration.py,sha256=GWBv2lDDLrkYCr9WsaEFS1hZstuXB7OgHdEnvC0Shc4,5565
|
||||
pydbus/request_name.py,sha256=jpWNd8H8rlRdJQiNAud-di1jU6gE31JvLOtoXQm_lHE,883
|
||||
pydbus/subscription.py,sha256=MBXKRIvLvSgYLIT7A0a3Qn6aAvT6Rcx1b6xudTsr7mQ,2166
|
||||
pydbus/timeout.py,sha256=G4o9dVwN7hb4MaPUHlcGpQKtlUSaTGjBTBC7nuza9mI,303
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
Wheel-Version: 1.0
|
||||
Generator: bdist_wheel (0.29.0)
|
||||
Root-Is-Purelib: true
|
||||
Tag: py2-none-any
|
||||
Tag: py3-none-any
|
||||
|
||||
|
|
@ -0,0 +1 @@
|
|||
{"classifiers": ["Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", "Natural Language :: English", "License :: OSI Approved :: GNU Lesser General Public License v2 or later (LGPLv2+)", "Programming Language :: Python", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.3", "Programming Language :: Python :: 3.4", "Programming Language :: Python :: 3.5", "Programming Language :: Python :: 2", "Programming Language :: Python :: 2.7"], "extensions": {"python.details": {"contacts": [{"email": "linus@lew21.net", "name": "Linus Lewandowski", "role": "author"}], "document_names": {"description": "DESCRIPTION.rst"}, "project_urls": {"Home": "https://github.com/LEW21/pydbus"}}}, "generator": "bdist_wheel (0.29.0)", "keywords": ["dbus"], "license": "LGPLv2+", "metadata_version": "2.0", "name": "pydbus", "summary": "Pythonic DBus library", "version": "0.6.0"}
|
||||
|
|
@ -0,0 +1 @@
|
|||
pydbus
|
||||
|
|
@ -0,0 +1 @@
|
|||
|
||||
|
|
@ -0,0 +1,4 @@
|
|||
from .bus import SystemBus, SessionBus, connect
|
||||
from gi.repository.GLib import Variant
|
||||
|
||||
__all__ = ["SystemBus", "SessionBus", "connect", "Variant"]
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
from collections import OrderedDict
|
||||
from inspect import getargspec
|
||||
|
||||
class _empty:
|
||||
pass
|
||||
|
||||
class Signature:
|
||||
empty = _empty
|
||||
|
||||
def __init__(self, parameters=None, return_annotation=_empty):
|
||||
self.parameters = OrderedDict(((param.name, param) for param in parameters))
|
||||
self.return_annotation = return_annotation
|
||||
|
||||
class Parameter:
|
||||
empty = _empty
|
||||
|
||||
POSITIONAL_ONLY = 0
|
||||
POSITIONAL_OR_KEYWORD = 1
|
||||
KEYWORD_ONLY = 999
|
||||
|
||||
def __init__(self, name, kind, default=_empty, annotation=_empty):
|
||||
self.name = name
|
||||
self.kind = kind
|
||||
self.annotation = annotation
|
||||
|
||||
def signature(f):
|
||||
parameters = [Parameter(arg, Parameter.POSITIONAL_OR_KEYWORD) for arg in getargspec(f).args]
|
||||
return Signature(parameters = parameters)
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
from gi.repository import Gio
|
||||
|
||||
def auto_bus_name(bus_name):
|
||||
if bus_name[0] == ".":
|
||||
#Default namespace
|
||||
bus_name = "org.freedesktop" + bus_name
|
||||
|
||||
if not Gio.dbus_is_name(bus_name):
|
||||
raise ValueError("invalid bus name")
|
||||
|
||||
return bus_name
|
||||
|
||||
def auto_object_path(bus_name, object_path=None):
|
||||
if object_path is None:
|
||||
# They always name it like that.
|
||||
object_path = "/" + bus_name.replace(".", "/")
|
||||
|
||||
if object_path[0] != "/":
|
||||
object_path = "/" + bus_name.replace(".", "/") + "/" + object_path
|
||||
|
||||
return object_path
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
from gi.repository import Gio
|
||||
from .proxy import ProxyMixin
|
||||
from .request_name import RequestNameMixin
|
||||
from .bus_names import OwnMixin, WatchMixin
|
||||
from .subscription import SubscriptionMixin
|
||||
from .registration import RegistrationMixin
|
||||
from .publication import PublicationMixin
|
||||
|
||||
def pydbus_property(self):
|
||||
try:
|
||||
return self._pydbus
|
||||
except AttributeError:
|
||||
self._pydbus = Bus(self)
|
||||
return self._pydbus
|
||||
|
||||
Gio.DBusConnection.pydbus = property(pydbus_property)
|
||||
|
||||
def bus_get(type):
|
||||
return Gio.bus_get_sync(type, None).pydbus
|
||||
|
||||
def connect(address):
|
||||
c = Gio.DBusConnection.new_for_address_sync(address, Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION, None, None)
|
||||
c.pydbus.autoclose = True
|
||||
return c.pydbus
|
||||
|
||||
class Bus(ProxyMixin, RequestNameMixin, OwnMixin, WatchMixin, SubscriptionMixin, RegistrationMixin, PublicationMixin):
|
||||
Type = Gio.BusType
|
||||
|
||||
def __init__(self, gio_con):
|
||||
self.con = gio_con
|
||||
self.autoclose = False
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, traceback):
|
||||
if self.autoclose:
|
||||
self.con.close_sync(None)
|
||||
|
||||
@property
|
||||
def dbus(self):
|
||||
try:
|
||||
return self._dbus
|
||||
except AttributeError:
|
||||
self._dbus = self.get(".DBus")[""]
|
||||
return self._dbus
|
||||
|
||||
@property
|
||||
def polkit_authority(self):
|
||||
try:
|
||||
return self._polkit_authority
|
||||
except AttributeError:
|
||||
self._polkit_authority = self.get(".PolicyKit1", "Authority")[""]
|
||||
return self._polkit_authority
|
||||
|
||||
def SystemBus():
|
||||
return bus_get(Bus.Type.SYSTEM)
|
||||
|
||||
def SessionBus():
|
||||
return bus_get(Bus.Type.SESSION)
|
||||
|
|
@ -0,0 +1,97 @@
|
|||
from gi.repository import Gio
|
||||
from .exitable import ExitableWithAliases
|
||||
import warnings
|
||||
|
||||
class NameOwner(ExitableWithAliases("unown")):
|
||||
Flags = Gio.BusNameOwnerFlags
|
||||
__slots__ = ()
|
||||
|
||||
def __init__(self, con, name, flags, name_aquired_handler, name_lost_handler):
|
||||
id = Gio.bus_own_name_on_connection(con, name, flags, name_aquired_handler, name_lost_handler)
|
||||
self._at_exit(lambda: Gio.bus_unown_name(id))
|
||||
|
||||
class NameWatcher(ExitableWithAliases("unwatch")):
|
||||
Flags = Gio.BusNameWatcherFlags
|
||||
__slots__ = ()
|
||||
|
||||
def __init__(self, con, name, flags, name_appeared_handler, name_vanished_handler):
|
||||
id = Gio.bus_watch_name_on_connection(con, name, flags, name_appeared_handler, name_vanished_handler)
|
||||
self._at_exit(lambda: Gio.bus_unwatch_name(id))
|
||||
|
||||
class OwnMixin(object):
|
||||
__slots__ = ()
|
||||
NameOwnerFlags = NameOwner.Flags
|
||||
|
||||
def own_name(self, name, flags=0, name_aquired=None, name_lost=None):
|
||||
"""[DEPRECATED] Asynchronously aquires a bus name.
|
||||
|
||||
Starts acquiring name on the bus specified by bus_type and calls
|
||||
name_acquired and name_lost when the name is acquired respectively lost.
|
||||
|
||||
To receive name_aquired and name_lost callbacks, you need an event loop.
|
||||
https://github.com/LEW21/pydbus/blob/master/doc/tutorial.rst#setting-up-an-event-loop
|
||||
|
||||
Parameters
|
||||
----------
|
||||
name : string
|
||||
Bus name to aquire
|
||||
flags : NameOwnerFlags, optional
|
||||
name_aquired : callable, optional
|
||||
Invoked when name is acquired
|
||||
name_lost : callable, optional
|
||||
Invoked when name is lost
|
||||
|
||||
Returns
|
||||
-------
|
||||
NameOwner
|
||||
An object you can use as a context manager to unown the name later.
|
||||
|
||||
See Also
|
||||
--------
|
||||
See https://developer.gnome.org/gio/2.44/gio-Owning-Bus-Names.html#g-bus-own-name
|
||||
for more information.
|
||||
"""
|
||||
warnings.warn("own_name() is deprecated, use request_name() instead.", DeprecationWarning)
|
||||
|
||||
name_aquired_handler = (lambda con, name: name_aquired()) if name_aquired is not None else None
|
||||
name_lost_handler = (lambda con, name: name_lost()) if name_lost is not None else None
|
||||
return NameOwner(self.con, name, flags, name_aquired_handler, name_lost_handler)
|
||||
|
||||
class WatchMixin(object):
|
||||
__slots__ = ()
|
||||
NameWatcherFlags = NameWatcher.Flags
|
||||
|
||||
def watch_name(self, name, flags=0, name_appeared=None, name_vanished=None):
|
||||
"""Asynchronously watches a bus name.
|
||||
|
||||
Starts watching name on the bus specified by bus_type and calls
|
||||
name_appeared and name_vanished when the name is known to have a owner
|
||||
respectively known to lose its owner.
|
||||
|
||||
To receive name_appeared and name_vanished callbacks, you need an event loop.
|
||||
https://github.com/LEW21/pydbus/blob/master/doc/tutorial.rst#setting-up-an-event-loop
|
||||
|
||||
Parameters
|
||||
----------
|
||||
name : string
|
||||
Bus name to watch
|
||||
flags : NameWatcherFlags, optional
|
||||
name_appeared : callable, optional
|
||||
Invoked when name is known to exist
|
||||
Called as name_appeared(name_owner).
|
||||
name_vanished : callable, optional
|
||||
Invoked when name is known to not exist
|
||||
|
||||
Returns
|
||||
-------
|
||||
NameWatcher
|
||||
An object you can use as a context manager to unwatch the name later.
|
||||
|
||||
See Also
|
||||
--------
|
||||
See https://developer.gnome.org/gio/2.44/gio-Watching-Bus-Names.html#g-bus-watch-name
|
||||
for more information.
|
||||
"""
|
||||
name_appeared_handler = (lambda con, name, name_owner: name_appeared(name_owner)) if name_appeared is not None else None
|
||||
name_vanished_handler = (lambda con, name: name_vanished()) if name_vanished is not None else None
|
||||
return NameWatcher(self.con, name, flags, name_appeared_handler, name_vanished_handler)
|
||||
|
|
@ -0,0 +1,52 @@
|
|||
import inspect
|
||||
|
||||
class Exitable(object):
|
||||
__slots__ = ("_at_exit_cbs")
|
||||
|
||||
def _at_exit(self, cb):
|
||||
try:
|
||||
self._at_exit_cbs
|
||||
except AttributeError:
|
||||
self._at_exit_cbs = []
|
||||
|
||||
self._at_exit_cbs.append(cb)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type = None, exc_value = None, traceback = None):
|
||||
if self._exited:
|
||||
return
|
||||
|
||||
for cb in reversed(self._at_exit_cbs):
|
||||
call_with_exc = True
|
||||
try:
|
||||
inspect.getcallargs(cb, exc_type, exc_value, traceback)
|
||||
except TypeError:
|
||||
call_with_exc = False
|
||||
|
||||
if call_with_exc:
|
||||
cb(exc_type, exc_value, traceback)
|
||||
else:
|
||||
cb()
|
||||
|
||||
self._at_exit_cbs = None
|
||||
|
||||
@property
|
||||
def _exited(self):
|
||||
try:
|
||||
return self._at_exit_cbs is None
|
||||
except AttributeError:
|
||||
return True
|
||||
|
||||
def ExitableWithAliases(*exit_methods):
|
||||
class CustomExitable(Exitable):
|
||||
pass
|
||||
|
||||
def exit(self):
|
||||
self.__exit__()
|
||||
|
||||
for exit_method_name in exit_methods:
|
||||
setattr(CustomExitable, exit_method_name, exit)
|
||||
|
||||
return CustomExitable
|
||||
|
|
@ -0,0 +1,105 @@
|
|||
"""Generic programming utilities.
|
||||
|
||||
Utilities implemented in this file are not dependent
|
||||
on dbus, they can be used everywhere.
|
||||
"""
|
||||
|
||||
class subscription(object):
|
||||
__slots__ = ("callback_list", "callback")
|
||||
|
||||
def __init__(self, callback_list, callback):
|
||||
self.callback_list = callback_list
|
||||
self.callback = callback
|
||||
self.callback_list.append(callback)
|
||||
|
||||
def unsubscribe(self):
|
||||
self.callback_list.remove(self.callback)
|
||||
self.callback_list = None
|
||||
self.callback = None
|
||||
|
||||
def disconnect(self):
|
||||
"""An alias for unsubscribe()"""
|
||||
self.unsubscribe()
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, traceback):
|
||||
if not self.callback is None:
|
||||
self.unsubscribe()
|
||||
|
||||
class bound_signal(object):
|
||||
__slots__ = ("__signal__", "__self__") # bound method uses ("__func__", "__self__")
|
||||
|
||||
def __init__(self, signal, instance):
|
||||
self.__signal__ = signal
|
||||
self.__self__ = instance
|
||||
|
||||
@property
|
||||
def callbacks(self):
|
||||
return self.__signal__.map[self.__self__]
|
||||
|
||||
def connect(self, callback):
|
||||
"""Subscribe to the signal."""
|
||||
return self.__signal__.connect(self.__self__, callback)
|
||||
|
||||
def emit(self, *args):
|
||||
"""Emit the signal."""
|
||||
self.__signal__.emit(self.__self__, *args)
|
||||
|
||||
def __call__(self, *args):
|
||||
"""Emit the signal."""
|
||||
self.emit(*args)
|
||||
|
||||
def __repr__(self):
|
||||
return "<bound signal " + self.__signal__.__qualname__ + " of " + repr(self.__self__) + ">"
|
||||
|
||||
class signal(object):
|
||||
"""Static signal object
|
||||
|
||||
You're expected to set it as a class property::
|
||||
|
||||
class A:
|
||||
SomethingHappened = signal()
|
||||
|
||||
Declared this way, it can be used on class instances
|
||||
to connect signal observers::
|
||||
|
||||
a = A()
|
||||
a.SomethingHappened.connect(func)
|
||||
|
||||
and emit the signal::
|
||||
|
||||
a.SomethingHappened()
|
||||
|
||||
You may pass any parameters to the emiting function
|
||||
- they will be forwarded to all subscribed callbacks.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.map = {}
|
||||
self.__qualname__ = "<anonymous signal>" # function uses <lambda> ;)
|
||||
self.__doc__ = "Signal."
|
||||
|
||||
def connect(self, object, callback):
|
||||
"""Subscribe to the signal."""
|
||||
return subscription(self.map.setdefault(object, []), callback)
|
||||
|
||||
def emit(self, object, *args):
|
||||
"""Emit the signal."""
|
||||
for cb in self.map.get(object, []):
|
||||
cb(*args)
|
||||
|
||||
def __get__(self, instance, owner):
|
||||
if instance is None:
|
||||
return self
|
||||
|
||||
return bound_signal(self, instance)
|
||||
|
||||
def __set__(self, instance, value):
|
||||
raise AttributeError("can't set attribute")
|
||||
|
||||
def __repr__(self):
|
||||
return "<signal " + self.__qualname__ + " at 0x" + format(id(self), "x") + ">"
|
||||
|
||||
bound_method = type(signal().emit) # TODO find a prettier way to get this type
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
|
||||
try:
|
||||
isident = str.isidentifier
|
||||
except:
|
||||
import re
|
||||
|
||||
isidentre = re.compile("^[a-zA-Z_][a-zA-Z0-9_]*$")
|
||||
def isident(s):
|
||||
return isidentre.match(s) is not None
|
||||
|
||||
def filter_identifier(name):
|
||||
name = name.replace("-", "_")
|
||||
|
||||
safe_name = ""
|
||||
for c in name:
|
||||
if not safe_name:
|
||||
if isident(c):
|
||||
safe_name += c
|
||||
else:
|
||||
if isident("a" + c):
|
||||
safe_name += c
|
||||
return safe_name
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
from gi.repository import GLib
|
||||
from collections import namedtuple
|
||||
|
||||
AuthorizationResult = namedtuple("AuthorizationResult", "is_authorized is_challenge details")
|
||||
|
||||
class MethodCallContext(object):
|
||||
def __init__(self, gdbus_method_invocation):
|
||||
self._mi = gdbus_method_invocation
|
||||
|
||||
@property
|
||||
def bus(self):
|
||||
return self._mi.get_connection().pydbus
|
||||
|
||||
@property
|
||||
def sender(self):
|
||||
return self._mi.get_sender()
|
||||
|
||||
@property
|
||||
def object_path(self):
|
||||
return self._mi.get_object_path()
|
||||
|
||||
@property
|
||||
def interface_name(self):
|
||||
return self._mi.get_interface_name()
|
||||
|
||||
@property
|
||||
def method_name(self):
|
||||
return self._mi.get_method_name()
|
||||
|
||||
def check_authorization(self, action_id, details, interactive=False):
|
||||
return AuthorizationResult(*self.bus.polkit_authority.CheckAuthorization(('system-bus-name', {'name': GLib.Variant("s", self.sender)}), action_id, details, 1 if interactive else 0, ''))
|
||||
|
||||
def is_authorized(self, action_id, details, interactive=False):
|
||||
return self.check_authorization(action_id, details, interactive).is_authorized
|
||||
|
|
@ -0,0 +1,123 @@
|
|||
from gi.repository import GLib
|
||||
from xml.etree import ElementTree as ET
|
||||
from .auto_names import *
|
||||
|
||||
from .proxy_method import ProxyMethod
|
||||
from .proxy_property import ProxyProperty
|
||||
from .proxy_signal import ProxySignal, OnSignal
|
||||
from .timeout import timeout_to_glib
|
||||
|
||||
class ProxyMixin(object):
|
||||
__slots__ = ()
|
||||
|
||||
def get(self, bus_name, object_path=None, **kwargs):
|
||||
"""Get a remote object.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
bus_name : string
|
||||
Name of the service that exposes this object.
|
||||
You may start with "." - then org.freedesktop will be automatically prepended.
|
||||
object_path : string, optional
|
||||
Path of the object. If not provided, bus_name translated to path format is used.
|
||||
|
||||
Returns
|
||||
-------
|
||||
ProxyObject implementing all the Interfaces exposed by the remote object.
|
||||
Note that it inherits from multiple Interfaces, so the method you want to use
|
||||
may be shadowed by another one, eg. from a newer version of the interface.
|
||||
Therefore, to interact with only a single interface, use:
|
||||
>>> bus.get("org.freedesktop.systemd1")["org.freedesktop.systemd1.Manager"]
|
||||
or simply
|
||||
>>> bus.get(".systemd1")[".Manager"]
|
||||
which will give you access to the one specific interface.
|
||||
"""
|
||||
# Python 2 sux
|
||||
for kwarg in kwargs:
|
||||
if kwarg not in ("timeout",):
|
||||
raise TypeError(self.__qualname__ + " got an unexpected keyword argument '{}'".format(kwarg))
|
||||
timeout = kwargs.get("timeout", None)
|
||||
|
||||
bus_name = auto_bus_name(bus_name)
|
||||
object_path = auto_object_path(bus_name, object_path)
|
||||
|
||||
ret = self.con.call_sync(
|
||||
bus_name, object_path,
|
||||
'org.freedesktop.DBus.Introspectable', "Introspect", None, GLib.VariantType.new("(s)"),
|
||||
0, timeout_to_glib(timeout), None)
|
||||
|
||||
if not ret:
|
||||
raise KeyError("no such object; you might need to pass object path as the 2nd argument for get()")
|
||||
|
||||
xml, = ret.unpack()
|
||||
|
||||
try:
|
||||
introspection = ET.fromstring(xml)
|
||||
except:
|
||||
raise KeyError("object provides invalid introspection XML")
|
||||
|
||||
return CompositeInterface(introspection)(self, bus_name, object_path)
|
||||
|
||||
class ProxyObject(object):
|
||||
def __init__(self, bus, bus_name, path, object=None):
|
||||
self._bus = bus
|
||||
self._bus_name = bus_name
|
||||
self._path = path
|
||||
self._object = object if object else self
|
||||
|
||||
def Interface(iface):
|
||||
|
||||
class interface(ProxyObject):
|
||||
@staticmethod
|
||||
def _Introspect():
|
||||
print(iface.attrib["name"] + ":")
|
||||
for member in iface:
|
||||
print("\t" + member.tag + " " + member.attrib["name"])
|
||||
print()
|
||||
|
||||
interface.__qualname__ = interface.__name__ = iface.attrib["name"]
|
||||
interface.__module__ = "DBUS"
|
||||
|
||||
for member in iface:
|
||||
member_name = member.attrib["name"]
|
||||
if member.tag == "method":
|
||||
setattr(interface, member_name, ProxyMethod(interface.__name__, member))
|
||||
elif member.tag == "property":
|
||||
setattr(interface, member_name, ProxyProperty(interface.__name__, member))
|
||||
elif member.tag == "signal":
|
||||
signal = ProxySignal(interface.__name__, member)
|
||||
setattr(interface, member_name, signal)
|
||||
setattr(interface, "on" + member_name, OnSignal(signal))
|
||||
|
||||
return interface
|
||||
|
||||
def CompositeInterface(introspection):
|
||||
class CompositeObject(ProxyObject):
|
||||
def __getitem__(self, iface):
|
||||
if iface == "" or iface[0] == ".":
|
||||
iface = self._path.replace("/", ".")[1:] + iface
|
||||
matching_bases = [base for base in type(self).__bases__ if base.__name__ == iface]
|
||||
|
||||
if len(matching_bases) == 0:
|
||||
raise KeyError(iface)
|
||||
assert(len(matching_bases) == 1)
|
||||
|
||||
iface_class = matching_bases[0]
|
||||
return iface_class(self._bus, self._bus_name, self._path, self)
|
||||
|
||||
@classmethod
|
||||
def _Introspect(cls):
|
||||
for iface in cls.__bases__:
|
||||
try:
|
||||
iface._Introspect()
|
||||
except:
|
||||
pass
|
||||
|
||||
ifaces = sorted([x for x in introspection if x.tag == "interface"], key=lambda x: int(x.attrib["name"].startswith("org.freedesktop.DBus.")))
|
||||
if not ifaces:
|
||||
raise KeyError("object does not export any interfaces; you might need to pass object path as the 2nd argument for get()")
|
||||
CompositeObject.__bases__ = tuple(Interface(iface) for iface in ifaces)
|
||||
CompositeObject.__name__ = "<CompositeObject>"
|
||||
CompositeObject.__qualname__ = "<CompositeObject>(" + "+".join(x.__name__ for x in CompositeObject.__bases__) + ")"
|
||||
CompositeObject.__module__ = "DBUS"
|
||||
return CompositeObject
|
||||
|
|
@ -0,0 +1,91 @@
|
|||
from gi.repository import GLib
|
||||
from .generic import bound_method
|
||||
from .identifier import filter_identifier
|
||||
from .timeout import timeout_to_glib
|
||||
|
||||
try:
|
||||
from inspect import Signature, Parameter
|
||||
put_signature_in_doc = False
|
||||
except:
|
||||
from ._inspect3 import Signature, Parameter
|
||||
put_signature_in_doc = True
|
||||
|
||||
class DBUSSignature(Signature):
|
||||
|
||||
def __str__(self):
|
||||
result = []
|
||||
for param in self.parameters.values():
|
||||
p = param.name if not param.name.startswith("arg") else ""
|
||||
if type(param.annotation) == str:
|
||||
p += ":" + param.annotation
|
||||
result.append(p)
|
||||
|
||||
rendered = '({})'.format(', '.join(result))
|
||||
|
||||
if self.return_annotation is not Signature.empty:
|
||||
rendered += ' -> {}'.format(self.return_annotation)
|
||||
|
||||
return rendered
|
||||
|
||||
class ProxyMethod(object):
|
||||
def __init__(self, iface_name, method):
|
||||
self._iface_name = iface_name
|
||||
self.__name__ = method.attrib["name"]
|
||||
self.__qualname__ = self._iface_name + "." + self.__name__
|
||||
|
||||
self._inargs = [(arg.attrib.get("name", ""), arg.attrib["type"]) for arg in method if arg.tag == "arg" and arg.attrib["direction"] == "in"]
|
||||
self._outargs = [arg.attrib["type"] for arg in method if arg.tag == "arg" and arg.attrib["direction"] == "out"]
|
||||
self._sinargs = "(" + "".join(x[1] for x in self._inargs) + ")"
|
||||
self._soutargs = "(" + "".join(self._outargs) + ")"
|
||||
|
||||
self_param = Parameter("self", Parameter.POSITIONAL_ONLY)
|
||||
pos_params = []
|
||||
for i, a in enumerate(self._inargs):
|
||||
name = filter_identifier(a[0])
|
||||
|
||||
if not name:
|
||||
name = "arg" + str(i)
|
||||
|
||||
param = Parameter(name, Parameter.POSITIONAL_ONLY, annotation=a[1])
|
||||
|
||||
pos_params.append(param)
|
||||
ret_type = Signature.empty if len(self._outargs) == 0 else self._outargs[0] if len(self._outargs) == 1 else "(" + ", ".join(self._outargs) + ")"
|
||||
|
||||
self.__signature__ = DBUSSignature([self_param] + pos_params, return_annotation=ret_type)
|
||||
|
||||
if put_signature_in_doc:
|
||||
self.__doc__ = self.__name__ + str(self.__signature__)
|
||||
|
||||
def __call__(self, instance, *args, **kwargs):
|
||||
argdiff = len(args) - len(self._inargs)
|
||||
if argdiff < 0:
|
||||
raise TypeError(self.__qualname__ + " missing {} required positional argument(s)".format(-argdiff))
|
||||
elif argdiff > 0:
|
||||
raise TypeError(self.__qualname__ + " takes {} positional argument(s) but {} was/were given".format(len(self._inargs), len(args)))
|
||||
|
||||
# Python 2 sux
|
||||
for kwarg in kwargs:
|
||||
if kwarg not in ("timeout",):
|
||||
raise TypeError(self.__qualname__ + " got an unexpected keyword argument '{}'".format(kwarg))
|
||||
timeout = kwargs.get("timeout", None)
|
||||
|
||||
ret = instance._bus.con.call_sync(
|
||||
instance._bus_name, instance._path,
|
||||
self._iface_name, self.__name__, GLib.Variant(self._sinargs, args), GLib.VariantType.new(self._soutargs),
|
||||
0, timeout_to_glib(timeout), None).unpack()
|
||||
|
||||
if len(self._outargs) == 0:
|
||||
return None
|
||||
elif len(self._outargs) == 1:
|
||||
return ret[0]
|
||||
else:
|
||||
return ret
|
||||
|
||||
def __get__(self, instance, owner):
|
||||
if instance is None:
|
||||
return self
|
||||
|
||||
return bound_method(self, instance)
|
||||
|
||||
def __repr__(self):
|
||||
return "<function " + self.__qualname__ + " at 0x" + format(id(self), "x") + ">"
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
from gi.repository import GLib
|
||||
|
||||
class ProxyProperty(object):
|
||||
def __init__(self, iface_name, property):
|
||||
self._iface_name = iface_name
|
||||
self.__name__ = property.attrib["name"]
|
||||
self.__qualname__ = self._iface_name + "." + self.__name__
|
||||
|
||||
self._type = property.attrib["type"]
|
||||
access = property.attrib["access"]
|
||||
self._readable = access.startswith("read")
|
||||
self._writeable = access.endswith("write")
|
||||
self.__doc__ = "(" + self._type + ") " + access
|
||||
|
||||
def __get__(self, instance, owner):
|
||||
if instance is None:
|
||||
return self
|
||||
|
||||
if not self._readable:
|
||||
raise AttributeError("unreadable attribute")
|
||||
|
||||
return instance._object["org.freedesktop.DBus.Properties"].Get(self._iface_name, self.__name__)
|
||||
|
||||
def __set__(self, instance, value):
|
||||
if instance is None or not self._writeable:
|
||||
raise AttributeError("can't set attribute")
|
||||
|
||||
instance._object["org.freedesktop.DBus.Properties"].Set(self._iface_name, self.__name__, GLib.Variant(self._type, value))
|
||||
|
||||
def __repr__(self):
|
||||
return "<property " + self.__qualname__ + " at 0x" + format(id(self), "x") + ">"
|
||||
|
|
@ -0,0 +1,66 @@
|
|||
from .generic import bound_signal
|
||||
|
||||
class ProxySignal(object):
|
||||
def __init__(self, iface_name, signal):
|
||||
self._iface_name = iface_name
|
||||
self.__name__ = signal.attrib["name"]
|
||||
self.__qualname__ = self._iface_name + "." + self.__name__
|
||||
|
||||
self._args = [arg.attrib["type"] for arg in signal if arg.tag == "arg"]
|
||||
self.__doc__ = "Signal. Callback: (" + ", ".join(self._args) + ")"
|
||||
|
||||
def connect(self, object, callback):
|
||||
"""Subscribe to the signal."""
|
||||
def signal_fired(sender, object, iface, signal, params):
|
||||
callback(*params)
|
||||
return object._bus.subscribe(sender=object._bus_name, object=object._path, iface=self._iface_name, signal=self.__name__, signal_fired=signal_fired)
|
||||
|
||||
def __get__(self, instance, owner):
|
||||
if instance is None:
|
||||
return self
|
||||
|
||||
return bound_signal(self, instance)
|
||||
|
||||
def __set__(self, instance, value):
|
||||
raise AttributeError("can't set attribute")
|
||||
|
||||
def __repr__(self):
|
||||
return "<signal " + self.__qualname__ + " at 0x" + format(id(self), "x") + ">"
|
||||
|
||||
class OnSignal(object):
|
||||
def __init__(self, signal):
|
||||
self.signal = signal
|
||||
self.__name__ = "on" + signal.__name__
|
||||
self.__qualname__ = signal._iface_name + "." + self.__name__
|
||||
self.__doc__ = "Assign a callback to subscribe to the signal. Assing None to unsubscribe. Callback: (" + ", ".join(signal._args) + ")"
|
||||
|
||||
def __get__(self, instance, owner):
|
||||
if instance is None:
|
||||
return self
|
||||
|
||||
try:
|
||||
return getattr(instance, "_on" + self.signal.__name__)
|
||||
except AttributeError:
|
||||
return None
|
||||
|
||||
def __set__(self, instance, value):
|
||||
if instance is None:
|
||||
raise AttributeError("can't set attribute")
|
||||
|
||||
try:
|
||||
old = getattr(instance, "_sub" + self.signal.__name__)
|
||||
old.unsubscribe()
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
if value is None:
|
||||
delattr(instance, "_on" + self.signal.__name__)
|
||||
delattr(instance, "_sub" + self.signal.__name__)
|
||||
return
|
||||
|
||||
sub = self.signal.connect(instance, value)
|
||||
setattr(instance, "_on" + self.signal.__name__, value)
|
||||
setattr(instance, "_sub" + self.signal.__name__, sub)
|
||||
|
||||
def __repr__(self):
|
||||
return "<descriptor " + self.__qualname__ + " at 0x" + format(id(self), "x") + ">"
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
from gi.repository import Gio
|
||||
from .exitable import ExitableWithAliases
|
||||
from .auto_names import *
|
||||
|
||||
class Publication(ExitableWithAliases("unpublish")):
|
||||
__slots__ = ()
|
||||
|
||||
def __init__(self, bus, bus_name, *objects, **kwargs): # allow_replacement=True, replace=False
|
||||
# Python 2 sux
|
||||
for kwarg in kwargs:
|
||||
if kwarg not in ("allow_replacement", "replace",):
|
||||
raise TypeError(self.__qualname__ + " got an unexpected keyword argument '{}'".format(kwarg))
|
||||
allow_replacement = kwargs.get("allow_replacement", True)
|
||||
replace = kwargs.get("replace", False)
|
||||
|
||||
bus_name = auto_bus_name(bus_name)
|
||||
|
||||
for object_info in objects:
|
||||
path, object, node_info = (None, None, None)
|
||||
|
||||
if type(object_info) == tuple:
|
||||
if len(object_info) == 3:
|
||||
path, object, node_info = object_info
|
||||
if len(object_info) == 2:
|
||||
path, object = object_info
|
||||
if len(object_info) == 1:
|
||||
object = object_info[0]
|
||||
else:
|
||||
object = object_info
|
||||
|
||||
path = auto_object_path(bus_name, path)
|
||||
self._at_exit(bus.register_object(path, object, node_info).__exit__)
|
||||
|
||||
# Request name only after registering all the objects.
|
||||
self._at_exit(bus.request_name(bus_name, allow_replacement=allow_replacement, replace=replace).__exit__)
|
||||
|
||||
class PublicationMixin(object):
|
||||
__slots__ = ()
|
||||
|
||||
def publish(self, bus_name, *objects):
|
||||
"""Expose objects on the bus."""
|
||||
return Publication(self, bus_name, *objects)
|
||||
|
|
@ -0,0 +1,156 @@
|
|||
from __future__ import print_function
|
||||
import sys, traceback
|
||||
from gi.repository import GLib, Gio
|
||||
from . import generic
|
||||
from .exitable import ExitableWithAliases
|
||||
from functools import partial
|
||||
from .method_call_context import MethodCallContext
|
||||
import logging
|
||||
|
||||
try:
|
||||
from inspect import signature, Parameter
|
||||
except:
|
||||
from ._inspect3 import signature, Parameter
|
||||
|
||||
class ObjectWrapper(ExitableWithAliases("unwrap")):
|
||||
__slots__ = ["object", "outargs", "readable_properties", "writable_properties"]
|
||||
|
||||
def __init__(self, object, interfaces):
|
||||
self.object = object
|
||||
|
||||
self.outargs = {}
|
||||
for iface in interfaces:
|
||||
for method in iface.methods:
|
||||
self.outargs[iface.name + "." + method.name] = [arg.signature for arg in method.out_args]
|
||||
|
||||
self.readable_properties = {}
|
||||
self.writable_properties = {}
|
||||
for iface in interfaces:
|
||||
for prop in iface.properties:
|
||||
if prop.flags & Gio.DBusPropertyInfoFlags.READABLE:
|
||||
self.readable_properties[iface.name + "." + prop.name] = prop.signature
|
||||
if prop.flags & Gio.DBusPropertyInfoFlags.WRITABLE:
|
||||
self.writable_properties[iface.name + "." + prop.name] = prop.signature
|
||||
|
||||
for iface in interfaces:
|
||||
for signal in iface.signals:
|
||||
s_name = signal.name
|
||||
def EmitSignal(iface, signal):
|
||||
return lambda *args: self.SignalEmitted(iface.name, signal.name, GLib.Variant("(" + "".join(s.signature for s in signal.args) + ")", args))
|
||||
self._at_exit(getattr(object, signal.name).connect(EmitSignal(iface, signal)).__exit__)
|
||||
|
||||
if "org.freedesktop.DBus.Properties" not in (iface.name for iface in interfaces):
|
||||
try:
|
||||
def onPropertiesChanged(iface, changed, invalidated):
|
||||
changed = {key: GLib.Variant(self.readable_properties[iface + "." + key], val) for key, val in changed.items()}
|
||||
args = GLib.Variant("(sa{sv}as)", (iface, changed, invalidated))
|
||||
self.SignalEmitted("org.freedesktop.DBus.Properties", "PropertiesChanged", args)
|
||||
self._at_exit(object.PropertiesChanged.connect(onPropertiesChanged).__exit__)
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
SignalEmitted = generic.signal()
|
||||
|
||||
def call_method(self, connection, sender, object_path, interface_name, method_name, parameters, invocation):
|
||||
try:
|
||||
try:
|
||||
outargs = self.outargs[interface_name + "." + method_name]
|
||||
method = getattr(self.object, method_name)
|
||||
except KeyError:
|
||||
if interface_name == "org.freedesktop.DBus.Properties":
|
||||
if method_name == "Get":
|
||||
method = self.Get
|
||||
outargs = ["v"]
|
||||
elif method_name == "GetAll":
|
||||
method = self.GetAll
|
||||
outargs = ["a{sv}"]
|
||||
elif method_name == "Set":
|
||||
method = self.Set
|
||||
outargs = []
|
||||
else:
|
||||
raise
|
||||
else:
|
||||
raise
|
||||
|
||||
sig = signature(method)
|
||||
|
||||
kwargs = {}
|
||||
if "dbus_context" in sig.parameters and sig.parameters["dbus_context"].kind in (Parameter.POSITIONAL_OR_KEYWORD, Parameter.KEYWORD_ONLY):
|
||||
kwargs["dbus_context"] = MethodCallContext(invocation)
|
||||
|
||||
result = method(*parameters, **kwargs)
|
||||
|
||||
if len(outargs) == 0:
|
||||
invocation.return_value(None)
|
||||
elif len(outargs) == 1:
|
||||
invocation.return_value(GLib.Variant("(" + "".join(outargs) + ")", (result,)))
|
||||
else:
|
||||
invocation.return_value(GLib.Variant("(" + "".join(outargs) + ")", result))
|
||||
|
||||
except Exception as e:
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.exception("Exception while handling %s.%s()", interface_name, method_name)
|
||||
|
||||
#TODO Think of a better way to translate Python exception types to DBus error types.
|
||||
e_type = type(e).__name__
|
||||
if not "." in e_type:
|
||||
e_type = "unknown." + e_type
|
||||
invocation.return_dbus_error(e_type, str(e))
|
||||
|
||||
def Get(self, interface_name, property_name):
|
||||
type = self.readable_properties[interface_name + "." + property_name]
|
||||
result = getattr(self.object, property_name)
|
||||
return GLib.Variant(type, result)
|
||||
|
||||
def GetAll(self, interface_name):
|
||||
ret = {}
|
||||
for name, type in self.readable_properties.items():
|
||||
ns, local = name.rsplit(".", 1)
|
||||
if ns == interface_name:
|
||||
ret[local] = GLib.Variant(type, getattr(self.object, local))
|
||||
return ret
|
||||
|
||||
def Set(self, interface_name, property_name, value):
|
||||
self.writable_properties[interface_name + "." + property_name]
|
||||
setattr(self.object, property_name, value)
|
||||
|
||||
class ObjectRegistration(ExitableWithAliases("unregister")):
|
||||
__slots__ = ()
|
||||
|
||||
def __init__(self, bus, path, interfaces, wrapper, own_wrapper=False):
|
||||
if own_wrapper:
|
||||
self._at_exit(wrapper.__exit__)
|
||||
|
||||
def func(interface_name, signal_name, parameters):
|
||||
bus.con.emit_signal(None, path, interface_name, signal_name, parameters)
|
||||
|
||||
self._at_exit(wrapper.SignalEmitted.connect(func).__exit__)
|
||||
|
||||
try:
|
||||
ids = [bus.con.register_object(path, interface, wrapper.call_method, None, None) for interface in interfaces]
|
||||
except TypeError as e:
|
||||
if str(e).startswith("argument vtable: Expected Gio.DBusInterfaceVTable"):
|
||||
raise Exception("GLib 2.46 is required to publish objects; it is impossible in older versions.")
|
||||
else:
|
||||
raise
|
||||
|
||||
self._at_exit(lambda: [bus.con.unregister_object(id) for id in ids])
|
||||
|
||||
class RegistrationMixin:
|
||||
__slots__ = ()
|
||||
|
||||
def register_object(self, path, object, node_info):
|
||||
if node_info is None:
|
||||
try:
|
||||
node_info = type(object).dbus
|
||||
except AttributeError:
|
||||
node_info = type(object).__doc__
|
||||
|
||||
if type(node_info) != list and type(node_info) != tuple:
|
||||
node_info = [node_info]
|
||||
|
||||
node_info = [Gio.DBusNodeInfo.new_for_xml(ni) for ni in node_info]
|
||||
interfaces = sum((ni.interfaces for ni in node_info), [])
|
||||
|
||||
wrapper = ObjectWrapper(object, interfaces)
|
||||
return ObjectRegistration(self, path, interfaces, wrapper, own_wrapper=True)
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
from .exitable import ExitableWithAliases
|
||||
|
||||
class NameOwner(ExitableWithAliases("unown")):
|
||||
__slots__ = ()
|
||||
|
||||
def __init__(self, bus, name, allow_replacement, replace):
|
||||
flags = 4 | (1 if allow_replacement else 0) | (2 if replace else 0)
|
||||
res = bus.dbus.RequestName(name, flags)
|
||||
if res == 1:
|
||||
self._at_exit(lambda: bus.dbus.ReleaseName(name))
|
||||
return # OK
|
||||
if res == 3:
|
||||
raise RuntimeError("name already exists on the bus")
|
||||
if res == 4:
|
||||
raise RuntimeError("you're already the owner of this name")
|
||||
raise RuntimeError("cannot take ownership of the name")
|
||||
|
||||
class RequestNameMixin(object):
|
||||
__slots__ = ()
|
||||
|
||||
def request_name(self, name, allow_replacement=True, replace=False):
|
||||
"""Aquires a bus name.
|
||||
|
||||
Returns
|
||||
-------
|
||||
NameOwner
|
||||
An object you can use as a context manager to unown the name later.
|
||||
"""
|
||||
return NameOwner(self, name, allow_replacement, replace)
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
from gi.repository import Gio
|
||||
from .exitable import ExitableWithAliases
|
||||
|
||||
class Subscription(ExitableWithAliases("unsubscribe", "disconnect")):
|
||||
Flags = Gio.DBusSignalFlags
|
||||
__slots__ = ()
|
||||
|
||||
def __init__(self, con, sender, iface, member, object, arg0, flags, callback):
|
||||
id = con.signal_subscribe(sender, iface, member, object, arg0, flags, callback)
|
||||
self._at_exit(lambda: con.signal_unsubscribe(id))
|
||||
|
||||
class SubscriptionMixin(object):
|
||||
__slots__ = ()
|
||||
SubscriptionFlags = Subscription.Flags
|
||||
|
||||
def subscribe(self, sender=None, iface=None, signal=None, object=None, arg0=None, flags=0, signal_fired=None):
|
||||
"""Subscribes to matching signals.
|
||||
|
||||
Subscribes to signals on connection and invokes signal_fired callback
|
||||
whenever the signal is received.
|
||||
|
||||
To receive signal_fired callback, you need an event loop.
|
||||
https://github.com/LEW21/pydbus/blob/master/doc/tutorial.rst#setting-up-an-event-loop
|
||||
|
||||
Parameters
|
||||
----------
|
||||
sender : string, optional
|
||||
Sender name to match on (unique or well-known name) or None to listen from all senders.
|
||||
iface : string, optional
|
||||
Interface name to match on or None to match on all interfaces.
|
||||
signal : string, optional
|
||||
Signal name to match on or None to match on all signals.
|
||||
object : string, optional
|
||||
Object path to match on or None to match on all object paths.
|
||||
arg0 : string, optional
|
||||
Contents of first string argument to match on or None to match on all kinds of arguments.
|
||||
flags : SubscriptionFlags, optional
|
||||
signal_fired : callable, optional
|
||||
Invoked when there is a signal matching the requested data.
|
||||
Parameters: sender, object, iface, signal, params
|
||||
|
||||
Returns
|
||||
-------
|
||||
Subscription
|
||||
An object you can use as a context manager to unsubscribe from the signal later.
|
||||
|
||||
See Also
|
||||
--------
|
||||
See https://developer.gnome.org/gio/2.44/GDBusConnection.html#g-dbus-connection-signal-subscribe
|
||||
for more information.
|
||||
"""
|
||||
callback = (lambda con, sender, object, iface, signal, params: signal_fired(sender, object, iface, signal, params.unpack())) if signal_fired is not None else lambda *args: None
|
||||
return Subscription(self.con, sender, iface, signal, object, arg0, flags, callback)
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
from gi.repository import GLib, GObject
|
||||
|
||||
def timeout_to_glib(timeout):
|
||||
if timeout is None:
|
||||
try:
|
||||
return GLib.MAXINT
|
||||
except AttributeError:
|
||||
# GLib < 2.46
|
||||
return GObject.G_MAXINT
|
||||
else:
|
||||
try:
|
||||
timeout = timeout.total_seconds()
|
||||
except AttributeError:
|
||||
pass
|
||||
return int(timeout * 1000)
|
||||
|
|
@ -17,12 +17,16 @@
|
|||
#
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
lib_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'lib')
|
||||
sys.path.insert(0, lib_dir)
|
||||
|
||||
import gettext
|
||||
import gi
|
||||
import locale
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import argparse
|
||||
|
||||
from logging.handlers import TimedRotatingFileHandler
|
||||
|
|
@ -41,7 +45,6 @@ locale.bindtextdomain('breezydesktop', locale_dir)
|
|||
gettext.bindtextdomain('breezydesktop', locale_dir)
|
||||
gettext.textdomain('breezydesktop')
|
||||
|
||||
|
||||
from gi.repository import Adw, Gtk, Gio
|
||||
from .licensedialog import LicenseDialog
|
||||
from .statemanager import StateManager
|
||||
|
|
|
|||
Loading…
Reference in New Issue