Vor einiger Zeit habe ich mich mit dem automatischen Abruf von Ereignissen aus einem iCal-Abo zur Anzeige auf einem (kleinen) ePaper Display beschäftigt.
Mangels Zeit zum weiter entwickeln bin ich noch bei einem einzigen Display und einer einzigen iCal URL.
Falls jemand einen ähnlichen Anwendungsfall hat, hier der Code den ich damals dafür geschrieben hatte:
(Ist nicht der schönste, aber vielleicht hilft es ja bei eigenen Projekten mit ähnlichen Anwendungsfällen, etc.)
Code: Alles auswählen
#!/usr/bin/env python3
import requests
from icalendar import Calendar, Event
from datetime import datetime, timedelta
import time
import xml.sax.saxutils
import pytz
import signal
import logging
import schedule
### Notes ###
# This is a very basic implementation fetching the contents of a single ical-url and updating a predefined display accordingly.
# The intention is to update a single display having it's own calendar.
# The event summary is displayed as 'purpose' and the event location as 'chair' in the template for purposes of simplicity.
# Furthermore, the start- and endtime of the event will be displayed as 'time' in the template.
# The room's name is predefined by a constant variable.
# Only one event will be displayed as it's intended for rendering on 4.2" displays
###
### Preferences ###
# Servers
ical_url = "https://ical_url" # iCal Url for retrieving the events; For nextcloud append '?export'
upload_url ="http://0.0.0.0:8001/service/task" # Url for uploading the xml-file to the server; Replace 0.0.0.0 with server ip (change port if necessary)
needsAuthentication = False # If Server requires authentication, set this to 'True' and provide a file named "credentials.txt" with one line containing user:password
# Times
retry_interval = 300 # interval for retrying after a failed down-/ upload
#If enabled, scheduler will use values below; Disable if you have problems with it. If disabled, update runs every hour at :50
#Some Ubuntu versions have problems with it; Still figuring out why...
advancedTimes = False
exec_start = 5 # first hour of execution (0-23)
exec_end = 17 # last hour of execution (0-23)
execute_minute = 50 # minute to execute the script (every hour)
# Label
label_id = "ABCD1234"
event_room = "Room" # Roomname displayed on label
external_id = "4711"
time_zone = pytz.timezone('Europe/Berlin') # Timezone; Should be the same as the calendar's
# Debug
writeXml = False # writeXML is fine for debugging; Not needed during daily operation
postXml = True # should always be True; If not, nothing will be uploaded to the Server
# logger
logging.basicConfig(filename='iCalToWEP.log', level=logging.INFO, format='%(asctime)s %(levelname)-8s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
logging.info('Started iCalToWEP')
###
# read credentials
if needsAuthentication :
with open("credentials.txt") as f:
username, password = f.read().strip().split(":")
# symbol filter
allowed_symbols = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789 äöüÄÖÜ")
# XML template
xml_template = '''<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<TaskOrder title="Refresh {labelid} for {room}">
<TemplateTask labelId="{labelid}" externalId="{extid}" template="ical_template_42.xsl">
<room roomName="{room}">
<field key="date" value="{date}"/>
<field key="time" value="{time}"/>
<field key="purpose" value="{purpose}"/>
<field key="chair" value="{chair}"/>
</room>
</TemplateTask>
</TaskOrder>
'''
# function to filter input string
def filter_string(input_string):
if input_string is not None:
return ''.join(filter(allowed_symbols.__contains__, input_string))
else:
return " "
# Define and register signal-handlers for SIGINT (Ctrl+C) and SIGTERM
def sigint_handler(signum, frame):
logging.info('Received SIGINT signal, exiting gracefully')
exit(0)
def sigterm_handler(signum, frame):
logging.info('Received SIGTERM signal, exiting gracefully')
exit(0)
signal.signal(signal.SIGINT, sigint_handler)
signal.signal(signal.SIGTERM, sigterm_handler)
# event tracking
last_event = None
# initialize past for writing empty xml
past = datetime.now(time_zone).date() - timedelta(days=1)
# Main function
def update_labels():
global label_id
global event_room
global external_id
global last_event
global past
# fetch calendar
while True:
try:
# If using your own ca, you can supply the path to the "ca_bundle.pem" instead of 'True' which uses the system's default ca-bundle; e.g. verify='/path/to/own/cert'
if needsAuthentication:
cal_content = requests.get(ical_url, timeout=10, verify=True, auth=(username,password)).text
else :
cal_content = requests.get(ical_url, timeout=10, verify=True).text
cal = Calendar.from_ical(cal_content) # Raise exception if not a valid calendar
except Exception as e:
print(f"Error fetching calendar: {e}")
logging.error("Exception occured while trying to fetch calendar", exc_info=True)
time.sleep(retry_interval)
continue
break
# get today's date and time
now = datetime.now(time_zone)
today = now.date()
current_time = now.time()
# get today's events
#events_today = [e for e in cal.walk() if e.name == "VEVENT" and isinstance(e.get('dtstart').dt, datetime) and e.get('dtstart').dt.date() == today and isinstance(e.get('dtend').dt, datetime) and e.get('dtend').dt.date() == today]
# get today's events
# ignore events with a date object instead of a datetime object (all-day events)
events_today = []
for e in cal.walk():
if e.name == "VEVENT":
try:
if e.get('dtstart').dt.date() == today:
events_today.append(e)
except AttributeError:
pass
# if there are events today
if len(events_today) > 0:
# get next event based on the difference of it's starttime to current time
next_event = min(events_today, key=lambda event: abs(event["DTSTART"].dt - now))
# check if next event is different from last event
#if next_event != last_event:
# do it the long way to prevent unnecessary updates caused by some ical-clients that "update" the event unnecessarily
if last_event is None or next_event["DTSTART"].dt != last_event["DTSTART"].dt or next_event["DTEND"].dt != last_event["DTEND"].dt or next_event.get("LOCATION") != last_event.get("LOCATION") or next_event["SUMMARY"] != last_event["SUMMARY"]:
logging.info('New event')
# update event tracking
last_event = next_event
# parse event data
event_start = next_event.get('dtstart').dt
event_end = next_event.get('dtend').dt
event_date = event_start.date().strftime('%d.%m.%Y')
event_start_time = event_start.time().strftime('%H:%M')
event_end_time = event_end.time().strftime('%H:%M')
event_summary = next_event.get('summary')
event_location = next_event.get('location')
#event_room = next_event.get('room') # Can be enabled to parse the room's name automatically (untested)
# filter event data
event_summary = filter_string(event_summary)
event_location = filter_string(event_location)
event_room = filter_string(event_room)
label_id = filter_string(label_id)
external_id = filter_string(external_id)
# escape XML characters using saxutils
# just in case the 'allowed_symbols' set was changed unintentionally
event_summary = xml.sax.saxutils.escape(event_summary)
event_location = xml.sax.saxutils.escape(event_location)
event_room = xml.sax.saxutils.escape(event_room)
label_id = xml.sax.saxutils.escape(label_id)
external_id = xml.sax.saxutils.escape(external_id)
# create XML
xml_data = xml_template.format(
labelid=label_id,
extid=external_id,
room=event_room,
date=event_date,
time=event_start_time + " - " + event_end_time,
purpose=event_summary,
chair=event_location
)
if writeXml:
# write XML to file
with open(f"{event_room}.xml", "w") as xml_file:
xml_file.write(xml_data)
logging.debug('Wrote event xml')
if postXml:
# Post XML to server
while True :
try:
headers = {'Content-Type': 'application/xml'}
response = requests.post(upload_url, data=xml_data, headers=headers, timeout=10)
response.raise_for_status()
logging.info('Posted event xml')
except Exception as e:
print(f"Error posting xml: {e}")
logging.error("Exception occured while trying to upload xml", exc_info=True)
time.sleep(retry_interval)
continue
break
#print(f"Response: {response}")
# Reset 'past' marker in case a reservation is made during the day and deleted on the same day (e.g. there were no reservations in the morning, a reservation is added at noon and deleted in the afternoon)
past = today - timedelta(days=1)
else:
logging.debug('No new event or change detected')
else:
logging.debug('No new event')
if today != past:
# Filter data fields
event_room = filter_string(event_room)
label_id = filter_string(label_id)
external_id = filter_string(external_id)
# escape XML characters
event_room = xml.sax.saxutils.escape(event_room)
label_id = xml.sax.saxutils.escape(label_id)
external_id = xml.sax.saxutils.escape(external_id)
# create xml
xml_data = xml_template.format(
labelid=label_id,
extid=external_id,
room=event_room,
date=today.strftime('%d.%m.%Y'),
time=" ",
purpose=" ",
chair=" "
)
# we only want to do this once per day to save battery
past = today
if writeXml:
# write XML to file
with open(f"{event_room}.xml", "w") as xml_file:
xml_file.write(xml_data)
logging.debug('Wrote empty xml for today')
if postXml:
while True :
# Post XML to server
try :
headers = {'Content-Type': 'application/xml'}
response = requests.post(upload_url, data=xml_data, headers=headers, timeout=10)
response.raise_for_status()
logging.info('Posted xml without events')
except Exception as e:
print(f"Error posting xml: {e}")
logging.error("Exception occured while trying to upload xml", exc_info=True)
time.sleep(retry_interval)
continue
break
#print(f"Response: {response}")
#Job scheduling
if advancedTimes:
if (exec_start <= exec_end) and (0 <= exec_start <= 23) and (0 <= exec_end <= 23):
execution_range = range(exec_start,exec_end+1)
#run job once after startup
update_labels()
#set scheduler
for hour in execution_range:
schedule.every().day.at(f"{hour}:{execute_minute}").do(update_labels)
while True:
schedule.run_pending()
time.sleep(60)
else:
logging.error('Value error for scheduler; Please check your settings')
exit(1)
else:
#run job once after startup
update_labels()
#set scheduler
schedule.every().hour.at(":50").do(update_labels)
while True:
schedule.run_pending()
time.sleep(60)
#Template for WEP-Server
#ical_template_42.xsl
#<?xml version="1.0" encoding="UTF-8"?>
#<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform" version="1.0">
#
# <xsl:template match="Record">
#
# <!-- Render for 4.2 inch labels -->
# <image height="300" width="400" rotation="0" font-family="Verdana">
#
# <!-- Room -->
# <field height="108" width="380" x="10" y="20">
# <text align="center" font-size="40" font-weight="bold">
# <utils method="toUpperCase">
# <xsl:value-of select="room/@roomName"/>
# </utils>
# </text>
#
# <!-- Date -->
# <text align="center" font-size="35" font-weight="bold" padding-top="10">
# <xsl:value-of select="room/field[@key='date']/@value"/>
# </text>
# </field>
#
# <line thickness="2" x-from="0" x-to="400" y-from="130" y-to="130"/>
#
# <!-- Time -->
# <field height="50" width="380" x="20" y="150">
# <text align="left" font-weight="bold" font-size="40">
# <xsl:value-of select="room/field[@key='time']/@value"/>
# </text>
# </field>
#
# <!-- Purpose -->
# <field height="50" width="380" x="20" y="200">
# <text align="left" font-size="40" condense="1, 0.8, 0.6, 0.5">
# <xsl:value-of select="room/field[@key='purpose']/@value"/>
# </text>
# </field>
#
# <!-- Chair -->
# <field height="40" width="370" x="20" y="250">
# <text align="left" font-weight="bold" font-size="30">
# <xsl:value-of select="room/field[@key='chair']/@value"/>
# </text>
# </field>
#
# </image>
# </xsl:template>
#</xsl:stylesheet>