diff --git a/custom_components/ble_monitor/__init__.py b/custom_components/ble_monitor/__init__.py index 07b141eb..2e03aded 100644 --- a/custom_components/ble_monitor/__init__.py +++ b/custom_components/ble_monitor/__init__.py @@ -21,7 +21,7 @@ from .ble_parser import BleParser from .bt_helpers import (BT_INTERFACES, BT_MULTI_SELECT, DEFAULT_BT_INTERFACE, - reset_bluetooth) + hci_get_all_mac, reset_bluetooth) from .const import (AES128KEY24_REGEX, AES128KEY32_REGEX, AUTO_BINARY_SENSOR_LIST, AUTO_MANUFACTURER_DICT, AUTO_SENSOR_LIST, CONF_ACTIVE_SCAN, CONF_BATT_ENTITIES, @@ -534,6 +534,15 @@ def __init__(self, config, dataqueue): self.evt_cnt = {} self.config = config self._interfaces = list(set(config[CONF_HCI_INTERFACE])) + # hci index of each configured adapter by MAC address, to follow an adapter + # that the kernel re-enumerates under another index (e.g. after a USB reset) + self._hci_by_mac = {} + if len(config[CONF_HCI_INTERFACE]) == len(config[CONF_BT_INTERFACE]): + self._hci_by_mac = { + bt_mac: hci + for hci, bt_mac in zip(config[CONF_HCI_INTERFACE], config[CONF_BT_INTERFACE]) + if bt_mac != "disable" + } self._active = int(config[CONF_ACTIVE_SCAN] is True) self._inactivity_timeout = config.get( CONF_HCI_INACTIVITY_TIMEOUT, DEFAULT_HCI_INACTIVITY_TIMEOUT @@ -705,6 +714,35 @@ def _schedule_scan_retry(self): self.SCAN_RETRY_INTERVAL, self._event_loop.stop ) + def _follow_renumbered_interfaces(self): + """Look up the current hci index of the configured adapters by MAC address. + + The kernel can re-enumerate an adapter under another hci index, e.g. after + a USB reset of a Bluetooth dongle. Without this, the scanner keeps using the + old index until Home Assistant is restarted. + """ + if not self._hci_by_mac: + return + available = {bt_mac: hci for hci, bt_mac in hci_get_all_mac().items()} + for bt_mac, hci in self._hci_by_mac.items(): + new_hci = available.get(bt_mac) + if new_hci is not None and new_hci != hci: + _LOGGER.warning( + "HCIdump thread: Bluetooth adapter %s moved from hci%i to hci%i", + bt_mac, + hci, + new_hci, + ) + self._hci_by_mac[bt_mac] = new_hci + self._interfaces = list(dict.fromkeys(self._hci_by_mac.values())) + + def _interface_mac(self, hci): + """Return the MAC address of the adapter at an hci index.""" + for bt_mac, bt_hci in self._hci_by_mac.items(): + if bt_hci == hci: + return bt_mac + return BT_INTERFACES.get(hci) + def run(self): """Run HCIdump thread.""" while True: @@ -721,6 +759,7 @@ def run(self): self._event_loop = asyncio.new_event_loop() asyncio.set_event_loop(self._event_loop) if "disable" not in self.config[CONF_BT_INTERFACE]: + self._follow_renumbered_interfaces() for hci in self._interfaces: interface_is_ok[hci] = False try: @@ -792,7 +831,7 @@ def run(self): "HCIdump thread: Trying to power cycle Bluetooth adapter hci%i %s," " will try to use it next scan period.", iface, - BT_INTERFACES[iface], + self._interface_mac(iface), ) reset_bluetooth(iface) self.last_bt_reset = ts_now diff --git a/custom_components/ble_monitor/bt_helpers.py b/custom_components/ble_monitor/bt_helpers.py index 19cb4d1c..baf6514d 100644 --- a/custom_components/ble_monitor/bt_helpers.py +++ b/custom_components/ble_monitor/bt_helpers.py @@ -81,22 +81,27 @@ def powered(self, new_state): # Bluetooth interfaces available on the system -def hci_get_mac(iface_list=None): - """Get dict of available bluetooth interfaces, returns hci and mac.""" +def hci_get_all_mac(): + """Get dict of all available bluetooth interfaces, returns hci and mac.""" # Result example: {0: 'F2:67:F3:5B:4D:FC', 1: '00:1A:7D:DA:71:11'} try: btctl = MGMTBluetoothCtl() except BluetoothSocketError as error: _LOGGER.debug("BluetoothSocketError: %s", error) return {} + return btctl.presented_list + + +def hci_get_mac(iface_list=None): + """Get dict of available bluetooth interfaces, returns hci and mac.""" + # Result example: {0: 'F2:67:F3:5B:4D:FC', 1: '00:1A:7D:DA:71:11'} + presented_list = hci_get_all_mac() q_iface_list = iface_list or [0] - btaddress_dict = {} - for hci_idx in q_iface_list: - try: - btaddress_dict[hci_idx] = btctl.presented_list[hci_idx] - except KeyError: - pass - return btaddress_dict + return { + hci_idx: presented_list[hci_idx] + for hci_idx in q_iface_list + if hci_idx in presented_list + } def reset_bluetooth(hci): diff --git a/custom_components/ble_monitor/test/test_hcidump_interfaces.py b/custom_components/ble_monitor/test/test_hcidump_interfaces.py new file mode 100644 index 00000000..734d3677 --- /dev/null +++ b/custom_components/ble_monitor/test/test_hcidump_interfaces.py @@ -0,0 +1,110 @@ +"""Tests for following Bluetooth adapters that change their hci index.""" +from unittest.mock import Mock, patch + +from ble_monitor.bt_helpers import hci_get_all_mac, hci_get_mac +from ble_monitor.const import (CONF_ACTIVE_SCAN, CONF_BT_INTERFACE, + CONF_HCI_INTERFACE, CONF_REPORT_UNKNOWN) +from btsocket.btmgmt_socket import BluetoothSocketError +from homeassistant.const import CONF_DEVICES, CONF_DISCOVERY + +from ble_monitor import HCIdump + +DONGLE = "00:E0:43:76:3E:03" +ONBOARD = "E4:5F:01:74:E1:FF" + + +def make_hcidump(bt_interfaces, hci_interfaces): + """Create an HCIdump for an adapter configuration without opening sockets.""" + config = { + CONF_ACTIVE_SCAN: False, + CONF_BT_INTERFACE: bt_interfaces, + CONF_DEVICES: [], + CONF_DISCOVERY: True, + CONF_HCI_INTERFACE: hci_interfaces, + CONF_REPORT_UNKNOWN: False, + } + dataqueue = {"binary": Mock(), "measuring": Mock(), "tracker": Mock()} + return HCIdump(config, dataqueue) + + +def test_adapter_is_followed_to_its_new_hci_index(): + """A re-enumerated adapter is used at its new index, found by its MAC address.""" + hcidump = make_hcidump([DONGLE], [2]) + + with patch("ble_monitor.hci_get_all_mac", return_value={0: DONGLE, 1: ONBOARD}): + hcidump._follow_renumbered_interfaces() + + assert hcidump._interfaces == [0] + assert hcidump._interface_mac(0) == DONGLE + + +def test_unchanged_adapter_keeps_its_hci_index(): + """Nothing changes while the adapter is still at its configured index.""" + hcidump = make_hcidump([DONGLE], [2]) + + with patch("ble_monitor.hci_get_all_mac", return_value={1: ONBOARD, 2: DONGLE}): + hcidump._follow_renumbered_interfaces() + + assert hcidump._interfaces == [2] + + +def test_missing_adapter_keeps_its_hci_index(): + """An adapter that is currently gone keeps its index, so power cycling still applies.""" + hcidump = make_hcidump([DONGLE], [2]) + + with patch("ble_monitor.hci_get_all_mac", return_value={1: ONBOARD}): + hcidump._follow_renumbered_interfaces() + + assert hcidump._interfaces == [2] + assert hcidump._interface_mac(2) == DONGLE + + +def test_adapters_swapping_hci_indexes(): + """Two configured adapters that swap their indexes are both followed.""" + hcidump = make_hcidump([DONGLE, ONBOARD], [0, 1]) + + with patch("ble_monitor.hci_get_all_mac", return_value={0: ONBOARD, 1: DONGLE}): + hcidump._follow_renumbered_interfaces() + + assert sorted(hcidump._interfaces) == [0, 1] + assert hcidump._interface_mac(0) == ONBOARD + assert hcidump._interface_mac(1) == DONGLE + + +def test_disabled_bluetooth_does_not_look_up_adapters(): + """Without a Bluetooth adapter, no adapter lookup is done.""" + hcidump = make_hcidump(["disable"], ["disable"]) + + with patch("ble_monitor.hci_get_all_mac") as get_all_mac: + hcidump._follow_renumbered_interfaces() + + get_all_mac.assert_not_called() + assert hcidump._interfaces == ["disable"] + + +def test_index_without_known_mac_does_not_raise(): + """An hci index without a known MAC address is logged without a KeyError.""" + hcidump = make_hcidump([], [4]) + + with patch("ble_monitor.BT_INTERFACES", {}): + assert hcidump._interface_mac(4) is None + + +def test_hci_get_mac_returns_requested_adapters_only(): + """hci_get_mac filters the available adapters, hci_get_all_mac returns all of them.""" + btctl = Mock(presented_list={0: ONBOARD, 4: DONGLE}) + + with patch("ble_monitor.bt_helpers.MGMTBluetoothCtl", return_value=btctl): + assert hci_get_all_mac() == {0: ONBOARD, 4: DONGLE} + assert hci_get_mac([0, 1, 2, 3]) == {0: ONBOARD} + assert hci_get_mac() == {0: ONBOARD} + + +def test_no_adapters_without_management_socket(): + """No adapters are returned when the Bluetooth management socket is unavailable.""" + with patch( + "ble_monitor.bt_helpers.MGMTBluetoothCtl", + side_effect=BluetoothSocketError("no socket"), + ): + assert hci_get_all_mac() == {} + assert hci_get_mac([0, 1, 2, 3]) == {}