spothole/providers/alert/ical_alert_provider.py

from datetime import datetime, time
from typing import cast

import pytz
from icalendar import Calendar, Event

from data.alert import Alert
from providers.alert.http_alert_provider import HTTPAlertProvider


class ICALAlertProvider(HTTPAlertProvider):
    """Generic alert provider for iCal calendars. Defines an abstract method event_to_alert(event) that subclasses must
    implement, and use it to convert an iCal event to an Alert object based on whatever format their iCal events use."""

    def __init__(self, name, provider_config, url, poll_interval):
        super().__init__(name, provider_config, url, poll_interval)

    def _http_response_to_alerts(self, http_response):
        new_alerts = []
        cal = Calendar.from_ical(http_response.content)

        # Iterate through events, passing each one in turn to the subclass' event_to_alert method to turn it into
        # a Spothole alert object
        for component in cal.walk():
            if component.name != "VEVENT":
                continue

            event = cast(Event, component)
            alert = self.event_to_alert(event)
            new_alerts.append(alert)
        return new_alerts

    def event_to_alert(self, event: Event) -> Alert:
        """Convert an ICal event to an Alert object. Subclasses must implement this method."""

    @staticmethod
    def _to_utc_timestamp(value):
        """Convert a date or datetime value from an iCal field into a UTC UNIX timestamp."""

        # Datetime object so we can treat it as-is, check if it has a non-UTC tz and convert it if necessary
        if isinstance(value, datetime):
            if value.tzinfo is None:
                value = pytz.UTC.localize(value)
            return value.astimezone(pytz.UTC).timestamp()

        # Date object so this is an all day event
        return pytz.UTC.localize(datetime.combine(value, time.min)).timestamp()