#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# This file is part of the `pypath` python module
#
# Copyright 2014-2023
# EMBL, EMBL-EBI, Uniklinik RWTH Aachen, Heidelberg University
#
# Authors: see the file `README.rst`
# Contact: Dénes Türei (turei.denes@gmail.com)
#
# Distributed under the GPLv3 License.
# See accompanying file LICENSE.txt or copy at
# https://www.gnu.org/licenses/gpl-3.0.html
#
# Website: https://pypath.omnipathdb.org/
#
import time
import os
import importlib as imp
import tqdm
import pypath.share.settings as settings
__all__ = ['Progress']
[docs]
class Progress(object):
"""
Before I had my custom progressbar here.
Now it is a wrapper around the great progressbar `tqdm`.
Old implementation moved to `OldProgress` class.
"""
[docs]
def __init__(
self,
total = None,
name = 'Progress',
interval = None,
percent = True,
status = 'initializing',
done = 0,
init = True,
unit = 'it',
off = None,
iterable = None,
):
if off is None:
self.off = not settings.get('progressbars')
else:
self.off = off
self.name = name
self.interval = (
interval
if interval is not None else
max(int(total / 100), 1)
if isinstance(total, (int, float)) else
1
)
self.total = total
self.iterable = iterable
self.done = done
self.status = status
self.unit = unit
self.start_time = time.time()
self.min_update_interval = 0.1
self.last_printed_value = 0
if init and not self.off:
self.init_tqdm()
def reload(self):
modname = self.__class__.__module__
mod = __import__(modname, fromlist=[modname.split('.')[0]])
imp.reload(mod)
new = getattr(mod, self.__class__.__name__)
setattr(self, '__class__', new)
[docs]
def init_tqdm(self):
"""
Creates a tqdm instance.
"""
self.tqdm = tqdm.tqdm(
iterable = self.iterable,
total = self.total,
desc = '%s: %s' % (self.name, self.status),
unit_scale = True,
unit = self.unit,
dynamic_ncols = True,
)
self.last_updated = time.time()
def __iter__(self):
iterable = (
self.tqdm.__iter__()
if hasattr(self, 'tqdm') else
self.iterable
if hasattr(self.iterable, '__iter__') else
()
)
for it in iterable:
self.set_status('busy')
yield it
self.set_status('finished')
[docs]
def step(self, step = 1, msg = None, status = 'busy', force = False):
"""
Updates the progressbar by the desired number of steps.
:param int step: Number of steps or items.
"""
self.done += step
if self.off:
return
if (
force or (
self.done % self.interval < 1.0 and
time.time() - self.last_updated > self.min_update_interval
)
):
self.set_status(status)
this_update = max(0, self.done - self.last_printed_value)
if this_update == 0:
self.tqdm.refresh()
self.tqdm.fp.flush()
else:
self.tqdm.update(int(this_update))
self.last_printed_value = self.done
self.last_updated = time.time()
[docs]
def terminate(self, status = 'finished'):
"""
Terminates the progressbar and destroys the tqdm object.
"""
if self.off:
return
self.step(self.total - self.done, force = True, status = status)
self.tqdm.close()
[docs]
def set_total(self, total):
"""
Changes the total value of the progress bar.
"""
if self.off:
return
self.total = total
self.tqdm.total = total
self.step(0)
[docs]
def set_done(self, done):
"""
Sets the position of the progress bar.
"""
if self.off:
return
self.done = done
self.tqdm.n = self.done
self.tqdm.last_print_n = self.done
self.step(0)
[docs]
def set_status(self, status):
"""
Changes the prefix of the progressbar.
"""
if not self.off and status != self.status:
self.status = status
self.tqdm.set_description(self.get_desc())
self.tqdm.refresh()
[docs]
def get_desc(self):
"""
Returns a formatted string of the description, consisted of
the name and the status. The name supposed something constant
within the life of the progressbar, while the status is there
to give information about the current stage of the task.
"""
return '%s%s%s%s' % (
' ' * 8,
self.name,
' -- ' if len(self.name) else '',
self.status,
)
