Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
device = Device(self, object_path,
AttrDictView(properties), itfc_prox)
yield Return(device)
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)
def create(cls):
service = (cls.BusName, cls.ObjectPath, cls.Interface)
proxy = yield connect_service(*service)
version = yield cls.get_version()
daemon = cls(proxy, version)
yield daemon._sync()
yield Return(daemon)
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)
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)
def get_interface(self, name):
"""
Get an interface proxy for this Dbus object.
:param str name: interface name
:returns: a proxy object for the other interface
:rtype: InterfaceProxy
"""
proxy = yield self._get_interface(name)
yield Return(InterfaceProxy(proxy))
def _sync(self):
"""Cache all device states."""
object_pathes = yield self._proxy.call('EnumerateDevices', '()')
yield AsyncList(map(self.update, object_pathes))
yield Return()