''' Utilities for showing progress bars, controlling default verbosity, etc. ''' # If the tqdm package is not available, then do not show progress bars; # just connect print_progress to print. import sys, types, builtins try: from tqdm import tqdm try: from tqdm.notebook import tqdm as tqdm_nb except: from tqdm import tqdm_notebook as tqdm_nb except: tqdm = None default_verbosity = True next_description = None python_print = builtins.print def post(**kwargs): ''' When within a progress loop, pbar.post(k=str) will display the given k=str status on the right-hand-side of the progress status bar. If not within a visible progress bar, does nothing. ''' innermost = innermost_tqdm() if innermost is not None: innermost.set_postfix(**kwargs) def desc(desc): ''' When within a progress loop, pbar.desc(str) changes the left-hand-side description of the loop toe the given description. ''' innermost = innermost_tqdm() if innermost is not None: innermost.set_description(str(desc)) def descnext(desc): ''' Called before starting a progress loop, pbar.descnext(str) sets the description text that will be used in the following loop. ''' global next_description if not default_verbosity or tqdm is None: return next_description = desc def print(*args): ''' When within a progress loop, will print above the progress loop. ''' global next_description next_description = None if default_verbosity: msg = ' '.join(str(s) for s in args) if tqdm is None: python_print(msg) else: tqdm.write(msg) def tqdm_terminal(it, *args, **kwargs): ''' Some settings for tqdm that make it run better in resizable terminals. ''' return tqdm(it, *args, dynamic_ncols=True, ascii=True, leave=(innermost_tqdm() is not None), **kwargs) def in_notebook(): ''' True if running inside a Jupyter notebook. ''' # From https://stackoverflow.com/a/39662359/265298 try: shell = get_ipython().__class__.__name__ if shell == 'ZMQInteractiveShell': return True # Jupyter notebook or qtconsole elif shell == 'TerminalInteractiveShell': return False # Terminal running IPython else: return False # Other type (?) except NameError: return False # Probably standard Python interpreter def innermost_tqdm(): ''' Returns the innermost active tqdm progress loop on the stack. ''' if hasattr(tqdm, '_instances') and len(tqdm._instances) > 0: return max(tqdm._instances, key=lambda x: x.pos) else: return None def reporthook(*args, **kwargs): ''' For use with urllib.request.urlretrieve. with pbar.reporthook() as hook: urllib.request.urlretrieve(url, filename, reporthook=hook) ''' kwargs2 = dict(unit_scale=True, miniters=1) kwargs2.update(kwargs) bar = __call__(None, *args, **kwargs2) class ReportHook(object): def __init__(self, t): self.t = t def __call__(self, b=1, bsize=1, tsize=None): if hasattr(self.t, 'total'): if tsize is not None: self.t.total = tsize if hasattr(self.t, 'update'): self.t.update(b * bsize - self.t.n) def __enter__(self): return self def __exit__(self, *exc): if hasattr(self.t, '__exit__'): self.t.__exit__(*exc) return ReportHook(bar) def __call__(x, *args, **kwargs): ''' Invokes a progress function that can wrap iterators to print progress messages, if verbose is True. If verbose is False or tqdm is unavailable, then a quiet non-printing identity function is used. verbose can also be set to a spefific progress function rather than True, and that function will be used. ''' global default_verbosity, next_description if not default_verbosity or tqdm is None: return x if default_verbosity == True: fn = tqdm_nb if in_notebook() else tqdm_terminal else: fn = default_verbosity if next_description is not None: kwargs = dict(kwargs) kwargs['desc'] = next_description next_description = None return fn(x, *args, **kwargs) class VerboseContextManager(): def __init__(self, v, entered=False): self.v, self.entered, self.saved = v, False, [] if entered: self.__enter__() self.entered = True def __enter__(self): global default_verbosity if self.entered: self.entered = False else: self.saved.append(default_verbosity) default_verbosity = self.v return self def __exit__(self, exc_type, exc_value, exc_traceback): global default_verbosity default_verbosity = self.saved.pop() def __call__(self, v=True): ''' Calling the context manager makes a new context that is pre-entered, so it works as both a plain function and as a factory for a context manager. ''' new_v = v if self.v else not v cm = VerboseContextManager(new_v, entered=True) default_verbosity = new_v return cm # Use as either "with pbar.verbose:" or "pbar.verbose(False)", or also # "with pbar.verbose(False):" verbose = VerboseContextManager(True) # Use as either "with @pbar.quiet" or "pbar.quiet(True)". or also # "with pbar.quiet(True):" quiet = VerboseContextManager(False) class CallableModule(types.ModuleType): def __init__(self): # or super().__init__(__name__) for Python 3 types.ModuleType.__init__(self, __name__) self.__dict__.update(sys.modules[__name__].__dict__) def __call__(self, x, *args, **kwargs): return __call__(x, *args, **kwargs) sys.modules[__name__] = CallableModule()