HABApp: Making Home Automation Easy and Fun - A Thank You Note to @Spaceman_Spiff

Thank you for your kind words. I’ve sunk way more than thousand hours into HABApp and sometimes I’ve wondered if it’s worth all the effort just that somebody else can reuse it, too.
So this post is (of course) very nice to hear and encouraging to keep going.

@RRoe I have some feedback. First if you define constants I’d use Parameter or DictParameter (Docs).
That way you can define them in a yaml file and they will be automatically reloaded if you change something.
You can also define them on a module level.

I’d use a fixed name for logger and not build it dynamically. The setLevel should be set in the logging.yaml as level. You can create the MyRule - logger there.

I’d put the NumberItem.get_item('EurToSekExchangeRate') into the __init__ and save the item as a member. That way you’re rule will fail fast if you have a typo in the item.

Scheduler is always without event, EventFilter always with event.

Other than that I’d say your rule looks pretty nice! :+1:


same, same

1 Like

I want also to appreciate the work of @Spaceman_Spiff.
In the last months I’ve reprogrammed all my codes within HABApp. That’s a variety from direct low level communication to my old heating controller (UVR1611/BLnet) to code which uses the deep integration of HABApp to OH (3.4) and the existing bindings or the interfaces of more modern devices.
I started to learn Python with this - and I had to learn a lot. After having learned something new, I reprogrammed and learned again… No rocket science and no professional programming style, but 16000 lines of fun.
As full examples are much to long, just two lines I’m using in all my functions as it is very helpful:

        LOGGER: str = "{}: ".format(inspect.currentframe().f_code.co_name)
        self.log.error(LOGGER + "some error message")

This starts the logged text with the name of the function.

Thanks for your feedback @Spaceman_Spiff

I will definitely adopt your suggestion to start using Parameter for storing my configuration constants in yml files instead. That will take me some time, but I believe it’s worth it.

Below is an excerpt showing how my one and only config file my_config.yml looks like:

    ADMIN_EMAIL: "some-dude@some-domain.nu"
    OPENHAB_HOST: "localhost"
    OPENHAB_PORT: "8090" # "8443"
    LOCAL_TIME_ZONE: "Europe/Stockholm"
    MY_LOGGER_NAME: "MyRule"
    AREA: 4

I have planned to use it like this:

    def __init__(self):
        # Get the parameters stored in the my_config.yml file
        self.system_configuration = Parameter('my_config', 'configuration', 'system', default_value=None).value
        self.entsoe_configuration = Parameter('my_config', 'configuration', 'entsoe', default_value=None).value
        time_from_iso = time.fromisoformat(self.entsoe_configuration['DAY_AHEAD_PRICES_ARRIVE_TIME_STR'])

I hope that my usage conforms to “best practise”, otherwise please let me know how it should be done.

I have actually created an entry for MyRule in logging.yml

    level: INFO
      - HABApp_default
    propagate: False

    level: INFO
      - EventFile
    propagate: False

  MyRule:   # <-- Name of the logger
    level: DEBUG
      - HABApp_default  # This logger uses the HABApp_default
    propagate: False

But I have a per rule setting, e.g. self.log.setLevel(logging.INFO) so that I can easily toggle logging on and off for the rule that I’m currently working on. I find that convenient and I cant see how to define per rule log levels in a different way.

I want to be able to see in my log what rule that created the log entry. I don’t know how to achieve the same in a different way.


Yes - but it’s DictParameter, Parameter is for simple values (e.g. upper boundaries).
If you want to reuse the DictParameter across multiple rules you can also create it as a global variable.
If you want to make it perfect you can also specify a dependency (with depends on) to the parameter file in the header of the rule file. That way the rule won’t be loaded in case you accidentally delete the yaml file (or create an invalid one).

I just mean if you don’t create the rules dynamically it doesn’t make any sense to create the logger dynamically. Either both static or both dynamic.

class EuroExchangeRate(HABApp.Rule):

    def __init__(self):
        self.log = logging.getLogger('MyRule.EuroExchangeRate')

# Create this additionally to MyRule
  MyRule.EuroExchangeRate:   # <-- Name of the logger (if you define it explicitly it won't be caught by MyRule)
    level: DEBUG             # <-- Use this to change the log level without having to reload the rule
      - HABApp_default
    propagate: False

By global variable, do you mean I should put the variable holding the DictParameter in a lib file and import it?

No, just make it a global variable in the module that holds the rule.
That way if you define multiple rules in one file all can use the global DictParameter.

Due to the possibility to automatically relad files importing into rule files is always difficult.
You can safely import libraries etc. but you should not share instances across files with the import mechanism.

1 Like

i created a rule for timed items. i use it for example for lights that should switch OFF automatically after a defined time, for exmple attic, garden or even my shaver that needs to be charged every week.

if somebody is wondering why i do the ‘boot_initialisations’ - earlier i often had issues with error messages when rebooting and the rule already loaded but the items still where not initialized.

# -*- coding: utf-8 -*-
import logging
from Email import email
from time import sleep

from HABApp import Rule
from HABApp.core.events import EventFilter
from HABApp.openhab.items import OpenhabItem, SwitchItem, NumberItem
from HABApp.openhab.events import ItemCommandEvent, ItemStateChangedEvent

class ItemTimer(Rule):
    Rule for automatic switching OFF a switch or a dimmer item with a individual timeout

    - Light item        Item to be timed, type must be 'Switch' or 'Dimmer'
    - Time left item    Remaining time until switch OFF, item type 'Number'
      (left_item_name)  if not specified it is the name of timer item + '_Left'
                        (Number iEK_Terrasse_Li_Left 'Restzeit[%d min]' <time>)
    - timer_time:       Time until switch OFF, passed as fix value (if not set then it uses the
                        value that is set in timer_item_name)
                        changing the dimmer level does not change the timer
                        for switch: timer restarts when press ON while running
    - timer_item_name:  Time until switch OFF, item type 'Number'
                        if not specified name of timer item + '_Timer'
                        must be persisted to be able to continue timer when running while shutdown
                        (Number iEK_Terrasse_Li_Timer 'Timer [%d min]' <settings> (iG_BtPersist) )
    and eventually:
    - unit:             unit of time, possible values 's' (seconds), 'm' (minutes), 'h' (hours)
                        if not specified time in minutes
        ItemTimer('iEK_Terrasse_Li', timer_item_name='iEK_Terrasse_Li_Timer',
                  left_item_name='iEK_Terrasse_Li_Left', unit='s')
    def __init__(self, item, timer_item_name=None, timer_time=None, left_item_name=None, unit='m'):
        self.log = logging.getLogger('My_HABApp.ItemTimer')
        self.log.debug(f'### init HABApp.Rule ItemTimer.py for {item:s} ###')

        self.item = item
        self.items_checklist = [self.item]
        self.timer_item_name = f'{self.item:s}_Timer' if timer_item_name is None else timer_item_name
        self.timer_time = timer_time
        if timer_time is None:
            self.timer_item_name = None
        self.left_item_name = f'{self.item:s}_Left' if left_item_name is None else left_item_name
        self.unit = unit

        self.is_dimmer = None
        self.timed_item = None
        self.time_left_item = None

        # wait when starting rule until all items are initialized
        self.init_wait_delta = 30       # Sec wait before next check
        self.init_wait_max = 600        # Sec wait before sending error message
        self.init_wait_curr = 0         # counter wait seconds (0 - init_wait_max)
        self.error_mailed = False       # to ensure errormail will be sent only once


    # *********************************************************************************************
    # check if all items are initialized to work properly
    # returns list with item names that are still None / UNDEF
    # *********************************************************************************************
    def items_offline_check(self):
        items_checklist = self.items_checklist[:]   # make a copy of self.items_checklist
        offline_list = []

        while items_checklist:
            item_name = items_checklist.pop()
            item_value = OpenhabItem.get_item(item_name).value

            if item_value is None:                  # check if NULL oder UNDEF:
                self.log.error(f'ItemTimer: {item_name:s} ist UNDEF')
                self.log.debug(f'### ItemTimer: {item_name:s} initialized ({item_value}) ###')

        self.log.debug(f'### ItemTimer: {item_name:s} offline check finished ###')
        return offline_list

    # *********************************************************************************************
    # - Items checken ob nicht None / UNDEF - erst danach Funktionalität aktivieren
    # - Listener für Dimmer bzw Switch aktivieren
    # *********************************************************************************************
    def boot_initialisations(self):
        offline_list = self.items_offline_check()
        if offline_list:
            if self.init_wait_curr >= self.init_wait_max and not self.error_mailed:
                self.log.error(f'ItemTimer: {offline_list:s} UNDEF, Timer deaktiviert')
                email.send_mail(f'Fehler bei Initialisierung ItemTimer {self.item}',
                f'ItemTimer Funktionen sind deaktiviert so lange {offline_list} UNDEF')
                self.error_mailed = True

            self.log.debug(f'### ItemTimer: Items needed for {self.item:s} UNDEF, wait '\
                           f'{self.init_wait_delta:d} seconds ###')
            self.init_wait_curr += self.init_wait_delta
            self.run.countdown(self.init_wait_delta, self.boot_initialisations).reset()

        if self.timer_time is None:  # fetch timer time from item when not passed
            self.timer_time = NumberItem.get_item(self.timer_item_name).value

        self.is_dimmer = bool(self.openhab.get_item(self.item).type == 'Dimmer')
        if self.is_dimmer:
            self.timed_item = OpenhabItem.get_item(self.item)
            self.timed_item.listen_event(self.timer_item_triggered, EventFilter(ItemStateChangedEvent))
            self.timed_item = SwitchItem.get_item(self.item)
            self.timed_item.listen_event(self.timer_item_triggered, EventFilter(ItemCommandEvent))

        self.time_left_item = NumberItem.get_item(self.left_item_name)

        if self.unit == 's':
            watch_change_sec = 1
        elif self.unit == 'm':
            watch_change_sec = 60
        elif self.unit == 'h':
            watch_change_sec = 3600
            self.log.critical(f'ItemTimer {self.item:s} ungültige Einheit übergeben, Abbruch!')

        watcher = self.time_left_item.watch_change(watch_change_sec)

        # continue timer if light was ON when rule loaded / systemstart
        if self.timed_item.is_on():
            time_left = self.time_left_item.value
            self.time_left_item.oh_post_update(99)  # time must change to trigger 'watch_change'
            self.log.info(f'### ItemTimer: {self.item:s} is ON at boot, continue timer '\
                          f'{time_left:d} {self.unit:s} ###')

        self.log.debug(f'### init HABApp.Rule ItemTimer.py for {self.item:s} finished ###')

    # *********************************************************************************************
    # one unit from countdown time elapsed, update item with new remaining time
    # *********************************************************************************************
    def time_left_elapsed(self, event):
        if self.time_left_item.value <= 0:
            return  # (needed because method also will be called if switched OFF)

        new_time = self.time_left_item.value - 1
        if new_time > 0:
            self.log.info(f'ItemTimer {self.item:s} Restzeit = {new_time:d} {self.unit:s}')

    # *********************************************************************************************
    # timer item switched ON or OFF, start or stop timer
    # *********************************************************************************************
    def timer_item_triggered(self, event: ItemCommandEvent):
        item_value = event.value

        if self.is_dimmer:
            if int(event.old_value) == 0:
                item_value = 'ON'
            elif int(item_value) > 0:

        if item_value == 'ON':
            self.log.info(f'ItemTimer {self.item:s} aktiviert für {self.timer_time:d} {self.unit:s}')

ItemTimer('iDachboden_Li', timer_time=30)
ItemTimer('iBad_Rasierer_Sd', timer_time=15, unit='h')

items needed:
Switch iDachboden_Li "Dachboden" <light> (iG_Licht) { channel="openwebnet:bus_on_off_switch:bticino:Dachboden_Li:switch" }
Number iDachboden_Li_Left "Restzeit[%d min]" <time> (iG_BtPersist)

Switch iBad_Rasierer_Sd "Bad Rasierer" <shaver> (iG_Steckdosen) { channel="openwebnet:bus_on_off_switch:bticino:Bad_Rasierer_Sd:switch" }
Number iBad_Rasierer_Sd_Left "Restzeit[%d Stunden]" <time> (iG_BtPersist)

No need to use **kwargs, you should use normal parameters

    def __init__(self, item, timer_item_name=None, timer_time=None, left_item_name=None, unit='m'):
        self.log = logging.getLogger('My_HABApp.ItemTimer')

Also as a general hint don’t use “%s” any more - use f strings (even if it’s a tiny bit slower, it’s much more readable)

self.log.error('ItemTimer: %s ist UNDEF', item_name)
self.log.error(f'ItemTimer: {item_name:s} ist UNDEF')

Also instead of an int you can always pass a timedelta object, so you don’t need the whole sec/min/hour logic.

thanks a lot for the tips, very kind of you!

i edited my post. not using **kwargs makes code much more readable.

the reason why i still used %s was the message from the python-linter in vsc that sais

Use lazy % formatting in logging functionsPylint(W1203:logging-fstring-interpolation)

but you are right, it is better readable and and no risk to forget adding the variable at the end :slight_smile:

The warning is correct.
But how often will this statement run? Even if it’ll run once a second you’ll only save a tiny amount of cpu time. Compare that to the much better readability. It’s a trade I’m always willing to make.
Maybe you can disable this warning in PyLint, that way you’re rules will look green. :wink:

1 Like

I’m giving all my code an overhaul and I have become aware that I should try to avoid my own little utility scripts whenever possible because it makes my scripts less readable and less usable when shared with others. For example I have a function named send_command_if_different which has been very convenient when creating one-liners. But I will ditch it so now I’m trying to figure out what to replace it with in my code. The idea is that I do not wish to alter an item in any way (causing rules to trigger or creating persistence points) if an item already has the target value that I want it to have.

    def switch_on_off(self, target_value):
        Turn the facade lights on or off depending on the target value and the current state of the lights
        self.log.debug(f'Turning {target_value} on facade lights')

        # One way to do it which also describes what I want to achieve
        if self.sch_facade_lights_item.get_value(ON) != target_value:

        # For post_value, there is the `oh_post_update_if` function
        self.sch_facade_lights_item.oh_post_update_if(target_value, not_equal=target_value)
        # For send_command, there's no `oh_send_command_if` function
        self.sch_facade_lights_item.oh_send_command_if(target_value, not_equal=target_value)
        # I have a function (which I don't want to use, due to that it's not very "HABAppian")
        send_command_if_different('Sch_Facade_Lights', 'ON')

Any comments highly appreciated and maybe @Spaceman_Spiff can comment on why we don’t have a oh_send_command_if function. :grinning:

The command does not translate to the item state and thus can lead to confusion.
That’s why I am hesitant to implement it for send_command because I (atm) believe it does not add clarity.

For switches this obviously works but for a dimmer a command will result in the item state 100.
So you would have to do something like dimmer.oh_send_command_if('ON', not_equal=100).
I’d rather use

if not self.my_dimmer:
# or
if not self.my_dimmer.is_on():

Typically I have some initializer logic and then HABApp is the only one to send commands so I guess for me it hasn’t made much sense to add this.
I normally work with item states and item state changes so that’s what will reduce the ammount of commands that will be sent.

because of a little bug in the openwebnet-binding i must not send an OFF command to a dimmer that already is OFF (this bug will be removed in oh4).

so my workaround for lights (switches and dimmers) is actually:

    def send_command_check_first(self, item, command):
        item_value = OpenhabItem.get_item(item).value

        if self.openhab.get_item(item).type == 'Dimmer':
            item_value = 'OFF' if int(item_value == 0) else 'ON'

        if not item_value == command:
            self.oh.send_command(item, str(command))

In my aim to eliminate all my personal helper scripts, (they make my scripts harder to share with others community members) I’ll try to use “vanilla” HABApp functionality. There are situations where post_update doesn’t seem to work well and I guess it’s like @watou once explained:

For example the following code won’t turn the wall plug on: (updated by MQTT binding)

self.wall_plug_kitchen_item.oh_post_update_if("ON", not_equal="ON")

However the following code works well:

self.wall_plug_kitchen_item.on() if not self.wall_plug_kitchen_item.is_on() else None


Why not do it like this? That way you have a dedicated branch for the workaround which you can remove once you get to OH4.

    def send_command_check_first(self, item, command):
        item = OpenhabItem.get_item(item)
        if isinstance(item, (DimmerItem, SwitchItem)):
            if item.is_on() or command != 'OFF':

I’m aware that post_update and send_command are two different things. One is to command devices and one is to update an internal state in openHAB.
However my argument is that except for switches the command that is sent to a devices does not translate to item state. So it seldom makes sense to compare the command with the item state and thus I’m not convinced a oh_send_command_if increases readability.

I’d probably write

if not self.wall_plug_kitchen_item.is_on():

A rule that runs at HABApp start. Some things need to be set up before other scripts can work properly. This script is named 001_init.py.

# HABApp:
#   depends on:
#    - params/my_config.yml

import logging
from datetime import datetime, timedelta

from HABApp import DictParameter, Rule
from HABApp.core.items import Item
from HABApp.openhab.definitions import OnOffValue
from HABApp.openhab.items import DatetimeItem, StringItem, SwitchItem
from nibe_f750_heat_pump import NibeF750HeatPump
from nord_pool_market_data import NordPoolMarketData

# Some useful constants
ON = OnOffValue.ON
OFF = OnOffValue.OFF

# Get the configurations parameters stored in the my_config.yml file
configuration = DictParameter('my_config', 'configuration', default_value=None).value

# Defining the following items here will detect any errors early in the development process.
clock_time_of_day_item = StringItem.get_item(configuration["custom_item_names"]['clock_time_of_day_item'])
its_not_early_morning_item = SwitchItem.get_item("Its_Not_Early_Morning")
solar_time_of_day_item = StringItem.get_item(configuration["custom_item_names"]['solar_time_of_day_item'])
v_civil_dawn_item = DatetimeItem.get_item('V_CivilDawn')
v_sunrise_item = DatetimeItem.get_item('V_Sunrise')
v_civil_dusk_start_item = DatetimeItem.get_item('V_CivilDuskStart')
v_civil_dusk_end_item =DatetimeItem.get_item('V_CivilDuskEnd')

class RunAtHABAppStart(Rule):
    A rule that runs at HABApp start.

    def __init__(self):
        logger_name = configuration["system"]["MY_LOGGER_NAME"]
        self.log = logging.getLogger(f"{logger_name}.{self.rule_name}")

    def init_routine(self):
        Initialization routine that is executed at HABApp start.
        self.log.info(f"[{self.rule_name}]: HABApp has started.")
        my_nord_pool_item = Item.get_create_item("MyNordPool", None)
        my_f750_item = Item.get_create_item("MyNibeF750", None)


class ClockTimeOfDay(Rule):
    A rule that determines the time of day based on the current hour.

    def __init__(self):
        logger_name = configuration["system"]["MY_LOGGER_NAME"]
        self.log = logging.getLogger(f"{logger_name}.{self.rule_name}")
        next_hour = (datetime.now() + timedelta(hours=1)).replace(minute=0, second=7)
        self.run.every(next_hour, timedelta(hours=1), self.process_changes)

    def process_changes(self):
        Process changes and update items based on the current hour.
        hour = datetime.now().hour
        clock_time_of_day = (
            if 6 <= hour <= 9
            else configuration["time_of_day"]["CLOCK_TIME_OF_DAY"][1]
            if 9 <= hour <= 17
            else configuration["time_of_day"]["CLOCK_TIME_OF_DAY"][2]
            if 18 <= hour <= 22
            else configuration["time_of_day"]["CLOCK_TIME_OF_DAY"][3]
        self.log.info("[%s]: The time of day (according to the clock) is [%s]", self.rule_name, clock_time_of_day)
        clock_time_of_day_item.oh_post_update_if(clock_time_of_day, not_equal=clock_time_of_day)
        target_value = OFF if 3 < hour < 9 else ON
        its_not_early_morning_item.oh_post_update_if(target_value, not_equal=target_value)


class SolarTimeOfDay(Rule):
    A rule that determines the solar time of day based on various events.

    def __init__(self):
        logger_name = configuration["system"]["MY_LOGGER_NAME"]
        self.log = logging.getLogger(f"{logger_name}.{self.rule_name}")
        self.run.soon(self.update_time_of_day, type_of_job='InitJob')
        self.run.on_sun_dawn(callback=self.update_time_of_day, type_of_job='DawnJob')
        self.run.on_sunrise(callback=self.update_time_of_day, type_of_job='SunriseJob')
        self.run.on_sun_dusk(callback=self.update_time_of_day, type_of_job='DuskJob')
        self.run.on_sunset(callback=self.update_time_of_day, type_of_job='SunsetJob')

    def update_time_of_day(self, type_of_job):
        Update the solar time of day based on the given type of job.
        solar_time_of_day = None
        if type_of_job == 'DawnJob':
            solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][0]
        elif type_of_job == 'SunriseJob':
            solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][1]
        elif type_of_job == 'SunsetJob':
            solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][2]
        elif type_of_job == 'DuskJob':
            solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][3]
            # Not triggered by a channel. Probably because the system just started.
            # Let's find out manually where we are then...
            datetime_now = datetime.now()

            dawn_start = v_civil_dawn_item.get_value(datetime_now)
            day_start = v_sunrise_item.get_value(datetime_now)
            dusk_start = v_civil_dusk_start_item.get_value(datetime_now)
            night_start = v_civil_dusk_end_item.get_value(datetime_now)

            if dawn_start <= datetime_now < day_start:
                solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][0]
            elif day_start <= datetime_now < dusk_start:
                solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][1]
            elif dusk_start <= datetime_now < night_start:
                solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][2]
                solar_time_of_day = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"][3]

        self.log.info("[%s]: The solar time of day is [%s]", self.rule_name, solar_time_of_day)


Same here. Much appreciation from me!


And me. Using openHAB’s own scripting language was painful - HABApp makes openHAB fantastic. Thank you!


This topic was automatically closed 41 days after the last reply. New replies are no longer allowed.