Following the discussion in this thread, I wrote a simple decorator function to debounce my Python rules (i.e. solving the problem of someone impatient ringing the doorbell ten times, and then the doorbell rule running ten times sequentially).
You can then simply add @debounce in front of a rule, and the decorator will enforce a two second gap between the rule ending and starting again (so when that idiot presses the doorbell ten times, the rule will only run once).
The function lives in my utils file (imported into each rules py) and looks like:
BouncingRuleTracker={}
def debounce(func):
def wrapper_debounce(event):
global BouncingRuleTracker
min_delay = 2
current_rule = str(func.__name__)
report = "Started rule " + current_rule + \
", triggered by " + str(event.itemName)
if current_rule in BouncingRuleTracker:
if BouncingRuleTracker[current_rule]["running"] == True:
# Note currently openhab seems to always run rules sequentially
# so this condition will never be satisfied if the rule is actually
# running - only if it failed with an error. Will need to revisit
# this if the sequential model ever changes
report += ". Did not previously exit cleanly (likely there was an error in the rule), but will permit it to run again."
else:
# we know rule previously finished - see how long ago that was
actualdelay = round(
(time.time() - BouncingRuleTracker[current_rule]["timetrack"]), 1)
report += ". Last finished " + \
str(actualdelay) + " seconds ago"
if actualdelay < min_delay:
report += ", so debouncing and not running rule"
log.info(report)
return
else:
report += ". This is the first time the rule has run."
BouncingRuleTracker[current_rule] = {}
BouncingRuleTracker[current_rule]["timetrack"] = time.time()
BouncingRuleTracker[current_rule]["running"] = True
log.info(report)
func(event)
ruletime = round(
(time.time() - BouncingRuleTracker[current_rule]["timetrack"]), 1)
log.info("Finishing rule " + current_rule + " triggered by " +
str(event.itemName) + ". Rule took " + str(ruletime) + " seconds to run.")
BouncingRuleTracker[current_rule]["timetrack"] = time.time()
BouncingRuleTracker[current_rule]["running"] = False
return wrapper_debounce
As a bonus, it gives you additional logging of when rules start and stop, and the time they took to run.
An example using this:
@rule("Main test function")
@when("Item TestSwitch received command ON")
@debounce
def main_test(event):
log.info("Test switch on")
sleep (5)
log.info("Test done")
You should then see that if TestSwitch goes ON multiple times in quick succession, then the rule will only actually run through once.
Lots of room for improvement by people with more python skill than me. Most obviously, would be nice to add an optional parameter for the wait delay (but I’m new to decorators and don’t know how to do that)
(v0.2 of wrapper - previously behaved badly if a run failed with an error, not letting the rule run again)