Fix connection problems in the Broadlink integration

.coveragerc

@@ -90,6 +90,7 @@ omit =
     homeassistant/components/braviatv/const.py
     homeassistant/components/braviatv/media_player.py
     homeassistant/components/broadlink/const.py
+    homeassistant/components/broadlink/device.py
     homeassistant/components/broadlink/remote.py
     homeassistant/components/broadlink/sensor.py
     homeassistant/components/broadlink/switch.py

homeassistant/components/broadlink/__init__.py

@@ -5,12 +5,11 @@
 from datetime import timedelta
 import logging
 import re
-import socket
 
+from broadlink.exceptions import BroadlinkException, ReadError
 import voluptuous as vol
 
 from homeassistant.const import CONF_HOST
-from homeassistant.core import callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.util.dt import utcnow
 
@@ -65,69 +64,62 @@ def mac_address(value):
 SERVICE_LEARN_SCHEMA = vol.Schema({vol.Required(CONF_HOST): cv.string})
 
 
-@callback
-def async_setup_service(hass, host, device):
+async def async_setup_service(hass, host, device):
     """Register a device for given host for use in services."""
     hass.data.setdefault(DOMAIN, {})[host] = device
 
     if hass.services.has_service(DOMAIN, SERVICE_LEARN):
         return
 
-    async def _learn_command(call):
+    async def async_learn_command(call):
         """Learn a packet from remote."""
 
         device = hass.data[DOMAIN][call.data[CONF_HOST]]
 
-        for retry in range(DEFAULT_RETRY):
-            try:
-                await hass.async_add_executor_job(device.enter_learning)
-                break
-            except (socket.timeout, ValueError):
-                try:
-                    await hass.async_add_executor_job(device.auth)
-                except socket.timeout:
-                    if retry == DEFAULT_RETRY - 1:
-                        _LOGGER.error("Failed to enter learning mode")
-                        return
+        try:
+            await device.async_request(device.api.enter_learning)
+        except BroadlinkException as err_msg:
+            _LOGGER.error("Failed to enter learning mode: %s", err_msg)
+            return
 
         _LOGGER.info("Press the key you want Home Assistant to learn")
         start_time = utcnow()
         while (utcnow() - start_time) < timedelta(seconds=20):
-            packet = await hass.async_add_executor_job(device.check_data)
-            if packet:
+            try:
+                packet = await device.async_request(device.api.check_data)
+            except ReadError:
+                await asyncio.sleep(1)
+            except BroadlinkException as err_msg:
+                _LOGGER.error("Failed to learn: %s", err_msg)
+                return
+            else:
                 data = b64encode(packet).decode("utf8")
                 log_msg = f"Received packet is: {data}"
                 _LOGGER.info(log_msg)
                 hass.components.persistent_notification.async_create(
                     log_msg, title="Broadlink switch"
                 )
                 return
-            await asyncio.sleep(1)
-        _LOGGER.error("No signal was received")
+        _LOGGER.error("Failed to learn: No signal received")
         hass.components.persistent_notification.async_create(
             "No signal was received", title="Broadlink switch"
         )
 
     hass.services.async_register(
-        DOMAIN, SERVICE_LEARN, _learn_command, schema=SERVICE_LEARN_SCHEMA
+        DOMAIN, SERVICE_LEARN, async_learn_command, schema=SERVICE_LEARN_SCHEMA
     )
 
-    async def _send_packet(call):
+    async def async_send_packet(call):
         """Send a packet."""
         device = hass.data[DOMAIN][call.data[CONF_HOST]]
         packets = call.data[CONF_PACKET]
         for packet in packets:
-            for retry in range(DEFAULT_RETRY):
-                try:
-                    await hass.async_add_executor_job(device.send_data, packet)
-                    break
-                except (socket.timeout, ValueError):
-                    try:
-                        await hass.async_add_executor_job(device.auth)
-                    except socket.timeout:
-                        if retry == DEFAULT_RETRY - 1:
-                            _LOGGER.error("Failed to send packet to device")
+            try:
+                await device.async_request(device.api.send_data, packet)
+            except BroadlinkException as err_msg:
+                _LOGGER.error("Failed to send packet: %s", err_msg)
+                return
 
     hass.services.async_register(
-        DOMAIN, SERVICE_SEND, _send_packet, schema=SERVICE_SEND_SCHEMA
+        DOMAIN, SERVICE_SEND, async_send_packet, schema=SERVICE_SEND_SCHEMA
     )

homeassistant/components/broadlink/device.py

@@ -0,0 +1,57 @@
+"""Support for Broadlink devices."""
+from functools import partial
+import logging
+
+from broadlink.exceptions import (
+    AuthorizationError,
+    BroadlinkException,
+    ConnectionClosedError,
+    DeviceOfflineError,
+)
+
+from .const import DEFAULT_RETRY
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class BroadlinkDevice:
+    """Manages a Broadlink device."""
+
+    def __init__(self, hass, api):
+        """Initialize the device."""
+        self.hass = hass
+        self.api = api
+        self.available = None
+
+    async def async_connect(self):
+        """Connect to the device."""
+        try:
+            await self.hass.async_add_executor_job(self.api.auth)
+        except BroadlinkException as err_msg:
+            if self.available:
+                self.available = False
+                _LOGGER.warning(
+                    "Disconnected from device at %s: %s", self.api.host[0], err_msg
+                )
+            return False
+        else:
+            if not self.available:
+                if self.available is not None:
+                    _LOGGER.warning("Connected to device at %s", self.api.host[0])
+                self.available = True
+            return True
+
+    async def async_request(self, function, *args, **kwargs):
+        """Send a request to the device."""
+        partial_function = partial(function, *args, **kwargs)
+        for attempt in range(DEFAULT_RETRY):
+            try:
+                result = await self.hass.async_add_executor_job(partial_function)
+            except (AuthorizationError, ConnectionClosedError, DeviceOfflineError):
+                if attempt == DEFAULT_RETRY - 1 or not await self.async_connect():
+                    raise
+            else:
+                if not self.available:
+                    self.available = True
+                    _LOGGER.warning("Connected to device at %s", self.api.host[0])
+                return result

homeassistant/components/broadlink/manifest.json

@@ -2,6 +2,6 @@
   "domain": "broadlink",
   "name": "Broadlink",
   "documentation": "https://www.home-assistant.io/integrations/broadlink",
-  "requirements": ["broadlink==0.13.2"],
+  "requirements": ["broadlink==0.14.0"],
   "codeowners": ["@danielhiversen", "@felipediel"]
 }

homeassistant/components/broadlink/remote.py

@@ -9,6 +9,12 @@
 import logging
 
 import broadlink as blk
+from broadlink.exceptions import (
+    AuthorizationError,
+    BroadlinkException,
+    DeviceOfflineError,
+    ReadError,
+)
 import voluptuous as vol
 
 from homeassistant.components.remote import (
@@ -36,11 +42,11 @@
     DEFAULT_LEARNING_TIMEOUT,
     DEFAULT_NAME,
     DEFAULT_PORT,
-    DEFAULT_RETRY,
     DEFAULT_TIMEOUT,
     RM4_TYPES,
     RM_TYPES,
 )
+from .device import BroadlinkDevice
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -103,17 +109,16 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
     else:
         api = blk.rm4((host, DEFAULT_PORT), mac_addr, None)
     api.timeout = timeout
+    device = BroadlinkDevice(hass, api)
+
     code_storage = Store(hass, CODE_STORAGE_VERSION, f"broadlink_{unique_id}_codes")
     flag_storage = Store(hass, FLAG_STORAGE_VERSION, f"broadlink_{unique_id}_flags")
-    remote = BroadlinkRemote(name, unique_id, api, code_storage, flag_storage)
 
-    connected, loaded = (False, False)
-    try:
-        connected, loaded = await asyncio.gather(
-            hass.async_add_executor_job(api.auth), remote.async_load_storage_files()
-        )
-    except OSError:
-        pass
+    remote = BroadlinkRemote(name, unique_id, device, code_storage, flag_storage)
+
+    connected, loaded = await asyncio.gather(
+        device.async_connect(), remote.async_load_storage_files()
+    )
     if not connected:
         hass.data[DOMAIN][COMPONENT].remove(unique_id)
         raise PlatformNotReady
@@ -127,11 +132,11 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
 class BroadlinkRemote(RemoteEntity):
     """Representation of a Broadlink remote."""
 
-    def __init__(self, name, unique_id, api, code_storage, flag_storage):
+    def __init__(self, name, unique_id, device, code_storage, flag_storage):
         """Initialize the remote."""
+        self.device = device
         self._name = name
         self._unique_id = unique_id
-        self._api = api
         self._code_storage = code_storage
         self._flag_storage = flag_storage
         self._codes = {}
@@ -157,7 +162,7 @@ def is_on(self):
     @property
     def available(self):
         """Return True if the remote is available."""
-        return self._available
+        return self.device.available
 
     @property
     def supported_features(self):
@@ -182,9 +187,9 @@ async def async_turn_off(self, **kwargs):
         self._state = False
 
     async def async_update(self):
-        """Update the availability of the remote."""
+        """Update the availability of the device."""
         if not self.available:
-            await self._async_connect()
+            await self.device.async_connect()
 
     async def async_load_storage_files(self):
         """Load codes and toggle flags from storage files."""
@@ -213,8 +218,10 @@ async def async_send_command(self, command, **kwargs):
                 should_delay = await self._async_send_code(
                     cmd, device, delay if should_delay else 0
                 )
-            except ConnectionError:
+            except (AuthorizationError, DeviceOfflineError):
                 break
+            except BroadlinkException:
+                pass
 
         self._flag_storage.async_delay_save(self.get_flags, FLAG_SAVE_DELAY)
 
@@ -227,7 +234,7 @@ async def _async_send_code(self, command, device, delay):
         try:
             code = self._codes[device][command]
         except KeyError:
-            _LOGGER.error("Failed to send '%s/%s': command not found", command, device)
+            _LOGGER.error("Failed to send '%s/%s': Command not found", command, device)
             return False
 
         if isinstance(code, list):
@@ -238,12 +245,14 @@ async def _async_send_code(self, command, device, delay):
         await asyncio.sleep(delay)
 
         try:
-            await self._async_attempt(self._api.send_data, data_packet(code))
+            await self.device.async_request(
+                self.device.api.send_data, data_packet(code)
+            )
         except ValueError:
-            _LOGGER.error("Failed to send '%s/%s': invalid code", command, device)
+            _LOGGER.error("Failed to send '%s/%s': Invalid code", command, device)
             return False
-        except ConnectionError:
-            _LOGGER.error("Failed to send '%s/%s': remote is offline", command, device)
+        except BroadlinkException as err_msg:
+            _LOGGER.error("Failed to send '%s/%s': %s", command, device, err_msg)
             raise
 
         if should_alternate:
@@ -268,8 +277,10 @@ async def async_learn_command(self, **kwargs):
                 should_store |= await self._async_learn_code(
                     command, device, toggle, timeout
                 )
-            except ConnectionError:
+            except (AuthorizationError, DeviceOfflineError):
                 break
+            except BroadlinkException:
+                pass
 
         if should_store:
             await self._code_storage.async_save(self._codes)
@@ -287,22 +298,19 @@ async def _async_learn_code(self, command, device, toggle, timeout):
                     await self._async_capture_code(command, timeout),
                     await self._async_capture_code(command, timeout),
                 ]
-        except (ValueError, TimeoutError):
-            _LOGGER.error(
-                "Failed to learn '%s/%s': no signal received", command, device
-            )
+        except TimeoutError:
+            _LOGGER.error("Failed to learn '%s/%s': No code received", command, device)
             return False
-        except ConnectionError:
-            _LOGGER.error("Failed to learn '%s/%s': remote is offline", command, device)
+        except BroadlinkException as err_msg:
+            _LOGGER.error("Failed to learn '%s/%s': %s", command, device, err_msg)
             raise
 
         self._codes.setdefault(device, {}).update({command: code})
-
         return True
 
     async def _async_capture_code(self, command, timeout):
         """Enter learning mode and capture a code from a remote."""
-        await self._async_attempt(self._api.enter_learning)
+        await self.device.async_request(self.device.api.enter_learning)
 
         self.hass.components.persistent_notification.async_create(
             f"Press the '{command}' button.",
@@ -313,44 +321,18 @@ async def _async_capture_code(self, command, timeout):
         code = None
         start_time = utcnow()
         while (utcnow() - start_time) < timedelta(seconds=timeout):
-            code = await self.hass.async_add_executor_job(self._api.check_data)
-            if code:
+            try:
+                code = await self.device.async_request(self.device.api.check_data)
+            except ReadError:
+                await asyncio.sleep(1)
+            else:
                 break
-            await asyncio.sleep(1)
 
         self.hass.components.persistent_notification.async_dismiss(
             notification_id="learn_command"
         )
 
-        if not code:
+        if code is None:
             raise TimeoutError
-        if all(not value for value in code):
-            raise ValueError
 
         return b64encode(code).decode("utf8")
-
-    async def _async_attempt(self, function, *args):
-        """Retry a socket-related function until it succeeds."""
-        for retry in range(DEFAULT_RETRY):
-            if retry and not await self._async_connect():
-                continue
-            try:
-                await self.hass.async_add_executor_job(function, *args)
-            except OSError:
-                continue
-            return
-        raise ConnectionError
-
-    async def _async_connect(self):
-        """Connect to the remote."""
-        try:
-            auth = await self.hass.async_add_executor_job(self._api.auth)
-        except OSError:
-            auth = False
-        if auth and not self._available:
-            _LOGGER.warning("Connected to the remote")
-            self._available = True
-        elif not auth and self._available:
-            _LOGGER.warning("Disconnected from the remote")
-            self._available = False
-        return auth