feat: add Waste Collection Schedule via HACS

custom_components/waste_collection_schedule/__init__.py

@@ -0,0 +1,237 @@
+"""Waste Collection Schedule Component."""
+import logging
+import site
+from pathlib import Path
+from random import randrange
+
+import homeassistant.helpers.config_validation as cv
+import homeassistant.util.dt as dt_util
+import voluptuous as vol
+from homeassistant.core import HomeAssistant, ServiceCall, callback
+from homeassistant.helpers.dispatcher import dispatcher_send
+
+from .const import DOMAIN, UPDATE_SENSORS_SIGNAL
+
+from homeassistant.helpers.event import async_call_later  # isort:skip
+from homeassistant.helpers.event import async_track_time_change  # isort:skip
+
+# add module directory to path
+package_dir = Path(__file__).resolve().parents[0]
+site.addsitedir(str(package_dir))
+from waste_collection_schedule import Customize, SourceShell  # type: ignore # isort:skip # noqa: E402
+
+_LOGGER = logging.getLogger(__name__)
+
+CONF_SOURCES = "sources"
+CONF_SOURCE_NAME = "name"
+CONF_SOURCE_ARGS = "args"  # source arguments
+CONF_SOURCE_CALENDAR_TITLE = "calendar_title"
+CONF_SEPARATOR = "separator"
+CONF_FETCH_TIME = "fetch_time"
+CONF_RANDOM_FETCH_TIME_OFFSET = "random_fetch_time_offset"
+CONF_DAY_SWITCH_TIME = "day_switch_time"
+
+CONF_CUSTOMIZE = "customize"
+CONF_TYPE = "type"
+CONF_ALIAS = "alias"
+CONF_SHOW = "show"
+CONF_ICON = "icon"
+CONF_PICTURE = "picture"
+CONF_USE_DEDICATED_CALENDAR = "use_dedicated_calendar"
+CONF_DEDICATED_CALENDAR_TITLE = "dedicated_calendar_title"
+
+CUSTOMIZE_CONFIG = vol.Schema(
+    {
+        vol.Optional(CONF_TYPE): cv.string,
+        vol.Optional(CONF_ALIAS): cv.string,
+        vol.Optional(CONF_SHOW): cv.boolean,
+        vol.Optional(CONF_ICON): cv.icon,
+        vol.Optional(CONF_PICTURE): cv.string,
+        vol.Optional(CONF_USE_DEDICATED_CALENDAR): cv.boolean,
+        vol.Optional(CONF_DEDICATED_CALENDAR_TITLE): cv.string,
+    }
+)
+
+SOURCE_CONFIG = vol.Schema(
+    {
+        vol.Required(CONF_SOURCE_NAME): cv.string,
+        vol.Required(CONF_SOURCE_ARGS): dict,
+        vol.Optional(CONF_CUSTOMIZE, default=[]): vol.All(
+            cv.ensure_list, [CUSTOMIZE_CONFIG]
+        ),
+        vol.Optional(CONF_SOURCE_CALENDAR_TITLE): cv.string,
+    }
+)
+
+CONFIG_SCHEMA = vol.Schema(
+    {
+        DOMAIN: vol.Schema(
+            {
+                vol.Required(CONF_SOURCES): vol.All(cv.ensure_list, [SOURCE_CONFIG]),
+                vol.Optional(CONF_SEPARATOR, default=", "): cv.string,
+                vol.Optional(CONF_FETCH_TIME, default="01:00"): cv.time,
+                vol.Optional(
+                    CONF_RANDOM_FETCH_TIME_OFFSET, default=60
+                ): cv.positive_int,
+                vol.Optional(CONF_DAY_SWITCH_TIME, default="10:00"): cv.time,
+            }
+        )
+    },
+    extra=vol.ALLOW_EXTRA,
+)
+
+
+async def async_setup(hass: HomeAssistant, config: dict):
+    """Set up the component. config contains data from configuration.yaml."""
+    # create empty api object as singleton
+    api = WasteCollectionApi(
+        hass,
+        separator=config[DOMAIN][CONF_SEPARATOR],
+        fetch_time=config[DOMAIN][CONF_FETCH_TIME],
+        random_fetch_time_offset=config[DOMAIN][CONF_RANDOM_FETCH_TIME_OFFSET],
+        day_switch_time=config[DOMAIN][CONF_DAY_SWITCH_TIME],
+    )
+
+    # create shells for source(s)
+    for source in config[DOMAIN][CONF_SOURCES]:
+        # create customize object
+        customize = {}
+        for c in source.get(CONF_CUSTOMIZE, {}):
+            customize[c[CONF_TYPE]] = Customize(
+                waste_type=c[CONF_TYPE],
+                alias=c.get(CONF_ALIAS),
+                show=c.get(CONF_SHOW, True),
+                icon=c.get(CONF_ICON),
+                picture=c.get(CONF_PICTURE),
+                use_dedicated_calendar=c.get(CONF_USE_DEDICATED_CALENDAR, False),
+                dedicated_calendar_title=c.get(CONF_DEDICATED_CALENDAR_TITLE, False),
+            )
+        api.add_source_shell(
+            source_name=source[CONF_SOURCE_NAME],
+            customize=customize,
+            calendar_title=source.get(CONF_SOURCE_CALENDAR_TITLE),
+            source_args=source.get(CONF_SOURCE_ARGS, {}),
+        )
+
+    # store api object
+    hass.data.setdefault(DOMAIN, api)
+
+    # load calendar platform
+    await hass.helpers.discovery.async_load_platform(
+        "calendar", DOMAIN, {"api": api}, config
+    )
+
+    # initial fetch of all data
+    hass.add_job(api._fetch)
+
+    async def async_fetch_data(service: ServiceCall) -> None:
+        hass.add_job(api._fetch)
+
+    # Register new Service fetch_data
+    hass.services.async_register(
+        DOMAIN, "fetch_data", async_fetch_data, schema=vol.Schema({})
+    )
+
+    return True
+
+
+class WasteCollectionApi:
+    def __init__(
+        self, hass, separator, fetch_time, random_fetch_time_offset, day_switch_time
+    ):
+        self._hass = hass
+        self._source_shells = []
+        self._separator = separator
+        self._fetch_time = fetch_time
+        self._random_fetch_time_offset = random_fetch_time_offset
+        self._day_switch_time = day_switch_time
+
+        # start timer to fetch date once per day
+        async_track_time_change(
+            hass,
+            self._fetch_callback,
+            self._fetch_time.hour,
+            self._fetch_time.minute,
+            self._fetch_time.second,
+        )
+
+        # start timer for day-switch time
+        if self._day_switch_time != self._fetch_time:
+            async_track_time_change(
+                hass,
+                self._update_sensors_callback,
+                self._day_switch_time.hour,
+                self._day_switch_time.minute,
+                self._day_switch_time.second,
+            )
+
+        # add a timer at midnight (if not already there) to update days-to
+        midnight = dt_util.parse_time("00:00")
+        if midnight != self._fetch_time and midnight != self._day_switch_time:
+            async_track_time_change(
+                hass,
+                self._update_sensors_callback,
+                midnight.hour,
+                midnight.minute,
+                midnight.second,
+            )
+
+    @property
+    def separator(self):
+        """Separator string, used to separator waste types."""
+        return self._separator
+
+    @property
+    def fetch_time(self):
+        """When to fetch to data."""
+        return self._fetch_time
+
+    @property
+    def day_switch_time(self):
+        """When to hide entries for today."""
+        return self._day_switch_time
+
+    def add_source_shell(
+        self,
+        source_name,
+        customize,
+        source_args,
+        calendar_title,
+    ):
+        self._source_shells.append(
+            SourceShell.create(
+                source_name=source_name,
+                customize=customize,
+                source_args=source_args,
+                calendar_title=calendar_title,
+            )
+        )
+
+    def _fetch(self, *_):
+        for shell in self._source_shells:
+            shell.fetch()
+
+        self._update_sensors_callback()
+
+    @property
+    def shells(self):
+        return self._source_shells
+
+    def get_shell(self, index):
+        return self._source_shells[index] if index < len(self._source_shells) else None
+
+    @callback
+    def _fetch_callback(self, *_):
+        async_call_later(
+            self._hass,
+            randrange(0, 60 * self._random_fetch_time_offset),
+            self._fetch_now_callback,
+        )
+
+    @callback
+    def _fetch_now_callback(self, *_):
+        self._hass.add_job(self._fetch)
+
+    @callback
+    def _update_sensors_callback(self, *_):
+        dispatcher_send(self._hass, UPDATE_SENSORS_SIGNAL)

custom_components/waste_collection_schedule/calendar.py

@@ -0,0 +1,127 @@
+"""Calendar platform support for Waste Collection Schedule."""
+
+import logging
+from datetime import datetime, timedelta
+
+from homeassistant.components.calendar import CalendarEntity, CalendarEvent
+from homeassistant.core import HomeAssistant
+
+# fmt: off
+from custom_components.waste_collection_schedule.waste_collection_schedule.collection_aggregator import \
+    CollectionAggregator
+from custom_components.waste_collection_schedule.waste_collection_schedule.source_shell import \
+    SourceShell
+
+# fmt: on
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
+    """Set up calendar platform."""
+    # We only want this platform to be set up via discovery.
+    if discovery_info is None:
+        return
+
+    entities = []
+
+    api = discovery_info["api"]
+
+    for shell in api.shells:
+        dedicated_calendar_types = shell.get_dedicated_calendar_types()
+        for type in dedicated_calendar_types:
+            entities.append(
+                WasteCollectionCalendar(
+                    api=api,
+                    aggregator=CollectionAggregator([shell]),
+                    name=shell.get_calendar_title_for_type(type),
+                    include_types={shell.get_collection_type_name(type)},
+                    unique_id=calc_unique_calendar_id(shell, type),
+                )
+            )
+
+        entities.append(
+            WasteCollectionCalendar(
+                api=api,
+                aggregator=CollectionAggregator([shell]),
+                name=shell.calendar_title,
+                exclude_types={
+                    shell.get_collection_type_name(type)
+                    for type in dedicated_calendar_types
+                },
+                unique_id=calc_unique_calendar_id(shell),
+            )
+        )
+
+    async_add_entities(entities)
+
+
+class WasteCollectionCalendar(CalendarEntity):
+    """Calendar entity class."""
+
+    def __init__(
+        self,
+        api,
+        aggregator,
+        name,
+        unique_id: str,
+        include_types=None,
+        exclude_types=None,
+    ):
+        self._api = api
+        self._aggregator = aggregator
+        self._name = name
+        self._include_types = include_types
+        self._exclude_types = exclude_types
+        self._unique_id = unique_id
+        self._attr_unique_id = unique_id
+
+    @property
+    def name(self):
+        """Return entity name."""
+        return self._name
+
+    @property
+    def event(self):
+        """Return next collection event."""
+        collections = self._aggregator.get_upcoming(
+            count=1,
+            include_today=True,
+            include_types=self._include_types,
+            exclude_types=self._exclude_types,
+        )
+
+        if len(collections) == 0:
+            return None
+        else:
+            return self._convert(collections[0])
+
+    async def async_get_events(
+        self, hass: HomeAssistant, start_date: datetime, end_date: datetime
+    ):
+        """Return all events within specified time span."""
+        events = []
+
+        for collection in self._aggregator.get_upcoming(
+            include_today=True,
+            include_types=self._include_types,
+            exclude_types=self._exclude_types,
+        ):
+            event = self._convert(collection)
+
+            if start_date <= event.start_datetime_local <= end_date:
+                events.append(event)
+
+        return events
+
+    def _convert(self, collection) -> CalendarEvent:
+        """Convert an collection into a Home Assistant calendar event."""
+        return CalendarEvent(
+            summary=collection.type,
+            start=collection.date,
+            end=collection.date + timedelta(days=1),
+        )
+
+
+def calc_unique_calendar_id(shell: SourceShell, type: str = None):
+    return shell.unique_id + ("_" + type if type is not None else "") + "_calendar"

custom_components/waste_collection_schedule/const.py

@@ -0,0 +1,6 @@
+"""Constants for the Waste Collection Schedule component."""
+
+# Component domain, used to store component data in hass data.
+DOMAIN = "waste_collection_schedule"
+
+UPDATE_SENSORS_SIGNAL = "wcs_update_sensors_signal"

custom_components/waste_collection_schedule/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "waste_collection_schedule",
+  "name": "waste_collection_schedule",
+  "codeowners": ["@mampfes"],
+  "dependencies": [],
+  "documentation": "https://github.com/mampfes/hacs_waste_collection_schedule#readme",
+  "integration_type": "hub",
+  "iot_class": "cloud_polling",
+  "requirements": ["icalendar", "recurring_ical_events", "icalevents", "bs4"],
+  "version": "1.41.0"
+}