In-memory measurement trend recording for one session.
Thread-safe by a single internal lock: the session worker thread appends
rows while request coroutines start/stop/read. Columns are frozen at start
(the API locks the measurement selection while recording, so row values
always align with them).
TrendRecorder(max_rows: int = MAX_LOG_ROWS)
Source code in scpi_control/server/recorder.py
| def __init__(self, max_rows: int = MAX_LOG_ROWS):
self.max_rows = max_rows
self._lock = threading.Lock()
self._state = "idle"
self._started_at: Optional[float] = None
self._columns: List[Tuple[int, str]] = []
self._rows: "deque" = deque(maxlen=max_rows)
|
start(columns: List[Tuple[int, str]], started_at: float) -> None
Freeze columns and begin a fresh recording (clears prior rows).
Source code in scpi_control/server/recorder.py
| def start(self, columns: List[Tuple[int, str]], started_at: float) -> None:
"""Freeze columns and begin a fresh recording (clears prior rows)."""
with self._lock:
self._state = "recording"
self._started_at = started_at
self._columns = list(columns)
self._rows.clear()
|
Stop appending; recorded rows stay available for export.
Source code in scpi_control/server/recorder.py
| def stop(self) -> None:
"""Stop appending; recorded rows stay available for export."""
with self._lock:
self._state = "idle"
|
append(timestamp: float, values: List[Optional[float]]) -> None
Record one sample row; a no-op unless recording.
Source code in scpi_control/server/recorder.py
| def append(self, timestamp: float, values: List[Optional[float]]) -> None:
"""Record one sample row; a no-op unless recording."""
with self._lock:
if self._state == "recording":
self._rows.append((timestamp, list(values)))
|