+
+
+class WaitableAutomaticStateTracker(AutomaticStateTracker):
+
+ def __init__(
+ self,
+ update_ids_to_update_secs: Dict[str, float],
+ *,
+ override_sleep_delay: Optional[float] = None,
+ ) -> None:
+ self._something_changed = threading.Event()
+ super().__init__(update_ids_to_update_secs,
+ override_sleep_delay=override_sleep_delay)
+
+ def something_changed(self):
+ self._something_changed.set()
+
+ def did_something_change(self) -> bool:
+ return self._something_changed.is_set()
+
+ def reset(self):
+ self._something_changed.clear()
+
+ def wait(self,
+ *,
+ timeout=None):
+ return self._something_changed.wait(
+ timeout=timeout
+ )