Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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(
"Retrieving descriptors for characteristic {}".format(cUUID)
)
descriptors = await cbapp.central_manager_delegate.connected_peripheral_delegate.discoverDescriptors_(
characteristic
)
self.services.add_characteristic(
BleakGATTCharacteristicCoreBluetooth(characteristic)
)
for descriptor in descriptors:
self.services.add_descriptor(
BleakGATTDescriptorCoreBluetooth(
descriptor, characteristic.UUID().UUIDString()
)
)
self._services_resolved = True
self._services = services
return self.services
def callback(sender, data):
print(f"{sender}: {data}")
client.start_notify(char_uuid, callback)
Args:
_uuid (str or UUID): The uuid of the characteristics to start notification/indication on.
callback (function): The function to be called on notification.
"""
_uuid = await self.get_appropriate_uuid(_uuid)
characteristic = self.services.get_characteristic(_uuid)
if not characteristic:
raise BleakError("Characteristic {} not found!".format(_uuid))
success = await cbapp.central_manager_delegate.connected_peripheral_delegate.startNotify_cb_(
characteristic.obj, callback
)
if not success:
raise BleakError(
"Could not start notify on {0}: {1}".format(
characteristic.uuid, success
)
async def write_gatt_descriptor(self, handle: int, data: bytearray) -> None:
"""Perform a write operation on the specified GATT descriptor.
Args:
handle (int): The handle of the descriptor to read from.
data (bytes or bytearray): The data to send.
"""
descriptor = self.services.get_descriptor(handle)
if not descriptor:
raise BleakError("Descriptor {} was not found!".format(handle))
value = NSData.alloc().initWithBytes_length_(data, len(data))
success = await cbapp.central_manager_delegate.connected_peripheral_delegate.writeDescriptor_value_(
descriptor.obj, value
)
if success:
logger.debug("Write Descriptor {0} : {1}".format(handle, data))
else:
raise BleakError(
"Could not write value {0} to descriptor {1}: {2}".format(
data, descriptor.uuid, success
)
) -> None:
"""Perform a write operation of the specified GATT characteristic.
Args:
_uuid (str or UUID): The uuid of the characteristics to write to.
data (bytes or bytearray): The data to send.
response (bool): If write-with-response operation should be done. Defaults to `False`.
"""
_uuid = await self.get_appropriate_uuid(_uuid)
characteristic = self.services.get_characteristic(_uuid)
if not characteristic:
raise BleakError("Characteristic {} was not found!".format(_uuid))
value = NSData.alloc().initWithBytes_length_(data, len(data))
success = await cbapp.central_manager_delegate.connected_peripheral_delegate.writeCharacteristic_value_(
characteristic.obj, value
)
if success:
logger.debug("Write Characteristic {0} : {1}".format(_uuid, data))
else:
raise BleakError(
"Could not write value {0} to characteristic {1}: {2}".format(
data, characteristic.uuid, success
)
Args:
_uuid (str or UUID): The uuid of the characteristics to read from.
use_cached (bool): `False` forces macOS to read the value from the
device again and not use its own cached value. Defaults to `False`.
Returns:
(bytearray) The read data.
"""
_uuid = await self.get_appropriate_uuid(_uuid)
characteristic = self.services.get_characteristic(_uuid)
if not characteristic:
raise BleakError("Characteristic {} was not found!".format(_uuid))
output = await cbapp.central_manager_delegate.connected_peripheral_delegate.readCharacteristic_(
characteristic.obj, use_cached=use_cached
)
value = bytearray(output)
logger.debug("Read Characteristic {0} : {1}".format(_uuid, value))
return value
) -> bytearray:
"""Perform read operation on the specified GATT descriptor.
Args:
handle (int): The handle of the descriptor to read from.
use_cached (bool): `False` forces Windows to read the value from the
device again and not use its own cached value. Defaults to `False`.
Returns:
(bytearray) The read data.
"""
descriptor = self.services.get_descriptor(handle)
if not descriptor:
raise BleakError("Descriptor {} was not found!".format(handle))
output = await cbapp.central_manager_delegate.connected_peripheral_delegate.readDescriptor_(
descriptor.obj, use_cached=use_cached
)
if isinstance(
output, str
): # Sometimes a `pyobjc_unicode`or `__NSCFString` is returned and they can be used as regular Python strings.
value = bytearray(output.encode("utf-8"))
else: # _NSInlineData
value = bytearray(output) # value.getBytes_length_(None, len(value))
logger.debug("Read Descriptor {0} : {1}".format(handle, value))
return value
"""Internal method performing call to BleakUWPBridge method.
Args:
characteristic_obj: The Managed Windows.Devices.Bluetooth.GenericAttributeProfile.GattCharacteristic Object
callback: The function to be called on notification.
Returns:
(int) The GattCommunicationStatus of the operation.
"""
_uuid = await self.get_appropriate_uuid(_uuid)
characteristic = self.services.get_characteristic(_uuid)
if not characteristic:
raise BleakError("Characteristic {} not found!".format(_uuid))
success = await cbapp.central_manager_delegate.connected_peripheral_delegate.stopNotify_(
characteristic.obj
)
if not success:
raise BleakError(
"Could not stop notify on {0}: {1}".format(characteristic.uuid, success)
)