289 lines
9.6 KiB
Python
289 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Sensor slot registry: addr → slot, persistence, stale/lost."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
|
|
from config import DEFAULT_REPORT_INTERVAL_MIN, N, STALE_PERIODS
|
|
from protocol import BATT_UNKNOWN, DIST_NO_TARGET, charge_percent, voltage_v
|
|
|
|
log = logging.getLogger("collector.registry")
|
|
|
|
|
|
@dataclass
|
|
class Sensor:
|
|
addr: str
|
|
slot: int
|
|
last_seen: float = 0.0
|
|
battery: int = BATT_UNKNOWN
|
|
flags: int = 0
|
|
dist: int = DIST_NO_TARGET
|
|
state: str = "new" # new|live|stale|lost
|
|
rssi: Optional[int] = None
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"addr": self.addr,
|
|
"slot": self.slot,
|
|
"last_seen": self.last_seen,
|
|
"battery": self.battery,
|
|
"flags": self.flags,
|
|
"dist": self.dist,
|
|
"state": self.state,
|
|
"rssi": self.rssi,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "Sensor":
|
|
return cls(
|
|
addr=str(d.get("addr", "")).upper(),
|
|
slot=int(d.get("slot", -1)),
|
|
last_seen=float(d.get("last_seen", 0.0)),
|
|
battery=int(d.get("battery", BATT_UNKNOWN)),
|
|
flags=int(d.get("flags", 0)),
|
|
dist=int(d.get("dist", DIST_NO_TARGET)),
|
|
state=str(d.get("state", "new")),
|
|
rssi=d.get("rssi", None),
|
|
)
|
|
|
|
|
|
class Registry:
|
|
def __init__(self, n: int = N, path: Optional[Path] = None):
|
|
self.n = n
|
|
self.path = path
|
|
self.slots: List[Optional[Sensor]] = [None] * n
|
|
self.by_addr: Dict[str, Sensor] = {}
|
|
|
|
def normalize_addr(self, addr: str) -> str:
|
|
return (addr or "").strip().upper()
|
|
|
|
def free_slot(self) -> Optional[int]:
|
|
for i, s in enumerate(self.slots):
|
|
if s is None:
|
|
return i
|
|
return None
|
|
|
|
def pool_full(self) -> bool:
|
|
return all(s is not None for s in self.slots)
|
|
|
|
def get(self, addr: str) -> Optional[Sensor]:
|
|
return self.by_addr.get(self.normalize_addr(addr))
|
|
|
|
def assign(self, addr: str) -> Optional[Sensor]:
|
|
"""Create sensor in a free slot. Returns None if pool full."""
|
|
addr = self.normalize_addr(addr)
|
|
existing = self.by_addr.get(addr)
|
|
if existing is not None:
|
|
return existing
|
|
slot = self.free_slot()
|
|
if slot is None:
|
|
return None
|
|
sensor = Sensor(addr=addr, slot=slot, last_seen=time.time(), state="new")
|
|
self.slots[slot] = sensor
|
|
self.by_addr[addr] = sensor
|
|
self._assert_invariants()
|
|
return sensor
|
|
|
|
def update_report(
|
|
self,
|
|
addr: str,
|
|
battery: int,
|
|
flags: int,
|
|
dist: int,
|
|
rssi: Optional[int] = None,
|
|
) -> Optional[Sensor]:
|
|
sensor = self.get(addr)
|
|
if sensor is None:
|
|
return None
|
|
sensor.battery = battery
|
|
sensor.flags = flags
|
|
sensor.dist = dist
|
|
sensor.last_seen = time.time()
|
|
if rssi is not None:
|
|
sensor.rssi = rssi
|
|
if sensor.state in ("new", "stale", "live"):
|
|
sensor.state = "live"
|
|
return sensor
|
|
|
|
def touch(self, addr: str, rssi: Optional[int] = None) -> Optional[Sensor]:
|
|
sensor = self.get(addr)
|
|
if sensor is None:
|
|
return None
|
|
sensor.last_seen = time.time()
|
|
if rssi is not None:
|
|
sensor.rssi = rssi
|
|
if sensor.state == "stale":
|
|
sensor.state = "live"
|
|
return sensor
|
|
|
|
def release(self, sensor: Sensor) -> None:
|
|
addr = self.normalize_addr(sensor.addr)
|
|
if 0 <= sensor.slot < self.n and self.slots[sensor.slot] is sensor:
|
|
self.slots[sensor.slot] = None
|
|
self.by_addr.pop(addr, None)
|
|
sensor.state = "lost"
|
|
|
|
def reevaluate(
|
|
self,
|
|
now: Optional[float] = None,
|
|
period_s: Optional[float] = None,
|
|
) -> bool:
|
|
"""Update stale/lost. Returns True if any slot freed or state changed."""
|
|
now = time.time() if now is None else now
|
|
if period_s is None:
|
|
period_s = float(DEFAULT_REPORT_INTERVAL_MIN * 60)
|
|
stale_after = period_s * STALE_PERIODS
|
|
lost_after = period_s * STALE_PERIODS * 2
|
|
changed = False
|
|
to_release: List[Sensor] = []
|
|
for sensor in list(self.by_addr.values()):
|
|
silence = now - sensor.last_seen
|
|
prev = sensor.state
|
|
if silence >= lost_after:
|
|
to_release.append(sensor)
|
|
elif silence >= stale_after:
|
|
if sensor.state != "stale":
|
|
sensor.state = "stale"
|
|
changed = True
|
|
elif sensor.state == "stale":
|
|
# only reports should revive; leave as-is here
|
|
pass
|
|
if prev != sensor.state:
|
|
changed = True
|
|
for sensor in to_release:
|
|
log.info("sensor lost slot=%s addr=%s", sensor.slot, sensor.addr)
|
|
self.release(sensor)
|
|
changed = True
|
|
if changed:
|
|
self._assert_invariants()
|
|
return changed
|
|
|
|
def freshest_report_bytes(self) -> Optional[bytes]:
|
|
"""Latest report among live/stale sensors by last_seen (for N=1 mirror)."""
|
|
from protocol import pack_report
|
|
|
|
best: Optional[Sensor] = None
|
|
for s in self.by_addr.values():
|
|
if s.state == "lost":
|
|
continue
|
|
if best is None or s.last_seen > best.last_seen:
|
|
best = s
|
|
if best is None:
|
|
return None
|
|
return pack_report(best.battery, best.flags, best.dist)
|
|
|
|
def mirror_payload_parts(self) -> tuple:
|
|
"""Return (bitmap, list of 4-byte reports per slot index)."""
|
|
from protocol import pack_report
|
|
|
|
bitmap = 0
|
|
reports = []
|
|
for i in range(self.n):
|
|
s = self.slots[i]
|
|
if s is None or s.state == "lost":
|
|
reports.append(bytes((0xFF, 0, 0xFF, 0xFF)))
|
|
else:
|
|
bitmap |= 1 << i
|
|
reports.append(pack_report(s.battery, s.flags, s.dist))
|
|
return bitmap, reports
|
|
|
|
def status_dict(
|
|
self,
|
|
names: Optional[Dict[str, str]] = None,
|
|
report_interval_min: int = DEFAULT_REPORT_INTERVAL_MIN,
|
|
) -> dict:
|
|
names = names or {}
|
|
now = time.time()
|
|
period_s = max(1, int(report_interval_min) * 60)
|
|
slots_out = []
|
|
for i in range(self.n):
|
|
s = self.slots[i]
|
|
if s is None:
|
|
slots_out.append({"slot": i, "empty": True})
|
|
else:
|
|
d = s.to_dict()
|
|
d["slot"] = i
|
|
d["empty"] = False
|
|
d["reading"] = None if s.dist == DIST_NO_TARGET else s.dist
|
|
volts = voltage_v(s.battery)
|
|
d["voltage_v"] = volts
|
|
d["charge_percent"] = None if volts is None else charge_percent(volts)
|
|
d["name"] = names.get(s.addr, "")
|
|
d["age_s"] = max(0, int(now - s.last_seen)) if s.last_seen else None
|
|
slots_out.append(d)
|
|
return {
|
|
"n": self.n,
|
|
"pool_full": self.pool_full(),
|
|
"period_s": period_s,
|
|
"report_interval_min": int(report_interval_min),
|
|
"slots": slots_out,
|
|
"ts": now,
|
|
}
|
|
|
|
def save(self) -> None:
|
|
if self.path is None:
|
|
return
|
|
data = {
|
|
"n": self.n,
|
|
"sensors": [s.to_dict() for s in self.by_addr.values() if s.state != "lost"],
|
|
}
|
|
tmp = Path(str(self.path) + ".tmp")
|
|
try:
|
|
tmp.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
|
tmp.replace(self.path)
|
|
except OSError as e:
|
|
log.error("registry save failed: %s", e)
|
|
|
|
def load(self) -> None:
|
|
if self.path is None or not self.path.exists():
|
|
return
|
|
try:
|
|
raw = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as e:
|
|
log.error("registry load failed: %s", e)
|
|
return
|
|
self.slots = [None] * self.n
|
|
self.by_addr = {}
|
|
for item in raw.get("sensors", []):
|
|
try:
|
|
s = Sensor.from_dict(item)
|
|
except (TypeError, ValueError) as e:
|
|
log.warning("skip bad sensor entry: %s", e)
|
|
continue
|
|
if s.slot < 0 or s.slot >= self.n:
|
|
continue
|
|
if self.slots[s.slot] is not None:
|
|
log.warning("duplicate slot %s, skip %s", s.slot, s.addr)
|
|
continue
|
|
if s.addr in self.by_addr:
|
|
continue
|
|
if s.state == "lost":
|
|
continue
|
|
# After reboot treat as stale until a fresh report.
|
|
if s.state == "live":
|
|
s.state = "stale"
|
|
self.slots[s.slot] = s
|
|
self.by_addr[s.addr] = s
|
|
self._assert_invariants()
|
|
log.info("loaded %d sensors from %s", len(self.by_addr), self.path)
|
|
|
|
def _assert_invariants(self) -> None:
|
|
seen_slots = set()
|
|
for addr, s in self.by_addr.items():
|
|
assert s.addr == addr
|
|
assert 0 <= s.slot < self.n
|
|
assert self.slots[s.slot] is s
|
|
assert s.slot not in seen_slots
|
|
seen_slots.add(s.slot)
|
|
for i, s in enumerate(self.slots):
|
|
if s is None:
|
|
continue
|
|
assert s.slot == i
|
|
assert self.by_addr.get(s.addr) is s
|
|
assert self.pool_full() == (None not in self.slots)
|