|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import itertools |
3 | 4 | import os |
| 5 | +import sys |
| 6 | +import threading |
| 7 | +import time |
4 | 8 | import warnings |
5 | 9 | from datetime import timedelta |
6 | 10 | from functools import lru_cache |
@@ -247,3 +251,46 @@ def _get_ship_config(expedition_dir: Path) -> ShipConfig: |
247 | 251 | raise FileNotFoundError( |
248 | 252 | f'Ship config not found. Save it to "{file_path}".' |
249 | 253 | ) from e |
| 254 | + |
| 255 | + |
| 256 | +class RotatePrint: |
| 257 | + """A rotating symbol for the end of simulation print statements, to indicate progress.""" |
| 258 | + |
| 259 | + def __init__(self, message="Processing...", delay=0.18): |
| 260 | + self.spinner_symbols = itertools.cycle(["-", "\\", "|", "/"]) |
| 261 | + self.delay = delay |
| 262 | + self.message = message |
| 263 | + self.running = False |
| 264 | + self.spinner_thread = None |
| 265 | + self.final_message_printed = False |
| 266 | + |
| 267 | + def _spinner_task(self): |
| 268 | + while self.running: |
| 269 | + sys.stdout.write(f"\r{self.message} {next(self.spinner_symbols)}") |
| 270 | + sys.stdout.flush() |
| 271 | + time.sleep(self.delay) |
| 272 | + |
| 273 | + # overwrite with the final message with completion message |
| 274 | + if not self.final_message_printed: |
| 275 | + sys.stdout.write(f"\r{self.message} [COMPLETED]\n") |
| 276 | + sys.stdout.flush() |
| 277 | + self.final_message_printed = True |
| 278 | + |
| 279 | + def start(self): |
| 280 | + self.running = True |
| 281 | + self.final_message_printed = False |
| 282 | + self.spinner_thread = threading.Thread( |
| 283 | + target=self._spinner_task |
| 284 | + ) # threading allows main function to complete and prevents rotator from hanging |
| 285 | + self.spinner_thread.daemon = True |
| 286 | + self.spinner_thread.start() |
| 287 | + |
| 288 | + def stop(self): |
| 289 | + self.running = False |
| 290 | + if self.spinner_thread and self.spinner_thread.is_alive(): |
| 291 | + self.spinner_thread.join() |
| 292 | + |
| 293 | + if not self.final_message_printed: |
| 294 | + sys.stdout.write(f"\r{self.message} [COMPLETED]\n") |
| 295 | + sys.stdout.flush() |
| 296 | + self.final_message_printed = True |
0 commit comments