util - helpers and utilities
The util package contains useful classes which make rule creation easier.
Functions
min
This function is very useful together with the all possible functions of ValueMode
for the
MultiModeItem
.
For example it can be used to automatically disable or calculate the new value of the ValueMode
It behaves like the standard python function except that it will ignore None
values which are sometimes set as the item state.
from HABApp.util.functions import min
print(min(1, 2, None))
- min(*args, default=None)
Behaves like the built-in min function but ignores any
None
values. e.g.min([1, None, 2]) == 1
. If the iterable is emptydefault
will be returned.- Parameters:
args – Single iterable or 1..n arguments
default – Value that will be returned if the iterable is empty
- Returns:
min value
max
This function is very useful together with the all possible functions of ValueMode
for the
MultiModeItem
.
For example it can be used to automatically disable or calculate the new value of the ValueMode
It behaves like the standard python function except that it will ignore None
values which are sometimes set as the item state.
from HABApp.util.functions import max
print(max(1, 2, None))
- max(*args, default=None)
Behaves like the built-in max function but ignores any
None
values. e.g.max([1, None, 2]) == 2
. If the iterable is emptydefault
will be returned.- Parameters:
args – Single iterable or 1..n arguments
default – Value that will be returned if the iterable is empty
- Returns:
max value
rgb_to_hsb
Converts a rgb value to hsb color space
from HABApp.util.functions import rgb_to_hsb
print(rgb_to_hsb(224, 201, 219))
- rgb_to_hsb(r, g, b, max_rgb_value=255, ndigits=2)
Convert from rgb to hsb/hsv
- Parameters:
- Return type:
- Returns:
Values for hue, saturation and brightness / value
hsb_to_rgb
Converts a hsb value to the rgb color space
from HABApp.util.functions import hsb_to_rgb
print(hsb_to_rgb(150, 40, 100))
- hsb_to_rgb(h, s, b, max_rgb_value=255)
Convert from rgb to hsv/hsb
Rate limiter
A simple rate limiter implementation which can be used in rules. The limiter is not rule bound so the same limiter can be used in multiples files. It also works as expected across rule reloads.
Defining limits
Limits can either be explicitly added or through a textual description. If the limit does already exist it will not be added again. It’s possible to explicitly create the limits or through some small textual description with the following syntax:
[count] [per|in|/] [count (optional)] [s|sec|second|m|min|minute|hour|h|day|month|year] [s (optional)]
Whitespaces are ignored and can be added as desired
Examples:
5 per minute
20 in 15 mins
300 / hour
Fixed window elastic expiry algorithm
This algorithm implements a fixed window with elastic expiry. That means if the limit is hit the interval time will be increased by the expiry time.
For example 3 per minute
:
First hit comes
00:00:00
. Two more hits at00:00:59
. All three pass, intervall goes from00:00:00
-00:01:00
. Another hit comes at00:01:01
an passes. The intervall now goes from00:01:01
-00:02:01
.First hit comes
00:00:00
. Two more hits at00:00:30
. All three pass. Another hit comes at00:00:45
, which gets rejected and the intervall now goes from00:00:00
-00:01:45
. A rejected hit makes the interval time longer by expiry time. If another hit comes at00:01:30
it will also get rejected and the intervall now goes from00:00:00
-00:02:30
.
Leaky bucket algorithm
The leaky bucket algorithm is based on the analogy of a bucket that leaks at a constant rate. As long as the bucket is not full the hits will pass. If the bucket overflows the hits will get rejected. Since the bucket leaks at a constant rate it will gradually get empty again thus allowing hits to pass again.
Example
from HABApp.util import RateLimiter
# Create or get existing, name is case insensitive
limiter = RateLimiter('MyRateLimiterName')
# define limits, duplicate limits of the same algorithm will only be added once
# These lines all define the same limit so it'll result in only one limiter added
limiter.add_limit(5, 60) # add limits explicitly
limiter.parse_limits('5 per minute').parse_limits('5 in 60s', '5/60seconds') # add limits through text
# add additional limit with leaky bucket algorithm
limiter.add_limit(10, 100, algorithm='leaky_bucket')
# add additional limit with fixed window elastic expiry algorithm
limiter.add_limit(10, 100, algorithm='fixed_window_elastic_expiry')
# Test the limit without increasing the hits
for _ in range(100):
assert limiter.test_allow()
# the limiter will allow 5 calls ...
for _ in range(5):
assert limiter.allow()
# and reject the 6th
assert not limiter.allow()
# It's possible to get statistics about the limiter and the corresponding windows
print(limiter.info())
# There is a counter that keeps track of the total skips that can be reset
print('Counter:')
print(limiter.total_skips)
limiter.reset() # Can be reset
print(limiter.total_skips)
LimiterInfo(skips=1, total_skips=1, limits=[LeakyBucketLimitInfo(hits=5, skips=1, limit=5, time_remaining=11.999655897999219), LeakyBucketLimitInfo(hits=5, skips=0, limit=10, time_remaining=9.999695871999393), FixedWindowElasticExpiryLimitInfo(hits=5, skips=0, limit=10, time_remaining=99.99997635700038)])
Counter:
1
0
Recommendation
Limiting external requests to an external API works well with the leaky bucket algorithm (maybe with some initial hits). For limiting notifications the best results can be achieved by combining both algorithms. Fixed window elastic expiry will notify but block until an issue is resolved, that’s why it’s more suited for small intervals. Leaky bucket will allow hits even while the issue persists, that’s why it’s more suited for larger intervals.
from HABApp.util import RateLimiter
limiter = RateLimiter('MyNotifications')
limiter.parse_limits('5 in 1 minute', algorithm='fixed_window_elastic_expiry')
limiter.parse_limits("20 in 1 hour", algorithm='leaky_bucket')
Documentation
- RateLimiter(name)
Create a new rate limiter or return an already existing one with a given name.
- class Limiter(name)
-
- add_limit(allowed, interval, *, initial_hits=0, algorithm='leaky_bucket')
Add a new rate limit
- Parameters:
- Return type:
- parse_limits(*text, initial_hits=0, algorithm='leaky_bucket')
Add one or more limits in textual form, e.g.
5 in 60s
,10 per hour
or10/15 mins
. If the limit does already exist it will not be added again.
- test_allow()
Test the limit(s) without hitting it. Calling this will not increase the hit counter.
- Return type:
- Returns:
True
if allowed,False
if forbidden
- info()
Get some info about the limiter and the defined windows
- Return type:
- class LimiterInfo(skips, total_skips, limits)
-
-
limits:
List
[Union
[FixedWindowElasticExpiryLimitInfo
,LeakyBucketLimitInfo
]] Info for every limit
-
limits:
- class FixedWindowElasticExpiryLimitInfo(hits, skips, limit, time_remaining)
Statistics
Example
s = Statistics(max_samples=4)
for i in range(1,4):
s.add_value(i)
print(s)
<Statistics sum: 1.0, min: 1.00, max: 1.00, mean: 1.00, median: 1.00>
<Statistics sum: 3.0, min: 1.00, max: 2.00, mean: 1.50, median: 1.50>
<Statistics sum: 6.0, min: 1.00, max: 3.00, mean: 2.00, median: 2.00>
Documentation
- class Statistics(max_age=None, max_samples=None)
Calculate mathematical statistics of numerical values.
- Variables:
sum – sum of all values
min – minimum of all values
max – maximum of all values
mean – mean of all values
median – median of all values
last_value – last added value
last_change – timestamp the last time a value was added
- update()
update values without adding a new value
- add_value(value)
Add a new value and recalculate statistical values
- Parameters:
value – new value
Fade
Fade is a helper class which allows to easily fade a value up or down.
Example
This example shows how to fade a Dimmer from 0 to 100 in 30 secs
from HABApp import Rule
from HABApp.openhab.items import DimmerItem
from HABApp.util import Fade
class FadeExample(Rule):
def __init__(self):
super().__init__()
self.dimmer = DimmerItem.get_item('Dimmer1')
self.fade = Fade(callback=self.fade_value) # self.dimmer.percent would also be a good callback in this example
# Setup the fade and schedule its execution
# Fade from 0 to 100 in 30s
self.fade.setup(0, 100, 30).schedule_fade()
def fade_value(self, value):
self.dimmer.percent(value)
FadeExample()
This example shows how to fade three values together (e.g. for an RGB strip)
from HABApp import Rule
from HABApp.openhab.items import DimmerItem
from HABApp.util import Fade
class Fade3Example(Rule):
def __init__(self):
super().__init__()
self.fade1 = Fade(callback=self.fade_value)
self.fade2 = Fade()
self.fade3 = Fade()
# Setup the fades and schedule the execution of one fade where the value gets updated every sec
self.fade3.setup(0, 100, 30)
self.fade2.setup(0, 50, 30)
self.fade1.setup(0, 25, 30, min_step_duration=1).schedule_fade()
def fade_value(self, value):
value1 = value
value2 = self.fade2.get_value()
value3 = self.fade3.get_value()
Fade3Example()
Documentation
- class Fade(callback=None, min_value=0, max_value=100)
Helper to easily fade values up/down
- Variables:
min_value – minimum valid value for the fade value
max_value – maximum valid value for the fade value
callback – Function with one argument that will be automatically called with the new values when the scheduled fade runs
- setup(start_value, stop_value, duration, min_step_duration=0.2, now=None)
Calculates everything that is needed to fade a value
- get_value(now=None)
Returns the current value. If the fade is finished it will always return the stop value.
- schedule_fade()
Automatically run the fade with the Scheduler. The callback can be used to set the current fade value e.g. on an item. Calling this on a running fade will restart the fade
- Return type:
- stop_fade()
Stop the scheduled fade. This can be called multiple times without error
EventListenerGroup
EventListenerGroup is a helper class which allows to subscribe to multiple items at once. All subscriptions can be canceled together, too. This is useful if e.g. something has to be done once after a sensor reports a value.
Example
This is a rule which will turn on the lights once (!) in a room on the first movement in the morning. The lights will only turn on after 4 and before 8 and two movement sensors are used to pick up movement.
from datetime import time
from HABApp import Rule
from HABApp.core.events import ValueChangeEventFilter
from HABApp.openhab.items import SwitchItem, NumberItem
from HABApp.util import EventListenerGroup
class EventListenerGroupExample(Rule):
def __init__(self):
super().__init__()
self.lights = SwitchItem.get_item('RoomLights')
self.sensor_move_1 = NumberItem.get_item('MovementSensor1')
self.sensor_move_2 = NumberItem.get_item('MovementSensor2')
# use a list of items which will be subscribed with the same callback and event
self.listeners = EventListenerGroup().add_listener(
[self.sensor_move_1, self.sensor_move_2], self.sensor_changed, ValueChangeEventFilter())
self.run.on_every_day(time(4), self.listen_sensors)
self.run.on_every_day(time(8), self.sensors_cancel)
def listen_sensors(self):
self.listeners.listen()
def sensors_cancel(self):
self.listeners.cancel()
def sensor_changed(self, event):
self.listeners.cancel()
self.lights.on()
EventListenerGroupExample()
Documentation
- class EventListenerGroup
Helper to create/cancel multiple event listeners simultaneously
- listen()
Create all event listeners. If the event listeners are already active this will do nothing.
- cancel()
Cancel the active event listeners. If the event listeners are not active this will do nothing.
- activate_listener(name)
Resume a previously deactivated listener creator in the group.
- Parameters:
name (
str
) – item name or alias of the listener- Returns:
True if it was activated, False if it was already active
- deactivate_listener(name, cancel_if_active=True)
Exempt the listener creator from further listener/cancel calls
- Parameters:
name (
str
) – item name or alias of the listenercancel_if_active – Cancel the listener if it is active
- Returns:
True if it was deactivated, False if it was already deactivated
- add_listener(item, callback, event_filter, alias=None)
Add an event listener to the group
- Parameters:
item (
Union
[TypeVar
(HINT_ITEM_OBJ
, bound=BaseItem
),Iterable
[TypeVar
(HINT_ITEM_OBJ
, bound=BaseItem
)]]) – Single or multiple itemsevent_filter (
TypeVar
(HINT_EVENT_FILTER_OBJ
, bound=EventFilterBase
)) – Event filter for the item(s)alias (
Optional
[str
]) – Alias if an item with the same name does already exist (e.g. if different callbacks shall be created for the same item)
- Return type:
- Returns:
self
- add_no_update_watcher(item, callback, seconds, alias=None)
- Add an no update watcher to the group. On
listen
this will create a no update watcher and the corresponding event listener that will trigger the callback
- Parameters:
item (
Union
[TypeVar
(HINT_ITEM_OBJ
, bound=BaseItem
),Iterable
[TypeVar
(HINT_ITEM_OBJ
, bound=BaseItem
)]]) – Single or multiple itemsseconds (
Union
[int
,float
,timedelta
]) – No update time for the no update watcheralias (
Optional
[str
]) – Alias if an item with the same name does already exist (e.g. if different callbacks shall be created for the same item)
- Return type:
- Returns:
self
- Add an no update watcher to the group. On
- add_no_change_watcher(item, callback, seconds, alias=None)
- Add a no change watcher to the group. On
listen
this will create a no change watcher and the corresponding event listener that will trigger the callback
- Parameters:
item (
Union
[TypeVar
(HINT_ITEM_OBJ
, bound=BaseItem
),Iterable
[TypeVar
(HINT_ITEM_OBJ
, bound=BaseItem
)]]) – Single or multiple itemsseconds (
Union
[int
,float
,timedelta
]) – No update time for the no change watcheralias (
Optional
[str
]) – Alias if an item with the same name does already exist (e.g. if different callbacks shall be created for the same item)
- Return type:
- Returns:
self
- Add a no change watcher to the group. On
MultiModeItem
Prioritizer item which automatically switches between values with different priorities. Very useful when different states or modes overlap, e.g. automatic and manual mode. etc.
Basic Example
import HABApp
from HABApp.core.events import ValueUpdateEventFilter
from HABApp.util.multimode import MultiModeItem, ValueMode
class MyMultiModeItemTestRule(HABApp.Rule):
def __init__(self):
super().__init__()
# create a new MultiModeItem
item = MultiModeItem.get_create_item('MultiModeTestItem')
item.listen_event(self.item_update, ValueUpdateEventFilter())
# create two different modes which we will use and add them to the item
auto = ValueMode('Automatic', initial_value=5)
manu = ValueMode('Manual', initial_value=0)
# Add the auto mode with priority 0 and the manual mode with priority 10
item.add_mode(0, auto).add_mode(10, manu)
# This shows how to enable/disable a mode and how to get a mode from the item
print('disable/enable the higher priority mode')
item.get_mode('manual').set_enabled(False) # disable mode
item.get_mode('manual').set_value(11) # setting a value will enable it again
# This shows that changes of the lower priority is only shown when
# the mode with the higher priority gets disabled
print('')
print('Set value of lower priority')
auto.set_value(55)
print('Disable higher priority')
manu.set_enabled(False)
def item_update(self, event):
print(f'State: {event.value}')
MyMultiModeItemTestRule()
disable/enable the higher priority mode
State: 5
State: 11
Set value of lower priority
State: 11
Disable higher priority
State: 55
Advanced Example
import logging
import HABApp
from HABApp.core.events import ValueUpdateEventFilter
from HABApp.util.multimode import MultiModeItem, ValueMode
class MyMultiModeItemTestRule(HABApp.Rule):
def __init__(self):
super().__init__()
# create a new MultiModeItem
item = MultiModeItem.get_create_item('MultiModeTestItem')
item.listen_event(self.item_update, ValueUpdateEventFilter())
# helper to print the heading so we have a nice output
def print_heading(_heading):
print('')
print('-' * 80)
print(_heading)
print('-' * 80)
for p, m in item.all_modes():
print(f'Prio {p:2d}: {m}')
print('')
log = logging.getLogger('AdvancedMultiMode')
# create modes and add them
auto = ValueMode('Automatic', initial_value=5, logger=log)
manu = ValueMode('Manual', initial_value=10, logger=log)
item.add_mode(0, auto).add_mode(10, manu)
# it is possible to automatically disable a mode
# this will disable the manual mode if the automatic mode
# sets a value greater equal manual mode
print_heading('Automatically disable mode')
# A custom function can also disable the mode:
manu.auto_disable_func = lambda low, own: low >= own
auto.set_value(11) # <-- manual now gets disabled because
auto.set_value(4) # the lower priority value is >= itself
# It is possible to use functions to calculate the new value for a mode.
# E.g. shutter control and the manual mode moves the shades. If it's dark the automatic
# mode closes the shutter again. This could be achieved by automatically disabling the
# manual mode or if the state should be remembered then the max function should be used
# create a move and use the max function for output calculation
manu = ValueMode('Manual', initial_value=5, logger=log, calc_value_func=max)
item.add_mode(10, manu) # overwrite the earlier added mode
print_heading('Use of functions')
auto.set_value(7) # manu uses max, so the value from auto is used
auto.set_value(3)
def item_update(self, event):
print(f'Item value: {event.value}')
MyMultiModeItemTestRule()
--------------------------------------------------------------------------------
Automatically disable mode
--------------------------------------------------------------------------------
Prio 0: <ValueMode Automatic enabled: True, value: 5>
Prio 10: <ValueMode Manual enabled: True, value: 10>
[AdvancedMultiMode] INFO | [x] Automatic: 11
[AdvancedMultiMode] INFO | [ ] Manual (function)
Item value: 11
[AdvancedMultiMode] INFO | [x] Automatic: 4
Item value: 4
--------------------------------------------------------------------------------
Use of functions
--------------------------------------------------------------------------------
Prio 0: <ValueMode Automatic enabled: True, value: 4>
Prio 10: <ValueMode Manual enabled: True, value: 5>
[AdvancedMultiMode] INFO | [x] Automatic: 7
Item value: 7
[AdvancedMultiMode] INFO | [x] Automatic: 3
Item value: 5
Example SwitchItemValueMode
The SwitchItemMode is same as ValueMode but enabled/disabled of the mode is controlled by a openHAB
SwitchItem
. This is very useful if the mode shall be deactivated from the openHAB sitemaps.
import HABApp
from HABApp.openhab.items import SwitchItem
from HABApp.util.multimode import MultiModeItem, SwitchItemValueMode, ValueMode
class MyMultiModeItemTestRule(HABApp.Rule):
def __init__(self):
super().__init__()
# create a new MultiModeItem
item = MultiModeItem.get_create_item('MultiModeTestItem')
# this switch allows to enable/disable the mode
switch = SwitchItem.get_item('Automatic_Enabled')
print(f'Switch is {switch}')
# this is how the switch gets linked to the mode
# if the switch is on, the mode is on, too
mode = SwitchItemValueMode('Automatic', switch)
print(mode)
# Use invert_switch if the desired behaviour is
# if the switch is off, the mode is on
mode = SwitchItemValueMode('AutomaticOff', switch, invert_switch=True)
print(mode)
# This shows how the SwitchItemValueMode can be used to disable any logic except for the manual mode.
# Now everything can be enabled/disabled from the openHAB sitemap
item.add_mode(100, mode)
item.add_mode(101, ValueMode('Manual'))
MyMultiModeItemTestRule()
Switch is ON
<SwitchItemValueMode Automatic enabled: True, value: None>
<SwitchItemValueMode AutomaticOff enabled: False, value: None>
Documentation
MultiModeItem
- class MultiModeItem()
Prioritizer
Item
- classmethod get_create_item(name, initial_value=None, default_value=<MISSING>)
Creates a new item in HABApp and returns it or returns the already existing one with the given name
- Parameters:
name (
str
) – item nameinitial_value – state the item will have if it gets created
default_value – Default value that will be sent if no mode is active
- Return type:
- Returns:
The created or existing item
- remove_mode(name)
Remove mode if it exists
- add_mode(priority, mode)
Add a new mode to the item, if it already exists it will be overwritten
- Parameters:
- Return type:
- all_modes()
Returns a sorted list containing tuples with the priority and the mode
- get_mode(name)
Returns a created mode
ValueMode
- class ValueMode(name, initial_value=None, enabled=None, enable_on_value=True, logger=None, auto_disable_after=None, auto_disable_func=None, calc_value_func=None)
- Variables:
last_update (datetime) – Timestamp of the last update/enable of this value
auto_disable_after (Optional[timedelta]) – Automatically disable this mode after a given timedelta on the next recalculation
auto_disable_func (Optional[Callable[[Any, Any], bool]]) – Function which can be used to disable this mode. Any function that accepts two Arguments can be used. First arg is value with lower priority, second argument is own value. Return
True
to disable this mode.calc_value_func (Optional[Callable[[Any, Any], Any]]) – Function to calculate the new value (e.g.
min
ormax
). Any function that accepts two Arguments can be used. First arg is value with lower priority, second argument is own value.
- property value
Returns the current value
- set_value(value, only_on_change=False)
Set new value and recalculate overall value. If
enable_on_value
is set, setting a value will also enable the mode.- Parameters:
value – new value
only_on_change (
bool
) – will set/enable the mode only if value differs or the mode is disabled
- Returns:
False if the value was not set, True otherwise
- set_enabled(value, only_on_change=False)
Enable or disable this value and recalculate overall value
- cancel()
Remove the mode from the parent
MultiModeItem
and stop processing it
SwitchItemValueMode
- class SwitchItemValueMode(name, switch_item, invert_switch=False, initial_value=None, logger=None, auto_disable_after=None, auto_disable_func=None, calc_value_func=None)
SwitchItemMode, same as ValueMode but enabled/disabled of the mode is controlled by a OpenHAB
SwitchItem
- Variables:
last_update (datetime) – Timestamp of the last update/enable of this value
auto_disable_after (Optional[timedelta]) – Automatically disable this mode after a given timedelta on the next recalculation
auto_disable_func (Optional[Callable[[Any, Any], bool]]) – Function which can be used to disable this mode. Any function that accepts two Arguments can be used. First arg is value with lower priority, second argument is own value. Return
True
to disable this mode.calc_value_func (Optional[Callable[[Any, Any], Any]]) – Function to calculate the new value (e.g.
min
ormax
). Any function that accepts two Arguments can be used. First arg is value with lower priority, second argument is own value.
- cancel()
Remove the mode from the parent
MultiModeItem
and stop processing it
- set_value(value, only_on_change=False)
Set new value and recalculate overall value. If
enable_on_value
is set, setting a value will also enable the mode.- Parameters:
value – new value
only_on_change (
bool
) – will set/enable the mode only if value differs or the mode is disabled
- Returns:
False if the value was not set, True otherwise
- property value
Returns the current value