Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def parse_msg(message):
if message.member == "InterfacesAdded":
msg_path = message.body[0]
try:
device_interface = message.body[1].get("org.bluez.Device1", {})
except Exception as e:
raise e
devices[msg_path] = (
{**devices[msg_path], **device_interface}
if msg_path in devices
else device_interface
)
elif message.member == "PropertiesChanged":
iface, changed, invalidated = message.body
if iface != defs.DEVICE_INTERFACE:
return
msg_path = message.path
# the PropertiesChanged signal only sends changed properties, so we
# need to get remaining properties from cached_devices. However, we
# don't want to add all cached_devices to the devices dict since
# they may not actually be nearby or powered on.
if msg_path not in devices and msg_path in cached_devices:
devices[msg_path] = cached_devices[msg_path]
devices[msg_path] = (
{**devices[msg_path], **changed} if msg_path in devices else changed
)
elif (
message.member == "InterfacesRemoved"
and message.body[1][0] == defs.BATTERY_INTERFACE
):
def _services_resolved_callback(message):
iface, changed, invalidated = message.body
is_resolved = defs.DEVICE_INTERFACE and changed.get(
"ServicesResolved", False
)
if iface == is_resolved:
logger.info("Services resolved.")
self.services_resolved = True
async def _get_device_properties(self, interface=defs.DEVICE_INTERFACE) -> dict:
"""Get properties of the connected device.
Args:
interface: Which DBus interface to get properties on. Defaults to `org.bluez.Device1`.
Returns:
(dict) The properties.
"""
return await self._bus.callRemote(
self._device_path,
"GetAll",
interface=defs.PROPERTIES_INTERFACE,
destination=defs.BLUEZ_SERVICE,
signature="s",
body=[interface],