#!/usr/bin/env python3
+"""Several helpers to keep track of internal state via periodic
+polling. StateTracker expects to be invoked periodically to maintain
+state whereas the others automatically update themselves and,
+optionally, expose an event for client code to wait on state changes."""
+
import datetime
import logging
import threading
"""
self.update_ids_to_update_secs = update_ids_to_update_secs
self.last_reminder_ts: Dict[str, Optional[datetime.datetime]] = {}
+ self.now: Optional[datetime.datetime] = None
for x in update_ids_to_update_secs.keys():
self.last_reminder_ts[x] = None
refresh_secs = self.update_ids_to_update_secs[update_id]
last_run = self.last_reminder_ts[update_id]
if last_run is None: # Never run before
- logger.debug(f'id {update_id} has never been run; running it now')
+ logger.debug('id %s has never been run; running it now', update_id)
self.update(update_id, self.now, self.last_reminder_ts[update_id])
self.last_reminder_ts[update_id] = self.now
else:
delta = self.now - last_run
if delta.total_seconds() >= refresh_secs: # Is overdue?
- logger.debug(f'id {update_id} is overdue; running it now')
+ logger.debug('id %s is overdue; running it now', update_id)
self.update(
update_id,
self.now,
"""
@background_thread
- def pace_maker(self, should_terminate) -> None:
+ def pace_maker(self, should_terminate: threading.Event) -> None:
"""Entry point for a background thread to own calling heartbeat()
at regular intervals so that the main thread doesn't need to do
so.
logger.debug('pace_maker noticed event; shutting down')
return
self.heartbeat()
- logger.debug(f'pace_maker is sleeping for {self.sleep_delay}s')
+ logger.debug('pace_maker is sleeping for %.1fs', self.sleep_delay)
time.sleep(self.sleep_delay)
def __init__(
super().__init__(update_ids_to_update_secs)
if override_sleep_delay is not None:
- logger.debug(f'Overriding sleep delay to {override_sleep_delay}')
+ logger.debug('Overriding sleep delay to %.1f', override_sleep_delay)
self.sleep_delay = override_sleep_delay
else:
periods_list = list(update_ids_to_update_secs.values())
self.sleep_delay = math_utils.gcd_float_sequence(periods_list)
- logger.info(f'Computed sleep_delay={self.sleep_delay}')
+ logger.info('Computed sleep_delay=%.1f', self.sleep_delay)
(thread, stop_event) = self.pace_maker()
self.should_terminate = stop_event
self.updater_thread = thread