auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-04-16 00:25:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #13 from google/gbg/standard-profiles

Browse Source 
support for type adapters and framework for standard GATT profiles
This commit is contained in:
Gilles Boccon-Gibod
2022-07-22 10:21:39 -07:00
committed by GitHub
parent 0ffed3deff e156ed3758
commit c66b357de6
19 changed files with 936 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -8,3 +8,4 @@ docs/mkdocs/site
tests/__pycache__
test-results.xml
bumble/transport/__pycache__
bumble/profiles/__pycache__

6
apps/gatt_dump.py
View File

@@ -32,10 +32,10 @@ async def dump_gatt_db(peer, done):
# Discover all services
print(color('### Discovering Services and Characteristics', 'magenta'))
await peer.discover_services()
await peer.discover_characteristics()
for service in peer.services:
await service.discover_characteristics()
for characteristic in service.characteristics:
await peer.discover_descriptors(characteristic)
await characteristic.discover_descriptors()
print(color('=== Services ===', 'yellow'))
show_services(peer.services)
@@ -47,7 +47,7 @@ async def dump_gatt_db(peer, done):
for attribute in attributes:
print(attribute)
try:
value = await peer.read_value(attribute)
value = await attribute.read_value()
print(color(f'{value.hex()}', 'green'))
except ProtocolError as error:
print(color(error, 'red'))

2
apps/gg_bridge.py
View File

@@ -73,7 +73,7 @@ class GattlinkHubBridge(Device.Listener):
gattlink_service = services[0]
# Discover all the characteristics for the service
characteristics = await self.peer.discover_characteristics(service = gattlink_service)
characteristics = await gattlink_service.discover_characteristics()
print(color('=== Characteristics discovered', 'yellow'))
for characteristic in characteristics:
if characteristic.uuid == GG_GATTLINK_RX_CHARACTERISTIC_UUID:

26
bumble/att.py
View File

@@ -682,11 +682,14 @@ class Attribute(EventEmitter):
def __init__(self, attribute_type, permissions, value = b''):
EventEmitter.__init__(self)
self.handle = 0
self.permissions = permissions
self.handle = 0
self.end_group_handle = 0
self.permissions = permissions
# Convert the type to a UUID
if type(attribute_type) is bytes:
# Convert the type to a UUID object if it isn't already
if type(attribute_type) is str:
self.type = UUID(attribute_type)
elif type(attribute_type) is bytes:
self.type = UUID.from_bytes(attribute_type)
else:
self.type = attribute_type
@@ -698,16 +701,13 @@ class Attribute(EventEmitter):
self.value = value
def read_value(self, connection):
if type(self.value) is bytes:
return self.value
if read := getattr(self.value, 'read', None):
try:
return read(connection)
except ATT_Error as error:
raise ATT_Error(error_code=error.error_code, att_handle=self.handle)
else:
if read := getattr(self.value, 'read', None):
try:
return read(connection)
except ATT_Error as error:
raise ATT_Error(error_code=error.error_code, att_handle=self.handle)
else:
return bytes(self.value)
return self.value
def write_value(self, connection, value):
if write := getattr(self.value, 'write', None):

11
bumble/device.py
View File

@@ -137,6 +137,17 @@ class Peer:
def get_characteristics_by_uuid(self, uuid, service = None):
return self.gatt_client.get_characteristics_by_uuid(uuid, service)
def create_service_proxy(self, proxy_class):
return proxy_class.from_client(self.gatt_client)
async def discover_service_and_create_proxy(self, proxy_class):
# Discover the first matching service and its characteristics
services = await self.discover_service(proxy_class.SERVICE_CLASS.UUID)
if services:
service = services[0]
await service.discover_characteristics()
return self.create_service_proxy(proxy_class)
# [Classic only]
async def request_name(self):
return await self.connection.request_remote_name()

250
bumble/gatt.py
View File

@@ -22,6 +22,8 @@
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import types
import logging
from colors import color
@@ -53,13 +55,13 @@ GATT_NEXT_DST_CHANGE_SERVICE = UUID.from_16_bits(0x1807, 'Next DS
GATT_GLUCOSE_SERVICE = UUID.from_16_bits(0x1808, 'Glucose')
GATT_HEALTH_THERMOMETER_SERVICE = UUID.from_16_bits(0x1809, 'Health Thermometer')
GATT_DEVICE_INFORMATION_SERVICE = UUID.from_16_bits(0x180A, 'Device Information')
GATT_DEVICE_HEART_RATE_SERVICE = UUID.from_16_bits(0x180D, 'Heart Rate')
GATT_PHONE_ALTERT_STATUS_SERVICE = UUID.from_16_bits(0x180E, 'Phone Alert Status')
GATT_DEVICE_BATTERY_SERVICE = UUID.from_16_bits(0x180F, 'Battery')
GATT_HEART_RATE_SERVICE = UUID.from_16_bits(0x180D, 'Heart Rate')
GATT_PHONE_ALERT_STATUS_SERVICE = UUID.from_16_bits(0x180E, 'Phone Alert Status')
GATT_BATTERY_SERVICE = UUID.from_16_bits(0x180F, 'Battery')
GATT_BLOOD_PRESSURE_SERVICE = UUID.from_16_bits(0x1810, 'Blood Pressure')
GATT_ALTERT_NOTIFICATION_SERVICE = UUID.from_16_bits(0x1811, 'Alert Notification')
GATT_DEVICE_HUMAN_INTERFACE_DEVICE_SERVICE = UUID.from_16_bits(0x1812, 'Human Interface Device')
GATT_DEVICE_SCAN_PARAMETERS_SERVICE = UUID.from_16_bits(0x1813, 'Scan Parameters')
GATT_ALERT_NOTIFICATION_SERVICE = UUID.from_16_bits(0x1811, 'Alert Notification')
GATT_HUMAN_INTERFACE_DEVICE_SERVICE = UUID.from_16_bits(0x1812, 'Human Interface Device')
GATT_SCAN_PARAMETERS_SERVICE = UUID.from_16_bits(0x1813, 'Scan Parameters')
GATT_RUNNING_SPEED_AND_CADENCE_SERVICE = UUID.from_16_bits(0x1814, 'Running Speed and Cadence')
GATT_AUTOMATION_IO_SERVICE = UUID.from_16_bits(0x1815, 'Automation IO')
GATT_CYCLING_SPEED_AND_CADENCE_SERVICE = UUID.from_16_bits(0x1816, 'Cycling Speed and Cadence')
@@ -119,7 +121,7 @@ GATT_ENVIRONMENTAL_SENSING_CONFIGURATION_DESCRIPTOR = UUID.from_16_bits(0x290B,
GATT_ENVIRONMENTAL_SENSING_MEASUREMENT_DESCRIPTOR = UUID.from_16_bits(0x290C, 'Environmental Sensing Measurement')
GATT_ENVIRONMENTAL_SENSING_TRIGGER_DESCRIPTOR = UUID.from_16_bits(0x290D, 'Environmental Sensing Trigger Setting')
GATT_TIME_TRIGGER_DESCRIPTOR = UUID.from_16_bits(0x290E, 'Time Trigger Setting')
GATT_COMPLETE_BE_EDR_TRANSPORT_BLOCK_DATA_DESCRIPTOR = UUID.from_16_bits(0x290F, 'Complete BR-EDR Transport Block Data')
GATT_COMPLETE_BR_EDR_TRANSPORT_BLOCK_DATA_DESCRIPTOR = UUID.from_16_bits(0x290F, 'Complete BR-EDR Transport Block Data')
# Device Information Service
GATT_SYSTEM_ID_CHARACTERISTIC = UUID.from_16_bits(0x2A23, 'System ID')
@@ -140,19 +142,19 @@ GATT_REPORT_CHARACTERISTIC = UUID.from_16_bits(0x2A4D, 'Report')
GATT_PROTOCOL_MODE_CHARACTERISTIC = UUID.from_16_bits(0x2A4E, 'Protocol Mode')
# Misc
GATT_DEVICE_NAME_CHARACTERISTIC = UUID.from_16_bits(0x2A00, 'Device Name')
GATT_APPEARANCE_CHARACTERISTIC = UUID.from_16_bits(0x2A01, 'Appearance')
GATT_PERIPHERAL_PRIVACY_FLAG_CHARACTERISTIC = UUID.from_16_bits(0x2A02, 'Peripheral Privacy Flag')
GATT_RECONNECTION_ADDRESS_CHARACTERISTIC = UUID.from_16_bits(0x2A03, 'Reconnection Address')
GATT_PERIPHERAL_PREFERRREED_CONNECTION_PARAMETERS_CHARACTERISTIC = UUID.from_16_bits(0x2A04, 'Peripheral Preferred Connection Parameters')
GATT_SERVICE_CHANGED_CHARACTERISTIC = UUID.from_16_bits(0x2A05, 'Service Changed')
GATT_ALERT_LEVEL_CHARACTERISTIC = UUID.from_16_bits(0x2A06, 'Alert Level')
GATT_TX_POWER_LEVEL_CHARACTERISTIC = UUID.from_16_bits(0x2A07, 'Tx Power Level')
GATT_BATTERY_LEVEL_CHARACTERISTIC = UUID.from_16_bits(0x2A19, 'Battery Level')
GATT_BOOT_KEYBOARD_INPUT_REPORT_CHARACTERISTIC = UUID.from_16_bits(0x2A22, 'Boot Keyboard Input Report')
GATT_CURRENT_TIME_CHARACTERISTIC = UUID.from_16_bits(0x2A2B, 'Current Time')
GATT_BOOT_KEYBOARD_OUTPUT_REPORT_CHARACTERISTIC = UUID.from_16_bits(0x2A32, 'Boot Keyboard Output Report')
GATT_CENTRAL_ADDRESS_RESOLUTION__CHARACTERISTIC = UUID.from_16_bits(0x2AA6, 'Central Address Resolution')
GATT_DEVICE_NAME_CHARACTERISTIC = UUID.from_16_bits(0x2A00, 'Device Name')
GATT_APPEARANCE_CHARACTERISTIC = UUID.from_16_bits(0x2A01, 'Appearance')
GATT_PERIPHERAL_PRIVACY_FLAG_CHARACTERISTIC = UUID.from_16_bits(0x2A02, 'Peripheral Privacy Flag')
GATT_RECONNECTION_ADDRESS_CHARACTERISTIC = UUID.from_16_bits(0x2A03, 'Reconnection Address')
GATT_PERIPHERAL_PREFERRED_CONNECTION_PARAMETERS_CHARACTERISTIC = UUID.from_16_bits(0x2A04, 'Peripheral Preferred Connection Parameters')
GATT_SERVICE_CHANGED_CHARACTERISTIC = UUID.from_16_bits(0x2A05, 'Service Changed')
GATT_ALERT_LEVEL_CHARACTERISTIC = UUID.from_16_bits(0x2A06, 'Alert Level')
GATT_TX_POWER_LEVEL_CHARACTERISTIC = UUID.from_16_bits(0x2A07, 'Tx Power Level')
GATT_BATTERY_LEVEL_CHARACTERISTIC = UUID.from_16_bits(0x2A19, 'Battery Level')
GATT_BOOT_KEYBOARD_INPUT_REPORT_CHARACTERISTIC = UUID.from_16_bits(0x2A22, 'Boot Keyboard Input Report')
GATT_CURRENT_TIME_CHARACTERISTIC = UUID.from_16_bits(0x2A2B, 'Current Time')
GATT_BOOT_KEYBOARD_OUTPUT_REPORT_CHARACTERISTIC = UUID.from_16_bits(0x2A32, 'Boot Keyboard Output Report')
GATT_CENTRAL_ADDRESS_RESOLUTION__CHARACTERISTIC = UUID.from_16_bits(0x2AA6, 'Central Address Resolution')
# -----------------------------------------------------------------------------
@@ -189,13 +191,24 @@ class Service(Attribute):
self.uuid = uuid
self.included_services = []
self.characteristics = characteristics[:]
self.end_group_handle = 0
self.primary = primary
def __str__(self):
return f'Service(handle=0x{self.handle:04X}, end=0x{self.end_group_handle:04X}, uuid={self.uuid}){"" if self.primary else "*"}'
# -----------------------------------------------------------------------------
class TemplateService(Service):
'''
Convenience abstract class that can be used by profile-specific subclasses that want
to expose their UUID as a class property
'''
UUID = None
def __init__(self, characteristics, primary=True):
super().__init__(self.UUID, characteristics, primary)
# -----------------------------------------------------------------------------
class Characteristic(Attribute):
'''
@@ -227,56 +240,34 @@ class Characteristic(Attribute):
def property_name(property):
return Characteristic.PROPERTY_NAMES.get(property, '')
@staticmethod
def properties_as_string(properties):
return ','.join([
Characteristic.property_name(p) for p in Characteristic.PROPERTY_NAMES.keys()
if properties & p
])
def __init__(self, uuid, properties, permissions, value = b'', descriptors = []):
# Convert the uuid to a UUID object if it isn't already
if type(uuid) is str:
uuid = UUID(uuid)
super().__init__(uuid, permissions, value)
self.uuid = uuid
self.properties = properties
self._descriptors = descriptors
self._descriptors_discovered = False
self.end_group_handle = 0
self.attach_descriptors()
def attach_descriptors(self):
""" Let all the descriptors know they are attached to this characteristic """
for descriptor in self._descriptors:
descriptor.characteristic = self
def add_descriptor(self, descriptor):
descriptor.characteristic = self
self.descriptors.append(descriptor)
self.uuid = self.type
self.properties = properties
self.descriptors = descriptors
def get_descriptor(self, descriptor_type):
for descriptor in self.descriptors:
if descriptor.uuid == descriptor_type:
return descriptor
@property
def descriptors(self):
return self._descriptors
@descriptors.setter
def descriptors(self, value):
self._descriptors = value
self._descriptors_discovered = True
self.attach_descriptors()
@property
def descriptors_discovered(self):
return self._descriptors_discovered
def get_properties_as_string(self):
return ','.join([self.property_name(p) for p in self.PROPERTY_NAMES.keys() if self.properties & p])
def __str__(self):
return f'Characteristic(handle=0x{self.handle:04X}, end=0x{self.end_group_handle:04X}, uuid={self.uuid}, properties={self.get_properties_as_string()})'
return f'Characteristic(handle=0x{self.handle:04X}, end=0x{self.end_group_handle:04X}, uuid={self.uuid}, properties={Characteristic.properties_as_string(self.properties)})'
# -----------------------------------------------------------------------------
class CharacteristicValue:
'''
Characteristic value where reading and/or writing is delegated to functions
passed as arguments to the constructor.
'''
def __init__(self, read=None, write=None):
self._read = read
self._write = write
@@ -289,20 +280,145 @@ class CharacteristicValue:
self._write(connection, value)
# -----------------------------------------------------------------------------
class CharacteristicAdapter:
'''
An adapter that can adapt any object with `read_value` and `write_value`
methods (like Characteristic and CharacteristicProxy objects) by wrapping
those methods with ones that return/accept encoded/decoded values.
Objects with async methods are considered proxies, so the adaptation is one
where the return value of `read_value` is decoded and the value passed to
`write_value` is encoded. Other objects are considered local characteristics
so the adaptation is one where the return value of `read_value` is encoded
and the value passed to `write_value` is decoded.
If the characteristic has a `subscribe` method, it is wrapped with one where
the values are decoded before being passed to the subscriber.
'''
def __init__(self, characteristic):
self.wrapped_characteristic = characteristic
if (
asyncio.iscoroutinefunction(characteristic.read_value) and
asyncio.iscoroutinefunction(characteristic.write_value)
):
self.read_value = self.read_decoded_value
self.write_value = self.write_decoded_value
else:
self.read_value = self.read_encoded_value
self.write_value = self.write_encoded_value
if hasattr(self.wrapped_characteristic, 'subscribe'):
self.subscribe = self.wrapped_subscribe
def __getattr__(self, name):
return getattr(self.wrapped_characteristic, name)
def read_encoded_value(self, connection):
return self.encode_value(self.wrapped_characteristic.read_value(connection))
def write_encoded_value(self, connection, value):
return self.wrapped_characteristic.write_value(connection, self.decode_value(value))
async def read_decoded_value(self):
return self.decode_value(await self.wrapped_characteristic.read_value())
async def write_decoded_value(self, value):
return await self.wrapped_characteristic.write_value(self.encode_value(value))
def encode_value(self, value):
return value
def decode_value(self, value):
return value
def wrapped_subscribe(self, subscriber=None):
return self.wrapped_characteristic.subscribe(
None if subscriber is None else lambda value: subscriber(self.decode_value(value))
)
# -----------------------------------------------------------------------------
class DelegatedCharacteristicAdapter(CharacteristicAdapter):
def __init__(self, characteristic, encode, decode):
super().__init__(characteristic)
self.encode = encode
self.decode = decode
def encode_value(self, value):
return self.encode(value)
def decode_value(self, value):
return self.decode(value)
# -----------------------------------------------------------------------------
class PackedCharacteristicAdapter(CharacteristicAdapter):
'''
Adapter that packs/unpacks characteristic values according to a standard
Python `struct` format.
For formats with a single value, the adapted `read_value` and `write_value`
methods return/accept single values. For formats with multiple values,
they return/accept a tuple with the same number of elements as is required for
the format.
'''
def __init__(self, characteristic, format):
super().__init__(characteristic)
self.struct = struct.Struct(format)
def pack(self, *values):
return self.struct.pack(*values)
def unpack(self, buffer):
return self.struct.unpack(buffer)
def encode_value(self, value):
return self.pack(*value if type(value) is tuple else (value,))
def decode_value(self, value):
unpacked = self.unpack(value)
return unpacked[0] if len(unpacked) == 1 else unpacked
# -----------------------------------------------------------------------------
class MappedCharacteristicAdapter(PackedCharacteristicAdapter):
'''
Adapter that packs/unpacks characteristic values according to a standard
Python `struct` format.
The adapted `read_value` and `write_value` methods return/accept aa dictionary which
is packed/unpacked according to format, with the arguments extracted from the dictionary
by key, in the same order as they occur in the `keys` parameter.
'''
def __init__(self, characteristic, format, keys):
super().__init__(characteristic, format)
self.keys = keys
def pack(self, values):
return super().pack(*(values[key] for key in self.keys))
def unpack(self, buffer):
return dict(zip(self.keys, super().unpack(buffer)))
# -----------------------------------------------------------------------------
class UTF8CharacteristicAdapter(CharacteristicAdapter):
'''
Adapter that converts strings to/from bytes using UTF-8 encoding
'''
def encode_value(self, value):
return value.encode('utf-8')
def decode_value(self, value):
return value.decode('utf-8')
# -----------------------------------------------------------------------------
class Descriptor(Attribute):
'''
See Vol 3, Part G - 3.3.3 Characteristic Descriptor Declarations
'''
def __init__(self, uuid, permissions, value = b''):
# Convert the uuid to a UUID object if it isn't already
if type(uuid) is str:
uuid = UUID(uuid)
super().__init__(uuid, permissions, value)
self.uuid = uuid
self.characteristic = None
def __init__(self, descriptor_type, permissions, value = b''):
super().__init__(descriptor_type, permissions, value)
def __str__(self):
return f'Descriptor(handle=0x{self.handle:04X}, uuid={self.uuid}, value={self.read_value(None).hex()})'
return f'Descriptor(handle=0x{self.handle:04X}, type={self.type}, value={self.read_value(None).hex()})'

117
bumble/gatt_client.py
View File

@@ -35,10 +35,9 @@ from .gatt import (
GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
GATT_REQUEST_TIMEOUT,
GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE,
GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE,
GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
Service,
Characteristic,
Descriptor
Characteristic
)
# -----------------------------------------------------------------------------
@@ -47,6 +46,91 @@ from .gatt import (
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
# Proxies
# -----------------------------------------------------------------------------
class AttributeProxy(EventEmitter):
def __init__(self, client, handle, end_group_handle, attribute_type):
EventEmitter.__init__(self)
self.client = client
self.handle = handle
self.end_group_handle = end_group_handle
self.type = attribute_type
async def read_value(self, no_long_read=False):
return await self.client.read_value(self.handle, no_long_read)
async def write_value(self, value, with_response=False):
return await self.client.write_value(self.handle, value, with_response)
def __str__(self):
return f'Attribute(handle=0x{self.handle:04X}, type={self.uuid})'
class ServiceProxy(AttributeProxy):
@staticmethod
def from_client(cls, client, service_uuid):
# The service and its characteristics are considered to have already been discovered
services = client.get_services_by_uuid(service_uuid)
service = services[0] if services else None
return cls(service) if service else None
def __init__(self, client, handle, end_group_handle, uuid, primary=True):
attribute_type = GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE if primary else GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE
super().__init__(client, handle, end_group_handle, attribute_type)
self.uuid = uuid
self.characteristics = []
async def discover_characteristics(self, uuids=[]):
return await self.client.discover_characteristics(uuids, self)
def get_characteristics_by_uuid(self, uuid):
return self.client.get_characteristics_by_uuid(uuid, self)
def __str__(self):
return f'Service(handle=0x{self.handle:04X}, uuid={self.uuid})'
class CharacteristicProxy(AttributeProxy):
def __init__(self, client, handle, end_group_handle, uuid, properties):
super().__init__(client, handle, end_group_handle, uuid)
self.uuid = uuid
self.properties = properties
self.descriptors = []
self.descriptors_discovered = False
def get_descriptor(self, descriptor_type):
for descriptor in self.descriptors:
if descriptor.type == descriptor_type:
return descriptor
async def discover_descriptors(self):
return await self.client.discover_descriptors(self)
async def subscribe(self, subscriber=None):
return await self.client.subscribe(self, subscriber)
def __str__(self):
return f'Characteristic(handle=0x{self.handle:04X}, uuid={self.uuid}, properties={Characteristic.properties_as_string(self.properties)})'
class DescriptorProxy(AttributeProxy):
def __init__(self, client, handle, descriptor_type):
super().__init__(client, handle, 0, descriptor_type)
def __str__(self):
return f'Descriptor(handle=0x{self.handle:04X}, type={self.type})'
class ProfileServiceProxy:
'''
Base class for profile-specific service proxies
'''
@classmethod
def from_client(cls, client):
return ServiceProxy.from_client(cls, client, cls.SERVICE_CLASS.UUID)
# -----------------------------------------------------------------------------
# GATT Client
# -----------------------------------------------------------------------------
@@ -173,10 +257,14 @@ class Client:
logger.warning(f'bogus handle values: {attribute_handle} {end_group_handle}')
return
# Create a primary service object
service = Service(UUID.from_bytes(attribute_value), [], True)
service.handle = attribute_handle
service.end_group_handle = end_group_handle
# Create a service proxy for this service
service = ServiceProxy(
self,
attribute_handle,
end_group_handle,
UUID.from_bytes(attribute_value),
True
)
# Filter out returned services based on the given uuids list
if (not uuids) or (service.uuid in uuids):
@@ -233,10 +321,8 @@ class Client:
logger.warning(f'bogus handle values: {attribute_handle} {end_group_handle}')
return
# Create a primary service object
service = Service(uuid, [], True)
service.handle = attribute_handle
service.end_group_handle = end_group_handle
# Create a service proxy for this service
service = ServiceProxy(self, attribute_handle, end_group_handle, uuid, True)
# Add the service to the peer's service list
services.append(service)
@@ -314,8 +400,7 @@ class Client:
properties, handle = struct.unpack_from('<BH', attribute_value)
characteristic_uuid = UUID.from_bytes(attribute_value[3:])
characteristic = Characteristic(characteristic_uuid, properties, 0)
characteristic.handle = handle
characteristic = CharacteristicProxy(self, handle, 0, characteristic_uuid, properties)
# Set the previous characteristic's end handle
if characteristics:
@@ -382,8 +467,7 @@ class Client:
logger.warning(f'bogus handle value: {attribute_handle}')
return []
descriptor = Descriptor(UUID.from_bytes(attribute_uuid), 0)
descriptor.handle = attribute_handle
descriptor = DescriptorProxy(self, attribute_handle, UUID.from_bytes(attribute_uuid))
descriptors.append(descriptor)
# TODO: read descriptor value
@@ -427,8 +511,7 @@ class Client:
logger.warning(f'bogus handle value: {attribute_handle}')
return []
attribute = Attribute(attribute_uuid, 0)
attribute.handle = attribute_handle
attribute = AttributeProxy(self, attribute_handle, 0, UUID.from_bytes(attribute_uuid))
attributes.append(attribute)
# Move on to the next attributes

13
bumble/profiles/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

61
bumble/profiles/battery_service.py Normal file
View File

@@ -0,0 +1,61 @@
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
from ..gatt_client import ProfileServiceProxy
from ..gatt import (
GATT_BATTERY_SERVICE,
GATT_BATTERY_LEVEL_CHARACTERISTIC,
TemplateService,
Characteristic,
CharacteristicValue,
PackedCharacteristicAdapter
)
# -----------------------------------------------------------------------------
class BatteryService(TemplateService):
UUID = GATT_BATTERY_SERVICE
BATTERY_LEVEL_FORMAT = 'B'
def __init__(self, read_battery_level):
self.battery_level_characteristic = PackedCharacteristicAdapter(
Characteristic(
GATT_BATTERY_LEVEL_CHARACTERISTIC,
Characteristic.READ | Characteristic.NOTIFY,
Characteristic.READABLE,
CharacteristicValue(read=read_battery_level)
),
format=BatteryService.BATTERY_LEVEL_FORMAT
)
super().__init__([self.battery_level_characteristic])
# -----------------------------------------------------------------------------
class BatteryServiceProxy(ProfileServiceProxy):
SERVICE_CLASS = BatteryService
def __init__(self, service_proxy):
self.service_proxy = service_proxy
if characteristics := service_proxy.get_characteristics_by_uuid(GATT_BATTERY_LEVEL_CHARACTERISTIC):
self.battery_level = PackedCharacteristicAdapter(
characteristics[0],
format=BatteryService.BATTERY_LEVEL_FORMAT
)
else:
self.battery_level = None

135
bumble/profiles/device_information_service.py Normal file
View File

@@ -0,0 +1,135 @@
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import struct
from typing import Tuple
from ..gatt_client import ProfileServiceProxy
from ..gatt import (
GATT_DEVICE_INFORMATION_SERVICE,
GATT_FIRMWARE_REVISION_STRING_CHARACTERISTIC,
GATT_HARDWARE_REVISION_STRING_CHARACTERISTIC,
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
GATT_MODEL_NUMBER_STRING_CHARACTERISTIC,
GATT_SERIAL_NUMBER_STRING_CHARACTERISTIC,
GATT_SOFTWARE_REVISION_STRING_CHARACTERISTIC,
GATT_SYSTEM_ID_CHARACTERISTIC,
GATT_REGULATORY_CERTIFICATION_DATA_LIST_CHARACTERISTIC,
TemplateService,
Characteristic,
DelegatedCharacteristicAdapter,
UTF8CharacteristicAdapter
)
# -----------------------------------------------------------------------------
class DeviceInformationService(TemplateService):
UUID = GATT_DEVICE_INFORMATION_SERVICE
@staticmethod
def pack_system_id(oui, manufacturer_id):
return struct.pack('<Q', oui << 40 | manufacturer_id)
@staticmethod
def unpack_system_id(buffer):
system_id = struct.unpack('<Q', buffer)[0]
return (system_id >> 40, system_id & 0xFFFFFFFFFF)
def __init__(
self,
manufacturer_name: str = None,
model_number: str = None,
serial_number: str = None,
hardware_revision: str = None,
firmware_revision: str = None,
software_revision: str = None,
system_id: Tuple[int, int] = None, # (OUI, Manufacturer ID)
ieee_regulatory_certification_data_list: bytes = None
# TODO: pnp_id
):
characteristics = [
Characteristic(
uuid,
Characteristic.READ,
Characteristic.READABLE,
field
)
for (field, uuid) in (
(manufacturer_name, GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC),
(model_number, GATT_MODEL_NUMBER_STRING_CHARACTERISTIC),
(serial_number, GATT_SERIAL_NUMBER_STRING_CHARACTERISTIC),
(hardware_revision, GATT_HARDWARE_REVISION_STRING_CHARACTERISTIC),
(firmware_revision, GATT_FIRMWARE_REVISION_STRING_CHARACTERISTIC),
(software_revision, GATT_SOFTWARE_REVISION_STRING_CHARACTERISTIC)
)
if field is not None
]
if system_id is not None:
characteristics.append(Characteristic(
GATT_SYSTEM_ID_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
self.pack_system_id(*system_id)
))
if ieee_regulatory_certification_data_list is not None:
characteristics.append(Characteristic(
GATT_REGULATORY_CERTIFICATION_DATA_LIST_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
ieee_regulatory_certification_data_list
))
super().__init__(characteristics)
# -----------------------------------------------------------------------------
class DeviceInformationServiceProxy(ProfileServiceProxy):
SERVICE_CLASS = DeviceInformationService
def __init__(self, service_proxy):
self.service_proxy = service_proxy
for (field, uuid) in (
('manufacturer_name', GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC),
('model_number', GATT_MODEL_NUMBER_STRING_CHARACTERISTIC),
('serial_number', GATT_SERIAL_NUMBER_STRING_CHARACTERISTIC),
('hardware_revision', GATT_HARDWARE_REVISION_STRING_CHARACTERISTIC),
('firmware_revision', GATT_FIRMWARE_REVISION_STRING_CHARACTERISTIC),
('software_revision', GATT_SOFTWARE_REVISION_STRING_CHARACTERISTIC)
):
if characteristics := service_proxy.get_characteristics_by_uuid(uuid):
characteristic = UTF8CharacteristicAdapter(characteristics[0])
else:
characteristic = None
self.__setattr__(field, characteristic)
if characteristics := service_proxy.get_characteristics_by_uuid(GATT_SYSTEM_ID_CHARACTERISTIC):
self.system_id = DelegatedCharacteristicAdapter(
characteristics[0],
encode=lambda v: DeviceInformationService.pack_system_id(*v),
decode=DeviceInformationService.unpack_system_id
)
else:
self.system_id = None
if characteristics := service_proxy.get_characteristics_by_uuid(GATT_REGULATORY_CERTIFICATION_DATA_LIST_CHARACTERISTIC):
self.ieee_regulatory_certification_data_list = characteristics[0]
else:
self.ieee_regulatory_certification_data_list = None

72
examples/battery_client.py Normal file
View File

@@ -0,0 +1,72 @@
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import sys
import os
import logging
from colors import color
from bumble.device import Device, Peer
from bumble.transport import open_transport
from bumble.profiles.battery_service import BatteryServiceProxy
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 3:
print('Usage: battery_client.py <transport-spec> <bluetooth-address>')
print('example: battery_client.py usb:0 E1:CA:72:48:C4:E8')
return
print('<<< connecting to HCI...')
async with await open_transport(sys.argv[1]) as (hci_source, hci_sink):
print('<<< connected')
# Create and start a device
device = Device.with_hci('Bumble', 'F0:F1:F2:F3:F4:F5', hci_source, hci_sink)
await device.power_on()
# Connect to the peer
target_address = sys.argv[2]
print(f'=== Connecting to {target_address}...')
connection = await device.connect(target_address)
print(f'=== Connected to {connection}')
# Discover the Battery Service
peer = Peer(connection)
print('=== Discovering Battery Service')
battery_service = await peer.discover_and_create_service_proxy(BatteryServiceProxy)
# Check that the service was found
if not battery_service:
print('!!! Service not found')
return
# Subscribe to and read the battery level
if battery_service.battery_level:
await battery_service.battery_level.subscribe(
lambda value: print(f'{color("Battery Level Update:", "green")} {value}')
)
value = await battery_service.battery_level.read_value()
print(f'{color("Initial Battery Level:", "green")} {value}')
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

46
examples/battery_service.py → examples/battery_server.py
View File

@@ -25,59 +25,41 @@ import struct
from bumble.core import AdvertisingData
from bumble.device import Device
from bumble.transport import open_transport_or_link
from bumble.gatt import (
Service,
Characteristic,
CharacteristicValue,
GATT_DEVICE_BATTERY_SERVICE,
GATT_BATTERY_LEVEL_CHARACTERISTIC
)
# -----------------------------------------------------------------------------
def read_battery_level(connection):
return bytes([random.randint(0, 100)])
from bumble.profiles.battery_service import BatteryService
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 3:
print('Usage: python battery_service.py <device-config> <transport-spec>')
print('example: python battery_service.py device1.json usb:0')
print('Usage: python battery_server.py <device-config> <transport-spec>')
print('example: python battery_server.py device1.json usb:0')
return
async with await open_transport_or_link(sys.argv[2]) as (hci_source, hci_sink):
# Create a device to manage the host
device = Device.from_config_file_with_hci(sys.argv[1], hci_source, hci_sink)
# Add a Battery Service to the GATT sever
device.add_services([
Service(
GATT_DEVICE_BATTERY_SERVICE,
[
Characteristic(
GATT_BATTERY_LEVEL_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
CharacteristicValue(read=read_battery_level)
)
]
)
])
# Add a Device Information Service and Battery Service to the GATT sever
battery_service = BatteryService(lambda _: random.randint(0, 100))
device.add_service(battery_service)
# Set the advertising data
device.advertising_data = bytes(
AdvertisingData([
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes('Bumble Battery', 'utf-8')),
(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, struct.pack('<H', 0x180F)),
(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, bytes(battery_service.uuid)),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340))
])
)
# Go!
await device.power_on()
await device.start_advertising()
await hci_source.wait_for_termination()
await device.start_advertising(auto_restart=True)
# Notify every 3 seconds
while True:
await asyncio.sleep(3.0)
await device.notify_subscribers(battery_service.battery_level_characteristic)
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())

80
examples/device_information_client.py Normal file
View File

@@ -0,0 +1,80 @@
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import sys
import os
import logging
from colors import color
from bumble.device import Device, Peer
from bumble.profiles.device_information_service import DeviceInformationServiceProxy
from bumble.transport import open_transport
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 3:
print('Usage: device_information_client.py <transport-spec> <bluetooth-address>')
print('example: device_information_client.py usb:0 E1:CA:72:48:C4:E8')
return
print('<<< connecting to HCI...')
async with await open_transport(sys.argv[1]) as (hci_source, hci_sink):
print('<<< connected')
# Create and start a device
device = Device.with_hci('Bumble', 'F0:F1:F2:F3:F4:F5', hci_source, hci_sink)
await device.power_on()
# Connect to the peer
target_address = sys.argv[2]
print(f'=== Connecting to {target_address}...')
connection = await device.connect(target_address)
print(f'=== Connected to {connection}')
# Discover the Device Information service
peer = Peer(connection)
print('=== Discovering Device Information Service')
device_information_service = await peer.discover_service_and_create_proxy(DeviceInformationServiceProxy)
# Check that the service was found
if device_information_service is None:
print('!!! Service not found')
return
# Read and print the fields
if device_information_service.manufacturer_name is not None:
print(color('Manufacturer Name: ', 'green'), await device_information_service.manufacturer_name.read_value())
if device_information_service.model_number is not None:
print(color('Model Number: ', 'green'), await device_information_service.model_number.read_value())
if device_information_service.serial_number is not None:
print(color('Serial Number: ', 'green'), await device_information_service.serial_number.read_value())
if device_information_service.hardware_revision is not None:
print(color('Hardware Revision: ', 'green'), await device_information_service.hardware_revision.read_value())
if device_information_service.firmware_revision is not None:
print(color('Firmware Revision: ', 'green'), await device_information_service.firmware_revision.read_value())
if device_information_service.software_revision is not None:
print(color('Software Revision: ', 'green'), await device_information_service.software_revision.read_value())
if device_information_service.system_id is not None:
print(color('System ID: ', 'green'), await device_information_service.system_id.read_value())
if device_information_service.ieee_regulatory_certification_data_list is not None:
print(color('Regulatory Certification:', 'green'), (await device_information_service.ieee_regulatory_certification_data_list.read_value()).hex())
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

66
examples/device_information_server.py Normal file
View File

@@ -0,0 +1,66 @@
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import sys
import os
import logging
import struct
from bumble.core import AdvertisingData
from bumble.device import Device
from bumble.transport import open_transport_or_link
from bumble.profiles.device_information_service import DeviceInformationService
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 3:
print('Usage: python device_info_server.py <device-config> <transport-spec>')
print('example: python device_info_server.py device1.json usb:0')
return
async with await open_transport_or_link(sys.argv[2]) as (hci_source, hci_sink):
device = Device.from_config_file_with_hci(sys.argv[1], hci_source, hci_sink)
# Add a Device Information Service to the GATT sever
device_information_service = DeviceInformationService(
manufacturer_name = 'ACME',
model_number = 'AB-102',
serial_number = '7654321',
hardware_revision = '1.1.3',
software_revision = '2.5.6',
system_id = (0x123456, 0x8877665544)
)
device.add_service(device_information_service)
# Set the advertising data
device.advertising_data = bytes(
AdvertisingData([
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes('Bumble Device', 'utf-8')),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340))
])
)
# Go!
await device.power_on()
await device.start_advertising(auto_restart=True)
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

96
examples/get_peer_device_info.py
View File

@@ -1,96 +0,0 @@
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import sys
import os
import logging
from colors import color
from bumble.device import Device, Peer
from bumble.host import Host
from bumble.transport import open_transport
from bumble.utils import AsyncRunner
from bumble import gatt
# -----------------------------------------------------------------------------
class Listener(Device.Listener):
def __init__(self, device):
self.device = device
self.done = asyncio.get_running_loop().create_future()
@AsyncRunner.run_in_task()
async def on_connection(self, connection):
print(f'=== Connected to {connection}')
# Discover the Device Info service
peer = Peer(connection)
print('=== Discovering Device Info')
await peer.discover_services([gatt.GATT_DEVICE_INFORMATION_SERVICE])
# Check that the service was found
device_info_services = peer.get_services_by_uuid(gatt.GATT_DEVICE_INFORMATION_SERVICE)
if not device_info_services:
print('!!! Service not found')
return
# Get the characteristics we want from the (first) device info service
service = device_info_services[0]
await peer.discover_characteristics([
gatt.GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC
], service)
# Read the manufacturer name
manufacturer_name = peer.get_characteristics_by_uuid(gatt.GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC, service)
if manufacturer_name:
value = await peer.read_value(manufacturer_name[0])
print(color('Manufacturer Name:', 'green'), value.decode('utf-8'))
else:
print('>>> No manufacturer name')
self.done.set_result(None)
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 3:
print('Usage: get_peer_device_info.py <transport-spec> <bluetooth-address>')
print('example: get_peer_device_info.py usb:0 E1:CA:72:48:C4:E8')
return
print('<<< connecting to HCI...')
packet_source, packet_sink = await open_transport(sys.argv[1])
print('<<< connected')
# Create a host using the packet source and sink as controller
host = Host(controller_source=packet_source, controller_sink=packet_sink)
# Create a device to manage the host, with a custom listener
device = Device('Bumble', address = 'F0:F1:F2:F3:F4:F5', host = host)
device.listener = Listener(device)
await device.power_on()
# Connect to a peer
target_address = sys.argv[2]
print(f'=== Connecting to {target_address}...')
await device.connect(target_address)
await device.listener.done
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

3
examples/run_controller.py
View File

@@ -45,8 +45,7 @@ async def main():
# Create a first controller using the packet source/sink as its host interface
controller1 = Controller('C1', host_source = hci_source, host_sink = hci_sink, link = link)
print("====", sys.argv)
controller1.address = sys.argv[1]
controller1.random_address = sys.argv[1]
# Create a second controller using the same link
controller2 = Controller('C2', link = link)

4
examples/run_gatt_client.py
View File

@@ -41,10 +41,10 @@ class Listener(Device.Listener):
print('=== Discovering services')
peer = Peer(connection)
await peer.discover_services()
await peer.discover_characteristics()
for service in peer.services:
await service.discover_characteristics()
for characteristic in service.characteristics:
await peer.discover_descriptors(characteristic)
await characteristic.discover_descriptors()
print('=== Services discovered')
show_services(peer.services)

70
examples/run_gatt_client_and_server.py
View File

@@ -25,7 +25,6 @@ from bumble.controller import Controller
from bumble.device import Device, Peer
from bumble.host import Host
from bumble.link import LocalLink
from bumble.utils import AsyncRunner
from bumble.gatt import (
Service,
Characteristic,
@@ -37,43 +36,6 @@ from bumble.gatt import (
)
# -----------------------------------------------------------------------------
class ClientListener(Device.Listener):
def __init__(self, device):
self.device = device
@AsyncRunner.run_in_task()
async def on_connection(self, connection):
print(f'=== Client: connected to {connection}')
# Discover all services
print('=== Discovering services')
peer = Peer(connection)
await peer.discover_services()
await peer.discover_characteristics()
for service in peer.services:
for characteristic in service.characteristics:
await peer.discover_descriptors(characteristic)
print('=== Services discovered')
show_services(peer.services)
# Discover all attributes
print('=== Discovering attributes')
attributes = await peer.discover_attributes()
for attribute in attributes:
print(attribute)
print('=== Attributes discovered')
# Read all attributes
for attribute in attributes:
try:
value = await peer.read_value(attribute)
print(color(f'0x{attribute.handle:04X} = {value.hex()}', 'green'))
except ProtocolError as error:
print(color(f'cannot read {attribute.handle:04X}:', 'red'), error)
# -----------------------------------------------------------------------------
class ServerListener(Device.Listener):
def on_connection(self, connection):
@@ -90,7 +52,6 @@ async def main():
client_host = Host()
client_host.controller = client_controller
client_device = Device("client", address = 'F0:F1:F2:F3:F4:F5', host = client_host)
client_device.listener = ClientListener(client_device)
await client_device.power_on()
# Setup a stack for the server
@@ -116,7 +77,36 @@ async def main():
server_device.add_service(device_info_service)
# Connect the client to the server
await client_device.connect(server_device.address)
connection = await client_device.connect(server_device.random_address)
print(f'=== Client: connected to {connection}')
# Discover all services
print('=== Discovering services')
peer = Peer(connection)
await peer.discover_services()
for service in peer.services:
await service.discover_characteristics()
for characteristic in service.characteristics:
await characteristic.discover_descriptors()
print('=== Services discovered')
show_services(peer.services)
# Discover all attributes
print('=== Discovering attributes')
attributes = await peer.discover_attributes()
for attribute in attributes:
print(attribute)
print('=== Attributes discovered')
# Read all attributes
for attribute in attributes:
try:
value = await attribute.read_value()
print(color(f'0x{attribute.handle:04X} = {value.hex()}', 'green'))
except ProtocolError as error:
print(color(f'cannot read {attribute.handle:04X}:', 'red'), error)
await asyncio.get_running_loop().create_future()
# -----------------------------------------------------------------------------

150
tests/gatt_test.py
View File

@@ -18,6 +18,7 @@
import asyncio
import logging
import os
import struct
import pytest
from bumble.controller import Controller
@@ -25,6 +26,12 @@ from bumble.link import LocalLink
from bumble.device import Device, Peer
from bumble.host import Host
from bumble.gatt import (
GATT_BATTERY_LEVEL_CHARACTERISTIC,
CharacteristicAdapter,
DelegatedCharacteristicAdapter,
PackedCharacteristicAdapter,
MappedCharacteristicAdapter,
UTF8CharacteristicAdapter,
Service,
Characteristic,
CharacteristicValue
@@ -91,6 +98,96 @@ def test_ATT_Read_By_Group_Type_Request():
basic_check(pdu)
# -----------------------------------------------------------------------------
def test_CharacteristicAdapter():
# Check that the CharacteristicAdapter base class is transparent
v = bytes([1, 2, 3])
c = Characteristic(GATT_BATTERY_LEVEL_CHARACTERISTIC, Characteristic.READ, Characteristic.READABLE, v)
a = CharacteristicAdapter(c)
value = a.read_value(None)
assert(value == v)
v = bytes([3, 4, 5])
a.write_value(None, v)
assert(c.value == v)
# Simple delegated adapter
a = DelegatedCharacteristicAdapter(c, lambda x: bytes(reversed(x)), lambda x: bytes(reversed(x)))
value = a.read_value(None)
assert(value == bytes(reversed(v)))
v = bytes([3, 4, 5])
a.write_value(None, v)
assert(a.value == bytes(reversed(v)))
# Packed adapter with single element format
v = 1234
pv = struct.pack('>H', v)
c.value = v
a = PackedCharacteristicAdapter(c, '>H')
value = a.read_value(None)
assert(value == pv)
c.value = None
a.write_value(None, pv)
assert(a.value == v)
# Packed adapter with multi-element format
v1 = 1234
v2 = 5678
pv = struct.pack('>HH', v1, v2)
c.value = (v1, v2)
a = PackedCharacteristicAdapter(c, '>HH')
value = a.read_value(None)
assert(value == pv)
c.value = None
a.write_value(None, pv)
assert(a.value == (v1, v2))
# Mapped adapter
v1 = 1234
v2 = 5678
pv = struct.pack('>HH', v1, v2)
mapped = {'v1': v1, 'v2': v2}
c.value = mapped
a = MappedCharacteristicAdapter(c, '>HH', ('v1', 'v2'))
value = a.read_value(None)
assert(value == pv)
c.value = None
a.write_value(None, pv)
assert(a.value == mapped)
# UTF-8 adapter
v = 'Hello π'
ev = v.encode('utf-8')
c.value = v
a = UTF8CharacteristicAdapter(c)
value = a.read_value(None)
assert(value == ev)
c.value = None
a.write_value(None, ev)
assert(a.value == v)
# -----------------------------------------------------------------------------
def test_CharacteristicValue():
b = bytes([1, 2, 3])
c = CharacteristicValue(read=lambda _: b)
x = c.read(None)
assert(x == b)
result = []
c = CharacteristicValue(write=lambda connection, value: result.append((connection, value)))
z = object()
c.write(z, b)
assert(result == [(z, b)])
# -----------------------------------------------------------------------------
class TwoDevices:
def __init__(self):
@@ -199,6 +296,56 @@ async def test_read_write():
assert(characteristic2._last_value[1] == b)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_read_write2():
[client, server] = TwoDevices().devices
v = bytes([0x11, 0x22, 0x33, 0x44])
characteristic1 = Characteristic(
'FDB159DB-036C-49E3-B3DB-6325AC750806',
Characteristic.READ | Characteristic.WRITE,
Characteristic.READABLE | Characteristic.WRITEABLE,
value=v
)
service1 = Service(
'3A657F47-D34F-46B3-B1EC-698E29B6B829',
[
characteristic1
]
)
server.add_services([service1])
await client.power_on()
await server.power_on()
connection = await client.connect(server.random_address)
peer = Peer(connection)
await peer.discover_services()
c = peer.get_services_by_uuid(service1.uuid)
assert(len(c) == 1)
s = c[0]
await s.discover_characteristics()
c = s.get_characteristics_by_uuid(characteristic1.uuid)
assert(len(c) == 1)
c1 = c[0]
v1 = await c1.read_value()
assert(v1 == v)
a1 = PackedCharacteristicAdapter(c1, '>I')
v1 = await a1.read_value()
assert(v1 == struct.unpack('>I', v)[0])
b = bytes([0x55, 0x66, 0x77, 0x88])
await a1.write_value(struct.unpack('>I', b)[0])
await async_barrier()
assert(characteristic1.value == b)
v1 = await a1.read_value()
assert(v1 == struct.unpack('>I', b)[0])
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_subscribe_notify():
@@ -330,6 +477,7 @@ async def test_subscribe_notify():
# -----------------------------------------------------------------------------
async def async_main():
await test_read_write()
await test_read_write2()
await test_subscribe_notify()
# -----------------------------------------------------------------------------
@@ -338,4 +486,6 @@ if __name__ == '__main__':
test_UUID()
test_ATT_Error_Response()
test_ATT_Read_By_Group_Type_Request()
test_CharacteristicValue()
test_CharacteristicAdapter()
asyncio.run(async_main())