auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-05-08 03:58:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #624 from zxzxwu/gatt

Browse Source 
Support GATT Service
This commit is contained in:
zxzxwu
2025-01-28 20:02:43 +08:00
committed by GitHub
parent a1bade6f20 1eb9d8d055
commit 079cf6b896
6 changed files with 398 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
bumble/device.py
View File

@@ -93,6 +93,7 @@ from bumble import smp
from bumble import sdp
from bumble import l2cap
from bumble import core
from bumble.profiles import gatt_service
if TYPE_CHECKING:
from .transport.common import TransportSource, TransportSink
@@ -1747,6 +1748,8 @@ class DeviceConfiguration:
cis_enabled: bool = False
identity_address_type: Optional[int] = None
io_capability: int = pairing.PairingDelegate.IoCapability.NO_OUTPUT_NO_INPUT
gap_service_enabled: bool = True
gatt_service_enabled: bool = True
def __post_init__(self) -> None:
self.gatt_services: list[Dict[str, Any]] = []
@@ -1929,6 +1932,7 @@ class Device(CompositeEventEmitter):
bis_links = dict[int, BisLink]()
big_syncs = dict[int, BigSync]()
_pending_cis: Dict[int, tuple[int, int]]
gatt_service: gatt_service.GenericAttributeProfileService | None = None
@composite_listener
class Listener:
@@ -1995,7 +1999,6 @@ class Device(CompositeEventEmitter):
address: Optional[hci.Address] = None,
config: Optional[DeviceConfiguration] = None,
host: Optional[Host] = None,
generic_access_service: bool = True,
) -> None:
super().__init__()
@@ -2142,7 +2145,10 @@ class Device(CompositeEventEmitter):
# Register the SDP server with the L2CAP Channel Manager
self.sdp_server.register(self.l2cap_channel_manager)
self.add_default_services(generic_access_service)
self.add_default_services(
add_gap_service=config.gap_service_enabled,
add_gatt_service=config.gatt_service_enabled,
)
self.l2cap_channel_manager.register_fixed_channel(ATT_CID, self.on_gatt_pdu)
# Forward some events
@@ -4506,10 +4512,15 @@ class Device(CompositeEventEmitter):
def add_services(self, services):
self.gatt_server.add_services(services)
def add_default_services(self, generic_access_service=True):
def add_default_services(
self, add_gap_service: bool = True, add_gatt_service: bool = True
) -> None:
# Add a GAP Service if requested
if generic_access_service:
if add_gap_service:
self.gatt_server.add_service(GenericAccessService(self.name))
if add_gatt_service:
self.gatt_service = gatt_service.GenericAttributeProfileService()
self.gatt_server.add_service(self.gatt_service)
async def notify_subscriber(self, connection, attribute, value=None, force=False):
await self.gatt_server.notify_subscriber(connection, attribute, value, force)

21
bumble/gatt.py
View File

@@ -48,7 +48,6 @@ from bumble.utils import ByteSerializable
if TYPE_CHECKING:
from bumble.gatt_client import AttributeProxy
from bumble.device import Connection
# -----------------------------------------------------------------------------
@@ -802,3 +801,23 @@ class ClientCharacteristicConfigurationBits(enum.IntFlag):
DEFAULT = 0x0000
NOTIFICATION = 0x0001
INDICATION = 0x0002
# -----------------------------------------------------------------------------
class ClientSupportedFeatures(enum.IntFlag):
'''
See Vol 3, Part G - 7.2 - Table 7.6: Client Supported Features bit assignments.
'''
ROBUST_CACHING = 0x01
ENHANCED_ATT_BEARER = 0x02
MULTIPLE_HANDLE_VALUE_NOTIFICATIONS = 0x04
# -----------------------------------------------------------------------------
class ServerSupportedFeatures(enum.IntFlag):
'''
See Vol 3, Part G - 7.4 - Table 7.11: Server Supported Features bit assignments.
'''
EATT_SUPPORTED = 0x01

166
bumble/profiles/gatt_service.py Normal file
View File

@@ -0,0 +1,166 @@
# Copyright 2021-2025 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.
from __future__ import annotations
import struct
from typing import TYPE_CHECKING
from bumble import att
from bumble import gatt
from bumble import gatt_client
from bumble import crypto
if TYPE_CHECKING:
from bumble import device
# -----------------------------------------------------------------------------
class GenericAttributeProfileService(gatt.TemplateService):
'''See Vol 3, Part G - 7 - DEFINED GENERIC ATTRIBUTE PROFILE SERVICE.'''
UUID = gatt.GATT_GENERIC_ATTRIBUTE_SERVICE
client_supported_features_characteristic: gatt.Characteristic | None = None
server_supported_features_characteristic: gatt.Characteristic | None = None
database_hash_characteristic: gatt.Characteristic | None = None
service_changed_characteristic: gatt.Characteristic | None = None
def __init__(
self,
server_supported_features: gatt.ServerSupportedFeatures | None = None,
database_hash_enabled: bool = True,
service_change_enabled: bool = True,
) -> None:
if server_supported_features is not None:
self.server_supported_features_characteristic = gatt.Characteristic(
uuid=gatt.GATT_SERVER_SUPPORTED_FEATURES_CHARACTERISTIC,
properties=gatt.Characteristic.Properties.READ,
permissions=gatt.Characteristic.Permissions.READABLE,
value=bytes([server_supported_features]),
)
if database_hash_enabled:
self.database_hash_characteristic = gatt.Characteristic(
uuid=gatt.GATT_DATABASE_HASH_CHARACTERISTIC,
properties=gatt.Characteristic.Properties.READ,
permissions=gatt.Characteristic.Permissions.READABLE,
value=gatt.CharacteristicValue(read=self.get_database_hash),
)
if service_change_enabled:
self.service_changed_characteristic = gatt.Characteristic(
uuid=gatt.GATT_SERVICE_CHANGED_CHARACTERISTIC,
properties=gatt.Characteristic.Properties.INDICATE,
permissions=gatt.Characteristic.Permissions(0),
value=b'',
)
if (database_hash_enabled and service_change_enabled) or (
server_supported_features
and (
server_supported_features & gatt.ServerSupportedFeatures.EATT_SUPPORTED
)
): # TODO: Support Multiple Handle Value Notifications
self.client_supported_features_characteristic = gatt.Characteristic(
uuid=gatt.GATT_CLIENT_SUPPORTED_FEATURES_CHARACTERISTIC,
properties=(
gatt.Characteristic.Properties.READ
| gatt.Characteristic.Properties.WRITE
),
permissions=(
gatt.Characteristic.Permissions.READABLE
| gatt.Characteristic.Permissions.WRITEABLE
),
value=bytes(1),
)
super().__init__(
characteristics=[
c
for c in (
self.service_changed_characteristic,
self.client_supported_features_characteristic,
self.database_hash_characteristic,
self.server_supported_features_characteristic,
)
if c is not None
],
primary=True,
)
@classmethod
def get_attribute_data(cls, attribute: att.Attribute) -> bytes:
if attribute.type in (
gatt.GATT_PRIMARY_SERVICE_ATTRIBUTE_TYPE,
gatt.GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE,
gatt.GATT_INCLUDE_ATTRIBUTE_TYPE,
gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
gatt.GATT_CHARACTERISTIC_EXTENDED_PROPERTIES_DESCRIPTOR,
):
return (
struct.pack("<H", attribute.handle)
+ attribute.type.to_bytes()
+ attribute.value
)
elif attribute.type in (
gatt.GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
gatt.GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
gatt.GATT_SERVER_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
gatt.GATT_CHARACTERISTIC_PRESENTATION_FORMAT_DESCRIPTOR,
gatt.GATT_CHARACTERISTIC_AGGREGATE_FORMAT_DESCRIPTOR,
):
return struct.pack("<H", attribute.handle) + attribute.type.to_bytes()
return b''
def get_database_hash(self, connection: device.Connection | None) -> bytes:
assert connection
m = b''.join(
[
self.get_attribute_data(attribute)
for attribute in connection.device.gatt_server.attributes
]
)
return crypto.aes_cmac(m=m, k=bytes(16))
class GenericAttributeProfileServiceProxy(gatt_client.ProfileServiceProxy):
SERVICE_CLASS = GenericAttributeProfileService
client_supported_features_characteristic: gatt_client.CharacteristicProxy | None = (
None
)
server_supported_features_characteristic: gatt_client.CharacteristicProxy | None = (
None
)
database_hash_characteristic: gatt_client.CharacteristicProxy | None = None
service_changed_characteristic: gatt_client.CharacteristicProxy | None = None
_CHARACTERISTICS = {
gatt.GATT_CLIENT_SUPPORTED_FEATURES_CHARACTERISTIC: 'client_supported_features_characteristic',
gatt.GATT_SERVER_SUPPORTED_FEATURES_CHARACTERISTIC: 'server_supported_features_characteristic',
gatt.GATT_DATABASE_HASH_CHARACTERISTIC: 'database_hash_characteristic',
gatt.GATT_SERVICE_CHANGED_CHARACTERISTIC: 'service_changed_characteristic',
}
def __init__(self, service_proxy: gatt_client.ServiceProxy) -> None:
self.service_proxy = service_proxy
for uuid, attribute_name in self._CHARACTERISTICS.items():
if characteristics := self.service_proxy.get_characteristics_by_uuid(uuid):
setattr(self, attribute_name, characteristics[0])