(self)
| 476 | self.socketio.emit("update", self.state(), to=self.room) |
| 477 | |
| 478 | def _watch_loop(self) -> None: |
| 479 | while self._watching and not self.shutdown_event.is_set(): |
| 480 | if not self.path.exists(): |
| 481 | if not self.missing: |
| 482 | self.missing = True |
| 483 | self.emit_update() |
| 484 | time.sleep(1) |
| 485 | continue |
| 486 | |
| 487 | self.missing = False |
| 488 | try: |
| 489 | stat = self.path.stat() |
| 490 | except Exception: |
| 491 | time.sleep(1) |
| 492 | continue |
| 493 | |
| 494 | size = stat.st_size |
| 495 | inode = stat.st_ino |
| 496 | |
| 497 | rotated = self._last_inode is not None and inode != self._last_inode |
| 498 | truncated = size < self._last_size |
| 499 | |
| 500 | if self._last_inode is None or rotated or truncated: |
| 501 | self._reset_parser() |
| 502 | with self.lock: |
| 503 | self._read_full() |
| 504 | self.emit_update() |
| 505 | time.sleep(0.2) |
| 506 | continue |
| 507 | |
| 508 | if size > self._last_size: |
| 509 | try: |
| 510 | with open(self.path, "r", encoding="utf-8", errors="replace") as f: |
| 511 | f.seek(self._last_position) |
| 512 | new_content = f.read() |
| 513 | self._last_position = f.tell() |
| 514 | self._last_size = size |
| 515 | with self.lock: |
| 516 | for line in new_content.splitlines(): |
| 517 | self.parser.parse_line(line.rstrip("\n")) |
| 518 | self.emit_update() |
| 519 | except Exception: |
| 520 | time.sleep(1) |
| 521 | continue |
| 522 | |
| 523 | time.sleep(0.5) |
| 524 | |
| 525 | |
| 526 | class LogManager: |
nothing calls this directly
no test coverage detected