Idea: use python to directly generate rules files

OH 2.5.0-M5-1

JAR:

/usr/share/openhab2/addons#
-rwxr-xr-x 1 openhab openhab   112191 Aug 15 08:22 org.openhab.binding.velbus-2.5.0-SNAPSHOT13d.jar
-rwxr-xr-x 1 openhab openhab 37403000 Jan  3 17:19 org.openhab.core.automation.module.script.scriptenginefactory.jython-2.5.0-SNAPSHOT.jar

Hmmmm, I’m not certain that the jar will work on a version of OH that old. You may need to move to 2.5 M6 or preferably 2.5.0 release.

oke. Last time I tried to update, I got the “famous” mqtt error (and while figuring that out, killed my installation). Will make a backup tonight and update.

5iver indeed states: Have OH 2.5 (S1778) or newer

So lets try.

Yes, you will need 2.5 in order to get a last minute PR for openHAB core that let’s this bundle work.

Progres!

Almost there. Upgrade to 2.5.0 worked. Jython is running and the script is loaded.

BUT, there is a small bug in the script . See below. So now my question is, how can I run the script without setting the crontab every time. What is the “commandline way” to run these scripts?

# File "/etc/openhab2/automation/lib/python/core/log.py", line 51, in wrapper
# return fn(*args, **kwargs)
# File "<script>", line 91, in send_trash_pickup_notifications
# AttributeError: 'java.time.ZonedDateTime' object has no attribute 'zonedDateTime'

So something is wrong with this part of the code and I like to debug / test various alternatives I found online.

def convert_date(day_text):
    day_list = day_text.split(" ")
    new_date = DateTimeType().zonedDateTime.withMonth(monthToNum(day_list[2])).withDayOfMonth(int(day_list[1])).withHour(0).withMinute(0).withSecond(0).withNano(0)
    return new_date

How to proceed?
(of course the bug fix is appreciated too, but the general question remains: how to run scripts on demand).

Thanks,
Matthijs

It’s quite easy to test scripts. Here are 3 options…

  1. Add a “System started” trigger, which will trigger the rule when the file is saved. This is how I tested this rule before posting.
@when("System started")
  1. Rather than trigger the rule, just call the rule function in the script. This can be tricky, depending on the event object that is expected. Fortunately, the event object for cron triggers and startup triggers are None.
send_trash_pickup_notifications(None)
  1. Just create another script and add the body of the rule function, or portions of it. I have several test scripts with commented out tests that I am constantly using when working on things.

To debug this, you will probably want to wrap things in try/except. This is done for you with the rule decorator, along with a traceback, using the core.log.log_traceback decorator, which comes in handy during testing.

As for the error, I’ve made a small fix in the 4th line from the end that I had commented out during testing because I don’t use Telegram. convert_date returns a ZonedDateTime, so pickup_date is a ZonedDateTime, not a DateTimeType.

In addition to the three options Scott provided I have a couple more.

  1. Click the play button next to the Rule in PaperUI. As with running the function from another script or System started, it may not work if you are expecting something specific in the event object. I suspect this can be done from the REST API docs and the Karaf console too.

  2. Manually command or update the triggering item using another rule, a script, the REST API docs, or the Karaf console. This approach will be the closest to how the rule will actually run in real life for Item or Member of triggered rules.

Guess what? Python is a scripting language used in programming.

Speaking from experience, it takes time and dedication to learn Python. I took the time myself to learn it in the past 18 months.

If you cannot follow simple programming. you could not implement your idea for this thread anyway.

Think I will be alright :slight_smile: Both options are working fine now. Got some very good help in this thread.

Controlled pretty complex machines over the last years with python. So I do have some experience, but not a programming background.

CU,
Matthijs

1 Like

Come on, Bruce! Your entire response here is harsh, uncalled for, and completely unhelpful! Definitely not the attitude that is encouraged here. The OP obviously has Python experience and had written a solution in Python that solved his needs. I only provided an example of using scripted automation w/ Jython so that he could learn from it.

1 Like

@matthijsfh, your post inspired me to solve the problem slightly differently. I made a new post here. Unfortunately, this does not work for your postal code.

@ArdKuijpers Nice one!

And very good to point out the option to develop parts of code in an environment with a proper debugger (and code completion). Having an IDE with those features helped me a lot with the initial code I made (outside OH jython). Fixing the small bug in the 5iver code did costs me relative much time due to the absence of a debugger.

Best of both worlds.

Grtz

It does work with a modified version of of the get_pickupdates function. Look here.

Yep, my fault. It DOES work for my address.

You first get at “BagID” (what is that?) then the actual data. That is pretty cool.

Great.

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