80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Per-sensor ring of raw reports. Newest sample is at the end."""
|
|
from __future__ import annotations
|
|
|
|
from collections import deque
|
|
from typing import Deque, Dict, List, Optional
|
|
|
|
HISTORY_LIMIT = 1000
|
|
|
|
|
|
class History:
|
|
def __init__(self, limit: int = HISTORY_LIMIT):
|
|
self.limit = max(1, int(limit))
|
|
self._by_mac: Dict[str, Deque[dict]] = {}
|
|
|
|
def add(
|
|
self,
|
|
mac: str,
|
|
ts: float,
|
|
raw_mm: Optional[int],
|
|
battery: int,
|
|
rssi: Optional[int],
|
|
) -> None:
|
|
addr = (mac or "").strip().upper()
|
|
if not addr:
|
|
return
|
|
bucket = self._by_mac.get(addr)
|
|
if bucket is None:
|
|
bucket = deque(maxlen=self.limit)
|
|
self._by_mac[addr] = bucket
|
|
bucket.append(
|
|
{
|
|
"ts": float(ts),
|
|
"raw_mm": None if raw_mm is None else int(raw_mm),
|
|
"battery": int(battery) & 0xFF,
|
|
"rssi": None if rssi is None else int(rssi),
|
|
}
|
|
)
|
|
|
|
def to_dict(self) -> Dict[str, List[dict]]:
|
|
return {mac: list(bucket) for mac, bucket in self._by_mac.items()}
|
|
|
|
def load_dict(self, raw: object) -> None:
|
|
self._by_mac.clear()
|
|
if not isinstance(raw, dict):
|
|
return
|
|
for mac, entries in raw.items():
|
|
addr = str(mac).strip().upper()
|
|
if not addr or not isinstance(entries, list):
|
|
continue
|
|
bucket: Deque[dict] = deque(maxlen=self.limit)
|
|
for entry in entries[-self.limit :]:
|
|
if not isinstance(entry, dict) or "ts" not in entry:
|
|
continue
|
|
try:
|
|
ts = float(entry["ts"])
|
|
except (TypeError, ValueError):
|
|
continue
|
|
raw_mm = entry.get("raw_mm")
|
|
if raw_mm is not None:
|
|
try:
|
|
raw_mm = int(raw_mm)
|
|
except (TypeError, ValueError):
|
|
raw_mm = None
|
|
try:
|
|
battery = int(entry.get("battery", 255)) & 0xFF
|
|
except (TypeError, ValueError):
|
|
battery = 255
|
|
rssi = entry.get("rssi")
|
|
if rssi is not None:
|
|
try:
|
|
rssi = int(rssi)
|
|
except (TypeError, ValueError):
|
|
rssi = None
|
|
bucket.append(
|
|
{"ts": ts, "raw_mm": raw_mm, "battery": battery, "rssi": rssi}
|
|
)
|
|
if bucket:
|
|
self._by_mac[addr] = bucket
|