X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=gcal_renderer.py;h=c6daf3f746c83d1aa892aa15151c13b07476ebab;hb=09215cf1a1498c99ee75a7cbef3ea62f58a56f0d;hp=e6657795ed91d1c057364b7f8f1ada105bf2bb1f;hpb=5e241dc47e497c547463cecc07946ea6882835a7;p=kiosk.git diff --git a/gcal_renderer.py b/gcal_renderer.py index e665779..c6daf3f 100644 --- a/gcal_renderer.py +++ b/gcal_renderer.py @@ -1,15 +1,28 @@ -from oauth2client.client import AccessTokenRefreshError -import constants +#!/usr/bin/env python3 + +"""Renders an upcoming events page and countdowns page based on the +contents of several Google calendars.""" + import datetime +import functools +import logging +import time +from typing import Any, Dict, List, Optional, Tuple + +from dateutil.parser import parse +import gdata_oauth +import pytz + +import kiosk_constants import file_writer -import gdata import globals -import os import renderer -import time -class gcal_renderer(renderer.debuggable_abstaining_renderer): +logger = logging.getLogger(__name__) + + +class gcal_renderer(renderer.abstaining_renderer): """A renderer to fetch upcoming events from www.google.com/calendar""" calendar_whitelist = frozenset( @@ -25,18 +38,29 @@ class gcal_renderer(renderer.debuggable_abstaining_renderer): ] ) + @functools.total_ordering class comparable_event(object): """A helper class to sort events.""" - def __init__(self, start_time, end_time, summary, calendar): + def __init__( + self, + start_time: Optional[datetime.datetime], + end_time: Optional[datetime.datetime], + summary: str, + calendar: str, + ) -> None: if start_time is None: assert end_time is None + else: + assert isinstance(start_time, datetime.datetime) + if end_time is not None: + assert isinstance(end_time, datetime.datetime) self.start_time = start_time self.end_time = end_time self.summary = summary self.calendar = calendar - def __lt__(self, that): + def __lt__(self, that) -> bool: if self.start_time is None and that.start_time is None: return self.summary < that.summary if self.start_time is None or that.start_time is None: @@ -48,185 +72,238 @@ class gcal_renderer(renderer.debuggable_abstaining_renderer): that.calendar, ) - def __str__(self): + def __eq__(self, that) -> bool: + return ( + self.start_time == that.start_time + and self.end_time == that.end_time + and self.summary == that.summary + and self.calendar == that.calendar + ) + + def __repr__(self) -> str: return "[%s] %s" % (self.timestamp(), self.friendly_name()) - def friendly_name(self): + def friendly_name(self) -> str: name = self.summary name = name.replace("countdown:", "") return "%s" % name - def timestamp(self): + def timestamp(self) -> str: + now = datetime.datetime.now(pytz.timezone("US/Pacific")) if self.start_time is None: return "None" - elif self.start_time.hour == 0: - return datetime.datetime.strftime(self.start_time, "%a %b %d %Y") + elif ( + self.start_time.hour == 0 + and self.start_time.minute == 0 + and self.start_time.second == 0 + ): + if self.start_time.year == now.year: + return datetime.datetime.strftime(self.start_time, "%a %b %d") + else: + return datetime.datetime.strftime(self.start_time, "%a %b %d, %Y") else: - return datetime.datetime.strftime( - self.start_time, "%a %b %d %Y %H:%M%p" + dt = self.start_time + zone = dt.tzinfo + local_dt = dt.replace(tzinfo=zone).astimezone( + tz=pytz.timezone("US/Pacific") ) + if local_dt.year == now.year: + return datetime.datetime.strftime(local_dt, "%a %b %d %I:%M%p") + else: + return datetime.datetime.strftime(local_dt, "%a %b %d, %Y %I:%M%p") - def __init__(self, name_to_timeout_dict, oauth): - super(gcal_renderer, self).__init__(name_to_timeout_dict, True) + def __init__( + self, name_to_timeout_dict: Dict[str, int], oauth: gdata_oauth.OAuth + ) -> None: + super().__init__(name_to_timeout_dict) self.oauth = oauth self.client = self.oauth.calendar_service() - self.sortable_events = [] - self.countdown_events = [] + self.sortable_events: List[gcal_renderer.comparable_event] = [] + self.countdown_events: List[gcal_renderer.comparable_event] = [] - def debug_prefix(self): + def debug_prefix(self) -> str: return "gcal" - def periodic_render(self, key): - self.debug_print('called for "%s"' % key) + def periodic_render(self, key: str) -> bool: + logger.debug(f'called for "{key}"') if key == "Render Upcoming Events": return self.render_upcoming_events() elif key == "Look For Triggered Events": return self.look_for_triggered_events() else: - raise error("Unexpected operation") - - def render_upcoming_events(self): - page_token = None - - def format_datetime(x): - return datetime.datetime.strftime(x, "%Y-%m-%dT%H:%M:%SZ") + raise Exception("Unexpected operation") - now = datetime.datetime.now() - time_min = now - datetime.timedelta(1) - time_max = now + datetime.timedelta(95) - time_min, time_max = list(map(format_datetime, (time_min, time_max))) - self.debug_print("time_min is %s" % time_min) - self.debug_print("time_max is %s" % time_max) + def get_min_max_timewindow(self) -> Tuple[str, str]: + now = datetime.datetime.now(pytz.timezone("US/Pacific")) + _time_min = now - datetime.timedelta(hours=6) + _time_max = now + datetime.timedelta(days=95) + time_min = datetime.datetime.strftime(_time_min, "%Y-%m-%dT%H:%M:%SZ") + time_max = datetime.datetime.strftime(_time_max, "%Y-%m-%dT%H:%M:%SZ") + logger.debug(f"time_min is {time_min}") + logger.debug(f"time_max is {time_max}") + return (time_min, time_max) - # Writes 2 files: - # + "upcoming events", - # + a countdown timer for a subser of events, - f = file_writer.file_writer("gcal_3_86400.html") - f.write("

Upcoming Calendar Events:

\n") - f.write("
\n") - - g = file_writer.file_writer("countdown_3_7200.html") - g.write("

Countdowns:

- - -\n""" - % (event.timestamp(), event.friendly_name()) +

Upcoming Calendar Events:

+
+
+
- %s - - %s -
+""" ) - f.write("
\n") - f.close() + upcoming_sortable_events = self.sortable_events[:12] + for n, event in enumerate(upcoming_sortable_events): + logger.debug(f"{n}/12: {event.friendly_name()} / {event.calendar}") + if n % 2 == 0: + color = "#c6b0b0" + else: + color = "#eeeeee" + f.write( + f""" + + + {event.timestamp()} + + + {event.friendly_name()} + + +""" + ) + f.write("\n") + # Populate the "Countdown" page. + logger.debug("Rendering the Countdowns page") self.countdown_events.sort() - upcoming_countdown_events = self.countdown_events[:12] - now = datetime.datetime.now() - count = 0 - timestamps = {} - for event in upcoming_countdown_events: - eventstamp = event.start_time - delta = eventstamp - now - name = event.friendly_name() - x = int(delta.total_seconds()) - if x > 0: - identifier = "id%d" % count - days = divmod(x, constants.seconds_per_day) - hours = divmod(days[1], constants.seconds_per_hour) - minutes = divmod(hours[1], constants.seconds_per_minute) - g.write( - '
  • %d days, %02d:%02d until %s
    • \n' - % (identifier, days[0], hours[0], minutes[0], name) - ) - timestamps[identifier] = time.mktime(eventstamp.timetuple()) - count += 1 - self.debug_print( - "countdown to %s is %dd %dh %dm" - % (name, days[0], hours[0], minutes[0]) - ) - g.write("") - g.write("""" - ) - g.close() + ) return True - except (gdata.service.RequestError, AccessTokenRefreshError): - print("********* TRYING TO REFRESH GCAL CLIENT *********") - self.oauth.refresh_token() - self.client = self.oauth.calendar_service() + except Exception as e: + logger.exception(e) return False - except: - raise - def look_for_triggered_events(self): - f = file_writer.file_writer(constants.gcal_imminent_pagename) - f.write("

    Imminent Upcoming Calendar Events:

    \n
    \n") - f.write("
    \n") - now = datetime.datetime.now() - count = 0 - for event in self.sortable_events: - eventstamp = event.start_time - delta = eventstamp - now - x = int(delta.total_seconds()) - if x > 0 and x <= constants.seconds_per_minute * 3: - days = divmod(x, constants.seconds_per_day) - hours = divmod(days[1], constants.seconds_per_hour) - minutes = divmod(hours[1], constants.seconds_per_minute) + def look_for_triggered_events(self) -> bool: + logger.debug("Looking for Imminent Upcoming Calendar events...") + with file_writer.file_writer(kiosk_constants.gcal_imminent_pagename) as f: + f.write("

    Imminent Upcoming Calendar Events:

    \n
    \n") + f.write("
    \n") + now = datetime.datetime.now(pytz.timezone("US/Pacific")) + count = 0 + for event in self.sortable_events: eventstamp = event.start_time - name = event.friendly_name() - calendar = event.calendar - f.write( - "
  • %s (%s) upcoming in %d minutes.\n" - % (name, calendar, minutes[0]) - ) - count += 1 - f.write("
    • ") - f.close() + if eventstamp is None: + continue + delta = eventstamp - now + x = int(delta.total_seconds()) + if x > 0 and x < 4 * kiosk_constants.seconds_per_minute: + days = divmod(x, kiosk_constants.seconds_per_day) + hours = divmod(days[1], kiosk_constants.seconds_per_hour) + minutes = divmod(hours[1], kiosk_constants.seconds_per_minute) + eventstamp = event.start_time + name = event.friendly_name() + calendar = event.calendar + logger.debug(f"Event {event} ({name}) triggered the page.") + f.write( + f"
  • {name} ({calendar}) upcoming in {int(minutes[0])} minutes.\n" + ) + count += 1 + f.write("") if count > 0: globals.put("gcal_triggered", True) else: globals.put("gcal_triggered", False) return True + + +# Test +# import kiosk_secrets as secrets +# +# oauth = gdata_oauth.OAuth(secrets.google_client_secret) +# x = gcal_renderer( +# {"Render Upcoming Events": 10000, "Look For Triggered Events": 1}, oauth +# ) +# x.periodic_render("Render Upcoming Events")