Get new output since last check, optionally filtered by regex.
(self, filter_pattern: str | None = None)
| 71 | self.output_lines.append(line) |
| 72 | |
| 73 | def get_new_output(self, filter_pattern: str | None = None) -> list[str]: |
| 74 | """Get new output since last check, optionally filtered by regex.""" |
| 75 | new_lines = self.output_lines[self.last_read_index :] |
| 76 | self.last_read_index = len(self.output_lines) |
| 77 | |
| 78 | if filter_pattern: |
| 79 | try: |
| 80 | pattern = re.compile(filter_pattern) |
| 81 | new_lines = [line for line in new_lines if pattern.search(line)] |
| 82 | except re.error: |
| 83 | # Invalid regex, return all lines |
| 84 | pass |
| 85 | |
| 86 | return new_lines |
| 87 | |
| 88 | def update_status(self, is_alive: bool, exit_code: int | None = None): |
| 89 | """Update process status.""" |