#!/usr/bin/env python3
# ----------------------------------------------------------------------------
#
# GCalert periodically checks all of your Google Calendars and displays a
# desktop notification whenever a reminder is set for an event.
#
# Only reminders set to 'popup' in Google Calendar will spawn a notification.
# This is the intended behavoir.
#
# Home: http://github.com/nejsan/gcalert
# Original project: http://github.com/raas/gcalert
#
# ----------------------------------------------------------------------------
#
# Copyright 2009 Andras Horvath (andras.horvath nospamat gmailcom) This
# program is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
#
# ----------------------------------------------------------------------------
from os import path, makedirs
from signal import signal, SIGINT
from threading import Thread, Lock
from getopt import GetoptError, getopt
from time import sleep, asctime, mktime
from datetime import datetime, timedelta
from sys import exit, argv, stdout, _getframe
from argparse import ArgumentParser, RawDescriptionHelpFormatter
# Dependencies below come from separate packages, the rest (above) are in the
# standard library so those are expected to work :)
try:
# libnotify handler
from notify2 import init, Notification, EXPIRES_NEVER
# Date parser and timezone handler
from dateutil.tz import tzlocal
from dateutil.parser import parse as parse_time
# For Google Calendar API v3
from httplib2 import Http
from oauth2client.file import Storage
from oauth2client.tools import run_flow, argparser
from googleapiclient.discovery import build
from oauth2client.client import OAuth2WebServerFlow
except ImportError as e:
print('Dependency was not found! {0}\n'.format(e))
print('For Debian/Ubuntu, try:')
print('\tsudo apt-get install python-notify python-dateutil python-googleapi notification-daemon')
exit(1)
#-----------------------------------------------------------------------------#
# Global Properties #
#-----------------------------------------------------------------------------#
__program__ = 'gcalert'
__version__ = '3.2'
__api_client_id__ = '447177524849-hh9ogtma7pgbkm39v1br6qa3h3cal9u9.apps.googleusercontent.com'
__api_client_secret__ = 'UECdkOkaoAnyYe5-4DBm31mu'
#-----------------------------------------------------------------------------#
# Console output functions #
#-----------------------------------------------------------------------------#
def message(message, *args, **kwargs):
"""Prints the given message and flushes the buffer; useful when redirected to a file."""
if not settings.quiet_flag or 'force' in kwargs:
print(message.format(*args, **kwargs))
stdout.flush()
def debug(message, *args, **kwargs):
"""Prints the given message if the debug_flag is set (running with -d or --debug)."""
if settings.debug_flag:
message = message.format(*args, **kwargs)
print('{0} in {1}: {2}'.format(
asctime(), _getframe(1).f_code.co_name, message))
stdout.flush()
def get_unix_timestamp(time):
"""Converts a datetime object to a UNIX timestamp int."""
return mktime(time.timetuple())
#-----------------------------------------------------------------------------#
# Calendar Notifications Class #
#-----------------------------------------------------------------------------#
class GCalertNotification(object):
"""Represents an instance of a calendar alarm for an event."""
def __init__(self, title, where, start_string, end_string, minutes):
"""Creates a new alarm for the given event.
Args:
title (str): The title of the event.
where (str): The location of the event, or an empty string.
start_string (str): The start time of the event as a string.
end_string (str): The end time of the event as a string.
minutes (int): How many minutes before the start of the event to set off the alarm.
"""
self.title = title
self.where = where
self.start = parse_time(start_string)
self.minutes = minutes
# Google sometimes does not supply timezones
# (for events that last more than a day and have no time set, apparently)
# python can't compare two dates if only one has TZ info
# this might screw us at, say, if DST changes between when we get the event and its alarm
try:
if not self.start.tzname():
self.start = self.start.replace(tzinfo=tzlocal())
except AttributeError:
self.start = self.start.replace(tzinfo=tzlocal())
self.start_str = self.start.astimezone(tzlocal()).strftime(settings.strftime_string)
self.reminder_time = get_unix_timestamp(self.start - timedelta(minutes=self.minutes)) # Store only the UNIX timestamp
self.start = get_unix_timestamp(self.start) # Store only the UNIX timestamp
def notify(self):
"""Show the alarm box for one event/recurrence"""
message(text.bold+'\n########## ALARM ##########'+text.normal)
message(self.get_formatted())
message(text.bold+'########## ALARM ##########\n'+text.normal)
if self.where:
a = Notification(self.title, 'Starting: {start}\nWhere: {location}'.format(start=self.start_str, location=self.where), settings.icon)
else:
a = Notification(self.title, 'Starting: {start}'.format(start=self.start_str), settings.icon)
# Display the alarm notification the user closes it manually
a.set_timeout(EXPIRES_NEVER)
if not a.show():
message('Failed to send alarm notification!')
def get_formatted(self):
"""Returns a string representation of this object's contents."""
representation = (
('Title:', self.title),
('Location:', self.where),
('Start time:', self.start),
('Reminder set:', '{0} minutes before'.format(self.minutes)),
)
return '\n'.join(map(lambda x: '{0:<15} {1}'.format(x[0], x[1]), representation))
def __str__(self):
"""Returns a string representation of this object."""
return 'GCalertNotification({title}, {location}, {start}, {minutes})'.format(
title = self.title,
location = self.where,
start = self.start,
minutes = self.minutes
)
def __eq__(self, other):
return hash(self) == hash(other)
def __hash__(self):
return hash(str(self))
#-----------------------------------------------------------------------------#
# GCalert Class #
# The main thread will start up and then launch the background alerts thread, #
# and proceed check the calendar every so often #
#-----------------------------------------------------------------------------#
class GCalert(object):
"""Connects to Google Calendar and notifies about events at their reminder time."""
def __init__(self):
super(GCalert, self).__init__()
# Create a global settings instance
settings.initialize_user_settings()
# Set GCalert properties
self.events = [] # All events seen so far that are yet to start
self.events_lock = Lock() # Hold to access events[]
self.notified_events = [] # All events which have had their timers registered
self.calendar_service = None
self.connected = False
self.do_login()
# Set up ^C handler
signal(SIGINT, self.stopthismadness)
# Start up
message('{0} {1} running', __program__, __version__)
debug('Settings: {0}', settings.get_settings())
# Start up the event processing thread
debug('Starting event processing thread')
Thread(target=self.process_events_thread, daemon=True).start()
self.update_events_thread()
#-----------------------------------------------------------------------------#
# Google Calendar Query Functions #
#-----------------------------------------------------------------------------#
def date_range_query(self, start_date=None, end_date=None):
"""
Get a list of events happening between the given dates in all calendars the user has
Each reminder occurrence creates a new event (GCalertNotification object).
Returns:
A tuple in the format (, ).
"""
debug('Querying for new events...')
google_events = [] # Events in all Google Calendars
event_list = [] # Our parsed events list
try:
# Get the id for each calendar
calendars = self.calendar_service.calendarList().list().execute()['items']
for calendar in calendars:
debug('Processing calendar: {0}', calendar['summary'])
query = self.calendar_service.events().list(
calendarId=calendar['id'],timeMin=start_date,timeMax=end_date,singleEvents=True).execute()
google_events += query['items']
debug('Events so far: {0}', len(google_events))
except Exception as error:
debug('Connection lost: {0}.', error)
try:
message('Connection lost ({0} {1}), will reconnect.', error.args[0]['status'], error.args[0]['reason'])
except Exception:
message('Connection lost with unknown error, will reconnect: {0}', error)
message('Please report this as a bug.')
self.connected = False
debug('Done querying for new events')
return []
for event in google_events:
# Not all events have 'where' fields, and that's okay
where = event['location'] if ('location' in event) else ''
# Skip events with not overrides key in their reminders dict
if not 'overrides' in event['reminders']:
continue
# Create a GCalertNotification out of each (event x reminder x occurrence)
for reminder in event['reminders']['overrides']:
debug('Event `{0}` notification method: {1}', event['summary'], reminder['method'])
if reminder['method'] == 'popup': # 'popup' in the web interface
# Event (one for each alarm instance) is done, add it to the list
this_event = GCalertNotification(
event['summary'],
where,
event['start']['dateTime'],
event['end']['dateTime'],
reminder['minutes'])
debug('New notification set: {0}', this_event)
event_list.append(this_event)
self.connected = True
debug('Done querying for new events')
return event_list
#-----------------------------------------------------------------------------#
# Authentication Functions #
#-----------------------------------------------------------------------------#
def do_login(self):
"""
Authenticates to Google Calendar.
Occassionally this fails or the connection dies, so this may need to be called again.
Return:
True if authentication succeeded, or False otherwise.
"""
try:
storage = Storage(settings.secrets_file)
credentials = storage.get()
if credentials is None or credentials.invalid:
flow = OAuth2WebServerFlow(
client_id = __api_client_id__,
client_secret = __api_client_secret__,
user_agent = __program__+'/'+__version__,
redirect_uri = 'urn:ietf:wg:oauth:2.0:oob:auto',
scope = 'https://www.googleapis.com/auth/calendar')
parser = ArgumentParser(
formatter_class = RawDescriptionHelpFormatter,
parents = [argparser])
# Parse the command-line flags
flags = parser.parse_args([])
credentials = run_flow(flow, storage, flags)
auth_http = credentials.authorize(Http())
self.calendar_service = build(serviceName='calendar', version='v3', http=auth_http)
except Exception as error:
debug('Failed to authenticate to Google: {0}', error)
message('Failed to authenticate to Google.')
self.connected = False # Login failed
return
message('Logged in to Google Calendar')
self.connected = True # We're logged in
#-----------------------------------------------------------------------------#
# Event Thread Handlers #
#-----------------------------------------------------------------------------#
def process_events_thread(self):
"""Process events and raise alarms via pynotify."""
# Initialize notification system
if not init(__program__+'/'+__version__):
message('Could not initialize pynotify/libnotify!')
exit(1)
sleep(settings.threads_offset) # Give a chance for the other thread to get some events
try:
while True:
now = get_unix_timestamp(datetime.now(tzlocal())) # Get the current UNIX timestamp
debug('Processing events...')
self.events_lock.acquire()
for event in self.events:
event_hash = hash(event)
if event.start < now:
debug('Removing event `{0}`', event)
self.events.remove(event)
# Also free up some memory
if event_hash in self.notified_events:
self.notified_events.remove(event_hash)
# If it starts in the future, check for alarm times if it wasn't alarmed yet
elif event_hash not in self.notified_events:
# Check the notification time. If it's now-ish, raise the notification,
# otherwise let the event sleep some more
# Notify now if the notification time has passed
if now >= event.reminder_time:
event.notify()
self.notified_events.append(event_hash)
else:
debug('Not ready to notify about event `{0}`', event)
else:
debug('Already notified for event `{0}`', event)
self.events_lock.release()
debug('Finished processing events')
# We can't just sleep until the next event as the other thread MIGHT add something new
sleep(settings.alarm_sleeptime)
except KeyboardInterrupt:
# Break if this thread was interrupted
pass
def update_events_thread(self):
"""Periodically syncs the 'events' list to what's in Google Calendar."""
while True:
debug('Updating events...')
# Today
range_start = datetime.now(tzlocal())
# A few days later
range_end = range_start + timedelta(days=settings.lookahead_days)
# Wait until we've obtained a connection and a list of events
new_events = self.date_range_query(range_start.isoformat(), range_end.isoformat())
while not self.connected:
sleep(settings.reconnect_sleeptime)
self.do_login()
new_events = self.date_range_query(range_start.isoformat(), range_end.isoformat())
self.events_lock.acquire()
# Remove events which were deleted or modified
for event in self.events:
if not event in new_events:
debug('Event deleted or modified: `{0}`', event)
self.events.remove(event)
# Add new events to the list
for event in new_events:
if not event in self.events and not hash(event) in self.notified_events:
debug('Event not seen before: `{0}`', event)
# Does it start in the future?
self.events.append(event)
else:
debug('Event already registered: `{0}`', event)
self.events_lock.release()
debug('Finished updating events')
sleep(settings.query_sleeptime)
#-----------------------------------------------------------------------------#
# Signal Handlers #
# Signal handlers are easier than wrapping everything in a giant try/except. #
# Additionally, we have 2 threads that we need to shut down #
#-----------------------------------------------------------------------------#
def stopthismadness(self, signal, frame):
"""Halts execution and exits. Intended for SIGINT (^C)."""
message('Shutting down on SIGINT.')
exit(0)
#-----------------------------------------------------------------------------#
# Text color constants #
#-----------------------------------------------------------------------------#
class text:
purple = '\033[95m'
cyan = '\033[96m'
darkcyan = '\033[36m'
blue = '\033[94m'
green = '\033[92m'
yellow = '\033[93m'
red = '\033[91m'
bold = '\033[1m'
underline = '\033[4m'
normal = '\033[0m'
#-----------------------------------------------------------------------------#
# GCalert Settings #
#-----------------------------------------------------------------------------#
class settings(object):
"""Stores all settings for this gcalert instance."""
config_directory = '~/.config/gcalert/'
abs_config_directory = None
secrets_filename = '.gcalert_oauth'
rc_filename = 'gcalertrc'
secrets_file = None
rc_file = None
alarm_sleeptime = 300 # Seconds between waking up to check the alarm list
query_sleeptime = 180 # Seconds between querying for new events
lookahead_days = 3 # Look this many days in the future
debug_flag = False # Display debug messages
quiet_flag = False # Suppresses all non-debug messages
reconnect_sleeptime = 300 # Seconds between reconnects in case of errors
threads_offset = 2 # Offset between the two threads' runs, in seconds
strftime_string = '%H:%M %Y-%m-%d' # String to format times with
icon = 'gtk-dialog-info' # Icon to use in notifications
@staticmethod
def initialize_user_settings():
"""Initializes user settings from their gcalertrc and then from their commandline arguments."""
# TODO Parse the command line rcfile argument before actually parsing the rcfile
settings.abs_config_directory = path.expanduser(settings.config_directory)
settings.secrets_file = path.join(settings.abs_config_directory, settings.secrets_filename)
settings.rc_file = path.join(settings.abs_config_directory, settings.rc_filename)
# Create the config directory if it doesn't already exist
if not path.exists(settings.abs_config_directory):
makedirs(settings.abs_config_directory)
# Handle gcalertrc file arguments
if path.exists(settings.rc_file):
with open(settings.rc_file, 'r') as rc_file:
rc_arguments = rc_file.read().splitlines()
settings.handle_arguments(rc_arguments)
# Handle command line arguments
settings.handle_arguments(argv[1:])
@staticmethod
def handle_arguments(args):
"""Parses the given list of commandline arguments."""
try:
opts, args = getopt(
args, 'hdqs:u:c:a:l:r:t:i:', ['help', 'debug', 'quiet', 'secret=', 'rc=', 'check=', 'alarm=', 'look=', 'retry=', 'timeformat=', 'icon='])
except GetoptError as err:
# Print help information and exit:
print(err) # Will print something like "option -a not recognized"
exit(2)
try:
for o, a in opts:
if o in ('-d', '--debug'):
settings.debug_flag = True
elif o in ('-h', '--help'):
message(settings.usage(), force=True)
exit()
elif o in ('-q', '--quiet'):
settings.quiet_flag = True
elif o in ('-s', '--secret'):
settings.secrets_file = a
debug('Secrets file set to {0}', settings.secrets_file)
elif o in ('-u', '--rc'):
settings.rc_file = a
debug('gcalertrc file set to {0}', settings.rc_file)
elif o in ('-c', '--check'):
settings.query_sleeptime = max(int(a), 5)
debug('Query sleep time set to {0}', settings.query_sleeptime)
elif o in ('-a', '--alarm'):
settings.alarm_sleeptime = int(a)
debug('Alarm sleep time set to {0}', settings.alarm_sleeptime)
elif o in ('-l', "--look"):
settings.lookahead_days = int(a)
debug('Lookahead days set to {0}', settings.lookahead_days)
elif o in ('-r', '--retry'):
settings.reconnect_sleeptime = int(a)
debug('Reconnect sleep time set to {0}', settings.reconnect_sleeptime)
elif o in ('-t', '--timeformat'):
settings.strftime_string = a
debug("strftime format string set to {0}", settings.strftime_string)
elif o in ('-i', '--icon'):
settings.icon = a
debug('Icon set to {0}', settings.icon)
else:
assert False, 'Unsupported argument'
except ValueError:
message('Option {0} requires an integer parameter; use \'-h\' for help.', o)
exit(1)
@staticmethod
def usage():
return '''{program} {version} - Displays reminder notifications for Google Calendar events.
Usage: {executable} [options]
-u
--rc Specifies the location of the gcalertrc file.
This file may contain one command line parameter
per line, and will be used to configure {program}
before any command line arguments are parsed.
(Default: {default_rc})
-s
--secret Specifies the location of the oauth credentials cache.
(Default: {default_secret})
-d
--debug Print debug messages.
-q
--quiet Disable all non-debug messages.
-c
--check Number of seconds between queries for new calendar events.
(Default: {default_query})
-a seconds
--alarm seconds
Number of seconds between checking for reminders to display.
(Default: {default_alarm})
-l days
--look days
Number of days to look ahead when checking for new events.
(Default: {default_look})
-r seconds
--retry seconds
Number of seconds to wait between reconnection attempts.
(Default: {default_retry})
-t format_string
--timeformat format_string
Formatting string to use for displaying event times.
Must be formatted according to strftime(3).
(Default: {default_timeformat})
-i icon_name
--icon icon_name
Sets the icon displayed in reminder notifications.
(Default: {default_icon})'''.format(
program = __program__,
version = __version__,
executable = argv[0],
default_rc = settings.config_directory + settings.rc_filename,
default_secret = settings.config_directory + settings.secrets_filename,
default_query = settings.query_sleeptime,
default_alarm = settings.alarm_sleeptime,
default_look = settings.lookahead_days,
default_retry = settings.reconnect_sleeptime,
default_timeformat = settings.strftime_string,
default_icon = settings.icon
)
@staticmethod
def get_settings():
"""Returns a string representation of the settings."""
return '''
Secrets file: {secrets_file}
Alarm sleep time: {alarm_time}
Query sleep time: {query_time}
Lookahead days: {lookahead}
Debug: {debug}
Quiet: {quiet}
Reconnect sleep time: {reconnect_time}
Thread offset: {thread_offset}
strftime format: {strftime_str}
Icon: {icon}
'''.format(
secrets_file = settings.secrets_file,
alarm_time = settings.alarm_sleeptime,
query_time = settings.query_sleeptime,
lookahead = settings.lookahead_days,
debug = settings.debug_flag,
quiet = settings.quiet_flag,
reconnect_time = settings.reconnect_sleeptime,
thread_offset = settings.threads_offset,
strftime_str = settings.strftime_string,
icon = settings.icon
)
#-----------------------------------------------------------------------------#
# Let's get started! #
#-----------------------------------------------------------------------------#
if __name__ == '__main__':
GCalert()
# vim: ai expandtab