state_tracker.py

   1 #!/usr/bin/env python3
   2
   3 from abc import ABC, abstractmethod
   4 import datetime
   5 import logging
   6 import time
   7 import threading
   8 from typing import Dict, Optional
   9
  10 import pytz
  11
  12 from thread_utils import background_thread
  13
  14 logger = logging.getLogger(__name__)
  15
  16
  17 class StateTracker(ABC):
  18     """A base class that maintains and updates a global state via an
  19     update routine.  Instances of this class should be periodically
  20     invoked via the heartbeat() method.  This method, in turn, invokes
  21     update() with update_ids according to a schedule / periodicity
  22     provided to the c'tor.
  23
  24     """
  25
  26     def __init__(self, update_ids_to_update_secs: Dict[str, float]) -> None:
  27         """The update_ids_to_update_secs dict parameter describes one or more
  28         update types (unique update_ids) and the periodicity(ies), in
  29         seconds, at which it/they should be invoked.
  30
  31         Note that, when more than one update is overdue, they will be
  32         invoked in order by their update_ids so care in choosing these
  33         identifiers may be in order.
  34
  35         """
  36         self.update_ids_to_update_secs = update_ids_to_update_secs
  37         self.last_reminder_ts: Dict[str, Optional[datetime.datetime]] = {}
  38         for x in update_ids_to_update_secs.keys():
  39             self.last_reminder_ts[x] = None
  40
  41     @abstractmethod
  42     def update(
  43         self,
  44         update_id: str,
  45         now: datetime.datetime,
  46         last_invocation: Optional[datetime.datetime],
  47     ) -> None:
  48         """Put whatever you want here.  The update_id will be the string
  49         passed to the c'tor as a key in the Dict.  It will only be
  50         tapped on the shoulder, at most, every update_secs seconds.
  51         The now param is the approximate current timestamp and the
  52         last_invocation param is the last time you were invoked (or
  53         None on the first invocation)
  54
  55         """
  56         pass
  57
  58     def heartbeat(self, *, force_all_updates_to_run: bool = False) -> None:
  59         """Invoke this method to cause the StateTracker instance to identify
  60         and invoke any overdue updates based on the schedule passed to
  61         the c'tor.  In the base StateTracker class, this method must
  62         be invoked manually with a thread from external code.
  63
  64         If more than one type of update (update_id) are overdue,
  65         they will be invoked in order based on their update_ids.
  66
  67         Setting force_all_updates_to_run will invoke all updates
  68         (ordered by update_id) immediately ignoring whether or not
  69         they are due.
  70
  71         """
  72         self.now = datetime.datetime.now(tz=pytz.timezone("US/Pacific"))
  73         for update_id in sorted(self.last_reminder_ts.keys()):
  74             if force_all_updates_to_run:
  75                 logger.debug('Forcing all updates to run')
  76                 self.update(
  77                     update_id, self.now, self.last_reminder_ts[update_id]
  78                 )
  79                 self.last_reminder_ts[update_id] = self.now
  80                 return
  81
  82             refresh_secs = self.update_ids_to_update_secs[update_id]
  83             last_run = self.last_reminder_ts[update_id]
  84             if last_run is None:  # Never run before
  85                 logger.debug(
  86                     f'id {update_id} has never been run; running it now'
  87                 )
  88                 self.update(
  89                     update_id, self.now, self.last_reminder_ts[update_id]
  90                 )
  91                 self.last_reminder_ts[update_id] = self.now
  92             else:
  93                 delta = self.now - last_run
  94                 if delta.total_seconds() >= refresh_secs:  # Is overdue?
  95                     logger.debug(f'id {update_id} is overdue; running it now')
  96                     self.update(
  97                         update_id,
  98                         self.now,
  99                         self.last_reminder_ts[update_id],
 100                     )
 101                     self.last_reminder_ts[update_id] = self.now
 102
 103
 104 class AutomaticStateTracker(StateTracker):
 105     """Just like HeartbeatCurrentState but you don't need to pump the
 106     heartbeat; it runs on a background thread.  Call .shutdown() to
 107     terminate the updates.
 108
 109     """
 110
 111     @background_thread
 112     def pace_maker(self, should_terminate) -> None:
 113         """Entry point for a background thread to own calling heartbeat()
 114         at regular intervals so that the main thread doesn't need to do
 115         so.
 116
 117         """
 118         while True:
 119             if should_terminate.is_set():
 120                 logger.debug('pace_maker noticed event; shutting down')
 121                 return
 122             self.heartbeat()
 123             logger.debug(f'pace_maker is sleeping for {self.sleep_delay}s')
 124             time.sleep(self.sleep_delay)
 125
 126     def __init__(
 127         self,
 128         update_ids_to_update_secs: Dict[str, float],
 129         *,
 130         override_sleep_delay: Optional[float] = None,
 131     ) -> None:
 132         import math_utils
 133
 134         super().__init__(update_ids_to_update_secs)
 135         if override_sleep_delay is not None:
 136             logger.debug(f'Overriding sleep delay to {override_sleep_delay}')
 137             self.sleep_delay = override_sleep_delay
 138         else:
 139             periods_list = list(update_ids_to_update_secs.values())
 140             self.sleep_delay = math_utils.gcd_float_sequence(periods_list)
 141             logger.info(f'Computed sleep_delay={self.sleep_delay}')
 142         (thread, stop_event) = self.pace_maker()
 143         self.should_terminate = stop_event
 144         self.updater_thread = thread
 145
 146     def shutdown(self):
 147         """Terminates the background thread and waits for it to tear down.
 148         This may block for as long as self.sleep_delay.
 149
 150         """
 151         logger.debug(
 152             'Setting shutdown event and waiting for background thread.'
 153         )
 154         self.should_terminate.set()
 155         self.updater_thread.join()
 156         logger.debug('Background thread terminated.')
 157
 158
 159 class WaitableAutomaticStateTracker(AutomaticStateTracker):
 160     """This is an AutomaticStateTracker that exposes a wait method which
 161     will block the calling thread until the state changes with an
 162     optional timeout.  The caller should check the return value of
 163     wait; it will be true if something changed and false if the wait
 164     simply timed out.  If the return value is true, the instance
 165     should be reset() before wait is called again.
 166
 167     Example usage:
 168
 169         detector = waitable_presence.WaitableAutomaticStateSubclass()
 170         while True:
 171             changed = detector.wait(timeout=60 * 5)
 172             if changed:
 173                 detector.reset()
 174                 # Figure out what changed and react
 175             else:
 176                 # Just a timeout; no need to reset.  Maybe do something
 177                 # else before looping up into wait again.
 178
 179     """
 180
 181     def __init__(
 182         self,
 183         update_ids_to_update_secs: Dict[str, float],
 184         *,
 185         override_sleep_delay: Optional[float] = None,
 186     ) -> None:
 187         self._something_changed = threading.Event()
 188         super().__init__(
 189             update_ids_to_update_secs, override_sleep_delay=override_sleep_delay
 190         )
 191
 192     def something_changed(self):
 193         self._something_changed.set()
 194
 195     def did_something_change(self) -> bool:
 196         return self._something_changed.is_set()
 197
 198     def reset(self):
 199         self._something_changed.clear()
 200
 201     def wait(self, *, timeout=None):
 202         return self._something_changed.wait(timeout=timeout)