Skip to content

Commit

Permalink
Added support for Google´s Bumble Bluetooth Controller stack
Browse files Browse the repository at this point in the history
The backend supports direct use with Bumble. The HCI Controller
is managed by the Bumble stack and the transport layer can
be defined by the user (e.g. VHCI, Serial, TCP, android-netsim).
  • Loading branch information
vChavezB committed Nov 1, 2024
1 parent c98883b commit b76d7f3
Show file tree
Hide file tree
Showing 14 changed files with 725 additions and 1 deletion.
1 change: 1 addition & 0 deletions AUTHORS.rst
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ Contributors
* David Johansen <[email protected]>
* JP Hutchins <[email protected]>
* Bram Duvigneau <[email protected]>
* Victor Chavez <[email protected]>

Sponsors
--------
Expand Down
4 changes: 4 additions & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@ and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0

`Unreleased`_
=============
Added
-----
* Added support for Google's Bumble Bluetooth stack.


`0.22.3`_ (2024-10-05)
======================
Expand Down
32 changes: 32 additions & 0 deletions bleak/backends/bumble/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2024 Victor Chavez
"""Bumble backend."""

from bumble.controller import Controller
from bumble.link import LocalLink
from bumble.transport import open_transport

transports = {}
link = LocalLink()


def get_default_transport():
return "tcp-server:_:1234"


async def start_transport(transport: str):
if transport not in transports.keys():
transports[transport] = await open_transport(transport)
Controller(
"ext",
host_source=transports[transport].source,
host_sink=transports[transport].sink,
link=link,
)

return transports[transport]


def get_link():
# Assume all transports are linked
return link
80 changes: 80 additions & 0 deletions bleak/backends/bumble/characteristic.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2024 Victor Chavez

from typing import Callable, List, Union
from uuid import UUID

from bumble.gatt_client import CharacteristicProxy, ServiceProxy

from ... import normalize_uuid_str
from ..characteristic import BleakGATTCharacteristic
from ..descriptor import BleakGATTDescriptor
from .utils import uuid_bytes_to_str


class BleakGATTCharacteristicBumble(BleakGATTCharacteristic):
"""GATT Characteristic implementation for the Bumble backend."""

def __init__(
self,
obj: CharacteristicProxy,
max_write_without_response_size: Callable[[], int],
svc: ServiceProxy,
):
super().__init__(obj, max_write_without_response_size)
self.__descriptors = []
self.__props = [str(prop) for prop in obj.properties]
self.__svc = svc
uuid = uuid_bytes_to_str(obj.uuid.uuid_bytes)
self.__uuid = normalize_uuid_str(uuid)

@property
def service_uuid(self) -> str:
"""The uuid of the Service containing this characteristic"""
return self.__svc.uuid

@property
def service_handle(self) -> int:
"""The integer handle of the Service containing this characteristic"""
return self.__svc.handle

@property
def handle(self) -> int:
"""The handle of this characteristic"""
return int(self.obj.handle)

@property
def uuid(self) -> str:
"""The uuid of this characteristic"""
return self.__uuid

@property
def properties(self) -> List[str]:
"""Properties of this characteristic"""
return self.__props

@property
def descriptors(self) -> List[BleakGATTDescriptor]:
"""List of descriptors for this characteristic"""
return self.__descriptors

def get_descriptor(
self, specifier: Union[int, str, UUID]
) -> Union[BleakGATTDescriptor, None]:
"""Get a descriptor by handle (int) or UUID (str or uuid.UUID)"""
try:
if isinstance(specifier, int):
return next(filter(lambda x: x.handle == specifier, self.descriptors))
else:
return next(
filter(lambda x: x.uuid == str(specifier), self.descriptors)
)
except StopIteration:
return None

def add_descriptor(self, descriptor: BleakGATTDescriptor):
"""Add a :py:class:`~BleakGATTDescriptor` to the characteristic.
Should not be used by end user, but rather by `bleak` itself.
"""
self.__descriptors.append(descriptor)
Loading

0 comments on commit b76d7f3

Please sign in to comment.