Spaces:
Running
Running
| import atexit | |
| from threading import Event, Thread, current_thread | |
| from time import time | |
| from warnings import warn | |
| __all__ = ["TMonitor", "TqdmSynchronisationWarning"] | |
| class TqdmSynchronisationWarning(RuntimeWarning): | |
| """tqdm multi-thread/-process errors which may cause incorrect nesting | |
| but otherwise no adverse effects""" | |
| pass | |
| class TMonitor(Thread): | |
| """ | |
| Monitoring thread for tqdm bars. | |
| Monitors if tqdm bars are taking too much time to display | |
| and readjusts miniters automatically if necessary. | |
| Parameters | |
| ---------- | |
| tqdm_cls : class | |
| tqdm class to use (can be core tqdm or a submodule). | |
| sleep_interval : float | |
| Time to sleep between monitoring checks. | |
| """ | |
| _test = {} # internal vars for unit testing | |
| def __init__(self, tqdm_cls, sleep_interval): | |
| Thread.__init__(self) | |
| self.daemon = True # kill thread when main killed (KeyboardInterrupt) | |
| self.woken = 0 # last time woken up, to sync with monitor | |
| self.tqdm_cls = tqdm_cls | |
| self.sleep_interval = sleep_interval | |
| self._time = self._test.get("time", time) | |
| self.was_killed = self._test.get("Event", Event)() | |
| atexit.register(self.exit) | |
| self.start() | |
| def exit(self): | |
| self.was_killed.set() | |
| if self is not current_thread(): | |
| self.join() | |
| return self.report() | |
| def get_instances(self): | |
| # returns a copy of started `tqdm_cls` instances | |
| return [i for i in self.tqdm_cls._instances.copy() | |
| # Avoid race by checking that the instance started | |
| if hasattr(i, 'start_t')] | |
| def run(self): | |
| cur_t = self._time() | |
| while True: | |
| # After processing and before sleeping, notify that we woke | |
| # Need to be done just before sleeping | |
| self.woken = cur_t | |
| # Sleep some time... | |
| self.was_killed.wait(self.sleep_interval) | |
| # Quit if killed | |
| if self.was_killed.is_set(): | |
| return | |
| # Then monitor! | |
| # Acquire lock (to access _instances) | |
| with self.tqdm_cls.get_lock(): | |
| cur_t = self._time() | |
| # Check tqdm instances are waiting too long to print | |
| instances = self.get_instances() | |
| for instance in instances: | |
| # Check event in loop to reduce blocking time on exit | |
| if self.was_killed.is_set(): | |
| return | |
| # Only if mininterval > 1 (else iterations are just slow) | |
| # and last refresh exceeded maxinterval | |
| if ( | |
| instance.miniters > 1 | |
| and (cur_t - instance.last_print_t) >= instance.maxinterval | |
| ): | |
| # force bypassing miniters on next iteration | |
| # (dynamic_miniters adjusts mininterval automatically) | |
| instance.miniters = 1 | |
| # Refresh now! (works only for manual tqdm) | |
| instance.refresh(nolock=True) | |
| # Remove accidental long-lived strong reference | |
| del instance | |
| if instances != self.get_instances(): # pragma: nocover | |
| warn("Set changed size during iteration" + | |
| " (see https://github.com/tqdm/tqdm/issues/481)", | |
| TqdmSynchronisationWarning, stacklevel=2) | |
| # Remove accidental long-lived strong references | |
| del instances | |
| def report(self): | |
| return not self.was_killed.is_set() | |