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