Add a GATT adapter that uses from_bytes and __bytes__ as conversion methods.

apps/auracast.py

@@ -60,7 +60,7 @@ AURACAST_DEFAULT_ATT_MTU = 256
 class BroadcastScanner(pyee.EventEmitter):
     @dataclasses.dataclass
     class Broadcast(pyee.EventEmitter):
-        name: str | None
+        name: str
         sync: bumble.device.PeriodicAdvertisingSync
         rssi: int = 0
         public_broadcast_announcement: Optional[
@@ -135,8 +135,7 @@ class BroadcastScanner(pyee.EventEmitter):
                 self.sync.advertiser_address,
                 color(self.sync.state.name, 'green'),
             )
-            if self.name is not None:
-                print(f'  {color("Name", "cyan")}:         {self.name}')
+            print(f'  {color("Name", "cyan")}:         {self.name}')
             if self.appearance:
                 print(f'  {color("Appearance", "cyan")}:   {str(self.appearance)}')
             print(f'  {color("RSSI", "cyan")}:         {self.rssi}')
@@ -175,7 +174,7 @@ class BroadcastScanner(pyee.EventEmitter):
                     print(color('      Codec ID:', 'yellow'))
                     print(
                         color('        Coding Format:           ', 'green'),
-                        subgroup.codec_id.codec_id.name,
+                        subgroup.codec_id.coding_format.name,
                     )
                     print(
                         color('        Company ID:              ', 'green'),
@@ -275,24 +274,13 @@ class BroadcastScanner(pyee.EventEmitter):
         await self.device.stop_scanning()
 
     def on_advertisement(self, advertisement: bumble.device.Advertisement) -> None:
-        if not (
-            ads := advertisement.data.get_all(
-                bumble.core.AdvertisingData.SERVICE_DATA_16_BIT_UUID
+        if (
+            broadcast_name := advertisement.data.get(
+                bumble.core.AdvertisingData.BROADCAST_NAME
             )
-        ) or not (
-            any(
-                ad
-                for ad in ads
-                if isinstance(ad, tuple)
-                and ad[0] == bumble.gatt.GATT_BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE
-            )
-        ):
+        ) is None:
             return
-
-        broadcast_name = advertisement.data.get(
-            bumble.core.AdvertisingData.BROADCAST_NAME
-        )
-        assert isinstance(broadcast_name, str) or broadcast_name is None
+        assert isinstance(broadcast_name, str)
 
         if broadcast := self.broadcasts.get(advertisement.address):
             broadcast.update(advertisement)
@@ -303,7 +291,7 @@ class BroadcastScanner(pyee.EventEmitter):
         )
 
     async def on_new_broadcast(
-        self, name: str | None, advertisement: bumble.device.Advertisement
+        self, name: str, advertisement: bumble.device.Advertisement
     ) -> None:
         periodic_advertising_sync = await self.device.create_periodic_advertising_sync(
             advertiser_address=advertisement.address,
@@ -311,7 +299,10 @@ class BroadcastScanner(pyee.EventEmitter):
             sync_timeout=self.sync_timeout,
             filter_duplicates=self.filter_duplicates,
         )
-        broadcast = self.Broadcast(name, periodic_advertising_sync)
+        broadcast = self.Broadcast(
+            name,
+            periodic_advertising_sync,
+        )
         broadcast.update(advertisement)
         self.broadcasts[advertisement.address] = broadcast
         periodic_advertising_sync.on('loss', lambda: self.on_broadcast_loss(broadcast))

apps/hci_bridge.py

@@ -83,7 +83,7 @@ async def async_main():
                         return_parameters=bytes([hci.HCI_SUCCESS]),
                     )
                     # Return a packet with 'respond to sender' set to True
-                    return (bytes(response), True)
+                    return (response.to_bytes(), True)
 
                 return None
 

apps/lea_unicast/app.py

@@ -486,12 +486,7 @@ class Speaker:
 
             def on_pdu(pdu: HCI_IsoDataPacket, ase: ascs.AseStateMachine):
                 codec_config = ase.codec_specific_configuration
-                if (
-                    not isinstance(codec_config, bap.CodecSpecificConfiguration)
-                    or codec_config.frame_duration is None
-                    or codec_config.audio_channel_allocation is None
-                ):
-                    return
+                assert isinstance(codec_config, bap.CodecSpecificConfiguration)
                 pcm = decode(
                     codec_config.frame_duration.us,
                     codec_config.audio_channel_allocation.channel_count,
@@ -500,17 +495,11 @@ class Speaker:
                 self.device.abort_on('disconnection', self.ui_server.send_audio(pcm))
 
             def on_ase_state_change(ase: ascs.AseStateMachine) -> None:
-                codec_config = ase.codec_specific_configuration
                 if ase.state == ascs.AseStateMachine.State.STREAMING:
+                    codec_config = ase.codec_specific_configuration
+                    assert isinstance(codec_config, bap.CodecSpecificConfiguration)
+                    assert ase.cis_link
                     if ase.role == ascs.AudioRole.SOURCE:
-                        if (
-                            not isinstance(codec_config, bap.CodecSpecificConfiguration)
-                            or ase.cis_link is None
-                            or codec_config.octets_per_codec_frame is None
-                            or codec_config.frame_duration is None
-                            or codec_config.codec_frames_per_sdu is None
-                        ):
-                            return
                         ase.cis_link.abort_on(
                             'disconnection',
                             lc3_source_task(
@@ -525,17 +514,10 @@ class Speaker:
                             ),
                         )
                     else:
-                        if not ase.cis_link:
-                            return
                         ase.cis_link.sink = functools.partial(on_pdu, ase=ase)
                 elif ase.state == ascs.AseStateMachine.State.CODEC_CONFIGURED:
-                    if (
-                        not isinstance(codec_config, bap.CodecSpecificConfiguration)
-                        or codec_config.sampling_frequency is None
-                        or codec_config.frame_duration is None
-                        or codec_config.audio_channel_allocation is None
-                    ):
-                        return
+                    codec_config = ase.codec_specific_configuration
+                    assert isinstance(codec_config, bap.CodecSpecificConfiguration)
                     if ase.role == ascs.AudioRole.SOURCE:
                         setup_encoders(
                             codec_config.sampling_frequency.hz,

bumble/att.py

@@ -291,6 +291,9 @@ class ATT_PDU:
     def init_from_bytes(self, pdu, offset):
         return HCI_Object.init_from_bytes(self, pdu, offset, self.fields)
 
+    def to_bytes(self):
+        return self.pdu
+
     @property
     def is_command(self):
         return ((self.op_code >> 6) & 1) == 1
@@ -300,7 +303,7 @@ class ATT_PDU:
         return ((self.op_code >> 7) & 1) == 1
 
     def __bytes__(self):
-        return self.pdu
+        return self.to_bytes()
 
     def __str__(self):
         result = color(self.name, 'yellow')
@@ -756,13 +759,13 @@ class AttributeValue:
     def __init__(
         self,
         read: Union[
-            Callable[[Optional[Connection]], bytes],
-            Callable[[Optional[Connection]], Awaitable[bytes]],
+            Callable[[Optional[Connection]], Any],
+            Callable[[Optional[Connection]], Awaitable[Any]],
             None,
         ] = None,
         write: Union[
-            Callable[[Optional[Connection], bytes], None],
-            Callable[[Optional[Connection], bytes], Awaitable[None]],
+            Callable[[Optional[Connection], Any], None],
+            Callable[[Optional[Connection], Any], Awaitable[None]],
             None,
         ] = None,
     ):
@@ -821,13 +824,13 @@ class Attribute(EventEmitter):
     READ_REQUIRES_AUTHORIZATION = Permissions.READ_REQUIRES_AUTHORIZATION
     WRITE_REQUIRES_AUTHORIZATION = Permissions.WRITE_REQUIRES_AUTHORIZATION
 
-    value: Union[bytes, AttributeValue]
+    value: Any
 
     def __init__(
         self,
         attribute_type: Union[str, bytes, UUID],
         permissions: Union[str, Attribute.Permissions],
-        value: Union[str, bytes, AttributeValue] = b'',
+        value: Any = b'',
     ) -> None:
         EventEmitter.__init__(self)
         self.handle = 0
@@ -845,11 +848,7 @@ class Attribute(EventEmitter):
         else:
             self.type = attribute_type
 
-        # Convert the value to a byte array
-        if isinstance(value, str):
-            self.value = bytes(value, 'utf-8')
-        else:
-            self.value = value
+        self.value = value
 
     def encode_value(self, value: Any) -> bytes:
         return value
@@ -892,6 +891,8 @@ class Attribute(EventEmitter):
         else:
             value = self.value
 
+        self.emit('read', connection, value)
+
         return self.encode_value(value)
 
     async def write_value(self, connection: Connection, value_bytes: bytes) -> None:

bumble/controller.py

@@ -314,7 +314,7 @@ class Controller:
             f'{color("CONTROLLER -> HOST", "green")}: {packet}'
         )
         if self.host:
-            self.host.on_packet(bytes(packet))
+            self.host.on_packet(packet.to_bytes())
 
     # This method allows the controller to emulate the same API as a transport source
     async def wait_for_termination(self):
@@ -1192,7 +1192,7 @@ class Controller:
         See Bluetooth spec Vol 4, Part E - 7.4.6 Read BD_ADDR Command
         '''
         bd_addr = (
-            bytes(self._public_address)
+            self._public_address.to_bytes()
             if self._public_address is not None
             else bytes(6)
         )
@@ -1543,41 +1543,6 @@ class Controller:
         }
         return bytes([HCI_SUCCESS])
 
-    def on_hci_le_set_advertising_set_random_address_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.52 LE Set Advertising Set Random Address
-        Command
-        '''
-        return bytes([HCI_SUCCESS])
-
-    def on_hci_le_set_extended_advertising_parameters_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.53 LE Set Extended Advertising Parameters
-        Command
-        '''
-        return bytes([HCI_SUCCESS, 0])
-
-    def on_hci_le_set_extended_advertising_data_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.54 LE Set Extended Advertising Data
-        Command
-        '''
-        return bytes([HCI_SUCCESS])
-
-    def on_hci_le_set_extended_scan_response_data_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.55 LE Set Extended Scan Response Data
-        Command
-        '''
-        return bytes([HCI_SUCCESS])
-
-    def on_hci_le_set_extended_advertising_enable_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.56 LE Set Extended Advertising Enable
-        Command
-        '''
-        return bytes([HCI_SUCCESS])
-
     def on_hci_le_read_maximum_advertising_data_length_command(self, _command):
         '''
         See Bluetooth spec Vol 4, Part E - 7.8.57 LE Read Maximum Advertising Data
@@ -1592,27 +1557,6 @@ class Controller:
         '''
         return struct.pack('<BB', HCI_SUCCESS, 0xF0)
 
-    def on_hci_le_set_periodic_advertising_parameters_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.61 LE Set Periodic Advertising Parameters
-        Command
-        '''
-        return bytes([HCI_SUCCESS])
-
-    def on_hci_le_set_periodic_advertising_data_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.62 LE Set Periodic Advertising Data
-        Command
-        '''
-        return bytes([HCI_SUCCESS])
-
-    def on_hci_le_set_periodic_advertising_enable_command(self, _command):
-        '''
-        See Bluetooth spec Vol 4, Part E - 7.8.63 LE Set Periodic Advertising Enable
-        Command
-        '''
-        return bytes([HCI_SUCCESS])
-
     def on_hci_le_read_transmit_power_command(self, _command):
         '''
         See Bluetooth spec Vol 4, Part E - 7.8.74 LE Read Transmit Power Command

bumble/device.py

@@ -557,15 +557,8 @@ class AdvertisingParameters:
 # -----------------------------------------------------------------------------
 @dataclass
 class PeriodicAdvertisingParameters:
-    periodic_advertising_interval_min: int = DEVICE_DEFAULT_ADVERTISING_INTERVAL
-    periodic_advertising_interval_max: int = DEVICE_DEFAULT_ADVERTISING_INTERVAL
-    periodic_advertising_properties: (
-        hci.HCI_LE_Set_Periodic_Advertising_Parameters_Command.Properties
-    ) = field(
-        default_factory=lambda: hci.HCI_LE_Set_Periodic_Advertising_Parameters_Command.Properties(
-            0
-        )
-    )
+    # TODO implement this class
+    pass
 
 
 # -----------------------------------------------------------------------------
@@ -582,7 +575,6 @@ class AdvertisingSet(EventEmitter):
     periodic_advertising_data: bytes
     selected_tx_power: int = 0
     enabled: bool = False
-    periodic_enabled: bool = False
 
     def __post_init__(self) -> None:
         super().__init__()
@@ -611,7 +603,7 @@ class AdvertisingSet(EventEmitter):
                     int(advertising_parameters.primary_advertising_interval_min / 0.625)
                 ),
                 primary_advertising_interval_max=(
-                    int(advertising_parameters.primary_advertising_interval_max / 0.625)
+                    int(advertising_parameters.primary_advertising_interval_min / 0.625)
                 ),
                 primary_advertising_channel_map=int(
                     advertising_parameters.primary_advertising_channel_map
@@ -679,26 +671,10 @@ class AdvertisingSet(EventEmitter):
     async def set_periodic_advertising_parameters(
         self, advertising_parameters: PeriodicAdvertisingParameters
     ) -> None:
-        await self.device.send_command(
-            hci.HCI_LE_Set_Periodic_Advertising_Parameters_Command(
-                advertising_handle=self.advertising_handle,
-                periodic_advertising_interval_min=advertising_parameters.periodic_advertising_interval_min,
-                periodic_advertising_interval_max=advertising_parameters.periodic_advertising_interval_max,
-                periodic_advertising_properties=advertising_parameters.periodic_advertising_properties,
-            ),
-            check_result=True,
-        )
         self.periodic_advertising_parameters = advertising_parameters
 
     async def set_periodic_advertising_data(self, advertising_data: bytes) -> None:
-        await self.device.send_command(
-            hci.HCI_LE_Set_Periodic_Advertising_Data_Command(
-                advertising_handle=self.advertising_handle,
-                operation=hci.HCI_LE_Set_Extended_Advertising_Data_Command.Operation.COMPLETE_DATA,
-                advertising_data=advertising_data,
-            ),
-            check_result=True,
-        )
+        # TODO: send command
         self.periodic_advertising_data = advertising_data
 
     async def set_random_address(self, random_address: hci.Address) -> None:
@@ -736,6 +712,17 @@ class AdvertisingSet(EventEmitter):
 
         self.emit('start')
 
+    async def start_periodic(self, include_adi: bool = False) -> None:
+        await self.device.send_command(
+            hci.HCI_LE_Set_Periodic_Advertising_Enable_Command(
+                enable=1 | (2 if include_adi else 0),
+                advertising_handles=self.advertising_handle,
+            ),
+            check_result=True,
+        )
+
+        self.emit('start_periodic')
+
     async def stop(self) -> None:
         await self.device.send_command(
             hci.HCI_LE_Set_Extended_Advertising_Enable_Command(
@@ -750,31 +737,14 @@ class AdvertisingSet(EventEmitter):
 
         self.emit('stop')
 
-    async def start_periodic(self, include_adi: bool = False) -> None:
-        if self.periodic_enabled:
-            return
-        await self.device.send_command(
-            hci.HCI_LE_Set_Periodic_Advertising_Enable_Command(
-                enable=1 | (2 if include_adi else 0),
-                advertising_handle=self.advertising_handle,
-            ),
-            check_result=True,
-        )
-        self.periodic_enabled = True
-
-        self.emit('start_periodic')
-
     async def stop_periodic(self) -> None:
-        if not self.periodic_enabled:
-            return
         await self.device.send_command(
             hci.HCI_LE_Set_Periodic_Advertising_Enable_Command(
                 enable=0,
-                advertising_handle=self.advertising_handle,
+                advertising_handles=self.advertising_handle,
             ),
             check_result=True,
         )
-        self.periodic_enabled = False
 
         self.emit('stop_periodic')
 
@@ -1572,10 +1542,6 @@ class DeviceConfiguration:
                 )
             )
 
-        # Load scan response data
-        if scan_response_data := config.pop('scan_response_data', None):
-            self.scan_response_data = bytes.fromhex(scan_response_data)
-
         # Load advertising interval (for backward compatibility)
         if advertising_interval := config.pop('advertising_interval', None):
             self.advertising_interval_min = advertising_interval
@@ -1986,7 +1952,7 @@ class Device(CompositeEventEmitter):
         check_address_type: bool = False,
     ) -> Optional[Connection]:
         for connection in self.connections.values():
-            if bytes(connection.peer_address) == bytes(bd_addr):
+            if connection.peer_address.to_bytes() == bd_addr.to_bytes():
                 if (
                     check_address_type
                     and connection.peer_address.address_type != bd_addr.address_type
@@ -2494,27 +2460,14 @@ class Device(CompositeEventEmitter):
         if advertising_parameters is None:
             advertising_parameters = AdvertisingParameters()
 
-        if periodic_advertising_data and periodic_advertising_parameters is None:
-            periodic_advertising_parameters = PeriodicAdvertisingParameters()
-
         if (
             not advertising_parameters.advertising_event_properties.is_legacy
             and advertising_data
             and scan_response_data
         ):
             raise InvalidArgumentError(
-                "Extended advertisements can't have both data and scan response data"
-            )
-
-        if periodic_advertising_parameters and (
-            advertising_parameters.advertising_event_properties.is_connectable
-            or advertising_parameters.advertising_event_properties.is_scannable
-            or advertising_parameters.advertising_event_properties.is_anonymous
-            or advertising_parameters.advertising_event_properties.is_legacy
-        ):
-            raise InvalidArgumentError(
-                "Periodic advertising set cannot be connectable, scannable, anonymous,"
-                "or legacy"
+                "Extended advertisements can't have both data and scan \
+                              response data"
             )
 
         # Allocate a new handle
@@ -2569,14 +2522,12 @@ class Device(CompositeEventEmitter):
                 await advertising_set.set_scan_response_data(scan_response_data)
 
             if periodic_advertising_parameters:
-                await advertising_set.set_periodic_advertising_parameters(
-                    periodic_advertising_parameters
-                )
+                # TODO: call LE Set Periodic Advertising Parameters command
+                raise NotImplementedError('periodic advertising not yet supported')
 
             if periodic_advertising_data:
-                await advertising_set.set_periodic_advertising_data(
-                    periodic_advertising_data
-                )
+                # TODO: call LE Set Periodic Advertising Data command
+                raise NotImplementedError('periodic advertising not yet supported')
 
         except hci.HCI_Error as error:
             # Remove the advertising set so that it doesn't stay dangling in the

bumble/gatt.py

@@ -28,12 +28,15 @@ import functools
 import logging
 import struct
 from typing import (
+    Any,
     Callable,
     Dict,
     Iterable,
     List,
     Optional,
     Sequence,
+    SupportsBytes,
+    Type,
     Union,
     TYPE_CHECKING,
 )
@@ -41,6 +44,7 @@ from typing import (
 from bumble.colors import color
 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
@@ -343,7 +347,7 @@ class Service(Attribute):
     def __init__(
         self,
         uuid: Union[str, UUID],
-        characteristics: List[Characteristic],
+        characteristics: Iterable[Characteristic],
         primary=True,
         included_services: Iterable[Service] = (),
     ) -> None:
@@ -362,7 +366,7 @@ class Service(Attribute):
         )
         self.uuid = uuid
         self.included_services = list(included_services)
-        self.characteristics = characteristics[:]
+        self.characteristics = list(characteristics)
         self.primary = primary
 
     def get_advertising_data(self) -> Optional[bytes]:
@@ -393,7 +397,7 @@ class TemplateService(Service):
 
     def __init__(
         self,
-        characteristics: List[Characteristic],
+        characteristics: Iterable[Characteristic],
         primary: bool = True,
         included_services: Iterable[Service] = (),
     ) -> None:
@@ -410,7 +414,7 @@ class IncludedServiceDeclaration(Attribute):
 
     def __init__(self, service: Service) -> None:
         declaration_bytes = struct.pack(
-            '<HH2s', service.handle, service.end_group_handle, bytes(service.uuid)
+            '<HH2s', service.handle, service.end_group_handle, service.uuid.to_bytes()
         )
         super().__init__(
             GATT_INCLUDE_ATTRIBUTE_TYPE, Attribute.READABLE, declaration_bytes
@@ -490,7 +494,7 @@ class Characteristic(Attribute):
         uuid: Union[str, bytes, UUID],
         properties: Characteristic.Properties,
         permissions: Union[str, Attribute.Permissions],
-        value: Union[str, bytes, CharacteristicValue] = b'',
+        value: Any = b'',
         descriptors: Sequence[Descriptor] = (),
     ):
         super().__init__(uuid, permissions, value)
@@ -525,7 +529,11 @@ class CharacteristicDeclaration(Attribute):
 
     characteristic: Characteristic
 
-    def __init__(self, characteristic: Characteristic, value_handle: int) -> None:
+    def __init__(
+        self,
+        characteristic: Characteristic,
+        value_handle: int,
+    ) -> None:
         declaration_bytes = (
             struct.pack('<BH', characteristic.properties, value_handle)
             + characteristic.uuid.to_pdu_bytes()
@@ -705,7 +713,7 @@ class MappedCharacteristicAdapter(PackedCharacteristicAdapter):
     '''
     Adapter that packs/unpacks characteristic values according to a standard
     Python `struct` format.
-    The adapted `read_value` and `write_value` methods return/accept aa dictionary which
+    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.
     '''
@@ -735,6 +743,24 @@ class UTF8CharacteristicAdapter(CharacteristicAdapter):
         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_client.py

@@ -292,7 +292,7 @@ class Client:
         logger.debug(
             f'GATT Command from client: [0x{self.connection.handle:04X}] {command}'
         )
-        self.send_gatt_pdu(bytes(command))
+        self.send_gatt_pdu(command.to_bytes())
 
     async def send_request(self, request: ATT_PDU):
         logger.debug(
@@ -310,7 +310,7 @@ class Client:
             self.pending_request = request
 
             try:
-                self.send_gatt_pdu(bytes(request))
+                self.send_gatt_pdu(request.to_bytes())
                 response = await asyncio.wait_for(
                     self.pending_response, GATT_REQUEST_TIMEOUT
                 )
@@ -328,7 +328,7 @@ class Client:
             f'GATT Confirmation from client: [0x{self.connection.handle:04X}] '
             f'{confirmation}'
         )
-        self.send_gatt_pdu(bytes(confirmation))
+        self.send_gatt_pdu(confirmation.to_bytes())
 
     async def request_mtu(self, mtu: int) -> int:
         # Check the range

bumble/gatt_server.py

@@ -28,7 +28,17 @@ import asyncio
 import logging
 from collections import defaultdict
 import struct
-from typing import List, Tuple, Optional, TypeVar, Type, Dict, Iterable, TYPE_CHECKING
+from typing import (
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Tuple,
+    TypeVar,
+    Type,
+    Union,
+    TYPE_CHECKING,
+)
 from pyee import EventEmitter
 
 from bumble.colors import color
@@ -68,6 +78,7 @@ from bumble.gatt import (
     GATT_REQUEST_TIMEOUT,
     GATT_SECONDARY_SERVICE_ATTRIBUTE_TYPE,
     Characteristic,
+    CharacteristicAdapter,
     CharacteristicDeclaration,
     CharacteristicValue,
     IncludedServiceDeclaration,
@@ -353,7 +364,7 @@ class Server(EventEmitter):
         logger.debug(
             f'GATT Response from server: [0x{connection.handle:04X}] {response}'
         )
-        self.send_gatt_pdu(connection.handle, bytes(response))
+        self.send_gatt_pdu(connection.handle, response.to_bytes())
 
     async def notify_subscriber(
         self,
@@ -450,7 +461,7 @@ class Server(EventEmitter):
             )
 
             try:
-                self.send_gatt_pdu(connection.handle, bytes(indication))
+                self.send_gatt_pdu(connection.handle, indication.to_bytes())
                 await asyncio.wait_for(pending_confirmation, GATT_REQUEST_TIMEOUT)
             except asyncio.TimeoutError as error:
                 logger.warning(color('!!! GATT Indicate timeout', 'red'))

bumble/hci.py

@@ -1482,7 +1482,7 @@ class CodingFormat:
     vendor_specific_codec_id: int = 0
 
     @classmethod
-    def parse_from_bytes(cls, data: bytes, offset: int) -> tuple[int, CodingFormat]:
+    def parse_from_bytes(cls, data: bytes, offset: int):
         (codec_id, company_id, vendor_specific_codec_id) = struct.unpack_from(
             '<BHH', data, offset
         )
@@ -1492,15 +1492,14 @@ class CodingFormat:
             vendor_specific_codec_id=vendor_specific_codec_id,
         )
 
-    @classmethod
-    def from_bytes(cls, data: bytes) -> CodingFormat:
-        return cls.parse_from_bytes(data, 0)[1]
-
-    def __bytes__(self) -> bytes:
+    def to_bytes(self) -> bytes:
         return struct.pack(
             '<BHH', self.codec_id, self.company_id, self.vendor_specific_codec_id
         )
 
+    def __bytes__(self) -> bytes:
+        return self.to_bytes()
+
 
 # -----------------------------------------------------------------------------
 class HCI_Constant:
@@ -1717,7 +1716,7 @@ class HCI_Object:
             field_length = len(field_bytes)
             field_bytes = bytes([field_length]) + field_bytes
         elif isinstance(field_value, (bytes, bytearray)) or hasattr(
-            field_value, '__bytes__'
+            field_value, 'to_bytes'
         ):
             field_bytes = bytes(field_value)
             if isinstance(field_type, int) and 4 < field_type <= 256:
@@ -1762,7 +1761,7 @@ class HCI_Object:
     def from_bytes(cls, data, offset, fields):
         return cls(fields, **cls.dict_from_bytes(data, offset, fields))
 
-    def __bytes__(self):
+    def to_bytes(self):
         return HCI_Object.dict_to_bytes(self.__dict__, self.fields)
 
     @staticmethod
@@ -1857,6 +1856,9 @@ class HCI_Object:
             for field_name, field_value in field_strings
         )
 
+    def __bytes__(self):
+        return self.to_bytes()
+
     def __init__(self, fields, **kwargs):
         self.fields = fields
         self.init_from_fields(self, fields, kwargs)
@@ -2031,6 +2033,9 @@ class Address:
     def is_static(self):
         return self.is_random and (self.address_bytes[5] >> 6 == 3)
 
+    def to_bytes(self):
+        return self.address_bytes
+
     def to_string(self, with_type_qualifier=True):
         '''
         String representation of the address, MSB first, with an optional type
@@ -2042,7 +2047,7 @@ class Address:
         return result + '/P'
 
     def __bytes__(self):
-        return self.address_bytes
+        return self.to_bytes()
 
     def __hash__(self):
         return hash(self.address_bytes)
@@ -2248,13 +2253,16 @@ class HCI_Command(HCI_Packet):
         self.op_code = op_code
         self.parameters = parameters
 
-    def __bytes__(self):
+    def to_bytes(self):
         parameters = b'' if self.parameters is None else self.parameters
         return (
             struct.pack('<BHB', HCI_COMMAND_PACKET, self.op_code, len(parameters))
             + parameters
         )
 
+    def __bytes__(self):
+        return self.to_bytes()
+
     def __str__(self):
         result = color(self.name, 'green')
         if fields := getattr(self, 'fields', None):
@@ -4319,61 +4327,6 @@ class HCI_LE_Clear_Advertising_Sets_Command(HCI_Command):
     '''
 
 
-# -----------------------------------------------------------------------------
-@HCI_Command.command(
-    [
-        ('advertising_handle', 1),
-        ('periodic_advertising_interval_min', 2),
-        ('periodic_advertising_interval_max', 2),
-        ('periodic_advertising_properties', 2),
-    ]
-)
-class HCI_LE_Set_Periodic_Advertising_Parameters_Command(HCI_Command):
-    '''
-    See Bluetooth spec @ 7.8.61 LE Set Periodic Advertising Parameters command
-    '''
-
-    class Properties(enum.IntFlag):
-        INCLUDE_TX_POWER = 1 << 6
-
-    advertising_handle: int
-    periodic_advertising_interval_min: int
-    periodic_advertising_interval_max: int
-    periodic_advertising_properties: int
-
-
-# -----------------------------------------------------------------------------
-@HCI_Command.command(
-    [
-        ('advertising_handle', 1),
-        (
-            'operation',
-            {
-                'size': 1,
-                'mapper': lambda x: HCI_LE_Set_Extended_Advertising_Data_Command.Operation(
-                    x
-                ).name,
-            },
-        ),
-        (
-            'advertising_data',
-            {
-                'parser': HCI_Object.parse_length_prefixed_bytes,
-                'serializer': HCI_Object.serialize_length_prefixed_bytes,
-            },
-        ),
-    ]
-)
-class HCI_LE_Set_Periodic_Advertising_Data_Command(HCI_Command):
-    '''
-    See Bluetooth spec @ 7.8.62 LE Set Periodic Advertising Data command
-    '''
-
-    advertising_handle: int
-    operation: int
-    advertising_data: bytes
-
-
 # -----------------------------------------------------------------------------
 @HCI_Command.command([('enable', 1), ('advertising_handle', 1)])
 class HCI_LE_Set_Periodic_Advertising_Enable_Command(HCI_Command):
@@ -5178,10 +5131,13 @@ class HCI_Event(HCI_Packet):
         self.event_code = event_code
         self.parameters = parameters
 
-    def __bytes__(self):
+    def to_bytes(self):
         parameters = b'' if self.parameters is None else self.parameters
         return bytes([HCI_EVENT_PACKET, self.event_code, len(parameters)]) + parameters
 
+    def __bytes__(self):
+        return self.to_bytes()
+
     def __str__(self):
         result = color(self.name, 'magenta')
         if fields := getattr(self, 'fields', None):
@@ -6732,7 +6688,7 @@ class HCI_AclDataPacket(HCI_Packet):
             connection_handle, pb_flag, bc_flag, data_total_length, data
         )
 
-    def __bytes__(self):
+    def to_bytes(self):
         h = (self.pb_flag << 12) | (self.bc_flag << 14) | self.connection_handle
         return (
             struct.pack('<BHH', HCI_ACL_DATA_PACKET, h, self.data_total_length)
@@ -6746,6 +6702,9 @@ class HCI_AclDataPacket(HCI_Packet):
         self.data_total_length = data_total_length
         self.data = data
 
+    def __bytes__(self):
+        return self.to_bytes()
+
     def __str__(self):
         return (
             f'{color("ACL", "blue")}: '
@@ -6779,7 +6738,7 @@ class HCI_SynchronousDataPacket(HCI_Packet):
             connection_handle, packet_status, data_total_length, data
         )
 
-    def __bytes__(self) -> bytes:
+    def to_bytes(self) -> bytes:
         h = (self.packet_status << 12) | self.connection_handle
         return (
             struct.pack('<BHB', HCI_SYNCHRONOUS_DATA_PACKET, h, self.data_total_length)
@@ -6798,6 +6757,9 @@ class HCI_SynchronousDataPacket(HCI_Packet):
         self.data_total_length = data_total_length
         self.data = data
 
+    def __bytes__(self) -> bytes:
+        return self.to_bytes()
+
     def __str__(self) -> str:
         return (
             f'{color("SCO", "blue")}: '
@@ -6870,6 +6832,9 @@ class HCI_IsoDataPacket(HCI_Packet):
         )
 
     def __bytes__(self) -> bytes:
+        return self.to_bytes()
+
+    def to_bytes(self) -> bytes:
         fmt = '<BHH'
         args = [
             HCI_ISO_DATA_PACKET,

bumble/hfp.py

@@ -141,7 +141,7 @@ class HfFeature(enum.IntFlag):
     """
     HF supported features (AT+BRSF=) (normative).
 
-    Hands-Free Profile v1.9, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
+    Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
     """
 
     EC_NR = 0x001  # Echo Cancel & Noise reduction
@@ -155,14 +155,14 @@ class HfFeature(enum.IntFlag):
     HF_INDICATORS = 0x100
     ESCO_S4_SETTINGS_SUPPORTED = 0x200
     ENHANCED_VOICE_RECOGNITION_STATUS = 0x400
-    VOICE_RECOGNITION_TEXT = 0x800
+    VOICE_RECOGNITION_TEST = 0x800
 
 
 class AgFeature(enum.IntFlag):
     """
     AG supported features (+BRSF:) (normative).
 
-    Hands-Free Profile v1.9, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
+    Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
     """
 
     THREE_WAY_CALLING = 0x001
@@ -178,7 +178,7 @@ class AgFeature(enum.IntFlag):
     HF_INDICATORS = 0x400
     ESCO_S4_SETTINGS_SUPPORTED = 0x800
     ENHANCED_VOICE_RECOGNITION_STATUS = 0x1000
-    VOICE_RECOGNITION_TEXT = 0x2000
+    VOICE_RECOGNITION_TEST = 0x2000
 
 
 class AudioCodec(enum.IntEnum):
@@ -1390,7 +1390,6 @@ class AgProtocol(pyee.EventEmitter):
 
     def _on_bac(self, *args) -> None:
         self.supported_audio_codecs = [AudioCodec(int(value)) for value in args]
-        self.emit('supported_audio_codecs', self.supported_audio_codecs)
         self.send_ok()
 
     def _on_bcs(self, codec: bytes) -> None:
@@ -1619,7 +1618,7 @@ class ProfileVersion(enum.IntEnum):
     """
     Profile version (normative).
 
-    Hands-Free Profile v1.8, 6.3 SDP Interoperability Requirements.
+    Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements.
     """
 
     V1_5 = 0x0105
@@ -1633,7 +1632,7 @@ class HfSdpFeature(enum.IntFlag):
     """
     HF supported features (normative).
 
-    Hands-Free Profile v1.9, 6.3 SDP Interoperability Requirements.
+    Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements.
     """
 
     EC_NR = 0x01  # Echo Cancel & Noise reduction
@@ -1641,17 +1640,16 @@ class HfSdpFeature(enum.IntFlag):
     CLI_PRESENTATION_CAPABILITY = 0x04
     VOICE_RECOGNITION_ACTIVATION = 0x08
     REMOTE_VOLUME_CONTROL = 0x10
-    WIDE_BAND_SPEECH = 0x20
+    WIDE_BAND = 0x20  # Wide band speech
     ENHANCED_VOICE_RECOGNITION_STATUS = 0x40
-    VOICE_RECOGNITION_TEXT = 0x80
-    SUPER_WIDE_BAND = 0x100
+    VOICE_RECOGNITION_TEST = 0x80
 
 
 class AgSdpFeature(enum.IntFlag):
     """
     AG supported features (normative).
 
-    Hands-Free Profile v1.9, 6.3 SDP Interoperability Requirements.
+    Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements.
     """
 
     THREE_WAY_CALLING = 0x01
@@ -1659,10 +1657,9 @@ class AgSdpFeature(enum.IntFlag):
     VOICE_RECOGNITION_FUNCTION = 0x04
     IN_BAND_RING_TONE_CAPABILITY = 0x08
     VOICE_TAG = 0x10  # Attach a number to voice tag
-    WIDE_BAND_SPEECH = 0x20
+    WIDE_BAND = 0x20  # Wide band speech
     ENHANCED_VOICE_RECOGNITION_STATUS = 0x40
-    VOICE_RECOGNITION_TEXT = 0x80
-    SUPER_WIDE_BAND_SPEED_SPEECH = 0x100
+    VOICE_RECOGNITION_TEST = 0x80
 
 
 def make_hf_sdp_records(
@@ -1695,11 +1692,11 @@ def make_hf_sdp_records(
         in configuration.supported_hf_features
     ):
         hf_supported_features |= HfSdpFeature.ENHANCED_VOICE_RECOGNITION_STATUS
-    if HfFeature.VOICE_RECOGNITION_TEXT in configuration.supported_hf_features:
-        hf_supported_features |= HfSdpFeature.VOICE_RECOGNITION_TEXT
+    if HfFeature.VOICE_RECOGNITION_TEST in configuration.supported_hf_features:
+        hf_supported_features |= HfSdpFeature.VOICE_RECOGNITION_TEST
 
     if AudioCodec.MSBC in configuration.supported_audio_codecs:
-        hf_supported_features |= HfSdpFeature.WIDE_BAND_SPEECH
+        hf_supported_features |= HfSdpFeature.WIDE_BAND
 
     return [
         sdp.ServiceAttribute(
@@ -1775,14 +1772,14 @@ def make_ag_sdp_records(
         in configuration.supported_ag_features
     ):
         ag_supported_features |= AgSdpFeature.ENHANCED_VOICE_RECOGNITION_STATUS
-    if AgFeature.VOICE_RECOGNITION_TEXT in configuration.supported_ag_features:
-        ag_supported_features |= AgSdpFeature.VOICE_RECOGNITION_TEXT
+    if AgFeature.VOICE_RECOGNITION_TEST in configuration.supported_ag_features:
+        ag_supported_features |= AgSdpFeature.VOICE_RECOGNITION_TEST
     if AgFeature.IN_BAND_RING_TONE_CAPABILITY in configuration.supported_ag_features:
         ag_supported_features |= AgSdpFeature.IN_BAND_RING_TONE_CAPABILITY
     if AgFeature.VOICE_RECOGNITION_FUNCTION in configuration.supported_ag_features:
         ag_supported_features |= AgSdpFeature.VOICE_RECOGNITION_FUNCTION
     if AudioCodec.MSBC in configuration.supported_audio_codecs:
-        ag_supported_features |= AgSdpFeature.WIDE_BAND_SPEECH
+        ag_supported_features |= AgSdpFeature.WIDE_BAND
 
     return [
         sdp.ServiceAttribute(

bumble/host.py

@@ -199,7 +199,7 @@ class Host(AbortableEventEmitter):
         check_address_type: bool = False,
     ) -> Optional[Connection]:
         for connection in self.connections.values():
-            if bytes(connection.peer_address) == bytes(bd_addr):
+            if connection.peer_address.to_bytes() == bd_addr.to_bytes():
                 if (
                     check_address_type
                     and connection.peer_address.address_type != bd_addr.address_type

bumble/l2cap.py

@@ -225,7 +225,7 @@ class L2CAP_PDU:
 
         return L2CAP_PDU(l2cap_pdu_cid, l2cap_pdu_payload)
 
-    def __bytes__(self) -> bytes:
+    def to_bytes(self) -> bytes:
         header = struct.pack('<HH', len(self.payload), self.cid)
         return header + self.payload
 
@@ -233,6 +233,9 @@ class L2CAP_PDU:
         self.cid = cid
         self.payload = payload
 
+    def __bytes__(self) -> bytes:
+        return self.to_bytes()
+
     def __str__(self) -> str:
         return f'{color("L2CAP", "green")} [CID={self.cid}]: {self.payload.hex()}'
 
@@ -330,9 +333,12 @@ class L2CAP_Control_Frame:
     def init_from_bytes(self, pdu, offset):
         return HCI_Object.init_from_bytes(self, pdu, offset, self.fields)
 
-    def __bytes__(self) -> bytes:
+    def to_bytes(self) -> bytes:
         return self.pdu
 
+    def __bytes__(self) -> bytes:
+        return self.to_bytes()
+
     def __str__(self) -> str:
         result = f'{color(self.name, "yellow")} [ID={self.identifier}]'
         if fields := getattr(self, 'fields', None):

bumble/profiles/aics.py

@@ -17,6 +17,7 @@
 # -----------------------------------------------------------------------------
 # Imports
 # -----------------------------------------------------------------------------
+from __future__ import annotations
 import logging
 import struct
 
@@ -28,10 +29,11 @@ from bumble.device import Connection
 from bumble.att import ATT_Error
 from bumble.gatt import (
     Characteristic,
-    DelegatedCharacteristicAdapter,
+    SerializableCharacteristicAdapter,
+    PackedCharacteristicAdapter,
     TemplateService,
     CharacteristicValue,
-    PackedCharacteristicAdapter,
+    UTF8CharacteristicAdapter,
     GATT_AUDIO_INPUT_CONTROL_SERVICE,
     GATT_AUDIO_INPUT_STATE_CHARACTERISTIC,
     GATT_GAIN_SETTINGS_ATTRIBUTE_CHARACTERISTIC,
@@ -95,7 +97,7 @@ class AudioInputStatus(OpenIntEnum):
     Cf. 3.4 Audio Input Status
     '''
 
-    INACTIVE = 0x00
+    INATIVE = 0x00
     ACTIVE = 0x01
 
 
@@ -104,7 +106,7 @@ class AudioInputControlPointOpCode(OpenIntEnum):
     Cf. 3.5.1 Audio Input Control Point procedure requirements
     '''
 
-    SET_GAIN_SETTING = 0x01
+    SET_GAIN_SETTING = 0x00
     UNMUTE = 0x02
     MUTE = 0x03
     SET_MANUAL_GAIN_MODE = 0x04
@@ -154,9 +156,6 @@ class AudioInputState:
             attribute=self.attribute_value, value=bytes(self)
         )
 
-    def on_read(self, _connection: Optional[Connection]) -> bytes:
-        return bytes(self)
-
 
 @dataclass
 class GainSettingsProperties:
@@ -173,7 +172,7 @@ class GainSettingsProperties:
         (gain_settings_unit, gain_settings_minimum, gain_settings_maximum) = (
             struct.unpack('BBB', data)
         )
-        GainSettingsProperties(
+        return GainSettingsProperties(
             gain_settings_unit, gain_settings_minimum, gain_settings_maximum
         )
 
@@ -186,9 +185,6 @@ class GainSettingsProperties:
             ]
         )
 
-    def on_read(self, _connection: Optional[Connection]) -> bytes:
-        return bytes(self)
-
 
 @dataclass
 class AudioInputControlPoint:
@@ -239,7 +235,7 @@ class AudioInputControlPoint:
             or gain_settings_operand
             > self.gain_settings_properties.gain_settings_maximum
         ):
-            logger.error("gain_settings value out of range")
+            logger.error("gain_seetings value out of range")
             raise ATT_Error(ErrorCode.VALUE_OUT_OF_RANGE)
 
         if self.audio_input_state.gain_settings != gain_settings_operand:
@@ -321,21 +317,14 @@ class AudioInputDescription:
     audio_input_description: str = "Bluetooth"
     attribute_value: Optional[CharacteristicValue] = None
 
-    @classmethod
-    def from_bytes(cls, data: bytes):
-        return cls(audio_input_description=data.decode('utf-8'))
+    def on_read(self, _connection: Optional[Connection]) -> str:
+        return self.audio_input_description
 
-    def __bytes__(self) -> bytes:
-        return self.audio_input_description.encode('utf-8')
-
-    def on_read(self, _connection: Optional[Connection]) -> bytes:
-        return self.audio_input_description.encode('utf-8')
-
-    async def on_write(self, connection: Optional[Connection], value: bytes) -> None:
+    async def on_write(self, connection: Optional[Connection], value: str) -> None:
         assert connection
         assert self.attribute_value
 
-        self.audio_input_description = value.decode('utf-8')
+        self.audio_input_description = value
         await connection.device.notify_subscribers(
             attribute=self.attribute_value, value=value
         )
@@ -375,26 +364,29 @@ class AICSService(TemplateService):
             self.audio_input_state, self.gain_settings_properties
         )
 
-        self.audio_input_state_characteristic = DelegatedCharacteristicAdapter(
+        self.audio_input_state_characteristic = SerializableCharacteristicAdapter(
             Characteristic(
                 uuid=GATT_AUDIO_INPUT_STATE_CHARACTERISTIC,
                 properties=Characteristic.Properties.READ
                 | Characteristic.Properties.NOTIFY,
                 permissions=Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
-                value=CharacteristicValue(read=self.audio_input_state.on_read),
+                value=self.audio_input_state,
             ),
-            encode=lambda value: bytes(value),
+            AudioInputState,
         )
         self.audio_input_state.attribute_value = (
             self.audio_input_state_characteristic.value
         )
 
-        self.gain_settings_properties_characteristic = DelegatedCharacteristicAdapter(
-            Characteristic(
-                uuid=GATT_GAIN_SETTINGS_ATTRIBUTE_CHARACTERISTIC,
-                properties=Characteristic.Properties.READ,
-                permissions=Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
-                value=CharacteristicValue(read=self.gain_settings_properties.on_read),
+        self.gain_settings_properties_characteristic = (
+            SerializableCharacteristicAdapter(
+                Characteristic(
+                    uuid=GATT_GAIN_SETTINGS_ATTRIBUTE_CHARACTERISTIC,
+                    properties=Characteristic.Properties.READ,
+                    permissions=Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
+                    value=self.gain_settings_properties,
+                ),
+                GainSettingsProperties,
             )
         )
 
@@ -402,7 +394,7 @@ class AICSService(TemplateService):
             uuid=GATT_AUDIO_INPUT_TYPE_CHARACTERISTIC,
             properties=Characteristic.Properties.READ,
             permissions=Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
-            value=audio_input_type,
+            value=bytes(audio_input_type, 'utf-8'),
         )
 
         self.audio_input_status_characteristic = Characteristic(
@@ -412,18 +404,14 @@ class AICSService(TemplateService):
             value=bytes([self.audio_input_status]),
         )
 
-        self.audio_input_control_point_characteristic = DelegatedCharacteristicAdapter(
-            Characteristic(
-                uuid=GATT_AUDIO_INPUT_CONTROL_POINT_CHARACTERISTIC,
-                properties=Characteristic.Properties.WRITE,
-                permissions=Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION,
-                value=CharacteristicValue(
-                    write=self.audio_input_control_point.on_write
-                ),
-            )
+        self.audio_input_control_point_characteristic = Characteristic(
+            uuid=GATT_AUDIO_INPUT_CONTROL_POINT_CHARACTERISTIC,
+            properties=Characteristic.Properties.WRITE,
+            permissions=Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION,
+            value=CharacteristicValue(write=self.audio_input_control_point.on_write),
         )
 
-        self.audio_input_description_characteristic = DelegatedCharacteristicAdapter(
+        self.audio_input_description_characteristic = UTF8CharacteristicAdapter(
             Characteristic(
                 uuid=GATT_AUDIO_INPUT_DESCRIPTION_CHARACTERISTIC,
                 properties=Characteristic.Properties.READ
@@ -469,8 +457,8 @@ class AICSServiceProxy(ProfileServiceProxy):
             )
         ):
             raise gatt.InvalidServiceError("Audio Input State Characteristic not found")
-        self.audio_input_state = DelegatedCharacteristicAdapter(
-            characteristic=characteristics[0], decode=AudioInputState.from_bytes
+        self.audio_input_state = SerializableCharacteristicAdapter(
+            characteristics[0], AudioInputState
         )
 
         if not (
@@ -481,9 +469,8 @@ class AICSServiceProxy(ProfileServiceProxy):
             raise gatt.InvalidServiceError(
                 "Gain Settings Attribute Characteristic not found"
             )
-        self.gain_settings_properties = PackedCharacteristicAdapter(
-            characteristics[0],
-            'BBB',
+        self.gain_settings_properties = SerializableCharacteristicAdapter(
+            characteristics[0], GainSettingsProperties
         )
 
         if not (
@@ -494,10 +481,7 @@ class AICSServiceProxy(ProfileServiceProxy):
             raise gatt.InvalidServiceError(
                 "Audio Input Status Characteristic not found"
             )
-        self.audio_input_status = PackedCharacteristicAdapter(
-            characteristics[0],
-            'B',
-        )
+        self.audio_input_status = PackedCharacteristicAdapter(characteristics[0], 'B')
 
         if not (
             characteristics := service_proxy.get_characteristics_by_uuid(
@@ -517,4 +501,4 @@ class AICSServiceProxy(ProfileServiceProxy):
             raise gatt.InvalidServiceError(
                 "Audio Input Description Characteristic not found"
             )
-        self.audio_input_description = characteristics[0]
+        self.audio_input_description = UTF8CharacteristicAdapter(characteristics[0])

bumble/profiles/bap.py

@@ -265,7 +265,7 @@ class UnicastServerAdvertisingData:
                         core.AdvertisingData.SERVICE_DATA_16_BIT_UUID,
                         struct.pack(
                             '<2sBIB',
-                            bytes(gatt.GATT_AUDIO_STREAM_CONTROL_SERVICE),
+                            gatt.GATT_AUDIO_STREAM_CONTROL_SERVICE.to_bytes(),
                             self.announcement_type,
                             self.available_audio_contexts,
                             len(self.metadata),
@@ -398,21 +398,18 @@ class CodecSpecificConfiguration:
         OCTETS_PER_FRAME         = 0x04
         CODEC_FRAMES_PER_SDU     = 0x05
 
-    sampling_frequency: SamplingFrequency | None = None
-    frame_duration: FrameDuration | None = None
-    audio_channel_allocation: AudioLocation | None = None
-    octets_per_codec_frame: int | None = None
-    codec_frames_per_sdu: int | None = None
+    sampling_frequency: SamplingFrequency
+    frame_duration: FrameDuration
+    audio_channel_allocation: AudioLocation
+    octets_per_codec_frame: int
+    codec_frames_per_sdu: int
 
     @classmethod
     def from_bytes(cls, data: bytes) -> CodecSpecificConfiguration:
         offset = 0
-        sampling_frequency: SamplingFrequency | None = None
-        frame_duration: FrameDuration | None = None
-        audio_channel_allocation: AudioLocation | None = None
-        octets_per_codec_frame: int | None = None
-        codec_frames_per_sdu: int | None = None
-
+        # Allowed default values.
+        audio_channel_allocation = AudioLocation.NOT_ALLOWED
+        codec_frames_per_sdu = 1
         while offset < len(data):
             length, type = struct.unpack_from('BB', data, offset)
             offset += 2
@@ -430,6 +427,8 @@ class CodecSpecificConfiguration:
             elif type == CodecSpecificConfiguration.Type.CODEC_FRAMES_PER_SDU:
                 codec_frames_per_sdu = value
 
+        # It is expected here that if some fields are missing, an error should be raised.
+        # pylint: disable=possibly-used-before-assignment,used-before-assignment
         return CodecSpecificConfiguration(
             sampling_frequency=sampling_frequency,
             frame_duration=frame_duration,
@@ -439,43 +438,23 @@ class CodecSpecificConfiguration:
         )
 
     def __bytes__(self) -> bytes:
-        return b''.join(
-            [
-                struct.pack(fmt, length, tag, value)
-                for fmt, length, tag, value in [
-                    (
-                        '<BBB',
-                        2,
-                        CodecSpecificConfiguration.Type.SAMPLING_FREQUENCY,
-                        self.sampling_frequency,
-                    ),
-                    (
-                        '<BBB',
-                        2,
-                        CodecSpecificConfiguration.Type.FRAME_DURATION,
-                        self.frame_duration,
-                    ),
-                    (
-                        '<BBI',
-                        5,
-                        CodecSpecificConfiguration.Type.AUDIO_CHANNEL_ALLOCATION,
-                        self.audio_channel_allocation,
-                    ),
-                    (
-                        '<BBH',
-                        3,
-                        CodecSpecificConfiguration.Type.OCTETS_PER_FRAME,
-                        self.octets_per_codec_frame,
-                    ),
-                    (
-                        '<BBB',
-                        2,
-                        CodecSpecificConfiguration.Type.CODEC_FRAMES_PER_SDU,
-                        self.codec_frames_per_sdu,
-                    ),
-                ]
-                if value is not None
-            ]
+        return struct.pack(
+            '<BBBBBBBBIBBHBBB',
+            2,
+            CodecSpecificConfiguration.Type.SAMPLING_FREQUENCY,
+            self.sampling_frequency,
+            2,
+            CodecSpecificConfiguration.Type.FRAME_DURATION,
+            self.frame_duration,
+            5,
+            CodecSpecificConfiguration.Type.AUDIO_CHANNEL_ALLOCATION,
+            self.audio_channel_allocation,
+            3,
+            CodecSpecificConfiguration.Type.OCTETS_PER_FRAME,
+            self.octets_per_codec_frame,
+            2,
+            CodecSpecificConfiguration.Type.CODEC_FRAMES_PER_SDU,
+            self.codec_frames_per_sdu,
         )
 
 
@@ -487,24 +466,6 @@ class BroadcastAudioAnnouncement:
     def from_bytes(cls, data: bytes) -> Self:
         return cls(int.from_bytes(data[:3], 'little'))
 
-    def __bytes__(self) -> bytes:
-        return self.broadcast_id.to_bytes(3, 'little')
-
-    def get_advertising_data(self) -> bytes:
-        return bytes(
-            core.AdvertisingData(
-                [
-                    (
-                        core.AdvertisingData.SERVICE_DATA_16_BIT_UUID,
-                        (
-                            bytes(gatt.GATT_BROADCAST_AUDIO_ANNOUNCEMENT_SERVICE)
-                            + bytes(self)
-                        ),
-                    )
-                ]
-            )
-        )
-
 
 @dataclasses.dataclass
 class BasicAudioAnnouncement:
@@ -513,37 +474,26 @@ class BasicAudioAnnouncement:
         index: int
         codec_specific_configuration: CodecSpecificConfiguration
 
-        def __bytes__(self) -> bytes:
-            codec_specific_configuration_bytes = bytes(
-                self.codec_specific_configuration
-            )
-            return (
-                bytes([self.index, len(codec_specific_configuration_bytes)])
-                + codec_specific_configuration_bytes
-            )
+    @dataclasses.dataclass
+    class CodecInfo:
+        coding_format: hci.CodecID
+        company_id: int
+        vendor_specific_codec_id: int
+
+        @classmethod
+        def from_bytes(cls, data: bytes) -> Self:
+            coding_format = hci.CodecID(data[0])
+            company_id = int.from_bytes(data[1:3], 'little')
+            vendor_specific_codec_id = int.from_bytes(data[3:5], 'little')
+            return cls(coding_format, company_id, vendor_specific_codec_id)
 
     @dataclasses.dataclass
     class Subgroup:
-        codec_id: hci.CodingFormat
+        codec_id: BasicAudioAnnouncement.CodecInfo
         codec_specific_configuration: CodecSpecificConfiguration
         metadata: le_audio.Metadata
         bis: List[BasicAudioAnnouncement.BIS]
 
-        def __bytes__(self) -> bytes:
-            metadata_bytes = bytes(self.metadata)
-            codec_specific_configuration_bytes = bytes(
-                self.codec_specific_configuration
-            )
-            return (
-                bytes([len(self.bis)])
-                + bytes(self.codec_id)
-                + bytes([len(codec_specific_configuration_bytes)])
-                + codec_specific_configuration_bytes
-                + bytes([len(metadata_bytes)])
-                + metadata_bytes
-                + b''.join(map(bytes, self.bis))
-            )
-
     presentation_delay: int
     subgroups: List[BasicAudioAnnouncement.Subgroup]
 
@@ -555,7 +505,7 @@ class BasicAudioAnnouncement:
         for _ in range(data[3]):
             num_bis = data[offset]
             offset += 1
-            codec_id = hci.CodingFormat.from_bytes(data[offset : offset + 5])
+            codec_id = cls.CodecInfo.from_bytes(data[offset : offset + 5])
             offset += 5
             codec_specific_configuration_length = data[offset]
             offset += 1
@@ -599,25 +549,3 @@ class BasicAudioAnnouncement:
             )
 
         return cls(presentation_delay, subgroups)
-
-    def __bytes__(self) -> bytes:
-        return (
-            self.presentation_delay.to_bytes(3, 'little')
-            + bytes([len(self.subgroups)])
-            + b''.join(map(bytes, self.subgroups))
-        )
-
-    def get_advertising_data(self) -> bytes:
-        return bytes(
-            core.AdvertisingData(
-                [
-                    (
-                        core.AdvertisingData.SERVICE_DATA_16_BIT_UUID,
-                        (
-                            bytes(gatt.GATT_BASIC_AUDIO_ANNOUNCEMENT_SERVICE)
-                            + bytes(self)
-                        ),
-                    )
-                ]
-            )
-        )

bumble/profiles/bass.py

@@ -276,10 +276,7 @@ class BroadcastReceiveState:
     subgroups: List[SubgroupInfo]
 
     @classmethod
-    def from_bytes(cls, data: bytes) -> Optional[BroadcastReceiveState]:
-        if not data:
-            return None
-
+    def from_bytes(cls, data: bytes) -> BroadcastReceiveState:
         source_id = data[0]
         _, source_address = hci.Address.parse_address_preceded_by_type(data, 2)
         source_adv_sid = data[8]
@@ -357,7 +354,7 @@ 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.SerializableCharacteristicAdapter]
 
     def __init__(self, service_proxy: gatt_client.ServiceProxy):
         self.service_proxy = service_proxy
@@ -381,8 +378,8 @@ class BroadcastAudioScanServiceProxy(gatt_client.ProfileServiceProxy):
                 "Broadcast Receive State characteristic not found"
             )
         self.broadcast_receive_states = [
-            gatt.DelegatedCharacteristicAdapter(
-                characteristic, decode=BroadcastReceiveState.from_bytes
+            gatt.SerializableCharacteristicAdapter(
+                characteristic, BroadcastReceiveState
             )
             for characteristic in characteristics
         ]

bumble/profiles/device_information_service.py

@@ -64,7 +64,10 @@ class DeviceInformationService(TemplateService):
     ):
         characteristics = [
             Characteristic(
-                uuid, Characteristic.Properties.READ, Characteristic.READABLE, field
+                uuid,
+                Characteristic.Properties.READ,
+                Characteristic.READABLE,
+                bytes(field, 'utf-8'),
             )
             for (field, uuid) in (
                 (manufacturer_name, GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC),

bumble/profiles/heart_rate_service.py

@@ -30,6 +30,7 @@ from ..gatt import (
     TemplateService,
     Characteristic,
     CharacteristicValue,
+    SerializableCharacteristicAdapter,
     DelegatedCharacteristicAdapter,
     PackedCharacteristicAdapter,
 )
@@ -150,15 +151,14 @@ class HeartRateService(TemplateService):
         body_sensor_location=None,
         reset_energy_expended=None,
     ):
-        self.heart_rate_measurement_characteristic = DelegatedCharacteristicAdapter(
+        self.heart_rate_measurement_characteristic = SerializableCharacteristicAdapter(
             Characteristic(
                 GATT_HEART_RATE_MEASUREMENT_CHARACTERISTIC,
                 Characteristic.Properties.NOTIFY,
                 0,
                 CharacteristicValue(read=read_heart_rate_measurement),
             ),
-            # pylint: disable=unnecessary-lambda
-            encode=lambda value: bytes(value),
+            HeartRateService.HeartRateMeasurement,
         )
         characteristics = [self.heart_rate_measurement_characteristic]
 
@@ -204,9 +204,8 @@ class HeartRateServiceProxy(ProfileServiceProxy):
         if characteristics := service_proxy.get_characteristics_by_uuid(
             GATT_HEART_RATE_MEASUREMENT_CHARACTERISTIC
         ):
-            self.heart_rate_measurement = DelegatedCharacteristicAdapter(
-                characteristics[0],
-                decode=HeartRateService.HeartRateMeasurement.from_bytes,
+            self.heart_rate_measurement = SerializableCharacteristicAdapter(
+                characteristics[0], HeartRateService.HeartRateMeasurement
             )
         else:
             self.heart_rate_measurement = None

bumble/sdp.py

@@ -344,6 +344,9 @@ class DataElement:
         ]  # Keep a copy so we can re-serialize to an exact replica
         return result
 
+    def to_bytes(self):
+        return bytes(self)
+
     def __bytes__(self):
         # Return early if we have a cache
         if self.bytes:
@@ -620,9 +623,12 @@ class SDP_PDU:
     def init_from_bytes(self, pdu, offset):
         return HCI_Object.init_from_bytes(self, pdu, offset, self.fields)
 
-    def __bytes__(self):
+    def to_bytes(self):
         return self.pdu
 
+    def __bytes__(self):
+        return self.to_bytes()
+
     def __str__(self):
         result = f'{color(self.name, "blue")} [TID={self.transaction_id}]'
         if fields := getattr(self, 'fields', None):

bumble/smp.py

@@ -298,9 +298,12 @@ class SMP_Command:
     def init_from_bytes(self, pdu: bytes, offset: int) -> None:
         return HCI_Object.init_from_bytes(self, pdu, offset, self.fields)
 
-    def __bytes__(self):
+    def to_bytes(self):
         return self.pdu
 
+    def __bytes__(self):
+        return self.to_bytes()
+
     def __str__(self):
         result = color(self.name, 'yellow')
         if fields := getattr(self, 'fields', None):
@@ -1946,7 +1949,7 @@ class Manager(EventEmitter):
             f'{connection.peer_address}: {command}'
         )
         cid = SMP_BR_CID if connection.transport == BT_BR_EDR_TRANSPORT else SMP_CID
-        connection.send_l2cap_pdu(cid, bytes(command))
+        connection.send_l2cap_pdu(cid, command.to_bytes())
 
     def on_smp_security_request_command(
         self, connection: Connection, request: SMP_Security_Request_Command

bumble/utils.py

@@ -24,17 +24,19 @@ import logging
 import sys
 import warnings
 from typing import (
-    Awaitable,
-    Set,
-    TypeVar,
-    List,
-    Tuple,
-    Callable,
     Any,
+    Awaitable,
+    Callable,
+    List,
     Optional,
+    Protocol,
+    Set,
+    Tuple,
+    TypeVar,
     Union,
     overload,
 )
+from typing_extensions import Self
 
 from pyee import EventEmitter
 
@@ -487,3 +489,16 @@ class OpenIntEnum(enum.IntEnum):
         obj._value_ = value
         obj._name_ = f"{cls.__name__}[{value}]"
         return obj
+
+
+# -----------------------------------------------------------------------------
+class ByteSerializable(Protocol):
+    """
+    Type protocol for classes that can be instantiated from bytes and serialized
+    to bytes.
+    """
+
+    @classmethod
+    def from_bytes(cls, data: bytes) -> Self: ...
+
+    def __bytes__(self) -> bytes: ...

examples/keyboard.py

@@ -282,7 +282,7 @@ async def keyboard_device(device, command):
                         GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
                         Characteristic.Properties.READ,
                         Characteristic.READABLE,
-                        'Bumble',
+                        bytes('Bumble', 'utf-8'),
                     )
                 ],
             ),

examples/run_gatt_server.py

@@ -127,7 +127,7 @@ async def main() -> None:
                     '486F64C6-4B5F-4B3B-8AFF-EDE134A8446A',
                     Characteristic.Properties.READ | Characteristic.Properties.NOTIFY,
                     Characteristic.READABLE,
-                    'hello',
+                    bytes('hello', 'utf-8'),
                 ),
             ],
         )

examples/run_gatt_with_adapters.py

@@ -0,0 +1,319 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# -----------------------------------------------------------------------------
+# Imports
+# -----------------------------------------------------------------------------
+from __future__ import annotations
+import asyncio
+import dataclasses
+import logging
+import os
+import random
+import struct
+import sys
+from typing import Any, List, Union
+
+from bumble.device import Connection, Device, Peer
+from bumble import transport
+from bumble import gatt
+from bumble import hci
+from bumble import core
+
+
+# -----------------------------------------------------------------------------
+SERVICE_UUID = core.UUID("50DB505C-8AC4-4738-8448-3B1D9CC09CC5")
+CHARACTERISTIC_UUID_BASE = "D901B45B-4916-412E-ACCA-0000000000"
+
+
+# -----------------------------------------------------------------------------
+@dataclasses.dataclass
+class CustomSerializableClass:
+    x: int
+    y: int
+
+    @classmethod
+    def from_bytes(cls, data: bytes) -> CustomSerializableClass:
+        return cls(*struct.unpack(">II", data))
+
+    def __bytes__(self) -> bytes:
+        return struct.pack(">II", self.x, self.y)
+
+
+# -----------------------------------------------------------------------------
+@dataclasses.dataclass
+class CustomClass:
+    a: int
+    b: int
+
+    @classmethod
+    def decode(cls, data: bytes) -> CustomClass:
+        return cls(*struct.unpack(">II", data))
+
+    def encode(self) -> bytes:
+        return struct.pack(">II", self.a, self.b)
+
+
+# -----------------------------------------------------------------------------
+async def client(device: Device, address: hci.Address) -> None:
+    print(f'=== Connecting to {address}...')
+    connection = await device.connect(address)
+    print('=== Connected')
+
+    # Discover all characteristics.
+    peer = Peer(connection)
+    print("*** Discovering services and characteristics...")
+    await peer.discover_all()
+    print("*** Discovery complete")
+
+    service = peer.get_services_by_uuid(SERVICE_UUID)[0]
+    characteristics = []
+    for index in range(1, 9):
+        characteristics.append(
+            service.get_characteristics_by_uuid(
+                CHARACTERISTIC_UUID_BASE + f"{index:02X}"
+            )[0]
+        )
+
+    # Read all characteristics as raw bytes.
+    for characteristic in characteristics:
+        value = await characteristic.read_value()
+        print(f"### {characteristic} = {value} ({value.hex()})")
+
+    # Static characteristic with a bytes value.
+    c1 = characteristics[0]
+    c1_value = await c1.read_value()
+    print(f"@@@ C1 {c1} value = {c1_value} (type={type(c1_value)})")
+    await c1.write_value("happy π day".encode("utf-8"))
+
+    # Static characteristic with a string value.
+    c2 = gatt.UTF8CharacteristicAdapter(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")
+
+    # Static characteristic with a tuple value.
+    c3 = gatt.PackedCharacteristicAdapter(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))
+
+    # Static characteristic with a named tuple value.
+    c4 = gatt.MappedCharacteristicAdapter(
+        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})
+
+    # Static characteristic with a serializable value.
+    c5 = gatt.SerializableCharacteristicAdapter(
+        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))
+
+    # Static characteristic with a delegated value.
+    c6 = gatt.DelegatedCharacteristicAdapter(
+        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))
+
+    # Dynamic characteristic with a bytes value.
+    c7 = characteristics[6]
+    c7_value = await c7.read_value()
+    print(f"@@@ C7 {c7} value = {c7_value} (type={type(c7_value)})")
+    await c7.write_value(bytes.fromhex("01020304"))
+
+    # Dynamic characteristic with a string value.
+    c8 = gatt.UTF8CharacteristicAdapter(characteristics[7])
+    c8_value = await c8.read_value()
+    print(f"@@@ C8 {c8} value = {c8_value} (type={type(c8_value)})")
+    await c8.write_value("howdy")
+
+
+# -----------------------------------------------------------------------------
+def dynamic_read(selector: str) -> Union[bytes, str]:
+    if selector == "bytes":
+        print("$$$ Returning random bytes")
+        return random.randbytes(7)
+    elif selector == "string":
+        print("$$$ Returning random string")
+        return random.randbytes(7).hex()
+
+    raise ValueError("invalid selector")
+
+
+# -----------------------------------------------------------------------------
+def dynamic_write(selector: str, value: Any) -> None:
+    print(f"$$$ Received[{selector}]: {value} (type={type(value)})")
+
+
+# -----------------------------------------------------------------------------
+def on_characteristic_read(characteristic: gatt.Characteristic, value: Any) -> None:
+    """Event listener invoked when a characteristic is read."""
+    print(f"<<< READ: {characteristic} -> {value} ({type(value)})")
+
+
+# -----------------------------------------------------------------------------
+def on_characteristic_write(characteristic: gatt.Characteristic, value: Any) -> None:
+    """Event listener invoked when a characteristic is written."""
+    print(f"<<< WRITE: {characteristic} <- {value}  ({type(value)})")
+
+
+# -----------------------------------------------------------------------------
+async def main() -> None:
+    if len(sys.argv) < 2:
+        print("Usage: run_gatt_with_adapters.py <transport-spec> [<bluetooth-address>]")
+        print("example: run_gatt_with_adapters.py usb:0 E1:CA:72:48:C4:E8")
+        return
+
+    async with await transport.open_transport(sys.argv[1]) as hci_transport:
+        # Create a device to manage the host
+        device = Device.with_hci(
+            "Bumble",
+            hci.Address("F0:F1:F2:F3:F4:F5"),
+            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 len(sys.argv) > 2:
+            await client(device, hci.Address(sys.argv[2]))
+        else:
+            await device.start_advertising(auto_restart=True)
+
+        await hci_transport.source.wait_for_termination()
+
+
+# -----------------------------------------------------------------------------
+logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
+asyncio.run(main())

examples/run_hfp_gateway.py

@@ -21,10 +21,10 @@ import sys
 import os
 import io
 import logging
-from typing import Iterable, Optional
-
 import websockets
 
+from typing import Optional
+
 import bumble.core
 from bumble.device import Device, ScoLink
 from bumble.transport import open_transport_or_link
@@ -82,10 +82,6 @@ def on_microphone_volume(level: int):
     send_message(type='microphone_volume', level=level)
 
 
-def on_supported_audio_codecs(codecs: Iterable[hfp.AudioCodec]):
-    send_message(type='supported_audio_codecs', codecs=[codec.name for codec in codecs])
-
-
 def on_sco_state_change(codec: int):
     if codec == hfp.AudioCodec.CVSD:
         sample_rate = 8000
@@ -211,7 +207,6 @@ async def main() -> None:
             ag_protocol = hfp.AgProtocol(dlc, configuration)
             ag_protocol.on('speaker_volume', on_speaker_volume)
             ag_protocol.on('microphone_volume', on_microphone_volume)
-            ag_protocol.on('supported_audio_codecs', on_supported_audio_codecs)
             on_hfp_state_change(True)
             dlc.multiplexer.l2cap_channel.on(
                 'close', lambda: on_hfp_state_change(False)
@@ -246,7 +241,7 @@ async def main() -> None:
             # Pick the first one
             channel, version, hf_sdp_features = hfp_record
             print(f'HF version: {version}')
-            print(f'HF features: {hf_sdp_features.name}')
+            print(f'HF features: {hf_sdp_features}')
 
             # Request authentication
             print('*** Authenticating...')

examples/run_unicast_server.py

@@ -161,13 +161,7 @@ async def main() -> None:
             else:
                 file_output = open(f'{datetime.datetime.now().isoformat()}.lc3', 'wb')
                 codec_configuration = ase.codec_specific_configuration
-                if (
-                    not isinstance(codec_configuration, CodecSpecificConfiguration)
-                    or codec_configuration.sampling_frequency is None
-                    or codec_configuration.audio_channel_allocation is None
-                    or codec_configuration.frame_duration is None
-                ):
-                    return
+                assert isinstance(codec_configuration, CodecSpecificConfiguration)
                 # Write a LC3 header.
                 file_output.write(
                     bytes([0x1C, 0xCC])  # Header.

tests/aics_test.py

@@ -28,6 +28,7 @@ from bumble.profiles.aics import (
     AudioInputState,
     AICSServiceProxy,
     GainMode,
+    GainSettingsProperties,
     AudioInputStatus,
     AudioInputControlPointOpCode,
     ErrorCode,
@@ -82,7 +83,12 @@ async def test_init_service(aics_client: AICSServiceProxy):
         gain_mode=GainMode.MANUAL,
         change_counter=0,
     )
-    assert await aics_client.gain_settings_properties.read_value() == (1, 0, 255)
+    assert (
+        await aics_client.gain_settings_properties.read_value()
+        == GainSettingsProperties(
+            gain_settings_unit=1, gain_settings_minimum=0, gain_settings_maximum=255
+        )
+    )
     assert await aics_client.audio_input_status.read_value() == (
         AudioInputStatus.ACTIVE
     )
@@ -481,12 +487,12 @@ async def test_set_automatic_gain_mode_when_automatic_only(
 @pytest.mark.asyncio
 async def test_audio_input_description_initial_value(aics_client: AICSServiceProxy):
     description = await aics_client.audio_input_description.read_value()
-    assert description.decode('utf-8') == "Bluetooth"
+    assert description == "Bluetooth"
 
 
 @pytest.mark.asyncio
 async def test_audio_input_description_write_and_read(aics_client: AICSServiceProxy):
-    new_description = "Line Input".encode('utf-8')
+    new_description = "Line Input"
 
     await aics_client.audio_input_description.write_value(new_description)
 

tests/bap_test.py

@@ -39,8 +39,6 @@ from bumble.profiles.ascs import (
 )
 from bumble.profiles.bap import (
     AudioLocation,
-    BasicAudioAnnouncement,
-    BroadcastAudioAnnouncement,
     SupportedFrameDuration,
     SupportedSamplingFrequency,
     SamplingFrequency,
@@ -202,56 +200,6 @@ def test_codec_specific_configuration() -> None:
     assert CodecSpecificConfiguration.from_bytes(bytes(config)) == config
 
 
-# -----------------------------------------------------------------------------
-def test_broadcast_audio_announcement() -> None:
-    broadcast_audio_announcement = BroadcastAudioAnnouncement(123456)
-    assert (
-        BroadcastAudioAnnouncement.from_bytes(bytes(broadcast_audio_announcement))
-        == broadcast_audio_announcement
-    )
-
-
-# -----------------------------------------------------------------------------
-def test_basic_audio_announcement() -> None:
-    basic_audio_announcement = BasicAudioAnnouncement(
-        presentation_delay=40000,
-        subgroups=[
-            BasicAudioAnnouncement.Subgroup(
-                codec_id=CodingFormat(codec_id=CodecID.LC3),
-                codec_specific_configuration=CodecSpecificConfiguration(
-                    sampling_frequency=SamplingFrequency.FREQ_48000,
-                    frame_duration=FrameDuration.DURATION_10000_US,
-                    octets_per_codec_frame=100,
-                ),
-                metadata=Metadata(
-                    [
-                        Metadata.Entry(tag=Metadata.Tag.LANGUAGE, data=b'eng'),
-                        Metadata.Entry(tag=Metadata.Tag.PROGRAM_INFO, data=b'Disco'),
-                    ]
-                ),
-                bis=[
-                    BasicAudioAnnouncement.BIS(
-                        index=0,
-                        codec_specific_configuration=CodecSpecificConfiguration(
-                            audio_channel_allocation=AudioLocation.FRONT_LEFT
-                        ),
-                    ),
-                    BasicAudioAnnouncement.BIS(
-                        index=1,
-                        codec_specific_configuration=CodecSpecificConfiguration(
-                            audio_channel_allocation=AudioLocation.FRONT_RIGHT
-                        ),
-                    ),
-                ],
-            )
-        ],
-    )
-    assert (
-        BasicAudioAnnouncement.from_bytes(bytes(basic_audio_announcement))
-        == basic_audio_announcement
-    )
-
-
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_pacs():

tests/device_test.py

@@ -19,7 +19,9 @@ import asyncio
 import functools
 import logging
 import os
+from types import LambdaType
 import pytest
+from unittest import mock
 
 from bumble.core import (
     BT_BR_EDR_TRANSPORT,
@@ -27,13 +29,7 @@ from bumble.core import (
     BT_PERIPHERAL_ROLE,
     ConnectionParameters,
 )
-from bumble.device import (
-    AdvertisingEventProperties,
-    AdvertisingParameters,
-    Connection,
-    Device,
-    PeriodicAdvertisingParameters,
-)
+from bumble.device import AdvertisingParameters, Connection, Device
 from bumble.host import AclPacketQueue, Host
 from bumble.hci import (
     HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
@@ -269,8 +265,7 @@ async def test_flush():
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_legacy_advertising():
-    device = TwoDevices()[0]
-    await device.power_on()
+    device = Device(host=mock.AsyncMock(Host))
 
     # Start advertising
     await device.start_advertising()
@@ -288,10 +283,7 @@ async def test_legacy_advertising():
 )
 @pytest.mark.asyncio
 async def test_legacy_advertising_disconnection(auto_restart):
-    devices = TwoDevices()
-    device = devices[0]
-    devices.controllers[0].le_features = bytes.fromhex('ffffffffffffffff')
-    await device.power_on()
+    device = Device(host=mock.AsyncMock(spec=Host))
     peer_address = Address('F0:F1:F2:F3:F4:F5')
     await device.start_advertising(auto_restart=auto_restart)
     device.on_connection(
@@ -313,11 +305,6 @@ async def test_legacy_advertising_disconnection(auto_restart):
     await async_barrier()
 
     if auto_restart:
-        assert device.legacy_advertising_set
-        started = asyncio.Event()
-        if not device.is_advertising:
-            device.legacy_advertising_set.once('start', started.set)
-            await asyncio.wait_for(started.wait(), _TIMEOUT)
         assert device.is_advertising
     else:
         assert not device.is_advertising
@@ -326,8 +313,7 @@ async def test_legacy_advertising_disconnection(auto_restart):
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_extended_advertising():
-    device = TwoDevices()[0]
-    await device.power_on()
+    device = Device(host=mock.AsyncMock(Host))
 
     # Start advertising
     advertising_set = await device.create_advertising_set()
@@ -346,8 +332,7 @@ async def test_extended_advertising():
 )
 @pytest.mark.asyncio
 async def test_extended_advertising_connection(own_address_type):
-    device = TwoDevices()[0]
-    await device.power_on()
+    device = Device(host=mock.AsyncMock(spec=Host))
     peer_address = Address('F0:F1:F2:F3:F4:F5')
     advertising_set = await device.create_advertising_set(
         advertising_parameters=AdvertisingParameters(own_address_type=own_address_type)
@@ -383,10 +368,8 @@ async def test_extended_advertising_connection(own_address_type):
 )
 @pytest.mark.asyncio
 async def test_extended_advertising_connection_out_of_order(own_address_type):
-    devices = TwoDevices()
-    device = devices[0]
-    devices.controllers[0].le_features = bytes.fromhex('ffffffffffffffff')
-    await device.power_on()
+    device = Device(host=mock.AsyncMock(spec=Host))
+    peer_address = Address('F0:F1:F2:F3:F4:F5')
     advertising_set = await device.create_advertising_set(
         advertising_parameters=AdvertisingParameters(own_address_type=own_address_type)
     )
@@ -399,7 +382,7 @@ async def test_extended_advertising_connection_out_of_order(own_address_type):
     device.on_connection(
         0x0001,
         BT_LE_TRANSPORT,
-        Address('F0:F1:F2:F3:F4:F5'),
+        peer_address,
         None,
         None,
         BT_PERIPHERAL_ROLE,
@@ -414,34 +397,6 @@ async def test_extended_advertising_connection_out_of_order(own_address_type):
     await async_barrier()
 
 
-# -----------------------------------------------------------------------------
-@pytest.mark.asyncio
-async def test_periodic_advertising():
-    device = TwoDevices()[0]
-    await device.power_on()
-
-    # Start advertising
-    advertising_set = await device.create_advertising_set(
-        advertising_parameters=AdvertisingParameters(
-            advertising_event_properties=AdvertisingEventProperties(
-                is_connectable=False
-            )
-        ),
-        advertising_data=b'123',
-        periodic_advertising_parameters=PeriodicAdvertisingParameters(),
-        periodic_advertising_data=b'abc',
-    )
-    assert device.extended_advertising_sets
-    assert advertising_set.enabled
-    assert not advertising_set.periodic_enabled
-
-    await advertising_set.start_periodic()
-    assert advertising_set.periodic_enabled
-
-    await advertising_set.stop_periodic()
-    assert not advertising_set.periodic_enabled
-
-
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_get_remote_le_features():

tests/gatt_test.py

@@ -15,11 +15,13 @@
 # -----------------------------------------------------------------------------
 # Imports
 # -----------------------------------------------------------------------------
+from __future__ import annotations
 import asyncio
 import logging
 import os
 import struct
 import pytest
+from typing_extensions import Self
 from unittest.mock import AsyncMock, Mock, ANY
 
 from bumble.controller import Controller
@@ -31,6 +33,7 @@ from bumble.gatt import (
     GATT_BATTERY_LEVEL_CHARACTERISTIC,
     GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR,
     CharacteristicAdapter,
+    SerializableCharacteristicAdapter,
     DelegatedCharacteristicAdapter,
     PackedCharacteristicAdapter,
     MappedCharacteristicAdapter,
@@ -57,7 +60,7 @@ from .test_utils import async_barrier
 
 # -----------------------------------------------------------------------------
 def basic_check(x):
-    pdu = bytes(x)
+    pdu = x.to_bytes()
     parsed = ATT_PDU.from_bytes(pdu)
     x_str = str(x)
     parsed_str = str(parsed)
@@ -74,7 +77,7 @@ def test_UUID():
     assert str(u) == '61A3512C-09BE-4DDC-A6A6-0B03667AAFC6'
     v = UUID(str(u))
     assert str(v) == '61A3512C-09BE-4DDC-A6A6-0B03667AAFC6'
-    w = UUID.from_bytes(bytes(v))
+    w = UUID.from_bytes(v.to_bytes())
     assert str(w) == '61A3512C-09BE-4DDC-A6A6-0B03667AAFC6'
 
     u1 = UUID.from_16_bits(0x1234)
@@ -310,7 +313,7 @@ async def test_attribute_getters():
 
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
-async def test_CharacteristicAdapter():
+async def test_CharacteristicAdapter() -> None:
     # Check that the CharacteristicAdapter base class is transparent
     v = bytes([1, 2, 3])
     c = Characteristic(
@@ -329,67 +332,94 @@ async def test_CharacteristicAdapter():
     assert c.value == v
 
     # Simple delegated adapter
-    a = DelegatedCharacteristicAdapter(
+    delegated = DelegatedCharacteristicAdapter(
         c, lambda x: bytes(reversed(x)), lambda x: bytes(reversed(x))
     )
 
-    value = await a.read_value(None)
-    assert value == bytes(reversed(v))
+    delegated_value = await delegated.read_value(None)
+    assert delegated_value == bytes(reversed(v))
 
-    v = bytes([3, 4, 5])
-    await a.write_value(None, v)
-    assert a.value == bytes(reversed(v))
+    delegated_value2 = bytes([3, 4, 5])
+    await delegated.write_value(None, delegated_value2)
+    assert delegated.value == bytes(reversed(delegated_value2))
 
     # Packed adapter with single element format
-    v = 1234
-    pv = struct.pack('>H', v)
-    c.value = v
-    a = PackedCharacteristicAdapter(c, '>H')
+    packed_value_ref = 1234
+    packed_value_bytes = struct.pack('>H', packed_value_ref)
+    c.value = packed_value_ref
+    packed = PackedCharacteristicAdapter(c, '>H')
 
-    value = await a.read_value(None)
-    assert value == pv
-    c.value = None
-    await a.write_value(None, pv)
-    assert a.value == v
+    packed_value_read = await packed.read_value(None)
+    assert packed_value_read == packed_value_bytes
+    c.value = b''
+    await packed.write_value(None, packed_value_bytes)
+    assert packed.value == packed_value_ref
 
     # Packed adapter with multi-element format
     v1 = 1234
     v2 = 5678
-    pv = struct.pack('>HH', v1, v2)
+    packed_multi_value_bytes = struct.pack('>HH', v1, v2)
     c.value = (v1, v2)
-    a = PackedCharacteristicAdapter(c, '>HH')
+    packed_multi = PackedCharacteristicAdapter(c, '>HH')
 
-    value = await a.read_value(None)
-    assert value == pv
-    c.value = None
-    await a.write_value(None, pv)
-    assert a.value == (v1, v2)
+    packed_multi_read_value = await packed_multi.read_value(None)
+    assert packed_multi_read_value == packed_multi_value_bytes
+    packed_multi.value = b''
+    await packed_multi.write_value(None, packed_multi_value_bytes)
+    assert packed_multi.value == (v1, v2)
 
     # Mapped adapter
     v1 = 1234
     v2 = 5678
-    pv = struct.pack('>HH', v1, v2)
+    packed_mapped_value_bytes = struct.pack('>HH', v1, v2)
     mapped = {'v1': v1, 'v2': v2}
     c.value = mapped
-    a = MappedCharacteristicAdapter(c, '>HH', ('v1', 'v2'))
+    packed_mapped = MappedCharacteristicAdapter(c, '>HH', ('v1', 'v2'))
 
-    value = await a.read_value(None)
-    assert value == pv
-    c.value = None
-    await a.write_value(None, pv)
-    assert a.value == mapped
+    packed_mapped_read_value = await packed_mapped.read_value(None)
+    assert packed_mapped_read_value == packed_mapped_value_bytes
+    c.value = b''
+    await packed_mapped.write_value(None, packed_mapped_value_bytes)
+    assert packed_mapped.value == mapped
 
     # UTF-8 adapter
-    v = 'Hello π'
-    ev = v.encode('utf-8')
-    c.value = v
-    a = UTF8CharacteristicAdapter(c)
+    string_value = 'Hello π'
+    string_value_bytes = string_value.encode('utf-8')
+    c.value = string_value
+    string_c = UTF8CharacteristicAdapter(c)
 
-    value = await a.read_value(None)
-    assert value == ev
-    c.value = None
-    await a.write_value(None, ev)
-    assert a.value == v
+    string_read_value = await string_c.read_value(None)
+    assert string_read_value == string_value_bytes
+    c.value = b''
+    await string_c.write_value(None, string_value_bytes)
+    assert string_c.value == string_value
+
+    # 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
+    class_c = SerializableCharacteristicAdapter(c, BlaBla)
+
+    class_read_value = await class_c.read_value(None)
+    assert class_read_value == class_value_bytes
+    c.value = b''
+    await class_c.write_value(None, class_value_bytes)
+    assert isinstance(c.value, BlaBla)
+    assert c.value.a == 3
+    assert c.value.b == 4
 
 
 # -----------------------------------------------------------------------------

tests/hci_test.py

@@ -75,13 +75,13 @@ from bumble.hci import (
 
 
 def basic_check(x):
-    packet = bytes(x)
+    packet = x.to_bytes()
     print(packet.hex())
     parsed = HCI_Packet.from_bytes(packet)
     x_str = str(x)
     parsed_str = str(parsed)
     print(x_str)
-    parsed_bytes = bytes(parsed)
+    parsed_bytes = parsed.to_bytes()
     assert x_str == parsed_str
     assert packet == parsed_bytes
 
@@ -188,7 +188,7 @@ def test_HCI_Command_Complete_Event():
         return_parameters=bytes([7]),
     )
     basic_check(event)
-    event = HCI_Packet.from_bytes(bytes(event))
+    event = HCI_Packet.from_bytes(event.to_bytes())
     assert event.return_parameters == 7
 
     # With a simple status as an integer status
@@ -562,7 +562,7 @@ def test_iso_data_packet():
         '6281bc77ed6a3206d984bcdabee6be831c699cb50e2'
     )
 
-    assert bytes(packet) == data
+    assert packet.to_bytes() == data
 
 
 # -----------------------------------------------------------------------------

tests/hfp_test.py

@@ -61,7 +61,7 @@ def _default_hf_configuration() -> hfp.HfConfiguration:
 # -----------------------------------------------------------------------------
 def _default_hf_sdp_features() -> hfp.HfSdpFeature:
     return (
-        hfp.HfSdpFeature.WIDE_BAND_SPEECH
+        hfp.HfSdpFeature.WIDE_BAND
         | hfp.HfSdpFeature.THREE_WAY_CALLING
         | hfp.HfSdpFeature.CLI_PRESENTATION_CAPABILITY
     )
@@ -108,7 +108,7 @@ def _default_ag_configuration() -> hfp.AgConfiguration:
 # -----------------------------------------------------------------------------
 def _default_ag_sdp_features() -> hfp.AgSdpFeature:
     return (
-        hfp.AgSdpFeature.WIDE_BAND_SPEECH
+        hfp.AgSdpFeature.WIDE_BAND
         | hfp.AgSdpFeature.IN_BAND_RING_TONE_CAPABILITY
         | hfp.AgSdpFeature.THREE_WAY_CALLING
     )

tests/self_test.py

@@ -240,7 +240,7 @@ async def test_self_gatt():
     result = await peer.discover_included_services(result[0])
     assert len(result) == 2
     # Service UUID is only present when the UUID is 16-bit Bluetooth UUID
-    assert bytes(result[1].uuid) == bytes(s3.uuid)
+    assert result[1].uuid.to_bytes() == s3.uuid.to_bytes()
 
 
 # -----------------------------------------------------------------------------