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

Hey everyone, B Synnerlig here! Today, I want to express my gratitude to @Spaceman_Spiff for creating HABApp - a fantastic environment for home automation scripting. As someone who’s not a super technical programmer, I can’t tell you how much I appreciate the countless hours he’s poured into this project. Thanks to HABApp, I’ve been able to write my own scripts and automate my home using OpenHAB in ways I never thought possible. In this thread, I’d like to share some of my scripts with you all and invite you to ask questions, provide feedback, and suggest improvements. Who knows, maybe my scripts will inspire some of you to try your hand at home automation too! So without further ado, let’s get started!

Oh, and if you’re one of those people who thinks reading technical documentation is about as fun as watching paint dry, let me tell you - @Spaceman_Spiff has somehow managed to make the HABApp documentation a delight to peruse. Yes, you read that right - delightful. Don’t believe me? Go check it out for yourself at https://habapp.readthedocs.io/en/latest/ - I promise you’ll be pleasantly surprised (and maybe even a little bit amused).

This thread is subject for discussions and might become long. That’s OK but for your convenience I have made a gitHub repository to which I will transfer the scripts that I share. I won’t share them all at once but will gradually add them there when I have checked them.


I aim to use the HABApp functionality as much as possible but sometimes I need to define my own little helper functions. I’ve put all those into a file named myutils.py that resides in the habapp/lib folder:

import json
import logging
import subprocess
from datetime import date, datetime
from typing import Dict, Optional, Union

from HABApp import DictParameter
from HABApp.mqtt.items import MqttItem
from HABApp.openhab.definitions import OnOffValue, OpenClosedValue, UpDownValue
from HABApp.openhab.items import NumberItem, StringItem

configuration = DictParameter('my_config', 'configuration', default_value=None).value
lighting_configuration = DictParameter('lighting_config', 'configuration', default_value=None).value
LIGHT_LEVEL = lighting_configuration["lighting"]["LIGHT_LEVEL"]
SOLAR_TIME_OF_DAY = configuration["time_of_day"]["SOLAR_TIME_OF_DAY"]
CLOCK_TIME_OF_DAY = configuration["time_of_day"]["CLOCK_TIME_OF_DAY"]
lametric_configuration = configuration["lametric"]
customItemNames = configuration["custom_item_names"]
sonos = configuration["sonos"]

log = logging.getLogger(f'{configuration["system"]["MY_LOGGER_NAME"]}.myutils')

# Some useful constants
ON = OnOffValue.ON
OFF = OnOffValue.OFF
OPEN = OpenClosedValue.OPEN
CLOSED = OpenClosedValue.CLOSED
UP = UpDownValue.UP
DOWN = UpDownValue.DOWN

SPC_AREA_MODE = {'unset': 0, 'partset_a': 1, 'partset_b': 2, 'set': 3}
PRIO = {'LOW': 0, 'MODERATE': 1, 'HIGH': 2, 'EMERGENCY': 3}

def calendar_days_between(start_date_or_datetime: Union[date, datetime], end_date_or_datetime: Union[date, datetime]) -> int:
    Get the number of calendar days between two dates or datetimes.
    start_date = start_date_or_datetime.date() if isinstance(start_date_or_datetime, datetime) else start_date_or_datetime
    end_date = end_date_or_datetime.date() if isinstance(end_date_or_datetime, datetime) else end_date_or_datetime
    num_days_between = (start_date - end_date).days
    return num_days_between

def get_key_for_value(dictionary: Dict[str, str], value: str) -> Optional[str]:
    In a given dictionary, get the first key that has a value matching the one provided.

        dictionary (dict): the dictionary to search
        value (str): the value to match to a key

        str or None: string representing the first key with a matching value, or
            None if the value is not found
    return next((k for k, v in dictionary.items() if v == value), None)

def spc_area_is_set():
    Returns True if the SPC alarm area 1 is either partial or fully set otherwise returns False
    item_value = StringItem.get_item('SPC_Area_1_Mode').value
    if item_value is not None:
        return SPC_AREA_MODE.get(item_value, 0) > 0
        return False

def spc_area_is_fully_set():
    Returns True if the SPC alarm area 1 is fully set otherwise returns False
    item_value = StringItem.get_item('SPC_Area_1_Mode').value
    if item_value is not None:
        return SPC_AREA_MODE.get(item_value, 0) == 3
        return False

def spc_area_is_partially_set():
    Returns True if the SPC alarm area 1 is fully set otherwise returns False
    item_value = StringItem.get_item('SPC_Area_1_Mode').value
    if item_value is not None:
        return SPC_AREA_MODE.get(item_value, 0) == 1
        return False

def get_compass_direction(degrees):
    Returns the compass direction abbreviation (Swedish) for the given compass degree
        (0, 22.5): 'N',
        (22.5, 67.5): 'NNO',
        (67.5, 112.5): 'O',
        (112.5, 157.5): 'OSO',
        (157.5, 202.5): 'S',
        (202.5, 247.5): 'SSO',
        (247.5, 292.5): 'V',
        (292.5, 337.5): 'VNV',
        (337.5, 360): 'N'
    for degree_range, direction in COMPASS_DIRECTIONS.items():
        if degree_range[0] <= degrees < degree_range[1]:
            return direction

def is_light_level_bright() -> bool:
    """Checks if the light level is bright."""
    light_level_item = NumberItem.get_item(customItemNames['sysLightLevel'])
    return light_level_item.value == LIGHT_LEVEL['BRIGHT'] if light_level_item.value is not None else False

def is_light_level_shady() -> bool:
    """Checks if the light level is shady."""
    light_level_item = NumberItem.get_item(customItemNames['sysLightLevel'])
    return light_level_item.value <= LIGHT_LEVEL['SHADY'] if light_level_item.value is not None else False

def is_light_level_dark() -> bool:
    """Checks if the light level is dark."""
    light_level_item = NumberItem.get_item(customItemNames['sysLightLevel'])
    return light_level_item.value <= LIGHT_LEVEL['DARK'] if light_level_item.value is not None else False

def is_light_level_black() -> bool:
    """Checks if the light level is black."""
    light_level_item = NumberItem.get_item(customItemNames['sysLightLevel'])
    return light_level_item.value <= LIGHT_LEVEL['BLACK'] if light_level_item.value is not None else False

def is_clock_night() -> bool:
    """Checks if the clock indicates night."""
    clock_time_item = StringItem.get_item(customItemNames['clock_time_of_day_item'])
    return clock_time_item.value == CLOCK_TIME_OF_DAY[3] if clock_time_item.value is not None else False

def is_solar_night() -> bool:
    """Checks if the solar position indicates night."""
    solar_time_item = StringItem.get_item(customItemNames['solar_time_of_day_item'])
    return solar_time_item.value == SOLAR_TIME_OF_DAY[3] if solar_time_item.value is not None else False

def calculate_speech_time_secs(text: str) -> int:
    '''Calculates the delay required for speaking a text string based on the character length and speaking speed.'''
    character_count = len(text)


class Notification:
    def __init__(self, notification_or_url, priority=PRIO['MODERATE'], **kwargs):
        self.notification_or_url = notification_or_url
        self.priority = priority
        self.room = kwargs.get('tts_room', NOTIFICATION_DEFAULT_ROOM)
        self.volume = kwargs.get('tts_volume', None)
        self.language = kwargs.get('tts_lang', sonos.get("rooms", {}).get(self.room, {}).get("tts_lang", NOTIFICATION_DEFAULT_LANGUAGE)) # Use param if exist, else config, else default
        self.voice = kwargs.get('tts_voice', sonos.get("rooms", {}).get(self.room, {}).get("tts_voice", NOTIFICATION_DEFAULT_VOICE))
        self.gender = kwargs.get('tts_gender', sonos.get("rooms", {}).get(self.room, {}).get("tts_gender", NOTIFICATION_DEFAULT_GENDER))
        self.engine = kwargs.get('tts_engine', sonos.get("rooms", {}).get(self.room, {}).get("tts_engine", NOTIFICATION_DEFAULT_ENGINE))
        self.only_when_playing = kwargs.get('only_when_palying', NOTIFICATION_DEFAULT_ONLY_WHEN_PLAYING)
        self.delay_ms = kwargs.get('delay_ms', 0) if 1 <= kwargs.get('delay_ms', 0) <= 1000 else 500
        self.timeout = kwargs.get('timeout', NOTIFICATION_DEFAULT_TIMEOUT)
        self.mp3_timeout = NOTIFICATION_DEFAULT_MP3_TIMEOUT
        self.language_server = f'http://{sonos["TTS_HOST"]}:5601/api/generate'
        self.mqtt_topic = ""
        self.payload = ""

    def volume(self):
        if not self._volume or self._volume >= 70:
            if self.priority == PRIO['LOW']:
                return 30
            elif self.priority == PRIO['MODERATE']:
                return 40
            elif self.priority == PRIO['HIGH']:
                return 60
            elif self.priority == PRIO['EMERGENCY']:
                return 70
                return 50
        return self._volume

    def volume(self, value):
        self._volume = value

    def should_play(self):
        # Get current hour
        current_hour = datetime.now().hour

        # Check if current hour is outside the range of 7 AM to 9 PM
        is_outside_range = current_hour < 7 or current_hour > 21

        # Check if outside range or if the SPC alarm is set and priority level is low
        if (is_outside_range or spc_area_is_set()) and self.priority <= PRIO['MODERATE']:
            log_message = f"Message priority [{get_key_for_value(PRIO, self.priority)}] is too low to play the notification '{self.notification_or_url}' at this moment."
            return False
        return True

    def play_mp3(self):
        # Get current hour
        return self.notification_or_url.lower().endswith('.mp3')

    def mqtt_topic(self):
        # Get the the mqtt topic
        return self._mqtt_topic

    def mqtt_topic(self, value):
        if self.play_mp3():
            self._mqtt_topic = 'sonos/set/notify' if self.room == "All" else f'sonos/set/{self.room}/notify'
            self._mqtt_topic = "sonos/set/speak" if self.room == "All" else f'sonos/set/{self.room}/speak'

    def payload(self):
        # Get the payload
        return self._payload

    def payload(self, value):
        if self.play_mp3():
            track_uri = f'http://{sonos["TTS_HOST"]}:5601/cache/sounds/{self.notification_or_url}'
            self._payload = { "trackUri": track_uri, "volume": self.volume, "timeout": self.mp3_timeout, "onlyWhenPlaying": self.only_when_playing }
            if self.delay_ms and 0 < self.delay_ms < 2001:
                self._payload['delayMs'] = self.delay_ms
            if self.timeout and 0 < self.timeout < 250:
                self._payload['timeout'] = self.timeout
            self._payload = { "text": self.notification_or_url, "endpoint": self.language_server, "lang": self.language, "gender": self.gender, "engine": self.engine, "volume": self.volume, "onlyWhenPlaying": self.only_when_playing }
            if self.voice is not None:
                self._payload['name'] = self.voice
            if self.delay_ms and 0 < self.delay_ms < 2001:
                self._payload['delayMs'] = self.delay_ms
            if self.timeout and 0 < self.timeout < 250:
                self._payload['timeout'] = self.timeout

    def play(self):
        # Play the notification
        if not self.should_play():
            return False
        mqtt_pub(self.mqtt_topic, json.dumps(self.payload))
        return True

def play_notification(notification_or_url, priority=PRIO['MODERATE'], **kwargs):
    Plays a notification on the Sonos system.
    The notification can be either a text string or an URL to an mp3 file.
    notification = Notification(notification_or_url, priority, **kwargs)
    return notification.play()

def speak_text(text_to_speak, priority=PRIO['MODERATE'], **keywords):
    Text To Speak function. First argument is positional and mandatory.
    Remaining arguments are optionally keyword arguments.
    Example: speak_text("Hello")
    Example: speak_text("Hello", PRIO['HIGH'], tts_room='Kitchen', tts_volume=42, tts_lang='en-GB', tts_voice='Brian')
    @param param1: Text to speak (positional argument)
    @param param2: priority as defined by PRIO. Defaults to PRIO['MODERATE']
    @param tts_room: Room to speak in. Defaults to "TV-Rummet".
    @return: this is a description of what is returned
    return play_notification(text_to_speak, priority, **keywords)

def play_sound(play_file, ttsPrio=PRIO['MODERATE'], **keywords):
    To play a short sound file as a notification
    return play_notification(play_file, ttsPrio, **keywords)

def mqtt_pub(topic, payload):
    Publishes a MQTT message on the default brooker
    log.debug('%s <- %s', topic, payload)

def send_notification_to_lametric(notification_text: str = 'HELLO!', notification_prio: int = PRIO['MODERATE'], **keywords: Dict) -> bool:
    Sends a notification to the LaMetric device.
    Documentation @ https://lametric-documentation.readthedocs.io/en/latest/reference-docs/device-notifications.html
    Possible keywords: sound, icon, autoDismiss, lifeTime, iconType
    log.debug('Sending a notification to LaMetric')

    def is_quiet_hours(start_hour, end_hour):
        """Returns True if the current time is within the quiet hours."""
        now_hour = datetime.now().hour
        return ((now_hour < start_hour) or (now_hour > end_hour))

    def should_play_sound(notification_prio):
        """Returns True if a sound should be played with the notification."""
        if notification_prio <= PRIO['MODERATE'] and spc_area_is_set():
            return False
        if is_quiet_hours(7, 21):
            return False
        return True    

    sound = lametric_configuration.get('DEFAULT_NOTIFICATION_SOUND') if 'sound' not in keywords else keywords['sound']
    icon = lametric_configuration.get('DEFAULT_ICON') if 'icon' not in keywords else keywords['icon']
    auto_dismiss = True if 'autoDismiss' not in keywords else keywords['autoDismiss']
    life_time = lametric_configuration.get('DEFAULT_LIFETIME') if 'lifeTime' not in keywords else keywords['lifeTime']
    icon_type = 'info' if 'iconType' not in keywords else keywords['iconType'] # [none|info|alert]

    url = f'https://{lametric_configuration["HOST"]}:{lametric_configuration["PORT"]}/api/v2/device/notifications'
    priority = 'critical' #"priority": "[info|warning|critical]" Must be critical to break through the app
    # cycles – the number of times message should be displayed. If cycles is set to 0, notification will stay on the screen until user dismisses it manually or you can dismiss it via the API (DELETE /api/v2/device/notifications/:id). By default it is set to 1.

    cycles = 1 if auto_dismiss else 0 # cycles – the number of times message should be displayed. If cycles is set to 0, notification will stay on the screen until user dismisses it manually or you can dismiss it via the API (DELETE /api/v2/device/notifications/:id). By default it is set to 1.
    payload = { 'priority': priority, 'icon_type': icon_type, 'lifeTime': life_time, 'model': { 'frames': [ { 'icon': icon, 'text': notification_text} ], 'cycles': cycles } }
    if should_play_sound(notification_prio):
        payload['model']['sound'] = { 'category': 'notifications', 'id': sound, 'repeat': 1 }
        log.info(f"The notification_prio argument {notification_prio} is too low to play a sound together with the notification at this moment")

    args_list = [
        'Content-Type: application/json',

    result = subprocess.run(args_list, capture_output=True)
    if result.returncode != 0:
        log.error(f"Error executing command: {result.returncode}")
        return False
    return True

def greeting():
    return f'God{StringItem.get_item(customItemNames["clock_time_of_day_item"]).value.lower()}'

                                 Safety pig has arrived!

                                  _._ _..._ .-',     _.._(`))
                                 '-. `     '  /-._.-'    ',/
                                    )         \            '.
                                   / _    _    |             \
                                  |  a    a    /              |
                                  \   .-.                     ;
                                   '-('' ).-'       ,'       ;
                                      '-;           |      .'
                                         \           \    /
                                         | 7  .__  _.-\   \
                                         | |  |  ``/  /`  /
                                        /,_|  |   /,_/   /
                                           /,_/      '`-'

I keep my configuration and settings in a separate file that I’ve put in the param directory.File name is my_configuration.yml

Just a small excerpt from that file:

      LOWEST: -2
      LOW: -1
      NORMAL: 0
      HIGH: 1
      EMERGENCY: 2
    api_token: 'aXXXXXXXXXXXXXXXXXXX364'
    SWEA_API_LATEST_OBSERVATIONS_URL: 'https://api-test.riksbank.se/swea/v1/Observations/Latest/SEKEURPMI'

I get access ro the configuration parameters through the DictParameter objec.

# HABApp:
#   depends on:
#    - rules/001_init.py
#    - params/my_config.yml
#   reloads on:
#    - params/my_config.yml

import json
import logging
import math
from HABApp import DictParameter, Rule

from HABApp.core.events import (EventFilter, ValueChangeEventFilter,
from HABApp.core.events.habapp_events import HABAppException
from HABApp.core.items import Item
from HABApp.mqtt.items import MqttItem
from HABApp.openhab.definitions import OnOffValue, OpenClosedValue, UpDownValue
from HABApp.openhab.events import ItemStateChangedEventFilter
from HABApp.openhab.items import (ContactItem, DatetimeItem, DimmerItem, GroupItem,
                                  NumberItem, OpenhabItem, StringItem,
... ... ... ... some code removed
# Get the configurations parameters stored in the my_config.yml file
configuration = DictParameter('my_config', 'configuration', default_value=None).value

autoremote_configuration = configuration["autoremote"]
robonect_configuration = configuration["robonect"]
riksbanken_configuration = configuration["riksbanken"]
surveillance_configuration = configuration["surveillance"]

Here goes my first script
It fetches the exchange rate for EUR to SEK and stores it in an OpenHAB Item. I haven’t included all the imports that I’m doing in the beginning of the file but you’ll probably figure that out anyway.

class EuroExchangeRate(Rule):
    Gets the latest exchange rate for EUR to SEK from Riksbanken

    def __init__(self):
        self.log = logging.getLogger(f'{configuration["system"]["MY_LOGGER_NAME"]}.{self.rule_name}')
        self.eur_to_sek_exchange_rate_item = NumberItem.get_item('EurToSekExchangeRate')
        self.eur_to_sek_exchange_rate_last_request_item = DatetimeItem.get_item('EurToSekExchangeRateLastRequest')
        self.run.every(timedelta(minutes=1), timedelta(hours=1), self.process_changes)

    def _on_subprocess_finished(self, process_output: str):
            # Parse the JSON response string into a Python dictionary
            response_dict = json.loads(process_output)
        except Exception as e:
            self.log.error(f"(Ο_Ο) Failed converting the response from Riksbanken to json when requesting latest exchange rate for EUR to SEK: {e}")

        # Check if the 'success' variable is set to True
        if 'value' in response_dict:
            # Extract the 'SEK' value as a float
            sek_value = float(response_dict['value'])
            self.log.info(f'-------------------> Latest exchange rate for 1 EUR to SEK is {round(sek_value, 2)}')
            self.log.error('(Ο_Ο) Requesting latest exchange rate for EUR to SEK from Riksbanken was not successful.')

    def process_changes(self, event=None):
        Go get the exchange rate for Euro to SEK from Riksbanken.
        # Read the time stamp value stored in item (Which might not be the same as the items last update time)
        self.datetime_now = datetime.now()
        last_request_made = self.eur_to_sek_exchange_rate_last_request_item.get_value(self.datetime_now - timedelta(hours=self.HOURS_BETWEEN_REQUESTS))

        if not (int((self.datetime_now - last_request_made).total_seconds()) > self.HOURS_BETWEEN_REQUESTS * 3600):
            #self.log.debug('Do not risk spamming the Riksbanken Exchange Rate API')

        url = f"{riksbanken_configuration['SWEA_API_LATEST_OBSERVATIONS_URL']}?seriesID=SEKEURPMI"
        args_list = [
            'Cache-Control: no-cache',
            'Accept: application/json',
            f'Ocp-Apim-Subscription-Key: {riksbanken_configuration["OCP_APIM_SUBSCRIPTION_KEY"]}',
        self.execute_subprocess(self._on_subprocess_finished, 'curl', *args_list, capture_output=True) # type: ignore


The API key can be obtained from: https://developer.api-test.riksbank.se/

i appreciate you want to share some of your scripts because is also like habapp wery much and want to say thank you to @Spaceman_Spiff for this excellent enviroment.

earlier i had jython installed, this was the first time i got in touch with python and found great possibillities to make your home more smart.

now i switched to habapp and most of my rules i created totally new. poor that i am still not a good coder at all (and surely will never become) i am very glad to find examples and how others do things.

1 Like

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)