Move Broadlink to the asynchronous python-broadlink library (#182999)

homeassistant/components/broadlink/config_flow.py

@@ -2,7 +2,6 @@
 
 from collections.abc import Mapping
 import errno
-from functools import partial
 import logging
 import socket
 from typing import Any, override
@@ -22,6 +21,7 @@ from homeassistant.config_entries import (
     ConfigFlowResult,
 )
 from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, CONF_TIMEOUT, CONF_TYPE
+from homeassistant.core import callback
 from homeassistant.data_entry_flow import AbortFlow
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
@@ -39,6 +39,17 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
 
     device: blk.Device
 
+    @override
+    @callback
+    def async_remove(self) -> None:
+        """Close the device's socket when the flow ends, however it ends.
+
+        The device created for the flow is only used to probe and
+        authenticate; the config entry builds its own.
+        """
+        if (device := getattr(self, "device", None)) is not None:
+            self.hass.async_create_task(device.aclose())
+
     async def async_set_device(
         self, device: blk.Device, raise_on_progress: bool = True
     ) -> None:
@@ -56,6 +67,10 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
         await self.async_set_unique_id(
             device.mac.hex(), raise_on_progress=raise_on_progress
         )
+        # A probe replaced after a failed auth() has an open endpoint.
+        previous = getattr(self, "device", None)
+        if previous is not None and previous is not device:
+            await previous.aclose()
         self.device = device
 
         self.context["title_placeholders"] = {
@@ -75,7 +90,7 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
         self._abort_if_unique_id_configured(updates={CONF_HOST: host})
 
         try:
-            device = await self.hass.async_add_executor_job(blk.hello, host)
+            device = await blk.hello(host)
 
         except NetworkTimeoutError:
             return self.async_abort(reason="cannot_connect")
@@ -103,8 +118,7 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
             timeout = user_input.get(CONF_TIMEOUT, DEFAULT_TIMEOUT)
 
             try:
-                hello = partial(blk.hello, host, timeout=timeout)
-                device = await self.hass.async_add_executor_job(hello)
+                device = await blk.hello(host, timeout=timeout)
 
             except NetworkTimeoutError:
                 errors["base"] = "cannot_connect"
@@ -162,7 +176,7 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
         errors: dict[str, str] = {}
 
         try:
-            await self.hass.async_add_executor_job(device.auth)
+            await device.auth()
 
         except AuthenticationError:
             errors["base"] = "invalid_auth"
@@ -252,7 +266,7 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
 
         elif user_input["unlock"]:
             try:
-                await self.hass.async_add_executor_job(device.set_lock, False)
+                await device.set_lock(False)
 
             except NetworkTimeoutError as err:
                 errors["base"] = "cannot_connect"

homeassistant/components/broadlink/device.py

@@ -1,7 +1,7 @@
 """Support for Broadlink devices."""
 
+from collections.abc import Awaitable, Callable
 from contextlib import suppress
-from functools import partial
 import logging
 
 import broadlink as blk
@@ -10,6 +10,7 @@ from broadlink.exceptions import (
     AuthorizationError,
     BroadlinkException,
     ConnectionClosedError,
+    EndpointClosedError,
     NetworkTimeoutError,
 )
 
@@ -88,11 +89,11 @@ class BroadlinkDevice[_ApiT: blk.Device = blk.Device]:
         device_registry.async_update_device(device_entry.id, name=entry.title)
         await hass.config_entries.async_reload(entry.entry_id)
 
-    def _get_firmware_version(self) -> int | None:
+    async def _async_get_firmware_version(self) -> int | None:
         """Get firmware version."""
-        self.api.auth()
+        await self.api.auth()
         with suppress(BroadlinkException, OSError):
-            return self.api.get_fwversion()
+            return await self.api.get_fwversion()
         return None
 
     async def async_setup(self) -> bool:
@@ -108,16 +109,18 @@ class BroadlinkDevice[_ApiT: blk.Device = blk.Device]:
         api.timeout = config.data[CONF_TIMEOUT]
         self.api = api
 
+        # The device is not registered yet, so a failure below must close
+        # the endpoint auth() opened; async_unload will not run for it.
         try:
-            self.fw_version = await self.hass.async_add_executor_job(
-                self._get_firmware_version
-            )
+            self.fw_version = await self._async_get_firmware_version()
 
         except AuthenticationError:
+            await api.aclose()
             await self._async_handle_auth_error()
             return False
 
         except (NetworkTimeoutError, OSError) as err:
+            await api.aclose()
             raise ConfigEntryNotReady(
                 translation_domain=DOMAIN,
                 translation_key="connect_failed",
@@ -128,6 +131,7 @@ class BroadlinkDevice[_ApiT: blk.Device = blk.Device]:
             ) from err
 
         except BroadlinkException as err:
+            await api.aclose()
             _LOGGER.error(
                 "Failed to authenticate to the device at %s: %s", api.host[0], err
             )
@@ -137,7 +141,11 @@ class BroadlinkDevice[_ApiT: blk.Device = blk.Device]:
 
         update_manager = get_update_manager(self)
         coordinator = update_manager.coordinator
-        await coordinator.async_config_entry_first_refresh()
+        try:
+            await coordinator.async_config_entry_first_refresh()
+        except ConfigEntryNotReady:
+            await api.aclose()
+            raise
 
         self.update_manager = update_manager
         # Uses legacy hass.data[DOMAIN] pattern
@@ -160,14 +168,17 @@ class BroadlinkDevice[_ApiT: blk.Device = blk.Device]:
         while self.reset_jobs:
             self.reset_jobs.pop()()
 
-        return await self.hass.config_entries.async_unload_platforms(
+        unloaded = await self.hass.config_entries.async_unload_platforms(
             self.config, get_domains(self.api.type)
         )
+        if unloaded:
+            await self.api.aclose()
+        return unloaded
 
     async def async_auth(self) -> bool:
         """Authenticate to the device."""
         try:
-            await self.hass.async_add_executor_job(self.api.auth)
+            await self.api.auth()
         except (BroadlinkException, OSError) as err:
             _LOGGER.debug(
                 "Failed to authenticate to the device at %s: %s", self.api.host[0], err
@@ -177,15 +188,25 @@ class BroadlinkDevice[_ApiT: blk.Device = blk.Device]:
             return False
         return True
 
-    async def async_request(self, function, *args, **kwargs):
-        """Send a request to the device."""
-        request = partial(function, *args, **kwargs)
+    async def async_request[**_P, _R](
+        self,
+        function: Callable[_P, Awaitable[_R]],
+        *args: _P.args,
+        **kwargs: _P.kwargs,
+    ) -> _R:
+        """Send a request to the device.
+
+        Re-authenticate and retry once on an authorization error; a request
+        that fails because the endpoint was closed on unload is not retried.
+        """
         try:
-            return await self.hass.async_add_executor_job(request)
+            return await function(*args, **kwargs)
+        except EndpointClosedError:
+            raise
         except AuthorizationError, ConnectionClosedError:
             if not await self.async_auth():
                 raise
-            return await self.hass.async_add_executor_job(request)
+            return await function(*args, **kwargs)
 
     async def _async_handle_auth_error(self) -> None:
         """Handle an authentication error."""

homeassistant/components/broadlink/heartbeat.py

@@ -48,14 +48,14 @@ class BroadlinkHeartbeat:
         hass = self._hass
         config_entries = hass.config_entries.async_entries(DOMAIN)
         hosts: set[str] = {entry.data[CONF_HOST] for entry in config_entries}
-        await hass.async_add_executor_job(self.heartbeat, hosts)
+        await self.heartbeat(hosts)
 
     @staticmethod
-    def heartbeat(hosts: set[str]) -> None:
+    async def heartbeat(hosts: set[str]) -> None:
         """Send packets to feed watchdog timers."""
         for host in hosts:
             try:
-                blk.ping(host)
+                await blk.ping(host)
             except OSError as err:
                 _LOGGER.debug("Failed to send heartbeat to %s: %s", host, err)
             else:

homeassistant/components/broadlink/manifest.json

@@ -39,5 +39,5 @@
   "integration_type": "device",
   "iot_class": "local_polling",
   "loggers": ["broadlink"],
-  "requirements": ["broadlink==0.19.0"]
+  "requirements": ["python-broadlink==1.0.6"]
 }

homeassistant/components/broadlink/radio_frequency.py

@@ -4,6 +4,7 @@ import logging
 from typing import override
 
 from broadlink.exceptions import BroadlinkException
+from broadlink.remote import TICK
 from rf_protocols import RadioFrequencyCommand
 
 from homeassistant.components.radio_frequency import RadioFrequencyTransmitterEntity
@@ -20,7 +21,9 @@ _LOGGER = logging.getLogger(__name__)
 
 PARALLEL_UPDATES = 0
 
-_TICK_US = 32.84
+# The device's timing unit in microseconds, taken from the library so the
+# IR and RF paths on the same hardware always agree.
+_TICK_US = TICK
 
 _RF_433_TYPE_BYTE = 0xB2
 _RF_315_TYPE_BYTE = 0xB4
@@ -60,8 +63,9 @@ def encode_rf_packet(
         bytes 4..N-1     pulses: 1 byte when ticks < 256, otherwise
                          0x00 followed by a 2-byte big-endian tick count
 
-    Each pulse is expressed as multiples of 32.84 µs ticks, which is the
-    timing resolution of the Broadlink RF front-end.
+    Each pulse is expressed as multiples of the Broadlink timing unit
+    (about 30.45 µs, ``broadlink.remote.TICK``), which is the timing
+    resolution of the RF front-end.
     """
     buf = bytearray([type_byte, repeat_count, 0, 0])
     for duration in timings_us:

requirements_all.txt

@@ -733,9 +733,6 @@ boto3==1.42.97
 # homeassistant.components.bring
 bring-api==1.1.2
 
-# homeassistant.components.broadlink
-broadlink==0.19.0
-
 # homeassistant.components.brother
 brother==6.2.0
 
@@ -2704,6 +2701,9 @@ python-awair==0.2.5
 # homeassistant.components.blockchain
 python-blockchain-api==0.0.2
 
+# homeassistant.components.broadlink
+python-broadlink==1.0.6
+
 # homeassistant.components.bsblan
 python-bsblan==6.1.8
 

tests/components/broadlink/__init__.py

@@ -1,7 +1,10 @@
 """Tests for the Broadlink integration."""
 
 from dataclasses import dataclass
-from unittest.mock import MagicMock, patch
+import inspect
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import broadlink as blk
 
 from homeassistant.components.broadlink.const import DOMAIN
 from homeassistant.core import HomeAssistant
@@ -103,6 +106,34 @@ BROADLINK_DEVICES = {
 }
 
 
+def _async_api_methods() -> frozenset[str]:
+    """Names of every coroutine method on any device class in the library."""
+    names: set[str] = set()
+    for _, cls in inspect.getmembers(blk, inspect.isclass):
+        if issubclass(cls, blk.Device):
+            names.update(
+                name for name, _ in inspect.getmembers(cls, inspect.iscoroutinefunction)
+            )
+    return frozenset(names)
+
+
+ASYNC_API_METHODS = _async_api_methods()
+
+
+def mock_api_base() -> MagicMock:
+    """Return a mock device whose coroutine methods await to plain mocks.
+
+    Attributes such as ``host`` and ``mac`` stay ordinary values; every
+    method that is a coroutine on the real library becomes an ``AsyncMock``
+    returning a ``MagicMock``, so code that reads the result (``data.get``)
+    does not receive another coroutine.
+    """
+    mock_api = MagicMock()
+    for name in ASYNC_API_METHODS:
+        setattr(mock_api, name, AsyncMock(return_value=MagicMock()))
+    return mock_api
+
+
 @dataclass
 class MockSetup:
     """Representation of a mock setup."""
@@ -160,7 +191,7 @@ class BroadlinkDevice:
 
     def get_mock_api(self):
         """Return a mock device (API)."""
-        mock_api = MagicMock()
+        mock_api = mock_api_base()
         mock_api.name = self.name
         mock_api.host = (self.host, 80)
         mock_api.mac = bytes.fromhex(self.mac)

tests/components/broadlink/test_config_flow.py

@@ -355,12 +355,17 @@ async def test_flow_reset_works(hass: HomeAssistant) -> None:
             {"host": device.host, "timeout": device.timeout},
         )
 
-    with patch(DEVICE_HELLO, return_value=device.get_mock_api()):
+    unlocked_api = device.get_mock_api()
+    with patch(DEVICE_HELLO, return_value=unlocked_api):
         result = await hass.config_entries.flow.async_configure(
             result["flow_id"],
             {"host": device.host, "timeout": device.timeout},
         )
 
+    # The first probe opened its endpoint in auth(); replacing it closes it.
+    assert mock_api.aclose.await_count == 1
+    assert unlocked_api.aclose.await_count == 0
+
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         {"name": device.name},
@@ -369,6 +374,8 @@ async def test_flow_reset_works(hass: HomeAssistant) -> None:
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["title"] == device.name
     assert result["data"] == device.get_entry_data()
+    await hass.async_block_till_done()
+    assert unlocked_api.aclose.await_count == 1
 
 
 async def test_flow_unlock_works(hass: HomeAssistant) -> None:

tests/components/broadlink/test_device.py

@@ -3,6 +3,7 @@
 from unittest.mock import patch
 
 import broadlink.exceptions as blke
+import pytest
 
 from homeassistant.components.broadlink.const import DOMAIN
 from homeassistant.components.broadlink.device import get_domains
@@ -29,6 +30,7 @@ async def test_device_setup(hass: HomeAssistant) -> None:
     assert mock_setup.entry.state is ConfigEntryState.LOADED
     assert mock_setup.api.auth.call_count == 1
     assert mock_setup.api.get_fwversion.call_count == 1
+    assert mock_setup.api.aclose.await_count == 0
     assert mock_setup.factory.call_count == 1
 
     forward_entries = set(mock_forward.mock_calls[0][1][1])
@@ -52,6 +54,7 @@ async def test_device_setup_authentication_error(hass: HomeAssistant) -> None:
 
     assert mock_setup.entry.state is ConfigEntryState.SETUP_ERROR
     assert mock_setup.api.auth.call_count == 1
+    assert mock_setup.api.aclose.await_count == 1
     assert mock_forward.call_count == 0
     assert mock_init.call_count == 1
     assert mock_init.mock_calls[0][2]["context"]["source"] == "reauth"
@@ -78,6 +81,7 @@ async def test_device_setup_network_timeout(hass: HomeAssistant) -> None:
         mock_setup.entry.reason == "Failed to connect to the device at 192.168.0.13: "
     )
     assert mock_setup.api.auth.call_count == 1
+    assert mock_setup.api.aclose.await_count == 1
     assert mock_forward.call_count == 0
     assert mock_init.call_count == 0
 
@@ -99,6 +103,7 @@ async def test_device_setup_os_error(hass: HomeAssistant) -> None:
         mock_setup.entry.reason == "Failed to connect to the device at 192.168.0.13: "
     )
     assert mock_setup.api.auth.call_count == 1
+    assert mock_setup.api.aclose.await_count == 1
     assert mock_forward.call_count == 0
     assert mock_init.call_count == 0
 
@@ -117,6 +122,7 @@ async def test_device_setup_broadlink_exception(hass: HomeAssistant) -> None:
 
     assert mock_setup.entry.state is ConfigEntryState.SETUP_ERROR
     assert mock_setup.api.auth.call_count == 1
+    assert mock_setup.api.aclose.await_count == 1
     assert mock_forward.call_count == 0
     assert mock_init.call_count == 0
 
@@ -135,11 +141,32 @@ async def test_device_setup_update_network_timeout(hass: HomeAssistant) -> None:
 
     assert mock_setup.entry.state is ConfigEntryState.SETUP_RETRY
     assert mock_setup.api.auth.call_count == 1
+    assert mock_setup.api.aclose.await_count == 1
     assert mock_setup.api.check_sensors.call_count == 1
     assert mock_forward.call_count == 0
     assert mock_init.call_count == 0
 
 
+async def test_device_request_endpoint_closed(hass: HomeAssistant) -> None:
+    """Test a request on a closed endpoint is raised without a retry."""
+    device = get_device("Office")
+    mock_api = device.get_mock_api()
+
+    with patch.object(hass.config_entries, "async_forward_entry_setups"):
+        mock_setup = await device.setup_entry(hass, mock_api=mock_api)
+
+    mock_api.check_sensors.side_effect = blke.EndpointClosedError()
+    mock_api.auth.reset_mock()
+    mock_api.check_sensors.reset_mock()
+
+    broadlink_device = hass.data[DOMAIN].devices[mock_setup.entry.entry_id]
+    with pytest.raises(blke.EndpointClosedError):
+        await broadlink_device.async_request(mock_api.check_sensors)
+
+    assert mock_api.check_sensors.call_count == 1
+    assert mock_api.auth.call_count == 0
+
+
 async def test_device_setup_update_authorization_error(hass: HomeAssistant) -> None:
     """Test we handle an authorization error in the update step."""
     device = get_device("Office")
@@ -181,6 +208,7 @@ async def test_device_setup_update_authentication_error(hass: HomeAssistant) ->
 
     assert mock_setup.entry.state is ConfigEntryState.SETUP_RETRY
     assert mock_setup.api.auth.call_count == 2
+    assert mock_setup.api.aclose.await_count == 1
     assert mock_setup.api.check_sensors.call_count == 1
     assert mock_forward.call_count == 0
     assert mock_init.call_count == 1
@@ -205,6 +233,7 @@ async def test_device_setup_update_broadlink_exception(hass: HomeAssistant) -> N
 
     assert mock_setup.entry.state is ConfigEntryState.SETUP_RETRY
     assert mock_setup.api.auth.call_count == 1
+    assert mock_setup.api.aclose.await_count == 1
     assert mock_setup.api.check_sensors.call_count == 1
     assert mock_forward.call_count == 0
     assert mock_init.call_count == 0
@@ -291,6 +320,7 @@ async def test_device_unload_works(hass: HomeAssistant) -> None:
         await hass.config_entries.async_unload(mock_setup.entry.entry_id)
 
     assert mock_setup.entry.state is ConfigEntryState.NOT_LOADED
+    assert mock_setup.api.aclose.await_count == 1
     forward_entries = {c[1][1] for c in mock_forward.mock_calls}
     domains = get_domains(mock_setup.api.type)
     assert mock_forward.call_count == len(domains)