Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def get_services(self) -> BleakGATTServiceCollection:
"""Get all services registered for this GATT server.
Returns:
A :py:class:`bleak.backends.service.BleakGATTServiceCollection` with this device's services tree.
"""
if self._services is not None:
return self._services
logger.debug("Retrieving services...")
services = (
await cbapp.central_manager_delegate.connected_peripheral_delegate.discoverServices()
)
for service in services:
serviceUUID = service.UUID().UUIDString()
logger.debug(
"Retrieving characteristics for service {}".format(serviceUUID)
)
characteristics = await cbapp.central_manager_delegate.connected_peripheral_delegate.discoverCharacteristics_(
service
)
self.services.add_service(BleakGATTServiceCoreBluetooth(service))
for characteristic in characteristics:
cUUID = characteristic.UUID().UUIDString()
logger.debug(
if not cbapp.central_manager_delegate.enabled:
raise BleakError("Bluetooth device is turned off")
scan_options = {"timeout": timeout}
await cbapp.central_manager_delegate.scanForPeripherals_(scan_options)
# CoreBluetooth doesn't explicitly use MAC addresses to identify peripheral
# devices because private devices may obscure their MAC addresses. To cope
# with this, CoreBluetooth utilizes UUIDs for each peripheral. We'll use
# this for the BLEDevice address on macOS
found = []
peripherals = cbapp.central_manager_delegate.peripheral_list
for i, peripheral in enumerate(peripherals):
address = peripheral.identifier().UUIDString()
name = peripheral.name() or "Unknown"
details = peripheral
advertisementData = cbapp.central_manager_delegate.advertisement_data_list[i]
manufacturer_binary_data = (
advertisementData["kCBAdvDataManufacturerData"]
if "kCBAdvDataManufacturerData" in advertisementData.keys()
else None
)
manufacturer_data = {}
if manufacturer_binary_data:
manufacturer_id = int.from_bytes(
manufacturer_binary_data[0:2], byteorder="little"
timeout = kwargs.get("timeout", self._timeout)
devices = await discover(timeout=timeout, loop=self.loop)
sought_device = list(
filter(lambda x: x.address.upper() == self.address.upper(), devices)
)
if len(sought_device):
self._device_info = sought_device[0].details
else:
raise BleakError(
"Device with address {} was not found".format(self.address)
)
logger.debug("Connecting to BLE device @ {}".format(self.address))
await cbapp.central_manager_delegate.connect_(sought_device[0].details)
# Now get services
await self.get_services()
return True
async def discover(
timeout: float = 5.0, loop: AbstractEventLoop = None, **kwargs
) -> List[BLEDevice]:
"""Perform a Bluetooth LE Scan.
Args:
timeout (float): duration of scaning period
loop (Event Loop): Event Loop to use
"""
loop = loop if loop else asyncio.get_event_loop()
devices = {}
if not cbapp.central_manager_delegate.enabled:
raise BleakError("Bluetooth device is turned off")
scan_options = {"timeout": timeout}
await cbapp.central_manager_delegate.scanForPeripherals_(scan_options)
# CoreBluetooth doesn't explicitly use MAC addresses to identify peripheral
# devices because private devices may obscure their MAC addresses. To cope
# with this, CoreBluetooth utilizes UUIDs for each peripheral. We'll use
# this for the BLEDevice address on macOS
found = []
peripherals = cbapp.central_manager_delegate.peripheral_list
for i, peripheral in enumerate(peripherals):