47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
import time
|
|
from collections import deque
|
|
|
|
from app.sample import Sample
|
|
|
|
|
|
class HistoryStore:
|
|
"""In-memory ring buffer of Sample points, oldest dropped first.
|
|
|
|
`maxlen` is derived from `DASH_RETENTION_MINUTES` / `DASH_SAMPLE_INTERVAL`
|
|
(see `Settings.history_maxlen`). All methods are called from the event
|
|
loop thread; the sampler's collection work happens in a worker thread
|
|
before `record` is called, so no locking is needed.
|
|
"""
|
|
|
|
def __init__(self, maxlen: int) -> None:
|
|
"""Create an empty store.
|
|
|
|
Args:
|
|
maxlen: maximum number of samples to keep.
|
|
"""
|
|
self._buf: deque[Sample] = deque(maxlen=maxlen)
|
|
|
|
def record(self, sample: Sample) -> None:
|
|
"""Stamp the sample with the current unix time and append it.
|
|
|
|
Args:
|
|
sample: sample to store; its `ts` field is overwritten.
|
|
"""
|
|
sample.ts = time.time()
|
|
self._buf.append(sample)
|
|
|
|
def snapshot(self) -> list[Sample]:
|
|
"""Return all stored samples, oldest first.
|
|
|
|
Returns:
|
|
A copy of the buffer contents as a list.
|
|
"""
|
|
return list(self._buf)
|
|
|
|
def latest(self) -> Sample | None:
|
|
"""Return the newest sample, or None if the store is empty."""
|
|
return self._buf[-1] if self._buf else None
|
|
|
|
def __len__(self) -> int:
|
|
"""Number of samples currently stored."""
|
|
return len(self._buf)
|