Provides periodic heartbeat output during long compilations to prevent CI timeouts.
| 49 | |
| 50 | |
| 51 | class CompilationHeartbeat: |
| 52 | """Provides periodic heartbeat output during long compilations to prevent CI timeouts.""" |
| 53 | |
| 54 | def __init__(self, interval_seconds: int = 30): |
| 55 | self.interval = interval_seconds |
| 56 | self._stop_event = threading.Event() |
| 57 | self._thread: threading.Thread | None = None |
| 58 | self._start_time = time.time() |
| 59 | self._last_message = "" |
| 60 | |
| 61 | def start(self, message: str = "Compiling") -> None: |
| 62 | """Start the heartbeat thread.""" |
| 63 | if not _IS_CI: |
| 64 | return # Only run heartbeat in CI environments |
| 65 | |
| 66 | self._last_message = message |
| 67 | self._start_time = time.time() |
| 68 | self._stop_event.clear() |
| 69 | self._thread = threading.Thread(target=self._heartbeat_loop, daemon=True) |
| 70 | self._thread.start() |
| 71 | |
| 72 | def update_message(self, message: str) -> None: |
| 73 | """Update the status message shown in heartbeat.""" |
| 74 | self._last_message = message |
| 75 | |
| 76 | def stop(self) -> None: |
| 77 | """Stop the heartbeat thread.""" |
| 78 | self._stop_event.set() |
| 79 | if self._thread: |
| 80 | self._thread.join(timeout=2.0) |
| 81 | self._thread = None |
| 82 | |
| 83 | def _heartbeat_loop(self) -> None: |
| 84 | """Periodically print status to keep CI alive and show progress.""" |
| 85 | heartbeat_count = 0 |
| 86 | while not self._stop_event.wait(self.interval): |
| 87 | heartbeat_count += 1 |
| 88 | elapsed = time.time() - self._start_time |
| 89 | minutes = int(elapsed // 60) |
| 90 | seconds = int(elapsed % 60) |
| 91 | |
| 92 | # Build heartbeat message with optional memory info |
| 93 | mem_mb = _get_memory_usage_mb() |
| 94 | if mem_mb is not None: |
| 95 | mem_info = f", mem: {mem_mb:.0f}MB" |
| 96 | else: |
| 97 | mem_info = "" |
| 98 | |
| 99 | # Print heartbeat with elapsed time and memory usage |
| 100 | _ts_print( |
| 101 | f"[HEARTBEAT {heartbeat_count}] {self._last_message} " |
| 102 | f"(elapsed: {minutes}m {seconds}s{mem_info})", |
| 103 | file=sys.stderr, |
| 104 | ) |
| 105 | # Flush to ensure CI captures the output |
| 106 | sys.stderr.flush() |
| 107 | sys.stdout.flush() |
| 108 |
no outgoing calls
no test coverage detected