blynktimer.py

# Copyright (c) 2019 Anton Morozenko
"""
Polling timers for functions.
Registers timers and performs run once or periodical function execution after defined time intervals.
"""
# select.select call used as polling waiter where it is possible
# cause time.sleep sometimes may load CPU up to 100% with small polling wait interval
try:
 # cpython
 import time
 import select

 polling_wait = lambda x: select.select([], [], [], x)
 polling_wait(0.01)
except OSError:
 # windows case where select.select call fails
 polling_wait = lambda x: time.sleep(x)

except ImportError:
 # micropython
 import utime as time

 try:
 from uselect import select as s_select

 polling_wait = lambda x: s_select([], [], [], x)
 except ImportError:
 # case when micropython port does not support select.select
 polling_wait = lambda x: time.sleep(x)

WAIT_SEC = 0.05
MAX_TIMERS = 16
DEFAULT_INTERVAL = 10


class TimerError(Exception):
 pass


class Timer(object):
 timers = {}

 def __init__(self, no_timers_err=True):
 self.no_timers_err = no_timers_err

 def _get_func_name(self, obj):
 """retrieves a suitable name for a function"""
 if hasattr(obj, 'func'):
 # handles nested decorators
 return self._get_func_name(obj.func)
 # simply returns 'timer' if on port without function attrs
 return getattr(obj, '__name__', 'timer')

 def register(blynk, *args, **kwargs):
 # kwargs with defaults are used cause PEP 3102 no supported by Python2
 interval = kwargs.pop('interval', DEFAULT_INTERVAL)
 run_once = kwargs.pop('run_once', False)
 stopped = kwargs.pop('stopped', False)

 class Deco(object):
 def __init__(self, func):
 self.func = func
 if len(list(Timer.timers.keys())) >= MAX_TIMERS:
 raise TimerError('Max allowed timers num={}'.format(MAX_TIMERS))
 _timer = _Timer(interval, func, run_once, stopped, *args, **kwargs)
 Timer.timers['{}_{}'.format(len(list(Timer.timers.keys())), blynk._get_func_name(func))] = _timer

 def __call__(self, *f_args, **f_kwargs):
 return self.func(*f_args, **f_kwargs)

 return Deco

 @staticmethod
 def stop(t_id):
 timer = Timer.timers.get(t_id, None)
 if timer is None:
 raise TimerError('Timer id={} not found'.format(t_id))
 Timer.timers[t_id].stopped = True

 @staticmethod
 def start(t_id):
 timer = Timer.timers.get(t_id, None)
 if timer is None:
 raise TimerError('Timer id={} not found'.format(t_id))
 Timer.timers[t_id].stopped = False
 Timer.timers[t_id].fire_time = None
 Timer.timers[t_id].fire_time_prev = None

 @staticmethod
 def is_stopped(t_id):
 timer = Timer.timers.get(t_id, None)
 if timer is None:
 raise TimerError('Timer id={} not found'.format(t_id))
 return timer.stopped

 def get_timers(self):
 states = {True: 'Stopped', False: 'Running'}
 return {k: states[v.stopped] for k, v in self.timers.items()}

 def run(self):
 polling_wait(WAIT_SEC)
 timers_intervals = [curr_timer.run() for curr_timer in Timer.timers.values() if not curr_timer.stopped]
 if not timers_intervals and self.no_timers_err:
 raise TimerError('Running timers not found')
 return timers_intervals


class _Timer(object):
 def __init__(self, interval, deco, run_once, stopped, *args, **kwargs):
 self.interval = interval
 self.deco = deco
 self.args = args
 self.run_once = run_once
 self.kwargs = kwargs
 self.fire_time = None
 self.fire_time_prev = None
 self.stopped = stopped

 def run(self):
 timer_real_interval = 0
 if self.fire_time is None:
 self.fire_time = time.time() + self.interval
 if self.fire_time_prev is None:
 self.fire_time_prev = time.time()
 curr_time = time.time()
 if curr_time >= self.fire_time:
 self.deco(*self.args, **self.kwargs)
 if self.run_once:
 self.stopped = True
 timer_real_interval = curr_time - self.fire_time_prev
 self.fire_time_prev = self.fire_time
 self.fire_time = curr_time + self.interval
 return timer_real_interval