Idea: use python to directly generate rules files

I’ve refactored the original code. If you want to test and debug outside Openhab, you need to use a different way to get the json response without using the core opehab libraries. That can be done with the following:

import requests
import logging

the_log = logging.getLogger('spam_application')
the_log.setLevel(logging.DEBUG)
the_log.addHandler(logging.StreamHandler())

def get_json_response(URI):
    log = the_log
    the_page = requests.get(URI)
    if not the_page or the_page.status_code != 200:
        log.warn("Could not get response from {}".format(URI))
        return
    try:
        #return json.loads(the_page.decode('utf-8'))
        return the_page.json()
    except Exception as ex:
        log.warn('Could not interpret json response from {}: {}'.format(URI, ex.message))

postcode = '5581BG'
huisnummer='17'
kalender_URI = 'https://afvalkalender.waalre.nl'
pickupdates = get_pickupdates_afvalkalender(postcode, huisnummer, kalender_URI)
the_log.info("Gegevens ophaaldagen voor {0}-{1}: {2}".format(postcode, huisnummer, pickupdates))

Hi @all,

despite the whole discussion I like the approach of a Python script generating openHAB events by “injecting” *.rules files. :sunglasses:

So let me share my piece of work here. My script generates rules up to two rules per day, if more than one event is found in the downloaded file. It shortens the message to fit on screen of my KNX display device:

#!/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 <https://www.gnu.org/licenses/>.

'''

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 = False

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()

waste_collection_notification.py.txt (6.9 KB)

To activate the script once a day in the early morning, one can add a cron job like this:
crontab -e
insert a line like this:
0 1 * * * /home/openhabian/waste_collection_notification/waste_collection_notification.py

For easy installation you can:

cd
mkdir waste_collection_notification
wget https://community.openhab.org/uploads/short-url/iWAB5yljPzJw8vlRuyKMKKgmZ4a.txt -O waste_collection_notification/waste_collection_notification.py
chmod +x waste_collection_notification/waste_collection_notification.py

If not yet done you need to install Python3 and Python3-vobject beforehand:
sudo apt install python3 python3-vobject

Have fun! :slight_smile:

Christian

1 Like

Looks good and glad I could inspire you.

Meanwhile I did another python script to feed OH with data. Just did not mention it here. Guess why :wink:

This time the script is for my solar panels (GoodWe). These powerinverters log their data to a database server somewhere (I guess far east). Only way to know what my panels are generating is by reading back from that server. Crappy but this inverter came with the panels and was affordable.

But ofcourse someone made a python script (concept is more like it) to read the server. I extended the script to push mqtt messages every 5 minutes with the last power actuals. So now any mqtt client can access that data. Also the screen I am working on for the living room which has no OH involvement.

Works like a charm. Having a good IDE to develop the code helped a lot in finding a nasty bug in the original code.

If anyone interested, let me know. I send you the GoodWe to mqtt code.

Greetings Matthijs

I pushed my solar panel dealer to install a Solar-Log interface next to the Inverter so I can use the corresponding openHAB binding ( https://www.openhab.org/addons/bindings/solarlog/).
:slight_smile: This way it’s pretty easy to get extensive statistics. :sunglasses:

Greetings
Christian

Since you really seem to enjoy coding in python in really encourage you to take a look at HABApp.
It allows you to seamlessly integrate your python code into openhab without the use of generation .rules files with a python script.
But creating a script that creates another script in another language feels so wrong if there are so many better solutions (HABApp & JSR223).

2 Likes

HABApp looks really good (briefly checked the documentation pages). Will give that a try soon.

1 Like