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=<MISSING>)
ValueChangeEventFilter
- class ValueChangeEventFilter(value=<MISSING>, old_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.at
.
Another very useful function is self.run.countdown
as it can simplify many rules!
Function |
Description |
---|---|
Run the callback as soon as possible (typically in the next second). |
|
Run the callback in x seconds or at a specified time. |
|
Run a function after a time has run down |
|
Run a function periodically |
|
Run a function every minute |
|
Run a function every hour |
|
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 on dawn |
|
Run a function on sunrise |
|
Run a function on sunset |
|
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
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- Returns:
Created job
- countdown(expire_time, callback, *args, **kwargs)
Run a job a specific time after calling
reset()
of the job. Another subsequent call toreset()
will start the countdown again.- Parameters:
expire_time (
Union
[timedelta
,float
,int
]) – countdown in seconds or a timedelta objcallback (
Callable
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- 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 timeinterval (
Union
[int
,float
,timedelta
]) – Interval how the job is repeatedcallback (
Callable
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- 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 runweekdays (
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
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- 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 runcallback (
Callable
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- on_sunrise(callback, *args, **kwargs)
Create a job that will run on sunrise, requires a location to be set
- Parameters:
callback (
Callable
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- 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
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- 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
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- 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
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- Returns:
Created job
- soon(callback, *args, **kwargs)
Run the callback as soon as possible.
- Parameters:
callback (
Callable
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- every_minute(callback, *args, **kwargs)
Picks a random second and runs the callback every minute
- Parameters:
callback (
Callable
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
- 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 runcallback – 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 runcallback – 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
- every_hour(callback, *args, **kwargs)
Picks a random minute and second and run the callback every hour
- Parameters:
callback (
Callable
[[ParamSpec
(HINT_CB_P
)],Any
]) – Function which will be calledargs (
ParamSpecArgs
) – Positional arguments that will be passed to the functionkwargs (
ParamSpecKwargs
) – Keyword arguments that will be passed to the function
- Return type:
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:
async_http – Async http connections
mqtt – MQTT interaction
openhab – openhab interaction
oh – short alias for openhab
- 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 toevent (
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 tocallback (
Callable
[[Any
],Any
]) – callback that accepts one parameter which will contain the eventevent_filter (
Optional
[TypeVar
(HINT_EVENT_FILTER_OBJ
, bound=EventFilterBase
)]) – Event filter. This is typicallyValueUpdateEventFilter
orValueChangeEventFilter
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. It is also possible to group filters logically with, e.g.AndFilterGroup
andOrFilterGroup
- Return type:
TypeVar
(HINT_EVENT_BUS_LISTENER
, bound=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
whenraw_info
isFalse
(default) else an instance ofFinishedProcessInfo
program (
Union
[str
,Path
]) – python module (path to file) or python packageargs (
Union
[str
,Path
]) – arguments passed to the module or to packageraw_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 ofFinishedProcessInfo
.capture_output (
bool
) – Capture program output, set toFalse
to only capture the return codeadditional_python_path (
Optional
[Iterable
[Union
[str
,Path
]]]) – additional folders which will be added to the env variablePYTHONPATH
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
whenraw_info
isFalse
(default) else an instance ofFinishedProcessInfo
module_or_package (
Union
[str
,Path
]) – python module (path to file) or python package (just the name)args (
Union
[str
,Path
]) – arguments passed to the module or to packageraw_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 ofFinishedProcessInfo
.capture_output (
bool
) – Capture program output, set toFalse
to only capture the return codeadditional_python_path (
Optional
[Iterable
[Union
[str
,Path
]]]) – additional folders which will be added to the env variablePYTHONPATH
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 (
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 classname (
Union
[str
,Pattern
[str
],None
]) – str (will be compiled) or regex that is used to search the Nametags (
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