Additional rule examples¶
Using the scheduler¶
from datetime import time, timedelta, datetime
from HABApp import Rule
class MyRule(Rule):
def __init__(self):
super().__init__()
self.run.on_day_of_week(time=time(14, 34, 20), weekdays=['Mo'], callback=self.run_mondays)
self.run.every(timedelta(seconds=5), 3, self.run_every_3s, 'arg 1', asdf='kwarg 1')
self.run.on_workdays(time(15, 00), self.run_workdays)
self.run.on_weekends(time(15, 00), self.run_weekends)
def run_every_3s(self, arg, asdf = None):
print(f'run_ever_3s: {datetime.now().replace(microsecond=0)} : {arg}, {asdf}')
def run_mondays(self):
print('Today is monday!')
def run_workdays(self):
print('Today is a workday!')
def run_weekends(self):
print('Finally weekend!')
MyRule()
Mirror openHAB events to a MQTT Broker¶
import HABApp
from HABApp.openhab.events import ItemStateEvent
from HABApp.openhab.items import OpenhabItem
class ExampleOpenhabToMQTTRule(HABApp.Rule):
"""This Rule mirrors all updates from OpenHAB to MQTT"""
def __init__(self):
super().__init__()
for item in self.get_items(OpenhabItem):
item.listen_event(self.process_update, ItemStateEvent)
def process_update(self, event):
assert isinstance(event, ItemStateEvent)
print(f'/openhab/{event.name} <- {event.value}')
self.mqtt.publish(f'/openhab/{event.name}', str(event.value))
ExampleOpenhabToMQTTRule()
Trigger an event when an item is constant¶
Get an even when the item is constant for 5 and for 10 seconds.
import HABApp
from HABApp.core.items import Item
from HABApp.core.events import ItemNoChangeEvent
class MyRule(HABApp.Rule):
def __init__(self):
super().__init__()
my_item = Item.get_item('test_watch')
# Create an event when the item doesn't change for 5 secs and
# create a watcher for ItemNoChangeEvent with 5s const time
my_item.watch_change(5).listen_event(self.item_constant_5s)
# Just create an event when the item doesn't change for 10 secs
my_item.watch_change(10)
# Listen to all ItemNoChangeEvents for the item
my_item.listen_event(self.item_constant, ItemNoChangeEvent)
# Set the item to a value to generate the ItemNoChangeEvent events
my_item.set_value('my_value')
def item_constant_5s(self, event):
print(f'Item 5s const: {event}')
def item_constant(self, event):
print(f'Item const: {event}')
MyRule()
Item 5s const: <ItemNoChangeEvent name: test_watch, seconds: 5>
Item const: <ItemNoChangeEvent name: test_watch, seconds: 5>
Item const: <ItemNoChangeEvent name: test_watch, seconds: 10>
Turn something off after movement¶
Turn a device off 30 seconds after one of the movement sensors in a room signals movement.
import HABApp
from HABApp.core.items import Item
from HABApp.core.events import ValueUpdateEvent
class MyCountdownRule(HABApp.Rule):
def __init__(self):
super().__init__()
self.countdown = self.run.countdown(30, self.switch_off)
self.device = Item.get_item('my_device')
self.movement1 = Item.get_item('movement_sensor1')
self.movement1.listen_event(self.movement, ValueUpdateEvent)
self.movement2 = Item.get_item('movement_sensor2')
self.movement2.listen_event(self.movement, ValueUpdateEvent)
def movement(self, event: ValueUpdateEvent):
if self.device != 'ON':
self.device.post_value('ON')
self.countdown.reset()
def switch_off(self):
self.device.post_value('OFF')
MyCountdownRule()
Process Errors in Rules¶
This example shows how to create a rule with a function which will be called when any rule throws an error. The rule function then can push the error message to an openhab item or e.g. use Pushover to send the error message to the mobile device (see Avanced Usage for more information).
import HABApp
from HABApp.core.events.habapp_events import HABAppException
class NotifyOnError(HABApp.Rule):
def __init__(self):
super().__init__()
# Listen to all errors
self.listen_event('HABApp.Errors', self.on_error, HABAppException)
def on_error(self, error_event: HABAppException):
msg = event.to_str() if isinstance(event, HABAppException) else event
print(msg)
NotifyOnError()
# this is a faulty example. Do not create this part!
class FaultyRule(HABApp.Rule):
def __init__(self):
super().__init__()
self.run.soon(self.faulty_function)
def faulty_function(self):
1 / 0
FaultyRule()
Error in FaultyRule.faulty_function: division by zero
File "/home/docs/checkouts/readthedocs.org/user_builds/habapp/checkouts/v0.31.2/src/HABApp/core/wrappedfunction.py", line 93, in __run
self._func(*args, **kwargs)
File "<string>", line 30, in faulty_function
ZeroDivisionError: division by zero
Error in NotifyOnError.on_error: name 'event' is not defined
File "/home/docs/checkouts/readthedocs.org/user_builds/habapp/checkouts/v0.31.2/src/HABApp/core/wrappedfunction.py", line 93, in __run
self._func(*args, **kwargs)
File "<string>", line 30, in faulty_function
ZeroDivisionError: division by zero
While handling the above exception, another exception occurred:
File "/home/docs/checkouts/readthedocs.org/user_builds/habapp/checkouts/v0.31.2/src/HABApp/core/wrappedfunction.py", line 93, in __run
self._func(*args, **kwargs)
File "<string>", line 17, in on_error
NameError: name 'event' is not defined