1 | import time |
||
2 | |||
3 | from cleo import ProgressBar as CleoProgressBar, Helper, Output |
||
4 | from cleo.exceptions import CleoException |
||
5 | |||
6 | |||
7 | View Code Duplication | class ProgressBar(CleoProgressBar): |
|
0 ignored issues
–
show
Duplication
introduced
by
![]() |
|||
8 | """ |
||
9 | Customized version of Cleo's ProgressBar. |
||
10 | """ |
||
11 | refresh_interval = 10 |
||
12 | """ |
||
13 | The refresh interval in seconds. |
||
14 | |||
15 | :type: int |
||
16 | """ |
||
17 | |||
18 | # ------------------------------------------------------------------------------------------------------------------ |
||
19 | def __init__(self, output: Output, maximum: int = 0): |
||
20 | """ |
||
21 | Constructor. |
||
22 | |||
23 | @param Output output: The output object. |
||
24 | @param int maximum: Maximum steps (0 if unknown). |
||
25 | """ |
||
26 | CleoProgressBar.__init__(self, output, maximum) |
||
27 | |||
28 | self.__last_display: int = -1 |
||
29 | """ |
||
30 | The last epoch the progress bar was printed. |
||
31 | """ |
||
32 | |||
33 | # Show now "0/max [>------] ) 0%" (instead of after step 1: "1/max [>------] 0% very long time". |
||
34 | self.set_format(' %current%/%max% [%bar%] %percent:3s%%') |
||
35 | self.set_progress(0) |
||
36 | |||
37 | # Display the remaining time. |
||
38 | self.set_format(' %current%/%max% [%bar%] %percent:3s%% %remaining%') |
||
39 | |||
40 | # ------------------------------------------------------------------------------------------------------------------ |
||
41 | def finish(self) -> None: |
||
42 | """ |
||
43 | Finish the progress output. |
||
44 | """ |
||
45 | # Force set_progress to call the parent method. |
||
46 | self.__last_display = -1 |
||
47 | |||
48 | # No more remaining time, display the total elapsed time. |
||
49 | self.set_format(' %current%/%max% [%bar%] %percent:3s%% %elapsed%') |
||
50 | |||
51 | CleoProgressBar.finish(self) |
||
52 | self._output.writeln('') |
||
53 | |||
54 | # ------------------------------------------------------------------------------------------------------------------ |
||
55 | def _formatter_remaining(self) -> str: |
||
56 | """ |
||
57 | Bug fix, see https://github.com/sdispater/cleo/pull/53. |
||
58 | |||
59 | :rtype: str |
||
60 | """ |
||
61 | if not self._max: |
||
62 | raise CleoException('Unable to display the remaining time if the maximum number of steps is not set.') |
||
63 | |||
64 | if not self._step: |
||
65 | remaining = 0 |
||
66 | else: |
||
67 | remaining = (round((time.time() - self._start_time) / self._step * (self._max - self._step))) |
||
68 | |||
69 | return Helper.format_time(remaining) |
||
70 | |||
71 | # ------------------------------------------------------------------------------------------------------------------ |
||
72 | def set_progress(self, step: int) -> None: |
||
73 | """ |
||
74 | Sets the current progress. |
||
75 | |||
76 | @param int step: The current progress. |
||
77 | """ |
||
78 | if self._should_overwrite: |
||
79 | now = time.time() |
||
80 | if self.__last_display == -1 or (now - self.__last_display) > ProgressBar.refresh_interval: |
||
81 | CleoProgressBar.set_progress(self, step) |
||
82 | self.__last_display = now |
||
83 | else: |
||
84 | if self._max and step > self._max: |
||
85 | self._max = step |
||
86 | elif step < 0: |
||
87 | step = 0 |
||
88 | self._step = step |
||
89 | |||
90 | else: |
||
91 | CleoProgressBar.set_progress(self, step) |
||
92 | |||
93 | # ---------------------------------------------------------------------------------------------------------------------- |
||
94 |