Rule

Interacting with items

Items are like variables. They have a name and a value (which can be anything). Items from openhab use the item name from openhab and get created when HABApp successfully connects to openhab or when the openhab configuration changes. Items from MQTT use the topic as item name and get created as soon as a message gets processed.

Some item types provide convenience functions, so it is advised to always set the correct item type.

The preferred way to get and create items is through the class factories get_item and get_create_item since this ensures the proper item class and provides type hints when using an IDE!

Example:

from HABApp.core.items import Item
my_item = Item.get_create_item('MyItem', initial_value=5)   # This will create the item if it does not exist
my_item = Item.get_item('MyItem')                           # This will raise an exception if the item is not found
print(my_item)

If an item value gets set there will be a ValueUpdateEvent on the event bus. If it changes there will be additionally a ValueChangeEvent, too.

It is possible to check the item value by comparing it

from HABApp.core.items import Item
my_item = Item.get_item('MyItem')

# this works
if my_item == 5:
    pass    # do something

# and is the same as this
if my_item.value == 5:
    pass    # do something
Inheritance diagram of HABApp.core.items.Item
class HABApp.core.items.Item(name, initial_value=None)

Simple item, used to store values in HABApp

classmethod get_create_item(name, initial_value=None)

Creates a new item in HABApp and returns it or returns the already existing one with the given name

Parameters
  • name (str) – item name

  • initial_value – state the item will have if it gets created

Returns

item

classmethod get_item(name)

Returns an already existing item. If it does not exist or has a different item type an exception will occur.

Parameters

name (str) – Name of the item

get_value(default_value=None)

Return the value of the item.

Parameters

default_value – Return this value if the item value is None

Return type

Any

Returns

value of the item

property last_change
Return type

datetime

Returns

Timestamp of the last time when the item has been changed (read only)

property last_update
Return type

datetime

Returns

Timestamp of the last time when the item has been updated (read only)

listen_event(callback, event_type)

Register an event listener which listens to all event that the item receives

Parameters
  • callback (Callable[[Any], Any]) – callback that accepts one parameter which will contain the event

  • event_type (Union[AllEvents, Any]) – Event filter. This is typically ValueUpdateEvent or ValueChangeEvent which will also trigger on changes/update from openHAB or mqtt.

Return type

EventBusListener

property name
Return type

str

Returns

Name of the item (read only)

watch_change(secs)

Generate an event if the item does not change for a certain period of time. Has to be called from inside a rule function.

Parameters

secs (Union[int, float]) – secs after which the event will occur, max 1 decimal digit for floats

Return type

BaseWatch

Returns

The watch obj which can be used to cancel the watch

watch_update(secs)

Generate an event if the item does not receive and update for a certain period of time. Has to be called from inside a rule function.

Parameters

secs (Union[int, float]) – secs after which the event will occur, max 1 decimal digit for floats

Return type

BaseWatch

Returns

The watch obj which can be used to cancel the watch

Events

Function

Description

listen_event()

Add a function which will be called as soon as an event occurs. The event will be passed as an argument into the function. There is the possibility to specify the event class which will reduce the function calls accordingly (typically ValueUpdateEvent or ValueChangeEvent).

Example:

def __init__(self):
    super().__init__()
    self.listen_event('MyOpenhabItem', self.on_change, ValueChangeEvent)
    self.listen_event('My/MQTT/Topic', self.on_update, ValueUpdateEvent)

    # If you already have an item you can use the more convenient method of the item
    # to listen to the item events
    my_item = Item.get_item('MyItem')
    my_item.listen_event(self.item_changed, ValueUpdateEvent)

def on_change(event):
    assert isinstance(event, ValueChangeEvent), type(event)

def on_update(event):
    assert isinstance(event, ValueUpdateEvent), type(event)

Scheduler

With the scheduler it is easy to call functions in the future or periodically. Do not use time.sleep but rather run_in().

Function

Description

run_soon()

Run the callback as soon as possible (typically in the next second).

run_in()

Run the callback in x seconds.

run_at()

Run a function at a specified date_time

run_every()

Run a function periodically

run_minutely()

Run a function every minute

run_hourly()

Run a function every hour

run_daily()

Run a function every day

run_on_every_day()

Run a function at a specific time every day

run_on_workdays()

Run a function at a specific time on workdays

run_on_weekends()

Run a function at a specific time on weekends

run_on_day_of_week()

Run a function at a specific time on specific days of the week

run_on_sun()

Run a function in relation to the sun (e.g. Sunrise, Sunset)

All functions return an instance of ScheduledCallbackBase

class HABApp.rule.scheduler.base.ScheduledCallbackBase(callback, *args, **kwargs)
earliest(time_obj)

Set earliest boundary as time of day

Parameters

time_obj (Optional[time]) – time obj, scheduler will not run earlier

Return type

ScheduledCallbackBase

latest(time_obj)

Set earliest boundary as time of day

Parameters

time_obj (Optional[time]) – time obj, scheduler will not run later

Return type

ScheduledCallbackBase

offset(timedelta_obj)

Set a constant offset to the calculation of the next run

Parameters

timedelta_obj (Optional[timedelta]) – constant offset

Return type

ScheduledCallbackBase

jitter(secs)

Add a random jitter per call in the intervall [(-1) * secs … secs] to the next run

Parameters

secs (Optional[int]) – jitter in secs

Return type

ScheduledCallbackBase

boundary_func(func)

Add a function which will be called when the datetime changes. Use this to implement custom boundaries

Parameters

func (Optional[Callable[[datetime], datetime]]) – Function which returns a datetime obj, arg is a datetime with the next call time

get_next_call()

Return the next execution timestamp

check_due(now)

Check whether the callback is due for execution

Parameters

now (datetime) –

Returns

execute()

Try to execute callback. If the callback is not due yet or execution has already finished nothing will happen

Return type

bool

Returns

True if callback has been executed else False

cancel()

Cancel execution

Running external tools

External tools can be run with the execute_subprocess() function. Once the process has finished the callback will be called with an FinishedProcessInfo instance as first argument. Example:

import HABApp

class MyExecutionRule(HABApp.Rule):

    def __init__(self):
        super().__init__()

        self.execute_subprocess( self.func_when_finished, 'path_to_program', 'arg1', capture_output=True)

    def func_when_finished(self, process_info):
        assert isinstance(process_info, HABApp.rule.FinishedProcessInfo)
        print(process_info)

MyExecutionRule()
class HABApp.rule.FinishedProcessInfo(returncode, stdout, stderr)

Information about the finished process.

Variables
  • returncode (int) – Return code of the process (0: IO, -1: Exception while starting process)

  • stdout (str) – Standard output of the process or None

  • stderr (str) – Error output of the process or None

How to properly use rules from other rule files

This example shows how to properly get a rule during runtime and execute one of its function. With the proper import and type hint this method provides syntax checks and auto complete.

Rule instances can be accessed by their name (typically the class name). In the HABApp.log you can see the name when the rule is loaded. If you want to assign a custom name, you can change the rule name easily by assigning it to self.rule_name in __init__.

Important

Always look up rule every time, never assign to a class member! The rule might get reloaded and then the class member will still point to the old unloaded instance.

rule_a.py:

class ClassA(Rule):
    ...

ClassA()

rule_b.py:

if typing.TYPE_CHECKING:            # This is only here to allow
    from .class_a import ClassA     # type hints for the IDE

class ClassB(Rule):
    ...

    def function_a(self):

        r = self.get_rule('ClassA')  # type: ClassA
        # The comment "# type: ClassA" will signal the IDE that the value returned from the
        # function is an instance of ClassA and thus provide checks and auto complete.

        # this calls the function on the instance
        r.function_b()

All available functions

class HABApp.Rule
Variables
post_event(name, event)

Post an event to the event bus

Parameters
  • name – name or item to post event to

  • event – Event class to be used (must be class instance)

Returns

listen_event(name, callback, event_type=<class 'HABApp.core.events.events.AllEvents'>)

Register an event listener

Parameters
  • name (Union[BaseValueItem, str]) – item or name to listen to. Use None to listen to all events

  • callback (Callable[[Any], Any]) – callback that accepts one parameter which will contain the event

  • event_type (Union[AllEvents, Any]) – Event filter. This is typically ValueUpdateEvent or ValueChangeEvent which will also trigger on changes/update from openhab or mqtt.

Return type

EventBusListener

execute_subprocess(callback, program, *args, capture_output=True)

Run another program

Parameters
  • callback – Function which will be called after process has finished. First parameter will be an instance of FinishedProcessInfo

  • program – program or path to program to run

  • args – Positional arguments that will be passed to the function

  • capture_output – Capture program output, set to False to only capture return code

Returns

run_every(time, interval, callback, *args, **kwargs)

Run a function periodically

Parameters
  • time (Union[None, datetime, timedelta, time]) – Use a datetime.time object to specify a certain time of day, a datetime.timedelta object to specify a time in the future or None to use the current time.

  • interval (Union[int, timedelta]) –

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

ReoccurringScheduledCallback

run_on_sun(sun_event, callback, *args, run_if_missed=False, **kwargs)

Run a function on sunrise/sunset etc

Parameters
  • sun_event (str) – ‘sunrise’, ‘sunset’, ‘dusk’, ‘dawn’

  • run_if_missed – run the event if we missed it for today

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

SunScheduledCallback

run_on_day_of_week(time, weekdays, callback, *args, **kwargs)
Parameters
  • time (time) – datetime.time

  • weekdays

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

DayOfWeekScheduledCallback

run_on_every_day(time, callback, *args, **kwargs)
Parameters
  • time (time) – datetime.time

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

DayOfWeekScheduledCallback

run_on_workdays(time, callback, *args, **kwargs)
Parameters
  • time (time) – datetime.time

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

DayOfWeekScheduledCallback

run_on_weekends(time, callback, *args, **kwargs)
Parameters
  • time (time) – datetime.time

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

DayOfWeekScheduledCallback

run_daily(callback, *args, **kwargs)

Picks a random hour, minute and second and runs the callback every day

Parameters
  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

ReoccurringScheduledCallback

run_hourly(callback, *args, **kwargs)

Picks a random minute and second and run the callback every hour

Parameters
  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

ReoccurringScheduledCallback

run_minutely(callback, *args, **kwargs)

Picks a random second and runs the callback every minute

Parameters
  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

ReoccurringScheduledCallback

run_at(date_time, callback, *args, **kwargs)

Run a function at a specified date_time

Parameters
  • date_time (Union[None, datetime, timedelta, time]) –

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

OneTimeCallback

run_in(seconds, callback, *args, **kwargs)

Run the callback in x seconds

Parameters
  • seconds (int) – Wait time in seconds or a timedelta obj before calling the function

  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

OneTimeCallback

run_soon(callback, *args, **kwargs)

Run the callback as soon as possible (typically in the next second).

Parameters
  • callback – Function which will be called

  • args – Positional arguments that will be passed to the function

  • kwargs – Keyword arguments that will be passed to the function

Return type

OneTimeCallback

register_on_unload(func)

Register a function with no parameters which will be called when the rule is unloaded. Use this for custom cleanup functions.

Parameters

func (Callable[[], Any]) – function which will be called

register_cancel_obj(obj)

Add a weakref to an obj which has a cancel function. When the rule gets unloaded the cancel function will be called (if the obj was not already garbage collected)

Parameters

obj