fix: follow Bluetooth adapters that change their hci index

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

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):

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]) == {}