diff --git a/custom_components/grocy/__init__.py b/custom_components/grocy/__init__.py index 014eb48..c32d9ae 100644 --- a/custom_components/grocy/__init__.py +++ b/custom_components/grocy/__init__.py @@ -1,5 +1,4 @@ -""" -Custom integration to integrate Grocy with Home Assistant. +"""Custom integration to integrate Grocy with Home Assistant. For more details about this integration, please refer to https://github.com/custom-components/grocy @@ -7,7 +6,6 @@ from __future__ import annotations import logging -from typing import List from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant @@ -67,7 +65,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> return unloaded -async def _async_get_available_entities(grocy_data: GrocyData) -> List[str]: +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() diff --git a/custom_components/grocy/binary_sensor.py b/custom_components/grocy/binary_sensor.py index c0becbc..99a7e4a 100644 --- a/custom_components/grocy/binary_sensor.py +++ b/custom_components/grocy/binary_sensor.py @@ -1,10 +1,10 @@ """Binary sensor platform for Grocy.""" from __future__ import annotations -import logging -from collections.abc import Callable, Mapping +from collections.abc import Callable from dataclasses import dataclass -from typing import Any, List +import logging +from typing import Any from homeassistant.components.binary_sensor import ( BinarySensorEntity, @@ -24,7 +24,7 @@ ATTR_OVERDUE_TASKS, DOMAIN, ) -from .coordinator import GrocyDataUpdateCoordinator +from .coordinator import GrocyCoordinatorData, GrocyDataUpdateCoordinator from .entity import GrocyEntity _LOGGER = logging.getLogger(__name__) @@ -35,7 +35,7 @@ async def async_setup_entry( config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ): - """Setup binary sensor platform.""" + """Initialize binary sensor platform.""" coordinator: GrocyDataUpdateCoordinator = hass.data[DOMAIN] entities = [] for description in BINARY_SENSORS: @@ -56,8 +56,8 @@ async def async_setup_entry( 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 + attributes_fn: Callable[[list[Any]], GrocyCoordinatorData | None] = lambda _: None + exists_fn: Callable[[list[str]], bool] = lambda _: True entity_registry_enabled_default: bool = False @@ -141,6 +141,6 @@ class GrocyBinarySensorEntity(GrocyEntity, BinarySensorEntity): @property 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) + entity_data = self.coordinator.data[self.entity_description.key] return len(entity_data) > 0 if entity_data else False diff --git a/custom_components/grocy/config_flow.py b/custom_components/grocy/config_flow.py index aabd540..97553f0 100644 --- a/custom_components/grocy/config_flow.py +++ b/custom_components/grocy/config_flow.py @@ -1,10 +1,11 @@ """Adds config flow for Grocy.""" -import logging from collections import OrderedDict +import logging +from pygrocy import Grocy import voluptuous as vol + from homeassistant import config_entries -from pygrocy import Grocy from .const import ( CONF_API_KEY, diff --git a/custom_components/grocy/const.py b/custom_components/grocy/const.py index 24824e8..88f0ffc 100644 --- a/custom_components/grocy/const.py +++ b/custom_components/grocy/const.py @@ -8,7 +8,7 @@ ISSUE_URL: Final = "https://github.com/custom-components/grocy/issues" -PLATFORMS: Final = ["binary_sensor", "sensor"] +PLATFORMS: Final = ["binary_sensor", "sensor", "todo"] SCAN_INTERVAL = timedelta(seconds=30) diff --git a/custom_components/grocy/coordinator.py b/custom_components/grocy/coordinator.py index 4aac982..2791b89 100644 --- a/custom_components/grocy/coordinator.py +++ b/custom_components/grocy/coordinator.py @@ -1,13 +1,19 @@ """Data update coordinator for Grocy.""" from __future__ import annotations +from dataclasses import dataclass import logging -from typing import Any, Dict, List + +from pygrocy import Grocy +from pygrocy.data_models.battery import Battery +from pygrocy.data_models.chore import Chore +from pygrocy.data_models.meal_items import MealPlanItem +from pygrocy.data_models.product import Product, ShoppingListProduct +from pygrocy.data_models.task import Task from homeassistant.core import HomeAssistant from homeassistant.helpers.entity import Entity from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed -from pygrocy import Grocy from .const import ( CONF_API_KEY, @@ -18,12 +24,35 @@ SCAN_INTERVAL, ) from .grocy_data import GrocyData -from .helpers import extract_base_url_and_path +from .helpers import MealPlanItemWrapper, extract_base_url_and_path _LOGGER = logging.getLogger(__name__) -class GrocyDataUpdateCoordinator(DataUpdateCoordinator[Dict[str, Any]]): +@dataclass +class GrocyCoordinatorData: + batteries: list[Battery] | None = None + chores: list[Chore] | None = None + expired_products: list[Product] | None = None + expiring_products: list[Product] | None = None + meal_plan: list[MealPlanItemWrapper] | None = None + missing_products: list[Product] | None = None + overdue_batteries: list[Battery] | None = None + overdue_chores: list[Chore] | None = None + overdue_products: list[Product] | None = None + overdue_tasks: list[Task] | None = None + shopping_list: list[ShoppingListProduct] | None = None + stock: list[Product] | None = None + tasks: list[Task] | None = None + + def __setitem__(self, key, value): + setattr(self, key, value) + + def __getitem__(self, key: str): + return getattr(self, key) + + +class GrocyDataUpdateCoordinator(DataUpdateCoordinator[GrocyCoordinatorData]): """Grocy data update coordinator.""" def __init__( @@ -50,16 +79,15 @@ def __init__( ) self.grocy_data = GrocyData(hass, self.grocy_api) - self.available_entities: List[str] = [] - self.entities: List[Entity] = [] + self.available_entities: list[str] = [] + self.entities: list[Entity] = [] - async def _async_update_data(self) -> dict[str, Any]: + async def _async_update_data(self) -> GrocyCoordinatorData: """Fetch data.""" - data: dict[str, Any] = {} - + data = GrocyCoordinatorData() for entity in self.entities: if not entity.enabled: - _LOGGER.debug("Entity %s is disabled.", entity.entity_id) + _LOGGER.debug("Entity %s is disabled", entity.entity_id) continue try: diff --git a/custom_components/grocy/entity.py b/custom_components/grocy/entity.py index 06bbe4f..d2065a8 100644 --- a/custom_components/grocy/entity.py +++ b/custom_components/grocy/entity.py @@ -2,8 +2,6 @@ from __future__ import annotations import json -from collections.abc import Mapping -from typing import Any from homeassistant.config_entries import ConfigEntry from homeassistant.helpers.device_registry import DeviceEntryType @@ -11,7 +9,7 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN, NAME, VERSION -from .coordinator import GrocyDataUpdateCoordinator +from .coordinator import GrocyCoordinatorData, GrocyDataUpdateCoordinator from .json_encoder import CustomJSONEncoder @@ -42,9 +40,9 @@ def device_info(self) -> DeviceInfo: ) @property - def extra_state_attributes(self) -> Mapping[str, Any] | None: + def extra_state_attributes(self) -> GrocyCoordinatorData | None: """Return the extra state attributes.""" - data = self.coordinator.data.get(self.entity_description.key) + data = self.coordinator.data[self.entity_description.key] if data and hasattr(self.entity_description, "attributes_fn"): return json.loads( json.dumps( diff --git a/custom_components/grocy/grocy_data.py b/custom_components/grocy/grocy_data.py index a4067cd..6e16eb5 100644 --- a/custom_components/grocy/grocy_data.py +++ b/custom_components/grocy/grocy_data.py @@ -1,16 +1,18 @@ """Communication with Grocy API.""" from __future__ import annotations -import logging from datetime import datetime, timedelta -from typing import List +import logging from aiohttp import hdrs, web +from pygrocy import Grocy +from pygrocy.data_models.battery import Battery +from pygrocy.data_models.chore import Chore + from homeassistant.components.http import HomeAssistantView from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.aiohttp_client import async_get_clientsession -from pygrocy.data_models.battery import Battery from .const import ( ATTR_BATTERIES, @@ -38,7 +40,7 @@ class GrocyData: """Handles communication and gets the data.""" - def __init__(self, hass, api): + def __init__(self, hass: HomeAssistant, api: Grocy) -> None: # noqa: D107 """Initialize Grocy data.""" self.hass = hass self.api = api @@ -70,7 +72,7 @@ async def async_update_stock(self): async def async_update_chores(self): """Update chores data.""" - def wrapper(): + def wrapper() -> list[Chore]: return self.api.chores(True) return await self.hass.async_add_executor_job(wrapper) @@ -103,7 +105,9 @@ async def async_update_overdue_tasks(self): and_query_filter = [ f"due_date<{datetime.now().date()}", - # It's not possible to pass an empty value to Grocy, so use a regex that matches non-empty values to exclude empty str due_date. + # It's not possible to pass an empty value to Grocy + # so use a regex that matches non-empty values + # to exclude empty str due_date. r"due_date§.*\S.*", ] @@ -155,7 +159,8 @@ def wrapper(): async def async_update_meal_plan(self): """Update meal plan data.""" - # The >= condition is broken before Grocy 3.3.1. So use > to maintain backward compatibility. + # The >= condition is broken before Grocy 3.3.1. + # So use > to maintain backward compatibility. yesterday = datetime.now() - timedelta(1) query_filter = [f"day>{yesterday.date()}"] @@ -166,7 +171,7 @@ def wrapper(): return await self.hass.async_add_executor_job(wrapper) - async def async_update_batteries(self) -> List[Battery]: + async def async_update_batteries(self) -> list[Battery]: """Update batteries.""" def wrapper(): @@ -174,7 +179,7 @@ def wrapper(): return await self.hass.async_add_executor_job(wrapper) - async def async_update_overdue_batteries(self) -> List[Battery]: + async def async_update_overdue_batteries(self) -> list[Battery]: """Update overdue batteries.""" def wrapper(): @@ -187,7 +192,7 @@ def wrapper(): async def async_setup_endpoint_for_image_proxy( hass: HomeAssistant, config_entry: ConfigEntry ): - """Setup and register the image api for grocy images with HA.""" + """Do setup and register the image api for grocy images with HA.""" session = async_get_clientsession(hass) url = config_entry.get(CONF_URL) @@ -210,7 +215,7 @@ class GrocyPictureView(HomeAssistantView): url = "/api/grocy/{picture_type}/{filename}" name = "api:grocy:picture" - def __init__(self, session, base_url, api_key): + def __init__(self, session, base_url, api_key) -> None: # noqa: D107 self._session = session self._base_url = base_url self._api_key = api_key diff --git a/custom_components/grocy/helpers.py b/custom_components/grocy/helpers.py index c8d9bb0..b0c1a8e 100644 --- a/custom_components/grocy/helpers.py +++ b/custom_components/grocy/helpers.py @@ -2,13 +2,13 @@ from __future__ import annotations import base64 -from typing import Any, Dict, Tuple +from typing import Any from urllib.parse import urlparse from pygrocy.data_models.meal_items import MealPlanItem -def extract_base_url_and_path(url: str) -> Tuple[str, str]: +def extract_base_url_and_path(url: str) -> tuple[str, str]: """Extract the base url and path from a given URL.""" parsed_url = urlparse(url) @@ -18,7 +18,7 @@ def extract_base_url_and_path(url: str) -> Tuple[str, str]: class MealPlanItemWrapper: """Wrapper around the pygrocy MealPlanItem.""" - def __init__(self, meal_plan: MealPlanItem): + def __init__(self, meal_plan: MealPlanItem) -> None: # noqa: D107 self._meal_plan = meal_plan @property @@ -35,7 +35,7 @@ def picture_url(self) -> str | None: return f"/api/grocy/recipepictures/{str(b64name, 'utf-8')}" return None - def as_dict(self) -> Dict[str, Any]: + def as_dict(self) -> dict[str, Any]: """Return attributes for the pygrocy MealPlanItem object including picture URL.""" props = self.meal_plan.as_dict() props["picture_url"] = self.picture_url diff --git a/custom_components/grocy/sensor.py b/custom_components/grocy/sensor.py index de041fb..b5e3679 100644 --- a/custom_components/grocy/sensor.py +++ b/custom_components/grocy/sensor.py @@ -1,10 +1,9 @@ """Sensor platform for Grocy.""" from __future__ import annotations -import logging -from collections.abc import Callable, Mapping +from collections.abc import Callable from dataclasses import dataclass -from typing import Any, List +import logging from homeassistant.components.sensor import ( SensorEntity, @@ -30,7 +29,7 @@ PRODUCTS, TASKS, ) -from .coordinator import GrocyDataUpdateCoordinator +from .coordinator import GrocyCoordinatorData, GrocyDataUpdateCoordinator from .entity import GrocyEntity _LOGGER = logging.getLogger(__name__) @@ -41,7 +40,7 @@ async def async_setup_entry( config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ): - """Setup sensor platform.""" + """Do setup sensor platform.""" coordinator: GrocyDataUpdateCoordinator = hass.data[DOMAIN] entities = [] for description in SENSORS: @@ -51,7 +50,7 @@ async def async_setup_entry( entities.append(entity) else: _LOGGER.debug( - "Entity description '%s' is not available.", + "Entity description '%s' is not available", description.key, ) @@ -62,8 +61,8 @@ async def async_setup_entry( class GrocySensorEntityDescription(SensorEntityDescription): """Grocy sensor entity description.""" - attributes_fn: Callable[[List[Any]], Mapping[str, Any] | None] = lambda _: None - exists_fn: Callable[[List[str]], bool] = lambda _: True + attributes_fn: Callable[GrocyCoordinatorData | None] = lambda _: None + exists_fn: Callable[[list[str]], bool] = lambda _: True entity_registry_enabled_default: bool = False @@ -149,6 +148,6 @@ class GrocySensorEntity(GrocyEntity, SensorEntity): @property def native_value(self) -> StateType: """Return the value reported by the sensor.""" - entity_data = self.coordinator.data.get(self.entity_description.key, None) + entity_data = self.coordinator.data[self.entity_description.key] return len(entity_data) if entity_data else 0 diff --git a/custom_components/grocy/services.py b/custom_components/grocy/services.py index 30899e6..ee9ff24 100644 --- a/custom_components/grocy/services.py +++ b/custom_components/grocy/services.py @@ -1,10 +1,11 @@ """Grocy services.""" from __future__ import annotations +from pygrocy import EntityType, TransactionType import voluptuous as vol + from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, ServiceCall -from pygrocy import EntityType, TransactionType from .const import ATTR_CHORES, ATTR_TASKS, DOMAIN from .coordinator import GrocyDataUpdateCoordinator @@ -12,6 +13,7 @@ SERVICE_PRODUCT_ID = "product_id" SERVICE_AMOUNT = "amount" SERVICE_PRICE = "price" +SERVICE_SHOPPING_LIST_ID = "shopping_list_id" SERVICE_SPOILED = "spoiled" SERVICE_SUBPRODUCT_SUBSTITUTION = "allow_subproduct_substitution" SERVICE_TRANSACTION_TYPE = "transaction_type" @@ -145,7 +147,8 @@ async def async_setup_services( - hass: HomeAssistant, config_entry: ConfigEntry # pylint: disable=unused-argument + hass: HomeAssistant, + config_entry: ConfigEntry, # pylint: disable=unused-argument ) -> None: """Set up services for Grocy integration.""" coordinator: GrocyDataUpdateCoordinator = hass.data[DOMAIN] @@ -200,7 +203,9 @@ async def async_unload_services(hass: HomeAssistant) -> None: hass.services.async_remove(DOMAIN, service) -async def async_add_product_service(hass, coordinator, data): +async def async_add_product_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Add a product in Grocy.""" product_id = data[SERVICE_PRODUCT_ID] amount = data[SERVICE_AMOUNT] @@ -212,7 +217,9 @@ def wrapper(): await hass.async_add_executor_job(wrapper) -async def async_open_product_service(hass, coordinator, data): +async def async_open_product_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Open a product in Grocy.""" product_id = data[SERVICE_PRODUCT_ID] amount = data[SERVICE_AMOUNT] @@ -226,7 +233,9 @@ def wrapper(): await hass.async_add_executor_job(wrapper) -async def async_consume_product_service(hass, coordinator, data): +async def async_consume_product_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Consume a product in Grocy.""" product_id = data[SERVICE_PRODUCT_ID] amount = data[SERVICE_AMOUNT] @@ -251,7 +260,9 @@ def wrapper(): await hass.async_add_executor_job(wrapper) -async def async_execute_chore_service(hass, coordinator, data): +async def async_execute_chore_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Execute a chore in Grocy.""" chore_id = data[SERVICE_CHORE_ID] done_by = data.get(SERVICE_DONE_BY, "") @@ -264,7 +275,9 @@ def wrapper(): await _async_force_update_entity(coordinator, ATTR_CHORES) -async def async_complete_task_service(hass, coordinator, data): +async def async_complete_task_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Complete a task in Grocy.""" task_id = data[SERVICE_TASK_ID] @@ -275,7 +288,9 @@ def wrapper(): await _async_force_update_entity(coordinator, ATTR_TASKS) -async def async_add_generic_service(hass, coordinator, data): +async def async_add_generic_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Add a generic entity in Grocy.""" entity_type_raw = data.get(SERVICE_ENTITY_TYPE, None) entity_type = EntityType.TASKS @@ -289,10 +304,12 @@ def wrapper(): coordinator.grocy_api.add_generic(entity_type, data) await hass.async_add_executor_job(wrapper) - await post_generic_refresh(coordinator, entity_type); + await _post_generic_refresh(coordinator, entity_type) -async def async_update_generic_service(hass, coordinator, data): +async def async_update_generic_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Update a generic entity in Grocy.""" entity_type_raw = data.get(SERVICE_ENTITY_TYPE, None) entity_type = EntityType.TASKS @@ -308,31 +325,36 @@ def wrapper(): coordinator.grocy_api.update_generic(entity_type, object_id, data) await hass.async_add_executor_job(wrapper) - await post_generic_refresh(coordinator, entity_type); + await _post_generic_refresh(coordinator, entity_type) -async def async_delete_generic_service(hass, coordinator, data): +async def async_delete_generic_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Delete a generic entity in Grocy.""" entity_type_raw = data.get(SERVICE_ENTITY_TYPE, None) - entity_type = EntityType.TASKS - if entity_type_raw is not None: - entity_type = EntityType(entity_type_raw) + entity_type = ( + EntityType(entity_type_raw) if entity_type_raw is not None else EntityType.TASKS + ) - object_id = data[SERVICE_OBJECT_ID] + object_id = int(data[SERVICE_OBJECT_ID]) def wrapper(): coordinator.grocy_api.delete_generic(entity_type, object_id) await hass.async_add_executor_job(wrapper) - await post_generic_refresh(coordinator, entity_type); + await _post_generic_refresh(coordinator, entity_type) -async def post_generic_refresh(coordinator, entity_type): - if entity_type == "tasks" or entity_type == "chores": +async def _post_generic_refresh(coordinator: GrocyDataUpdateCoordinator, entity_type): + if entity_type in ("tasks", "chores"): await _async_force_update_entity(coordinator, entity_type) -async def async_consume_recipe_service(hass, coordinator, data): + +async def async_consume_recipe_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Consume a recipe in Grocy.""" recipe_id = data[SERVICE_RECIPE_ID] @@ -342,7 +364,25 @@ def wrapper(): await hass.async_add_executor_job(wrapper) -async def async_track_battery_service(hass, coordinator, data): +async def async_remove_product_in_shopping_list( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): + """Consume a recipe in Grocy.""" + product_id = data[SERVICE_PRODUCT_ID] + shopping_list_id = data[SERVICE_SHOPPING_LIST_ID] + amount = data[SERVICE_AMOUNT] + + def wrapper(): + coordinator.grocy_api.remove_product_in_shopping_list( + product_id, shopping_list_id, amount + ) + + await hass.async_add_executor_job(wrapper) + + +async def async_track_battery_service( + hass: HomeAssistant, coordinator: GrocyDataUpdateCoordinator, data +): """Track a battery in Grocy.""" battery_id = data[SERVICE_BATTERY_ID] diff --git a/custom_components/grocy/todo.py b/custom_components/grocy/todo.py new file mode 100644 index 0000000..def861d --- /dev/null +++ b/custom_components/grocy/todo.py @@ -0,0 +1,443 @@ +"""Todo platform for Grocy.""" +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +import datetime +import logging +from typing import Any + +from pygrocy.data_models.battery import Battery +from pygrocy.data_models.chore import Chore +from pygrocy.data_models.meal_items import MealPlanItem +from pygrocy.data_models.product import Product, ShoppingListProduct +from pygrocy.data_models.task import Task + +from homeassistant.components.todo import ( + TodoItem, + TodoItemStatus, + TodoListEntity, + TodoListEntityFeature, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity import EntityDescription +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .const import ( + ATTR_BATTERIES, + ATTR_CHORES, + ATTR_MEAL_PLAN, + ATTR_SHOPPING_LIST, + ATTR_STOCK, + ATTR_TASKS, + DOMAIN, +) +from .coordinator import GrocyCoordinatorData, GrocyDataUpdateCoordinator +from .entity import GrocyEntity +from .helpers import MealPlanItemWrapper +from .services import ( + SERVICE_AMOUNT, + SERVICE_BATTERY_ID, + SERVICE_CHORE_ID, + SERVICE_DATA, + SERVICE_DONE_BY, + SERVICE_ENTITY_TYPE, + SERVICE_OBJECT_ID, + SERVICE_PRODUCT_ID, + SERVICE_RECIPE_ID, + SERVICE_SHOPPING_LIST_ID, + SERVICE_SKIPPED, + SERVICE_TASK_ID, + async_add_generic_service, + async_complete_task_service, + async_consume_product_service, + async_consume_recipe_service, + async_delete_generic_service, + async_execute_chore_service, + async_remove_product_in_shopping_list, + async_track_battery_service, +) + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +): + """Do setup todo platform.""" + coordinator: GrocyDataUpdateCoordinator = hass.data[DOMAIN] + entities = [] + for description in TODOS: + if description.exists_fn(coordinator.available_entities): + entity = GrocyTodoListEntity(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) + + +@dataclass +class GrocyTodoListEntityDescription(EntityDescription): + """Grocy todo entity description.""" + + attributes_fn: Callable[[list[Any]], GrocyCoordinatorData | None] = lambda _: None + exists_fn: Callable[[list[str]], bool] = lambda _: True + entity_registry_enabled_default: bool = False + + +TODOS: tuple[GrocyTodoListEntityDescription, ...] = ( + GrocyTodoListEntityDescription( + key=ATTR_BATTERIES, + name="Grocy batteries", + icon="mdi:battery", + exists_fn=lambda entities: ATTR_BATTERIES in entities, + ), + GrocyTodoListEntityDescription( + key=ATTR_CHORES, + name="Grocy chores", + icon="mdi:broom", + exists_fn=lambda entities: ATTR_CHORES in entities, + ), + GrocyTodoListEntityDescription( + key=ATTR_MEAL_PLAN, + name="Grocy meal plan", + icon="mdi:silverware-variant", + exists_fn=lambda entities: ATTR_MEAL_PLAN in entities, + ), + GrocyTodoListEntityDescription( + key=ATTR_SHOPPING_LIST, + name="Grocy shopping list", + icon="mdi:cart-outline", + exists_fn=lambda entities: ATTR_SHOPPING_LIST in entities, + ), + GrocyTodoListEntityDescription( + key=ATTR_STOCK, + name="Grocy stock", + icon="mdi:fridge-outline", + exists_fn=lambda entities: ATTR_STOCK in entities, + ), + GrocyTodoListEntityDescription( + key=ATTR_TASKS, + name="Grocy tasks", + icon="mdi:checkbox-marked-circle-outline", + exists_fn=lambda entities: ATTR_TASKS in entities, + ), +) + + +def _calculate_days_until( + due: datetime.datetime | datetime.date | None, date_only: bool = False +) -> int: + return ( + ( + (due.date() if isinstance(due, datetime.datetime) else due) + - datetime.date.today() + if date_only + else due - datetime.datetime.now() + ).days + if due + else 0 + ) + + +def _calculate_item_status(daysUntilDue: int): + return TodoItemStatus.NEEDS_ACTION if daysUntilDue < 1 else TodoItemStatus.COMPLETED + + +class GrocyTodoItem(TodoItem): + def __init__( + self, + item: Chore + | Battery + | MealPlanItem + | MealPlanItemWrapper + | Product + | ShoppingListProduct + | Task + | None = None, + key: str = "", + ): + if isinstance(item, Chore): + due = item.next_estimated_execution_time + days_until = _calculate_days_until(due, item.track_date_only) + super().__init__( + uid=item.id.__str__(), + summary=item.name, + due=due, + status=_calculate_item_status(days_until), + description=item.description or None, + ) + elif isinstance(item, Battery): + due = item.next_estimated_charge_time + days_until = _calculate_days_until(due, True) + super().__init__( + uid=item.id.__str__(), + summary=item.name, + due=due, + status=_calculate_item_status(days_until), + description=item.description or None, + ) + elif isinstance(item, MealPlanItem): + due = item.day + days_until = _calculate_days_until(due, True) + super().__init__( + uid=item.id.__str__(), + summary=item.recipe.name, + due=due, + status=_calculate_item_status(days_until), + description=item.recipe.description or None, + ) + elif isinstance(item, MealPlanItemWrapper): + due = item.meal_plan.day + days_until = _calculate_days_until(due, True) + super().__init__( + uid=item.meal_plan.id.__str__(), + summary=item.meal_plan.recipe.name, + due=due, + status=_calculate_item_status(days_until), + description=item.meal_plan.recipe.description or None, + ) + elif isinstance(item, Product): + super().__init__( + uid=item.id.__str__(), + summary=f"{item.available_amount:.2f}x {item.name}", + status=TodoItemStatus.NEEDS_ACTION + if (item.available_amount or 0) > 0 + else TodoItemStatus.COMPLETED, + # TODO, the description attribute isn't pulled for products in pygrocy + description=None, + ) + elif isinstance(item, ShoppingListProduct): + super().__init__( + uid=item.id.__str__(), + summary=f"{item.amount:.2f}x {item.product.name}", + due=None, + status=TodoItemStatus.NEEDS_ACTION + # TODO, needs the 'done' attribute instead; however, this isn't supported by pygrocy yet. + if item.amount > 0 + else TodoItemStatus.COMPLETED, + description=item.note or None, + ) + elif isinstance(item, Task): + due = item.due_date + days_until = _calculate_days_until(due, True) + super().__init__( + uid=item.id.__str__(), + summary=item.name, + due=due, + status=_calculate_item_status(days_until), + description=item.description or None, + ) + else: + raise NotImplementedError(f"{key} => {type(item)}") + + +class GrocyTodoListEntity(GrocyEntity, TodoListEntity): + """Grocy todo entity definition.""" + + def __init__( + self, + coordinator: GrocyDataUpdateCoordinator, + description: EntityDescription, + config_entry: ConfigEntry, + ): + self._attr_supported_features = ( + TodoListEntityFeature.UPDATE_TODO_ITEM + | TodoListEntityFeature.DELETE_TODO_ITEM + ) + if description.key in [ATTR_BATTERIES, ATTR_CHORES, ATTR_TASKS]: + self._attr_supported_features |= TodoListEntityFeature.CREATE_TODO_ITEM + if description.key in []: + self._attr_supported_features |= ( + TodoListEntityFeature.SET_DESCRIPTION_ON_ITEM + ) + if description.key in []: + self._attr_supported_features |= TodoListEntityFeature.SET_DUE_DATE_ON_ITEM + if description.key in []: + self._attr_supported_features |= ( + TodoListEntityFeature.SET_DUE_DATETIME_ON_ITEM + ) + super().__init__(coordinator, description, config_entry) + + def _get_grocy_item(self, item_id: str): + entity_data = self.coordinator.data[self.entity_description.key] + return [ + item + for item in entity_data + if (item.id if hasattr(item, "id") else item.meal_plan.id).__str__() + == item_id + ][0] or None + + @property + def todo_items(self) -> list[TodoItem] | None: + """Return the value reported by the todo.""" + entity_data = self.coordinator.data[self.entity_description.key] + return ( + [GrocyTodoItem(item, self.entity_description.key) for item in entity_data] + if entity_data + else [] + ) + + async def async_create_todo_item(self, item: TodoItem) -> None: + """Add an item to the To-do list.""" + if self.entity_description.key == ATTR_BATTERIES: + # TODO pygrocy needs support for empty description, empty used_in + await async_add_generic_service( + self.hass, + self.coordinator, + { + SERVICE_ENTITY_TYPE: "batteries", + SERVICE_DATA: { + "name": item.summary, + "description": item.description or "generic", + "used_in": "generic", + "charge_interval_days": "0", + }, + }, + ) + elif self.entity_description.key == ATTR_CHORES: + await async_add_generic_service( + self.hass, + self.coordinator, + { + SERVICE_ENTITY_TYPE: "chores", + SERVICE_DATA: { + "name": item.summary, + "description": item.description or "", + # "due_date": item.due, + "period_type": "manually", + "period_days": 0, + }, + }, + ) + elif self.entity_description.key == ATTR_TASKS: + # In Validation + await async_add_generic_service( + self.hass, + self.coordinator, + { + SERVICE_ENTITY_TYPE: "tasks", + SERVICE_DATA: { + "name": item.summary, + "description": item.description, + "due_date": (item.due or datetime.date.today()).isoformat(), + }, + }, + ) + else: + raise NotImplementedError(self.entity_description.key) + # Meal Plan, Stock, Shopping List are not intuitive to add. + # (Requires nested IDs, which need to be provided by the user) + await self.coordinator.async_refresh() + + async def async_update_todo_item(self, item: GrocyTodoItem) -> None: + """Update an item in the To-do list.""" + # My template Update handler + if self.entity_description.key == ATTR_BATTERIES: + if item.status == TodoItemStatus.COMPLETED: + await async_track_battery_service( + self.hass, self.coordinator, {SERVICE_BATTERY_ID: item.uid} + ) + else: + raise NotImplementedError(self.entity_description.key) + elif self.entity_description.key == ATTR_CHORES: + if item.status == TodoItemStatus.COMPLETED: + data: dict[str, Any] = { + SERVICE_CHORE_ID: item.uid, + SERVICE_DONE_BY: 1, + SERVICE_SKIPPED: False, + } + await async_execute_chore_service(self.hass, self.coordinator, data) + else: + # I Probably need to cache the chore completion, so that I can undo it... + raise NotImplementedError(self.entity_description.key) + elif self.entity_description.key == ATTR_MEAL_PLAN: + if item.status == TodoItemStatus.COMPLETED: + grocy_item = self._get_grocy_item(item.uid) + await async_consume_recipe_service( + self.hass, + self.coordinator, + {SERVICE_RECIPE_ID: grocy_item.meal_plan.recipe.id}, + ) + await async_delete_generic_service( + self.hass, + self.coordinator, + { + SERVICE_ENTITY_TYPE: "meal_plan", + SERVICE_OBJECT_ID: item.uid, + }, + ) + else: + # I Probably need to cache the chore completion, so that I can undo it... + raise NotImplementedError(self.entity_description.key) + elif self.entity_description.key == ATTR_SHOPPING_LIST: + if item.status == TodoItemStatus.COMPLETED: + # TODO pygrocy doesn't track shopping lists, but they are needed here + grocy_item = self._get_grocy_item(item.uid) + await async_remove_product_in_shopping_list( + self.hass, + self.coordinator, + { + SERVICE_SHOPPING_LIST_ID: 1, + SERVICE_PRODUCT_ID: grocy_item.product_id, + SERVICE_AMOUNT: grocy_item.amount, + }, + ) + else: + raise NotImplementedError(self.entity_description.key) + elif self.entity_description.key == ATTR_STOCK: + if item.status == TodoItemStatus.COMPLETED: + grocy_item = self._get_grocy_item(item.uid) + await async_consume_product_service( + self.hass, + self.coordinator, + { + SERVICE_PRODUCT_ID: item.uid, + SERVICE_AMOUNT: grocy_item.available_amount, + }, + ) + else: + raise NotImplementedError(self.entity_description.key) + elif self.entity_description.key == ATTR_TASKS: + # In Validation, process executes; however, throws error about hass being undefined. (NOTE Action is still performed) + if item.status == TodoItemStatus.COMPLETED: + data: dict[str, Any] = { + SERVICE_TASK_ID: item.uid, + } + await async_complete_task_service(self.hass, self.coordinator, data) + else: + raise NotImplementedError(self.entity_description.key) + # My template Update handler + elif self.entity_description.key == "unsupported": + if item.status == TodoItemStatus.COMPLETED: + raise NotImplementedError(self.entity_description.key) + else: + raise NotImplementedError(self.entity_description.key) + else: + raise NotImplementedError(self.entity_description.key) + await self.coordinator.async_refresh() + + async def async_delete_todo_items(self, uids: list[str]) -> None: + """Delete an item in the To-do list.""" + routines = [ + async_delete_generic_service( + self.hass, + self.coordinator, + { + SERVICE_ENTITY_TYPE: self.entity_description.key, + SERVICE_OBJECT_ID: uid, + }, + ) + for uid in uids + ] + for routine in routines: + await routine + await self.coordinator.async_refresh()