gardentech
f34817d5b6
FastAPI backend + vanilla JS frontend for LiDAR/camera startup, camera settings, and rosbag recording, unifying the previous scan_gui/scan_gui_dual/ scan_gui_triple desktop tools into one web app.
55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
"""Tails the lidar/camera/recording log files for the browser log panel.
|
|
|
|
The original PyQt5 scan GUIs had a single unified QPlainTextEdit showing all
|
|
subprocess stdout with a [tag] prefix — this is the web equivalent. Each
|
|
ManagedProcess re-opens its log file in "w" mode on every start (see
|
|
process_manager.py), so a tailer must detect truncation (file shrank since
|
|
last read) and reset to the top rather than seeking past EOF forever.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
|
|
class LogTailer:
|
|
def __init__(self, sources: dict[str, Path]):
|
|
self._sources = sources
|
|
self._offsets: dict[str, int] = {tag: 0 for tag in sources}
|
|
|
|
def read_tail(self, tag: str, max_lines: int = 200) -> list[str]:
|
|
"""Backlog for a fresh websocket connection — last N lines, and
|
|
advances this tag's offset to end-of-file so read_new() only
|
|
reports genuinely new lines afterwards."""
|
|
path = self._sources.get(tag)
|
|
if path is None or not path.exists():
|
|
return []
|
|
try:
|
|
with open(path, "r", errors="replace") as f:
|
|
lines = f.readlines()
|
|
self._offsets[tag] = f.tell()
|
|
except OSError:
|
|
return []
|
|
return [ln.rstrip("\n") for ln in lines[-max_lines:]]
|
|
|
|
def read_new(self) -> Iterable[tuple[str, str]]:
|
|
"""Yields (tag, line) for every line appended since the last call."""
|
|
for tag, path in self._sources.items():
|
|
if not path.exists():
|
|
continue
|
|
try:
|
|
size = path.stat().st_size
|
|
if size < self._offsets.get(tag, 0):
|
|
self._offsets[tag] = 0 # file was truncated (process restarted)
|
|
with open(path, "r", errors="replace") as f:
|
|
f.seek(self._offsets[tag])
|
|
new_text = f.read()
|
|
self._offsets[tag] = f.tell()
|
|
except OSError:
|
|
continue
|
|
if not new_text:
|
|
continue
|
|
for line in new_text.splitlines():
|
|
if line:
|
|
yield tag, line
|