6133538591

Committed 09 Sep 2023 10:41PM UTC coverage: 16.096% (+5.4%) from 10.746%

Build # 6133538591

Build Type

push

github

Committed by

yeliudev

Commit Message

Support iterable progress bar

Run Details

12 of 12 new or added lines in 1 file covered. (100.0%)

649 of 4032 relevant lines covered (16.1%)

3.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

14.86

/nncore/utils/progress.py

# Copyright (c) Ye Liu. Licensed under the MIT License.

from math import ceil
from shutil import get_terminal_size

from .timer import Timer


class ProgressBar(object):
    """
    A progress bar which showing the state of a progress.

    Args:
        iterable (iterable | None, optional): The iterable object to decorate
            with a progress bar. Default: ``None``.
        num_tasks (int | None, optional): The number of expected iterations.
            If not specified, the length of iterable object will be used.
            Default: ``None``.
        active (bool | None, optional): Whether the progress bar is active. If
            not specified, only the main process will show the progree bar.
            Default: ``None``.

    Example:
        >>> for item in ProgressBar(range(10)):
        ...     # do processing

        >>> prog_bar = ProgressBar(num_tasks=10)
        >>> for item in range(10):
        ...     # do processing
        ...     prog_bar.update()
    """

    _wb = '\r[{{}}] {}/{}, {:.1f} task/s, elapsed: {}, eta: {}{}'
    _ob = '\rCompleted: {}, elapsed: {}, {:.1f} tasks/s'

    def __init__(self, iterable=None, num_tasks=None, active=None):
        self._iterable = iterable
        self._num_tasks = num_tasks or (len(iterable) if hasattr(
            iterable, '__len__') else None)
        self._completed = 0

        if active is None:
            try:
                from nncore.engine import is_main_process
                self._active = is_main_process()
            except ImportError:
                self._active = True
        else:
            self._active = active

        if self._active:
            if self._num_tasks is not None:
                msg = self._wb.format(0, self._num_tasks, 0, 0, 0, '')
                msg = msg.format(' ' * self._get_bar_width(msg))
            else:
                msg = self._ob.format(0, 0, 0)

            print(msg, end='')
            self._last_length = len(msg)

            self._timer = Timer()

    def __iter__(self):
        for item in self._iterable:
            yield item
            self.update()

    def _get_bar_width(self, msg):
        width, _ = get_terminal_size()
        bar_width = min(int(width - len(msg)) + 2, int(width * 0.6), 40)
        return max(2, bar_width)

    def _get_time_str(self, second):
        if second >= 86400:
            day = second // 86400
            second -= (day * 86400)
            day = '{}d'.format(day)
        else:
            day = ''

        if second >= 3600:
            hour = second // 3600
            second -= (hour * 3600)
            hour = '{}h'.format(hour)
        else:
            hour = ''

        if second >= 60:
            minute = second // 60
            second -= minute * 60
            minute = '{}m'.format(minute)
        else:
            minute = ''

        if second > 0:
            second = '{}s'.format(second)
        else:
            second = ''

        time_str = '{}{}{}{}'.format(day, hour, minute, second) or '0s'
        return time_str

    def update(self, times=1):
        if not self._active:
            return

        for _ in range(times):
            self._completed += 1
            ela = self._timer.seconds()
            fps = self._completed / ela
            ela_str = self._get_time_str(ceil(ela))

            if self._num_tasks is not None:
                perc = self._completed / float(self._num_tasks)
                eta = int(ela * (1 - perc) / perc + 0.5)
                eta_str = self._get_time_str(ceil(eta))
                msg = self._wb.format(
                    self._completed, self._num_tasks, fps, ela_str, eta_str,
                    '\n' if self._num_tasks == self._completed else '')
                bar_width = self._get_bar_width(msg)
                mark_width = int(bar_width * perc)
                chars = '>' * mark_width + ' ' * (bar_width - mark_width)
                msg = msg.format(chars)
            else:
                msg = self._ob.format(self._completed, ela_str, fps)

            print(msg.ljust(self._last_length), end='')
            self._last_length = len(msg)

1	# Copyright (c) Ye Liu. Licensed under the MIT License.
2
3	from math import ceil	20✔
4	from shutil import get_terminal_size	20✔
5
6	from .timer import Timer	20✔
7
8
9	class ProgressBar(object):	20✔
10	"""
11	A progress bar which showing the state of a progress.
12
13	Args:
14	iterable (iterable \| None, optional): The iterable object to decorate
15	with a progress bar. Default: ``None``.
16	num_tasks (int \| None, optional): The number of expected iterations.
17	If not specified, the length of iterable object will be used.
18	Default: ``None``.
19	active (bool \| None, optional): Whether the progress bar is active. If
20	not specified, only the main process will show the progree bar.
21	Default: ``None``.
22
23	Example:
24	>>> for item in ProgressBar(range(10)):
25	... # do processing
26
27	>>> prog_bar = ProgressBar(num_tasks=10)
28	>>> for item in range(10):
29	... # do processing
30	... prog_bar.update()
31	"""
32
33	_wb = '\r[{{}}] {}/{}, {:.1f} task/s, elapsed: {}, eta: {}{}'	20✔
34	_ob = '\rCompleted: {}, elapsed: {}, {:.1f} tasks/s'	20✔
35
36	def __init__(self, iterable=None, num_tasks=None, active=None):	20✔
37	self._iterable = iterable	×
38	self._num_tasks = num_tasks or (len(iterable) if hasattr(	×
39	iterable, '__len__') else None)
40	self._completed = 0	×
41
42	if active is None:	×
43	try:	×
44	from nncore.engine import is_main_process	×
45	self._active = is_main_process()	×
46	except ImportError:	×
47	self._active = True	×
48	else:
49	self._active = active	×
50
51	if self._active:	×
52	if self._num_tasks is not None:	×
53	msg = self._wb.format(0, self._num_tasks, 0, 0, 0, '')	×
54	msg = msg.format(' ' * self._get_bar_width(msg))	×
55	else:
56	msg = self._ob.format(0, 0, 0)	×
57
58	print(msg, end='')	×
59	self._last_length = len(msg)	×
60
61	self._timer = Timer()	×
62
63	def __iter__(self):	20✔
64	for item in self._iterable:	×
65	yield item	×
66	self.update()	×
67
68	def _get_bar_width(self, msg):	20✔
69	width, _ = get_terminal_size()	×
70	bar_width = min(int(width - len(msg)) + 2, int(width * 0.6), 40)	×
71	return max(2, bar_width)	×
72
73	def _get_time_str(self, second):	20✔
74	if second >= 86400:	×
75	day = second // 86400	×
76	second -= (day * 86400)	×
77	day = '{}d'.format(day)	×
78	else:
79	day = ''	×
80
81	if second >= 3600:	×
82	hour = second // 3600	×
83	second -= (hour * 3600)	×
84	hour = '{}h'.format(hour)	×
85	else:
86	hour = ''	×
87
88	if second >= 60:	×
89	minute = second // 60	×
90	second -= minute * 60	×
91	minute = '{}m'.format(minute)	×
92	else:
93	minute = ''	×
94
95	if second > 0:	×
96	second = '{}s'.format(second)	×
97	else:
98	second = ''	×
99
100	time_str = '{}{}{}{}'.format(day, hour, minute, second) or '0s'	×
101	return time_str	×
102
103	def update(self, times=1):	20✔
104	if not self._active:	×
105	return	×
106
107	for _ in range(times):	×
108	self._completed += 1	×
109	ela = self._timer.seconds()	×
110	fps = self._completed / ela	×
111	ela_str = self._get_time_str(ceil(ela))	×
112
113	if self._num_tasks is not None:	×
114	perc = self._completed / float(self._num_tasks)	×
115	eta = int(ela * (1 - perc) / perc + 0.5)	×
116	eta_str = self._get_time_str(ceil(eta))	×
117	msg = self._wb.format(	×
118	self._completed, self._num_tasks, fps, ela_str, eta_str,
119	'\n' if self._num_tasks == self._completed else '')
120	bar_width = self._get_bar_width(msg)	×
121	mark_width = int(bar_width * perc)	×
122	chars = '>' * mark_width + ' ' * (bar_width - mark_width)	×
123	msg = msg.format(chars)	×
124	else:
125	msg = self._ob.format(self._completed, ela_str, fps)	×
126
127	print(msg.ljust(self._last_length), end='')	×
128	self._last_length = len(msg)	×

yeliudev / nncore / 6133538591

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous