Simple debouncing of Python rules in the new rule engine

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:


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."


                # 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"

            report += ". This is the first time the rule has run."
            BouncingRuleTracker[current_rule] = {}

        BouncingRuleTracker[current_rule]["timetrack"] = time.time()
        BouncingRuleTracker[current_rule]["running"] = True

        ruletime = round(
            (time.time() - BouncingRuleTracker[current_rule]["timetrack"]), 1)"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")
def main_test(event):"Test switch on")
   sleep (5)"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)


Nice! Running this through pylint would help with some cleanup. If you submit it as a PR, this would fit nicely into core.utils.

1 Like

That’s kind! Although sadly I have no idea how to use pylint or make a PR… I will look into it!

Adding to GH is easy through the web UI. Just open the file for editing, paste in your code, and save to new branch (be sure to sign the commit). If you’re not comfortable using GitHub, I can add it in. I can also document the steps too.

Thanks - I’ve added it but not sure I want to sign it.

This topic was automatically closed 41 days after the last reply. New replies are no longer allowed.