Bumble Python API

Classes

Address

Bluetooth Address (see Bluetooth spec Vol 6, Part B - 1.3 DEVICE ADDRESS) NOTE: the address bytes are stored in little-endian byte order here, so address[0] is the LSB of the address, address[5] is the MSB.

Source code in bumble/hci.py

class Address:
    '''
    Bluetooth Address (see Bluetooth spec Vol 6, Part B - 1.3 DEVICE ADDRESS)
    NOTE: the address bytes are stored in little-endian byte order here, so
    address[0] is the LSB of the address, address[5] is the MSB.
    '''

    PUBLIC_DEVICE_ADDRESS = 0x00
    RANDOM_DEVICE_ADDRESS = 0x01
    PUBLIC_IDENTITY_ADDRESS = 0x02
    RANDOM_IDENTITY_ADDRESS = 0x03

    ADDRESS_TYPE_NAMES = {
        PUBLIC_DEVICE_ADDRESS: 'PUBLIC_DEVICE_ADDRESS',
        RANDOM_DEVICE_ADDRESS: 'RANDOM_DEVICE_ADDRESS',
        PUBLIC_IDENTITY_ADDRESS: 'PUBLIC_IDENTITY_ADDRESS',
        RANDOM_IDENTITY_ADDRESS: 'RANDOM_IDENTITY_ADDRESS',
    }

    # Type declarations
    NIL: Address
    ANY: Address
    ANY_RANDOM: Address

    # pylint: disable-next=unnecessary-lambda
    ADDRESS_TYPE_SPEC = {'size': 1, 'mapper': lambda x: Address.address_type_name(x)}

    @staticmethod
    def address_type_name(address_type):
        return name_or_number(Address.ADDRESS_TYPE_NAMES, address_type)

    @staticmethod
    def from_string_for_transport(string, transport):
        if transport == BT_BR_EDR_TRANSPORT:
            address_type = Address.PUBLIC_DEVICE_ADDRESS
        else:
            address_type = Address.RANDOM_DEVICE_ADDRESS
        return Address(string, address_type)

    @staticmethod
    def parse_address(data, offset):
        # Fix the type to a default value. This is used for parsing type-less Classic
        # addresses
        return Address.parse_address_with_type(
            data, offset, Address.PUBLIC_DEVICE_ADDRESS
        )

    @staticmethod
    def parse_address_with_type(data, offset, address_type):
        return offset + 6, Address(data[offset : offset + 6], address_type)

    @staticmethod
    def parse_address_preceded_by_type(data, offset):
        address_type = data[offset - 1]
        return Address.parse_address_with_type(data, offset, address_type)

    def __init__(
        self, address: Union[bytes, str], address_type: int = RANDOM_DEVICE_ADDRESS
    ):
        '''
        Initialize an instance. `address` may be a byte array in little-endian
        format, or a hex string in big-endian format (with optional ':'
        separators between the bytes).
        If the address is a string suffixed with '/P', `address_type` is ignored and
        the type is set to PUBLIC_DEVICE_ADDRESS.
        '''
        if isinstance(address, bytes):
            self.address_bytes = address
        else:
            # Check if there's a '/P' type specifier
            if address.endswith('P'):
                address_type = Address.PUBLIC_DEVICE_ADDRESS
                address = address[:-2]

            if len(address) == 12 + 5:
                # Form with ':' separators
                address = address.replace(':', '')
            self.address_bytes = bytes(reversed(bytes.fromhex(address)))

        if len(self.address_bytes) != 6:
            raise ValueError('invalid address length')

        self.address_type = address_type

    def clone(self):
        return Address(self.address_bytes, self.address_type)

    @property
    def is_public(self):
        return self.address_type in (
            self.PUBLIC_DEVICE_ADDRESS,
            self.PUBLIC_IDENTITY_ADDRESS,
        )

    @property
    def is_random(self):
        return not self.is_public

    @property
    def is_resolved(self):
        return self.address_type in (
            self.PUBLIC_IDENTITY_ADDRESS,
            self.RANDOM_IDENTITY_ADDRESS,
        )

    @property
    def is_resolvable(self):
        return self.address_type == self.RANDOM_DEVICE_ADDRESS and (
            self.address_bytes[5] >> 6 == 1
        )

    @property
    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
        qualifier.
        '''
        result = ':'.join([f'{x:02X}' for x in reversed(self.address_bytes)])
        if not with_type_qualifier or not self.is_public:
            return result
        return result + '/P'

    def __bytes__(self):
        return self.to_bytes()

    def __hash__(self):
        return hash(self.address_bytes)

    def __eq__(self, other):
        return (
            self.address_bytes == other.address_bytes
            and self.is_public == other.is_public
        )

    def __str__(self):
        return self.to_string()

__init__(address, address_type=RANDOM_DEVICE_ADDRESS)

Initialize an instance. address may be a byte array in little-endian format, or a hex string in big-endian format (with optional ':' separators between the bytes). If the address is a string suffixed with '/P', address_type is ignored and the type is set to PUBLIC_DEVICE_ADDRESS.

Source code in bumble/hci.py

def __init__(
    self, address: Union[bytes, str], address_type: int = RANDOM_DEVICE_ADDRESS
):
    '''
    Initialize an instance. `address` may be a byte array in little-endian
    format, or a hex string in big-endian format (with optional ':'
    separators between the bytes).
    If the address is a string suffixed with '/P', `address_type` is ignored and
    the type is set to PUBLIC_DEVICE_ADDRESS.
    '''
    if isinstance(address, bytes):
        self.address_bytes = address
    else:
        # Check if there's a '/P' type specifier
        if address.endswith('P'):
            address_type = Address.PUBLIC_DEVICE_ADDRESS
            address = address[:-2]

        if len(address) == 12 + 5:
            # Form with ':' separators
            address = address.replace(':', '')
        self.address_bytes = bytes(reversed(bytes.fromhex(address)))

    if len(self.address_bytes) != 6:
        raise ValueError('invalid address length')

    self.address_type = address_type

to_string(with_type_qualifier=True)

String representation of the address, MSB first, with an optional type qualifier.

Source code in bumble/hci.py

def to_string(self, with_type_qualifier=True):
    '''
    String representation of the address, MSB first, with an optional type
    qualifier.
    '''
    result = ':'.join([f'{x:02X}' for x in reversed(self.address_bytes)])
    if not with_type_qualifier or not self.is_public:
        return result
    return result + '/P'

HCI_Packet

Abstract Base class for HCI packets

Source code in bumble/hci.py

class HCI_Packet:
    '''
    Abstract Base class for HCI packets
    '''

    hci_packet_type: int

    @staticmethod
    def from_bytes(packet):
        packet_type = packet[0]

        if packet_type == HCI_COMMAND_PACKET:
            return HCI_Command.from_bytes(packet)

        if packet_type == HCI_ACL_DATA_PACKET:
            return HCI_AclDataPacket.from_bytes(packet)

        if packet_type == HCI_EVENT_PACKET:
            return HCI_Event.from_bytes(packet)

        return HCI_CustomPacket(packet)

    def __init__(self, name):
        self.name = name

    def __bytes__(self) -> bytes:
        raise NotImplementedError

    def __repr__(self) -> str:
        return self.name

HCI Commands

HCI_Command

Bases: HCI_Packet

See Bluetooth spec @ Vol 2, Part E - 5.4.1 HCI Command Packet

Source code in bumble/hci.py

class HCI_Command(HCI_Packet):
    '''
    See Bluetooth spec @ Vol 2, Part E - 5.4.1 HCI Command Packet
    '''

    hci_packet_type = HCI_COMMAND_PACKET
    command_classes: Dict[int, Type[HCI_Command]] = {}

    @staticmethod
    def command(fields=(), return_parameters_fields=()):
        '''
        Decorator used to declare and register subclasses
        '''

        def inner(cls):
            cls.name = cls.__name__.upper()
            cls.op_code = key_with_value(HCI_COMMAND_NAMES, cls.name)
            if cls.op_code is None:
                raise KeyError(f'command {cls.name} not found in HCI_COMMAND_NAMES')
            cls.fields = fields
            cls.return_parameters_fields = return_parameters_fields

            # Patch the __init__ method to fix the op_code
            if fields is not None:

                def init(self, parameters=None, **kwargs):
                    return HCI_Command.__init__(self, cls.op_code, parameters, **kwargs)

                cls.__init__ = init

            # Register a factory for this class
            HCI_Command.command_classes[cls.op_code] = cls

            return cls

        return inner

    @staticmethod
    def from_bytes(packet):
        op_code, length = struct.unpack_from('<HB', packet, 1)
        parameters = packet[4:]
        if len(parameters) != length:
            raise ValueError('invalid packet length')

        # Look for a registered class
        cls = HCI_Command.command_classes.get(op_code)
        if cls is None:
            # No class registered, just use a generic instance
            return HCI_Command(op_code, parameters)

        # Create a new instance
        if (fields := getattr(cls, 'fields', None)) is not None:
            self = cls.__new__(cls)
            HCI_Command.__init__(self, op_code, parameters)
            HCI_Object.init_from_bytes(self, parameters, 0, fields)
            return self

        return cls.from_parameters(parameters)

    @staticmethod
    def command_name(op_code):
        name = HCI_COMMAND_NAMES.get(op_code)
        if name is not None:
            return name
        return f'[OGF=0x{op_code >> 10:02x}, OCF=0x{op_code & 0x3FF:04x}]'

    @classmethod
    def create_return_parameters(cls, **kwargs):
        return HCI_Object(cls.return_parameters_fields, **kwargs)

    def __init__(self, op_code, parameters=None, **kwargs):
        super().__init__(HCI_Command.command_name(op_code))
        if (fields := getattr(self, 'fields', None)) and kwargs:
            HCI_Object.init_from_fields(self, fields, kwargs)
            if parameters is None:
                parameters = HCI_Object.dict_to_bytes(kwargs, fields)
        self.op_code = op_code
        self.parameters = parameters

    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):
            result += ':\n' + HCI_Object.format_fields(self.__dict__, fields, '  ')
        else:
            if self.parameters:
                result += f': {self.parameters.hex()}'
        return result

command(fields=(), return_parameters_fields=()) staticmethod

Decorator used to declare and register subclasses

Source code in bumble/hci.py

@staticmethod
def command(fields=(), return_parameters_fields=()):
    '''
    Decorator used to declare and register subclasses
    '''

    def inner(cls):
        cls.name = cls.__name__.upper()
        cls.op_code = key_with_value(HCI_COMMAND_NAMES, cls.name)
        if cls.op_code is None:
            raise KeyError(f'command {cls.name} not found in HCI_COMMAND_NAMES')
        cls.fields = fields
        cls.return_parameters_fields = return_parameters_fields

        # Patch the __init__ method to fix the op_code
        if fields is not None:

            def init(self, parameters=None, **kwargs):
                return HCI_Command.__init__(self, cls.op_code, parameters, **kwargs)

            cls.__init__ = init

        # Register a factory for this class
        HCI_Command.command_classes[cls.op_code] = cls

        return cls

    return inner

HCI_Disconnect_Command

Bases: HCI_Command

See Bluetooth spec @ 7.1.6 Disconnect Command

Source code in bumble/hci.py

@HCI_Command.command(
    [
        ('connection_handle', 2),
        ('reason', {'size': 1, 'mapper': HCI_Constant.error_name}),
    ]
)
class HCI_Disconnect_Command(HCI_Command):
    '''
    See Bluetooth spec @ 7.1.6 Disconnect Command
    '''