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!
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
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 parameter
(typically ValueUpdateEvent
or ValueChangeEvent
).
An overview over the events can be found on the HABApp event section, the openhab event section and the the mqtt event section
from HABApp import Rule
from HABApp.core.events import ValueChangeEvent, ValueUpdateEvent
from HABApp.core.items import Item
class MyRule(Rule):
def __init__(self):
super().__init__()
self.listen_event('MyOpenhabItem', self.on_change, ValueChangeEvent) # will trigger only on ValueChangeEvent
self.listen_event('My/MQTT/Topic', self.on_update, ValueUpdateEvent) # will 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, ValueUpdateEvent)
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 an instance of EventFilter as event type.
There are convenience Filters (e.g. ValueUpdateEventFilter
and
ValueChangeEventFilter
) for the most used event types that provide type hints.
-
class
HABApp.core.events.
EventFilter
(event_type, **kwargs)¶
-
class
HABApp.core.events.
ValueUpdateEventFilter
(value)¶
-
class
HABApp.core.events.
ValueChangeEventFilter
(value=<Missing>, old_value=<Missing>)¶
from HABApp import Rule
from HABApp.core.events import EventFilter, ValueUpdateEventFilter, ValueUpdateEvent
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'))
def on_val_my_value(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 run_in()
.
Function |
Description |
---|---|
Run the callback as soon as possible (typically in the next second). |
|
Run the callback in x seconds. |
|
Run a function at a specified date_time |
|
Run a function periodically |
|
Run a function every minute |
|
Run a function every hour |
|
Run a function every day |
|
Run a function at a specific time every day |
|
Run a function at a specific time on workdays |
|
Run a function at a specific time on weekends |
|
Run a function at a specific time on specific days of the week |
|
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.
None
will disable boundary.- Parameters
time_obj (
Optional
[time
]) – time obj, scheduler will not run earlier- Return type
-
latest
(time_obj)¶ Set earliest boundary as time of day.
None
will disable boundary.- Parameters
time_obj (
Optional
[time
]) – time obj, scheduler will not run later- Return type
-
offset
(timedelta_obj)¶ Set a constant offset to the calculation of the next run.
None
will disable the offset.- Parameters
timedelta_obj (
Optional
[timedelta
]) – constant offset- Return type
-
jitter
(secs)¶ Add a random jitter per call in the intervall [(-1) * secs … secs] to the next run.
None
will disable jitter.- Parameters
secs (
Optional
[int
]) – jitter in secs- Return type
-
boundary_func
(func)¶ Add a function which will be called when the datetime changes. Use this to implement custom boundaries. Use
None
to disable the boundary function.- 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:
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
HABApp.
Rule
¶ - Variables
async_http – Async http connections
openhab – Openhab interaction
oh – short alias for
openhab
openhab
-
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 eventscallback (
Callable
[[Any
],Any
]) – callback that accepts one parameter which will contain the eventevent_type (
Union
[Type
[AllEvents
],EventFilter
,Any
]) – Event filter. This is typicallyValueUpdateEvent
orValueChangeEvent
which will also trigger on changes/update from openhab or mqtt. Additionally it can be an instance ofEventFilter
which additionally filters on the values of the event. There are also templates for the most common filters, e.g.ValueUpdateEventFilter
andValueChangeEventFilter
- 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.timeweekdays –
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.timecallback – 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.timecallback – 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.timecallback – 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 acancel
function. When the rule gets unloaded the cancel function will be called (if the obj was not already garbage collected)- Parameters
obj –