reset parser before raising exception

Enumify: PhysicalTransport, Role, AddressType

apps/auracast.py

@@ -27,7 +27,6 @@ import logging
 import os
 import struct
 from typing import (
-    cast,
     Any,
     AsyncGenerator,
     Coroutine,
@@ -100,6 +99,25 @@ def codec_config_string(
     return '\n'.join(indent + line for line in lines)
 
 
+def broadcast_code_bytes(broadcast_code: str) -> bytes:
+    """
+    Convert a broadcast code string to a 16-byte value.
+
+    If `broadcast_code` is `0x` followed by 32 hex characters, it is interpreted as a
+    raw 16-byte raw broadcast code in big-endian byte order.
+    Otherwise, `broadcast_code` is converted to a 16-byte value as specified in
+    BLUETOOTH CORE SPECIFICATION Version 6.0 | Vol 3, Part C , section 3.2.6.3
+    """
+    if broadcast_code.startswith("0x") and len(broadcast_code) == 34:
+        return bytes.fromhex(broadcast_code[2:])[::-1]
+
+    broadcast_code_utf8 = broadcast_code.encode("utf-8")
+    if len(broadcast_code_utf8) > 16:
+        raise ValueError("broadcast code must be <= 16 bytes in utf-8 encoding")
+    padding = bytes(16 - len(broadcast_code_utf8))
+    return broadcast_code_utf8 + padding
+
+
 # -----------------------------------------------------------------------------
 # Scan For Broadcasts
 # -----------------------------------------------------------------------------
@@ -127,11 +145,9 @@ class BroadcastScanner(pyee.EventEmitter):
         def update(self, advertisement: bumble.device.Advertisement) -> None:
             self.rssi = advertisement.rssi
             for service_data in advertisement.data.get_all(
-                core.AdvertisingData.SERVICE_DATA
+                core.AdvertisingData.Type.SERVICE_DATA_16_BIT_UUID
             ):
-                assert isinstance(service_data, tuple)
                 service_uuid, data = service_data
-                assert isinstance(data, bytes)
 
                 if service_uuid == gatt.GATT_PUBLIC_BROADCAST_ANNOUNCEMENT_SERVICE:
                     self.public_broadcast_announcement = (
@@ -145,16 +161,14 @@ class BroadcastScanner(pyee.EventEmitter):
                     )
                     continue
 
-            self.appearance = advertisement.data.get(  # type: ignore[assignment]
+            self.appearance = advertisement.data.get(
-                core.AdvertisingData.APPEARANCE
+                core.AdvertisingData.Type.APPEARANCE
             )
 
             if manufacturer_data := advertisement.data.get(
-                core.AdvertisingData.MANUFACTURER_SPECIFIC_DATA
+                core.AdvertisingData.Type.MANUFACTURER_SPECIFIC_DATA
             ):
-                assert isinstance(manufacturer_data, tuple)
+                company_id, data = manufacturer_data
-                company_id = cast(int, manufacturer_data[0])
-                data = cast(bytes, manufacturer_data[1])
                 self.manufacturer_data = (
                     company_ids.COMPANY_IDENTIFIERS.get(
                         company_id, f'0x{company_id:04X}'
@@ -239,22 +253,14 @@ class BroadcastScanner(pyee.EventEmitter):
 
             if self.biginfo:
                 print(color('  BIG:', 'cyan'))
-                print(
+                print(color('    Number of BIS:', 'magenta'), self.biginfo.num_bis)
-                    color('    Number of BIS:', 'magenta'),
+                print(color('    ISO Interval: ', 'magenta'), self.biginfo.iso_interval)
-                    self.biginfo.num_bis,
+                print(color('    Max PDU:      ', 'magenta'), self.biginfo.max_pdu)
-                )
+                print(color('    SDU Interval: ', 'magenta'), self.biginfo.sdu_interval)
-                print(
+                print(color('    Max SDU:      ', 'magenta'), self.biginfo.max_sdu)
-                    color('    PHY:          ', 'magenta'),
+                print(color('    PHY:          ', 'magenta'), self.biginfo.phy.name)
-                    self.biginfo.phy.name,
+                print(color('    Framed:       ', 'magenta'), self.biginfo.framed)
-                )
+                print(color('    Encrypted:    ', 'magenta'), self.biginfo.encrypted)
-                print(
-                    color('    Framed:       ', 'magenta'),
-                    self.biginfo.framed,
-                )
-                print(
-                    color('    Encrypted:    ', 'magenta'),
-                    self.biginfo.encrypted,
-                )
 
         def on_sync_establishment(self) -> None:
             self.emit('sync_establishment')
@@ -271,11 +277,9 @@ class BroadcastScanner(pyee.EventEmitter):
                 return
 
             for service_data in advertisement.data.get_all(
-                core.AdvertisingData.SERVICE_DATA
+                core.AdvertisingData.Type.SERVICE_DATA_16_BIT_UUID
             ):
-                assert isinstance(service_data, tuple)
                 service_uuid, data = service_data
-                assert isinstance(data, bytes)
 
                 if service_uuid == gatt.GATT_BASIC_AUDIO_ANNOUNCEMENT_SERVICE:
                     self.basic_audio_announcement = (
@@ -316,24 +320,23 @@ class BroadcastScanner(pyee.EventEmitter):
     def on_advertisement(self, advertisement: bumble.device.Advertisement) -> None:
         if not (
             ads := advertisement.data.get_all(
-                core.AdvertisingData.SERVICE_DATA_16_BIT_UUID
+                core.AdvertisingData.Type.SERVICE_DATA_16_BIT_UUID
             )
         ) or not (
             broadcast_audio_announcement := next(
                 (
                     ad
                     for ad in ads
-                    if isinstance(ad, tuple)
+                    if ad[0] == gatt.GATT_BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE
-                    and ad[0] == gatt.GATT_BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE
                 ),
                 None,
             )
         ):
             return
 
-        broadcast_name = advertisement.data.get(core.AdvertisingData.BROADCAST_NAME)
+        broadcast_name = advertisement.data.get_all(
-        assert isinstance(broadcast_name, str) or broadcast_name is None
+            core.AdvertisingData.Type.BROADCAST_NAME
-        assert isinstance(broadcast_audio_announcement[1], bytes)
+        )
 
         if broadcast := self.broadcasts.get(advertisement.address):
             broadcast.update(advertisement)
@@ -341,7 +344,7 @@ class BroadcastScanner(pyee.EventEmitter):
 
         bumble.utils.AsyncRunner.spawn(
             self.on_new_broadcast(
-                broadcast_name,
+                broadcast_name[0] if broadcast_name else None,
                 advertisement,
                 bap.BroadcastAudioAnnouncement.from_bytes(
                     broadcast_audio_announcement[1]
@@ -522,14 +525,19 @@ async def run_assist(
             return
 
         # Subscribe to and read the broadcast receive state characteristics
+        def on_broadcast_receive_state_update(
+            value: bass.BroadcastReceiveState, index: int
+        ) -> None:
+            print(
+                f"{color(f'Broadcast Receive State Update [{index}]:', 'green')} {value}"
+            )
+
         for i, broadcast_receive_state in enumerate(
             bass_client.broadcast_receive_states
         ):
             try:
                 await broadcast_receive_state.subscribe(
-                    lambda value, i=i: print(
+                    functools.partial(on_broadcast_receive_state_update, index=i)
-                        f"{color(f'Broadcast Receive State Update [{i}]:', 'green')} {value}"
-                    )
                 )
             except core.ProtocolError as error:
                 print(
@@ -705,14 +713,13 @@ async def run_receive(
 
         def on_change() -> None:
             if (
-                broadcast.basic_audio_announcement
+                broadcast.basic_audio_announcement and broadcast.biginfo
-                and not basic_audio_announcement_scanned.is_set()
+            ) and not basic_audio_announcement_scanned.is_set():
-            ):
                 basic_audio_announcement_scanned.set()
 
         broadcast.on('change', on_change)
-        if not broadcast.basic_audio_announcement:
+        if not broadcast.basic_audio_announcement or not broadcast.biginfo:
-            print('Wait for Basic Audio Announcement...')
+            print('Wait for Basic Audio Announcement and BIG Info...')
             await basic_audio_announcement_scanned.wait()
         print('Basic Audio Announcement found')
         broadcast.print()
@@ -733,7 +740,7 @@ async def run_receive(
                 big_sync_timeout=0x4000,
                 bis=[bis.index for bis in subgroup.bis],
                 broadcast_code=(
-                    bytes.fromhex(broadcast_code) if broadcast_code else None
+                    broadcast_code_bytes(broadcast_code) if broadcast_code else None
                 ),
             ),
         )
@@ -790,16 +797,8 @@ async def run_receive(
             for i, bis_link in enumerate(big_sync.bis_links):
                 print(f'Setup ISO for BIS {bis_link.handle}')
                 bis_link.sink = functools.partial(sink, lc3_queues[i])
-                await device.send_command(
+                await bis_link.setup_data_path(
-                    hci.HCI_LE_Setup_ISO_Data_Path_Command(
+                    direction=bis_link.Direction.CONTROLLER_TO_HOST
-                        connection_handle=bis_link.handle,
-                        data_path_direction=hci.HCI_LE_Setup_ISO_Data_Path_Command.Direction.CONTROLLER_TO_HOST,
-                        data_path_id=0,
-                        codec_id=hci.CodingFormat(codec_id=hci.CodecID.TRANSPARENT),
-                        controller_delay=0,
-                        codec_configuration=b'',
-                    ),
-                    check_result=True,
                 )
 
             terminated = asyncio.Event()
@@ -955,14 +954,19 @@ async def run_transmit(
                     max_transport_latency=65,
                     rtn=4,
                     broadcast_code=(
-                        bytes.fromhex(broadcast_code) if broadcast_code else None
+                        broadcast_code_bytes(broadcast_code) if broadcast_code else None
                     ),
                 ),
             )
+            for bis_link in big.bis_links:
+                print(f'Setup ISO for BIS {bis_link.handle}')
+                await bis_link.setup_data_path(
+                    direction=bis_link.Direction.HOST_TO_CONTROLLER
+                )
 
             iso_queues = [
-                bumble.device.IsoPacketStream(big.bis_links[0], 64),
+                bumble.device.IsoPacketStream(bis_link, 64)
-                bumble.device.IsoPacketStream(big.bis_links[1], 64),
+                for bis_link in big.bis_links
             ]
 
             def on_flow():
@@ -1094,7 +1098,7 @@ def pair(ctx, transport, address):
     '--broadcast-code',
     metavar='BROADCAST_CODE',
     type=str,
-    help='Broadcast encryption code in hex format',
+    help='Broadcast encryption code (string or raw hex format prefixed with 0x)',
 )
 @click.option(
     '--sync-timeout',

apps/gg_bridge.py

@@ -234,7 +234,7 @@ class GattlinkNodeBridge(GattlinkL2capEndpoint, Device.Listener):
             Characteristic.WRITEABLE,
             CharacteristicValue(write=self.on_rx_write),
         )
-        self.tx_characteristic = Characteristic(
+        self.tx_characteristic: Characteristic[bytes] = Characteristic(
             GG_GATTLINK_TX_CHARACTERISTIC_UUID,
             Characteristic.Properties.NOTIFY,
             Characteristic.READABLE,

bumble/att.py

@@ -29,13 +29,14 @@ import functools
 import inspect
 import struct
 from typing import (
-    Any,
     Awaitable,
     Callable,
+    Generic,
     Dict,
     List,
     Optional,
     Type,
+    TypeVar,
     Union,
     TYPE_CHECKING,
 )
@@ -43,13 +44,18 @@ from typing import (
 from pyee import EventEmitter
 
 from bumble import utils
-from bumble.core import UUID, name_or_number, ProtocolError
+from bumble.core import UUID, name_or_number, InvalidOperationError, ProtocolError
 from bumble.hci import HCI_Object, key_with_value
 from bumble.colors import color
 
+# -----------------------------------------------------------------------------
+# Typing
+# -----------------------------------------------------------------------------
 if TYPE_CHECKING:
     from bumble.device import Connection
 
+_T = TypeVar('_T')
+
 # -----------------------------------------------------------------------------
 # Constants
 # -----------------------------------------------------------------------------
@@ -748,7 +754,7 @@ class ATT_Handle_Value_Confirmation(ATT_PDU):
 
 
 # -----------------------------------------------------------------------------
-class AttributeValue:
+class AttributeValue(Generic[_T]):
     '''
     Attribute value where reading and/or writing is delegated to functions
     passed as arguments to the constructor.
@@ -757,33 +763,34 @@ class AttributeValue:
     def __init__(
         self,
         read: Union[
-            Callable[[Optional[Connection]], Any],
+            Callable[[Optional[Connection]], _T],
-            Callable[[Optional[Connection]], Awaitable[Any]],
+            Callable[[Optional[Connection]], Awaitable[_T]],
             None,
         ] = None,
         write: Union[
-            Callable[[Optional[Connection], Any], None],
+            Callable[[Optional[Connection], _T], None],
-            Callable[[Optional[Connection], Any], Awaitable[None]],
+            Callable[[Optional[Connection], _T], Awaitable[None]],
             None,
         ] = None,
     ):
         self._read = read
         self._write = write
 
-    def read(self, connection: Optional[Connection]) -> Union[bytes, Awaitable[bytes]]:
+    def read(self, connection: Optional[Connection]) -> Union[_T, Awaitable[_T]]:
-        return self._read(connection) if self._read else b''
+        if self._read is None:
+            raise InvalidOperationError('AttributeValue has no read function')
+        return self._read(connection)
 
     def write(
-        self, connection: Optional[Connection], value: bytes
+        self, connection: Optional[Connection], value: _T
     ) -> Union[Awaitable[None], None]:
-        if self._write:
+        if self._write is None:
-            return self._write(connection, value)
+            raise InvalidOperationError('AttributeValue has no write function')
-
+        return self._write(connection, value)
-        return None
 
 
 # -----------------------------------------------------------------------------
-class Attribute(EventEmitter):
+class Attribute(EventEmitter, Generic[_T]):
     class Permissions(enum.IntFlag):
         READABLE = 0x01
         WRITEABLE = 0x02
@@ -822,13 +829,13 @@ class Attribute(EventEmitter):
     READ_REQUIRES_AUTHORIZATION = Permissions.READ_REQUIRES_AUTHORIZATION
     WRITE_REQUIRES_AUTHORIZATION = Permissions.WRITE_REQUIRES_AUTHORIZATION
 
-    value: Any
+    value: Union[AttributeValue[_T], _T, None]
 
     def __init__(
         self,
         attribute_type: Union[str, bytes, UUID],
         permissions: Union[str, Attribute.Permissions],
-        value: Any = b'',
+        value: Union[AttributeValue[_T], _T, None] = None,
     ) -> None:
         EventEmitter.__init__(self)
         self.handle = 0
@@ -848,11 +855,11 @@ class Attribute(EventEmitter):
 
         self.value = value
 
-    def encode_value(self, value: Any) -> bytes:
+    def encode_value(self, value: _T) -> bytes:
-        return value
+        return value  # type: ignore
 
-    def decode_value(self, value_bytes: bytes) -> Any:
+    def decode_value(self, value: bytes) -> _T:
-        return value_bytes
+        return value  # type: ignore
 
     async def read_value(self, connection: Optional[Connection]) -> bytes:
         if (
@@ -877,11 +884,14 @@ class Attribute(EventEmitter):
                 error_code=ATT_INSUFFICIENT_AUTHORIZATION_ERROR, att_handle=self.handle
             )
 
-        if hasattr(self.value, 'read'):
+        value: Union[_T, None]
+        if isinstance(self.value, AttributeValue):
             try:
-                value = self.value.read(connection)
+                read_value = self.value.read(connection)
-                if inspect.isawaitable(value):
+                if inspect.isawaitable(read_value):
-                    value = await value
+                    value = await read_value
+                else:
+                    value = read_value
             except ATT_Error as error:
                 raise ATT_Error(
                     error_code=error.error_code, att_handle=self.handle
@@ -889,20 +899,24 @@ class Attribute(EventEmitter):
         else:
             value = self.value
 
-        self.emit('read', connection, value)
+        self.emit('read', connection, b'' if value is None else value)
 
-        return self.encode_value(value)
+        return b'' if value is None else self.encode_value(value)
 
-    async def write_value(self, connection: Connection, value_bytes: bytes) -> None:
+    async def write_value(self, connection: Optional[Connection], value: bytes) -> None:
         if (
-            self.permissions & self.WRITE_REQUIRES_ENCRYPTION
+            (self.permissions & self.WRITE_REQUIRES_ENCRYPTION)
-        ) and not connection.encryption:
+            and connection is not None
+            and not connection.encryption
+        ):
             raise ATT_Error(
                 error_code=ATT_INSUFFICIENT_ENCRYPTION_ERROR, att_handle=self.handle
             )
         if (
-            self.permissions & self.WRITE_REQUIRES_AUTHENTICATION
+            (self.permissions & self.WRITE_REQUIRES_AUTHENTICATION)
-        ) and not connection.authenticated:
+            and connection is not None
+            and not connection.authenticated
+        ):
             raise ATT_Error(
                 error_code=ATT_INSUFFICIENT_AUTHENTICATION_ERROR, att_handle=self.handle
             )
@@ -912,11 +926,11 @@ class Attribute(EventEmitter):
                 error_code=ATT_INSUFFICIENT_AUTHORIZATION_ERROR, att_handle=self.handle
             )
 
-        value = self.decode_value(value_bytes)
+        decoded_value = self.decode_value(value)
 
-        if hasattr(self.value, 'write'):
+        if isinstance(self.value, AttributeValue):
             try:
-                result = self.value.write(connection, value)
+                result = self.value.write(connection, decoded_value)
                 if inspect.isawaitable(result):
                     await result
             except ATT_Error as error:
@@ -924,9 +938,9 @@ class Attribute(EventEmitter):
                     error_code=error.error_code, att_handle=self.handle
                 ) from error
         else:
-            self.value = value
+            self.value = decoded_value
 
-        self.emit('write', connection, value)
+        self.emit('write', connection, decoded_value)
 
     def __repr__(self):
         if isinstance(self.value, bytes):

bumble/controller.py

@@ -25,8 +25,6 @@ import random
 import struct
 from bumble.colors import color
 from bumble.core import (
-    BT_CENTRAL_ROLE,
-    BT_PERIPHERAL_ROLE,
     BT_LE_TRANSPORT,
     BT_BR_EDR_TRANSPORT,
 )
@@ -47,6 +45,7 @@ from bumble.hci import (
     HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR,
     HCI_VERSION_BLUETOOTH_CORE_5_0,
     Address,
+    Role,
     HCI_AclDataPacket,
     HCI_AclDataPacketAssembler,
     HCI_Command_Complete_Event,
@@ -98,7 +97,7 @@ class CisLink:
 class Connection:
     controller: Controller
     handle: int
-    role: int
+    role: Role
     peer_address: Address
     link: Any
     transport: int
@@ -390,7 +389,7 @@ class Controller:
             connection = Connection(
                 controller=self,
                 handle=connection_handle,
-                role=BT_PERIPHERAL_ROLE,
+                role=Role.PERIPHERAL,
                 peer_address=peer_address,
                 link=self.link,
                 transport=BT_LE_TRANSPORT,
@@ -450,7 +449,7 @@ class Controller:
                 connection = Connection(
                     controller=self,
                     handle=connection_handle,
-                    role=BT_CENTRAL_ROLE,
+                    role=Role.CENTRAL,
                     peer_address=peer_address,
                     link=self.link,
                     transport=BT_LE_TRANSPORT,
@@ -469,7 +468,7 @@ class Controller:
             HCI_LE_Connection_Complete_Event(
                 status=status,
                 connection_handle=connection.handle if connection else 0,
-                role=BT_CENTRAL_ROLE,
+                role=Role.CENTRAL,
                 peer_address_type=le_create_connection_command.peer_address_type,
                 peer_address=le_create_connection_command.peer_address,
                 connection_interval=le_create_connection_command.connection_interval_min,
@@ -693,7 +692,7 @@ class Controller:
                     controller=self,
                     handle=connection_handle,
                     # Role doesn't matter in Classic because they are managed by HCI_Role_Change and HCI_Role_Discovery
-                    role=BT_CENTRAL_ROLE,
+                    role=Role.CENTRAL,
                     peer_address=peer_address,
                     link=self.link,
                     transport=BT_BR_EDR_TRANSPORT,
@@ -761,7 +760,7 @@ class Controller:
                 controller=self,
                 handle=connection_handle,
                 # Role doesn't matter in SCO.
-                role=BT_CENTRAL_ROLE,
+                role=Role.CENTRAL,
                 peer_address=peer_address,
                 link=self.link,
                 transport=BT_BR_EDR_TRANSPORT,

bumble/core.py

@@ -1,4 +1,4 @@
-# Copyright 2021-2022 Google LLC
+# 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.
@@ -16,10 +16,10 @@
 # Imports
 # -----------------------------------------------------------------------------
 from __future__ import annotations
-import dataclasses
+
 import enum
 import struct
-from typing import List, Optional, Tuple, Union, cast, Dict
+from typing import cast, overload, Literal, Union, Optional
 from typing_extensions import Self
 
 from bumble.company_ids import COMPANY_IDENTIFIERS
@@ -31,11 +31,12 @@ from bumble.utils import OpenIntEnum
 # -----------------------------------------------------------------------------
 # fmt: off
 
-BT_CENTRAL_ROLE    = 0
+class PhysicalTransport(enum.IntEnum):
-BT_PERIPHERAL_ROLE = 1
+    BR_EDR = 0
+    LE     = 1
 
-BT_BR_EDR_TRANSPORT = 0
+BT_BR_EDR_TRANSPORT = PhysicalTransport.BR_EDR
-BT_LE_TRANSPORT     = 1
+BT_LE_TRANSPORT     = PhysicalTransport.LE
 
 
 # fmt: on
@@ -57,7 +58,7 @@ def bit_flags_to_strings(bits, bit_flag_names):
     return names
 
 
-def name_or_number(dictionary: Dict[int, str], number: int, width: int = 2) -> str:
+def name_or_number(dictionary: dict[int, str], number: int, width: int = 2) -> str:
     name = dictionary.get(number)
     if name is not None:
         return name
@@ -200,7 +201,7 @@ class UUID:
     '''
 
     BASE_UUID = bytes.fromhex('00001000800000805F9B34FB')[::-1]  # little-endian
-    UUIDS: List[UUID] = []  # Registry of all instances created
+    UUIDS: list[UUID] = []  # Registry of all instances created
 
     uuid_bytes: bytes
     name: Optional[str]
@@ -259,11 +260,11 @@ class UUID:
         return cls.from_bytes(struct.pack('<I', uuid_32), name)
 
     @classmethod
-    def parse_uuid(cls, uuid_as_bytes: bytes, offset: int) -> Tuple[int, UUID]:
+    def parse_uuid(cls, uuid_as_bytes: bytes, offset: int) -> tuple[int, UUID]:
         return len(uuid_as_bytes), cls.from_bytes(uuid_as_bytes[offset:])
 
     @classmethod
-    def parse_uuid_2(cls, uuid_as_bytes: bytes, offset: int) -> Tuple[int, UUID]:
+    def parse_uuid_2(cls, uuid_as_bytes: bytes, offset: int) -> tuple[int, UUID]:
         return offset + 2, cls.from_bytes(uuid_as_bytes[offset : offset + 2])
 
     def to_bytes(self, force_128: bool = False) -> bytes:
@@ -1280,13 +1281,13 @@ class Appearance:
 # Advertising Data
 # -----------------------------------------------------------------------------
 AdvertisingDataObject = Union[
-    List[UUID],
+    list[UUID],
-    Tuple[UUID, bytes],
+    tuple[UUID, bytes],
     bytes,
     str,
     int,
-    Tuple[int, int],
+    tuple[int, int],
-    Tuple[int, bytes],
+    tuple[int, bytes],
     Appearance,
 ]
 
@@ -1295,116 +1296,116 @@ class AdvertisingData:
     # fmt: off
     # pylint: disable=line-too-long
 
-    FLAGS                                            = 0x01
+    class Type(OpenIntEnum):
-    INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS    = 0x02
+        FLAGS                                               = 0x01
-    COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS      = 0x03
+        INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS       = 0x02
-    INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS    = 0x04
+        COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS         = 0x03
-    COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS      = 0x05
+        INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS       = 0x04
-    INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS   = 0x06
+        COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS         = 0x05
-    COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS     = 0x07
+        INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS      = 0x06
-    SHORTENED_LOCAL_NAME                             = 0x08
+        COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS        = 0x07
-    COMPLETE_LOCAL_NAME                              = 0x09
+        SHORTENED_LOCAL_NAME                                = 0x08
-    TX_POWER_LEVEL                                   = 0x0A
+        COMPLETE_LOCAL_NAME                                 = 0x09
-    CLASS_OF_DEVICE                                  = 0x0D
+        TX_POWER_LEVEL                                      = 0x0A
-    SIMPLE_PAIRING_HASH_C                            = 0x0E
+        CLASS_OF_DEVICE                                     = 0x0D
-    SIMPLE_PAIRING_HASH_C_192                        = 0x0E
+        SIMPLE_PAIRING_HASH_C                               = 0x0E
-    SIMPLE_PAIRING_RANDOMIZER_R                      = 0x0F
+        SIMPLE_PAIRING_HASH_C_192                           = 0x0E
-    SIMPLE_PAIRING_RANDOMIZER_R_192                  = 0x0F
+        SIMPLE_PAIRING_RANDOMIZER_R                         = 0x0F
-    DEVICE_ID                                        = 0x10
+        SIMPLE_PAIRING_RANDOMIZER_R_192                     = 0x0F
-    SECURITY_MANAGER_TK_VALUE                        = 0x10
+        DEVICE_ID                                           = 0x10
-    SECURITY_MANAGER_OUT_OF_BAND_FLAGS               = 0x11
+        SECURITY_MANAGER_TK_VALUE                           = 0x10
-    PERIPHERAL_CONNECTION_INTERVAL_RANGE             = 0x12
+        SECURITY_MANAGER_OUT_OF_BAND_FLAGS                  = 0x11
-    LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS        = 0x14
+        PERIPHERAL_CONNECTION_INTERVAL_RANGE                = 0x12
-    LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS       = 0x15
+        LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS           = 0x14
-    SERVICE_DATA                                     = 0x16
+        LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS          = 0x15
-    SERVICE_DATA_16_BIT_UUID                         = 0x16
+        SERVICE_DATA_16_BIT_UUID                            = 0x16
-    PUBLIC_TARGET_ADDRESS                            = 0x17
+        PUBLIC_TARGET_ADDRESS                               = 0x17
-    RANDOM_TARGET_ADDRESS                            = 0x18
+        RANDOM_TARGET_ADDRESS                               = 0x18
-    APPEARANCE                                       = 0x19
+        APPEARANCE                                          = 0x19
-    ADVERTISING_INTERVAL                             = 0x1A
+        ADVERTISING_INTERVAL                                = 0x1A
-    LE_BLUETOOTH_DEVICE_ADDRESS                      = 0x1B
+        LE_BLUETOOTH_DEVICE_ADDRESS                         = 0x1B
-    LE_ROLE                                          = 0x1C
+        LE_ROLE                                             = 0x1C
-    SIMPLE_PAIRING_HASH_C_256                        = 0x1D
+        SIMPLE_PAIRING_HASH_C_256                           = 0x1D
-    SIMPLE_PAIRING_RANDOMIZER_R_256                  = 0x1E
+        SIMPLE_PAIRING_RANDOMIZER_R_256                     = 0x1E
-    LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS        = 0x1F
+        LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS           = 0x1F
-    SERVICE_DATA_32_BIT_UUID                         = 0x20
+        SERVICE_DATA_32_BIT_UUID                            = 0x20
-    SERVICE_DATA_128_BIT_UUID                        = 0x21
+        SERVICE_DATA_128_BIT_UUID                           = 0x21
-    LE_SECURE_CONNECTIONS_CONFIRMATION_VALUE         = 0x22
+        LE_SECURE_CONNECTIONS_CONFIRMATION_VALUE            = 0x22
-    LE_SECURE_CONNECTIONS_RANDOM_VALUE               = 0x23
+        LE_SECURE_CONNECTIONS_RANDOM_VALUE                  = 0x23
-    URI                                              = 0x24
+        URI                                                 = 0x24
-    INDOOR_POSITIONING                               = 0x25
+        INDOOR_POSITIONING                                  = 0x25
-    TRANSPORT_DISCOVERY_DATA                         = 0x26
+        TRANSPORT_DISCOVERY_DATA                            = 0x26
-    LE_SUPPORTED_FEATURES                            = 0x27
+        LE_SUPPORTED_FEATURES                               = 0x27
-    CHANNEL_MAP_UPDATE_INDICATION                    = 0x28
+        CHANNEL_MAP_UPDATE_INDICATION                       = 0x28
-    PB_ADV                                           = 0x29
+        PB_ADV                                              = 0x29
-    MESH_MESSAGE                                     = 0x2A
+        MESH_MESSAGE                                        = 0x2A
-    MESH_BEACON                                      = 0x2B
+        MESH_BEACON                                         = 0x2B
-    BIGINFO                                          = 0x2C
+        BIGINFO                                             = 0x2C
-    BROADCAST_CODE                                   = 0x2D
+        BROADCAST_CODE                                      = 0x2D
-    RESOLVABLE_SET_IDENTIFIER                        = 0x2E
+        RESOLVABLE_SET_IDENTIFIER                           = 0x2E
-    ADVERTISING_INTERVAL_LONG                        = 0x2F
+        ADVERTISING_INTERVAL_LONG                           = 0x2F
-    BROADCAST_NAME                                   = 0x30
+        BROADCAST_NAME                                      = 0x30
-    ENCRYPTED_ADVERTISING_DATA                       = 0X31
+        ENCRYPTED_ADVERTISING_DATA                          = 0x31
-    PERIODIC_ADVERTISING_RESPONSE_TIMING_INFORMATION = 0X32
+        PERIODIC_ADVERTISING_RESPONSE_TIMING_INFORMATION    = 0x32
-    ELECTRONIC_SHELF_LABEL                           = 0X34
+        ELECTRONIC_SHELF_LABEL                              = 0x34
-    THREE_D_INFORMATION_DATA                         = 0x3D
+        THREE_D_INFORMATION_DATA                            = 0x3D
-    MANUFACTURER_SPECIFIC_DATA                       = 0xFF
+        MANUFACTURER_SPECIFIC_DATA                          = 0xFF
 
-    AD_TYPE_NAMES = {
+    # For backward-compatibility
-        FLAGS:                                            'FLAGS',
+    FLAGS                                            = Type.FLAGS
-        INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS:    'INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS',
+    INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS    = Type.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS
-        COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS:      'COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS',
+    COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS      = Type.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS
-        INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS:    'INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS',
+    INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS    = Type.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS
-        COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS:      'COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS',
+    COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS      = Type.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS
-        INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS:   'INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS',
+    INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS   = Type.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS
-        COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS:     'COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS',
+    COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS     = Type.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS
-        SHORTENED_LOCAL_NAME:                             'SHORTENED_LOCAL_NAME',
+    SHORTENED_LOCAL_NAME                             = Type.SHORTENED_LOCAL_NAME
-        COMPLETE_LOCAL_NAME:                              'COMPLETE_LOCAL_NAME',
+    COMPLETE_LOCAL_NAME                              = Type.COMPLETE_LOCAL_NAME
-        TX_POWER_LEVEL:                                   'TX_POWER_LEVEL',
+    TX_POWER_LEVEL                                   = Type.TX_POWER_LEVEL
-        CLASS_OF_DEVICE:                                  'CLASS_OF_DEVICE',
+    CLASS_OF_DEVICE                                  = Type.CLASS_OF_DEVICE
-        SIMPLE_PAIRING_HASH_C:                            'SIMPLE_PAIRING_HASH_C',
+    SIMPLE_PAIRING_HASH_C                            = Type.SIMPLE_PAIRING_HASH_C
-        SIMPLE_PAIRING_HASH_C_192:                        'SIMPLE_PAIRING_HASH_C_192',
+    SIMPLE_PAIRING_HASH_C_192                        = Type.SIMPLE_PAIRING_HASH_C_192
-        SIMPLE_PAIRING_RANDOMIZER_R:                      'SIMPLE_PAIRING_RANDOMIZER_R',
+    SIMPLE_PAIRING_RANDOMIZER_R                      = Type.SIMPLE_PAIRING_RANDOMIZER_R
-        SIMPLE_PAIRING_RANDOMIZER_R_192:                  'SIMPLE_PAIRING_RANDOMIZER_R_192',
+    SIMPLE_PAIRING_RANDOMIZER_R_192                  = Type.SIMPLE_PAIRING_RANDOMIZER_R_192
-        DEVICE_ID:                                        'DEVICE_ID',
+    DEVICE_ID                                        = Type.DEVICE_ID
-        SECURITY_MANAGER_TK_VALUE:                        'SECURITY_MANAGER_TK_VALUE',
+    SECURITY_MANAGER_TK_VALUE                        = Type.SECURITY_MANAGER_TK_VALUE
-        SECURITY_MANAGER_OUT_OF_BAND_FLAGS:               'SECURITY_MANAGER_OUT_OF_BAND_FLAGS',
+    SECURITY_MANAGER_OUT_OF_BAND_FLAGS               = Type.SECURITY_MANAGER_OUT_OF_BAND_FLAGS
-        PERIPHERAL_CONNECTION_INTERVAL_RANGE:             'PERIPHERAL_CONNECTION_INTERVAL_RANGE',
+    PERIPHERAL_CONNECTION_INTERVAL_RANGE             = Type.PERIPHERAL_CONNECTION_INTERVAL_RANGE
-        LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS:        'LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS',
+    LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS        = Type.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS
-        LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS:       'LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS',
+    LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS       = Type.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS
-        SERVICE_DATA_16_BIT_UUID:                         'SERVICE_DATA_16_BIT_UUID',
+    SERVICE_DATA                                     = Type.SERVICE_DATA_16_BIT_UUID
-        PUBLIC_TARGET_ADDRESS:                            'PUBLIC_TARGET_ADDRESS',
+    SERVICE_DATA_16_BIT_UUID                         = Type.SERVICE_DATA_16_BIT_UUID
-        RANDOM_TARGET_ADDRESS:                            'RANDOM_TARGET_ADDRESS',
+    PUBLIC_TARGET_ADDRESS                            = Type.PUBLIC_TARGET_ADDRESS
-        APPEARANCE:                                       'APPEARANCE',
+    RANDOM_TARGET_ADDRESS                            = Type.RANDOM_TARGET_ADDRESS
-        ADVERTISING_INTERVAL:                             'ADVERTISING_INTERVAL',
+    APPEARANCE                                       = Type.APPEARANCE
-        LE_BLUETOOTH_DEVICE_ADDRESS:                      'LE_BLUETOOTH_DEVICE_ADDRESS',
+    ADVERTISING_INTERVAL                             = Type.ADVERTISING_INTERVAL
-        LE_ROLE:                                          'LE_ROLE',
+    LE_BLUETOOTH_DEVICE_ADDRESS                      = Type.LE_BLUETOOTH_DEVICE_ADDRESS
-        SIMPLE_PAIRING_HASH_C_256:                        'SIMPLE_PAIRING_HASH_C_256',
+    LE_ROLE                                          = Type.LE_ROLE
-        SIMPLE_PAIRING_RANDOMIZER_R_256:                  'SIMPLE_PAIRING_RANDOMIZER_R_256',
+    SIMPLE_PAIRING_HASH_C_256                        = Type.SIMPLE_PAIRING_HASH_C_256
-        LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS:        'LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS',
+    SIMPLE_PAIRING_RANDOMIZER_R_256                  = Type.SIMPLE_PAIRING_RANDOMIZER_R_256
-        SERVICE_DATA_32_BIT_UUID:                         'SERVICE_DATA_32_BIT_UUID',
+    LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS        = Type.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS
-        SERVICE_DATA_128_BIT_UUID:                        'SERVICE_DATA_128_BIT_UUID',
+    SERVICE_DATA_32_BIT_UUID                         = Type.SERVICE_DATA_32_BIT_UUID
-        LE_SECURE_CONNECTIONS_CONFIRMATION_VALUE:         'LE_SECURE_CONNECTIONS_CONFIRMATION_VALUE',
+    SERVICE_DATA_128_BIT_UUID                        = Type.SERVICE_DATA_128_BIT_UUID
-        LE_SECURE_CONNECTIONS_RANDOM_VALUE:               'LE_SECURE_CONNECTIONS_RANDOM_VALUE',
+    LE_SECURE_CONNECTIONS_CONFIRMATION_VALUE         = Type.LE_SECURE_CONNECTIONS_CONFIRMATION_VALUE
-        URI:                                              'URI',
+    LE_SECURE_CONNECTIONS_RANDOM_VALUE               = Type.LE_SECURE_CONNECTIONS_RANDOM_VALUE
-        INDOOR_POSITIONING:                               'INDOOR_POSITIONING',
+    URI                                              = Type.URI
-        TRANSPORT_DISCOVERY_DATA:                         'TRANSPORT_DISCOVERY_DATA',
+    INDOOR_POSITIONING                               = Type.INDOOR_POSITIONING
-        LE_SUPPORTED_FEATURES:                            'LE_SUPPORTED_FEATURES',
+    TRANSPORT_DISCOVERY_DATA                         = Type.TRANSPORT_DISCOVERY_DATA
-        CHANNEL_MAP_UPDATE_INDICATION:                    'CHANNEL_MAP_UPDATE_INDICATION',
+    LE_SUPPORTED_FEATURES                            = Type.LE_SUPPORTED_FEATURES
-        PB_ADV:                                           'PB_ADV',
+    CHANNEL_MAP_UPDATE_INDICATION                    = Type.CHANNEL_MAP_UPDATE_INDICATION
-        MESH_MESSAGE:                                     'MESH_MESSAGE',
+    PB_ADV                                           = Type.PB_ADV
-        MESH_BEACON:                                      'MESH_BEACON',
+    MESH_MESSAGE                                     = Type.MESH_MESSAGE
-        BIGINFO:                                          'BIGINFO',
+    MESH_BEACON                                      = Type.MESH_BEACON
-        BROADCAST_CODE:                                   'BROADCAST_CODE',
+    BIGINFO                                          = Type.BIGINFO
-        RESOLVABLE_SET_IDENTIFIER:                        'RESOLVABLE_SET_IDENTIFIER',
+    BROADCAST_CODE                                   = Type.BROADCAST_CODE
-        ADVERTISING_INTERVAL_LONG:                        'ADVERTISING_INTERVAL_LONG',
+    RESOLVABLE_SET_IDENTIFIER                        = Type.RESOLVABLE_SET_IDENTIFIER
-        BROADCAST_NAME:                                   'BROADCAST_NAME',
+    ADVERTISING_INTERVAL_LONG                        = Type.ADVERTISING_INTERVAL_LONG
-        ENCRYPTED_ADVERTISING_DATA:                       'ENCRYPTED_ADVERTISING_DATA',
+    BROADCAST_NAME                                   = Type.BROADCAST_NAME
-        PERIODIC_ADVERTISING_RESPONSE_TIMING_INFORMATION: 'PERIODIC_ADVERTISING_RESPONSE_TIMING_INFORMATION',
+    ENCRYPTED_ADVERTISING_DATA                       = Type.ENCRYPTED_ADVERTISING_DATA
-        ELECTRONIC_SHELF_LABEL:                           'ELECTRONIC_SHELF_LABEL',
+    PERIODIC_ADVERTISING_RESPONSE_TIMING_INFORMATION = Type.PERIODIC_ADVERTISING_RESPONSE_TIMING_INFORMATION
-        THREE_D_INFORMATION_DATA:                         'THREE_D_INFORMATION_DATA',
+    ELECTRONIC_SHELF_LABEL                           = Type.ELECTRONIC_SHELF_LABEL
-        MANUFACTURER_SPECIFIC_DATA:                       'MANUFACTURER_SPECIFIC_DATA'
+    THREE_D_INFORMATION_DATA                         = Type.THREE_D_INFORMATION_DATA
-    }
+    MANUFACTURER_SPECIFIC_DATA                       = Type.MANUFACTURER_SPECIFIC_DATA
 
     LE_LIMITED_DISCOVERABLE_MODE_FLAG = 0x01
     LE_GENERAL_DISCOVERABLE_MODE_FLAG = 0x02
@@ -1412,12 +1413,12 @@ class AdvertisingData:
     BR_EDR_CONTROLLER_FLAG            = 0x08
     BR_EDR_HOST_FLAG                  = 0x10
 
-    ad_structures: List[Tuple[int, bytes]]
+    ad_structures: list[tuple[int, bytes]]
 
     # fmt: on
     # pylint: enable=line-too-long
 
-    def __init__(self, ad_structures: Optional[List[Tuple[int, bytes]]] = None) -> None:
+    def __init__(self, ad_structures: Optional[list[tuple[int, bytes]]] = None) -> None:
         if ad_structures is None:
             ad_structures = []
         self.ad_structures = ad_structures[:]
@@ -1444,7 +1445,7 @@ class AdvertisingData:
         return ','.join(bit_flags_to_strings(flags, flag_names))
 
     @staticmethod
-    def uuid_list_to_objects(ad_data: bytes, uuid_size: int) -> List[UUID]:
+    def uuid_list_to_objects(ad_data: bytes, uuid_size: int) -> list[UUID]:
         uuids = []
         offset = 0
         while (offset + uuid_size) <= len(ad_data):
@@ -1461,8 +1462,8 @@ class AdvertisingData:
             ]
         )
 
-    @staticmethod
+    @classmethod
-    def ad_data_to_string(ad_type, ad_data):
+    def ad_data_to_string(cls, ad_type: int, ad_data: bytes) -> str:
         if ad_type == AdvertisingData.FLAGS:
             ad_type_str = 'Flags'
             ad_data_str = AdvertisingData.flags_to_string(ad_data[0], short=True)
@@ -1521,72 +1522,72 @@ class AdvertisingData:
             ad_type_str = 'Broadcast Name'
             ad_data_str = ad_data.decode('utf-8')
         else:
-            ad_type_str = AdvertisingData.AD_TYPE_NAMES.get(ad_type, f'0x{ad_type:02X}')
+            ad_type_str = AdvertisingData.Type(ad_type).name
             ad_data_str = ad_data.hex()
 
         return f'[{ad_type_str}]: {ad_data_str}'
 
     # pylint: disable=too-many-return-statements
-    @staticmethod
+    @classmethod
-    def ad_data_to_object(ad_type: int, ad_data: bytes) -> AdvertisingDataObject:
+    def ad_data_to_object(cls, ad_type: int, ad_data: bytes) -> AdvertisingDataObject:
         if ad_type in (
-            AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
-            AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
-            AdvertisingData.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS,
+            AdvertisingData.Type.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS,
         ):
             return AdvertisingData.uuid_list_to_objects(ad_data, 2)
 
         if ad_type in (
-            AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
-            AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
-            AdvertisingData.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS,
+            AdvertisingData.Type.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS,
         ):
             return AdvertisingData.uuid_list_to_objects(ad_data, 4)
 
         if ad_type in (
-            AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
-            AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
-            AdvertisingData.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS,
+            AdvertisingData.Type.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS,
         ):
             return AdvertisingData.uuid_list_to_objects(ad_data, 16)
 
-        if ad_type == AdvertisingData.SERVICE_DATA_16_BIT_UUID:
+        if ad_type == AdvertisingData.Type.SERVICE_DATA_16_BIT_UUID:
             return (UUID.from_bytes(ad_data[:2]), ad_data[2:])
 
-        if ad_type == AdvertisingData.SERVICE_DATA_32_BIT_UUID:
+        if ad_type == AdvertisingData.Type.SERVICE_DATA_32_BIT_UUID:
             return (UUID.from_bytes(ad_data[:4]), ad_data[4:])
 
-        if ad_type == AdvertisingData.SERVICE_DATA_128_BIT_UUID:
+        if ad_type == AdvertisingData.Type.SERVICE_DATA_128_BIT_UUID:
             return (UUID.from_bytes(ad_data[:16]), ad_data[16:])
 
         if ad_type in (
-            AdvertisingData.SHORTENED_LOCAL_NAME,
+            AdvertisingData.Type.SHORTENED_LOCAL_NAME,
-            AdvertisingData.COMPLETE_LOCAL_NAME,
+            AdvertisingData.Type.COMPLETE_LOCAL_NAME,
-            AdvertisingData.URI,
+            AdvertisingData.Type.URI,
-            AdvertisingData.BROADCAST_NAME,
+            AdvertisingData.Type.BROADCAST_NAME,
         ):
             return ad_data.decode("utf-8")
 
-        if ad_type in (AdvertisingData.TX_POWER_LEVEL, AdvertisingData.FLAGS):
+        if ad_type in (AdvertisingData.Type.TX_POWER_LEVEL, AdvertisingData.Type.FLAGS):
             return cast(int, struct.unpack('B', ad_data)[0])
 
-        if ad_type in (AdvertisingData.ADVERTISING_INTERVAL,):
+        if ad_type in (AdvertisingData.Type.ADVERTISING_INTERVAL,):
             return cast(int, struct.unpack('<H', ad_data)[0])
 
-        if ad_type == AdvertisingData.CLASS_OF_DEVICE:
+        if ad_type == AdvertisingData.Type.CLASS_OF_DEVICE:
             return cast(int, struct.unpack('<I', bytes([*ad_data, 0]))[0])
 
-        if ad_type == AdvertisingData.PERIPHERAL_CONNECTION_INTERVAL_RANGE:
+        if ad_type == AdvertisingData.Type.PERIPHERAL_CONNECTION_INTERVAL_RANGE:
-            return cast(Tuple[int, int], struct.unpack('<HH', ad_data))
+            return cast(tuple[int, int], struct.unpack('<HH', ad_data))
 
-        if ad_type == AdvertisingData.MANUFACTURER_SPECIFIC_DATA:
+        if ad_type == AdvertisingData.Type.APPEARANCE:
-            return (cast(int, struct.unpack_from('<H', ad_data, 0)[0]), ad_data[2:])
-
-        if ad_type == AdvertisingData.APPEARANCE:
             return Appearance.from_int(
                 cast(int, struct.unpack_from('<H', ad_data, 0)[0])
             )
 
+        if ad_type == AdvertisingData.Type.MANUFACTURER_SPECIFIC_DATA:
+            return (cast(int, struct.unpack_from('<H', ad_data, 0)[0]), ad_data[2:])
+
         return ad_data
 
     def append(self, data: bytes) -> None:
@@ -1600,7 +1601,80 @@ class AdvertisingData:
                 self.ad_structures.append((ad_type, ad_data))
             offset += length
 
-    def get_all(self, type_id: int, raw: bool = False) -> List[AdvertisingDataObject]:
+    @overload
+    def get_all(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS,
+            AdvertisingData.Type.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS,
+            AdvertisingData.Type.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS,
+        ],
+        raw: Literal[False] = False,
+    ) -> list[list[UUID]]: ...
+    @overload
+    def get_all(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.SERVICE_DATA_16_BIT_UUID,
+            AdvertisingData.Type.SERVICE_DATA_32_BIT_UUID,
+            AdvertisingData.Type.SERVICE_DATA_128_BIT_UUID,
+        ],
+        raw: Literal[False] = False,
+    ) -> list[tuple[UUID, bytes]]: ...
+    @overload
+    def get_all(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.SHORTENED_LOCAL_NAME,
+            AdvertisingData.Type.COMPLETE_LOCAL_NAME,
+            AdvertisingData.Type.URI,
+            AdvertisingData.Type.BROADCAST_NAME,
+        ],
+        raw: Literal[False] = False,
+    ) -> list[str]: ...
+    @overload
+    def get_all(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.TX_POWER_LEVEL,
+            AdvertisingData.Type.FLAGS,
+            AdvertisingData.Type.ADVERTISING_INTERVAL,
+            AdvertisingData.Type.CLASS_OF_DEVICE,
+        ],
+        raw: Literal[False] = False,
+    ) -> list[int]: ...
+    @overload
+    def get_all(
+        self,
+        type_id: Literal[AdvertisingData.Type.PERIPHERAL_CONNECTION_INTERVAL_RANGE,],
+        raw: Literal[False] = False,
+    ) -> list[tuple[int, int]]: ...
+    @overload
+    def get_all(
+        self,
+        type_id: Literal[AdvertisingData.Type.MANUFACTURER_SPECIFIC_DATA,],
+        raw: Literal[False] = False,
+    ) -> list[tuple[int, bytes]]: ...
+    @overload
+    def get_all(
+        self,
+        type_id: Literal[AdvertisingData.Type.APPEARANCE,],
+        raw: Literal[False] = False,
+    ) -> list[Appearance]: ...
+    @overload
+    def get_all(self, type_id: int, raw: Literal[True]) -> list[bytes]: ...
+    @overload
+    def get_all(
+        self, type_id: int, raw: bool = False
+    ) -> list[AdvertisingDataObject]: ...
+
+    def get_all(self, type_id: int, raw: bool = False) -> list[AdvertisingDataObject]:  # type: ignore[misc]
         '''
         Get Advertising Data Structure(s) with a given type
 
@@ -1612,6 +1686,79 @@ class AdvertisingData:
 
         return [process_ad_data(ad[1]) for ad in self.ad_structures if ad[0] == type_id]
 
+    @overload
+    def get(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS,
+            AdvertisingData.Type.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS,
+            AdvertisingData.Type.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
+            AdvertisingData.Type.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS,
+        ],
+        raw: Literal[False] = False,
+    ) -> Optional[list[UUID]]: ...
+    @overload
+    def get(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.SERVICE_DATA_16_BIT_UUID,
+            AdvertisingData.Type.SERVICE_DATA_32_BIT_UUID,
+            AdvertisingData.Type.SERVICE_DATA_128_BIT_UUID,
+        ],
+        raw: Literal[False] = False,
+    ) -> Optional[tuple[UUID, bytes]]: ...
+    @overload
+    def get(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.SHORTENED_LOCAL_NAME,
+            AdvertisingData.Type.COMPLETE_LOCAL_NAME,
+            AdvertisingData.Type.URI,
+            AdvertisingData.Type.BROADCAST_NAME,
+        ],
+        raw: Literal[False] = False,
+    ) -> Optional[Optional[str]]: ...
+    @overload
+    def get(
+        self,
+        type_id: Literal[
+            AdvertisingData.Type.TX_POWER_LEVEL,
+            AdvertisingData.Type.FLAGS,
+            AdvertisingData.Type.ADVERTISING_INTERVAL,
+            AdvertisingData.Type.CLASS_OF_DEVICE,
+        ],
+        raw: Literal[False] = False,
+    ) -> Optional[int]: ...
+    @overload
+    def get(
+        self,
+        type_id: Literal[AdvertisingData.Type.PERIPHERAL_CONNECTION_INTERVAL_RANGE,],
+        raw: Literal[False] = False,
+    ) -> Optional[tuple[int, int]]: ...
+    @overload
+    def get(
+        self,
+        type_id: Literal[AdvertisingData.Type.MANUFACTURER_SPECIFIC_DATA,],
+        raw: Literal[False] = False,
+    ) -> Optional[tuple[int, bytes]]: ...
+    @overload
+    def get(
+        self,
+        type_id: Literal[AdvertisingData.Type.APPEARANCE,],
+        raw: Literal[False] = False,
+    ) -> Optional[Appearance]: ...
+    @overload
+    def get(self, type_id: int, raw: Literal[True]) -> Optional[bytes]: ...
+    @overload
+    def get(
+        self, type_id: int, raw: bool = False
+    ) -> Optional[AdvertisingDataObject]: ...
+
     def get(self, type_id: int, raw: bool = False) -> Optional[AdvertisingDataObject]:
         '''
         Get Advertising Data Structure(s) with a given type

bumble/device.py

@@ -53,14 +53,12 @@ from pyee import EventEmitter
 
 from .colors import color
 from .att import ATT_CID, ATT_DEFAULT_MTU, ATT_PDU
-from .gatt import Characteristic, Descriptor, Service
+from .gatt import Attribute, Characteristic, Descriptor, Service
 from .host import DataPacketQueue, Host
 from .profiles.gap import GenericAccessService
 from .core import (
     BT_BR_EDR_TRANSPORT,
-    BT_CENTRAL_ROLE,
     BT_LE_TRANSPORT,
-    BT_PERIPHERAL_ROLE,
     AdvertisingData,
     BaseBumbleError,
     ConnectionParameterUpdateError,
@@ -1555,13 +1553,13 @@ class IsoPacketStream:
 class Connection(CompositeEventEmitter):
     device: Device
     handle: int
-    transport: int
+    transport: core.PhysicalTransport
     self_address: hci.Address
     self_resolvable_address: Optional[hci.Address]
     peer_address: hci.Address
     peer_resolvable_address: Optional[hci.Address]
     peer_le_features: Optional[hci.LeFeatureMask]
-    role: int
+    role: hci.Role
     encryption: int
     authenticated: bool
     sc: bool
@@ -1569,8 +1567,8 @@ class Connection(CompositeEventEmitter):
     gatt_client: gatt_client.Client
     pairing_peer_io_capability: Optional[int]
     pairing_peer_authentication_requirements: Optional[int]
-    cs_configs: dict[int, ChannelSoundingConfig] = {}  # Config ID to Configuration
+    cs_configs: dict[int, ChannelSoundingConfig]  # Config ID to Configuration
-    cs_procedures: dict[int, ChannelSoundingProcedure] = {}  # Config ID to Procedures
+    cs_procedures: dict[int, ChannelSoundingProcedure]  # Config ID to Procedures
 
     @composite_listener
     class Listener:
@@ -1637,6 +1635,8 @@ class Connection(CompositeEventEmitter):
         self.pairing_peer_io_capability = None
         self.pairing_peer_authentication_requirements = None
         self.peer_le_features = None
+        self.cs_configs = {}
+        self.cs_procedures = {}
 
     # [Classic only]
     @classmethod
@@ -1672,9 +1672,9 @@ class Connection(CompositeEventEmitter):
     def role_name(self):
         if self.role is None:
             return 'NOT-SET'
-        if self.role == BT_CENTRAL_ROLE:
+        if self.role == hci.Role.CENTRAL:
             return 'CENTRAL'
-        if self.role == BT_PERIPHERAL_ROLE:
+        if self.role == hci.Role.PERIPHERAL:
             return 'PERIPHERAL'
         return f'UNKNOWN[{self.role}]'
 
@@ -1732,7 +1732,7 @@ class Connection(CompositeEventEmitter):
     async def encrypt(self, enable: bool = True) -> None:
         return await self.device.encrypt(self, enable)
 
-    async def switch_role(self, role: int) -> None:
+    async def switch_role(self, role: hci.Role) -> None:
         return await self.device.switch_role(self, role)
 
     async def sustain(self, timeout: Optional[float] = None) -> None:
@@ -2046,9 +2046,9 @@ class Device(CompositeEventEmitter):
     legacy_advertiser: Optional[LegacyAdvertiser]
     sco_links: Dict[int, ScoLink]
     cis_links: Dict[int, CisLink]
-    bigs = dict[int, Big]()
+    bigs: dict[int, Big]
-    bis_links = dict[int, BisLink]()
+    bis_links: dict[int, BisLink]
-    big_syncs = dict[int, BigSync]()
+    big_syncs: dict[int, BigSync]
     _pending_cis: Dict[int, tuple[int, int]]
     gatt_service: gatt_service.GenericAttributeProfileService | None = None
 
@@ -2141,6 +2141,9 @@ class Device(CompositeEventEmitter):
         self.sco_links = {}  # ScoLinks, by connection handle (BR/EDR only)
         self.cis_links = {}  # CisLinks, by connection handle (LE only)
         self._pending_cis = {}  # (CIS_ID, CIG_ID), by CIS_handle
+        self.bigs = {}
+        self.bis_links = {}
+        self.big_syncs = {}
         self.classic_enabled = False
         self.inquiry_response = None
         self.address_resolver = None
@@ -2218,7 +2221,7 @@ class Device(CompositeEventEmitter):
                         permissions=descriptor["permissions"],
                     )
                     descriptors.append(new_descriptor)
-                new_characteristic = Characteristic(
+                new_characteristic: Characteristic[bytes] = Characteristic(
                     uuid=characteristic["uuid"],
                     properties=Characteristic.Properties.from_string(
                         characteristic["properties"]
@@ -2708,7 +2711,7 @@ class Device(CompositeEventEmitter):
         if phy == hci.HCI_LE_1M_PHY:
             return True
 
-        feature_map = {
+        feature_map: dict[int, hci.LeFeatureMask] = {
             hci.HCI_LE_2M_PHY: hci.LeFeatureMask.LE_2M_PHY,
             hci.HCI_LE_CODED_PHY: hci.LeFeatureMask.LE_CODED_PHY,
         }
@@ -2729,7 +2732,7 @@ class Device(CompositeEventEmitter):
         self,
         advertising_type: AdvertisingType = AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE,
         target: Optional[hci.Address] = None,
-        own_address_type: int = hci.OwnAddressType.RANDOM,
+        own_address_type: hci.OwnAddressType = hci.OwnAddressType.RANDOM,
         auto_restart: bool = False,
         advertising_data: Optional[bytes] = None,
         scan_response_data: Optional[bytes] = None,
@@ -3010,7 +3013,7 @@ class Device(CompositeEventEmitter):
         active: bool = True,
         scan_interval: float = DEVICE_DEFAULT_SCAN_INTERVAL,  # Scan interval in ms
         scan_window: float = DEVICE_DEFAULT_SCAN_WINDOW,  # Scan window in ms
-        own_address_type: int = hci.OwnAddressType.RANDOM,
+        own_address_type: hci.OwnAddressType = hci.OwnAddressType.RANDOM,
         filter_duplicates: bool = False,
         scanning_phys: Sequence[int] = (hci.HCI_LE_1M_PHY, hci.HCI_LE_CODED_PHY),
     ) -> None:
@@ -3086,7 +3089,7 @@ class Device(CompositeEventEmitter):
                 # pylint: disable=line-too-long
                 hci.HCI_LE_Set_Scan_Parameters_Command(
                     le_scan_type=scan_type,
-                    le_scan_interval=int(scan_window / 0.625),
+                    le_scan_interval=int(scan_interval / 0.625),
                     le_scan_window=int(scan_window / 0.625),
                     own_address_type=own_address_type,
                     scanning_filter_policy=hci.HCI_LE_Set_Scan_Parameters_Command.BASIC_UNFILTERED_POLICY,
@@ -3376,11 +3379,11 @@ class Device(CompositeEventEmitter):
     async def connect(
         self,
         peer_address: Union[hci.Address, str],
-        transport: int = BT_LE_TRANSPORT,
+        transport: core.PhysicalTransport = BT_LE_TRANSPORT,
         connection_parameters_preferences: Optional[
-            Dict[int, ConnectionParametersPreferences]
+            dict[hci.Phy, ConnectionParametersPreferences]
         ] = None,
-        own_address_type: int = hci.OwnAddressType.RANDOM,
+        own_address_type: hci.OwnAddressType = hci.OwnAddressType.RANDOM,
         timeout: Optional[float] = DEVICE_DEFAULT_CONNECT_TIMEOUT,
         always_resolve: bool = False,
     ) -> Connection:
@@ -3428,6 +3431,7 @@ class Device(CompositeEventEmitter):
         # Check parameters
         if transport not in (BT_LE_TRANSPORT, BT_BR_EDR_TRANSPORT):
             raise InvalidArgumentError('invalid transport')
+        transport = core.PhysicalTransport(transport)
 
         # Adjust the transport automatically if we need to
         if transport == BT_LE_TRANSPORT and not self.le_enabled:
@@ -3623,7 +3627,7 @@ class Device(CompositeEventEmitter):
             else:
                 # Save pending connection
                 self.pending_connections[peer_address] = Connection.incomplete(
-                    self, peer_address, BT_CENTRAL_ROLE
+                    self, peer_address, hci.Role.CENTRAL
                 )
 
                 # TODO: allow passing other settings
@@ -3678,7 +3682,7 @@ class Device(CompositeEventEmitter):
     async def accept(
         self,
         peer_address: Union[hci.Address, str] = hci.Address.ANY,
-        role: int = BT_PERIPHERAL_ROLE,
+        role: hci.Role = hci.Role.PERIPHERAL,
         timeout: Optional[float] = DEVICE_DEFAULT_CONNECT_TIMEOUT,
     ) -> Connection:
         '''
@@ -3764,12 +3768,12 @@ class Device(CompositeEventEmitter):
         self.on('connection', on_connection)
         self.on('connection_failure', on_connection_failure)
 
-        # Save pending connection, with the Peripheral role.
+        # Save pending connection, with the Peripheral hci.role.
         # Even if we requested a role switch in the hci.HCI_Accept_Connection_Request
         # command, this connection is still considered Peripheral until an eventual
         # role change event.
         self.pending_connections[peer_address] = Connection.incomplete(
-            self, peer_address, BT_PERIPHERAL_ROLE
+            self, peer_address, hci.Role.PERIPHERAL
         )
 
         try:
@@ -3898,7 +3902,7 @@ class Device(CompositeEventEmitter):
         '''
 
         if use_l2cap:
-            if connection.role != BT_PERIPHERAL_ROLE:
+            if connection.role != hci.Role.PERIPHERAL:
                 raise InvalidStateError(
                     'only peripheral can update connection parameters with l2cap'
                 )
@@ -4005,13 +4009,12 @@ class Device(CompositeEventEmitter):
         # Create a future to wait for an address to be found
         peer_address = asyncio.get_running_loop().create_future()
 
-        def on_peer_found(address, ad_data):
+        def on_peer_found(address: hci.Address, ad_data: AdvertisingData) -> None:
-            local_name = ad_data.get(AdvertisingData.COMPLETE_LOCAL_NAME, raw=True)
+            local_name = ad_data.get(
-            if local_name is None:
+                AdvertisingData.Type.COMPLETE_LOCAL_NAME
-                local_name = ad_data.get(AdvertisingData.SHORTENED_LOCAL_NAME, raw=True)
+            ) or ad_data.get(AdvertisingData.Type.SHORTENED_LOCAL_NAME)
-            if local_name is not None:
+            if local_name == name:
-                if local_name.decode('utf-8') == name:
+                peer_address.set_result(address)
-                    peer_address.set_result(address)
 
         listener = None
         was_scanning = self.scanning
@@ -4144,10 +4147,10 @@ class Device(CompositeEventEmitter):
                 if keys.ltk:
                     return keys.ltk.value
 
-                if connection.role == BT_CENTRAL_ROLE and keys.ltk_central:
+                if connection.role == hci.Role.CENTRAL and keys.ltk_central:
                     return keys.ltk_central.value
 
-                if connection.role == BT_PERIPHERAL_ROLE and keys.ltk_peripheral:
+                if connection.role == hci.Role.PERIPHERAL and keys.ltk_peripheral:
                     return keys.ltk_peripheral.value
         return None
 
@@ -4299,7 +4302,7 @@ class Device(CompositeEventEmitter):
             self.emit('key_store_update')
 
     # [Classic only]
-    async def switch_role(self, connection: Connection, role: int):
+    async def switch_role(self, connection: Connection, role: hci.Role):
         pending_role_change = asyncio.get_running_loop().create_future()
 
         def on_role_change(new_role):
@@ -4919,16 +4922,84 @@ class Device(CompositeEventEmitter):
             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):
+    async def notify_subscriber(
+        self,
+        connection: Connection,
+        attribute: Attribute,
+        value: Optional[Any] = None,
+        force: bool = False,
+    ) -> None:
+        """
+        Send a notification to an attribute subscriber.
+
+        Args:
+           connection:
+             The connection of the subscriber.
+           attribute:
+             The attribute whose value is notified.
+           value:
+             The value of the attribute (if None, the value is read from the attribute)
+            force:
+              If True, send a notification even if there is no subscriber.
+        """
         await self.gatt_server.notify_subscriber(connection, attribute, value, force)
 
-    async def notify_subscribers(self, attribute, value=None, force=False):
+    async def notify_subscribers(
+        self, attribute: Attribute, value=None, force=False
+    ) -> None:
+        """
+        Send a notification to all the subscribers of an attribute.
+
+        Args:
+           attribute:
+             The attribute whose value is notified.
+           value:
+             The value of the attribute (if None, the value is read from the attribute)
+            force:
+              If True, send a notification for every connection even if there is no
+              subscriber.
+        """
         await self.gatt_server.notify_subscribers(attribute, value, force)
 
-    async def indicate_subscriber(self, connection, attribute, value=None, force=False):
+    async def indicate_subscriber(
+        self,
+        connection: Connection,
+        attribute: Attribute,
+        value: Optional[Any] = None,
+        force: bool = False,
+    ):
+        """
+        Send an indication to an attribute subscriber.
+
+        This method returns when the response to the indication has been received.
+
+        Args:
+           connection:
+             The connection of the subscriber.
+           attribute:
+             The attribute whose value is indicated.
+           value:
+             The value of the attribute (if None, the value is read from the attribute)
+            force:
+              If True, send an indication even if there is no subscriber.
+        """
         await self.gatt_server.indicate_subscriber(connection, attribute, value, force)
 
-    async def indicate_subscribers(self, attribute, value=None, force=False):
+    async def indicate_subscribers(
+        self, attribute: Attribute, value: Optional[Any] = None, force: bool = False
+    ):
+        """
+        Send an indication to all the subscribers of an attribute.
+
+        Args:
+           attribute:
+             The attribute whose value is notified.
+           value:
+             The value of the attribute (if None, the value is read from the attribute)
+            force:
+              If True, send an indication for every connection even if there is no
+              subscriber.
+        """
         await self.gatt_server.indicate_subscribers(attribute, value, force)
 
     @host_event_handler
@@ -5106,11 +5177,11 @@ class Device(CompositeEventEmitter):
     def on_connection(
         self,
         connection_handle: int,
-        transport: int,
+        transport: core.PhysicalTransport,
         peer_address: hci.Address,
         self_resolvable_address: Optional[hci.Address],
         peer_resolvable_address: Optional[hci.Address],
-        role: int,
+        role: hci.Role,
         connection_parameters: ConnectionParameters,
     ) -> None:
         # Convert all-zeros addresses into None.
@@ -5153,7 +5224,7 @@ class Device(CompositeEventEmitter):
                         peer_address = resolved_address
 
         self_address = None
-        own_address_type: Optional[int] = None
+        own_address_type: Optional[hci.OwnAddressType] = None
         if role == hci.HCI_CENTRAL_ROLE:
             own_address_type = self.connect_own_address_type
             assert own_address_type is not None
@@ -5281,7 +5352,7 @@ class Device(CompositeEventEmitter):
         elif self.classic_accept_any:
             # Save pending connection
             self.pending_connections[bd_addr] = Connection.incomplete(
-                self, bd_addr, BT_PERIPHERAL_ROLE
+                self, bd_addr, hci.Role.PERIPHERAL
             )
 
             self.host.send_command_sync(

bumble/gatt.py

@@ -27,28 +27,16 @@ import enum
 import functools
 import logging
 import struct
-from typing import (
+from typing import Iterable, List, Optional, Sequence, TypeVar, Union
-    Any,
-    Callable,
-    Dict,
-    Iterable,
-    List,
-    Optional,
-    Sequence,
-    SupportsBytes,
-    Type,
-    Union,
-    TYPE_CHECKING,
-)
 
 from bumble.colors import color
-from bumble.core import BaseBumbleError, InvalidOperationError, UUID
+from bumble.core import BaseBumbleError, UUID
 from bumble.att import Attribute, AttributeValue
-from bumble.utils import ByteSerializable
-
-if TYPE_CHECKING:
-    from bumble.gatt_client import AttributeProxy
 
+# -----------------------------------------------------------------------------
+# Typing
+# -----------------------------------------------------------------------------
+_T = TypeVar('_T')
 
 # -----------------------------------------------------------------------------
 # Logging
@@ -436,7 +424,7 @@ class IncludedServiceDeclaration(Attribute):
 
 
 # -----------------------------------------------------------------------------
-class Characteristic(Attribute):
+class Characteristic(Attribute[_T]):
     '''
     See Vol 3, Part G - 3.3 CHARACTERISTIC DEFINITION
     '''
@@ -499,7 +487,7 @@ class Characteristic(Attribute):
         uuid: Union[str, bytes, UUID],
         properties: Characteristic.Properties,
         permissions: Union[str, Attribute.Permissions],
-        value: Any = b'',
+        value: Union[AttributeValue[_T], _T, None] = None,
         descriptors: Sequence[Descriptor] = (),
     ):
         super().__init__(uuid, permissions, value)
@@ -559,217 +547,10 @@ class CharacteristicDeclaration(Attribute):
 
 
 # -----------------------------------------------------------------------------
-class CharacteristicValue(AttributeValue):
+class CharacteristicValue(AttributeValue[_T]):
     """Same as AttributeValue, for backward compatibility"""
 
 
-# -----------------------------------------------------------------------------
-class CharacteristicAdapter:
-    '''
-    An adapter that can adapt Characteristic and AttributeProxy objects
-    by wrapping their `read_value()` and `write_value()` methods with ones that
-    return/accept encoded/decoded values.
-
-    For proxies (i.e used by a GATT client), the adaptation is one where the return
-    value of `read_value()` is decoded and the value passed to `write_value()` is
-    encoded. The `subscribe()` method, is wrapped with one where the values are decoded
-    before being passed to the subscriber.
-
-    For local values (i.e hosted by a GATT server) the adaptation is one where the
-    return value of `read_value()` is encoded and the value passed to `write_value()`
-    is decoded.
-    '''
-
-    read_value: Callable
-    write_value: Callable
-
-    def __init__(self, characteristic: Union[Characteristic, AttributeProxy]):
-        self.wrapped_characteristic = characteristic
-        self.subscribers: Dict[Callable, Callable] = (
-            {}
-        )  # Map from subscriber to proxy subscriber
-
-        if isinstance(characteristic, Characteristic):
-            self.read_value = self.read_encoded_value
-            self.write_value = self.write_encoded_value
-        else:
-            self.read_value = self.read_decoded_value
-            self.write_value = self.write_decoded_value
-            self.subscribe = self.wrapped_subscribe
-            self.unsubscribe = self.wrapped_unsubscribe
-
-    def __getattr__(self, name):
-        return getattr(self.wrapped_characteristic, name)
-
-    def __setattr__(self, name, value):
-        if name in (
-            'wrapped_characteristic',
-            'subscribers',
-            'read_value',
-            'write_value',
-            'subscribe',
-            'unsubscribe',
-        ):
-            super().__setattr__(name, value)
-        else:
-            setattr(self.wrapped_characteristic, name, value)
-
-    async def read_encoded_value(self, connection):
-        return self.encode_value(
-            await self.wrapped_characteristic.read_value(connection)
-        )
-
-    async def write_encoded_value(self, connection, value):
-        return await 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, with_response=False):
-        return await self.wrapped_characteristic.write_value(
-            self.encode_value(value), with_response
-        )
-
-    def encode_value(self, value):
-        return value
-
-    def decode_value(self, value):
-        return value
-
-    def wrapped_subscribe(self, subscriber=None):
-        if subscriber is not None:
-            if subscriber in self.subscribers:
-                # We already have a proxy subscriber
-                subscriber = self.subscribers[subscriber]
-            else:
-                # Create and register a proxy that will decode the value
-                original_subscriber = subscriber
-
-                def on_change(value):
-                    original_subscriber(self.decode_value(value))
-
-                self.subscribers[subscriber] = on_change
-                subscriber = on_change
-
-        return self.wrapped_characteristic.subscribe(subscriber)
-
-    def wrapped_unsubscribe(self, subscriber=None):
-        if subscriber in self.subscribers:
-            subscriber = self.subscribers.pop(subscriber)
-
-        return self.wrapped_characteristic.unsubscribe(subscriber)
-
-    def __str__(self) -> str:
-        wrapped = str(self.wrapped_characteristic)
-        return f'{self.__class__.__name__}({wrapped})'
-
-
-# -----------------------------------------------------------------------------
-class DelegatedCharacteristicAdapter(CharacteristicAdapter):
-    '''
-    Adapter that converts bytes values using an encode and a decode function.
-    '''
-
-    def __init__(self, characteristic, encode=None, decode=None):
-        super().__init__(characteristic)
-        self.encode = encode
-        self.decode = decode
-
-    def encode_value(self, value):
-        if self.encode is None:
-            raise InvalidOperationError('delegated adapter does not have an encoder')
-        return self.encode(value)
-
-    def decode_value(self, value):
-        if self.decode is None:
-            raise InvalidOperationError('delegate adapter does not have a decoder')
-        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, pack_format):
-        super().__init__(characteristic)
-        self.struct = struct.Struct(pack_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 isinstance(value, 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 a 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, pack_format, keys):
-        super().__init__(characteristic, pack_format)
-        self.keys = keys
-
-    # pylint: disable=arguments-differ
-    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: str) -> bytes:
-        return value.encode('utf-8')
-
-    def decode_value(self, value: bytes) -> str:
-        return value.decode('utf-8')
-
-
-# -----------------------------------------------------------------------------
-class SerializableCharacteristicAdapter(CharacteristicAdapter):
-    '''
-    Adapter that converts any class to/from bytes using the class'
-    `to_bytes` and `__bytes__` methods, respectively.
-    '''
-
-    def __init__(self, characteristic, cls: Type[ByteSerializable]):
-        super().__init__(characteristic)
-        self.cls = cls
-
-    def encode_value(self, value: SupportsBytes) -> bytes:
-        return bytes(value)
-
-    def decode_value(self, value: bytes) -> Any:
-        return self.cls.from_bytes(value)
-
-
 # -----------------------------------------------------------------------------
 class Descriptor(Attribute):
     '''

bumble/gatt_adapters.py

@@ -0,0 +1,374 @@
+# Copyright 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.
+
+# -----------------------------------------------------------------------------
+# GATT - Type Adapters
+# -----------------------------------------------------------------------------
+
+# -----------------------------------------------------------------------------
+# Imports
+# -----------------------------------------------------------------------------
+from __future__ import annotations
+import struct
+from typing import (
+    Any,
+    Callable,
+    Generic,
+    Iterable,
+    Literal,
+    Optional,
+    Type,
+    TypeVar,
+)
+
+from bumble.core import InvalidOperationError
+from bumble.gatt import Characteristic
+from bumble.gatt_client import CharacteristicProxy
+from bumble.utils import ByteSerializable, IntConvertible
+
+
+# -----------------------------------------------------------------------------
+# Typing
+# -----------------------------------------------------------------------------
+_T = TypeVar('_T')
+_T2 = TypeVar('_T2', bound=ByteSerializable)
+_T3 = TypeVar('_T3', bound=IntConvertible)
+
+
+# -----------------------------------------------------------------------------
+class CharacteristicAdapter(Characteristic, Generic[_T]):
+    '''Base class for GATT Characteristic adapters.'''
+
+    def __init__(self, characteristic: Characteristic) -> None:
+        super().__init__(
+            characteristic.uuid,
+            characteristic.properties,
+            characteristic.permissions,
+            characteristic.value,
+            characteristic.descriptors,
+        )
+
+
+# -----------------------------------------------------------------------------
+class CharacteristicProxyAdapter(CharacteristicProxy[_T]):
+    '''Base class for GATT CharacteristicProxy adapters.'''
+
+    def __init__(self, characteristic_proxy: CharacteristicProxy):
+        super().__init__(
+            characteristic_proxy.client,
+            characteristic_proxy.handle,
+            characteristic_proxy.end_group_handle,
+            characteristic_proxy.uuid,
+            characteristic_proxy.properties,
+        )
+
+
+# -----------------------------------------------------------------------------
+class DelegatedCharacteristicAdapter(CharacteristicAdapter[_T]):
+    '''
+    Adapter that converts bytes values using an encode and/or a decode function.
+    '''
+
+    def __init__(
+        self,
+        characteristic: Characteristic,
+        encode: Optional[Callable[[_T], bytes]] = None,
+        decode: Optional[Callable[[bytes], _T]] = None,
+    ):
+        super().__init__(characteristic)
+        self.encode = encode
+        self.decode = decode
+
+    def encode_value(self, value: _T) -> bytes:
+        if self.encode is None:
+            raise InvalidOperationError('delegated adapter does not have an encoder')
+        return self.encode(value)
+
+    def decode_value(self, value: bytes) -> _T:
+        if self.decode is None:
+            raise InvalidOperationError('delegate adapter does not have a decoder')
+        return self.decode(value)
+
+
+# -----------------------------------------------------------------------------
+class DelegatedCharacteristicProxyAdapter(CharacteristicProxyAdapter[_T]):
+    '''
+    Adapter that converts bytes values using an encode and a decode function.
+    '''
+
+    def __init__(
+        self,
+        characteristic_proxy: CharacteristicProxy,
+        encode: Optional[Callable[[_T], bytes]] = None,
+        decode: Optional[Callable[[bytes], _T]] = None,
+    ):
+        super().__init__(characteristic_proxy)
+        self.encode = encode
+        self.decode = decode
+
+    def encode_value(self, value: _T) -> bytes:
+        if self.encode is None:
+            raise InvalidOperationError('delegated adapter does not have an encoder')
+        return self.encode(value)
+
+    def decode_value(self, value: bytes) -> _T:
+        if self.decode is None:
+            raise InvalidOperationError('delegate adapter does not have a decoder')
+        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: Characteristic, pack_format: str) -> None:
+        super().__init__(characteristic)
+        self.struct = struct.Struct(pack_format)
+
+    def pack(self, *values) -> bytes:
+        return self.struct.pack(*values)
+
+    def unpack(self, buffer: bytes) -> tuple:
+        return self.struct.unpack(buffer)
+
+    def encode_value(self, value: Any) -> bytes:
+        return self.pack(*value if isinstance(value, tuple) else (value,))
+
+    def decode_value(self, value: bytes) -> Any:
+        unpacked = self.unpack(value)
+        return unpacked[0] if len(unpacked) == 1 else unpacked
+
+
+# -----------------------------------------------------------------------------
+class PackedCharacteristicProxyAdapter(CharacteristicProxyAdapter):
+    '''
+    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_proxy, pack_format):
+        super().__init__(characteristic_proxy)
+        self.struct = struct.Struct(pack_format)
+
+    def pack(self, *values) -> bytes:
+        return self.struct.pack(*values)
+
+    def unpack(self, buffer: bytes) -> tuple:
+        return self.struct.unpack(buffer)
+
+    def encode_value(self, value: Any) -> bytes:
+        return self.pack(*value if isinstance(value, tuple) else (value,))
+
+    def decode_value(self, value: bytes) -> Any:
+        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 a 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: Characteristic, pack_format: str, keys: Iterable[str]
+    ) -> None:
+        super().__init__(characteristic, pack_format)
+        self.keys = keys
+
+    # pylint: disable=arguments-differ
+    def pack(self, values) -> bytes:
+        return super().pack(*(values[key] for key in self.keys))
+
+    def unpack(self, buffer: bytes) -> Any:
+        return dict(zip(self.keys, super().unpack(buffer)))
+
+
+# -----------------------------------------------------------------------------
+class MappedCharacteristicProxyAdapter(PackedCharacteristicProxyAdapter):
+    '''
+    Adapter that packs/unpacks characteristic values according to a standard
+    Python `struct` format.
+    The adapted `read_value` and `write_value` methods return/accept a 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_proxy: CharacteristicProxy,
+        pack_format: str,
+        keys: Iterable[str],
+    ) -> None:
+        super().__init__(characteristic_proxy, pack_format)
+        self.keys = keys
+
+    # pylint: disable=arguments-differ
+    def pack(self, values) -> bytes:
+        return super().pack(*(values[key] for key in self.keys))
+
+    def unpack(self, buffer: bytes) -> Any:
+        return dict(zip(self.keys, super().unpack(buffer)))
+
+
+# -----------------------------------------------------------------------------
+class UTF8CharacteristicAdapter(CharacteristicAdapter[str]):
+    '''
+    Adapter that converts strings to/from bytes using UTF-8 encoding
+    '''
+
+    def encode_value(self, value: str) -> bytes:
+        return value.encode('utf-8')
+
+    def decode_value(self, value: bytes) -> str:
+        return value.decode('utf-8')
+
+
+# -----------------------------------------------------------------------------
+class UTF8CharacteristicProxyAdapter(CharacteristicProxyAdapter[str]):
+    '''
+    Adapter that converts strings to/from bytes using UTF-8 encoding
+    '''
+
+    def encode_value(self, value: str) -> bytes:
+        return value.encode('utf-8')
+
+    def decode_value(self, value: bytes) -> str:
+        return value.decode('utf-8')
+
+
+# -----------------------------------------------------------------------------
+class SerializableCharacteristicAdapter(CharacteristicAdapter[_T2]):
+    '''
+    Adapter that converts any class to/from bytes using the class'
+    `to_bytes` and `__bytes__` methods, respectively.
+    '''
+
+    def __init__(self, characteristic: Characteristic, cls: Type[_T2]) -> None:
+        super().__init__(characteristic)
+        self.cls = cls
+
+    def encode_value(self, value: _T2) -> bytes:
+        return bytes(value)
+
+    def decode_value(self, value: bytes) -> _T2:
+        return self.cls.from_bytes(value)
+
+
+# -----------------------------------------------------------------------------
+class SerializableCharacteristicProxyAdapter(CharacteristicProxyAdapter[_T2]):
+    '''
+    Adapter that converts any class to/from bytes using the class'
+    `to_bytes` and `__bytes__` methods, respectively.
+    '''
+
+    def __init__(
+        self, characteristic_proxy: CharacteristicProxy, cls: Type[_T2]
+    ) -> None:
+        super().__init__(characteristic_proxy)
+        self.cls = cls
+
+    def encode_value(self, value: _T2) -> bytes:
+        return bytes(value)
+
+    def decode_value(self, value: bytes) -> _T2:
+        return self.cls.from_bytes(value)
+
+
+# -----------------------------------------------------------------------------
+class EnumCharacteristicAdapter(CharacteristicAdapter[_T3]):
+    '''
+    Adapter that converts int-enum-like classes to/from bytes using the class'
+    `int().to_bytes()` and `from_bytes()` methods, respectively.
+    '''
+
+    def __init__(
+        self,
+        characteristic: Characteristic,
+        cls: Type[_T3],
+        length: int,
+        byteorder: Literal['little', 'big'] = 'little',
+    ):
+        """
+        Initialize an instance.
+
+        Params:
+          characteristic: the Characteristic to adapt to/from
+          cls: the class to/from which to convert integer values
+          length: number of bytes used to represent integer values
+          byteorder: byte order of the byte representation of integers.
+        """
+        super().__init__(characteristic)
+        self.cls = cls
+        self.length = length
+        self.byteorder = byteorder
+
+    def encode_value(self, value: _T3) -> bytes:
+        return int(value).to_bytes(self.length, self.byteorder)
+
+    def decode_value(self, value: bytes) -> _T3:
+        int_value = int.from_bytes(value, self.byteorder)
+        return self.cls(int_value)
+
+
+# -----------------------------------------------------------------------------
+class EnumCharacteristicProxyAdapter(CharacteristicProxyAdapter[_T3]):
+    '''
+    Adapter that converts int-enum-like classes to/from bytes using the class'
+    `int().to_bytes()` and `from_bytes()` methods, respectively.
+    '''
+
+    def __init__(
+        self,
+        characteristic_proxy: CharacteristicProxy,
+        cls: Type[_T3],
+        length: int,
+        byteorder: Literal['little', 'big'] = 'little',
+    ):
+        """
+        Initialize an instance.
+
+        Params:
+          characteristic_proxy: the CharacteristicProxy to adapt to/from
+          cls: the class to/from which to convert integer values
+          length: number of bytes used to represent integer values
+          byteorder: byte order of the byte representation of integers.
+        """
+        super().__init__(characteristic_proxy)
+        self.cls = cls
+        self.length = length
+        self.byteorder = byteorder
+
+    def encode_value(self, value: _T3) -> bytes:
+        return int(value).to_bytes(self.length, self.byteorder)
+
+    def decode_value(self, value: bytes) -> _T3:
+        int_value = int.from_bytes(value, self.byteorder)
+        a = self.cls(int_value)
+        return self.cls(int_value)

bumble/gatt_client.py

@@ -29,16 +29,18 @@ import logging
 import struct
 from datetime import datetime
 from typing import (
+    Any,
+    Callable,
+    Dict,
+    Generic,
+    Iterable,
     List,
     Optional,
-    Dict,
-    Tuple,
-    Callable,
-    Union,
-    Any,
-    Iterable,
-    Type,
     Set,
+    Tuple,
+    Union,
+    Type,
+    TypeVar,
     TYPE_CHECKING,
 )
 
@@ -82,9 +84,14 @@ from .gatt import (
     TemplateService,
 )
 
+# -----------------------------------------------------------------------------
+# Typing
+# -----------------------------------------------------------------------------
 if TYPE_CHECKING:
     from bumble.device import Connection
 
+_T = TypeVar('_T')
+
 # -----------------------------------------------------------------------------
 # Logging
 # -----------------------------------------------------------------------------
@@ -110,7 +117,7 @@ def show_services(services: Iterable[ServiceProxy]) -> None:
 # -----------------------------------------------------------------------------
 # Proxies
 # -----------------------------------------------------------------------------
-class AttributeProxy(EventEmitter):
+class AttributeProxy(EventEmitter, Generic[_T]):
     def __init__(
         self, client: Client, handle: int, end_group_handle: int, attribute_type: UUID
     ) -> None:
@@ -120,21 +127,21 @@ class AttributeProxy(EventEmitter):
         self.end_group_handle = end_group_handle
         self.type = attribute_type
 
-    async def read_value(self, no_long_read: bool = False) -> bytes:
+    async def read_value(self, no_long_read: bool = False) -> _T:
         return self.decode_value(
             await self.client.read_value(self.handle, no_long_read)
         )
 
-    async def write_value(self, value, with_response=False):
+    async def write_value(self, value: _T, with_response=False):
         return await self.client.write_value(
             self.handle, self.encode_value(value), with_response
         )
 
-    def encode_value(self, value: Any) -> bytes:
+    def encode_value(self, value: _T) -> bytes:
-        return value
+        return value  # type: ignore
 
-    def decode_value(self, value_bytes: bytes) -> Any:
+    def decode_value(self, value: bytes) -> _T:
-        return value_bytes
+        return value  # type: ignore
 
     def __str__(self) -> str:
         return f'Attribute(handle=0x{self.handle:04X}, type={self.type})'
@@ -184,19 +191,19 @@ class ServiceProxy(AttributeProxy):
         return f'Service(handle=0x{self.handle:04X}, uuid={self.uuid})'
 
 
-class CharacteristicProxy(AttributeProxy):
+class CharacteristicProxy(AttributeProxy[_T]):
     properties: Characteristic.Properties
     descriptors: List[DescriptorProxy]
-    subscribers: Dict[Any, Callable[[bytes], Any]]
+    subscribers: Dict[Any, Callable[[_T], Any]]
 
     def __init__(
         self,
-        client,
+        client: Client,
-        handle,
+        handle: int,
-        end_group_handle,
+        end_group_handle: int,
-        uuid,
+        uuid: UUID,
         properties: int,
-    ):
+    ) -> None:
         super().__init__(client, handle, end_group_handle, uuid)
         self.uuid = uuid
         self.properties = Characteristic.Properties(properties)
@@ -204,21 +211,21 @@ class CharacteristicProxy(AttributeProxy):
         self.descriptors_discovered = False
         self.subscribers = {}  # Map from subscriber to proxy subscriber
 
-    def get_descriptor(self, descriptor_type):
+    def get_descriptor(self, descriptor_type: UUID) -> Optional[DescriptorProxy]:
         for descriptor in self.descriptors:
             if descriptor.type == descriptor_type:
                 return descriptor
 
         return None
 
-    async def discover_descriptors(self):
+    async def discover_descriptors(self) -> list[DescriptorProxy]:
         return await self.client.discover_descriptors(self)
 
     async def subscribe(
         self,
-        subscriber: Optional[Callable[[bytes], Any]] = None,
+        subscriber: Optional[Callable[[_T], Any]] = None,
         prefer_notify: bool = True,
-    ):
+    ) -> None:
         if subscriber is not None:
             if subscriber in self.subscribers:
                 # We already have a proxy subscriber
@@ -233,13 +240,13 @@ class CharacteristicProxy(AttributeProxy):
                 self.subscribers[subscriber] = on_change
                 subscriber = on_change
 
-        return await self.client.subscribe(self, subscriber, prefer_notify)
+        await self.client.subscribe(self, subscriber, prefer_notify)
 
-    async def unsubscribe(self, subscriber=None, force=False):
+    async def unsubscribe(self, subscriber=None, force=False) -> None:
         if subscriber in self.subscribers:
             subscriber = self.subscribers.pop(subscriber)
 
-        return await self.client.unsubscribe(self, subscriber, force)
+        await self.client.unsubscribe(self, subscriber, force)
 
     def __str__(self) -> str:
         return (
@@ -250,7 +257,7 @@ class CharacteristicProxy(AttributeProxy):
 
 
 class DescriptorProxy(AttributeProxy):
-    def __init__(self, client, handle, descriptor_type):
+    def __init__(self, client: Client, handle: int, descriptor_type: UUID) -> None:
         super().__init__(client, handle, 0, descriptor_type)
 
     def __str__(self) -> str:
@@ -679,7 +686,7 @@ class Client:
 
                     properties, handle = struct.unpack_from('<BH', attribute_value)
                     characteristic_uuid = UUID.from_bytes(attribute_value[3:])
-                    characteristic = CharacteristicProxy(
+                    characteristic: CharacteristicProxy = CharacteristicProxy(
                         self, handle, 0, characteristic_uuid, properties
                     )
 
@@ -805,7 +812,7 @@ class Client:
                     logger.warning(f'bogus handle value: {attribute_handle}')
                     return []
 
-                attribute = AttributeProxy(
+                attribute: AttributeProxy = AttributeProxy(
                     self, attribute_handle, 0, UUID.from_bytes(attribute_uuid)
                 )
                 attributes.append(attribute)
@@ -818,7 +825,7 @@ class Client:
     async def subscribe(
         self,
         characteristic: CharacteristicProxy,
-        subscriber: Optional[Callable[[bytes], Any]] = None,
+        subscriber: Optional[Callable[[Any], Any]] = None,
         prefer_notify: bool = True,
     ) -> None:
         # If we haven't already discovered the descriptors for this characteristic,
@@ -868,7 +875,7 @@ class Client:
     async def unsubscribe(
         self,
         characteristic: CharacteristicProxy,
-        subscriber: Optional[Callable[[bytes], Any]] = None,
+        subscriber: Optional[Callable[[Any], Any]] = None,
         force: bool = False,
     ) -> None:
         '''

bumble/gatt_server.py

@@ -36,7 +36,6 @@ from typing import (
     Tuple,
     TypeVar,
     Type,
-    Union,
     TYPE_CHECKING,
 )
 from pyee import EventEmitter
@@ -78,7 +77,6 @@ from bumble.gatt import (
     GATT_REQUEST_TIMEOUT,
     GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE,
     Characteristic,
-    CharacteristicAdapter,
     CharacteristicDeclaration,
     CharacteristicValue,
     IncludedServiceDeclaration,
@@ -469,7 +467,7 @@ class Server(EventEmitter):
             finally:
                 self.pending_confirmations[connection.handle] = None
 
-    async def notify_or_indicate_subscribers(
+    async def _notify_or_indicate_subscribers(
         self,
         indicate: bool,
         attribute: Attribute,
@@ -503,7 +501,9 @@ class Server(EventEmitter):
         value: Optional[bytes] = None,
         force: bool = False,
     ):
-        return await self.notify_or_indicate_subscribers(False, attribute, value, force)
+        return await self._notify_or_indicate_subscribers(
+            False, attribute, value, force
+        )
 
     async def indicate_subscribers(
         self,
@@ -511,7 +511,7 @@ class Server(EventEmitter):
         value: Optional[bytes] = None,
         force: bool = False,
     ):
-        return await self.notify_or_indicate_subscribers(True, attribute, value, force)
+        return await self._notify_or_indicate_subscribers(True, attribute, value, force)
 
     def on_disconnection(self, connection: Connection) -> None:
         if connection.handle in self.subscribers:

bumble/hci.py

@@ -24,6 +24,7 @@ import logging
 import secrets
 import struct
 from typing import Any, Callable, Dict, Iterable, List, Optional, Type, Union, ClassVar
+from typing_extensions import Self
 
 from bumble import crypto
 from bumble.colors import color
@@ -34,6 +35,7 @@ from bumble.core import (
     InvalidArgumentError,
     InvalidPacketError,
     ProtocolError,
+    PhysicalTransport,
     bit_flags_to_strings,
     name_or_number,
     padded_bytes,
@@ -94,7 +96,7 @@ def map_class_of_device(class_of_device):
     )
 
 
-def phy_list_to_bits(phys: Optional[Iterable[int]]) -> int:
+def phy_list_to_bits(phys: Optional[Iterable[Phy]]) -> int:
     if phys is None:
         return 0
 
@@ -700,30 +702,22 @@ HCI_ERROR_NAMES[HCI_SUCCESS] = 'HCI_SUCCESS'
 HCI_COMMAND_STATUS_PENDING = 0
 
 
+class Phy(enum.IntEnum):
+    LE_1M    = 1
+    LE_2M    = 2
+    LE_CODED = 3
+
+
 # ACL
 HCI_ACL_PB_FIRST_NON_FLUSHABLE = 0
 HCI_ACL_PB_CONTINUATION        = 1
 HCI_ACL_PB_FIRST_FLUSHABLE     = 2
 HCI_ACK_PB_COMPLETE_L2CAP      = 3
 
-# Roles
+HCI_LE_PHY_NAMES: dict[int,str] = {
-HCI_CENTRAL_ROLE    = 0
+    Phy.LE_1M:    'LE 1M',
-HCI_PERIPHERAL_ROLE = 1
+    Phy.LE_2M:    'LE 2M',
-
+    Phy.LE_CODED: 'LE Coded'
-HCI_ROLE_NAMES = {
-    HCI_CENTRAL_ROLE:    'CENTRAL',
-    HCI_PERIPHERAL_ROLE: 'PERIPHERAL'
-}
-
-# LE PHY Types
-HCI_LE_1M_PHY    = 1
-HCI_LE_2M_PHY    = 2
-HCI_LE_CODED_PHY = 3
-
-HCI_LE_PHY_NAMES = {
-    HCI_LE_1M_PHY:    'LE 1M',
-    HCI_LE_2M_PHY:    'LE 2M',
-    HCI_LE_CODED_PHY: 'LE Coded'
 }
 
 HCI_LE_1M_PHY_BIT    = 0
@@ -732,19 +726,13 @@ HCI_LE_CODED_PHY_BIT = 2
 
 HCI_LE_PHY_BIT_NAMES = ['LE_1M_PHY', 'LE_2M_PHY', 'LE_CODED_PHY']
 
-HCI_LE_PHY_TYPE_TO_BIT = {
+HCI_LE_PHY_TYPE_TO_BIT: dict[Phy, int] = {
-    HCI_LE_1M_PHY:    HCI_LE_1M_PHY_BIT,
+    Phy.LE_1M:    HCI_LE_1M_PHY_BIT,
-    HCI_LE_2M_PHY:    HCI_LE_2M_PHY_BIT,
+    Phy.LE_2M:    HCI_LE_2M_PHY_BIT,
-    HCI_LE_CODED_PHY: HCI_LE_CODED_PHY_BIT
+    Phy.LE_CODED: HCI_LE_CODED_PHY_BIT,
 }
 
 
-class Phy(enum.IntEnum):
-    LE_1M    = HCI_LE_1M_PHY
-    LE_2M    = HCI_LE_2M_PHY
-    LE_CODED = HCI_LE_CODED_PHY
-
-
 class PhyBit(enum.IntFlag):
     LE_1M    = 1 << HCI_LE_1M_PHY_BIT
     LE_2M    = 1 << HCI_LE_2M_PHY_BIT
@@ -811,6 +799,19 @@ class CsSubeventAbortReason(OpenIntEnum):
     SCHEDULING_CONFLICT_OR_LIMITED_RESOURCES = 0x03
     UNSPECIFIED = 0x0F
 
+class Role(enum.IntEnum):
+    CENTRAL    = 0
+    PERIPHERAL = 1
+
+# For Backward Compatibility.
+HCI_CENTRAL_ROLE    = Role.CENTRAL
+HCI_PERIPHERAL_ROLE = Role.PERIPHERAL
+
+
+HCI_LE_1M_PHY    = Phy.LE_1M
+HCI_LE_2M_PHY    = Phy.LE_2M
+HCI_LE_CODED_PHY = Phy.LE_CODED
+
 
 # Connection Parameters
 HCI_CONNECTION_INTERVAL_MS_PER_UNIT = 1.25
@@ -889,10 +890,15 @@ HCI_LINK_TYPE_NAMES = {
 }
 
 # Address types
-HCI_PUBLIC_DEVICE_ADDRESS_TYPE   = 0x00
+class AddressType(OpenIntEnum):
-HCI_RANDOM_DEVICE_ADDRESS_TYPE   = 0x01
+    PUBLIC_DEVICE   = 0x00
-HCI_PUBLIC_IDENTITY_ADDRESS_TYPE = 0x02
+    RANDOM_DEVICE   = 0x01
-HCI_RANDOM_IDENTITY_ADDRESS_TYPE = 0x03
+    PUBLIC_IDENTITY = 0x02
+    RANDOM_IDENTITY = 0x03
+    # (Directed Only) Address is RPA, but controller cannot resolve.
+    UNABLE_TO_RESOLVE = 0xFE
+    # (Extended Only) No address.
+    ANONYMOUS       = 0xFF
 
 # Supported Commands Masks
 # See Bluetooth spec @ 6.27 SUPPORTED COMMANDS
@@ -1582,8 +1588,8 @@ class HCI_Constant:
         return HCI_ERROR_NAMES.get(status, f'0x{status:02X}')
 
     @staticmethod
-    def role_name(role):
+    def role_name(role: int) -> str:
-        return HCI_ROLE_NAMES.get(role, str(role))
+        return Role(role).name
 
     @staticmethod
     def le_phy_name(phy):
@@ -1949,17 +1955,10 @@ class Address:
     address[0] is the LSB of the address, address[5] is the MSB.
     '''
 
-    PUBLIC_DEVICE_ADDRESS = 0x00
+    PUBLIC_DEVICE_ADDRESS = AddressType.PUBLIC_DEVICE
-    RANDOM_DEVICE_ADDRESS = 0x01
+    RANDOM_DEVICE_ADDRESS = AddressType.RANDOM_DEVICE
-    PUBLIC_IDENTITY_ADDRESS = 0x02
+    PUBLIC_IDENTITY_ADDRESS = AddressType.PUBLIC_IDENTITY
-    RANDOM_IDENTITY_ADDRESS = 0x03
+    RANDOM_IDENTITY_ADDRESS = AddressType.RANDOM_IDENTITY
-
-    ADDRESS_TYPE_NAMES = {
-        PUBLIC_DEVICE_ADDRESS: 'PUBLIC_DEVICE_ADDRESS',
-        RANDOM_DEVICE_ADDRESS: 'RANDOM_DEVICE_ADDRESS',
-        PUBLIC_IDENTITY_ADDRESS: 'PUBLIC_IDENTITY_ADDRESS',
-        RANDOM_IDENTITY_ADDRESS: 'RANDOM_IDENTITY_ADDRESS',
-    }
 
     # Type declarations
     NIL: Address
@@ -1969,40 +1968,44 @@ class Address:
     # pylint: disable-next=unnecessary-lambda
     ADDRESS_TYPE_SPEC = {'size': 1, 'mapper': lambda x: Address.address_type_name(x)}
 
-    @staticmethod
+    @classmethod
-    def address_type_name(address_type):
+    def address_type_name(cls: type[Self], address_type: int) -> str:
-        return name_or_number(Address.ADDRESS_TYPE_NAMES, address_type)
+        return AddressType(address_type).name
 
-    @staticmethod
+    @classmethod
-    def from_string_for_transport(string, transport):
+    def from_string_for_transport(
+        cls: type[Self], string: str, transport: PhysicalTransport
+    ) -> Self:
         if transport == BT_BR_EDR_TRANSPORT:
             address_type = Address.PUBLIC_DEVICE_ADDRESS
         else:
             address_type = Address.RANDOM_DEVICE_ADDRESS
-        return Address(string, address_type)
+        return cls(string, address_type)
 
-    @staticmethod
+    @classmethod
-    def parse_address(data, offset):
+    def parse_address(cls: type[Self], data: bytes, offset: int) -> tuple[int, Self]:
         # Fix the type to a default value. This is used for parsing type-less Classic
         # addresses
-        return Address.parse_address_with_type(
+        return cls.parse_address_with_type(data, offset, Address.PUBLIC_DEVICE_ADDRESS)
-            data, offset, Address.PUBLIC_DEVICE_ADDRESS
-        )
 
-    @staticmethod
+    @classmethod
-    def parse_random_address(data, offset):
+    def parse_random_address(
-        return Address.parse_address_with_type(
+        cls: type[Self], data: bytes, offset: int
-            data, offset, Address.RANDOM_DEVICE_ADDRESS
+    ) -> tuple[int, Self]:
-        )
+        return cls.parse_address_with_type(data, offset, Address.RANDOM_DEVICE_ADDRESS)
 
-    @staticmethod
+    @classmethod
-    def parse_address_with_type(data, offset, address_type):
+    def parse_address_with_type(
-        return offset + 6, Address(data[offset : offset + 6], address_type)
+        cls: type[Self], data: bytes, offset: int, address_type: AddressType
+    ) -> tuple[int, Self]:
+        return offset + 6, cls(data[offset : offset + 6], address_type)
 
-    @staticmethod
+    @classmethod
-    def parse_address_preceded_by_type(data, offset):
+    def parse_address_preceded_by_type(
-        address_type = data[offset - 1]
+        cls: type[Self], data: bytes, offset: int
-        return Address.parse_address_with_type(data, offset, address_type)
+    ) -> tuple[int, Self]:
+        address_type = AddressType(data[offset - 1])
+        return cls.parse_address_with_type(data, offset, address_type)
 
     @classmethod
     def generate_static_address(cls) -> Address:
@@ -2042,8 +2045,10 @@ class Address:
         )
 
     def __init__(
-        self, address: Union[bytes, str], address_type: int = RANDOM_DEVICE_ADDRESS
+        self,
-    ):
+        address: Union[bytes, str],
+        address_type: AddressType = RANDOM_DEVICE_ADDRESS,
+    ) -> None:
         '''
         Initialize an instance. `address` may be a byte array in little-endian
         format, or a hex string in big-endian format (with optional ':'

bumble/host.py

@@ -44,6 +44,7 @@ from bumble import hci
 from bumble.core import (
     BT_BR_EDR_TRANSPORT,
     BT_LE_TRANSPORT,
+    PhysicalTransport,
     ConnectionPHY,
     ConnectionParameters,
 )
@@ -186,7 +187,11 @@ class DataPacketQueue(pyee.EventEmitter):
 # -----------------------------------------------------------------------------
 class Connection:
     def __init__(
-        self, host: Host, handle: int, peer_address: hci.Address, transport: int
+        self,
+        host: Host,
+        handle: int,
+        peer_address: hci.Address,
+        transport: PhysicalTransport,
     ):
         self.host = host
         self.handle = handle
@@ -235,7 +240,7 @@ class Host(AbortableEventEmitter):
     cis_links: Dict[int, IsoLink]
     bis_links: Dict[int, IsoLink]
     sco_links: Dict[int, ScoLink]
-    bigs: dict[int, set[int]] = {}  # BIG Handle to BIS Handles
+    bigs: dict[int, set[int]]
     acl_packet_queue: Optional[DataPacketQueue] = None
     le_acl_packet_queue: Optional[DataPacketQueue] = None
     iso_packet_queue: Optional[DataPacketQueue] = None
@@ -259,6 +264,7 @@ class Host(AbortableEventEmitter):
         self.cis_links = {}  # CIS links, by connection handle
         self.bis_links = {}  # BIS links, by connection handle
         self.sco_links = {}  # SCO links, by connection handle
+        self.bigs = {}  # BIG Handle to BIS Handles
         self.pending_command = None
         self.pending_response: Optional[asyncio.Future[Any]] = None
         self.number_of_supported_advertising_sets = 0
@@ -978,7 +984,7 @@ class Host(AbortableEventEmitter):
                 event.peer_address,
                 getattr(event, 'local_resolvable_private_address', None),
                 getattr(event, 'peer_resolvable_private_address', None),
-                event.role,
+                hci.Role(event.role),
                 connection_parameters,
             )
         else:
@@ -1061,8 +1067,10 @@ class Host(AbortableEventEmitter):
             )
 
             # Flush the data queues
-            self.acl_packet_queue.flush(handle)
+            if self.acl_packet_queue:
-            self.le_acl_packet_queue.flush(handle)
+                self.acl_packet_queue.flush(handle)
+            if self.le_acl_packet_queue:
+                self.le_acl_packet_queue.flush(handle)
             if self.iso_packet_queue:
                 self.iso_packet_queue.flush(handle)
         else:
@@ -1334,7 +1342,7 @@ class Host(AbortableEventEmitter):
                 f'role change for {event.bd_addr}: '
                 f'{hci.HCI_Constant.role_name(event.new_role)}'
             )
-            self.emit('role_change', event.bd_addr, event.new_role)
+            self.emit('role_change', event.bd_addr, hci.Role(event.new_role))
         else:
             logger.debug(
                 f'role change for {event.bd_addr} failed: '

bumble/l2cap.py

@@ -42,7 +42,6 @@ from typing import (
 from .utils import deprecated
 from .colors import color
 from .core import (
-    BT_CENTRAL_ROLE,
     InvalidStateError,
     InvalidArgumentError,
     InvalidPacketError,
@@ -52,6 +51,7 @@ from .core import (
 from .hci import (
     HCI_LE_Connection_Update_Command,
     HCI_Object,
+    Role,
     key_with_value,
     name_or_number,
 )
@@ -1908,7 +1908,7 @@ class ChannelManager:
     def on_l2cap_connection_parameter_update_request(
         self, connection: Connection, cid: int, request
     ):
-        if connection.role == BT_CENTRAL_ROLE:
+        if connection.role == Role.CENTRAL:
             self.send_control_frame(
                 connection,
                 cid,

bumble/link.py

@@ -20,7 +20,6 @@ import asyncio
 from functools import partial
 
 from bumble.core import (
-    BT_PERIPHERAL_ROLE,
     BT_BR_EDR_TRANSPORT,
     BT_LE_TRANSPORT,
     InvalidStateError,
@@ -28,6 +27,7 @@ from bumble.core import (
 from bumble.colors import color
 from bumble.hci import (
     Address,
+    Role,
     HCI_SUCCESS,
     HCI_CONNECTION_ACCEPT_TIMEOUT_ERROR,
     HCI_CONNECTION_TIMEOUT_ERROR,
@@ -292,7 +292,7 @@ class LocalLink:
             return
 
         async def task():
-            if responder_role != BT_PERIPHERAL_ROLE:
+            if responder_role != Role.PERIPHERAL:
                 initiator_controller.on_classic_role_change(
                     responder_controller.public_address, int(not (responder_role))
                 )

bumble/pairing.py

@@ -1,4 +1,4 @@
-# Copyright 2021-2023 Google LLC
+# 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.
@@ -76,18 +76,18 @@ class OobData:
         return instance
 
     def to_ad(self) -> AdvertisingData:
-        ad_structures = []
+        ad_structures: list[tuple[int, bytes]] = []
         if self.address is not None:
             ad_structures.append(
-                (AdvertisingData.LE_BLUETOOTH_DEVICE_ADDRESS, bytes(self.address))
+                (AdvertisingData.Type.LE_BLUETOOTH_DEVICE_ADDRESS, bytes(self.address))
             )
         if self.role is not None:
-            ad_structures.append((AdvertisingData.LE_ROLE, bytes([self.role])))
+            ad_structures.append((AdvertisingData.Type.LE_ROLE, bytes([self.role])))
         if self.shared_data is not None:
             ad_structures.extend(self.shared_data.to_ad().ad_structures)
         if self.legacy_context is not None:
             ad_structures.append(
-                (AdvertisingData.SECURITY_MANAGER_TK_VALUE, self.legacy_context.tk)
+                (AdvertisingData.Type.SECURITY_MANAGER_TK_VALUE, self.legacy_context.tk)
             )
 
         return AdvertisingData(ad_structures)

bumble/pandora/host.py

@@ -25,7 +25,6 @@ from .config import Config
 from bumble.core import (
     BT_BR_EDR_TRANSPORT,
     BT_LE_TRANSPORT,
-    BT_PERIPHERAL_ROLE,
     UUID,
     AdvertisingData,
     Appearance,
@@ -47,6 +46,8 @@ from bumble.hci import (
     HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR,
     Address,
     Phy,
+    Role,
+    OwnAddressType,
 )
 from google.protobuf import any_pb2  # pytype: disable=pyi-error
 from google.protobuf import empty_pb2  # pytype: disable=pyi-error
@@ -114,11 +115,11 @@ SECONDARY_PHY_TO_BUMBLE_PHY_MAP: Dict[SecondaryPhy, Phy] = {
     SECONDARY_CODED: Phy.LE_CODED,
 }
 
-OWN_ADDRESS_MAP: Dict[host_pb2.OwnAddressType, bumble.hci.OwnAddressType] = {
+OWN_ADDRESS_MAP: Dict[host_pb2.OwnAddressType, OwnAddressType] = {
-    host_pb2.PUBLIC: bumble.hci.OwnAddressType.PUBLIC,
+    host_pb2.PUBLIC: OwnAddressType.PUBLIC,
-    host_pb2.RANDOM: bumble.hci.OwnAddressType.RANDOM,
+    host_pb2.RANDOM: OwnAddressType.RANDOM,
-    host_pb2.RESOLVABLE_OR_PUBLIC: bumble.hci.OwnAddressType.RESOLVABLE_OR_PUBLIC,
+    host_pb2.RESOLVABLE_OR_PUBLIC: OwnAddressType.RESOLVABLE_OR_PUBLIC,
-    host_pb2.RESOLVABLE_OR_RANDOM: bumble.hci.OwnAddressType.RESOLVABLE_OR_RANDOM,
+    host_pb2.RESOLVABLE_OR_RANDOM: OwnAddressType.RESOLVABLE_OR_RANDOM,
 }
 
 
@@ -250,7 +251,7 @@ class HostService(HostServicer):
             connection = await self.device.connect(
                 address,
                 transport=BT_LE_TRANSPORT,
-                own_address_type=request.own_address_type,
+                own_address_type=OwnAddressType(request.own_address_type),
             )
         except ConnectionError as e:
             if e.error_code == HCI_PAGE_TIMEOUT_ERROR:
@@ -378,7 +379,7 @@ class HostService(HostServicer):
             def on_connection(connection: bumble.device.Connection) -> None:
                 if (
                     connection.transport == BT_LE_TRANSPORT
-                    and connection.role == BT_PERIPHERAL_ROLE
+                    and connection.role == Role.PERIPHERAL
                 ):
                     connections.put_nowait(connection)
 
@@ -496,7 +497,7 @@ class HostService(HostServicer):
             def on_connection(connection: bumble.device.Connection) -> None:
                 if (
                     connection.transport == BT_LE_TRANSPORT
-                    and connection.role == BT_PERIPHERAL_ROLE
+                    and connection.role == Role.PERIPHERAL
                 ):
                     connections.put_nowait(connection)
 
@@ -509,7 +510,7 @@ class HostService(HostServicer):
                     await self.device.start_advertising(
                         target=target,
                         advertising_type=advertising_type,
-                        own_address_type=request.own_address_type,
+                        own_address_type=OwnAddressType(request.own_address_type),
                     )
 
                 if not request.connectable:
@@ -558,7 +559,7 @@ class HostService(HostServicer):
         await self.device.start_scanning(
             legacy=request.legacy,
             active=not request.passive,
-            own_address_type=request.own_address_type,
+            own_address_type=OwnAddressType(request.own_address_type),
             scan_interval=(
                 int(request.interval)
                 if request.interval

bumble/pandora/security.py

@@ -24,11 +24,10 @@ from bumble import hci
 from bumble.core import (
     BT_BR_EDR_TRANSPORT,
     BT_LE_TRANSPORT,
-    BT_PERIPHERAL_ROLE,
     ProtocolError,
 )
 from bumble.device import Connection as BumbleConnection, Device
-from bumble.hci import HCI_Error
+from bumble.hci import HCI_Error, Role
 from bumble.utils import EventWatcher
 from bumble.pairing import PairingConfig, PairingDelegate as BasePairingDelegate
 from google.protobuf import any_pb2  # pytype: disable=pyi-error
@@ -318,7 +317,7 @@ class SecurityService(SecurityServicer):
 
                     if (
                         connection.transport == BT_LE_TRANSPORT
-                        and connection.role == BT_PERIPHERAL_ROLE
+                        and connection.role == Role.PERIPHERAL
                     ):
                         connection.request_pairing()
                     else:

bumble/pandora/utils.py

@@ -20,11 +20,11 @@ import inspect
 import logging
 
 from bumble.device import Device
-from bumble.hci import Address
+from bumble.hci import Address, AddressType
 from google.protobuf.message import Message  # pytype: disable=pyi-error
 from typing import Any, Dict, Generator, MutableMapping, Optional, Tuple
 
-ADDRESS_TYPES: Dict[str, int] = {
+ADDRESS_TYPES: Dict[str, AddressType] = {
     "public": Address.PUBLIC_DEVICE_ADDRESS,
     "random": Address.RANDOM_DEVICE_ADDRESS,
     "public_identity": Address.PUBLIC_IDENTITY_ADDRESS,

bumble/profiles/aics.py

@@ -24,16 +24,13 @@ import struct
 from dataclasses import dataclass
 from typing import Optional
 
-from bumble import gatt
 from bumble.device import Connection
 from bumble.att import ATT_Error
 from bumble.gatt import (
+    Attribute,
     Characteristic,
-    SerializableCharacteristicAdapter,
-    PackedCharacteristicAdapter,
     TemplateService,
     CharacteristicValue,
-    UTF8CharacteristicAdapter,
     GATT_AUDIO_INPUT_CONTROL_SERVICE,
     GATT_AUDIO_INPUT_STATE_CHARACTERISTIC,
     GATT_GAIN_SETTINGS_ATTRIBUTE_CHARACTERISTIC,
@@ -42,6 +39,14 @@ from bumble.gatt import (
     GATT_AUDIO_INPUT_CONTROL_POINT_CHARACTERISTIC,
     GATT_AUDIO_INPUT_DESCRIPTION_CHARACTERISTIC,
 )
+from bumble.gatt_adapters import (
+    CharacteristicProxy,
+    PackedCharacteristicProxyAdapter,
+    SerializableCharacteristicAdapter,
+    SerializableCharacteristicProxyAdapter,
+    UTF8CharacteristicAdapter,
+    UTF8CharacteristicProxyAdapter,
+)
 from bumble.gatt_client import ProfileServiceProxy, ServiceProxy
 from bumble.utils import OpenIntEnum
 
@@ -124,7 +129,7 @@ class AudioInputState:
     mute: Mute = Mute.NOT_MUTED
     gain_mode: GainMode = GainMode.MANUAL
     change_counter: int = 0
-    attribute_value: Optional[CharacteristicValue] = None
+    attribute: Optional[Attribute] = None
 
     def __bytes__(self) -> bytes:
         return bytes(
@@ -151,10 +156,8 @@ class AudioInputState:
         self.change_counter = (self.change_counter + 1) % (CHANGE_COUNTER_MAX_VALUE + 1)
 
     async def notify_subscribers_via_connection(self, connection: Connection) -> None:
-        assert self.attribute_value is not None
+        assert self.attribute is not None
-        await connection.device.notify_subscribers(
+        await connection.device.notify_subscribers(attribute=self.attribute)
-            attribute=self.attribute_value, value=bytes(self)
-        )
 
 
 @dataclass
@@ -315,24 +318,28 @@ class AudioInputDescription:
     '''
 
     audio_input_description: str = "Bluetooth"
-    attribute_value: Optional[CharacteristicValue] = None
+    attribute: Optional[Attribute] = None
 
     def on_read(self, _connection: Optional[Connection]) -> str:
         return self.audio_input_description
 
     async def on_write(self, connection: Optional[Connection], value: str) -> None:
         assert connection
-        assert self.attribute_value
+        assert self.attribute
 
         self.audio_input_description = value
-        await connection.device.notify_subscribers(
+        await connection.device.notify_subscribers(attribute=self.attribute)
-            attribute=self.attribute_value, value=value
-        )
 
 
 class AICSService(TemplateService):
     UUID = GATT_AUDIO_INPUT_CONTROL_SERVICE
 
+    audio_input_state_characteristic: Characteristic[AudioInputState]
+    audio_input_type_characteristic: Characteristic[bytes]
+    audio_input_status_characteristic: Characteristic[bytes]
+    audio_input_control_point_characteristic: Characteristic[bytes]
+    gain_settings_properties_characteristic: Characteristic[GainSettingsProperties]
+
     def __init__(
         self,
         audio_input_state: Optional[AudioInputState] = None,
@@ -374,9 +381,7 @@ class AICSService(TemplateService):
             ),
             AudioInputState,
         )
-        self.audio_input_state.attribute_value = (
+        self.audio_input_state.attribute = self.audio_input_state_characteristic
-            self.audio_input_state_characteristic.value
-        )
 
         self.gain_settings_properties_characteristic = (
             SerializableCharacteristicAdapter(
@@ -425,8 +430,8 @@ class AICSService(TemplateService):
                 ),
             )
         )
-        self.audio_input_description.attribute_value = (
+        self.audio_input_description.attribute = (
-            self.audio_input_control_point_characteristic.value
+            self.audio_input_control_point_characteristic
         )
 
         super().__init__(
@@ -448,24 +453,29 @@ class AICSService(TemplateService):
 class AICSServiceProxy(ProfileServiceProxy):
     SERVICE_CLASS = AICSService
 
+    audio_input_state: CharacteristicProxy[AudioInputState]
+    gain_settings_properties: CharacteristicProxy[GainSettingsProperties]
+    audio_input_status: CharacteristicProxy[int]
+    audio_input_control_point: CharacteristicProxy[bytes]
+
     def __init__(self, service_proxy: ServiceProxy) -> None:
         self.service_proxy = service_proxy
 
-        self.audio_input_state = SerializableCharacteristicAdapter(
+        self.audio_input_state = SerializableCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_AUDIO_INPUT_STATE_CHARACTERISTIC
             ),
             AudioInputState,
         )
 
-        self.gain_settings_properties = SerializableCharacteristicAdapter(
+        self.gain_settings_properties = SerializableCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_GAIN_SETTINGS_ATTRIBUTE_CHARACTERISTIC
             ),
             GainSettingsProperties,
         )
 
-        self.audio_input_status = PackedCharacteristicAdapter(
+        self.audio_input_status = PackedCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_AUDIO_INPUT_STATUS_CHARACTERISTIC
             ),
@@ -478,7 +488,7 @@ class AICSServiceProxy(ProfileServiceProxy):
             )
         )
 
-        self.audio_input_description = UTF8CharacteristicAdapter(
+        self.audio_input_description = UTF8CharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_AUDIO_INPUT_DESCRIPTION_CHARACTERISTIC
             )

bumble/profiles/ascs.py

@@ -301,7 +301,7 @@ class AseStateMachine(gatt.Characteristic):
     presentation_delay = 0
 
     # Additional parameters in ENABLING, STREAMING, DISABLING State
-    metadata = le_audio.Metadata()
+    metadata: le_audio.Metadata
 
     def __init__(
         self,
@@ -313,6 +313,7 @@ class AseStateMachine(gatt.Characteristic):
         self.ase_id = ase_id
         self._state = AseStateMachine.State.IDLE
         self.role = role
+        self.metadata = le_audio.Metadata()
 
         uuid = (
             gatt.GATT_SINK_ASE_CHARACTERISTIC

bumble/profiles/asha.py

@@ -134,12 +134,14 @@ class AshaService(gatt.TemplateService):
             ),
         )
 
-        self.audio_control_point_characteristic = gatt.Characteristic(
+        self.audio_control_point_characteristic: gatt.Characteristic[bytes] = (
-            gatt.GATT_ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC,
+            gatt.Characteristic(
-            gatt.Characteristic.Properties.WRITE
+                gatt.GATT_ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC,
-            | gatt.Characteristic.Properties.WRITE_WITHOUT_RESPONSE,
+                gatt.Characteristic.Properties.WRITE
-            gatt.Characteristic.WRITEABLE,
+                | gatt.Characteristic.Properties.WRITE_WITHOUT_RESPONSE,
-            gatt.CharacteristicValue(write=self._on_audio_control_point_write),
+                gatt.Characteristic.WRITEABLE,
+                gatt.CharacteristicValue(write=self._on_audio_control_point_write),
+            )
         )
         self.audio_status_characteristic = gatt.Characteristic(
             gatt.GATT_ASHA_AUDIO_STATUS_CHARACTERISTIC,
@@ -147,7 +149,7 @@ class AshaService(gatt.TemplateService):
             gatt.Characteristic.READABLE,
             bytes([AudioStatus.OK]),
         )
-        self.volume_characteristic = gatt.Characteristic(
+        self.volume_characteristic: gatt.Characteristic[bytes] = gatt.Characteristic(
             gatt.GATT_ASHA_VOLUME_CHARACTERISTIC,
             gatt.Characteristic.Properties.WRITE_WITHOUT_RESPONSE,
             gatt.Characteristic.WRITEABLE,
@@ -166,13 +168,13 @@ class AshaService(gatt.TemplateService):
             struct.pack('<H', self.psm),
         )
 
-        characteristics = [
+        characteristics = (
             self.read_only_properties_characteristic,
             self.audio_control_point_characteristic,
             self.audio_status_characteristic,
             self.volume_characteristic,
             self.le_psm_out_characteristic,
-        ]
+        )
 
         super().__init__(characteristics)
 

bumble/profiles/bass.py

@@ -20,11 +20,12 @@ from __future__ import annotations
 import dataclasses
 import logging
 import struct
-from typing import ClassVar, List, Optional, Sequence
+from typing import ClassVar, Optional, Sequence
 
 from bumble import core
 from bumble import device
 from bumble import gatt
+from bumble import gatt_adapters
 from bumble import gatt_client
 from bumble import hci
 from bumble import utils
@@ -52,7 +53,7 @@ def encode_subgroups(subgroups: Sequence[SubgroupInfo]) -> bytes:
     )
 
 
-def decode_subgroups(data: bytes) -> List[SubgroupInfo]:
+def decode_subgroups(data: bytes) -> list[SubgroupInfo]:
     num_subgroups = data[0]
     offset = 1
     subgroups = []
@@ -273,7 +274,7 @@ class BroadcastReceiveState:
     pa_sync_state: PeriodicAdvertisingSyncState
     big_encryption: BigEncryption
     bad_code: bytes
-    subgroups: List[SubgroupInfo]
+    subgroups: list[SubgroupInfo]
 
     @classmethod
     def from_bytes(cls, data: bytes) -> BroadcastReceiveState:
@@ -354,7 +355,9 @@ class BroadcastAudioScanServiceProxy(gatt_client.ProfileServiceProxy):
     SERVICE_CLASS = BroadcastAudioScanService
 
     broadcast_audio_scan_control_point: gatt_client.CharacteristicProxy
-    broadcast_receive_states: List[gatt.DelegatedCharacteristicAdapter]
+    broadcast_receive_states: list[
+        gatt_client.CharacteristicProxy[Optional[BroadcastReceiveState]]
+    ]
 
     def __init__(self, service_proxy: gatt_client.ServiceProxy):
         self.service_proxy = service_proxy
@@ -366,7 +369,7 @@ class BroadcastAudioScanServiceProxy(gatt_client.ProfileServiceProxy):
         )
 
         self.broadcast_receive_states = [
-            gatt.DelegatedCharacteristicAdapter(
+            gatt_adapters.DelegatedCharacteristicProxyAdapter(
                 characteristic,
                 decode=lambda x: BroadcastReceiveState.from_bytes(x) if x else None,
             )

bumble/profiles/battery_service.py

@@ -16,14 +16,20 @@
 # -----------------------------------------------------------------------------
 # Imports
 # -----------------------------------------------------------------------------
-from ..gatt_client import ProfileServiceProxy
+from typing import Optional
-from ..gatt import (
+
+from bumble.gatt_client import ProfileServiceProxy
+from bumble.gatt import (
     GATT_BATTERY_SERVICE,
     GATT_BATTERY_LEVEL_CHARACTERISTIC,
     TemplateService,
     Characteristic,
     CharacteristicValue,
+)
+from bumble.gatt_client import CharacteristicProxy
+from bumble.gatt_adapters import (
     PackedCharacteristicAdapter,
+    PackedCharacteristicProxyAdapter,
 )
 
 
@@ -32,6 +38,8 @@ class BatteryService(TemplateService):
     UUID = GATT_BATTERY_SERVICE
     BATTERY_LEVEL_FORMAT = 'B'
 
+    battery_level_characteristic: Characteristic[int]
+
     def __init__(self, read_battery_level):
         self.battery_level_characteristic = PackedCharacteristicAdapter(
             Characteristic(
@@ -49,13 +57,15 @@ class BatteryService(TemplateService):
 class BatteryServiceProxy(ProfileServiceProxy):
     SERVICE_CLASS = BatteryService
 
+    battery_level: Optional[CharacteristicProxy[int]]
+
     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(
+            self.battery_level = PackedCharacteristicProxyAdapter(
                 characteristics[0], pack_format=BatteryService.BATTERY_LEVEL_FORMAT
             )
         else:

bumble/profiles/device_information_service.py

@@ -19,7 +19,6 @@
 import struct
 from typing import Optional, Tuple
 
-from bumble.gatt_client import ServiceProxy, ProfileServiceProxy, CharacteristicProxy
 from bumble.gatt import (
     GATT_DEVICE_INFORMATION_SERVICE,
     GATT_FIRMWARE_REVISION_STRING_CHARACTERISTIC,
@@ -32,9 +31,12 @@ from bumble.gatt import (
     GATT_REGULATORY_CERTIFICATION_DATA_LIST_CHARACTERISTIC,
     TemplateService,
     Characteristic,
-    DelegatedCharacteristicAdapter,
-    UTF8CharacteristicAdapter,
 )
+from bumble.gatt_adapters import (
+    DelegatedCharacteristicProxyAdapter,
+    UTF8CharacteristicProxyAdapter,
+)
+from bumble.gatt_client import CharacteristicProxy, ProfileServiceProxy, ServiceProxy
 
 
 # -----------------------------------------------------------------------------
@@ -62,7 +64,7 @@ class DeviceInformationService(TemplateService):
         ieee_regulatory_certification_data_list: Optional[bytes] = None,
         # TODO: pnp_id
     ):
-        characteristics = [
+        characteristics: list[Characteristic[bytes]] = [
             Characteristic(
                 uuid,
                 Characteristic.Properties.READ,
@@ -107,14 +109,14 @@ class DeviceInformationService(TemplateService):
 class DeviceInformationServiceProxy(ProfileServiceProxy):
     SERVICE_CLASS = DeviceInformationService
 
-    manufacturer_name: Optional[UTF8CharacteristicAdapter]
+    manufacturer_name: Optional[CharacteristicProxy[str]]
-    model_number: Optional[UTF8CharacteristicAdapter]
+    model_number: Optional[CharacteristicProxy[str]]
-    serial_number: Optional[UTF8CharacteristicAdapter]
+    serial_number: Optional[CharacteristicProxy[str]]
-    hardware_revision: Optional[UTF8CharacteristicAdapter]
+    hardware_revision: Optional[CharacteristicProxy[str]]
-    firmware_revision: Optional[UTF8CharacteristicAdapter]
+    firmware_revision: Optional[CharacteristicProxy[str]]
-    software_revision: Optional[UTF8CharacteristicAdapter]
+    software_revision: Optional[CharacteristicProxy[str]]
-    system_id: Optional[DelegatedCharacteristicAdapter]
+    system_id: Optional[CharacteristicProxy[tuple[int, int]]]
-    ieee_regulatory_certification_data_list: Optional[CharacteristicProxy]
+    ieee_regulatory_certification_data_list: Optional[CharacteristicProxy[bytes]]
 
     def __init__(self, service_proxy: ServiceProxy):
         self.service_proxy = service_proxy
@@ -128,7 +130,7 @@ class DeviceInformationServiceProxy(ProfileServiceProxy):
             ('software_revision', GATT_SOFTWARE_REVISION_STRING_CHARACTERISTIC),
         ):
             if characteristics := service_proxy.get_characteristics_by_uuid(uuid):
-                characteristic = UTF8CharacteristicAdapter(characteristics[0])
+                characteristic = UTF8CharacteristicProxyAdapter(characteristics[0])
             else:
                 characteristic = None
             self.__setattr__(field, characteristic)
@@ -136,7 +138,7 @@ class DeviceInformationServiceProxy(ProfileServiceProxy):
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_SYSTEM_ID_CHARACTERISTIC
         ):
-            self.system_id = DelegatedCharacteristicAdapter(
+            self.system_id = DelegatedCharacteristicProxyAdapter(
                 characteristics[0],
                 encode=lambda v: DeviceInformationService.pack_system_id(*v),
                 decode=DeviceInformationService.unpack_system_id,

bumble/profiles/gap.py

@@ -25,14 +25,15 @@ from bumble.core import Appearance
 from bumble.gatt import (
     TemplateService,
     Characteristic,
-    CharacteristicAdapter,
-    DelegatedCharacteristicAdapter,
-    UTF8CharacteristicAdapter,
     GATT_GENERIC_ACCESS_SERVICE,
     GATT_DEVICE_NAME_CHARACTERISTIC,
     GATT_APPEARANCE_CHARACTERISTIC,
 )
-from bumble.gatt_client import ProfileServiceProxy, ServiceProxy
+from bumble.gatt_adapters import (
+    DelegatedCharacteristicProxyAdapter,
+    UTF8CharacteristicProxyAdapter,
+)
+from bumble.gatt_client import CharacteristicProxy, ProfileServiceProxy, ServiceProxy
 
 # -----------------------------------------------------------------------------
 # Logging
@@ -49,6 +50,9 @@ logger = logging.getLogger(__name__)
 class GenericAccessService(TemplateService):
     UUID = GATT_GENERIC_ACCESS_SERVICE
 
+    device_name_characteristic: Characteristic[bytes]
+    appearance_characteristic: Characteristic[bytes]
+
     def __init__(
         self, device_name: str, appearance: Union[Appearance, Tuple[int, int], int] = 0
     ):
@@ -84,8 +88,8 @@ class GenericAccessService(TemplateService):
 class GenericAccessServiceProxy(ProfileServiceProxy):
     SERVICE_CLASS = GenericAccessService
 
-    device_name: Optional[CharacteristicAdapter]
+    device_name: Optional[CharacteristicProxy[str]]
-    appearance: Optional[DelegatedCharacteristicAdapter]
+    appearance: Optional[CharacteristicProxy[Appearance]]
 
     def __init__(self, service_proxy: ServiceProxy):
         self.service_proxy = service_proxy
@@ -93,14 +97,14 @@ class GenericAccessServiceProxy(ProfileServiceProxy):
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_DEVICE_NAME_CHARACTERISTIC
         ):
-            self.device_name = UTF8CharacteristicAdapter(characteristics[0])
+            self.device_name = UTF8CharacteristicProxyAdapter(characteristics[0])
         else:
             self.device_name = None
 
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_APPEARANCE_CHARACTERISTIC
         ):
-            self.appearance = DelegatedCharacteristicAdapter(
+            self.appearance = DelegatedCharacteristicProxyAdapter(
                 characteristics[0],
                 decode=lambda value: Appearance.from_int(
                     struct.unpack_from('<H', value, 0)[0],

bumble/profiles/gatt_service.py

@@ -110,6 +110,7 @@ class GenericAttributeProfileService(gatt.TemplateService):
             gatt.GATT_CHARACTERISTIC_ATTRIBUTE_TYPE,
             gatt.GATT_CHARACTERISTIC_EXTENDED_PROPERTIES_DESCRIPTOR,
         ):
+            assert isinstance(attribute.value, bytes)
             return (
                 struct.pack("<H", attribute.handle)
                 + attribute.type.to_bytes()

bumble/profiles/gmap.py

@@ -22,7 +22,6 @@ from typing import Optional
 
 from bumble.gatt import (
     TemplateService,
-    DelegatedCharacteristicAdapter,
     Characteristic,
     GATT_GAMING_AUDIO_SERVICE,
     GATT_GMAP_ROLE_CHARACTERISTIC,
@@ -31,7 +30,8 @@ from bumble.gatt import (
     GATT_BGS_FEATURES_CHARACTERISTIC,
     GATT_BGR_FEATURES_CHARACTERISTIC,
 )
-from bumble.gatt_client import ProfileServiceProxy, ServiceProxy
+from bumble.gatt_adapters import DelegatedCharacteristicProxyAdapter
+from bumble.gatt_client import CharacteristicProxy, ProfileServiceProxy, ServiceProxy
 from enum import IntFlag
 
 
@@ -150,10 +150,15 @@ class GamingAudioService(TemplateService):
 class GamingAudioServiceProxy(ProfileServiceProxy):
     SERVICE_CLASS = GamingAudioService
 
+    ugg_features: Optional[CharacteristicProxy[UggFeatures]] = None
+    ugt_features: Optional[CharacteristicProxy[UgtFeatures]] = None
+    bgs_features: Optional[CharacteristicProxy[BgsFeatures]] = None
+    bgr_features: Optional[CharacteristicProxy[BgrFeatures]] = None
+
     def __init__(self, service_proxy: ServiceProxy) -> None:
         self.service_proxy = service_proxy
 
-        self.gmap_role = DelegatedCharacteristicAdapter(
+        self.gmap_role = DelegatedCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_GMAP_ROLE_CHARACTERISTIC
             ),
@@ -163,31 +168,31 @@ class GamingAudioServiceProxy(ProfileServiceProxy):
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_UGG_FEATURES_CHARACTERISTIC
         ):
-            self.ugg_features = DelegatedCharacteristicAdapter(
+            self.ugg_features = DelegatedCharacteristicProxyAdapter(
-                characteristic=characteristics[0],
+                characteristics[0],
                 decode=lambda value: UggFeatures(value[0]),
             )
 
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_UGT_FEATURES_CHARACTERISTIC
         ):
-            self.ugt_features = DelegatedCharacteristicAdapter(
+            self.ugt_features = DelegatedCharacteristicProxyAdapter(
-                characteristic=characteristics[0],
+                characteristics[0],
                 decode=lambda value: UgtFeatures(value[0]),
             )
 
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_BGS_FEATURES_CHARACTERISTIC
         ):
-            self.bgs_features = DelegatedCharacteristicAdapter(
+            self.bgs_features = DelegatedCharacteristicProxyAdapter(
-                characteristic=characteristics[0],
+                characteristics[0],
                 decode=lambda value: BgsFeatures(value[0]),
             )
 
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_BGR_FEATURES_CHARACTERISTIC
         ):
-            self.bgr_features = DelegatedCharacteristicAdapter(
+            self.bgr_features = DelegatedCharacteristicProxyAdapter(
-                characteristic=characteristics[0],
+                characteristics[0],
                 decode=lambda value: BgrFeatures(value[0]),
             )

bumble/profiles/hap.py

@@ -18,14 +18,15 @@
 from __future__ import annotations
 import asyncio
 import functools
-from bumble import att, gatt, gatt_client
+from dataclasses import dataclass, field
+import logging
+from typing import Any, Dict, List, Optional, Set, Union
+
+from bumble import att, gatt, gatt_adapters, gatt_client
 from bumble.core import InvalidArgumentError, InvalidStateError
 from bumble.device import Device, Connection
 from bumble.utils import AsyncRunner, OpenIntEnum
 from bumble.hci import Address
-from dataclasses import dataclass, field
-import logging
-from typing import Any, Dict, List, Optional, Set, Union
 
 
 # -----------------------------------------------------------------------------
@@ -631,11 +632,12 @@ class HearingAccessServiceProxy(gatt_client.ProfileServiceProxy):
 
     hearing_aid_preset_control_point: gatt_client.CharacteristicProxy
     preset_control_point_indications: asyncio.Queue
+    active_preset_index_notification: asyncio.Queue
 
     def __init__(self, service_proxy: gatt_client.ServiceProxy) -> None:
         self.service_proxy = service_proxy
 
-        self.server_features = gatt.PackedCharacteristicAdapter(
+        self.server_features = gatt_adapters.PackedCharacteristicProxyAdapter(
             service_proxy.get_characteristics_by_uuid(
                 gatt.GATT_HEARING_AID_FEATURES_CHARACTERISTIC
             )[0],
@@ -648,7 +650,7 @@ class HearingAccessServiceProxy(gatt_client.ProfileServiceProxy):
             )[0]
         )
 
-        self.active_preset_index = gatt.PackedCharacteristicAdapter(
+        self.active_preset_index = gatt_adapters.PackedCharacteristicProxyAdapter(
             service_proxy.get_characteristics_by_uuid(
                 gatt.GATT_ACTIVE_PRESET_INDEX_CHARACTERISTIC
             )[0],

bumble/profiles/heart_rate_service.py

@@ -16,13 +16,14 @@
 # -----------------------------------------------------------------------------
 # Imports
 # -----------------------------------------------------------------------------
+from __future__ import annotations
 from enum import IntEnum
 import struct
+from typing import Optional
 
 from bumble import core
-from ..gatt_client import ProfileServiceProxy
+from bumble.att import ATT_Error
-from ..att import ATT_Error
+from bumble.gatt import (
-from ..gatt import (
     GATT_HEART_RATE_SERVICE,
     GATT_HEART_RATE_MEASUREMENT_CHARACTERISTIC,
     GATT_BODY_SENSOR_LOCATION_CHARACTERISTIC,
@@ -30,10 +31,13 @@ from ..gatt import (
     TemplateService,
     Characteristic,
     CharacteristicValue,
-    SerializableCharacteristicAdapter,
+)
+from bumble.gatt_adapters import (
     DelegatedCharacteristicAdapter,
     PackedCharacteristicAdapter,
+    SerializableCharacteristicAdapter,
 )
+from bumble.gatt_client import CharacteristicProxy, ProfileServiceProxy
 
 
 # -----------------------------------------------------------------------------
@@ -43,6 +47,10 @@ class HeartRateService(TemplateService):
     CONTROL_POINT_NOT_SUPPORTED = 0x80
     RESET_ENERGY_EXPENDED = 0x01
 
+    heart_rate_measurement_characteristic: Characteristic[HeartRateMeasurement]
+    body_sensor_location_characteristic: Characteristic[BodySensorLocation]
+    heart_rate_control_point_characteristic: Characteristic[int]
+
     class BodySensorLocation(IntEnum):
         OTHER = 0
         CHEST = 1
@@ -198,6 +206,14 @@ class HeartRateService(TemplateService):
 class HeartRateServiceProxy(ProfileServiceProxy):
     SERVICE_CLASS = HeartRateService
 
+    heart_rate_measurement: Optional[
+        CharacteristicProxy[HeartRateService.HeartRateMeasurement]
+    ]
+    body_sensor_location: Optional[
+        CharacteristicProxy[HeartRateService.BodySensorLocation]
+    ]
+    heart_rate_control_point: Optional[CharacteristicProxy[int]]
+
     def __init__(self, service_proxy):
         self.service_proxy = service_proxy
 

bumble/profiles/mcp.py

@@ -208,7 +208,7 @@ class MediaControlService(gatt.TemplateService):
             properties=gatt.Characteristic.Properties.READ
             | gatt.Characteristic.Properties.NOTIFY,
             permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
-            value=media_player_name or 'Bumble Player',
+            value=(media_player_name or 'Bumble Player').encode(),
         )
         self.track_changed_characteristic = gatt.Characteristic(
             uuid=gatt.GATT_TRACK_CHANGED_CHARACTERISTIC,
@@ -247,14 +247,16 @@ class MediaControlService(gatt.TemplateService):
             permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
             value=b'',
         )
-        self.media_control_point_characteristic = gatt.Characteristic(
+        self.media_control_point_characteristic: gatt.Characteristic[bytes] = (
-            uuid=gatt.GATT_MEDIA_CONTROL_POINT_CHARACTERISTIC,
+            gatt.Characteristic(
-            properties=gatt.Characteristic.Properties.WRITE
+                uuid=gatt.GATT_MEDIA_CONTROL_POINT_CHARACTERISTIC,
-            | gatt.Characteristic.Properties.WRITE_WITHOUT_RESPONSE
+                properties=gatt.Characteristic.Properties.WRITE
-            | gatt.Characteristic.Properties.NOTIFY,
+                | gatt.Characteristic.Properties.WRITE_WITHOUT_RESPONSE
-            permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION
+                | gatt.Characteristic.Properties.NOTIFY,
-            | gatt.Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION,
+                permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION
-            value=gatt.CharacteristicValue(write=self.on_media_control_point),
+                | gatt.Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION,
+                value=gatt.CharacteristicValue(write=self.on_media_control_point),
+            )
         )
         self.media_control_point_opcodes_supported_characteristic = gatt.Characteristic(
             uuid=gatt.GATT_MEDIA_CONTROL_POINT_OPCODES_SUPPORTED_CHARACTERISTIC,

bumble/profiles/pacs.py

@@ -25,6 +25,7 @@ from typing import Optional, Sequence, Union
 from bumble.profiles.bap import AudioLocation, CodecSpecificCapabilities, ContextType
 from bumble.profiles import le_audio
 from bumble import gatt
+from bumble import gatt_adapters
 from bumble import gatt_client
 from bumble import hci
 
@@ -185,34 +186,42 @@ class PublishedAudioCapabilitiesService(gatt.TemplateService):
 class PublishedAudioCapabilitiesServiceProxy(gatt_client.ProfileServiceProxy):
     SERVICE_CLASS = PublishedAudioCapabilitiesService
 
-    sink_pac: Optional[gatt.DelegatedCharacteristicAdapter] = None
+    sink_pac: Optional[gatt_client.CharacteristicProxy[list[PacRecord]]] = None
-    sink_audio_locations: Optional[gatt.DelegatedCharacteristicAdapter] = None
+    sink_audio_locations: Optional[gatt_client.CharacteristicProxy[AudioLocation]] = (
-    source_pac: Optional[gatt.DelegatedCharacteristicAdapter] = None
+        None
-    source_audio_locations: Optional[gatt.DelegatedCharacteristicAdapter] = None
+    )
-    available_audio_contexts: gatt.DelegatedCharacteristicAdapter
+    source_pac: Optional[gatt_client.CharacteristicProxy[list[PacRecord]]] = None
-    supported_audio_contexts: gatt.DelegatedCharacteristicAdapter
+    source_audio_locations: Optional[gatt_client.CharacteristicProxy[AudioLocation]] = (
+        None
+    )
+    available_audio_contexts: gatt_client.CharacteristicProxy[tuple[ContextType, ...]]
+    supported_audio_contexts: gatt_client.CharacteristicProxy[tuple[ContextType, ...]]
 
     def __init__(self, service_proxy: gatt_client.ServiceProxy):
         self.service_proxy = service_proxy
 
-        self.available_audio_contexts = gatt.DelegatedCharacteristicAdapter(
+        self.available_audio_contexts = (
-            service_proxy.get_required_characteristic_by_uuid(
+            gatt_adapters.DelegatedCharacteristicProxyAdapter(
-                gatt.GATT_AVAILABLE_AUDIO_CONTEXTS_CHARACTERISTIC
+                service_proxy.get_required_characteristic_by_uuid(
-            ),
+                    gatt.GATT_AVAILABLE_AUDIO_CONTEXTS_CHARACTERISTIC
-            decode=lambda x: tuple(map(ContextType, struct.unpack('<HH', x))),
+                ),
+                decode=lambda x: tuple(map(ContextType, struct.unpack('<HH', x))),
+            )
         )
 
-        self.supported_audio_contexts = gatt.DelegatedCharacteristicAdapter(
+        self.supported_audio_contexts = (
-            service_proxy.get_required_characteristic_by_uuid(
+            gatt_adapters.DelegatedCharacteristicProxyAdapter(
-                gatt.GATT_SUPPORTED_AUDIO_CONTEXTS_CHARACTERISTIC
+                service_proxy.get_required_characteristic_by_uuid(
-            ),
+                    gatt.GATT_SUPPORTED_AUDIO_CONTEXTS_CHARACTERISTIC
-            decode=lambda x: tuple(map(ContextType, struct.unpack('<HH', x))),
+                ),
+                decode=lambda x: tuple(map(ContextType, struct.unpack('<HH', x))),
+            )
         )
 
         if characteristics := service_proxy.get_characteristics_by_uuid(
             gatt.GATT_SINK_PAC_CHARACTERISTIC
         ):
-            self.sink_pac = gatt.DelegatedCharacteristicAdapter(
+            self.sink_pac = gatt_adapters.DelegatedCharacteristicProxyAdapter(
                 characteristics[0],
                 decode=PacRecord.list_from_bytes,
             )
@@ -220,7 +229,7 @@ class PublishedAudioCapabilitiesServiceProxy(gatt_client.ProfileServiceProxy):
         if characteristics := service_proxy.get_characteristics_by_uuid(
             gatt.GATT_SOURCE_PAC_CHARACTERISTIC
         ):
-            self.source_pac = gatt.DelegatedCharacteristicAdapter(
+            self.source_pac = gatt_adapters.DelegatedCharacteristicProxyAdapter(
                 characteristics[0],
                 decode=PacRecord.list_from_bytes,
             )
@@ -228,15 +237,19 @@ class PublishedAudioCapabilitiesServiceProxy(gatt_client.ProfileServiceProxy):
         if characteristics := service_proxy.get_characteristics_by_uuid(
             gatt.GATT_SINK_AUDIO_LOCATION_CHARACTERISTIC
         ):
-            self.sink_audio_locations = gatt.DelegatedCharacteristicAdapter(
+            self.sink_audio_locations = (
-                characteristics[0],
+                gatt_adapters.DelegatedCharacteristicProxyAdapter(
-                decode=lambda x: AudioLocation(struct.unpack('<I', x)[0]),
+                    characteristics[0],
+                    decode=lambda x: AudioLocation(struct.unpack('<I', x)[0]),
+                )
             )
 
         if characteristics := service_proxy.get_characteristics_by_uuid(
             gatt.GATT_SOURCE_AUDIO_LOCATION_CHARACTERISTIC
         ):
-            self.source_audio_locations = gatt.DelegatedCharacteristicAdapter(
+            self.source_audio_locations = (
-                characteristics[0],
+                gatt_adapters.DelegatedCharacteristicProxyAdapter(
-                decode=lambda x: AudioLocation(struct.unpack('<I', x)[0]),
+                    characteristics[0],
+                    decode=lambda x: AudioLocation(struct.unpack('<I', x)[0]),
+                )
             )

bumble/profiles/pbp.py

@@ -40,7 +40,7 @@ class PublicBroadcastAnnouncement:
     def from_bytes(cls, data: bytes) -> Self:
         features = cls.Features(data[0])
         metadata_length = data[1]
-        metadata_ltv = data[1 : 1 + metadata_length]
+        metadata_ltv = data[2 : 2 + metadata_length]
         return cls(
             features=features, metadata=le_audio.Metadata.from_bytes(metadata_ltv)
         )

bumble/profiles/tmap.py

@@ -24,11 +24,11 @@ import struct
 from bumble.gatt import (
     TemplateService,
     Characteristic,
-    DelegatedCharacteristicAdapter,
     GATT_TELEPHONY_AND_MEDIA_AUDIO_SERVICE,
     GATT_TMAP_ROLE_CHARACTERISTIC,
 )
-from bumble.gatt_client import ProfileServiceProxy, ServiceProxy
+from bumble.gatt_adapters import DelegatedCharacteristicProxyAdapter
+from bumble.gatt_client import CharacteristicProxy, ProfileServiceProxy, ServiceProxy
 
 
 # -----------------------------------------------------------------------------
@@ -53,6 +53,8 @@ class Role(enum.IntFlag):
 class TelephonyAndMediaAudioService(TemplateService):
     UUID = GATT_TELEPHONY_AND_MEDIA_AUDIO_SERVICE
 
+    role_characteristic: Characteristic[bytes]
+
     def __init__(self, role: Role):
         self.role_characteristic = Characteristic(
             GATT_TMAP_ROLE_CHARACTERISTIC,
@@ -68,12 +70,12 @@ class TelephonyAndMediaAudioService(TemplateService):
 class TelephonyAndMediaAudioServiceProxy(ProfileServiceProxy):
     SERVICE_CLASS = TelephonyAndMediaAudioService
 
-    role: DelegatedCharacteristicAdapter
+    role: CharacteristicProxy[Role]
 
     def __init__(self, service_proxy: ServiceProxy):
         self.service_proxy = service_proxy
 
-        self.role = DelegatedCharacteristicAdapter(
+        self.role = DelegatedCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_TMAP_ROLE_CHARACTERISTIC
             ),

bumble/profiles/vcs.py

@@ -25,6 +25,7 @@ from typing import Optional, Sequence
 from bumble import att
 from bumble import device
 from bumble import gatt
+from bumble import gatt_adapters
 from bumble import gatt_client
 
 
@@ -209,14 +210,14 @@ class VolumeControlService(gatt.TemplateService):
 class VolumeControlServiceProxy(gatt_client.ProfileServiceProxy):
     SERVICE_CLASS = VolumeControlService
 
-    volume_control_point: gatt_client.CharacteristicProxy
+    volume_control_point: gatt_client.CharacteristicProxy[bytes]
-    volume_state: gatt.SerializableCharacteristicAdapter
+    volume_state: gatt_client.CharacteristicProxy[VolumeState]
-    volume_flags: gatt.DelegatedCharacteristicAdapter
+    volume_flags: gatt_client.CharacteristicProxy[VolumeFlags]
 
     def __init__(self, service_proxy: gatt_client.ServiceProxy) -> None:
         self.service_proxy = service_proxy
 
-        self.volume_state = gatt.SerializableCharacteristicAdapter(
+        self.volume_state = gatt_adapters.SerializableCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 gatt.GATT_VOLUME_STATE_CHARACTERISTIC
             ),
@@ -227,7 +228,7 @@ class VolumeControlServiceProxy(gatt_client.ProfileServiceProxy):
             gatt.GATT_VOLUME_CONTROL_POINT_CHARACTERISTIC
         )
 
-        self.volume_flags = gatt.DelegatedCharacteristicAdapter(
+        self.volume_flags = gatt_adapters.DelegatedCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 gatt.GATT_VOLUME_FLAGS_CHARACTERISTIC
             ),

bumble/profiles/vocs.py

@@ -24,17 +24,19 @@ from bumble.device import Connection
 from bumble.att import ATT_Error
 from bumble.gatt import (
     Characteristic,
-    DelegatedCharacteristicAdapter,
     TemplateService,
     CharacteristicValue,
-    SerializableCharacteristicAdapter,
-    UTF8CharacteristicAdapter,
     GATT_VOLUME_OFFSET_CONTROL_SERVICE,
     GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC,
     GATT_AUDIO_LOCATION_CHARACTERISTIC,
     GATT_VOLUME_OFFSET_CONTROL_POINT_CHARACTERISTIC,
     GATT_AUDIO_OUTPUT_DESCRIPTION_CHARACTERISTIC,
 )
+from bumble.gatt_adapters import (
+    DelegatedCharacteristicProxyAdapter,
+    SerializableCharacteristicProxyAdapter,
+    UTF8CharacteristicProxyAdapter,
+)
 from bumble.gatt_client import ProfileServiceProxy, ServiceProxy
 from bumble.utils import OpenIntEnum
 from bumble.profiles.bap import AudioLocation
@@ -67,7 +69,7 @@ class ErrorCode(OpenIntEnum):
 class VolumeOffsetState:
     volume_offset: int = 0
     change_counter: int = 0
-    attribute_value: Optional[CharacteristicValue] = None
+    attribute: Optional[Characteristic] = None
 
     def __bytes__(self) -> bytes:
         return struct.pack('<hB', self.volume_offset, self.change_counter)
@@ -81,8 +83,8 @@ class VolumeOffsetState:
         self.change_counter = (self.change_counter + 1) % (CHANGE_COUNTER_MAX_VALUE + 1)
 
     async def notify_subscribers_via_connection(self, connection: Connection) -> None:
-        assert self.attribute_value is not None
+        assert self.attribute is not None
-        await connection.device.notify_subscribers(attribute=self.attribute_value)
+        await connection.device.notify_subscribers(attribute=self.attribute)
 
     def on_read(self, _connection: Optional[Connection]) -> bytes:
         return bytes(self)
@@ -91,7 +93,7 @@ class VolumeOffsetState:
 @dataclass
 class VocsAudioLocation:
     audio_location: AudioLocation = AudioLocation.NOT_ALLOWED
-    attribute_value: Optional[CharacteristicValue] = None
+    attribute: Optional[Characteristic] = None
 
     def __bytes__(self) -> bytes:
         return struct.pack('<I', self.audio_location)
@@ -106,10 +108,10 @@ class VocsAudioLocation:
 
     async def on_write(self, connection: Optional[Connection], value: bytes) -> None:
         assert connection
-        assert self.attribute_value
+        assert self.attribute
 
         self.audio_location = AudioLocation(int.from_bytes(value, 'little'))
-        await connection.device.notify_subscribers(attribute=self.attribute_value)
+        await connection.device.notify_subscribers(attribute=self.attribute)
 
 
 @dataclass
@@ -148,7 +150,7 @@ class VolumeOffsetControlPoint:
 @dataclass
 class AudioOutputDescription:
     audio_output_description: str = ''
-    attribute_value: Optional[CharacteristicValue] = None
+    attribute: Optional[Characteristic] = None
 
     @classmethod
     def from_bytes(cls, data: bytes):
@@ -162,10 +164,10 @@ class AudioOutputDescription:
 
     async def on_write(self, connection: Optional[Connection], value: bytes) -> None:
         assert connection
-        assert self.attribute_value
+        assert self.attribute
 
         self.audio_output_description = value.decode('utf-8')
-        await connection.device.notify_subscribers(attribute=self.attribute_value)
+        await connection.device.notify_subscribers(attribute=self.attribute)
 
 
 # -----------------------------------------------------------------------------
@@ -197,7 +199,7 @@ class VolumeOffsetControlService(TemplateService):
             VolumeOffsetControlPoint(self.volume_offset_state)
         )
 
-        self.volume_offset_state_characteristic = Characteristic(
+        self.volume_offset_state_characteristic: Characteristic[bytes] = Characteristic(
             uuid=GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC,
             properties=(
                 Characteristic.Properties.READ | Characteristic.Properties.NOTIFY
@@ -206,7 +208,7 @@ class VolumeOffsetControlService(TemplateService):
             value=CharacteristicValue(read=self.volume_offset_state.on_read),
         )
 
-        self.audio_location_characteristic = Characteristic(
+        self.audio_location_characteristic: Characteristic[bytes] = Characteristic(
             uuid=GATT_AUDIO_LOCATION_CHARACTERISTIC,
             properties=(
                 Characteristic.Properties.READ
@@ -222,33 +224,39 @@ class VolumeOffsetControlService(TemplateService):
                 write=self.audio_location.on_write,
             ),
         )
-        self.audio_location.attribute_value = self.audio_location_characteristic.value
+        self.audio_location.attribute = self.audio_location_characteristic
 
-        self.volume_offset_control_point_characteristic = Characteristic(
+        self.volume_offset_control_point_characteristic: Characteristic[bytes] = (
-            uuid=GATT_VOLUME_OFFSET_CONTROL_POINT_CHARACTERISTIC,
+            Characteristic(
-            properties=Characteristic.Properties.WRITE,
+                uuid=GATT_VOLUME_OFFSET_CONTROL_POINT_CHARACTERISTIC,
-            permissions=Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION,
+                properties=Characteristic.Properties.WRITE,
-            value=CharacteristicValue(write=self.volume_offset_control_point.on_write),
+                permissions=Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION,
+                value=CharacteristicValue(
+                    write=self.volume_offset_control_point.on_write
+                ),
+            )
         )
 
-        self.audio_output_description_characteristic = Characteristic(
+        self.audio_output_description_characteristic: Characteristic[bytes] = (
-            uuid=GATT_AUDIO_OUTPUT_DESCRIPTION_CHARACTERISTIC,
+            Characteristic(
-            properties=(
+                uuid=GATT_AUDIO_OUTPUT_DESCRIPTION_CHARACTERISTIC,
-                Characteristic.Properties.READ
+                properties=(
-                | Characteristic.Properties.NOTIFY
+                    Characteristic.Properties.READ
-                | Characteristic.Properties.WRITE_WITHOUT_RESPONSE
+                    | Characteristic.Properties.NOTIFY
-            ),
+                    | Characteristic.Properties.WRITE_WITHOUT_RESPONSE
-            permissions=(
+                ),
-                Characteristic.Permissions.READ_REQUIRES_ENCRYPTION
+                permissions=(
-                | Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION
+                    Characteristic.Permissions.READ_REQUIRES_ENCRYPTION
-            ),
+                    | Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION
-            value=CharacteristicValue(
+                ),
-                read=self.audio_output_description.on_read,
+                value=CharacteristicValue(
-                write=self.audio_output_description.on_write,
+                    read=self.audio_output_description.on_read,
-            ),
+                    write=self.audio_output_description.on_write,
+                ),
+            )
         )
-        self.audio_output_description.attribute_value = (
+        self.audio_output_description.attribute = (
-            self.audio_output_description_characteristic.value
+            self.audio_output_description_characteristic
         )
 
         super().__init__(
@@ -271,14 +279,14 @@ class VolumeOffsetControlServiceProxy(ProfileServiceProxy):
     def __init__(self, service_proxy: ServiceProxy) -> None:
         self.service_proxy = service_proxy
 
-        self.volume_offset_state = SerializableCharacteristicAdapter(
+        self.volume_offset_state = SerializableCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC
             ),
             VolumeOffsetState,
         )
 
-        self.audio_location = DelegatedCharacteristicAdapter(
+        self.audio_location = DelegatedCharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_AUDIO_LOCATION_CHARACTERISTIC
             ),
@@ -292,7 +300,7 @@ class VolumeOffsetControlServiceProxy(ProfileServiceProxy):
             )
         )
 
-        self.audio_output_description = UTF8CharacteristicAdapter(
+        self.audio_output_description = UTF8CharacteristicProxyAdapter(
             service_proxy.get_required_characteristic_by_uuid(
                 GATT_AUDIO_OUTPUT_DESCRIPTION_CHARACTERISTIC
             )

bumble/smp.py

@@ -46,13 +46,13 @@ from pyee import EventEmitter
 from .colors import color
 from .hci import (
     Address,
+    Role,
     HCI_LE_Enable_Encryption_Command,
     HCI_Object,
     key_with_value,
 )
 from .core import (
     BT_BR_EDR_TRANSPORT,
-    BT_CENTRAL_ROLE,
     BT_LE_TRANSPORT,
     AdvertisingData,
     InvalidArgumentError,
@@ -1975,7 +1975,7 @@ class Manager(EventEmitter):
 
         # Look for a session with this connection, and create one if none exists
         if not (session := self.sessions.get(connection.handle)):
-            if connection.role == BT_CENTRAL_ROLE:
+            if connection.role == Role.CENTRAL:
                 logger.warning('Remote starts pairing as Peripheral!')
             pairing_config = self.pairing_config_factory(connection)
             session = self.session_proxy(
@@ -1995,7 +1995,7 @@ class Manager(EventEmitter):
 
     async def pair(self, connection: Connection) -> None:
         # TODO: check if there's already a session for this connection
-        if connection.role != BT_CENTRAL_ROLE:
+        if connection.role != Role.CENTRAL:
             logger.warning('Start pairing as Peripheral!')
         pairing_config = self.pairing_config_factory(connection)
         session = self.session_proxy(

bumble/transport/common.py

@@ -139,6 +139,7 @@ class PacketParser:
                         packet_type
                     ) or self.extended_packet_info.get(packet_type)
                     if self.packet_info is None:
+                        self.reset()
                         raise core.InvalidPacketError(
                             f'invalid packet type {packet_type}'
                         )
@@ -302,7 +303,10 @@ class ParserSource(BaseSource):
 # -----------------------------------------------------------------------------
 class StreamPacketSource(asyncio.Protocol, ParserSource):
     def data_received(self, data: bytes) -> None:
-        self.parser.feed_data(data)
+        try:
+            self.parser.feed_data(data)
+        except core.InvalidPacketError:
+            logger.warning("invalid packet, ignoring data")
 
 
 # -----------------------------------------------------------------------------

bumble/transport/usb.py

@@ -115,9 +115,7 @@ async def open_usb_transport(spec: str) -> Transport:
             self.acl_out = acl_out
             self.acl_out_transfer = device.getTransfer()
             self.acl_out_transfer_ready = asyncio.Semaphore(1)
-            self.packets: asyncio.Queue[bytes] = (
+            self.packets = asyncio.Queue[bytes]()  # Queue of packets waiting to be sent
-                asyncio.Queue()
-            )  # Queue of packets waiting to be sent
             self.loop = asyncio.get_running_loop()
             self.queue_task = None
             self.cancel_done = self.loop.create_future()

bumble/utils.py

@@ -502,3 +502,13 @@ class ByteSerializable(Protocol):
     def from_bytes(cls, data: bytes) -> Self: ...
 
     def __bytes__(self) -> bytes: ...
+
+
+# -----------------------------------------------------------------------------
+class IntConvertible(Protocol):
+    """
+    Type protocol for classes that can be instantiated from int and converted to int.
+    """
+
+    def __init__(self, value: int) -> None: ...
+    def __int__(self) -> int: ...

docs/mkdocs/src/apps_and_tools/auracast.md

@@ -0,0 +1,202 @@
+AURACAST TOOL
+=============
+
+The "auracast" tool implements commands that implement broadcasting, receiving
+and controlling LE Audio broadcasts.
+
+=== "Running as an installed package"
+    ```
+    $ bumble-auracast
+    ```
+
+=== "Running from source"
+    ```
+    $ python3 apps/auracast.py <args>
+    ```
+
+# Python Dependencies
+Try installing the optional `[auracast]` dependencies:
+
+=== "From source"
+    ```bash
+    $ python3 -m pip install ".[auracast]"
+    ```
+
+=== "From PyPI"
+    ```bash
+    $ python3 -m pip install "bumble[auracast]"
+    ```
+
+## LC3
+The `auracast` app depends on the `lc3` python module, which is available
+either as PyPI module (currently only available for Linux x86_64).
+When installing Bumble with the optional `auracast` dependency, the `lc3`
+module will be installed from the `lc3py` PyPI package if available.
+If not, you will need to install it separately. This can be done with:
+```bash
+$ python3 -m pip install "git+https://github.com/google/liblc3.git"
+```
+
+## SoundDevice
+The `sounddevice` module is required for audio output to the host's sound
+output device(s) and/or input from the host's input device(s).
+If not installed, the `auracast` app is still functional, but will be limited
+to non-device inputs and output (files, external processes, ...)
+
+On macOS and Windows, the `sounddevice` module gets installed with the
+native PortAudio libraries included.
+
+For Linux, however, PortAudio must be installed separately.
+This is typically done with a command like:
+```bash
+$ sudo apt install libportaudio2
+```
+
+Visit the [sounddevice documentation](https://python-sounddevice.readthedocs.io/)
+for details.
+
+
+# General Usage
+```
+Usage: bumble-auracast [OPTIONS] COMMAND [ARGS]...
+
+Options:
+  --help  Show this message and exit.
+
+Commands:
+  assist    Scan for broadcasts on behalf of an audio server
+  pair      Pair with an audio server
+  receive   Receive a broadcast source
+  scan      Scan for public broadcasts
+  transmit  Transmit a broadcast source
+```
+
+Use `bumble-auracast <command> --help` to get more detailed usage information
+for a specific `<command>`.
+
+## `assist`
+Act as a broadcast assistant.
+
+Use `bumble-auracast assist --help` for details on the commands and options.
+
+The assistant commands are:
+
+### `monitor-state`
+Subscribe to the state characteristic and monitor changes.
+
+### `add-source`
+Add a broadcast source. This will instruct the device to start
+receiving a broadcast.
+
+### `modify-source`
+Modify a broadcast source.
+
+### `remove-source`
+Remote a broadcast source.
+
+## `pair`
+Pair with a device.
+
+## `receive`
+Receive a broadcast source.
+
+The `--output` option specifies where to send the decoded audio samples.
+The following outputs are supported:
+
+### Sound Device
+The `--output` argument is either `device`, to send the audio to the hosts's default sound device, or `device:<DEVICE_ID>` where `<DEVICE_ID>`
+is the integer ID of one of the available sound devices.
+When invoked with `--output "device:?"`, a list of available devices and
+their IDs is printed out.
+
+### Standard Output
+With `--output stdout`, the decoded audio samples are written to the
+standard output (currently always as float32 PCM samples)
+
+### FFPlay
+With `--output ffplay`, the decoded audio samples are piped to `ffplay`
+in a child process. This option is only available if `ffplay` is a command that is available on the host.
+
+### File
+With `--output <filename>` or `--output file:<filename>`, the decoded audio
+samples are written to a file (currently always as float32 PCM)
+
+## `transmit`
+Broadcast an audio source as a transmitter.
+
+The `--input` and `--input-format` options specify what audio input
+source to transmit.
+The following inputs are supported:
+
+### Sound Device
+The `--input` argument is either `device`, to use the host's default sound
+device (typically a builtin microphone), or `device:<DEVICE_ID>` where
+`<DEVICE_ID>` is the integer ID of one of the available sound devices.
+When invoked with `--input "device:?"`, a list of available devices and their
+IDs is printed out.
+
+### Standard Input
+With `--input stdout`, the audio samples are read from the standard input.
+(currently always as int16 PCM).
+
+### File
+With `--input <filename>` or `--input file:<filename>`, the audio samples
+are read from a .wav or raw PCM file.
+
+Use the `--input-format <FORMAT>` option to specify the format of the audio
+samples in raw PCM files. `<FORMAT>` is expressed as:
+`<sample-type>,<sample-rate>,<channels>`
+(the only supported <sample-type> currently is 'int16le' for 16 bit signed integers with little-endian byte order)
+
+## `scan`
+Scan for public broadcasts.
+
+A live display of the available broadcasts is displayed continuously.
+
+# Compatibility With Some Products
+The `auracast` app has been tested for compatibility with a few products.
+The list is still very limited. Please let us know if there are products
+that are not working well, or if there are specific instructions that should
+be shared to allow better compatibiity with certain products.
+
+## Transmitters
+
+The `receive` command has been tested to successfully receive broadcasts from
+the following transmitters:
+
+  * JBL GO 4
+  * Flairmesh FlooGoo FMA120
+  * Eppfun AK3040Pro Max
+  * HIGHGAZE BA-25T
+  * Nexum Audio VOCE and USB dongle
+
+## Receivers
+
+### Pixel Buds Pro 2
+
+The Pixel Buds Pro 2 can be used as a broadcast receiver, controlled by the
+`auracast assist` command, instructing the buds to receive a broadcast.
+
+Use the `assist --command add-source` command to tell the buds to receive a
+broadcast.
+
+Use the `assist --command monitor-state` command to monitor the current sync/receive
+state of the buds.
+
+### JBL
+The JBL GO 4 and other JBL products that support the Auracast feature can be used
+as transmitters or receivers.
+
+When running in receiver mode (pressing the Auracast button while not already playing),
+the JBL speaker will scan for broadcast advertisements with a specific manufacturer data.
+Use the `--manufacturer-data` option of the `transmit` command in order to include data
+that will let the speaker recognize the broadcast as a compatible source.
+
+The manufacturer ID for JBL is 87.
+Using an option like `--manufacturer-data 87:00000000000000000000000000000000dffd` should work (tested on the
+JBL GO 4. The `dffd` value at the end of the payload may be different on other models?).
+
+
+### Others
+
+  * Nexum Audio VOCE and USB dongle

examples/heart_rate_server.py

@@ -102,7 +102,6 @@ async def main() -> None:
         )
 
         # Notify subscribers of the current value as soon as they subscribe
-        @heart_rate_service.heart_rate_measurement_characteristic.on('subscription')
         def on_subscription(connection, notify_enabled, indicate_enabled):
             if notify_enabled or indicate_enabled:
                 AsyncRunner.spawn(
@@ -112,6 +111,10 @@ async def main() -> None:
                     )
                 )
 
+        heart_rate_service.heart_rate_measurement_characteristic.on(
+            'subscription', on_subscription
+        )
+
         # Go!
         await device.power_on()
         await device.start_advertising(auto_restart=True)

examples/run_gatt_client_and_server.py

@@ -70,13 +70,13 @@ async def main() -> None:
     descriptor = Descriptor(
         GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
         Descriptor.READABLE,
-        'My Description',
+        'My Description'.encode(),
     )
-    manufacturer_name_characteristic = Characteristic(
+    manufacturer_name_characteristic = Characteristic[bytes](
         GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
         Characteristic.Properties.READ,
         Characteristic.READABLE,
-        "Fitbit",
+        "Fitbit".encode(),
         [descriptor],
     )
     device_info_service = Service(

examples/run_gatt_server.py

@@ -94,13 +94,13 @@ async def main() -> None:
         descriptor = Descriptor(
             GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
             Descriptor.READABLE,
-            'My Description',
+            'My Description'.encode(),
         )
         manufacturer_name_characteristic = Characteristic(
             GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
             Characteristic.Properties.READ,
             Characteristic.READABLE,
-            'Fitbit',
+            'Fitbit'.encode(),
             [descriptor],
         )
         device_info_service = Service(

examples/run_gatt_with_adapters.py

@@ -1,4 +1,4 @@
-# Copyright 2024 Google LLC
+# Copyright 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.
@@ -18,6 +18,8 @@
 from __future__ import annotations
 import asyncio
 import dataclasses
+import functools
+import enum
 import logging
 import os
 import random
@@ -28,6 +30,8 @@ from typing import Any, List, Union
 from bumble.device import Device, Peer
 from bumble import transport
 from bumble import gatt
+from bumble import gatt_adapters
+from bumble import gatt_client
 from bumble import hci
 from bumble import core
 
@@ -36,6 +40,9 @@ from bumble import core
 SERVICE_UUID = core.UUID("50DB505C-8AC4-4738-8448-3B1D9CC09CC5")
 CHARACTERISTIC_UUID_BASE = "D901B45B-4916-412E-ACCA-0000000000"
 
+DEFAULT_CLIENT_ADDRESS = "F0:F1:F2:F3:F4:F5"
+DEFAULT_SERVER_ADDRESS = "F1:F2:F3:F4:F5:F6"
+
 
 # -----------------------------------------------------------------------------
 @dataclasses.dataclass
@@ -65,6 +72,12 @@ class CustomClass:
         return struct.pack(">II", self.a, self.b)
 
 
+# -----------------------------------------------------------------------------
+class CustomEnum(enum.IntEnum):
+    FOO = 1234
+    BAR = 5678
+
+
 # -----------------------------------------------------------------------------
 async def client(device: Device, address: hci.Address) -> None:
     print(f'=== Connecting to {address}...')
@@ -78,8 +91,8 @@ async def client(device: Device, address: hci.Address) -> None:
     print("*** Discovery complete")
 
     service = peer.get_services_by_uuid(SERVICE_UUID)[0]
-    characteristics = []
+    characteristics: list[gatt_client.CharacteristicProxy] = []
-    for index in range(1, 9):
+    for index in range(1, 10):
         characteristics.append(
             service.get_characteristics_by_uuid(
                 core.UUID(CHARACTERISTIC_UUID_BASE + f"{index:02X}")
@@ -91,59 +104,92 @@ async def client(device: Device, address: hci.Address) -> None:
         value = await characteristic.read_value()
         print(f"### {characteristic} = {value!r} ({value.hex()})")
 
+    # Subscribe to all characteristics as a raw bytes listener.
+    def on_raw_characteristic_update(characteristic, value):
+        print(f"^^^ Update[RAW] {characteristic.uuid} value = {value.hex()}")
+
+    for characteristic in characteristics:
+        await characteristic.subscribe(
+            functools.partial(on_raw_characteristic_update, characteristic)
+        )
+
+    # Function to subscribe to adapted characteristics
+    def on_adapted_characteristic_update(characteristic, value):
+        print(
+            f"^^^ Update[ADAPTED] {characteristic.uuid} value = {value!r}, "
+            f"type={type(value)}"
+        )
+
     # Static characteristic with a bytes value.
     c1 = characteristics[0]
     c1_value = await c1.read_value()
     print(f"@@@ C1 {c1} value = {c1_value!r} (type={type(c1_value)})")
     await c1.write_value("happy π day".encode("utf-8"))
+    await c1.subscribe(functools.partial(on_adapted_characteristic_update, c1))
 
     # Static characteristic with a string value.
-    c2 = gatt.UTF8CharacteristicAdapter(characteristics[1])
+    c2 = gatt_adapters.UTF8CharacteristicProxyAdapter(characteristics[1])
     c2_value = await c2.read_value()
     print(f"@@@ C2 {c2} value = {c2_value} (type={type(c2_value)})")
     await c2.write_value("happy π day")
+    await c2.subscribe(functools.partial(on_adapted_characteristic_update, c2))
 
     # Static characteristic with a tuple value.
-    c3 = gatt.PackedCharacteristicAdapter(characteristics[2], ">III")
+    c3 = gatt_adapters.PackedCharacteristicProxyAdapter(characteristics[2], ">III")
     c3_value = await c3.read_value()
     print(f"@@@ C3 {c3} value = {c3_value} (type={type(c3_value)})")
     await c3.write_value((2001, 2002, 2003))
+    await c3.subscribe(functools.partial(on_adapted_characteristic_update, c3))
 
     # Static characteristic with a named tuple value.
-    c4 = gatt.MappedCharacteristicAdapter(
+    c4 = gatt_adapters.MappedCharacteristicProxyAdapter(
         characteristics[3], ">III", ["f1", "f2", "f3"]
     )
     c4_value = await c4.read_value()
     print(f"@@@ C4 {c4} value = {c4_value} (type={type(c4_value)})")
     await c4.write_value({"f1": 4001, "f2": 4002, "f3": 4003})
+    await c4.subscribe(functools.partial(on_adapted_characteristic_update, c4))
 
     # Static characteristic with a serializable value.
-    c5 = gatt.SerializableCharacteristicAdapter(
+    c5 = gatt_adapters.SerializableCharacteristicProxyAdapter(
         characteristics[4], CustomSerializableClass
     )
     c5_value = await c5.read_value()
     print(f"@@@ C5 {c5} value = {c5_value} (type={type(c5_value)})")
     await c5.write_value(CustomSerializableClass(56, 57))
+    await c5.subscribe(functools.partial(on_adapted_characteristic_update, c5))
 
     # Static characteristic with a delegated value.
-    c6 = gatt.DelegatedCharacteristicAdapter(
+    c6 = gatt_adapters.DelegatedCharacteristicProxyAdapter(
         characteristics[5], encode=CustomClass.encode, decode=CustomClass.decode
     )
     c6_value = await c6.read_value()
     print(f"@@@ C6 {c6} value = {c6_value} (type={type(c6_value)})")
     await c6.write_value(CustomClass(6, 7))
+    await c6.subscribe(functools.partial(on_adapted_characteristic_update, c6))
 
     # Dynamic characteristic with a bytes value.
     c7 = characteristics[6]
     c7_value = await c7.read_value()
     print(f"@@@ C7 {c7} value = {c7_value!r} (type={type(c7_value)})")
     await c7.write_value(bytes.fromhex("01020304"))
+    await c7.subscribe(functools.partial(on_adapted_characteristic_update, c7))
 
     # Dynamic characteristic with a string value.
-    c8 = gatt.UTF8CharacteristicAdapter(characteristics[7])
+    c8 = gatt_adapters.UTF8CharacteristicProxyAdapter(characteristics[7])
     c8_value = await c8.read_value()
     print(f"@@@ C8 {c8} value = {c8_value} (type={type(c8_value)})")
     await c8.write_value("howdy")
+    await c8.subscribe(functools.partial(on_adapted_characteristic_update, c8))
+
+    # Static characteristic with an enum value
+    c9 = gatt_adapters.EnumCharacteristicProxyAdapter(
+        characteristics[8], CustomEnum, 3, 'big'
+    )
+    c9_value = await c9.read_value()
+    print(f"@@@ C9 {c9} value = {c9_value.name} (type={type(c9_value)})")
+    await c9.write_value(CustomEnum.BAR)
+    await c9.subscribe(functools.partial(on_adapted_characteristic_update, c9))
 
 
 # -----------------------------------------------------------------------------
@@ -175,142 +221,213 @@ def on_characteristic_write(characteristic: gatt.Characteristic, value: Any) ->
     print(f"<<< WRITE: {characteristic} <- {value}  ({type(value)})")
 
 
+# -----------------------------------------------------------------------------
+async def server(device: Device) -> None:
+    # Static characteristic with a bytes value.
+    c1 = gatt.Characteristic(
+        CHARACTERISTIC_UUID_BASE + "01",
+        gatt.Characteristic.Properties.READ
+        | gatt.Characteristic.Properties.WRITE
+        | gatt.Characteristic.Properties.NOTIFY,
+        gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+        b'hello',
+    )
+
+    # Static characteristic with a string value.
+    c2 = gatt_adapters.UTF8CharacteristicAdapter(
+        gatt.Characteristic(
+            CHARACTERISTIC_UUID_BASE + "02",
+            gatt.Characteristic.Properties.READ
+            | gatt.Characteristic.Properties.WRITE
+            | gatt.Characteristic.Properties.NOTIFY,
+            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+            'hello',
+        )
+    )
+
+    # Static characteristic with a tuple value.
+    c3 = gatt_adapters.PackedCharacteristicAdapter(
+        gatt.Characteristic(
+            CHARACTERISTIC_UUID_BASE + "03",
+            gatt.Characteristic.Properties.READ
+            | gatt.Characteristic.Properties.WRITE
+            | gatt.Characteristic.Properties.NOTIFY,
+            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+            (1007, 1008, 1009),
+        ),
+        ">III",
+    )
+
+    # Static characteristic with a named tuple value.
+    c4 = gatt_adapters.MappedCharacteristicAdapter(
+        gatt.Characteristic(
+            CHARACTERISTIC_UUID_BASE + "04",
+            gatt.Characteristic.Properties.READ
+            | gatt.Characteristic.Properties.WRITE
+            | gatt.Characteristic.Properties.NOTIFY,
+            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+            {"f1": 3007, "f2": 3008, "f3": 3009},
+        ),
+        ">III",
+        ["f1", "f2", "f3"],
+    )
+
+    # Static characteristic with a serializable value.
+    c5 = gatt_adapters.SerializableCharacteristicAdapter(
+        gatt.Characteristic(
+            CHARACTERISTIC_UUID_BASE + "05",
+            gatt.Characteristic.Properties.READ
+            | gatt.Characteristic.Properties.WRITE
+            | gatt.Characteristic.Properties.NOTIFY,
+            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+            CustomSerializableClass(11, 12),
+        ),
+        CustomSerializableClass,
+    )
+
+    # Static characteristic with a delegated value.
+    c6 = gatt_adapters.DelegatedCharacteristicAdapter(
+        gatt.Characteristic(
+            CHARACTERISTIC_UUID_BASE + "06",
+            gatt.Characteristic.Properties.READ
+            | gatt.Characteristic.Properties.WRITE
+            | gatt.Characteristic.Properties.NOTIFY,
+            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+            CustomClass(1, 2),
+        ),
+        encode=CustomClass.encode,
+        decode=CustomClass.decode,
+    )
+
+    # Dynamic characteristic with a bytes value.
+    c7 = gatt.Characteristic(
+        CHARACTERISTIC_UUID_BASE + "07",
+        gatt.Characteristic.Properties.READ
+        | gatt.Characteristic.Properties.WRITE
+        | gatt.Characteristic.Properties.NOTIFY,
+        gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+        gatt.CharacteristicValue(
+            read=lambda connection: dynamic_read("bytes"),
+            write=lambda connection, value: dynamic_write("bytes", value),
+        ),
+    )
+
+    # Dynamic characteristic with a string value.
+    c8 = gatt_adapters.UTF8CharacteristicAdapter(
+        gatt.Characteristic(
+            CHARACTERISTIC_UUID_BASE + "08",
+            gatt.Characteristic.Properties.READ
+            | gatt.Characteristic.Properties.WRITE
+            | gatt.Characteristic.Properties.NOTIFY,
+            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+            gatt.CharacteristicValue(
+                read=lambda connection: dynamic_read("string"),
+                write=lambda connection, value: dynamic_write("string", value),
+            ),
+        )
+    )
+
+    # Static characteristic with an enum value
+    c9 = gatt_adapters.EnumCharacteristicAdapter(
+        gatt.Characteristic(
+            CHARACTERISTIC_UUID_BASE + "09",
+            gatt.Characteristic.Properties.READ
+            | gatt.Characteristic.Properties.WRITE
+            | gatt.Characteristic.Properties.NOTIFY,
+            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
+            CustomEnum.FOO,
+        ),
+        cls=CustomEnum,
+        length=3,
+        byteorder='big',
+    )
+
+    characteristics: List[gatt.Characteristic] = [
+        c1,
+        c2,
+        c3,
+        c4,
+        c5,
+        c6,
+        c7,
+        c8,
+        c9,
+    ]
+
+    # Listen for read and write events.
+    for characteristic in characteristics:
+        characteristic.on(
+            "read",
+            lambda _, value, c=characteristic: on_characteristic_read(c, value),
+        )
+        characteristic.on(
+            "write",
+            lambda _, value, c=characteristic: on_characteristic_write(c, value),
+        )
+
+    device.add_service(gatt.Service(SERVICE_UUID, characteristics))
+
+    # Notify every 3 seconds
+    i = 0
+    while True:
+        await asyncio.sleep(3)
+
+        # Notifying can be done with the characteristic's current value, or
+        # by explicitly passing a value to notify with. Both variants are used
+        # here: for c1..c4 we set the value and then notify, for c4..c9 we notify
+        # with an explicit value.
+        c1.value = f'hello c1 {i}'.encode()
+        await device.notify_subscribers(c1)
+        c2.value = f'hello c2 {i}'
+        await device.notify_subscribers(c2)
+        c3.value = (1000 + i, 2000 + i, 3000 + i)
+        await device.notify_subscribers(c3)
+        c4.value = {"f1": 4000 + i, "f2": 5000 + i, "f3": 6000 + i}
+        await device.notify_subscribers(c4)
+        await device.notify_subscribers(c5, CustomSerializableClass(1000 + i, 2000 + i))
+        await device.notify_subscribers(c6, CustomClass(3000 + i, 4000 + i))
+        await device.notify_subscribers(c7, bytes([1, 2, 3, i % 256]))
+        await device.notify_subscribers(c8, f'hello c8 {i}')
+        await device.notify_subscribers(
+            c9, CustomEnum.FOO if i % 2 == 0 else CustomEnum.BAR
+        )
+
+        i += 1
+
+
 # -----------------------------------------------------------------------------
 async def main() -> None:
     if len(sys.argv) < 2:
-        print("Usage: run_gatt_with_adapters.py <transport-spec> [<bluetooth-address>]")
+        print("Usage: run_gatt_with_adapters.py <transport-spec> client|server")
-        print("example: run_gatt_with_adapters.py usb:0 E1:CA:72:48:C4:E8")
+        print("example: run_gatt_with_adapters.py usb:0 F0:F1:F2:F3:F4:F5")
         return
 
     async with await transport.open_transport(sys.argv[1]) as hci_transport:
+        is_client = sys.argv[2] == "client"
+
         # Create a device to manage the host
         device = Device.with_hci(
             "Bumble",
-            hci.Address("F0:F1:F2:F3:F4:F5"),
+            hci.Address(
+                DEFAULT_CLIENT_ADDRESS if is_client else DEFAULT_SERVER_ADDRESS
+            ),
             hci_transport.source,
             hci_transport.sink,
         )
 
-        # Static characteristic with a bytes value.
-        c1 = gatt.Characteristic(
-            CHARACTERISTIC_UUID_BASE + "01",
-            gatt.Characteristic.Properties.READ | gatt.Characteristic.Properties.WRITE,
-            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-            b'hello',
-        )
-
-        # Static characteristic with a string value.
-        c2 = gatt.UTF8CharacteristicAdapter(
-            gatt.Characteristic(
-                CHARACTERISTIC_UUID_BASE + "02",
-                gatt.Characteristic.Properties.READ
-                | gatt.Characteristic.Properties.WRITE,
-                gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-                'hello',
-            )
-        )
-
-        # Static characteristic with a tuple value.
-        c3 = gatt.PackedCharacteristicAdapter(
-            gatt.Characteristic(
-                CHARACTERISTIC_UUID_BASE + "03",
-                gatt.Characteristic.Properties.READ
-                | gatt.Characteristic.Properties.WRITE,
-                gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-                (1007, 1008, 1009),
-            ),
-            ">III",
-        )
-
-        # Static characteristic with a named tuple value.
-        c4 = gatt.MappedCharacteristicAdapter(
-            gatt.Characteristic(
-                CHARACTERISTIC_UUID_BASE + "04",
-                gatt.Characteristic.Properties.READ
-                | gatt.Characteristic.Properties.WRITE,
-                gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-                {"f1": 3007, "f2": 3008, "f3": 3009},
-            ),
-            ">III",
-            ["f1", "f2", "f3"],
-        )
-
-        # Static characteristic with a serializable value.
-        c5 = gatt.SerializableCharacteristicAdapter(
-            gatt.Characteristic(
-                CHARACTERISTIC_UUID_BASE + "05",
-                gatt.Characteristic.Properties.READ
-                | gatt.Characteristic.Properties.WRITE,
-                gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-                CustomSerializableClass(11, 12),
-            ),
-            CustomSerializableClass,
-        )
-
-        # Static characteristic with a delegated value.
-        c6 = gatt.DelegatedCharacteristicAdapter(
-            gatt.Characteristic(
-                CHARACTERISTIC_UUID_BASE + "06",
-                gatt.Characteristic.Properties.READ
-                | gatt.Characteristic.Properties.WRITE,
-                gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-                CustomClass(1, 2),
-            ),
-            encode=CustomClass.encode,
-            decode=CustomClass.decode,
-        )
-
-        # Dynamic characteristic with a bytes value.
-        c7 = gatt.Characteristic(
-            CHARACTERISTIC_UUID_BASE + "07",
-            gatt.Characteristic.Properties.READ | gatt.Characteristic.Properties.WRITE,
-            gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-            gatt.CharacteristicValue(
-                read=lambda connection: dynamic_read("bytes"),
-                write=lambda connection, value: dynamic_write("bytes", value),
-            ),
-        )
-
-        # Dynamic characteristic with a string value.
-        c8 = gatt.UTF8CharacteristicAdapter(
-            gatt.Characteristic(
-                CHARACTERISTIC_UUID_BASE + "08",
-                gatt.Characteristic.Properties.READ
-                | gatt.Characteristic.Properties.WRITE,
-                gatt.Characteristic.READABLE | gatt.Characteristic.WRITEABLE,
-                gatt.CharacteristicValue(
-                    read=lambda connection: dynamic_read("string"),
-                    write=lambda connection, value: dynamic_write("string", value),
-                ),
-            )
-        )
-
-        characteristics: List[
-            Union[gatt.Characteristic, gatt.CharacteristicAdapter]
-        ] = [c1, c2, c3, c4, c5, c6, c7, c8]
-
-        # Listen for read and write events.
-        for characteristic in characteristics:
-            characteristic.on(
-                "read",
-                lambda _, value, c=characteristic: on_characteristic_read(c, value),
-            )
-            characteristic.on(
-                "write",
-                lambda _, value, c=characteristic: on_characteristic_write(c, value),
-            )
-
-        device.add_service(gatt.Service(SERVICE_UUID, characteristics))  # type: ignore
-
         # Get things going
         await device.power_on()
 
-        # Connect to a peer
+        if is_client:
-        if len(sys.argv) > 2:
+            # Connect a client to a peer
-            await client(device, hci.Address(sys.argv[2]))
+            await client(device, hci.Address(DEFAULT_SERVER_ADDRESS))
         else:
+            # Advertise so a peer can connect
             await device.start_advertising(auto_restart=True)
 
+            # Setup a server
+            await server(device)
+
         await hci_transport.source.wait_for_termination()
 
 

pyproject.toml

@@ -57,7 +57,7 @@ development = [
 ]
 avatar = [
     "pandora-avatar == 0.0.10",
-    "rootcanal == 1.10.0 ; python_version>='3.10'",
+    "rootcanal == 1.11.1 ; python_version>='3.10'",
 ]
 pandora = ["bt-test-interfaces >= 0.0.6"]
 documentation = [
@@ -66,7 +66,7 @@ documentation = [
     "mkdocstrings[python] >= 0.27.0",
 ]
 auracast = [
-    "lc3py ; python_version>='3.10' and platform_system=='Linux' and platform_machine=='x86_64'",
+    "lc3py >= 1.1.3; python_version>='3.10' and ((platform_system=='Linux' and platform_machine=='x86_64') or (platform_system=='Darwin' and platform_machine=='arm64'))",
     "sounddevice >= 0.5.1",
 ]
 

tests/device_test.py

@@ -24,7 +24,6 @@ import pytest
 from bumble.core import (
     BT_BR_EDR_TRANSPORT,
     BT_LE_TRANSPORT,
-    BT_PERIPHERAL_ROLE,
     ConnectionParameters,
 )
 from bumble.device import (
@@ -43,6 +42,7 @@ from bumble.hci import (
     HCI_CONNECTION_FAILED_TO_BE_ESTABLISHED_ERROR,
     Address,
     OwnAddressType,
+    Role,
     HCI_Command_Complete_Event,
     HCI_Command_Status_Event,
     HCI_Connection_Complete_Event,
@@ -295,7 +295,7 @@ async def test_legacy_advertising_disconnection(auto_restart):
         peer_address,
         None,
         None,
-        BT_PERIPHERAL_ROLE,
+        Role.PERIPHERAL,
         ConnectionParameters(0, 0, 0),
     )
 
@@ -353,7 +353,7 @@ async def test_extended_advertising_connection(own_address_type):
         peer_address,
         None,
         None,
-        BT_PERIPHERAL_ROLE,
+        Role.PERIPHERAL,
         ConnectionParameters(0, 0, 0),
     )
     device.on_advertising_set_termination(
@@ -397,7 +397,7 @@ async def test_extended_advertising_connection_out_of_order(own_address_type):
         Address('F0:F1:F2:F3:F4:F5'),
         None,
         None,
-        BT_PERIPHERAL_ROLE,
+        Role.PERIPHERAL,
         ConnectionParameters(0, 0, 0),
     )
 

tests/gatt_service_test.py

@@ -16,6 +16,7 @@
 # Imports
 # -----------------------------------------------------------------------------
 from __future__ import annotations
+import pytest
 
 from . import test_utils
 
@@ -25,6 +26,7 @@ from bumble.profiles import gatt_service
 
 
 # -----------------------------------------------------------------------------
+@pytest.mark.asyncio
 async def test_database_hash():
     devices = await test_utils.TwoDevices.create_with_connection()
     devices[0].gatt_server.services.clear()
@@ -118,6 +120,7 @@ async def test_database_hash():
 
 
 # -----------------------------------------------------------------------------
+@pytest.mark.asyncio
 async def test_service_changed():
     devices = await test_utils.TwoDevices.create_with_connection()
     assert (service := devices[0].gatt_service)

tests/gatt_test.py

@@ -17,32 +17,43 @@
 # -----------------------------------------------------------------------------
 from __future__ import annotations
 import asyncio
+import enum
 import logging
 import os
 import struct
 import pytest
+from typing import Any
 from typing_extensions import Self
 from unittest.mock import AsyncMock, Mock, ANY
 
 from bumble.controller import Controller
-from bumble.gatt_client import CharacteristicProxy
 from bumble.link import LocalLink
 from bumble.device import Device, Peer
 from bumble.host import Host
 from bumble.gatt import (
     GATT_BATTERY_LEVEL_CHARACTERISTIC,
     GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
-    CharacteristicAdapter,
-    SerializableCharacteristicAdapter,
-    DelegatedCharacteristicAdapter,
-    PackedCharacteristicAdapter,
-    MappedCharacteristicAdapter,
-    UTF8CharacteristicAdapter,
     Service,
     Characteristic,
     CharacteristicValue,
     Descriptor,
 )
+from bumble.gatt_client import CharacteristicProxy
+from bumble.gatt_adapters import (
+    CharacteristicProxyAdapter,
+    SerializableCharacteristicAdapter,
+    SerializableCharacteristicProxyAdapter,
+    DelegatedCharacteristicAdapter,
+    DelegatedCharacteristicProxyAdapter,
+    PackedCharacteristicAdapter,
+    PackedCharacteristicProxyAdapter,
+    MappedCharacteristicAdapter,
+    MappedCharacteristicProxyAdapter,
+    UTF8CharacteristicAdapter,
+    UTF8CharacteristicProxyAdapter,
+    EnumCharacteristicAdapter,
+    EnumCharacteristicProxyAdapter,
+)
 from bumble.transport import AsyncPipeSink
 from bumble.core import UUID
 from bumble.att import (
@@ -199,7 +210,7 @@ async def test_characteristic_encoding():
     await async_barrier()
     assert characteristic.value == bytes([125])
 
-    cd = DelegatedCharacteristicAdapter(c, encode=lambda x: bytes([x // 2]))
+    cd = DelegatedCharacteristicProxyAdapter(c, encode=lambda x: bytes([x // 2]))
     await cd.write_value(100, with_response=True)
     await async_barrier()
     assert characteristic.value == bytes([50])
@@ -207,7 +218,7 @@ async def test_characteristic_encoding():
     c2 = peer.get_characteristics_by_uuid(async_characteristic.uuid)
     assert len(c2) == 1
     c2 = c2[0]
-    cd2 = PackedCharacteristicAdapter(c2, ">I")
+    cd2 = PackedCharacteristicProxyAdapter(c2, ">I")
     cd2v = await cd2.read_value()
     assert cd2v == 0x05060708
 
@@ -249,7 +260,7 @@ async def test_characteristic_encoding():
     await async_barrier()
     assert last_change is None
 
-    cd = DelegatedCharacteristicAdapter(c, decode=lambda x: x[0])
+    cd = DelegatedCharacteristicProxyAdapter(c, decode=lambda x: x[0])
     await cd.subscribe(on_change)
     await server.notify_subscribers(characteristic)
     await async_barrier()
@@ -314,21 +325,16 @@ async def test_attribute_getters():
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_CharacteristicAdapter() -> None:
-    # Check that the CharacteristicAdapter base class is transparent
     v = bytes([1, 2, 3])
-    c = Characteristic(
+    c: Characteristic[Any] = Characteristic(
         GATT_BATTERY_LEVEL_CHARACTERISTIC,
         Characteristic.Properties.READ,
         Characteristic.READABLE,
         v,
     )
-    a = CharacteristicAdapter(c)
-
-    value = await a.read_value(None)
-    assert value == v
 
     v = bytes([3, 4, 5])
-    await a.write_value(None, v)
+    await c.write_value(None, v)
     assert c.value == v
 
     # Simple delegated adapter
@@ -415,11 +421,171 @@ async def test_CharacteristicAdapter() -> None:
 
     class_read_value = await class_c.read_value(None)
     assert class_read_value == class_value_bytes
-    c.value = b''
+    class_c.value = b''
     await class_c.write_value(None, class_value_bytes)
-    assert isinstance(c.value, BlaBla)
+    assert isinstance(class_c.value, BlaBla)
-    assert c.value.a == 3
+    assert class_c.value.a == class_value.a
-    assert c.value.b == 4
+    assert class_c.value.b == class_value.b
+
+    # Enum adapter
+    class MyEnum(enum.IntEnum):
+        ENUM_1 = 1234
+        ENUM_2 = 5678
+
+    enum_value = MyEnum.ENUM_2
+    enum_value_bytes = int(enum_value).to_bytes(3, 'big')
+    c.value = enum_value
+    enum_c = EnumCharacteristicAdapter(c, MyEnum, 3, 'big')
+    enum_read_value = await enum_c.read_value(None)
+    assert enum_read_value == enum_value_bytes
+    enum_c.value = b''
+    await enum_c.write_value(None, enum_value_bytes)
+    assert isinstance(enum_c.value, MyEnum)
+    assert enum_c.value == enum_value
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_CharacteristicProxyAdapter() -> None:
+    class Client:
+        def __init__(self, value):
+            self.value = value
+
+        async def read_value(self, handle, no_long_read=False) -> bytes:
+            return self.value
+
+        async def write_value(self, handle, value, with_response=False):
+            self.value = value
+
+    class TestAttributeProxy(CharacteristicProxy):
+        def __init__(self, value) -> None:
+            super().__init__(Client(value), 0, 0, None, 0)  # type: ignore
+
+        @property
+        def value(self):
+            return self.client.value
+
+        @value.setter
+        def value(self, value):
+            self.client.value = value
+
+    v = bytes([1, 2, 3])
+    c = TestAttributeProxy(v)
+    a: CharacteristicProxyAdapter = CharacteristicProxyAdapter(c)
+
+    value = await a.read_value()
+    assert value == v
+
+    v = bytes([3, 4, 5])
+    await a.write_value(v)
+    assert c.value == v
+
+    # Simple delegated adapter
+    delegated = DelegatedCharacteristicProxyAdapter(
+        c, lambda x: bytes(reversed(x)), lambda x: bytes(reversed(x))
+    )
+
+    delegated_value = await delegated.read_value()
+    assert delegated_value == bytes(reversed(v))
+
+    delegated_value2 = bytes([3, 4, 5])
+    await delegated.write_value(delegated_value2)
+    assert c.value == bytes(reversed(delegated_value2))
+
+    # Packed adapter with single element format
+    packed_value_ref = 1234
+    packed_value_bytes = struct.pack('>H', packed_value_ref)
+    c.value = packed_value_bytes
+    packed = PackedCharacteristicProxyAdapter(c, '>H')
+
+    packed_value_read = await packed.read_value()
+    assert packed_value_read == packed_value_ref
+    c.value = None
+    await packed.write_value(packed_value_ref)
+    assert c.value == packed_value_bytes
+
+    # Packed adapter with multi-element format
+    v1 = 1234
+    v2 = 5678
+    packed_multi_value_bytes = struct.pack('>HH', v1, v2)
+    c.value = packed_multi_value_bytes
+    packed_multi = PackedCharacteristicProxyAdapter(c, '>HH')
+
+    packed_multi_read_value = await packed_multi.read_value()
+    assert packed_multi_read_value == (v1, v2)
+    c.value = b''
+    await packed_multi.write_value((v1, v2))
+    assert c.value == packed_multi_value_bytes
+
+    # Mapped adapter
+    v1 = 1234
+    v2 = 5678
+    packed_mapped_value_bytes = struct.pack('>HH', v1, v2)
+    mapped = {'v1': v1, 'v2': v2}
+    c.value = packed_mapped_value_bytes
+    packed_mapped = MappedCharacteristicProxyAdapter(c, '>HH', ('v1', 'v2'))
+
+    packed_mapped_read_value = await packed_mapped.read_value()
+    assert packed_mapped_read_value == mapped
+    c.value = b''
+    await packed_mapped.write_value(mapped)
+    assert c.value == packed_mapped_value_bytes
+
+    # UTF-8 adapter
+    string_value = 'Hello π'
+    string_value_bytes = string_value.encode('utf-8')
+    c.value = string_value_bytes
+    string_c = UTF8CharacteristicProxyAdapter(c)
+
+    string_read_value = await string_c.read_value()
+    assert string_read_value == string_value
+    c.value = b''
+    await string_c.write_value(string_value)
+    assert c.value == string_value_bytes
+
+    # Class adapter
+    class BlaBla:
+        def __init__(self, a: int, b: int) -> None:
+            self.a = a
+            self.b = b
+
+        @classmethod
+        def from_bytes(cls, data: bytes) -> Self:
+            a, b = struct.unpack(">II", data)
+            return cls(a, b)
+
+        def __bytes__(self) -> bytes:
+            return struct.pack(">II", self.a, self.b)
+
+    class_value = BlaBla(3, 4)
+    class_value_bytes = struct.pack(">II", 3, 4)
+    c.value = class_value_bytes
+    class_c = SerializableCharacteristicProxyAdapter(c, BlaBla)
+
+    class_read_value = await class_c.read_value()
+    assert isinstance(class_read_value, BlaBla)
+    assert class_read_value.a == class_value.a
+    assert class_read_value.b == class_value.b
+    c.value = b''
+    await class_c.write_value(class_value)
+    assert c.value == class_value_bytes
+
+    # Enum adapter
+    class MyEnum(enum.IntEnum):
+        ENUM_1 = 1234
+        ENUM_2 = 5678
+
+    enum_value = MyEnum.ENUM_1
+    enum_value_bytes = int(enum_value).to_bytes(3, 'little')
+    c.value = enum_value_bytes
+    enum_c = EnumCharacteristicProxyAdapter(c, MyEnum, 3)
+
+    enum_read_value = await enum_c.read_value()
+    assert isinstance(enum_read_value, MyEnum)
+    assert enum_read_value == enum_value
+    c.value = b''
+    await enum_c.write_value(enum_value)
+    assert c.value == enum_value_bytes
 
 
 # -----------------------------------------------------------------------------
@@ -601,7 +767,7 @@ async def test_read_write2():
     v1 = await c1.read_value()
     assert v1 == v
 
-    a1 = PackedCharacteristicAdapter(c1, '>I')
+    a1 = PackedCharacteristicProxyAdapter(c1, '>I')
     v1 = await a1.read_value()
     assert v1 == struct.unpack('>I', v)[0]
 
@@ -1114,6 +1280,7 @@ async def async_main():
     await test_CharacteristicValue()
     await test_CharacteristicValue_async()
     await test_CharacteristicAdapter()
+    await test_CharacteristicProxyAdapter()
 
 
 # -----------------------------------------------------------------------------

tests/gmap_test.py

@@ -78,7 +78,11 @@ async def test_init_service(gmap_client: GamingAudioServiceProxy):
         | GmapRole.BROADCAST_GAME_RECEIVER
         | GmapRole.BROADCAST_GAME_SENDER
     )
+    assert gmap_client.ugg_features is not None
     assert await gmap_client.ugg_features.read_value() == UggFeatures.UGG_MULTISINK
+    assert gmap_client.ugt_features is not None
     assert await gmap_client.ugt_features.read_value() == UgtFeatures.UGT_SOURCE
+    assert gmap_client.bgr_features is not None
     assert await gmap_client.bgr_features.read_value() == BgrFeatures.BGR_MULTISINK
+    assert gmap_client.bgs_features is not None
     assert await gmap_client.bgs_features.read_value() == BgsFeatures.BGS_96_KBPS

tests/self_test.py

@@ -24,7 +24,7 @@ import pytest
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from bumble.controller import Controller
-from bumble.core import BT_BR_EDR_TRANSPORT, BT_PERIPHERAL_ROLE, BT_CENTRAL_ROLE
+from bumble.core import BT_BR_EDR_TRANSPORT, BT_LE_TRANSPORT
 from bumble.link import LocalLink
 from bumble.device import Device, Peer
 from bumble.host import Host
@@ -39,6 +39,7 @@ from bumble.smp import (
 )
 from bumble.core import ProtocolError
 from bumble.keys import PairingKeys
+from bumble.hci import Role
 
 
 # -----------------------------------------------------------------------------
@@ -111,7 +112,7 @@ async def test_self_connection():
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     'responder_role,',
-    (BT_CENTRAL_ROLE, BT_PERIPHERAL_ROLE),
+    (Role.CENTRAL, Role.PERIPHERAL),
 )
 async def test_self_classic_connection(responder_role):
     # Create two devices, each with a controller, attached to the same link