Calendar notifications with a Matrix.org bot
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.