HomeAssistant_Config/custom_components/waste_collection_schedule/__init__.py

"""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, 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, Scraper  # type: ignore # isort:skip # noqa: E402

_LOGGER = logging.getLogger(__name__)

CONF_SOURCES = "sources"
CONF_SOURCE_NAME = "name"
CONF_SOURCE_ARGS = "args"  # scraper-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 scraper(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_scraper(
            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)

    return True


class WasteCollectionApi:
    def __init__(
        self, hass, separator, fetch_time, random_fetch_time_offset, day_switch_time
    ):
        self._hass = hass
        self._scrapers = []
        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_scraper(
        self,
        source_name,
        customize,
        source_args,
        calendar_title,
    ):
        self._scrapers.append(
            Scraper.create(
                source_name=source_name,
                customize=customize,
                source_args=source_args,
                calendar_title=calendar_title,
            )
        )

    def _fetch(self, *_):
        for scraper in self._scrapers:
            scraper.fetch()

        self._update_sensors_callback()

    @property
    def scrapers(self):
        return self._scrapers

    def get_scraper(self, index):
        return self._scrapers[index] if index < len(self._scrapers) 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)