#!/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)