Additional rule examples

Using the scheduler

from datetime import datetime, time, timedelta

from HABApp import Rule

class MyRule(Rule):

    def __init__(self):
        super().__init__(), 34, 20), weekdays=['Mo'], callback=self.run_mondays), 3, self.run_every_3s, 'arg 1', asdf='kwarg 1'), 00), self.run_workdays), 00), self.run_weekends)

    def run_every_3s(self, arg, asdf = None):
        print(f'run_ever_3s: {} : {arg}, {asdf}')

    def run_mondays(self):
        print('Today is monday!')

    def run_workdays(self):
        print('Today is a workday!')

    def run_weekends(self):
        print('Finally weekend!')


Mirror openHAB events to a MQTT Broker

import HABApp
from import ItemStateEvent, ItemStateUpdatedEventFilter
from HABApp.openhab.items import OpenhabItem

class ExampleOpenhabToMQTTRule(HABApp.Rule):
    """This Rule mirrors all updates from OpenHAB to MQTT"""

    def __init__(self):

        for item in self.get_items(OpenhabItem):
            item.listen_event(self.process_update, ItemStateUpdatedEventFilter())

    def process_update(self, event):
        assert isinstance(event, ItemStateEvent)

        print(f'/openhab/{} <- {event.value}')
        self.mqtt.publish(f'/openhab/{}', str(event.value))


Trigger an event when an item is constant

Get an even when the item is constant for 5 and for 10 seconds.

import HABApp
from HABApp.core.items import Item
from import ItemNoChangeEvent, EventFilter

class MyRule(HABApp.Rule):
    def __init__(self):

        my_item = Item.get_item('test_watch')

        # Create an event when the item doesn't change for 5 secs and
        # create a watcher for ItemNoChangeEvent with 5s const time

        # Just create an event when the item doesn't change for 10 secs

        # Listen to all ItemNoChangeEvents for the item
        my_item.listen_event(self.item_constant, EventFilter(ItemNoChangeEvent))

        # Set the item to a value to generate the ItemNoChangeEvent events

    def item_constant_5s(self, event):
        print(f'Item 5s const: {event}')

    def item_constant(self, event):
        print(f'Item const: {event}')

Item 5s const: <ItemNoChangeEvent name: test_watch, seconds: 5>
Item const: <ItemNoChangeEvent name: test_watch, seconds: 5>
Item const: <ItemNoChangeEvent name: test_watch, seconds: 10>

Turn something off after movement

Turn a device off 30 seconds after one of the movement sensors in a room signals movement.

import HABApp
from HABApp.core.items import Item
from import ValueUpdateEvent, ValueUpdateEventFilter

class MyCountdownRule(HABApp.Rule):
    def __init__(self):

        self.countdown =, self.switch_off)
        self.device = Item.get_item('my_device')

        self.movement1 = Item.get_item('movement_sensor1')
        self.movement1.listen_event(self.movement, ValueUpdateEventFilter())

        self.movement2 = Item.get_item('movement_sensor2')
        self.movement2.listen_event(self.movement, ValueUpdateEventFilter())

    def movement(self, event: ValueUpdateEvent):
        if self.device != 'ON':


    def switch_off(self):


Process Errors in Rules

This example shows how to create a rule with a function which will be called when any rule throws an error. The rule function then can push the error message to an openHAB item, use a notification service to send the error message to the mobile device or send an email with the error message. See Advanced Usage for more information about the available internal topics. It also uses the built in rate limiter to limit the amount of notifications.

import HABApp
from import HABAppException
from import EventFilter
from HABApp.util import RateLimiter

# Set up rate limiter to limit the amount of notifications
LIMITER = RateLimiter('MyNotifications')
LIMITER.parse_limits('5 in 1 minute', algorithm='fixed_window_elastic_expiry')
LIMITER.parse_limits("20 in 1 hour", algorithm='leaky_bucket')

class NotifyOnError(HABApp.Rule):
    def __init__(self):

        # Listen to all errors
        self.listen_event('HABApp.Errors', self.on_error, EventFilter(HABAppException))

    def on_error(self, error_event: HABAppException):
        msg = error_event.to_str() if isinstance(error_event, HABAppException) else error_event

        # use limiter
        if not LIMITER.allow():
            return None

        # Replace this part with your notification logic
        print('Error in rules:')


# this is a faulty rule as an example. Do not create this part!
class FaultyRule(HABApp.Rule):
    def __init__(self):

    def faulty_function(self):
        1 / 0
Error in rules:
Exception in TestRule.FaultyRule.faulty_function: division by zero
File "<string>", line 46 in faulty_function

Traceback (most recent call last):
  File "/home/docs/checkouts/", line 94, in run
    self.func_obj(*self.func_args, **self.func_kwargs)
  File "<string>", line 46, in faulty_function
ZeroDivisionError: division by zero