Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@Coroutine.from_generator_function
def update(self, object_path):
device = yield self._get_updated_device(object_path)
if device is not None:
self._devices[object_path] = device
yield Return(device)
@Coroutine.from_generator_function
def get_property_interface(self, interface_name=None):
proxy = yield self._get_interface(PropertiesProxy.Interface)
yield Return(PropertiesProxy(proxy, interface_name))
@Coroutine.from_generator_function
def call(self, interface_name, method_name, signature='()', *args):
proxy = yield self.get_interface(interface_name)
result = yield proxy.call(method_name, signature, *args)
yield Return(result)
@Coroutine.from_generator_function
def _get_updated_device(self, object_path):
obj = self._proxy.object.bus.get_object(object_path)
try:
prop_prox = self._property_proxy[object_path]
except KeyError:
prop_prox = yield obj.get_property_interface(Device.Interface)
self._property_proxy[object_path] = prop_prox
try:
properties = yield prop_prox.GetAll()
except GLib.GError:
self._invalidate(object_path)
yield Return(None)
# TODO: return something useful? (NullDevice)
else:
itfc_prox = MethodsProxy(obj, Device.Interface)
@Coroutine.from_generator_function
def get_version(cls):
service = (cls.BusName, cls.ObjectPath,
'org.freedesktop.DBus.Properties')
manager = yield connect_service(*service)
version = yield DBusCall(manager._proxy, 'Get', '(ss)', (
cls.Interface, 'DaemonVersion'))
yield Return(version)
@Coroutine.from_generator_function
def _sync(self):
"""Cache all device states."""
object_pathes = yield self._proxy.call('EnumerateDevices', '()')
yield AsyncList(map(self.update, object_pathes))
yield Return()