1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
from ..proxy_object import BaseProxyObject, BaseProxyInterface
from ..message_bus import BaseMessageBus
from ..message import Message, MessageFlag
from ..signature import Variant
from ..errors import DBusError
from ..constants import ErrorType
from .._private.util import replace_idx_with_fds, replace_fds_with_idx
from .. import introspection as intr
import xml.etree.ElementTree as ET
from typing import Union, List
class ProxyInterface(BaseProxyInterface):
"""A class representing a proxy to an interface exported on the bus by
another client for the asyncio :class:`MessageBus
<dbus_next.aio.MessageBus>` implementation.
This class is not meant to be constructed directly by the user. Use
:func:`ProxyObject.get_interface()
<dbus_next.aio.ProxyObject.get_interface>` on a asyncio proxy object to get
a proxy interface.
This class exposes methods to call DBus methods, listen to signals, and get
and set properties on the interface that are created dynamically based on
the introspection data passed to the proxy object that made this proxy
interface.
A *method call* takes this form:
.. code-block:: python3
result = await interface.call_[METHOD](*args)
Where ``METHOD`` is the name of the method converted to snake case.
DBus methods are exposed as coroutines that take arguments that correpond
to the *in args* of the interface method definition and return a ``result``
that corresponds to the *out arg*. If the method has more than one out arg,
they are returned within a :class:`list`.
To *listen to a signal* use this form:
.. code-block:: python3
interface.on_[SIGNAL](callback)
To *stop listening to a signal* use this form:
.. code-block:: python3
interface.off_[SIGNAL](callback)
Where ``SIGNAL`` is the name of the signal converted to snake case.
DBus signals are exposed with an event-callback interface. The provided
``callback`` will be called when the signal is emitted with arguments that
correspond to the *out args* of the interface signal definition.
To *get or set a property* use this form:
.. code-block:: python3
value = await interface.get_[PROPERTY]()
await interface.set_[PROPERTY](value)
Where ``PROPERTY`` is the name of the property converted to snake case.
DBus property getters and setters are exposed as coroutines. The ``value``
must correspond to the type of the property in the interface definition.
If the service returns an error for a DBus call, a :class:`DBusError
<dbus_next.DBusError>` will be raised with information about the error.
"""
def _add_method(self, intr_method):
async def method_fn(*args, flags=MessageFlag.NONE):
input_body, unix_fds = replace_fds_with_idx(intr_method.in_signature, list(args))
msg = await self.bus.call(
Message(destination=self.bus_name,
path=self.path,
interface=self.introspection.name,
member=intr_method.name,
signature=intr_method.in_signature,
body=input_body,
flags=flags,
unix_fds=unix_fds))
if flags & MessageFlag.NO_REPLY_EXPECTED:
return None
BaseProxyInterface._check_method_return(msg, intr_method.out_signature)
out_len = len(intr_method.out_args)
body = replace_idx_with_fds(msg.signature_tree, msg.body, msg.unix_fds)
if not out_len:
return None
elif out_len == 1:
return body[0]
else:
return body
method_name = f'call_{BaseProxyInterface._to_snake_case(intr_method.name)}'
setattr(self, method_name, method_fn)
def _add_property(self, intr_property):
async def property_getter():
msg = await self.bus.call(
Message(destination=self.bus_name,
path=self.path,
interface='org.freedesktop.DBus.Properties',
member='Get',
signature='ss',
body=[self.introspection.name, intr_property.name]))
BaseProxyInterface._check_method_return(msg, 'v')
variant = msg.body[0]
if variant.signature != intr_property.signature:
raise DBusError(ErrorType.CLIENT_ERROR,
f'property returned unexpected signature "{variant.signature}"',
msg)
return replace_idx_with_fds('v', msg.body, msg.unix_fds)[0].value
async def property_setter(val):
variant = Variant(intr_property.signature, val)
body, unix_fds = replace_fds_with_idx(
'ssv', [self.introspection.name, intr_property.name, variant])
msg = await self.bus.call(
Message(destination=self.bus_name,
path=self.path,
interface='org.freedesktop.DBus.Properties',
member='Set',
signature='ssv',
body=body,
unix_fds=unix_fds))
BaseProxyInterface._check_method_return(msg)
snake_case = BaseProxyInterface._to_snake_case(intr_property.name)
setattr(self, f'get_{snake_case}', property_getter)
setattr(self, f'set_{snake_case}', property_setter)
class ProxyObject(BaseProxyObject):
"""The proxy object implementation for the GLib :class:`MessageBus <dbus_next.glib.MessageBus>`.
For more information, see the :class:`BaseProxyObject <dbus_next.proxy_object.BaseProxyObject>`.
"""
def __init__(self, bus_name: str, path: str, introspection: Union[intr.Node, str, ET.Element],
bus: BaseMessageBus):
super().__init__(bus_name, path, introspection, bus, ProxyInterface)
def get_interface(self, name: str) -> ProxyInterface:
return super().get_interface(name)
def get_children(self) -> List['ProxyObject']:
return super().get_children()
|
