refactor: handle binary sensors updates in the coordinator (#1003)

custom_components/mail_and_packages/__init__.py

@@ -3,6 +3,7 @@
 import asyncio
 import logging
 from datetime import timedelta
+import os
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST, CONF_RESOURCES
@@ -11,6 +12,9 @@
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import (
+    ATTR_AMAZON_IMAGE,
+    ATTR_IMAGE_NAME,
+    ATTR_IMAGE_PATH,
     CONF_AMAZON_DAYS,
     CONF_AMAZON_DOMAIN,
     CONF_AMAZON_FWDS,
@@ -29,7 +33,7 @@
     PLATFORMS,
     VERSION,
 )
-from .helpers import process_emails
+from .helpers import default_image_path, hash_file, process_emails
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -210,4 +214,52 @@ async def _async_update_data(self):
 
             if data:
                 self._data = data
+                await self._binary_sensor_update()
             return self._data
+
+    async def _binary_sensor_update(self):
+        """Update binary sensor states."""
+        attributes = (ATTR_IMAGE_NAME, ATTR_IMAGE_PATH)
+        if set(attributes).issubset(self._data.keys()):
+            image = self._data[ATTR_IMAGE_NAME]
+            path = default_image_path(self.hass, self.config)
+            usps_image = f"{path}/{image}"
+            usps_none = f"{os.path.dirname(__file__)}/mail_none.gif"
+            usps_check = os.path.exists(usps_image)
+            _LOGGER.debug("USPS Check: %s", usps_check)
+            if usps_check:
+                image_hash = await self.hass.async_add_executor_job(
+                    hash_file, usps_image
+                )
+                none_hash = await self.hass.async_add_executor_job(hash_file, usps_none)
+
+                _LOGGER.debug("USPS Image hash: %s", image_hash)
+                _LOGGER.debug("USPS None hash: %s", none_hash)
+
+                if image_hash != none_hash:
+                    self._data["usps_update"] = True
+                else:
+                    self._data["usps_update"] = False
+        attributes = (ATTR_AMAZON_IMAGE, ATTR_IMAGE_PATH)
+        if set(attributes).issubset(self._data.keys()):
+            image = self._data[ATTR_AMAZON_IMAGE]
+            path = f"{default_image_path(self.hass, self.config)}/amazon/"
+            amazon_image = f"{path}{image}"
+            amazon_none = f"{os.path.dirname(__file__)}/no_deliveries.jpg"
+            amazon_check = os.path.exists(amazon_image)
+            _LOGGER.debug("Amazon Check: %s", amazon_check)
+            if amazon_check:
+                image_hash = await self.hass.async_add_executor_job(
+                    hash_file, amazon_image
+                )
+                none_hash = await self.hass.async_add_executor_job(
+                    hash_file, amazon_none
+                )
+
+                _LOGGER.debug("Amazon Image hash: %s", image_hash)
+                _LOGGER.debug("Amazon None hash: %s", none_hash)
+
+                if image_hash != none_hash:
+                    self._data["amazon_update"] = True
+                else:
+                    self._data["amazon_update"] = False

custom_components/mail_and_packages/binary_sensor.py

@@ -1,8 +1,6 @@
 """Binary sensors for Mail and Packages."""
 
-import asyncio
 import logging
-import os
 
 from homeassistant.components.binary_sensor import (
     BinarySensorEntity,
@@ -16,15 +14,11 @@
 )
 
 from .const import (
-    ATTR_AMAZON_IMAGE,
-    ATTR_IMAGE_NAME,
-    ATTR_IMAGE_PATH,
     BINARY_SENSORS,
     COORDINATOR,
     DOMAIN,
     VERSION,
 )
-from .helpers import default_image_path, hash_file
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -85,47 +79,6 @@ def available(self) -> bool:
     @property
     def is_on(self) -> bool:
         """Return True if the image is updated."""
-        if self._type == "usps_update":
-            attributes = (ATTR_IMAGE_NAME, ATTR_IMAGE_PATH)
-            if set(attributes).issubset(self.coordinator.data.keys()):
-                image = self.coordinator.data[ATTR_IMAGE_NAME]
-                path = default_image_path(self.hass, self._config)
-                usps_image = f"{path}/{image}"
-                usps_none = f"{os.path.dirname(__file__)}/mail_none.gif"
-                usps_check = os.path.exists(usps_image)
-                _LOGGER.debug("USPS Check: %s", usps_check)
-                if usps_check:
-                    loop = asyncio.get_running_loop()
-                    image_hash = loop.run_in_executor(None, hash_file, usps_image)
-                    none_hash = loop.run_in_executor(None, hash_file, usps_none)
-
-                    _LOGGER.debug("USPS Image hash: %s", image_hash)
-                    _LOGGER.debug("USPS None hash: %s", none_hash)
-
-                    if image_hash != none_hash:
-                        return True
-                return False
-
-        if self._type == "amazon_update":
-            attributes = (ATTR_AMAZON_IMAGE, ATTR_IMAGE_PATH)
-            if set(attributes).issubset(self.coordinator.data.keys()):
-                image = self.coordinator.data[ATTR_AMAZON_IMAGE]
-                path = f"{default_image_path(self.hass, self._config)}/amazon/"
-                amazon_image = f"{path}{image}"
-                amazon_none = f"{os.path.dirname(__file__)}/no_deliveries.jpg"
-                amazon_check = os.path.exists(amazon_image)
-                _LOGGER.debug("Amazon Check: %s", amazon_check)
-                if amazon_check:
-                    loop = asyncio.get_running_loop()
-                    image_hash = loop.run_in_executor(None, hash_file, amazon_image)
-                    none_hash = loop.run_in_executor(None, hash_file, amazon_none)
-
-                    _LOGGER.debug("Amazon Image hash: %s", image_hash)
-                    _LOGGER.debug("Amazon None hash: %s", none_hash)
-
-                    if image_hash != none_hash:
-                        return True
-                return False
         if self._type in self.coordinator.data.keys():
             _LOGGER.debug(
                 "binary_sensor: %s value: %s",

tests/test_helpers.py

@@ -1140,10 +1140,9 @@ async def test_email_search_none(mock_imap_search_error_none, caplog):
 
 @pytest.mark.asyncio
 async def test_amazon_shipped_fwd(hass, mock_imap_amazon_fwd, caplog):
-    result = get_items(mock_imap_amazon_fwd, "order", fwds="[email protected]" ,the_domain="amazon.com")
-    assert (
-        "Amazon email list: ['[email protected]', 'amazon.com']"
-        in caplog.text
+    result = get_items(
+        mock_imap_amazon_fwd, "order", fwds="[email protected]", the_domain="amazon.com"
     )
+    assert "Amazon email list: ['[email protected]', 'amazon.com']" in caplog.text
     assert result == ["123-1234567-1234567"]
     assert "First pass: Tuesday, January 11" in caplog.text