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)

<Item name: MyItem, value: 5, last_change: InstantView(2024-12-09T13:53:44.146314193+00:00), last_update: InstantView(2024-12-09T13:53:44.146314193+00:00)>

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=<Missing>)

ValueChangeEventFilter

class ValueChangeEventFilter(value=<Missing>, old_value=<Missing>)

ValueCommandEventFilter

class ValueCommandEventFilter(value=<Missing>)

AndFilterGroup

class AndFilterGroup(*args)

All child filters have to match

OrFilterGroup

class OrFilterGroup(*args)

Only one child filter has to match

Example

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
        my_item.listen_event(self.on_val_my_value, ValueUpdateEventFilter())

        # 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.once. 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).

once()

Run the callback at a specified time.

countdown()

Run a function after a time has run down

at()

Run the callback when a trigger fires.

Scheduler

class HABAppJobBuilder(context)
countdown(secs, callback, *args, job_id=None, **kwargs)

Create a job that count town a certain time and then execute.

Parameters:

  • secs (timedelta | TimeDelta | int | float | str) – countdown time in seconds

  • callback (Callable[[ParamSpec(HINT_CB_P, bound= None)], Any]) – Function which will be called

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

  • job_id (Hashable | None)

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

Return type:

CountdownJobControl

Returns:

Created job

once(instant, callback, *args, job_id=None, **kwargs)

Create a job that runs once.

Parameters:
Return type:

OneTimeJobControl

Returns:

Created job

at(trigger, callback, *args, job_id=None, **kwargs)

Create a job that will run when a provided trigger occurs.

Parameters:

  • trigger (TriggerObject)

  • coro_func – Function which will be called

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

  • job_id (Hashable | None)

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

Return type:

DateTimeJobControl

Returns:

Created job

soon(callback, *args, job_id=None, **kwargs)

Run the callback as soon as possible.

Parameters:

  • callback (Callable[[ParamSpec(HINT_CB_P, bound= None)], Any]) – Function which will be called

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

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

Return type:

OneTimeJobControl

Example

from HABApp import Rule
from HABApp.rule.scheduler import filter, trigger

class MyTriggerRule(Rule):
    def __init__(self):
        super().__init__()

        # Run the function every day at 12
        self.run.at(self.run.trigger.time('12:00:00'), self.dummy_func)

        # ------------------------------------------------------------------------------
        # The trigger and filter factories are available as a property on self.run,
        # however they can also be used separately when imported
        # This is exactly the same as above
        self.run.at(trigger.time('12:00:00'), self.dummy_func)


        # ------------------------------------------------------------------------------
        # It's possible to trigger on sun position
        # ------------------------------------------------------------------------------

        # Run the function at sunrise
        self.run.at(self.run.trigger.sunrise(), self.dummy_func)


        # ------------------------------------------------------------------------------
        # Filters can be used to restrict the trigger
        # ------------------------------------------------------------------------------

        # Run the function every workday at 12
        self.run.at(
            self.run.trigger.time('12:00:00').only_on(self.run.filter.weekdays('Mo-Fr')),
            self.dummy_func
        )


        # ------------------------------------------------------------------------------
        # Triggers offer operations which can shift the trigger time
        # ------------------------------------------------------------------------------

        # Run the function one hour after sunrise
        self.run.at(self.run.trigger.sunrise().offset(3600), self.dummy_func)

        # Run the function one hour after sunrise, but but earliest at 8
        self.run.at(self.run.trigger.sunrise().offset(3600).earliest('08:00:00'), self.dummy_func)


        # ------------------------------------------------------------------------------
        # Triggers can be grouped together
        # ------------------------------------------------------------------------------

        # Run the function every workday at 12, but on the weekends at 8
        self.run.at(
            self.run.trigger.group(
                self.run.trigger.time('12:00:00').only_on(self.run.filter.weekdays('Mo-Fr')),
                self.run.trigger.time('08:00:00').only_on(self.run.filter.weekdays('Sa,So')),
            ),
            self.dummy_func
        )


        # ------------------------------------------------------------------------------
        # Filters can be grouped together
        # ------------------------------------------------------------------------------

        # Run the function at the first Sunday of every month at 12
        self.run.at(
            self.run.trigger.time('12:00:00').only_on(
                self.run.filter.all(
                    self.run.filter.weekdays('So'),
                    self.run.filter.days('1-7')
                )
            ),
            self.dummy_func
        )


    def dummy_func():
        pass

MyTriggerRule()

Reoccuring Jobs

Reoccuring jobs are created with at(). The point in time when the job is executed is described by Triggers. These triggers can be combined and/or restricted with filters.

class TriggerBuilder
static dawn()

Triggers at dawn.

Return type:

TriggerObject

static sunrise()

Triggers at sunrise.

Return type:

TriggerObject

static noon()

Triggers at noon.

Return type:

TriggerObject

static sunset()

Triggers at sunset.

Return type:

TriggerObject

static dusk()

Triggers at dusk.

Return type:

TriggerObject

static sun_elevation(elevation, direction)

Triggers at a specific sun elevation

Parameters:

  • elevation (float) – Sun elevation in degrees

  • direction (Literal['rising', 'setting']) – rising or falling

Return type:

TriggerObject

static sun_azimuth(azimuth)

Triggers at a specific sun azimuth

Parameters:

azimuth (float) – Sun azimuth in degrees

Return type:

TriggerObject

static group(*builders)

Group multiple triggers together. The triggers will be checked and the trigger that runs next will be used

Parameters:

builders (TriggerObject) – Triggers that should be grouped together

Return type:

TriggerObject

static interval(start, interval)

Triggers at a fixed interval from a given start time.

Parameters:

  • start (datetime | None | str | time | Time | timedelta | TimeDelta | int | float | SystemDateTime) – When this trigger will run for the first time. Note: It’s not possible to specify a start time greater than the interval time. Since the producer is stateless it will automatically select the next appropriate run time. Example: start 90 minutes, interval 60 minutes -> first run will be in 30 minutes. This makes it easy to ensure that the job will always run at a certain time by specifying the start time isntead of a delta and the interval. E.g. start 00:15:00 and interval 1 hour

  • interval (timedelta | TimeDelta | int | float | str) – The interval how this trigger will be repeated

Return type:

TriggerObject

static time(time, *, clock_forward=None, clock_backward=None)

Triggers at a specific time of day. When the time of day is during a daylight saving time transition it has to be explicitly specified how the transition should be handled.

Parameters:

  • time (time | Time | str) – The time of day the trigger should fire

  • clock_forward (Union[SkippedTimeBehavior, Literal['skip', 'earlier', 'later', 'after'], None]) – How to handle the transition when the clock moves forward

  • clock_backward (Union[RepeatedTimeBehavior, Literal['skip', 'earlier', 'later', 'twice'], None]) – How to handle the transition when the clock moves backward

Return type:

TriggerObject

class TriggerObject(producer)
offset(offset)

Offset the time returned by the trigger

Parameters:

offset (timedelta | TimeDelta | int | float | str) – The offset (positive or negative)

Return type:

Self

earliest(earliest, *, clock_forward=None, clock_backward=None)

Set the earliest time of day the trigger can fire. If the trigger would fire before the earliest time, the earliest time will be used instead.

Parameters:

  • earliest (time | Time | str) – The time of day before the trigger can not fire

  • clock_forward (Union[SkippedTimeBehavior, Literal['skip', 'earlier', 'later', 'after'], None]) – How to handle the transition when the clock moves forward

  • clock_backward (Union[RepeatedTimeBehavior, Literal['skip', 'earlier', 'later', 'twice'], None]) – How to handle the transition when the clock moves backward

Return type:

Self

latest(latest, *, clock_forward=None, clock_backward=None)

Set the latest time of day the trigger can fire. If the trigger would fire after the latest time, the latest time will be used instead.

Parameters:

  • latest (time | Time | str) – The time of day before the trigger can not fire

  • clock_forward (Union[SkippedTimeBehavior, Literal['skip', 'earlier', 'later', 'after'], None]) – How to handle the transition when the clock moves forward

  • clock_backward (Union[RepeatedTimeBehavior, Literal['skip', 'earlier', 'later', 'twice'], None]) – How to handle the transition when the clock moves backward

Return type:

Self

jitter(low, high=None)

Add jitter to the time returned by the trigger.

Parameters:
Return type:

Self

only_on(filter)

Add a filter to the trigger which can be used to allow or disallow certain times.

Parameters:

filter (FilterObject) – The filter to apply to the trigger

Return type:

Self

only_at(filter)

Add a filter to the trigger which can be used to allow or disallow certain times.

Parameters:

filter (FilterObject) – The filter to apply to the trigger

Return type:

Self

class FilterBuilder
static any(*filters)

Passes an instant as soon as any of the filters passes it

Return type:

FilterObject

static all(*filters)

Passes an instant only when all the filters pass it

Return type:

FilterObject

static not_(filter)

Invert a filter

Return type:

FilterObject

static time(lower=None, upper=None)

Restrict the trigger instant to a time range

Return type:

FilterObject

static weekdays(*weekdays)

Let only certain weekdays pass through

Parameters:

weekdays (int | str | Iterable[int | str]) – days of week as str, int, or str range ('Mon-Fri', 'Sun-Mon', 'Mo-Mi, 'Fr-So')

Return type:

FilterObject

static days(*days)

Let only certain days of the month pass through

Parameters:

days (int | str | Iterable[int | str]) – days of the month as str, int, or str range ('1-7', '1-5,10-15', '1,5,7')

Return type:

FilterObject

static months(*months)

Let only certain months pass through

Parameters:

months (int | str | Iterable[int | str]) – Month as str, int, or str range ('Jan-Jun', '1-3,10-12', 'Oct-Feb', 'Jan, March')

Return type:

FilterObject

static holidays(holidays=None)

Let only holidays pass through

Parameters:

holidays (HolidayBase | None) – Optional holiday object to use. If not provided the default holiday object will be used Userful if you want to use holidays from e.g. another country or another subdivision

Return type:

FilterObject

static work_days(holidays=None)

Let only working days pass through

Parameters:

holidays (HolidayBase | None) – Optional holiday object to use. If not provided the default holiday object will be used Userful if you want to use holidays from e.g. another country or another subdivision

Return type:

FilterObject

static not_work_days(holidays=None)

Let only days pass through that are not working days

Parameters:

holidays (HolidayBase | None) – Optional holiday object to use. If not provided the default holiday object will be used Userful if you want to use holidays from e.g. another country or another subdivision

Return type:

FilterObject

Job Control

class OneTimeJobControl()
cancel()

Cancel the job

Return type:

Self

property id: Hashable

Get the job’s id

property last_run_datetime: datetime | None

Get the last run time as a naive datetime object (without timezone set) or None if yet run

property next_run_datetime: datetime | None

Get the next run time as a naive datetime object (without timezone set) or None if not scheduled

property status: JobStatusEnum

Get the status of the job

to_item(item)

Sends the next execution (date)time to an item. Sends None if the job is not scheduled. Every time the scheduler updates to a new (date)time the item will also receive the updated time.

Parameters:

item (str | BaseValueItem | None) – item name or item, None to disable

Return type:

None

class CountdownJobControl()
set_countdown(secs)

Set the countdown time

Return type:

Self

stop()

Stop the countdown

Return type:

Self

reset()

Start the countdown again

Return type:

Self

cancel()

Cancel the job

Return type:

Self

property id: Hashable

Get the job’s id

property last_run_datetime: datetime | None

Get the last run time as a naive datetime object (without timezone set) or None if yet run

property next_run_datetime: datetime | None

Get the next run time as a naive datetime object (without timezone set) or None if not scheduled

property status: JobStatusEnum

Get the status of the job

to_item(item)

Sends the next execution (date)time to an item. Sends None if the job is not scheduled. Every time the scheduler updates to a new (date)time the item will also receive the updated time.

Parameters:

item (str | BaseValueItem | None) – item name or item, None to disable

Return type:

None

class DateTimeJobControl()
pause()

Stop executing this job

Return type:

Self

resume()

Resume executing this job

Return type:

Self

cancel()

Cancel the job

Return type:

Self

property id: Hashable

Get the job’s id

property last_run_datetime: datetime | None

Get the last run time as a naive datetime object (without timezone set) or None if yet run

property next_run_datetime: datetime | None

Get the next run time as a naive datetime object (without timezone set) or None if not scheduled

property status: JobStatusEnum

Get the status of the job

to_item(item)

Sends the next execution (date)time to an item. Sends None if the job is not scheduled. Every time the scheduler updates to a new (date)time the item will also receive the updated time.

Parameters:

item (str | BaseValueItem | None) – item name or item, None to disable

Return type:

None

Other tools and scripts

HABApp provides convenience functions to run other tools and scripts. The working directory for the new process is by default the folder of the HABApp configuration file.

Running tools

External tools can be run with the execute_subprocess() function. Once the process has finished the callback will be called with the captured output of the process. Example:

import HABApp

class MyExecutionRule(HABApp.Rule):

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

        self.execute_subprocess( self.func_when_finished, 'path_to_program', 'arg1_for_program')

    def func_when_finished(self, process_output: str):
        print(process_output)

MyExecutionRule()

Running python scripts or modules

Python scripts can be run with the execute_python() function. The working directory for a script is by default the folder of the script. Once the script or module has finished the callback will be called with the captured output of the module/script. Example:

import HABApp

class MyExecutionRule(HABApp.Rule):

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

        self.execute_python( self.func_when_finished, '/path/to/python/script.py', 'arg1_for_script')

    def func_when_finished(self, module_output: str):
        print(module_output)

MyExecutionRule()

FinishedProcessInfo

It’s possible to get the raw process output instead of just the captured string. See execute_subprocess() or execute_python() on how to enable it.

class FinishedProcessInfo(returncode, stdout, stderr)

Information about the finished process.

Variables:

  • returncode (int) – Return code of the process

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

  • stderr (Optional[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

Return type:

None

on_rule_removed()

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

Return type:

None

post_event(name, event)

Post an event to the event bus

Parameters:

  • name (BaseItem | str) – name or item to post event to

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

Return type:

None

Returns:

listen_event(name, callback, event_filter=None)

Register an event listener

Parameters:

  • name (BaseItem | str) – item or name to listen to

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

  • event_filter (EventFilterBase | None) – 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:

EventBusListener

execute_subprocess(callback, program, *args, additional_python_path=None, capture_output=True, raw_info=False, **kwargs)

Run another program

Parameters:

  • callback – Function that will be called when the process has finished. First parameter takes a str when raw_info is False (default) else an instance of FinishedProcessInfo

  • program (str | Path) – python module (path to file) or python package

  • args (str | Path) – arguments passed to the module or to package

  • raw_info (bool) –

    False: Return only the textual process output. In case of failure (return code != 0) a log entry and an error event will be created. This is the default and should be fine for almost all use cases.

    True: The callback will always be called with an instance of FinishedProcessInfo.

  • capture_output (bool) – Capture program output, set to False to only capture the return code

  • additional_python_path (Iterable[str | Path] | None) – additional folders which will be added to the env variable PYTHONPATH

  • kwargs – Additional kwargs that will be passed to asyncio.create_subprocess_exec

Returns:

execute_python(callback, module_or_package, *args, additional_python_path=None, capture_output=True, raw_info=False, **kwargs)

Run a python module or package as a new process. The python environment that is used to run HABApp will be to run the module or package.

Parameters:

  • callback – Function that will be called when the process has finished. First parameter takes a str when raw_info is False (default) else an instance of FinishedProcessInfo

  • module_or_package (str | Path) – python module (path to file) or python package (just the name)

  • args (str | Path) – arguments passed to the module or to package

  • raw_info (bool) –

    False: Return only the textual process output. In case of failure (return code != 0) a log entry and an error event will be created. This is the default and should be fine for almost all use cases.

    True: The callback will always be called with an instance of FinishedProcessInfo.

  • capture_output (bool) – Capture program output, set to False to only capture the return code

  • additional_python_path (Iterable[str | Path] | None) – additional folders which will be added to the env variable PYTHONPATH

  • kwargs – Additional kwargs that will be passed to asyncio.create_subprocess_exec

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 (tuple[type[TypeVar(ITEM_TYPE, bound= BaseItem)], ...] | type[TypeVar(ITEM_TYPE, bound= BaseItem)] | None) – item has to be an instance of this class

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

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

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

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

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

Return type:

list[TypeVar(ITEM_TYPE, bound= BaseItem)] | list[BaseItem]

Returns:

Items that match all the passed criteria