My goal is to forward Bluetooth commands without using a Bluetooth adapter.
I’m attempting to develop a Bluetooth proxy component and have encountered some issues.
I submit Bluetooth advertisement data to an event and then create a scanner that inherits from the BaseHaRemoteScanner class to listen for events and receive advertisement data
The code is roughly as follows.
bluetooth.py
from base64 import b64decode
from habluetooth import BaseHaRemoteScanner, HaBluetoothConnector
from homeassistant.components.bluetooth import MONOTONIC_TIME, async_register_scanner
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback, Event
from homeassistant.config_entries import ConfigEntry
from .const import LOGGER, BLE_SCAN_RESULT_EVENT, DOMAIN
def decode_data(data: dict[str, str]) -> dict[str, bytes]:
return {int(k): b64decode(v.encode()) for k, v in data.items()}
class MyBLEScanner(BaseHaRemoteScanner):
def async_on_event(self, event: Event) -> None:
if event.event_type != BLE_SCAN_RESULT_EVENT:
return
LOGGER.debug(f"Received advertisement data: {event.data}")
service_data = decode_data(event.data["service_data"])
manufacturer_data = decode_data(event.data["manufacturer_data"])
self._async_on_advertisement(
address=event.data["address"],
rssi=event.data.get("rssi"),
local_name=event.data.get("local_name"),
service_data=service_data,
service_uuids=event.data.get("service_uuids"),
manufacturer_data=manufacturer_data,
tx_power=None,
details={},
advertisement_monotonic_time=MONOTONIC_TIME(),
)
def async_connect_scanner(hass: HomeAssistant, entry: ConfigEntry) -> tuple[MyBLEScanner, CALLBACK_TYPE]:
"""Connect scanner"""
assert entry.unique_id is not None
source = str(entry.unique_id)
LOGGER.debug(f"{entry.title} [{source}]: Connecting scanner")
scanner = MyBLEScanner(
scanner_id=source,
name=entry.title,
connector=HaBluetoothConnector(client=None, source=source, can_connect=lambda: True),
connectable=True,
)
unload_callbacks = [
async_register_scanner(hass, scanner, source_domain=DOMAIN, source_config_entry_id=entry.entry_id),
scanner.async_setup(),
hass.bus.async_listen(BLE_SCAN_RESULT_EVENT, scanner.async_on_event),
]
@callback
def _async_unload() -> None:
for unloader in unload_callbacks:
unloader()
return (scanner, _async_unload)
__init__.py
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
entry.runtime_data = MyUpdateCoordinator(hass=hass, entry=entry)
entry.runtime_data.async_setup()
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
entry.runtime_data.async_on_disconnect()
return unload_ok
coordinator.py
from homeassistant.core import callback, HomeAssistant, CALLBACK_TYPE
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import LOGGER
from .bluetooth import async_connect_scanner
class MyUpdateCoordinator(DataUpdateCoordinator):
def __init__(self, hass: HomeAssistant, entry: ConfigEntry):
super().__init__(hass, LOGGER, config_entry=entry, name="test")
self._disconnected_callbacks: list[CALLBACK_TYPE] = []
def async_setup(self) -> None:
self.config_entry.async_create_task(self.hass, self._async_connected())
async def _async_connected(self):
_, unload_scaner = async_connect_scanner(self.hass, self.config_entry)
self._disconnected_callbacks.append(unload_scaner)
@callback
def async_on_disconnect(self) -> None:
for disconnected_callback in self._disconnected_callbacks:
disconnected_callback()
self._disconnected_callbacks.clear()
I am able to receive advertisement data brought by the event, but after calling the BaseHaRemoteScanner._async_on_advertisement function, the device is still not discoverable on the web
How can I create a scanner that can discover devices?