#!/usr/bin/python3 ''' Author of this script: Christian Pohl, https://www.chpohl.de. DISCLAIMER: I'm not responsible for any damage or data loss on your system. Use at your own risk! You have been warned, I did this quick and dirty. ;-) This script is inspired by matthijsfh (https://community.openhab.org/u/matthijsfh) and his idea: https://community.openhab.org/t/idea-use-python-to-directly-generate-rules-files/89055. License: GPLv3 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 . ''' import os import time import datetime import vobject # sudo apt install python3-vobject import urllib.request ''' ICS file download: https://www.awsh.de/service/abfuhrtermine/ Direct download: https://www.awsh.de/api_v2/collection_dates/1/ort/90/strasse/1950/hausnummern/0/abfallarten/R04-B02-D02-P04-W0/kalender.ics ''' ''' If you wan't to debug this script, you can set DEBUG = True to get much sonsole output and let the script read calendar data from local file. ''' DEBUG = True def print_log(message): if DEBUG: print(message) class WasteCollectionEvent: def __init__(self, event_type = '', date = datetime.datetime.now()): self.event_type = event_type self.date = date class WasteCollectionNotification: rules_path = '/etc/openhab2/rules/' # Don't forget the trailing slash ('/')! rule_file_name = 'waste_' rules_file_extension = '.rules' MAX_EVENT_COUNT = 2 # My KNX display device has two slots for free information display. MAX_CHAR_PER_EVENT_COUNT = 14 # My KNX display device has two slots for free information display. ''' Enter a valid direct download URL to the ICS file of your refuse disposal service for your address. If there's no direct download URL, you can create a file on your own and put it next to the this script. Have a look at "DEBUG" ''' direct_calendar_url = "https://www.awsh.de/api_v2/collection_dates/1/ort/90/strasse/1950/hausnummern/0/abfallarten/R04-B02-D02-P04-W0/kalender.ics" def __init__(self): self.events = list() def parse_input_ics(self): if DEBUG: # Read calendar data from the file: data = open('Abfuhrtermine.ics').read() else: # Read calendar data from direct download URL: response = urllib.request.urlopen(self.direct_calendar_url) data = response.read().decode('utf-8') # iterate through the contents for cal in vobject.readComponents(data): for component in cal.components(): if component.name == "VEVENT": self.events.append(component) def find_todays_events(self): today = datetime.datetime.now() print_log('Removing all auto generated rule files...') self.remove_all_generated_rule_files() today_events = list() print_log('today: ' + str(today.year) + '-' + str(today.month) + '-' + str(today.day)) for ev in self.events: event_date = ev.contents['dtstart'][0].value event_type = ev.summary.valueRepr() if not '(' in ev.summary.valueRepr() else \ ev.summary.valueRepr().split('(')[0] event_is_today = False if today.day == event_date.day and today.month == event_date.month and today.year == event_date.year: today_events.append(WasteCollectionEvent(event_type, event_date)) event_is_today = True print_log(('* ' if event_is_today else '') + 'Event: ' + str(event_date.year) + '-' + str(event_date.month) + '-' + str( event_date.day) + ' ' + event_type) self.add_openhab_rules(today_events) def remove_all_generated_rule_files(self): files = os.listdir(self.rules_path) print_log('Scanning files for deletion...') for file in files: print_log('Scanning file "' + file + '".') if os.path.isfile(self.rules_path + file): if file.startswith(self.rule_file_name) and file.endswith(self.rules_file_extension): print_log('Removing file "' + self.rules_path + file + '"') os.remove(self.rules_path + file) def add_openhab_rules(self, events): filepath = self.rules_path + self.rule_file_name + str(int(datetime.datetime.now().timestamp()*1000)) + self.rules_file_extension print_log('File to open: ' + filepath + '.') file = open(filepath, 'w') for i in range (0, self.MAX_EVENT_COUNT): if i < len(events): self.write_openhab_cron_rule(file, events[i], i+1) else: self.write_openhab_cron_rule(file, WasteCollectionEvent(), i+1) file.flush() file.close() print_log('File closed.') def write_openhab_cron_rule(self, file, waste_event, slot): # Using current time for openHAB cron job to make the rule fire shortly after creation. current_time = datetime.datetime.now() minute = current_time.minute + 1 # Give openHAB more than enough time to parse the rule. hour = current_time.hour if minute > 59: minute = minute - 60 if hour < 23: hour = hour + 1 else: # It's way to late... tonight nobody will work anymore... ;-) pass # Please note: It is extremely important that each rule is given a unique name. file.write('rule "Date: ' + str(waste_event.date) + ', slot: ' + str(slot) + ', type: ' + waste_event.event_type + '."\n') file.write('when\n') file.write(' // sec min hr dom mon dow yr\n') file.write(' Time cron "12 ' + str(minute) + ' ' + str(hour) + ' ' + str(waste_event.date.day) + ' ' + str(waste_event.date.month) + ' ?"\n') file.write('then\n') file.write(' logInfo("rules", "Auto generated rule by ' + os.path.basename(__file__) + ', date: ' + str(waste_event.date) + ', type: ' + waste_event.event_type + '")\n') file.write(' GF_Hallway_StatusText' + str(slot) + '.sendCommand("' + waste_event.event_type[0:self.MAX_CHAR_PER_EVENT_COUNT] + '") // postUpdate() does not trigger knxd to relay data to the KNX bus.\n') file.write('end\n\n') if __name__ == '__main__': wcn = WasteCollectionNotification() wcn.parse_input_ics() wcn.find_todays_events()