Merge pull request #228 from marcelvriend/feature/refactor-legacy-code

Bring code more up to current

.devcontainer/configuration.yaml

@@ -4,4 +4,4 @@ logger:
   default: info
   logs:
     custom_components.grocy: debug
-
+    pygrocy.grocy_api_client: debug

custom_components/grocy/__init__.py

@@ -4,151 +4,104 @@
 For more details about this integration, please refer to
 https://github.com/custom-components/grocy
 """
+from __future__ import annotations
+
 import logging
-from datetime import timedelta
 from typing import Any, List
 
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.core import Config, HomeAssistant
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
-from pygrocy import Grocy
-
-from .helpers import extract_base_url_and_path
+from homeassistant.core import HomeAssistant
 
 from .const import (
-    CONF_API_KEY,
-    CONF_PORT,
-    CONF_URL,
-    CONF_VERIFY_SSL,
+    ATTR_CHORES,
+    ATTR_EXPIRED_PRODUCTS,
+    ATTR_EXPIRING_PRODUCTS,
+    ATTR_MEAL_PLAN,
+    ATTR_MISSING_PRODUCTS,
+    ATTR_OVERDUE_CHORES,
+    ATTR_OVERDUE_TASKS,
+    ATTR_SHOPPING_LIST,
+    ATTR_STOCK,
+    ATTR_TASKS,
     DOMAIN,
-    GrocyEntityType,
     PLATFORMS,
     STARTUP_MESSAGE,
 )
-from .grocy_data import GrocyData, async_setup_image_api
+from .coordinator import GrocyDataUpdateCoordinator
+from .grocy_data import GrocyData, async_setup_endpoint_for_image_proxy
 from .services import async_setup_services, async_unload_services
 
-SCAN_INTERVAL = timedelta(seconds=30)
-
 _LOGGER = logging.getLogger(__name__)
 
 
 async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
     """Set up this integration using UI."""
-    hass.data.setdefault(DOMAIN, {})
     _LOGGER.info(STARTUP_MESSAGE)
 
-    coordinator = GrocyDataUpdateCoordinator(
-        hass,
-        config_entry.data[CONF_URL],
-        config_entry.data[CONF_API_KEY],
-        config_entry.data[CONF_PORT],
-        config_entry.data[CONF_VERIFY_SSL],
+    coordinator: GrocyDataUpdateCoordinator = GrocyDataUpdateCoordinator(hass)
+    coordinator.available_entities = await _async_get_available_entities(
+        coordinator.grocy_data
     )
-
     await coordinator.async_config_entry_first_refresh()
-
+    hass.data.setdefault(DOMAIN, {})
     hass.data[DOMAIN] = coordinator
 
-    for platform in PLATFORMS:
-        hass.async_add_job(
-            hass.config_entries.async_forward_entry_setup(config_entry, platform)
-        )
-
+    hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
     await async_setup_services(hass, config_entry)
-
-    # Setup http endpoint for proxying images from grocy
-    await async_setup_image_api(hass, config_entry.data)
+    await async_setup_endpoint_for_image_proxy(hass, config_entry.data)
 
     return True
 
 
-class GrocyDataUpdateCoordinator(DataUpdateCoordinator):
-    """Class to manage fetching data from the API."""
-
-    def __init__(self, hass, url, api_key, port_number, verify_ssl):
-        """Initialize."""
-        super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL)
-        (base_url, path) = extract_base_url_and_path(url)
-        self.api = Grocy(
-            base_url, api_key, path=path, port=port_number, verify_ssl=verify_ssl
-        )
-        self.entities = []
-        self.data = {}
-
-    async def _async_update_data(self):
-        """Update data via library."""
-        grocy_data = GrocyData(self.hass, self.api)
-        data = {}
-        features = await async_supported_features(grocy_data)
-        if not features:
-            raise UpdateFailed("No features enabled")
-
-        for entity in self.entities:
-            if not entity.enabled:
-                continue
-            if not entity.entity_type in features:
-                _LOGGER.debug(
-                    "You have enabled the entity for '%s', but this feature is not enabled in Grocy",
-                    entity.name,
-                )
-                continue
-
-            try:
-                data[entity.entity_type] = await grocy_data.async_update_data(
-                    entity.entity_type
-                )
-            except Exception as exception:  # pylint: disable=broad-except
-                _LOGGER.error(
-                    "Update of %s failed with %s",
-                    entity.entity_type,
-                    exception,
-                )
-        return data
-
-
-async def async_supported_features(grocy_data: GrocyData) -> List[str]:
-    """Return a list of supported features."""
-    features = []
-    config = await grocy_data.async_get_config()
-    if config:
-        if is_enabled_grocy_feature(config, "FEATURE_FLAG_STOCK"):
-            features.append(GrocyEntityType.STOCK)
-            features.append(GrocyEntityType.PRODUCTS)
-            features.append(GrocyEntityType.MISSING_PRODUCTS)
-            features.append(GrocyEntityType.EXPIRED_PRODUCTS)
-            features.append(GrocyEntityType.EXPIRING_PRODUCTS)
-
-        if is_enabled_grocy_feature(config, "FEATURE_FLAG_SHOPPINGLIST"):
-            features.append(GrocyEntityType.SHOPPING_LIST)
-
-        if is_enabled_grocy_feature(config, "FEATURE_FLAG_TASKS"):
-            features.append(GrocyEntityType.TASKS)
-            features.append(GrocyEntityType.OVERDUE_TASKS)
-
-        if is_enabled_grocy_feature(config, "FEATURE_FLAG_CHORES"):
-            features.append(GrocyEntityType.CHORES)
-            features.append(GrocyEntityType.OVERDUE_CHORES)
-
-        if is_enabled_grocy_feature(config, "FEATURE_FLAG_RECIPES"):
-            features.append(GrocyEntityType.MEAL_PLAN)
-
-    return features
-
-
-def is_enabled_grocy_feature(grocy_config: Any, feature_setting_key: str) -> bool:
+async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    await async_unload_services(hass)
+    if unloaded := await hass.config_entries.async_unload_platforms(
+        config_entry, PLATFORMS
+    ):
+        del hass.data[DOMAIN]
+
+    return unloaded
+
+
+async def _async_get_available_entities(grocy_data: GrocyData) -> List[str]:
+    """Return a list of available entities based on enabled Grocy features."""
+    available_entities = []
+    grocy_config = await grocy_data.async_get_config()
+    if grocy_config:
+        if _is_enabled_grocy_feature(grocy_config, "FEATURE_FLAG_STOCK"):
+            available_entities.append(ATTR_STOCK)
+            available_entities.append(ATTR_MISSING_PRODUCTS)
+            available_entities.append(ATTR_EXPIRED_PRODUCTS)
+            available_entities.append(ATTR_EXPIRING_PRODUCTS)
+
+        if _is_enabled_grocy_feature(grocy_config, "FEATURE_FLAG_SHOPPINGLIST"):
+            available_entities.append(ATTR_SHOPPING_LIST)
+
+        if _is_enabled_grocy_feature(grocy_config, "FEATURE_FLAG_TASKS"):
+            available_entities.append(ATTR_TASKS)
+            available_entities.append(ATTR_OVERDUE_TASKS)
+
+        if _is_enabled_grocy_feature(grocy_config, "FEATURE_FLAG_CHORES"):
+            available_entities.append(ATTR_CHORES)
+            available_entities.append(ATTR_OVERDUE_CHORES)
+
+        if _is_enabled_grocy_feature(grocy_config, "FEATURE_FLAG_RECIPES"):
+            available_entities.append(ATTR_MEAL_PLAN)
+
+    _LOGGER.debug("Available entities: %s", available_entities)
+
+    return available_entities
+
+
+def _is_enabled_grocy_feature(grocy_config: Any, feature_setting_key: str) -> bool:
     """
     Return whether the Grocy feature is enabled or not, default is enabled.
     Setting value received from Grocy can be a str or bool.
     """
     feature_setting_value = grocy_config[feature_setting_key]
-    return feature_setting_value not in (False, "0")
-
-
-async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
-    """Unload a config entry."""
-    await async_unload_services(hass)
-    if unloaded := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
-        del hass.data[DOMAIN]
+    _LOGGER.debug(
+        "Grocy feature '%s' has value '%s'.", feature_setting_key, feature_setting_value
+    )
 
-    return unloaded
+    return feature_setting_value not in (False, "0")

custom_components/grocy/binary_sensor.py

@@ -1,45 +1,124 @@
 """Binary sensor platform for Grocy."""
+from __future__ import annotations
+
 import logging
-from homeassistant.components.binary_sensor import BinarySensorEntity
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Any, List
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-# pylint: disable=relative-beyond-top-level
 from .const import (
+    ATTR_EXPIRED_PRODUCTS,
+    ATTR_EXPIRING_PRODUCTS,
+    ATTR_MISSING_PRODUCTS,
+    ATTR_OVERDUE_CHORES,
+    ATTR_OVERDUE_TASKS,
     DOMAIN,
-    GrocyEntityType,
 )
+from .coordinator import GrocyDataUpdateCoordinator
 from .entity import GrocyEntity
 
 _LOGGER = logging.getLogger(__name__)
-BINARY_SENSOR_TYPES = [
-    GrocyEntityType.EXPIRED_PRODUCTS,
-    GrocyEntityType.EXPIRING_PRODUCTS,
-    GrocyEntityType.MISSING_PRODUCTS,
-    GrocyEntityType.OVERDUE_CHORES,
-    GrocyEntityType.OVERDUE_TASKS,
-]
 
 
-async def async_setup_entry(hass, entry, async_add_entities):
-    """Setup binary_sensor platform."""
-    coordinator = hass.data[DOMAIN]
-
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+):
+    """Setup binary sensor platform."""
+    coordinator: GrocyDataUpdateCoordinator = hass.data[DOMAIN]
     entities = []
-    for binary_sensor in BINARY_SENSOR_TYPES:
-        _LOGGER.debug("Adding %s binary sensor", binary_sensor)
-        entity = GrocyBinarySensor(coordinator, entry, binary_sensor)
-        coordinator.entities.append(entity)
-        entities.append(entity)
+    for description in BINARY_SENSORS:
+        if description.exists_fn(coordinator.available_entities):
+            entity = GrocyBinarySensorEntity(coordinator, description, config_entry)
+            coordinator.entities.append(entity)
+            entities.append(entity)
+        else:
+            _LOGGER.debug(
+                "Entity description '%s' is not available.",
+                description.key,
+            )
 
     async_add_entities(entities, True)
 
 
-class GrocyBinarySensor(GrocyEntity, BinarySensorEntity):
-    """Grocy binary_sensor class."""
+@dataclass
+class GrocyBinarySensorEntityDescription(BinarySensorEntityDescription):
+    """Grocy binary sensor entity description."""
+
+    attributes_fn: Callable[[List[Any]], Mapping[str, Any] | None] = lambda _: None
+    exists_fn: Callable[[List[str]], bool] = lambda _: True
+    entity_registry_enabled_default: bool = False
+
+
+BINARY_SENSORS: tuple[GrocyBinarySensorEntityDescription, ...] = (
+    GrocyBinarySensorEntityDescription(
+        key=ATTR_EXPIRED_PRODUCTS,
+        name="Grocy expired products",
+        icon="mdi:delete-alert-outline",
+        exists_fn=lambda entities: ATTR_EXPIRED_PRODUCTS in entities,
+        attributes_fn=lambda data: {
+            "expired_products": [x.as_dict() for x in data],
+            "count": len(data),
+        },
+    ),
+    GrocyBinarySensorEntityDescription(
+        key=ATTR_EXPIRING_PRODUCTS,
+        name="Grocy expiring products",
+        icon="mdi:clock-fast",
+        exists_fn=lambda entities: ATTR_EXPIRING_PRODUCTS in entities,
+        attributes_fn=lambda data: {
+            "expiring_products": [x.as_dict() for x in data],
+            "count": len(data),
+        },
+    ),
+    GrocyBinarySensorEntityDescription(
+        key=ATTR_MISSING_PRODUCTS,
+        name="Grocy missing products",
+        icon="mdi:flask-round-bottom-empty-outline",
+        exists_fn=lambda entities: ATTR_MISSING_PRODUCTS in entities,
+        attributes_fn=lambda data: {
+            "missing_products": [x.as_dict() for x in data],
+            "count": len(data),
+        },
+    ),
+    GrocyBinarySensorEntityDescription(
+        key=ATTR_OVERDUE_CHORES,
+        name="Grocy overdue chores",
+        icon="mdi:alert-circle-check-outline",
+        exists_fn=lambda entities: ATTR_OVERDUE_CHORES in entities,
+        attributes_fn=lambda data: {
+            "overdue_chores": [x.as_dict() for x in data],
+            "count": len(data),
+        },
+    ),
+    GrocyBinarySensorEntityDescription(
+        key=ATTR_OVERDUE_TASKS,
+        name="Grocy overdue tasks",
+        icon="mdi:alert-circle-check-outline",
+        exists_fn=lambda entities: ATTR_OVERDUE_TASKS in entities,
+        attributes_fn=lambda data: {
+            "overdue_tasks": [x.as_dict() for x in data],
+            "count": len(data),
+        },
+    ),
+)
+
+
+class GrocyBinarySensorEntity(GrocyEntity, BinarySensorEntity):
+    """Grocy binary sensor entity definition."""
 
     @property
-    def is_on(self):
-        """Return true if the binary_sensor is on."""
-        if not self.entity_data:
-            return
+    def is_on(self) -> bool | None:
+        """Return true if the binary sensor is on."""
+        entity_data = self.coordinator.data.get(self.entity_description.key, None)
 
-        return len(self.entity_data) > 0
+        return len(entity_data) > 0 if entity_data else False