trigger.py

Go to the documentation of this file.

"""Offer calendar automation rules."""
 
from __future__ import annotations
 
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
import datetime
import logging
from typing import Any
 
import voluptuous as vol
 
from homeassistant.const import CONF_ENTITY_ID, CONF_EVENT, CONF_OFFSET, CONF_PLATFORM
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.event import (
    async_track_point_in_time,
    async_track_time_interval,
)
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util
 
from . import CalendarEntity, CalendarEvent
from .const import DATA_COMPONENT, DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
EVENT_START = "start"
EVENT_END = "end"
UPDATE_INTERVAL = datetime.timedelta(minutes=15)
 
TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
    {
        vol.Required(CONF_PLATFORM): DOMAIN,
        vol.Required(CONF_ENTITY_ID): cv.entity_id,
        vol.Optional(CONF_EVENT, default=EVENT_START): vol.In({EVENT_START, EVENT_END}),
        vol.Optional(CONF_OFFSET, default=datetime.timedelta(0)): cv.time_period,
    }
)
 
# mypy: disallow-any-generics
 
 
@dataclass
class QueuedCalendarEvent:
    """An event that is queued to be fired in the future."""
 
    trigger_time: datetime.datetime
    event: CalendarEvent
 
 
@dataclass
class Timespan:
    """A time range part of start/end dates, used for considering active events."""
 
    start: datetime.datetime
    """The start datetime of the interval."""
 
    end: datetime.datetime
    """The end datetime (exclusive) of the interval."""
 
    def with_offset(self, offset: datetime.timedelta) -> Timespan:
        """Return a new interval shifted by the specified offset."""
        return Timespan(self.start + offset, self.end + offset)
 
    def __contains__(self, trigger: datetime.datetime) -> bool:
        """Return true if the trigger time is within the time span."""
        return self.start <= trigger < self.end
 
    def next_upcoming(
        self, now: datetime.datetime, interval: datetime.timedelta
    ) -> Timespan:
        """Return a subsequent time span following the current time span.
 
        This effectively gives us a cursor like interface for advancing through
        time using the interval as a hint. The returned span may have a
        different interval than the one specified. For example, time span may
        be longer during a daylight saving time transition, or may extend due to
        drift if the current interval is old. The returned time span is
        adjacent and non-overlapping.
        """
        return Timespan(self.end, max(self.end, now) + interval)
 
    def __str__(self) -> str:
        """Return a string representing the half open interval time span."""
        return f"[{self.start}, {self.end})"
 
 
type EventFetcher = Callable[[Timespan], Awaitable[list[CalendarEvent]]]
type QueuedEventFetcher = Callable[[Timespan], Awaitable[list[QueuedCalendarEvent]]]
 
 
def get_entity(hass: HomeAssistant, entity_id: str) -> CalendarEntity:
    """Get the calendar entity for the provided entity_id."""
    component: EntityComponent[CalendarEntity] = hass.data[DATA_COMPONENT]
    if not (entity := component.get_entity(entity_id)) or not isinstance(
        entity, CalendarEntity
    ):
        raise HomeAssistantError(
            f"Entity does not exist {entity_id} or is not a calendar entity"
        )
    return entity
 
 
def event_fetcher(hass: HomeAssistant, entity_id: str) -> EventFetcher:
    """Build an async_get_events wrapper to fetch events during a time span."""
 
    async def async_get_events(timespan: Timespan) -> list[CalendarEvent]:
        """Return events active in the specified time span."""
        entity = get_entity(hass, entity_id)
        # Expand by one second to make the end time exclusive
        end_time = timespan.end + datetime.timedelta(seconds=1)
        return await entity.async_get_events(hass, timespan.start, end_time)
 
    return async_get_events
 
 
def queued_event_fetcher(
    fetcher: EventFetcher, event_type: str, offset: datetime.timedelta
) -> QueuedEventFetcher:
    """Build a fetcher that produces a schedule of upcoming trigger events."""
 
    def get_trigger_time(event: CalendarEvent) -> datetime.datetime:
        if event_type == EVENT_START:
            return event.start_datetime_local
        return event.end_datetime_local
 
    async def async_get_events(timespan: Timespan) -> list[QueuedCalendarEvent]:
        """Get calendar event triggers eligible to fire in the time span."""
        offset_timespan = timespan.with_offset(-1 * offset)
        active_events = await fetcher(offset_timespan)
 
        # Determine the trigger eligibility of events during this time span.
        # Example: For an EVENT_END trigger the event may start during this
        # time span, but need to be triggered later when the end happens.
        results = []
        for trigger_time, event in zip(
            map(get_trigger_time, active_events), active_events, strict=False
        ):
            if trigger_time not in offset_timespan:
                continue
            results.append(QueuedCalendarEvent(trigger_time + offset, event))
 
        _LOGGER.debug(
            "Scan events @ %s%s found %s eligible of %s active",
            offset_timespan,
            f" (offset={offset})" if offset else "",
            len(results),
            len(active_events),
        )
        results.sort(key=lambda x: x.trigger_time)
        return results
 
    return async_get_events
 
 
class CalendarEventListener:
    """Helper class to listen to calendar events.
 
    This listener will poll every UPDATE_INTERVAL to fetch a set of upcoming
    calendar events in the upcoming window of time, putting them into a queue.
    The queue is drained by scheduling an alarm for the next upcoming event
    trigger time, one event at a time.
    """
 
    def __init__(
        self,
        hass: HomeAssistant,
        job: HassJob[..., Coroutine[Any, Any, None]],
        trigger_data: dict[str, Any],
        fetcher: QueuedEventFetcher,
    ) -> None:
        """Initialize CalendarEventListener."""
        self._hass_hass = hass
        self._job_job = job
        self._trigger_data_trigger_data = trigger_data
        self._unsub_event_unsub_event: CALLBACK_TYPE | None = None
        self._unsub_refresh_unsub_refresh: CALLBACK_TYPE | None = None
        self._fetcher_fetcher = fetcher
        now = dt_util.now()
        self._timespan_timespan = Timespan(now, now + UPDATE_INTERVAL)
        self._events: list[QueuedCalendarEvent] = []
 
    async def async_attach(self) -> None:
        """Attach a calendar event listener."""
        self._events.extend(await self._fetcher_fetcher(self._timespan_timespan))
        self._unsub_refresh_unsub_refresh = async_track_time_interval(
            self._hass_hass, self._handle_refresh_handle_refresh, UPDATE_INTERVAL
        )
        self._listen_next_calendar_event_listen_next_calendar_event()
 
    @callback
    def async_detach(self) -> None:
        """Detach the calendar event listener."""
        self._clear_event_listener_clear_event_listener()
        if self._unsub_refresh_unsub_refresh:
            self._unsub_refresh_unsub_refresh()
        self._unsub_refresh_unsub_refresh = None
 
    @callback
    def _listen_next_calendar_event(self) -> None:
        """Set up the calendar event listener."""
        if not self._events:
            return
 
        _LOGGER.debug(
            "Scheduled next event trigger for %s", self._events[0].trigger_time
        )
        self._unsub_event_unsub_event = async_track_point_in_time(
            self._hass_hass,
            self._handle_calendar_event_handle_calendar_event,
            self._events[0].trigger_time,
        )
 
    def _clear_event_listener(self) -> None:
        """Reset the event listener."""
        if self._unsub_event_unsub_event:
            self._unsub_event_unsub_event()
        self._unsub_event_unsub_event = None
 
    async def _handle_calendar_event(self, now: datetime.datetime) -> None:
        """Handle calendar event."""
        _LOGGER.debug("Calendar event @ %s", dt_util.as_local(now))
        self._dispatch_events_dispatch_events(now)
        self._clear_event_listener_clear_event_listener()
        self._listen_next_calendar_event_listen_next_calendar_event()
 
    def _dispatch_events(self, now: datetime.datetime) -> None:
        """Dispatch all events that are eligible to fire."""
        while self._events and self._events[0].trigger_time <= now:
            queued_event = self._events.pop(0)
            _LOGGER.debug("Dispatching event: %s", queued_event.event)
            self._hass_hass.async_run_hass_job(
                self._job_job,
                {
                    "trigger": {
                        **self._trigger_data_trigger_data,
                        "calendar_event": queued_event.event.as_dict(),
                    }
                },
            )
 
    async def _handle_refresh(self, now_utc: datetime.datetime) -> None:
        """Handle core config update."""
        now = dt_util.as_local(now_utc)
        _LOGGER.debug("Refresh events @ %s", now)
        # Dispatch any eligible events in the boundary case where refresh
        # fires before the calendar event.
        self._dispatch_events_dispatch_events(now)
        self._clear_event_listener_clear_event_listener()
        self._timespan_timespan = self._timespan_timespan.next_upcoming(now, UPDATE_INTERVAL)
        try:
            self._events.extend(await self._fetcher_fetcher(self._timespan_timespan))
        except HomeAssistantError as ex:
            _LOGGER.error("Calendar trigger failed to fetch events: %s", ex)
        self._listen_next_calendar_event_listen_next_calendar_event()
 
 
async def async_attach_trigger(
    hass: HomeAssistant,
    config: ConfigType,
    action: TriggerActionType,
    trigger_info: TriggerInfo,
) -> CALLBACK_TYPE:
    """Attach trigger for the specified calendar."""
    entity_id = config[CONF_ENTITY_ID]
    event_type = config[CONF_EVENT]
    offset = config[CONF_OFFSET]
 
    # Validate the entity id is valid
    get_entity(hass, entity_id)
 
    trigger_data = {
        **trigger_info["trigger_data"],
        "platform": DOMAIN,
        "event": event_type,
        "offset": offset,
    }
    listener = CalendarEventListener(
        hass,
        HassJob(action),
        trigger_data,
        queued_event_fetcher(event_fetcher(hass, entity_id), event_type, offset),
    )
    await listener.async_attach()
    return listener.async_detach