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

An overview over the item types can be found on the HABApp item section, the openHAB item section and the the mqtt item section

Interacting with events

It is possible to listen to events through the listen_event() function. The passed function will be called as soon as an event occurs and the event will pe passed as an argument into the function.

There is the possibility to reduce the function calls to a certain event type with an additional event filter (typically ValueUpdateEventFilter or ValueChangeEventFilter).

An overview over the events can be found on the HABApp event section, the openHAB event section and the the MQTT event section

Example

from HABApp import Rule
from HABApp.core.events import ValueChangeEvent, ValueUpdateEvent, ValueChangeEventFilter, ValueUpdateEventFilter
from HABApp.core.items import Item

class MyRule(Rule):
    def __init__(self):
        super().__init__()
        self.listen_event('MyOpenhabItem', self.on_change, ValueChangeEventFilter())  # trigger only on ValueChangeEvent
        self.listen_event('My/MQTT/Topic', self.on_update, ValueUpdateEventFilter())  # trigger only on ValueUpdateEvent

        # If you already have an item you can and should 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.on_change, ValueUpdateEventFilter())

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

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

MyRule()

Additionally there is the possibility to filter not only on the event type but on the event values, too. This can be achieved by passing the value to the event filter. There are convenience Filters (e.g. ValueUpdateEventFilter and ValueChangeEventFilter) for the most used event types that provide type hints.

NoEventFilter

class NoEventFilter

Triggers on all events

EventFilter

class EventFilter(event_class, **kwargs)

Triggers on event types and optionally on their values, too

ValueUpdateEventFilter

class ValueUpdateEventFilter(value=_MissingType.MISSING)

ValueChangeEventFilter

class ValueChangeEventFilter(value=_MissingType.MISSING, old_value=_MissingType.MISSING)

AndFilterGroup

class AndFilterGroup(*args)

All child filters have to match

OrFilterGroup

class OrFilterGroup(*args)

Only one child filter has to match

Example

from HABApp import Rule
from HABApp.core.events import EventFilter, ValueUpdateEventFilter, ValueUpdateEvent, OrFilterGroup
from HABApp.core.items import Item

class MyRule(Rule):
    def __init__(self):
        super().__init__()
        my_item = Item.get_item('MyItem')

        # This will only call the callback for ValueUpdateEvents where the value==my_value
        my_item.listen_event(self.on_val_my_value, ValueUpdateEventFilter(value='my_value'))

        # This is the same as above but with the generic filter
        my_item.listen_event(self.on_val_my_value, EventFilter(ValueUpdateEvent, value='my_value'))

        # trigger if the value is 1 or 2 by using both filters with or
        my_item.listen_event(
            self.value_1_or_2,
            OrFilterGroup(
                ValueUpdateEventFilter(value=1), ValueUpdateEventFilter(value=2)
            )
        )

    def on_val_my_value(self, event: ValueUpdateEvent):
        assert isinstance(event, ValueUpdateEvent), type(event)

    def value_1_or_2(self, event: ValueUpdateEvent):
        assert isinstance(event, ValueUpdateEvent), type(event)

MyRule()

Scheduler

With the scheduler it is easy to call functions in the future or periodically. Do not use time.sleep but rather self.run.at. Another very useful function is self.run.countdown as it can simplify many rules!

Function

Description

soon()

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

at()

Run the callback in x seconds or at a specified time.

countdown()

Run a function after a time has run down

every()

Run a function periodically

every_minute()

Run a function every minute

every_hour()

Run a function every hour

on_every_day()

Run a function at a specific time every day

on_workdays()

Run a function at a specific time on workdays

on_weekends()

Run a function at a specific time on weekends

on_day_of_week()

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

on_sun_dawn()

Run a function on dawn

on_sunrise()

Run a function on sunrise

on_sunset()

Run a function on sunset

on_sun_dusk()

Run a function on dusk

All functions return an instance of ScheduledCallbackBase

class HABAppSchedulerView(context)
at(time, callback, *args, **kwargs)

Create a a job that will run at a specified time.

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

  • callback (Callable[[], Any]) – 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

OneTimeJob

Returns

Created job

countdown(expire_time, callback, *args, **kwargs)

Run a job a specific time after calling reset() of the job. Another subsequent call to reset() will start the countdown again.

Parameters
  • expire_time (Union[timedelta, float, int]) – countdown in seconds or a timedelta obj

  • callback (Callable[[], Any]) – 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

CountdownJob

Returns

Created job

every(start_time, interval, callback, *args, **kwargs)

Create a job that will run at a specific interval.

Parameters
  • start_time (Union[None, datetime, timedelta, time, int]) – First execution time

  • interval (Union[int, float, timedelta]) – Interval how the job is repeated

  • callback (Callable[[], Any]) – 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

ReoccurringJob

Returns

Created job

on_day_of_week(time, weekdays, callback, *args, **kwargs)

Create a job that will run at a certain time on certain days during the week.

Parameters
  • time (Union[time, datetime]) – Time when the job will run

  • weekdays (Union[str, Iterable[Union[str, int]]]) – Day group names (e.g. 'all', 'weekend', 'workdays'), an iterable with day names (e.g. ['Mon', 'Fri']) or an iterable with the isoweekday values (e.g. [1, 5]).

  • callback (Callable[[], Any]) – 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

DayOfWeekJob

Returns

Created job

on_every_day(time, callback, *args, **kwargs)

Create a job that will run at a certain time of day

Parameters
  • time (Union[time, datetime]) – Time when the job will run

  • callback (Callable[[], Any]) – 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

DayOfWeekJob

on_sunrise(callback, *args, **kwargs)

Create a job that will run on sunrise, requires a location to be set

Parameters
  • callback (Callable[[], Any]) – 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

SunriseJob

Returns

Created job

on_sunset(callback, *args, **kwargs)

Create a job that will run on sunset, requires a location to be set

Parameters
  • callback (Callable[[], Any]) – 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

SunsetJob

Returns

Created job

on_sun_dawn(callback, *args, **kwargs)

Create a job that will run on dawn, requires a location to be set

Parameters
  • callback (Callable[[], Any]) – 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

DawnJob

Returns

Created job

on_sun_dusk(callback, *args, **kwargs)

Create a job that will run on dusk, requires a location to be set

Parameters
  • callback (Callable[[], Any]) – 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

DuskJob

Returns

Created job

soon(callback, *args, **kwargs)

Run the callback as soon as possible.

Parameters
  • callback (Callable[[], Any]) – 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

OneTimeJob

every_minute(callback, *args, **kwargs)

Picks a random second and runs the callback every minute

Parameters
  • callback (Callable[[], Any]) – 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

ReoccurringJob

every_hour(callback, *args, **kwargs)

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

Parameters
  • callback (Callable[[], Any]) – 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

ReoccurringJob

on_weekends(time, callback, *args, **kwargs)

Create a job that will run at a certain time on weekends.

Parameters
  • time (Union[time, datetime]) – Time when the job will run

  • 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

DayOfWeekJob

Returns

Created job

on_workdays(time, callback, *args, **kwargs)

Create a job that will run at a certain time on workdays.

Parameters
  • time (Union[time, datetime]) – Time when the job will run

  • 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

DayOfWeekJob

Returns

Created job

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 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:

import HABApp

class ClassA(HABApp.Rule):
    ...

    def function_a(self):
      ...

ClassA()

rule_b.py:

import HABApp
import typing

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

class ClassB(HABApp.Rule):
    ...

    def function_b(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_a()

All available functions

class Rule
Variables
on_rule_loaded()

Override this to implement logic that will be called when the rule and the file has been successfully loaded

on_rule_removed()

Override this to implement logic that will be called when the rule has been unloaded.

post_event(name, event)

Post an event to the event bus

Parameters
  • name (Union[TypeVar(HINT_ITEM_OBJ, bound= BaseItem), str]) – name or item to post event to

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

Returns

listen_event(name, callback, event_filter=None)

Register an event listener

Parameters
  • name (Union[TypeVar(HINT_ITEM_OBJ, bound= BaseItem), str]) – item or name to listen to

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

  • event_filter (Optional[TypeVar(HINT_EVENT_FILTER_OBJ, bound= EventFilterBase)]) – Event filter. This is typically ValueUpdateEventFilter or ValueChangeEventFilter which will also trigger on changes/update from openhab or mqtt. Additionally it can be an instance of EventFilter which additionally filters on the values of the event. It is also possible to group filters logically with, e.g. AndFilterGroup and OrFilterGroup

Return type

TypeVar(HINT_EVENT_BUS_LISTENER, bound= EventBusListener)

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

Run another program

Parameters
  • callback (Callable[[Any], Any]) – 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

static get_items(type=None, name=None, tags=None, groups=None, metadata=None, metadata_value=None)

Search the HABApp item registry and return the found items.

Parameters
  • type (Union[Tuple[Type[TypeVar(HINT_ITEM_OBJ, bound= BaseItem)], ...], Type[TypeVar(HINT_ITEM_OBJ, bound= BaseItem)], None]) – item has to be an instance of this class

  • name (Union[str, Pattern[str], None]) – str (will be compiled) or regex that is used to search the Name

  • tags (Union[str, Iterable[str], None]) – item must have these tags (will return only instances of OpenhabItem)

  • groups (Union[str, Iterable[str], None]) – item must be a member of these groups (will return only instances of OpenhabItem)

  • metadata (Union[str, Pattern[str], None]) – str (will be compiled) or regex that is used to search the metadata (e.g. ‘homekit’)

  • metadata_value (Union[str, Pattern[str], None]) – str (will be compiled) or regex that is used to search the metadata value (e.g. ‘TargetTemperature’)

Return type

Union[List[TypeVar(HINT_ITEM_OBJ, bound= BaseItem)], List[BaseItem]]

Returns

Items that match all the passed criteria