-
Notifications
You must be signed in to change notification settings - Fork 306
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Added support for Google´s Bumble Bluetooth Controller stack #1681
base: develop
Are you sure you want to change the base?
Conversation
b76d7f3
to
62411ce
Compare
Fixed some linting and doc issues. |
It looks like you may not have updated the lock file. Please run |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks amazing!
Starting with a first round of type safety. While Bleak does not have 100% type coverage, I won't support any new python - in any context - that does not have 100% type coverage. That's just my opinion, and I'm happy to PR for it if it's too much of a lift.
I'll stay tuned to this and test it out on HW ASAP!
bleak/backends/bumble/__init__.py
Outdated
def get_default_transport(): | ||
return "tcp-server:_:1234" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can it be:
def get_default_transport() -> str:
return "tcp-server:_:1234"
or better:
def get_default_transport() -> Literal["tcp-server:_:1234"]:
return "tcp-server:_:1234"
Why does it need a function?
bleak/backends/bumble/__init__.py
Outdated
from bumble.link import LocalLink | ||
from bumble.transport import open_transport | ||
|
||
transports = {} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What are these? Bumble has poor type safety, but it seems we can get some of it:
async def open_transport(name: str) -> Transport:
"""
Open a transport by name.
The name must be <type>:<metadata><parameters>
Where <parameters> depend on the type (and may be empty for some types), and
<metadata> is either omitted, or a ,-separated list of <key>=<value> pairs,
enclosed in [].
If there are not metadata or parameter, the : after the <type> may be omitted.
Examples:
* usb:0
* usb:[driver=rtk]0
* android-netsim
The supported types are:
* serial
* udp
* tcp-client
* tcp-server
* ws-client
* ws-server
* pty
* file
* vhci
* hci-socket
* usb
* pyusb
* android-emulator
* android-netsim
"""
Likely this API should not use strings. Aren't these really enums?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Like why are these serialized to strings? OMG, it parses them:
scheme, *tail = name.split(':', 1)
spec = tail[0] if tail else None
metadata = None
if spec:
# Metadata may precede the spec
if spec.startswith('['):
metadata_str, *tail = spec[1:].split(']')
spec = tail[0] if tail else None
metadata = dict([entry.split('=') for entry in metadata_str.split(',')])
🤦♀️
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When Guido left Google he should have taken Python with him...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think bumble uses directly strings. See here.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Actually there are hidden imports here which is going to be a problem for compiling binaries... lazy imports are nice until they aren't.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think bumble uses directly strings. See here.
Agreed - that API is bumming me out. It should deserialize to some well defined types and then pass those to open the transport. I see it was done this way to allow user input from the command line into this function.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looking into this more, I don't seen any reason to use the open_transport function. Possibly can do away with the stuff in the __init__.py
file entirely. Can the user be responsible to import the transport from Bumble and inject it themselves?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I checked the Transports from bumble and I am not sure its trivial. The implementation specific transports are defined locally inside the function that opens each transport.
Example for the TCP Server
This requires a PR to the bumble project to allow users to access the Transport implementations or copy paste all transports and make them accessible while its not possible directly from the bumble project.
What I could do in the meantime is add enums to the strings of the transport with argument adapter
and a separate argument string for the specific parameters that the transport needs.
bleak/backends/bumble/__init__.py
Outdated
from bumble.transport import open_transport | ||
|
||
transports = {} | ||
link = LocalLink() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't know if these mutable objects are intended to export to the entire application. If they must be at file scope, then some safety can be added, for example:
_link: Final = LocalLink()
bleak/backends/bumble/__init__.py
Outdated
def get_link(): | ||
# Assume all transports are linked | ||
return link |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
def get_link() -> LocalLink:
# Assume all transports are linked
return link
I think I see that these getters are intended to protect the global object.
bleak/backends/bumble/__init__.py
Outdated
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] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
async def start_transport(transport: str) -> Transport:
bleak/backends/bumble/client.py
Outdated
async def disconnect(self) -> bool: | ||
"""Disconnect from the specified GATT server. | ||
|
||
Returns: | ||
Boolean representing connection status. | ||
|
||
""" | ||
await self._dev.disconnect( | ||
self._connection, HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR | ||
) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
missing return
bleak/backends/bumble/client.py
Outdated
""" | ||
descriptor = self.services.get_descriptor(handle) | ||
if not descriptor: | ||
raise BleakError("Descriptor {} was not found!".format(handle)) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Prefer f strings
bleak/backends/bumble/client.py
Outdated
if not descriptor: | ||
raise BleakError("Descriptor {} was not found!".format(handle)) | ||
val = await descriptor.obj.read_value() | ||
logger.debug("Read Descriptor {0} : {1}".format(handle, val)) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Prefer f-strings
Re: the overall API. If this is merged, I think that it should be importable as a separate class. BleakClient can remain as an OS-selected class, but BumbleClient should be separately importable. BumbleClient would be fully compatible with BleakClient - that is, they implement the same interface, Protocol, ABC, whatever you want to call it. |
I agree, the Bumble client is largely OS independent (except for the transports that use the Linux HCI device). So it could be imported separately. Thanks for the feedback on type safety! It really helps to remove ambiguity in Python and help other developers understand the intent of the code. I will apply the suggestions you mentioned. |
Glad you like it! I was annoyed when I first saw it being required in projects but now I'm a "kool-aid drinker" 🤣! It does in fact eliminate an entire class of runtime bugs that Python applications tend to suffer from. You'll need mypy or pyright LSPs/extensions running to really see anything. By default, if a function has no return type specified, the linters ignore the typing in the function body. Best to simply specify a return type for all functions, even if it is None. |
dd907d4
to
7cf2739
Compare
I have applied some of the suggestions from @JPHutchins. I will also check the typesafety of the changes in this PR. |
7a1fc78
to
d49fe6f
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This looks really nice!
I think it will be useful for people who have devices that don't work so well with the various OS Bluetooth stacks. And it could be useful for providing a test backend for anyone that needs some serious testing.
I think that it should be importable as a separate class.
I disagree with this suggestion. I would like to be able to run any script with or without the Bumble backend, so no code changes should be required to use the Bumble backend. I also don't want a bunch of features being added that aren't supported on other backends, so having a separate class should not be necessary.
from bleak.backends.descriptor import BleakGATTDescriptor | ||
|
||
|
||
class BleakGATTCharacteristicBumble(BleakGATTCharacteristic): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've been hoping we could get rid of the subclasses of BleakGATTService
, BleakGATTCharacteristic
and BleakGATTDescriptor
and just use that class directly everywhere since the subclasses are 90% duplicate code.
Not sure if that is something that is feasible to do before this PR or not.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Currently I put this PR in Draft to get some feedback. If you plan to refactor the classes I am open to rebase my fork and change the backend.
await self._dev.power_on() | ||
await self._dev.connect(self.address) | ||
|
||
self.services: BleakGATTServiceCollection = await self.get_services() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If getting services fails, we should disconnect.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What do you mean by disconnect?
should I do wrap it around a try
statement ?
or do you mean if the BLE device does not have any services we should disconnect?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I did not find any similar logic in the backends that refer to check if the service fails. The only one is the WinRT where this is raised and is specific for Windows:
bleak/bleak/backends/winrt/client.py
Line 796 in e01e264
raise |
From bumble I always get a List of services, if not, then its empty and thus will have an empty BleakGATTServiceCollection
|
||
""" | ||
await start_transport(self._adapter) | ||
await self._dev.power_on() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm... power isn't something Bleak has had to deal with before. Since up to now, we have only used OS APIs, it has been up to the OS to control power.
Likely this should be moved to a new adapter class. See #1060
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is because Bumble does not interact with the OS. It has its own HCI stack, so I have to "turn on" the Bumble BLE device, which internally resets the bumble HCI controller.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If there is a BleakAdapter proposed I am glad to use it for this PR
docs/backends/bumble.rst
Outdated
|
||
.. note:: | ||
|
||
To select this backend set the environmental variable ``BLEAK_BUMBLE``. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The environment variable should contain the transport info.
The idea is that users should just have to set the environment variable and use any Bleak script with the Bumble backend without having to modify any code.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
My idea was that I could use bumble for functional testing with a virtual link. The user can then modify their code to select the appropiate transport.
Also what happens if the user wants to use both bumble and their physical OS HCI Controller.
Example:
- Native HCI from Linux : hci0
- HCI Serial controller that does not work well with Linux : bumble over serial
I opted for the bumble backend with an env. variable for tests only. it would be good to exactly define the use cases of bumble for bleak.
- Do we accept bumble as a virtual transport to extend the limitations of OS drivers and hardware issues?
- Should bumble cohabitate with the backend of each native OS ?
- Use case for this is when you still want to use the native hardware from your OS and as mentioned a serial/USB controller.
- Should bumble be flexible to use multiple transports?
- This allows for example to connect an application running over Android simulator, a VHCI (linux), a serial HCI Controller and real hardware on the same network. This would for example enhance functional cross-platform testing of applications.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I will add an env. variable to define use of bumble + transport options. In addition, I will just mention in the docs that if a specific backend must be selected then it can be changed over the BleakClient
argument backend
👍
pyproject.toml
Outdated
@@ -38,6 +38,7 @@ bleak-winrt = { version = "^1.2.0", markers = "platform_system=='Windows'", pyth | |||
"winrt-Windows.Foundation.Collections" = { version = "^2", markers = "platform_system=='Windows'", python = ">=3.12" } | |||
"winrt-Windows.Storage.Streams" = { version = "^2", markers = "platform_system=='Windows'", python = ">=3.12" } | |||
dbus-fast = { version = ">=1.83.0, < 3", markers = "platform_system == 'Linux'" } | |||
bumble = "^0.0.201" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this should be an optional dependency (i.e. only installed if pip install bleak[bumble]
). Otherwise it is going to bring in a lot of other dependencies that most people don't need.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yes, seems fine by me. Then I would add this in the docs for the backend on how to install.
Resolved some of the reviews pending is:
|
1fcba15
to
e6c8182
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looking good! Will be testing with an nrf52840dk momentarily!
Overall, a lot of time will be saved by getting mypy working in your editor.
@property | ||
def service_uuid(self) -> str: | ||
"""The uuid of the Service containing this characteristic""" | ||
return self.__svc.uuid |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Type error, got UUID, expected str
bleak/backends/bumble/client.py
Outdated
value = await self._peer.read_characteristics_by_uuid(characteristic.obj.uuid) | ||
value = bytearray(value[0]) | ||
logger.debug(f"Read Characteristic {characteristic.uuid} : {value}") | ||
return value |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So, the peer returns a list of bytes, then we extract the first one and return it? What is in the rest of the list?
There are type errors here because value is mutated. Best to use a different variable name for the data that's extracted from the list of bytes.
@property | ||
def uuid(self) -> str: | ||
"""UUID for this descriptor""" | ||
return str(self.obj.uuid) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
"DescriptorProxy" has no attribute "uuid"
|
||
device = self.create_or_update_device( | ||
str(advertisement.address), | ||
local_name, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Argument 2 to "create_or_update_device" of "BaseBleakScanner" has incompatible type "list[UUID] | tuple[UUID, bytes] | bytes | str | int | tuple[int, int] | tuple[int, bytes] | Appearance | None"; expected "str"
bleak/backends/bumble/scanner.py
Outdated
for service_data in advertisement.data.get_all(AdvertisingData.SERVICE_DATA): | ||
service_uuid, data = service_data | ||
service_data[str(service_uuid)] = data |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Incompatible types in assignment (expression has type "list[UUID] | tuple[UUID, bytes] | bytes | str | int | tuple[int, int] | tuple[int, bytes] | Appearance", variable has type "dict[str, list[UUID] | tuple[UUID, bytes] | bytes | str | int | tuple[int, int] | tuple[int, bytes] | Appearance]")
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Service data expects dict[str, bytes
, but the for loop mutates the dict and fills some of the values with strings. Is this intentional?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Most of the type errors can be resolved by avoiding the reuse of variable names for objects that are different types.
bleak/backends/bumble/service.py
Outdated
|
||
def __init__(self, obj: ServiceProxy): | ||
super().__init__(obj) | ||
self.__characteristics = [] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Replace all __
with _
. Add type annotation.
bleak/backends/bumble/service.py
Outdated
def add_characteristic(self, characteristic: CharacteristicProxy): | ||
"""Add a :py:class:`~BleakGATTCharacteristicWinRT` to the service. | ||
|
||
Should not be used by end user, but rather by `bleak` itself. | ||
""" | ||
self.__characteristics.append(characteristic) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Argument 1 of "add_characteristic" is incompatible with supertype "BleakGATTService"; supertype defines the argument type as "BleakGATTCharacteristic"
bleak/backends/bumble/utils.py
Outdated
""" | ||
Convert a native Bumble UUID to a string representation. | ||
Reason for this conversion is that the string representation (__str__) done in bumble | ||
is not a standard UUID string representation. | ||
In addition, the byte representation from data member uuid_bytes of bumble.core.UUID | ||
is represented in reverse order. | ||
Example: | ||
Shortened 16 bit UUID: 0x1800Bumble UUID class | ||
bumble UUID string representation: 'UUID-16:1800 (Generic Access)' | ||
bumble UUID bytes representation: b'\x00\x18' | ||
""" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This docstring isn't getting parsed correctly, needs to be updated to conform to google spec.
bleak/backends/bumble/descriptor.py
Outdated
def __init__(self, obj: DescriptorProxy, characteristic: CharacteristicProxy): | ||
super(BleakGATTDescriptorBumble, self).__init__(obj) | ||
self.obj: Final = obj |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Update to:
def __init__(self, obj: DescriptorProxy, characteristic: CharacteristicProxy):
super().__init__(obj)
which is the same, AFAICT.
Sounds like a job for a separate repo. |
Build HCI USB sample for NRF52840DK:
Flash
NRF52840DK device shows up! Confirmed
Tried lower case hex for VID/PID as well. Reference is here: https://google.github.io/bumble/transports/usb.html OK, updated to:
And now I get:
Installing Zadig 🤣 ... Tried libusb-win32 driver - discover script hangs until I unplug the board. Tried libusbK driver - same Tried WinUSB - same Could test from linux first, for example, then try Windows. |
Thanks for the review. I will definitely check again with the typesafety extensions in pycharm. It would be nice to get some field tests to check if everything works. So far I have only tested with virtual devices (Zephyr posix and renode). |
Would you believe that the PID is 000B but I'd set 0008 every time? 😭. Will run through again ASAP. |
OK, picking up where I left off. Starting by trying to revert the drivers to Windows defaults. It's on libwdi now, after unisntalling and restart. Set logging and correct USB VID:
Still hang. No Ctrl-C. At least it's find the USB device now. Can exit by removing the Bumble device. Call stack reveals it's maybe doing some USB stuff (I didn't include whole stack)
So... maybe it's blocking on some USB transfer that never happens. Adding logging to examples.discover: logging.basicConfig(level=logging.DEBUG) I mean... this looks promising!
So maybe there are 2 problems left:
|
So, it looks like Bumble uses a flexible event callback system. Unfortunately, they're using strings again, but here's the magic string: "advertisement". Found in source as self.emit('advertisement', advertisement) So it seems like intended usage may be with a closure, like this: async def start(self) -> None:
def on_advertisement(advertisement: Advertisement):
logger.debug(f"Received advertisement: {advertisement}")
service_uuids: List[str] = []
service_data: Dict[str, AdvertisingDataObject] = {}
local_name = advertisement.data.get(AdvertisingData.COMPLETE_LOCAL_NAME)
if not local_name:
local_name = advertisement.data.get(
AdvertisingData.SHORTENED_LOCAL_NAME
)
manuf_data = advertisement.data.get(
AdvertisingData.MANUFACTURER_SPECIFIC_DATA
)
for uuid_type in SERVICE_UUID_TYPES:
adv_uuids = advertisement.data.get(uuid_type)
if adv_uuids is None:
continue
if not isinstance(adv_uuids, list):
continue
for uuid in adv_uuids: # type: UUID
if uuid not in service_uuids:
service_uuids.append(str(uuid))
for service_data in advertisement.data.get_all(
AdvertisingData.SERVICE_DATA
):
service_uuid, data = service_data
service_data[str(service_uuid)] = data
advertisement_data = AdvertisementData(
local_name=local_name,
manufacturer_data=manuf_data,
service_data=service_data,
service_uuids=service_uuids,
tx_power=advertisement.tx_power,
rssi=advertisement.rssi,
platform_data=(None, None),
)
device = self.create_or_update_device(
str(advertisement.address),
local_name,
{},
advertisement_data,
)
self.call_detection_callbacks(device, advertisement_data)
self.device.on("advertisement", on_advertisement)
await start_transport(self._adapter)
await self.device.power_on()
await self.device.start_scanning(active=self._scan_active) That said, I still have no success. After adding some logging to bumble and pyee I can confirm that no events are being emitted, so I'm not really sure where to look next. |
Aw, I see this was supposed to work with multiple inheritance of |
Can confirm the scanner (Zephyr FW) is working well with Bumble's example!
|
I think this can be refactored to align with the example more closely: https://github.com/google/bumble/blob/5e959d638e6a9c99e62536d0a3472cf4e6616ccf/examples/run_scanner.py#L36-L78 Specifically, class consrtuctors in Python async should not do any IO. So we can wait until some action is taken to open connections and do async IO, for example. |
The following scanner implementation is working well for me. Obviously would need to be adapted to fit with the rest of the API, but it's nice to see it running. # SPDX-License-Identifier: MIT
# Copyright (c) 2024 Victor Chavez
import logging
import os
from typing import Dict, Final, List, Literal, Optional
from bumble.core import AdvertisingData, AdvertisingDataObject
from bumble.device import Advertisement, Device
from bumble.hci import Address
from bumble.transport import open_transport
from bleak.backends.scanner import (
AdvertisementData,
AdvertisementDataCallback,
BaseBleakScanner,
)
logger = logging.getLogger(__name__)
SERVICE_UUID_TYPES = [
AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS,
AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS,
]
# Arbitrary BD_ADDR for the scanner device
SCANNER_BD_ADDR = "F0:F1:F2:F3:F4:F5"
class BleakScannerBumble(BaseBleakScanner):
"""
Interface for Bleak Bluetooth LE Scanners
Args:
detection_callback:
Optional function that will be called each time a device is
discovered or advertising data has changed.
service_uuids:
Optional list of service UUIDs to filter on. Only advertisements
containing this advertising data will be received.
Keyword Args:
adapter (BumbleTransport): Bumble transport adapter to use.
"""
def __init__(
self,
detection_callback: Optional[AdvertisementDataCallback],
service_uuids: Optional[List[str]],
scanning_mode: Literal["active", "passive"],
**kwargs,
):
super().__init__(detection_callback, service_uuids)
self._scanning_mode: Final = scanning_mode
self._device: Optional[Device] = None
async def on_connection(self, connection):
pass
async def start(self) -> None:
def on_advertisement(advertisement: Advertisement):
logger.debug(f"Received advertisement: {advertisement}")
service_uuids: List[str] = []
service_data: Dict[str, AdvertisingDataObject] = {}
local_name = advertisement.data.get(AdvertisingData.COMPLETE_LOCAL_NAME)
if not local_name:
local_name = advertisement.data.get(
AdvertisingData.SHORTENED_LOCAL_NAME
)
manuf_data = advertisement.data.get(
AdvertisingData.MANUFACTURER_SPECIFIC_DATA
)
for uuid_type in SERVICE_UUID_TYPES:
adv_uuids = advertisement.data.get(uuid_type)
if adv_uuids is None:
continue
if not isinstance(adv_uuids, list):
continue
for uuid in adv_uuids:
if uuid not in service_uuids:
service_uuids.append(str(uuid))
for service_data in advertisement.data.get_all(
AdvertisingData.SERVICE_DATA
):
service_uuid, data = service_data
service_data[str(service_uuid)] = data
advertisement_data = AdvertisementData(
local_name=local_name,
manufacturer_data=manuf_data,
service_data=service_data,
service_uuids=service_uuids,
tx_power=advertisement.tx_power,
rssi=advertisement.rssi,
platform_data=(None, None),
)
device = self.create_or_update_device(
str(advertisement.address),
local_name,
{},
advertisement_data,
)
self.call_detection_callbacks(device, advertisement_data)
hci_transport: Final = await open_transport(os.environ["BLEAK_BUMBLE"])
self._device = Device.with_hci(
"scanner_dev",
Address(SCANNER_BD_ADDR),
hci_transport.source,
hci_transport.sink,
)
self._device.on("advertisement", on_advertisement)
await self._device.power_on()
await self._device.start_scanning(active=self._scanning_mode == "active")
async def stop(self) -> None:
if self._device is None:
raise RuntimeError("Scanner not started")
await self._device.stop_scanning()
await self._device.power_off()
self._device = None
def set_scanning_filter(self, **kwargs) -> None:
# Implement scanning filter setup
pass |
There's another wrinkle with Bumble. The library defines Defining the USBTransport class in a factory function is a problem. |
Thanks for the tests and suggestions to make it work with usb 👍 |
I have done some more type safety changes, and moved operation calls of the Bumble Controller stack to function calls. In addition, I removed the listeners and use the |
I think the modifications you made were due to having an HCI controller (Zephyr MCU). graph LR
A[Bleak Bumble - HCI Host] <-->B[Zephyr - HCI Controller]<--> C[Bluetooth Radio]
For my use case its the other way around. graph LR
A[Bleak Bumble - HCI Controller] <-->B[Zephyr - HCI Host]
Perhaps there should be an env variable or setting to set the backend mode:
|
Just a short update. You can now set the HCI Mode (Host or controller). To set as HCI Host env. variable |
1a4da61
to
168ed72
Compare
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).
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).
Use cases for this backend are:
support virtualization are Android Emulator and Zephyr RTOS.
radio network (virtual or physical).
To enable this backend the env. variable
BLEAK_BUMBLE
must be set.At the moment this is an experimental backend and would like to get some feedback from users and the devs of bleak :)