Feed a serial line and return any decoded output lines. Returns: Empty list during normal output. When a crash dump ends, returns a list of formatted strings (banner + decoded trace).
(self, line: str)
| 107 | return self._crash_detected |
| 108 | |
| 109 | def process_line(self, line: str) -> list[str]: |
| 110 | """Feed a serial line and return any decoded output lines. |
| 111 | |
| 112 | Returns: |
| 113 | Empty list during normal output. When a crash dump ends, returns |
| 114 | a list of formatted strings (banner + decoded trace). |
| 115 | """ |
| 116 | stripped = line.strip() |
| 117 | |
| 118 | if not self._in_crash: |
| 119 | if self._is_crash_start(stripped): |
| 120 | self._in_crash = True |
| 121 | self._crash_detected = True |
| 122 | self._crash_lines = [stripped] |
| 123 | self._grace_count = 0 |
| 124 | return [] |
| 125 | |
| 126 | # Already accumulating a crash. |
| 127 | if self._is_crash_continuation(stripped): |
| 128 | self._crash_lines.append(stripped) |
| 129 | self._grace_count = 0 |
| 130 | elif stripped == "": |
| 131 | # Blank lines are common within dumps; allow a few. |
| 132 | self._grace_count += 1 |
| 133 | else: |
| 134 | self._grace_count += 1 |
| 135 | |
| 136 | if ( |
| 137 | self._grace_count >= _END_GRACE_LINES |
| 138 | or len(self._crash_lines) >= _MAX_CRASH_LINES |
| 139 | ): |
| 140 | return self._finish_crash() |
| 141 | |
| 142 | return [] |
| 143 | |
| 144 | def flush(self) -> list[str]: |
| 145 | """Force-finish any in-progress crash accumulation.""" |
no test coverage detected