from __future__ import annotations from collections import deque from dataclasses import dataclass from typing import Deque, Sequence from ameba_control_panel import config from ameba_control_panel.config import Direction from ameba_control_panel.utils.timeutils import timestamp_ms @dataclass class LogLine: text: str direction: Direction timestamp: str def as_display(self) -> str: return f"{self.timestamp} {self.text}" class LogBuffer: """Keeps a bounded archive plus a bounded UI tail.""" def __init__(self, max_tail: int = config.UI_LOG_TAIL_LINES) -> None: self._max_tail = max_tail self._tail: Deque[LogLine] = deque(maxlen=max_tail) self._archive: Deque[LogLine] = deque(maxlen=config.LOG_ARCHIVE_MAX) def append(self, text: str, direction: Direction) -> LogLine: line = LogLine(text=text.rstrip("\n"), direction=direction, timestamp=timestamp_ms()) self._tail.append(line) self._archive.append(line) return line def extend(self, items: Sequence[LogLine]) -> None: for line in items: self._tail.append(line) self._archive.append(line) def tail(self) -> Deque[LogLine]: return self._tail def archive(self) -> Deque[LogLine]: return self._archive def clear(self) -> None: self._tail.clear() self._archive.clear() def as_text(self, full: bool = False) -> str: source = self._archive if full else self._tail return "\n".join(line.as_display() for line in source)