Show a progress bar and measure elapsed time until .end() is called
| 135 | |
| 136 | |
| 137 | class TimedProgress: |
| 138 | """Show a progress bar and measure elapsed time until .end() is called""" |
| 139 | |
| 140 | def __init__(self, seconds, prefix=''): |
| 141 | |
| 142 | self.SHOW_PROGRESS = SHOW_PROGRESS |
| 143 | if self.SHOW_PROGRESS: |
| 144 | self.p = Process(target=progress_bar, args=(seconds, prefix)) |
| 145 | self.p.start() |
| 146 | |
| 147 | self.stats = {'start_ts': datetime.now(timezone.utc), 'end_ts': None} |
| 148 | |
| 149 | def end(self): |
| 150 | """immediately end progress, clear the progressbar line, and save end_ts""" |
| 151 | |
| 152 | |
| 153 | end_ts = datetime.now(timezone.utc) |
| 154 | self.stats['end_ts'] = end_ts |
| 155 | |
| 156 | if self.SHOW_PROGRESS: |
| 157 | # terminate if we havent already terminated |
| 158 | try: |
| 159 | # kill the progress bar subprocess |
| 160 | try: |
| 161 | self.p.close() # must be closed *before* its terminnated |
| 162 | except (KeyboardInterrupt, SystemExit): |
| 163 | print() |
| 164 | raise |
| 165 | except BaseException: # lgtm [py/catch-base-exception] |
| 166 | pass |
| 167 | self.p.terminate() |
| 168 | self.p.join() |
| 169 | |
| 170 | |
| 171 | # clear whole terminal line |
| 172 | try: |
| 173 | sys.stdout.write('\r{}{}\r'.format((' ' * TERM_WIDTH()), ANSI['reset'])) |
| 174 | except (IOError, BrokenPipeError): |
| 175 | # ignore when the parent proc has stopped listening to our stdout |
| 176 | pass |
| 177 | except ValueError: |
| 178 | pass |
| 179 | |
| 180 | |
| 181 | @enforce_types |
no outgoing calls
no test coverage detected