My semi-final version of this idea follows. I’ve veered a little bit from my original plans. Instead of creating something to help manage just flapping timers, I’ve created a class that helps manage lots of Timers when you need one or more Timer per Item. The important features are:
- each Item can have more than one Timer
- moves all of the book keeping logic to the new class
- minimal set of functions
- as encapsulated as possible.
"""
A class to centralize management of multiple Timers in cases where one needs to keep
track of one timer per Item
"""
from core.jsr223.scope import ir
from core.jsr223.scope import items
from core.actions import ScriptExecution
from org.joda.time import DateTime
from core.log import log_traceback
class timer_mgr(object):
"""
Keeps and manages a dictionary of Timers keyed on an Item name.
"""
def __init__(self):
self.timers = {}
@log_traceback
def __not_flapping(self, itemName):
"""
Called when the Timer expires. Call the function and delete the timer from the dict.
This function ensures that the dict get's cleaned up.
Args:
itemName: the name of the Item associated with the Timer
"""
try:
self.timers[itemName]['not_flapping']()
finally:
del self.timers[itemName]
@log_traceback
def timer_check(self, itemName, interval, function, flapping_function=None, reschedule=False):
"""
Call to check whether an Item is flapping. If a Timer exists, create a new timer to run the passed
in function.
If a Timer exists, reschedule it if reschedule is True. If a flapping_function was passed, run it.
Args:
itemName: The name of the Item we are monitoring for flapping
interval: How long the Item needs to remain the same state before calling function, in milliseconds
function: Function to call when the timer expired
flapping_function: Optional function to call if the Item is found to be flapping
reschedule: Optional flag that causes the Timer to be rescheduled when the Item is flapping
"""
# Timer exists, reschedule and call flapping_lambda if necessary
if itemName in self.timers:
# Reschedule the Timer if desired
if reschedule: self.timers[itemName]['timer'].reschedule(DateTime.now().plusMillis(interval))
# Cancel the Timer if not rescheduling
else:
self.timers[itemName]['timer'].cancel()
del self.timers[itemName]
# Call the flapping function
if flapping_function: flapping_function()
# No timer exists, create the Timer
else:
item = ir.getItem(itemName)
self.timers[itemName] = { 'orig_state': item.state,
'timer': ScriptExecution.createTimer(DateTime.now().plusMillis(interval), lambda: self.__not_flapping(itemName)),
'flapping': flapping_function,
'not_flapping': function }
def has_timer(self, itemName):
"""
Returns True if there is a Timer for this Item, False otherwise
"""
return True if itemName in self.timers else False
@log_traceback
def cancel(self, itemName):
"""
Cancels the Timer assocaited with this Timer if one exists
"""
if not self.has_timer(itemName): return
self.timers[itemName]['timer'].cancel()
del self.timers[itemName]
And a test script showing how to use it:
from core.log import logging, LOG_PREFIX
log = logging.getLogger("{}.TEST".format(LOG_PREFIX))
import personal.timer_mgr
reload(personal.timer_mgr)
from personal.timer_mgr import timer_mgr
timers = timer_mgr()
def flapping():
log.info("Test was found to be flapping")
def not_flapping():
log.info("Test was found not to be flapping")
from time import sleep
log.info("TEST: Flapping, no lambda, no reschedule: no output")
timers.timer_check("Test", 1000, not_flapping)
sleep(0.5)
timers.timer_check("Test", 1000, not_flapping)
sleep(0.5)
log.info("TEST: Flapping, with lambda, no reschedule: Test as found to be flapping")
timers.timer_check("Test", 1000, not_flapping, flapping_function=flapping)
sleep(0.5)
timers.timer_check("Test", 1000, not_flapping, flapping_function=flapping)
sleep(0.5)
log.info("TEST: Flapping, with lambda and reschedule: Test was found to be flapping, Test was found not to be flapping")
timers.timer_check("Test", 1000, not_flapping, flapping_function=flapping, reschedule=True)
sleep(0.5)
timers.timer_check("Test", 1000, not_flapping, flapping_function=flapping, reschedule=True)
sleep(1.1)
log.info("TEST: Not flapping: Test was found not to be flapping")
timers.timer_check("Test", 1000, not_flapping)
sleep(1.1)
log.info("TEST: hasTimer and cancel")
timers.timer_check("Test", 1000, not_flapping)
sleep(0.5)
if timers.has_timer("Test"): log.info("Timer exists")
log.info("Cancelling timer")
timers.cancel("Test")
sleep(0.6)
log.info("Cancelling non-existant timer")
if not timers.has_timer("Test"): log.info("No timer exists")
timers.cancel("Test")
log.info("Done")