-
Notifications
You must be signed in to change notification settings - Fork 32
/
CounterProgress.py
66 lines (52 loc) · 1.75 KB
/
CounterProgress.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import os
from sys import stdout
from collections import deque
from progress import Infinite
try:
from time import monotonic
except ImportError:
from time import time as monotonic
class CounterProgress(Infinite):
file = stdout
def is_tty(self):
return False
def write(self, s):
if self.file:
line = self.message + s.ljust(self._width)
print('\r' + line, end='', file=self.file)
self._width = max(self._width, len(s))
self.file.flush()
def finish(self):
if self.file:
print(file=self.file)
def __init__(self, message='', **kwargs):
self.encoding = kwargs.get('encoding', 'cp1252')
super(CounterProgress, self).__init__(**kwargs)
self.index = 0
self.start_ts = monotonic()
self.avg = 0
self._avg_update_ts = self.start_ts
self._ts = self.start_ts
self._xput = deque(maxlen=self.sma_window)
self.max = kwargs.get('max', 100)
for key, val in kwargs.items():
setattr(self, key, val)
self._width = 0
self.message = message
if 'DYNO' not in os.environ:
self.file.reconfigure(encoding=self.encoding)
if self.file:
print(self.message, end='', file=self.file)
self.file.flush()
@property
def progress(self):
return min(1, self.index / self.max)
@property
def percent(self):
return self.progress * 100
def update(self):
self.write(" - {:.2f}% [{}/{}]".format(self.percent, self.index, self.max))
def finish_update_with_error(self, count):
self.write(" | Synced ({} files failed to copy)".format(count))
def finish_update(self):
self.write(" | Synced")