After setting up Radicale as my self-hosted calendar system (CalDav), I hit the issue that my wife and I have a shared calendar and I couldn't find a good way to get notifications that a new event has been added/changed there, which was a nice feature of the iCloud calendar, which is what we were using before.

At first I set up a python script that would monitor the calendar every 30s and send us an e-mail if there's anything new, that worked fine but it wasn't very elegant: I'm already flooded with e-mails and this wouldn't make it any better. Also, if we wanted to talk about the event it would require forwarding the e-mail and have the discussion on another medium, while we've been mostly using Matrix and Riot to chat already.

So I decided to try out the matrix-python-sdk library and build a bot that would send a message to our chat room when there's something new on our calendar.

Making a basic matrix.org bot

First, let's create a Python 3 virtual environment and install the packages we're going to use:

$ virtualenv --python=python3 venv
$ source venv/bin/activate
$ pip install matrix_client caldav icalendar

From there, let's assume you have a matrix server running on my-matrix.com with an user called @my-user:my-matrix.com already joined a room with internal ID of !ABcDEFgHizIJlmnop:my-matrix.com.

from matrix_client.client import MatrixClient
client = MatrixClient("https://my-matrix.com")
token = client.login_with_password(username="my-user", password="my-password")
room = client.get_rooms()["!ABcDEFgHizIJlmnop:my-matrix.com"]
room.send_text("Hello World!")

And that's it! Can't get simpler than that!

Caldav

Now, let's connect to the calendar:

import caldav
from icalendar import Calendar, Event
client = caldav.DAVClient(caldav_server,
                          username=caldav_user,
                          password=caldav_passwd)

From there we can list the events in the calendar:

principal = self.client.principal()
for calendar in principal.calendars():
    for event in calendar.events():
        asciidata = event.data.encode("ascii","ignore")
        c = Calendar.from_ical(asciidata)
        for comp in c.walk():
            if comp.name == "VEVENT":
                v = [d.strftime("%Y-%m-%d %H:%M:%S") \
                     for d in (comp[u"DTSTART"].dt,  \
                               comp[u"DTEND"].dt,    \
                               comp[u"DTSTAMP"].dt)]

The idea is to just gather that list every 30s and see if there's anything new. The full source is on the bottom of this post.

Systemd service

Now, I want to run that script on start-up time on my server in the background, so let's create a systemd unit file called "caldav_bot.service":

[Unit]
Description=calbot

[Service]
Type=simple
User=myuser
Group=mygroup
WorkingDirectory=/home/myuser/caldav_bot
ExecStart=/home/myuser/caldav_bot/venv/bin/python /home/myuser/caldav_bot/caldav_bot.py
Restart=always

[Install]
WantedBy=multi-user.target

And let's create a link for that in /etc/systemd/system:

$ sudo su
# cd /etc/systemd/system
# ln -s /home/myuser/caldav_bot/caldav_bot.service .
# systemctl start caldav_bot
# systemctl status caldav_bot

And the status message should say the daemon is running!

Full source for Calbot

Edit: Thanks to dubtooth on reddit for pointing out that the script needs a token for your matrix user, you can get one by doing:

curl -XPOST -d '{"type":"m.login.password", "user":"my-user", "password":"my-password"}' "https://my-matrix.com/_matrix/client/r0/login"

#!/usr/bin/env python
from matrix_client.client import MatrixClient
import caldav
from icalendar import Calendar, Event
import pickle
import time
import os

config = {
    "caldav" : {
        "url"      : "",
        "user"     : "",
        "password" : ""
    },
    "matrix" : {
        "user_id" : "",
        "room" : "",
        "base_url" : "",
        "token" : ""
    }
}

class CaldavWatcher(object):
    def __init__(self, caldav_server, caldav_user, caldav_passwd):
        self.client = caldav.DAVClient(caldav_server, username=caldav_user, password=caldav_passwd)
        self.data_loc = "/tmp/caldav_data.set"
        open(self.data_loc, 'a+').close()
        os.chmod(self.data_loc, 700) # accessible only by root

    def sync(self):
        try:
            with open(self.data_loc, 'rb') as f:
                old_s = pickle.load(f)
        except:
            old_s = set()

        store = set()

        principal = self.client.principal()
        for calendar in principal.calendars():
            for event in calendar.events():
                asciidata = event.data.encode("ascii","ignore")
                c = Calendar.from_ical(asciidata)
                for comp in c.walk():
                    if comp.name == "VEVENT":
                        v = [d.strftime("%Y-%m-%d %H:%M:%S") for d in (comp[u"DTSTART"].dt, comp[u"DTEND"].dt, comp[u"DTSTAMP"].dt)]
                        store.add( (comp["SUMMARY"], v[0], v[1], v[2]) )

        with open(self.data_loc, 'wb') as f:
            pickle.dump(store, f)

        print("finished caldav sync")
        k = store - old_s
        return k

    def run(self):
        d = self.sync()
        if len(d) > 0:
            print("new events: %d" % len(d))
            msg = ""
            for k in d:
                summary = k[0]
                start = k[1]
                end   = k[2]
                msg += "New event: %s\n\tStart:\t%s\n\tEnd:\t%s\n" % (summary,start,end)
            if len(d) > 5: return "There is %d new events" % len(d)
            else: return msg
        else:
            return None

def main(config):

    matrix_config = config["matrix"]
    caldav_config = config["caldav"]

    # setup api/endpoint
    client = MatrixClient(matrix_config["base_url"],
            token=matrix_config["token"],
            user_id=matrix_config["user_id"])

    room = client.get_rooms()[matrix_config["room"]]

    cal = CaldavWatcher(caldav_config["url"],
            caldav_config["user"],
            caldav_config["password"])

    while True:
        m = cal.run()
        print("message", m)
        if m:
            room.send_notice("New Calendar Events:\n" + m)

        time.sleep(30)

if __name__ == '__main__':
    main(config)

Edit: discussion on reddit.