FM / tqdm

/usr/local/lib/python3.6/site-packages/tqdm/

/usr/local/lib/python3.6/site-packages/tqdm

Name	Size	Mode	Actions
contrib/	-	0755	rm
__pycache__/	-	0755	rm
asyncio.py	2789	0644	edit dl rm
auto.py	1106	0644	edit dl rm
autonotebook.py	956	0644	edit dl rm
cli.py	10871	0644	edit dl rm
completion.sh	946	0755	edit dl rm
dask.py	1377	0644	edit dl rm
gui.py	5943	0644	edit dl rm
keras.py	4409	0644	edit dl rm
notebook.py	11317	0644	edit dl rm
rich.py	5152	0644	edit dl rm
std.py	58341	0644	edit dl rm
tk.py	6948	0644	edit dl rm
tqdm.1	7997	0644	edit dl rm
utils.py	9803	0644	edit dl rm
version.py	333	0644	edit dl rm
_dist_ver.py	23	0644	edit dl rm
_main.py	283	0644	edit dl rm
_monitor.py	3699	0644	edit dl rm
_tqdm.py	283	0644	edit dl rm
_tqdm_gui.py	287	0644	edit dl rm
_tqdm_notebook.py	307	0644	edit dl rm
_tqdm_pandas.py	888	0644	edit dl rm
_utils.py	596	0644	edit dl rm
__init__.py	1639	0644	edit dl rm
__main__.py	30	0644	edit dl rm

Edit: /usr/local/lib/python3.6/site-packages/tqdm/_monitor.py (3699B)

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()

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()