Source code for mlthon.basics.timer_mgr

import datetime
import typing

from mlthon.basics import utils


class _Timer(object):
    def __init__(self, id: int, duration_ms: int, next_expo_ts: int, callback_func: typing.Callable[[], bool]):
        self.id_ = id
        self.duration_ms_ = duration_ms
        self.next_expiration_ts_ = next_expo_ts
        self.callback_func_ = callback_func


class TimerMgr(object):
    def __init__(self):
        self._MIN_TIMER_DURATION_MS = 250
        self._next_timer_id_ = 0
        self._timers_list_ = []
        self._next_check_ts_ = 0

    def setup_timer(self, duration_ms: int, callback_func: typing.Callable[[], bool]):
        if duration_ms < self._MIN_TIMER_DURATION_MS:
            duration_ms = self._MIN_TIMER_DURATION_MS
        now_ts = utils.get_now_ts()
        next_expiration_ts = now_ts + duration_ms
        new_timer = _Timer(self._next_timer_id_, duration_ms, next_expiration_ts, callback_func)
        self._timers_list_.append(new_timer)
        self._next_timer_id_ = self._next_timer_id_ + 1

    def setup_daily_callback(self, time: str, callback_func: typing.Callable[[], bool]):
        now_date_time = datetime.datetime.now()

        date_time = datetime.datetime.strptime(time, '%H:%M:%S')
        date_time = date_time.replace(now_date_time.year, now_date_time.month, now_date_time.day)

        if date_time < now_date_time:
            date_time += datetime.timedelta(days=1)

        first_duration_ms = round((date_time - now_date_time).total_seconds() * 1000)
        now_ts = utils.get_now_ts()
        next_expiration_ts = now_ts + first_duration_ms
        duration_ms = 24 * 3600 * 1000
        new_timer = _Timer(self._next_timer_id_, duration_ms, next_expiration_ts, callback_func)
        self._timers_list_.append(new_timer)
        self._next_timer_id_ = self._next_timer_id_ + 1

    def process_timers(self):
        now_ts = utils.get_now_ts()
        if now_ts > self._next_check_ts_:
            self._next_check_ts_ = now_ts + self._MIN_TIMER_DURATION_MS
            timers_to_delete = []
            for timer in self._timers_list_:
                if now_ts > timer.next_expiration_ts_:
                    ret_val = timer.callback_func_()
                    if ret_val is not None and ret_val:
                        timer.next_expiration_ts_ = utils.get_now_ts() + timer.duration_ms_
                    else:
                        timers_to_delete.append(timer)

            for timer in timers_to_delete:
                self._timers_list_.remove(timer)