rich
01c49d7bdd
Repeaters now report their firmware version (REQ_TYPE_GET_OWNER_INFO), checked daily and after a reboot, into meshcore_repeater_info. Firmware changes show as Grafana annotations and in the web UI node list. Repeaters on keymind/Cascade firmware keep 7 days of half-hourly battery voltage and MCU temperature. With history = true and an admin login, the collector fetches enough days via the remote CLI to fill gaps after missed polls, writing them to meshcore_repeater_history.
1407 lines
62 KiB
Python
1407 lines
62 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Poll MeshCore repeaters (status, telemetry, neighbours) and remote companions
|
|
(telemetry, where their owners allow it) over the mesh via a locally attached
|
|
companion radio, plus that companion's own stats, and write it all to InfluxDB.
|
|
Includes a web interface for managing nodes and the InfluxDB target.
|
|
|
|
companion radio (USB / TCP / BLE) -> mesh -> repeater(s)
|
|
|
|
|
this script -> InfluxDB -> Grafana
|
|
|
|
|
web UI (:8080)
|
|
|
|
Usage:
|
|
python meshcore_influx.py -c config.toml # run forever (+ web UI)
|
|
python meshcore_influx.py -c config.toml --once # poll due nodes (cron)
|
|
python meshcore_influx.py -c config.toml --once --dry-run # print, don't write
|
|
python meshcore_influx.py -c config.toml --set-password # set web UI password
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import base64
|
|
import collections
|
|
import getpass
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import signal
|
|
import sys
|
|
import time
|
|
from dataclasses import asdict, dataclass, field, fields
|
|
from typing import Optional
|
|
|
|
try:
|
|
import tomllib
|
|
except ModuleNotFoundError: # Python 3.10
|
|
import tomli as tomllib
|
|
import tomli_w
|
|
|
|
from meshcore import MeshCore, EventType
|
|
from meshcore.packets import BinaryReqType
|
|
from influxdb_client import Point, WritePrecision
|
|
from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
|
|
|
|
log = logging.getLogger("meshcore_influx")
|
|
|
|
# Status response key (meshcore lib) -> InfluxDB field name.
|
|
# Counters are cumulative since the repeater last booted (they live in RAM),
|
|
# so graph them with non_negative_derivative / increase().
|
|
log = logging.getLogger("meshcore_influx")
|
|
|
|
# Status response key (meshcore lib) -> InfluxDB field name.
|
|
# Counters are cumulative since the repeater last booted (they live in RAM),
|
|
# so graph them with non_negative_derivative / increase().
|
|
STATUS_FIELDS = {
|
|
"bat": "battery_mv",
|
|
"tx_queue_len": "tx_queue_len",
|
|
"noise_floor": "noise_floor_dbm",
|
|
"last_rssi": "last_rssi_dbm",
|
|
"last_snr": "last_snr_db",
|
|
"nb_recv": "packets_recv",
|
|
"nb_sent": "packets_sent",
|
|
"airtime": "tx_airtime_s",
|
|
"rx_airtime": "rx_airtime_s",
|
|
"uptime": "uptime_s",
|
|
"sent_flood": "sent_flood",
|
|
"sent_direct": "sent_direct",
|
|
"recv_flood": "recv_flood",
|
|
"recv_direct": "recv_direct",
|
|
"direct_dups": "direct_dups", # uint16 on the wire: wraps at 65536
|
|
"flood_dups": "flood_dups", # uint16 on the wire: wraps at 65536
|
|
"full_evts": "err_flags", # sticky bitmask: 1=queue full, 2=CAD timeout, 4=startRX timeout
|
|
"recv_errors": "recv_errors", # firmware >= v1.12
|
|
}
|
|
FLOAT_STATUS_FIELDS = {"last_snr"}
|
|
|
|
|
|
REQUEST_KINDS = ("status", "telemetry", "neighbours")
|
|
|
|
|
|
@dataclass
|
|
class BatteryPolicy:
|
|
"""Low-battery back-off with hysteresis: enter below low_mv, leave at/above recover_mv."""
|
|
low_mv: int = 0 # 0 = disabled
|
|
recover_mv: int = 0
|
|
interval: int = 3600 # poll interval while in low-power mode
|
|
skip: tuple[str, ...] = ("neighbours",) # requests to skip while in low-power mode
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return self.low_mv > 0
|
|
|
|
|
|
# Keys that may be set in [poll] and overridden per [[repeater]].
|
|
POLICY_KEYS = ("interval_seconds", "low_battery_mv", "recover_battery_mv",
|
|
"low_battery_interval_seconds", "low_battery_skip")
|
|
REPEATER_KEYS = ("name", "password", "pubkey_prefix", "enabled", *REQUEST_KINDS, "history")
|
|
COMPANION_KEYS = ("name", "pubkey_prefix", "enabled", "location")
|
|
NODE_SECTIONS = {"repeater": REPEATER_KEYS, "companion": COMPANION_KEYS} # TOML [[section]] -> keys
|
|
|
|
# Telemetry permission bits (firmware SensorManager.h)
|
|
TELEM_PERM_BASE, TELEM_PERM_LOCATION, TELEM_PERM_ENVIRONMENT = 0x01, 0x02, 0x04
|
|
|
|
|
|
@dataclass
|
|
class NodeCfg:
|
|
"""A polled node: a repeater (login + status/telemetry/neighbours) or a
|
|
remote companion (telemetry only, if its owner allows it)."""
|
|
name: str # contact name as seen by the companion
|
|
kind: str = "repeater" # "repeater" or "companion"
|
|
password: str = "" # repeater: admin or guest password ("" = blank/guest login)
|
|
pubkey_prefix: str = "" # optional, used instead of name if set
|
|
enabled: bool = True
|
|
status: bool = True
|
|
telemetry: bool = True
|
|
neighbours: bool = True
|
|
location: bool = False # companion: also request GPS location telemetry
|
|
history: bool = False # repeater: backfill gaps from its on-device history (admin only)
|
|
interval: int = 900
|
|
battery: BatteryPolicy = field(default_factory=BatteryPolicy)
|
|
|
|
|
|
def _int(v, what: str, minimum: int = 0) -> int:
|
|
try:
|
|
i = int(v)
|
|
except (TypeError, ValueError):
|
|
raise ValueError(f"{what}: {v!r} is not a whole number") from None
|
|
if i < minimum:
|
|
raise ValueError(f"{what}: must be >= {minimum}")
|
|
return i
|
|
|
|
|
|
def build_node(raw: dict, poll: dict, kind: str = "repeater") -> NodeCfg:
|
|
name = str(raw.get("name", "")).strip()
|
|
if not name:
|
|
raise ValueError(f"{kind} name is required")
|
|
unknown = set(raw) - set(NODE_SECTIONS[kind]) - set(POLICY_KEYS)
|
|
if unknown:
|
|
raise ValueError(f"{name}: unknown setting(s) {', '.join(sorted(unknown))}")
|
|
eff = {k: raw.get(k, poll.get(k)) for k in POLICY_KEYS}
|
|
low = _int(eff["low_battery_mv"] or 0, f"{name}: low_battery_mv")
|
|
recover = _int(eff["recover_battery_mv"] or low, f"{name}: recover_battery_mv")
|
|
if recover < low:
|
|
raise ValueError(f"{name}: recover_battery_mv ({recover}) must be >= low_battery_mv ({low})")
|
|
skip = eff["low_battery_skip"] if eff["low_battery_skip"] is not None else ["neighbours"]
|
|
if bad := set(skip) - set(REQUEST_KINDS):
|
|
raise ValueError(f"{name}: low_battery_skip has unknown item(s) {', '.join(sorted(bad))}")
|
|
pol = BatteryPolicy(
|
|
low_mv=low,
|
|
recover_mv=recover,
|
|
interval=_int(eff["low_battery_interval_seconds"] or 3600, f"{name}: low_battery_interval_seconds", 60),
|
|
skip=tuple(skip),
|
|
)
|
|
prefix = str(raw.get("pubkey_prefix", "")).strip().lower()
|
|
if prefix and any(c not in "0123456789abcdef" for c in prefix):
|
|
raise ValueError(f"{name}: pubkey_prefix must be hex")
|
|
if kind == "companion":
|
|
# Companions only answer telemetry requests; there is no login.
|
|
raw = {**raw, "status": False, "telemetry": True, "neighbours": False, "history": False}
|
|
return NodeCfg(
|
|
name=name,
|
|
kind=kind,
|
|
password=str(raw.get("password", "")),
|
|
pubkey_prefix=prefix,
|
|
enabled=bool(raw.get("enabled", True)),
|
|
status=bool(raw.get("status", True)),
|
|
telemetry=bool(raw.get("telemetry", True)),
|
|
neighbours=bool(raw.get("neighbours", True)),
|
|
location=bool(raw.get("location", False)),
|
|
history=bool(raw.get("history", False)),
|
|
interval=_int(eff["interval_seconds"] or 900, f"{name}: interval_seconds", 60),
|
|
battery=pol,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Config:
|
|
connection: dict
|
|
influx: dict
|
|
web: dict
|
|
retries: int = 3
|
|
timeout: float = 0 # 0 = use the radio's suggested timeout
|
|
state_file: str = "state.json"
|
|
local_interval: int = 60 # local companion stats; 0 = off
|
|
location_interval: int = 900 # node locations from the contact list; 0 = off
|
|
advert_hours: int = 0 # automatic self-advert interval; 0 = off
|
|
advert_flood: bool = True # automatic adverts: flood (whole mesh) or zero-hop
|
|
nodes: list[NodeCfg] = field(default_factory=list)
|
|
|
|
@property
|
|
def companion_configured(self) -> bool:
|
|
c = self.connection
|
|
return bool({"serial": c.get("port"), "tcp": c.get("host"), "ble": c.get("address")}.get(c.get("type")))
|
|
|
|
@property
|
|
def influx_configured(self) -> bool:
|
|
return bool(self.influx.get("url") and self.influx.get("bucket"))
|
|
|
|
@classmethod
|
|
def from_raw(cls, raw: dict) -> "Config":
|
|
"""Build and validate. Raises ValueError with a user-readable message."""
|
|
poll = raw.get("poll", {})
|
|
conn = dict(raw.get("connection", {}))
|
|
conn.setdefault("type", "serial")
|
|
if conn["type"] not in ("serial", "tcp", "ble"):
|
|
raise ValueError(f"connection type must be serial, tcp or ble (got {conn['type']!r})")
|
|
nodes = [build_node(r, poll, kind) for kind in NODE_SECTIONS for r in raw.get(kind, [])]
|
|
names = [n.name.lower() for n in nodes]
|
|
if len(names) != len(set(names)):
|
|
raise ValueError("node names must be unique (across repeaters and companions)")
|
|
local_interval = _int(poll.get("local_stats_interval_seconds", 60), "local_stats_interval_seconds")
|
|
if 0 < local_interval < 10:
|
|
raise ValueError("local_stats_interval_seconds: must be 0 (off) or >= 10")
|
|
location_interval = _int(poll.get("location_interval_seconds", 900), "location_interval_seconds")
|
|
if 0 < location_interval < 60:
|
|
raise ValueError("location_interval_seconds: must be 0 (off) or >= 60")
|
|
adv = raw.get("advert", {})
|
|
advert_hours = _int(adv.get("interval_hours", 0), "advert interval_hours")
|
|
if advert_hours and not 3 <= advert_hours <= 168:
|
|
# Same range the repeater firmware enforces for its own flood adverts.
|
|
raise ValueError("advert interval_hours: must be 0 (off) or 3-168")
|
|
return cls(
|
|
connection=conn,
|
|
influx=dict(raw.get("influxdb", {})),
|
|
web=dict(raw.get("web", {})),
|
|
retries=_int(poll.get("retries", 3), "retries", 1),
|
|
timeout=float(poll.get("timeout_seconds", 0) or 0),
|
|
state_file=poll.get("state_file", "state.json"),
|
|
local_interval=local_interval,
|
|
location_interval=location_interval,
|
|
advert_hours=advert_hours,
|
|
advert_flood=bool(adv.get("flood", True)),
|
|
nodes=nodes,
|
|
)
|
|
|
|
|
|
def read_raw(path: str) -> dict:
|
|
with open(path, "rb") as f:
|
|
return tomllib.load(f)
|
|
|
|
|
|
def write_raw(path: str, raw: dict) -> None:
|
|
"""Atomically rewrite the config (comments are not preserved), keeping it private."""
|
|
order = ["connection", "influxdb", "poll", "advert", "web"]
|
|
rest = {k: raw[k] for k in order if k in raw}
|
|
rest.update({k: v for k, v in raw.items() if k not in order and k not in NODE_SECTIONS})
|
|
text = "# Managed by the meshcore-influx web UI; hand edits are picked up automatically.\n\n"
|
|
text += tomli_w.dumps(rest)
|
|
# tomli_w inlines short arrays of tables; keep [[repeater]]/[[companion]] blocks for readability.
|
|
for section in NODE_SECTIONS:
|
|
for r in raw.get(section, []):
|
|
text += f"\n[[{section}]]\n" + tomli_w.dumps(r)
|
|
tmp = path + ".tmp"
|
|
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w") as f:
|
|
f.write(text)
|
|
os.replace(tmp, path)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Per-repeater schedule / low-power state (persisted so --once from cron works)
|
|
# --------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class RepeaterState:
|
|
low_power: Optional[bool] = None # None = no battery reading yet
|
|
last_poll: float = 0.0 # wall-clock epoch of last poll attempt
|
|
battery_mv: Optional[int] = None
|
|
last_success: float = 0.0 # epoch of last fully successful poll
|
|
last_result: dict = field(default_factory=dict) # summary for the web UI
|
|
firmware: str = "" # version string from the owner-info request
|
|
firmware_checked: float = 0.0 # epoch of the last owner-info request
|
|
uptime_s: Optional[int] = None # last reported uptime, to spot reboots (and possible upgrades)
|
|
history_through: float = 0.0 # epoch up to which battery/temperature are written (live or backfilled)
|
|
|
|
def interval(self, rpt: NodeCfg) -> int:
|
|
return rpt.battery.interval if self.low_power else rpt.interval
|
|
|
|
def next_due(self, rpt: NodeCfg) -> float:
|
|
return self.last_poll + self.interval(rpt)
|
|
|
|
def update_battery(self, rpt: NodeCfg, mv: Optional[int]) -> None:
|
|
pol = rpt.battery
|
|
if mv is None or mv <= 0: # no reading (failed poll, or board can't measure)
|
|
return
|
|
self.battery_mv = mv
|
|
if not pol.enabled:
|
|
self.low_power = False
|
|
return
|
|
was = self.low_power
|
|
if was is None:
|
|
# Unknown history (first run): be conservative for solar nodes.
|
|
self.low_power = mv < pol.recover_mv
|
|
elif was and mv >= pol.recover_mv:
|
|
self.low_power = False
|
|
elif not was and mv < pol.low_mv:
|
|
self.low_power = True
|
|
if self.low_power != was:
|
|
log.log(logging.INFO if was is None else logging.WARNING,
|
|
"%s: battery %d mV -> %s mode, polling every %ds", rpt.name, mv,
|
|
"LOW-POWER" if self.low_power else "normal", self.interval(rpt))
|
|
|
|
|
|
def load_state(path: str) -> dict[str, RepeaterState]:
|
|
known = {f.name for f in fields(RepeaterState)}
|
|
try:
|
|
with open(path) as f:
|
|
return {k: RepeaterState(**{kk: vv for kk, vv in v.items() if kk in known})
|
|
for k, v in json.load(f).items()}
|
|
except FileNotFoundError:
|
|
return {}
|
|
except Exception as e: # noqa: BLE001 - corrupt state is not fatal
|
|
log.warning("ignoring unreadable state file %s: %s", path, e)
|
|
return {}
|
|
|
|
|
|
def save_state(path: str, states: dict[str, RepeaterState]) -> None:
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump({k: asdict(v) for k, v in states.items()}, f, indent=2)
|
|
os.replace(tmp, path)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Companion radio
|
|
# --------------------------------------------------------------------------
|
|
|
|
async def connect_companion(cx: dict) -> MeshCore:
|
|
kind = cx.get("type", "serial")
|
|
if kind == "serial":
|
|
mc = await MeshCore.create_serial(cx["port"], int(cx.get("baudrate", 115200)))
|
|
elif kind == "tcp":
|
|
mc = await MeshCore.create_tcp(cx["host"], int(cx.get("port", 5000)))
|
|
elif kind == "ble":
|
|
mc = await MeshCore.create_ble(cx.get("address"), pin=cx.get("pin"))
|
|
else:
|
|
raise ValueError(f"unknown connection type {kind!r}")
|
|
if mc is None:
|
|
raise ConnectionError(f"could not connect to companion radio ({kind})")
|
|
log.info("connected to companion %s", mc.self_info.get("name", "?"))
|
|
return mc
|
|
|
|
|
|
def find_contact(mc: MeshCore, rpt: NodeCfg) -> Optional[dict]:
|
|
if rpt.pubkey_prefix:
|
|
return mc.get_contact_by_key_prefix(rpt.pubkey_prefix)
|
|
return mc.get_contact_by_name(rpt.name)
|
|
|
|
|
|
async def retry(what: str, attempts: int, coro_fn):
|
|
"""Mesh requests are lossy; retry a few times before giving up."""
|
|
for i in range(1, attempts + 1):
|
|
try:
|
|
res = await coro_fn()
|
|
if res is not None:
|
|
return res
|
|
log.warning("%s: no response (attempt %d/%d)", what, i, attempts)
|
|
except Exception as e: # noqa: BLE001 - lib raises assorted errors
|
|
log.warning("%s: %s (attempt %d/%d)", what, e, i, attempts)
|
|
if i < attempts:
|
|
await asyncio.sleep(2 * i)
|
|
return None
|
|
|
|
|
|
async def login(mc: MeshCore, contact: dict, rpt: NodeCfg, cfg: Config) -> Optional[dict]:
|
|
"""Returns the LOGIN_SUCCESS payload (is_admin, server_timestamp, ...), or None."""
|
|
async def _do():
|
|
ev = await mc.commands.send_login_sync(contact, rpt.password, timeout=cfg.timeout)
|
|
if ev is None or ev.type != EventType.LOGIN_SUCCESS:
|
|
return None
|
|
return ev
|
|
ev = await retry(f"{rpt.name} login", cfg.retries, _do)
|
|
if ev is None:
|
|
return None
|
|
log.info("%s: logged in (admin=%s)", rpt.name, ev.payload.get("is_admin"))
|
|
return ev.payload or {}
|
|
|
|
|
|
REQ_TYPE_GET_OWNER_INFO = 0x07 # repeater firmware; not in meshcore's BinaryReqType yet
|
|
|
|
|
|
async def req_owner_info(mc: MeshCore, contact: dict, timeout: float) -> Optional[dict]:
|
|
"""Firmware version, node name and owner info ("ver\\nname\\nowner"). Guests may ask."""
|
|
res = await mc.commands.send(b"\x32" + bytes.fromhex(contact["public_key"]) + bytes([REQ_TYPE_GET_OWNER_INFO]),
|
|
[EventType.MSG_SENT, EventType.ERROR])
|
|
if res is None or res.type == EventType.ERROR:
|
|
return None
|
|
wait = timeout or res.payload["suggested_timeout"] / 800
|
|
ev = await mc.dispatcher.wait_for_event(EventType.BINARY_RESPONSE,
|
|
attribute_filters={"tag": res.payload["expected_ack"].hex()},
|
|
timeout=wait)
|
|
if ev is None:
|
|
return None
|
|
text = bytes.fromhex(ev.payload["data"]).decode("utf-8", "replace").strip("\x00")
|
|
firmware, _, rest = text.partition("\n")
|
|
_, _, owner = rest.partition("\n")
|
|
return {"firmware": firmware.strip(), "owner": owner.strip()} if firmware.strip() else None
|
|
|
|
|
|
class CliSession:
|
|
"""Remote CLI commands to a logged-in repeater. The repeater drops a command whose
|
|
timestamp isn't newer than the last one it saw from us (including the login, which
|
|
the companion stamps with its own clock), and replays its cached reply when the same
|
|
timestamp is resent, so each command gets a fresh, increasing timestamp and retries
|
|
reuse it."""
|
|
|
|
def __init__(self, mc: MeshCore, contact: dict, timeout: float):
|
|
self.mc, self.contact, self.timeout = mc, contact, timeout
|
|
self.last_ts = 0
|
|
|
|
async def start(self) -> None:
|
|
dev = 0
|
|
try:
|
|
ev = await self.mc.commands.get_time()
|
|
if ev is not None and ev.type != EventType.ERROR:
|
|
dev = int(ev.payload.get("time", 0))
|
|
except Exception as e: # noqa: BLE001
|
|
log.debug("companion clock read failed: %s", e)
|
|
self.last_ts = max(int(time.time()), dev)
|
|
|
|
def next_ts(self) -> int:
|
|
self.last_ts = max(int(time.time()), self.last_ts + 1)
|
|
return self.last_ts
|
|
|
|
async def send(self, cmd: str, ts: int) -> Optional[str]:
|
|
res = await self.mc.commands.send_cmd(self.contact, cmd, timestamp=ts)
|
|
if res is None or res.type == EventType.ERROR:
|
|
return None
|
|
deadline = time.monotonic() + (self.timeout or res.payload["suggested_timeout"] / 800)
|
|
prefix = self.contact["public_key"][:12]
|
|
# The reply is queued on the companion like any message; fetch until it turns up.
|
|
while time.monotonic() < deadline:
|
|
while True:
|
|
ev = await self.mc.commands.get_msg()
|
|
if ev is None or ev.type in (EventType.NO_MORE_MSGS, EventType.ERROR):
|
|
break
|
|
p = ev.payload
|
|
if ev.type == EventType.CONTACT_MSG_RECV and p.get("pubkey_prefix") == prefix and p.get("txt_type") == 1:
|
|
return p.get("text", "")
|
|
log.info("discarded a %s message from %s while waiting for a CLI reply",
|
|
ev.type.value, p.get("pubkey_prefix") or p.get("channel_idx", "?"))
|
|
await asyncio.sleep(1)
|
|
return None
|
|
|
|
|
|
# On-device telemetry history (keymind/Cascade repeater firmware): `get telemetry.temp N` and
|
|
# `get telemetry.volt N` return one day (48 half-hour samples) per page as Base64. Page 1 is
|
|
# the newest; there are 7. The history lives in RAM, so a reboot clears it.
|
|
HISTORY_TEMPERATURE, HISTORY_VOLTAGE = 0x11, 0x12
|
|
HISTORY_PAGES = 7
|
|
HISTORY_COMMANDS = {"temperature_c": "get telemetry.temp", "battery_v": "get telemetry.volt"}
|
|
|
|
|
|
def decode_history_page(reply: str) -> tuple[str, list[tuple[int, float]]]:
|
|
"""Decode one history page into (field, [(epoch, value), ...]).
|
|
Missing and out-of-range samples are dropped. Raises ValueError if it isn't a page."""
|
|
text = reply.strip()
|
|
if not text.startswith(">"):
|
|
raise ValueError(text[:80] or "empty reply")
|
|
raw = base64.b64decode(text[1:].strip(), validate=True)
|
|
if len(raw) < 7:
|
|
raise ValueError("page too short")
|
|
kind, start, step, count = raw[0], int.from_bytes(raw[1:5], "little"), raw[5] * 60, raw[6]
|
|
body = raw[7:]
|
|
out: list[tuple[int, float]] = []
|
|
if kind == HISTORY_VOLTAGE:
|
|
if len(body) < count:
|
|
raise ValueError("voltage page truncated")
|
|
# 0 = none, 1 = below 1.88 V, 2..254 = 1.88..4.40 V in 10 mV steps, 255 = above 4.40 V
|
|
for i, code in enumerate(body[:count]):
|
|
if 2 <= code <= 254:
|
|
out.append((start + i * step, round((1880 + (code - 2) * 10) / 1000, 2)))
|
|
return "battery_v", out
|
|
if kind == HISTORY_TEMPERATURE:
|
|
# 2-bit status per sample (1 = value), then 7-bit values (0 = -50 C), both MSB first.
|
|
status_len, data_len = (count * 2 + 7) // 8, (count * 7 + 7) // 8
|
|
if len(body) < status_len + data_len:
|
|
raise ValueError("temperature page truncated")
|
|
status = int.from_bytes(body[:status_len], "big")
|
|
values = int.from_bytes(body[status_len:status_len + data_len], "big")
|
|
for i in range(count):
|
|
if (status >> (status_len * 8 - 2 * (i + 1))) & 0x3 == 1:
|
|
out.append((start + i * step, float(((values >> (data_len * 8 - 7 * (i + 1))) & 0x7F) - 50)))
|
|
return "temperature_c", out
|
|
raise ValueError(f"unknown page type 0x{kind:02x}")
|
|
|
|
|
|
def history_pages_needed(state: "RepeaterState", interval: int, now: float) -> int:
|
|
"""How many days of history cover the gap since the last data we wrote (0 = no gap)."""
|
|
if not state.history_through:
|
|
return 1
|
|
gap = now - state.history_through
|
|
if gap < max(3600, 1.5 * interval): # samples are half-hourly; smaller gaps aren't worth the airtime
|
|
return 0
|
|
return min(HISTORY_PAGES, math.ceil(gap / 86400))
|
|
|
|
|
|
async def fetch_history(cli: CliSession, rpt: NodeCfg, key: str, pages: int, cfg: Config,
|
|
now: float) -> Optional[list[Point]]:
|
|
"""Fetch and decode `pages` days of battery and temperature history. None on failure."""
|
|
pts: list[Point] = []
|
|
for fld, cmd in HISTORY_COMMANDS.items():
|
|
for page in range(1, pages + 1):
|
|
ts = cli.next_ts()
|
|
reply = await retry(f"{rpt.name} {cmd} {page}", cfg.retries, lambda: cli.send(f"{cmd} {page}", ts))
|
|
if reply is None:
|
|
return None
|
|
if "history is empty" in reply:
|
|
break
|
|
try:
|
|
got, samples = decode_history_page(reply)
|
|
except ValueError as e:
|
|
log.warning("%s: %s %d: unexpected reply: %s", rpt.name, cmd, page, e)
|
|
return None
|
|
if got != fld:
|
|
log.warning("%s: %s %d: got %s page", rpt.name, cmd, page, got)
|
|
return None
|
|
for t, v in samples:
|
|
if SANE_EPOCH <= t <= now + 1800:
|
|
pts.append(Point("meshcore_repeater_history").tag("repeater", rpt.name)
|
|
.tag("pubkey", key[:12]).field(fld, v).time(t, WritePrecision.S))
|
|
return pts
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Conversion to InfluxDB points
|
|
# --------------------------------------------------------------------------
|
|
|
|
def status_point(rpt: NodeCfg, key: str, st: dict, ts: int) -> Optional[Point]:
|
|
p = Point("meshcore_repeater_status").tag("repeater", rpt.name).tag("pubkey", key[:12])
|
|
n = 0
|
|
for src, dst in STATUS_FIELDS.items():
|
|
if src in st and st[src] is not None:
|
|
v = float(st[src]) if src in FLOAT_STATUS_FIELDS else int(st[src])
|
|
p = p.field(dst, v)
|
|
n += 1
|
|
if "bat" in st:
|
|
p = p.field("battery_v", st["bat"] / 1000.0)
|
|
return p.time(ts, WritePrecision.S) if n else None
|
|
|
|
|
|
def telemetry_points(rpt: NodeCfg, key: str, lpp: list[dict], ts: int,
|
|
measurement: str = "meshcore_repeater_telemetry", tag: str = "repeater") -> list[Point]:
|
|
"""One point per LPP entry: tags channel + sensor type, value(s) as float fields."""
|
|
pts = []
|
|
for item in lpp or []:
|
|
p = (Point(measurement)
|
|
.tag(tag, rpt.name).tag("pubkey", key[:12])
|
|
.tag("channel", str(item.get("channel")))
|
|
.tag("sensor", str(item.get("type"))))
|
|
val = item.get("value")
|
|
if isinstance(val, dict):
|
|
for k, v in val.items():
|
|
p = p.field(k, float(v))
|
|
elif isinstance(val, (list, tuple)):
|
|
for i, v in enumerate(val):
|
|
p = p.field(f"value_{i}", float(v))
|
|
elif val is not None:
|
|
p = p.field("value", float(val))
|
|
else:
|
|
continue
|
|
pts.append(p.time(ts, WritePrecision.S))
|
|
return pts
|
|
|
|
|
|
def neighbour_points(rpt: NodeCfg, key: str, nb: dict, mc: MeshCore, ts: int) -> list[Point]:
|
|
pts = [Point("meshcore_repeater_neighbours")
|
|
.tag("repeater", rpt.name).tag("pubkey", key[:12])
|
|
.field("count", int(nb.get("neighbours_count", 0)))
|
|
.time(ts, WritePrecision.S)]
|
|
for n in nb.get("neighbours", []):
|
|
known = mc.get_contact_by_key_prefix(n["pubkey"])
|
|
name = known.get("adv_name", "") if known else ""
|
|
pts.append(Point("meshcore_repeater_neighbour")
|
|
.tag("repeater", rpt.name).tag("pubkey", key[:12])
|
|
.tag("neighbour", n["pubkey"])
|
|
.tag("neighbour_name", name or n["pubkey"])
|
|
.field("snr_db", float(n["snr"]))
|
|
.field("secs_ago", int(n["secs_ago"]))
|
|
.field("last_heard", int(ts - n["secs_ago"]))
|
|
.time(ts, WritePrecision.S))
|
|
return pts
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Polling
|
|
# --------------------------------------------------------------------------
|
|
|
|
def battery_from_telemetry(lpp: Optional[list[dict]]) -> Optional[int]:
|
|
"""Fallback battery reading: LPP voltage on the 'self' channel (1)."""
|
|
for item in lpp or []:
|
|
if item.get("type") == "voltage" and item.get("channel") == 1:
|
|
try:
|
|
return int(round(float(item["value"]) * 1000))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return None
|
|
|
|
|
|
|
|
async def poll_repeater(mc: MeshCore, rpt: NodeCfg, cfg: Config, state: RepeaterState) -> list[Point]:
|
|
t0 = time.monotonic()
|
|
ts = int(time.time())
|
|
state.last_poll = time.time()
|
|
pts: list[Point] = []
|
|
low = bool(state.low_power) and rpt.battery.enabled
|
|
wanted = {k: getattr(rpt, k) and not (low and k in rpt.battery.skip) for k in REQUEST_KINDS}
|
|
ok = {k: False for k in wanted}
|
|
battery_mv: Optional[int] = None
|
|
|
|
await mc.ensure_contacts(follow=True)
|
|
contact = find_contact(mc, rpt)
|
|
if contact is None:
|
|
msg = "not in companion's contact list (wait for its advert, or add it)"
|
|
log.error("%s: %s", rpt.name, msg)
|
|
state.last_result = {"ts": ts, "success": False, "error": msg}
|
|
return [Point("meshcore_poll").tag("repeater", rpt.name).field("success", 0)
|
|
.field("error", "contact not found").time(ts, WritePrecision.S)]
|
|
key = contact["public_key"]
|
|
|
|
session = await login(mc, contact, rpt, cfg)
|
|
logged_in = session is not None
|
|
rebooted = False
|
|
history: Optional[str] = None # backfill outcome for the UI: None = not attempted
|
|
if logged_in:
|
|
# Status first: it carries the battery reading.
|
|
if wanted["status"]:
|
|
st = await retry(f"{rpt.name} status", cfg.retries,
|
|
lambda: mc.commands.req_status_sync(contact, timeout=cfg.timeout))
|
|
if st and (p := status_point(rpt, key, st, ts)):
|
|
pts.append(p)
|
|
ok["status"] = True
|
|
battery_mv = st.get("bat")
|
|
if (up := st.get("uptime")) is not None:
|
|
rebooted = state.uptime_s is not None and up < state.uptime_s
|
|
state.uptime_s = int(up)
|
|
|
|
if wanted["telemetry"]:
|
|
lpp = await retry(f"{rpt.name} telemetry", cfg.retries,
|
|
lambda: mc.commands.req_telemetry_sync(contact, timeout=cfg.timeout))
|
|
if lpp is not None:
|
|
pts += telemetry_points(rpt, key, lpp, ts)
|
|
ok["telemetry"] = True
|
|
if not battery_mv:
|
|
battery_mv = battery_from_telemetry(lpp)
|
|
|
|
if wanted["neighbours"]:
|
|
nb = await retry(f"{rpt.name} neighbours", cfg.retries,
|
|
lambda: mc.commands.fetch_all_neighbours(contact, timeout=cfg.timeout))
|
|
if nb is not None:
|
|
pts += neighbour_points(rpt, key, nb, mc, ts)
|
|
ok["neighbours"] = True
|
|
|
|
# Firmware version: daily, and after a reboot (which is how an upgrade shows up).
|
|
if not state.firmware or rebooted or time.time() - state.firmware_checked > 86400:
|
|
info = await retry(f"{rpt.name} owner info", cfg.retries,
|
|
lambda: req_owner_info(mc, contact, cfg.timeout))
|
|
state.firmware_checked = time.time() # also on failure: old firmware may never answer
|
|
if info:
|
|
changed = bool(state.firmware) and info["firmware"] != state.firmware
|
|
if changed:
|
|
log.warning("%s: firmware changed %s -> %s", rpt.name, state.firmware, info["firmware"])
|
|
state.firmware = info["firmware"]
|
|
pts.append(Point("meshcore_repeater_info")
|
|
.tag("repeater", rpt.name).tag("pubkey", key[:12])
|
|
.field("firmware", info["firmware"]).field("owner", info["owner"])
|
|
.field("changed", int(changed))
|
|
.time(ts, WritePrecision.S))
|
|
|
|
# Fill the gap since the last good poll from the repeater's own half-hourly history.
|
|
live = ok["status"] or ok["telemetry"]
|
|
if live and rpt.history and (pages := history_pages_needed(state, state.interval(rpt), ts)):
|
|
clock = session.get("server_timestamp")
|
|
if not session.get("is_admin"):
|
|
history = "needs the admin password"
|
|
elif clock and abs(clock - ts) > 3600:
|
|
history = f"repeater clock is off by {int(clock - ts)}s"
|
|
else:
|
|
cli = CliSession(mc, contact, cfg.timeout)
|
|
await cli.start()
|
|
got = await fetch_history(cli, rpt, key, pages, cfg, ts)
|
|
if got is None:
|
|
history = "failed (firmware without telemetry history?)"
|
|
else:
|
|
pts += got
|
|
history = f"backfilled {len(got)} samples from {pages} day(s)"
|
|
state.history_through = ts
|
|
log.info("%s: history %s", rpt.name, history)
|
|
if live and history is None: # no gap to fill (or backfill is off)
|
|
state.history_through = ts
|
|
|
|
try:
|
|
await mc.commands.send_logout(contact)
|
|
except Exception: # noqa: BLE001 - best effort
|
|
pass
|
|
|
|
state.update_battery(rpt, battery_mv)
|
|
success = logged_in and all(ok[k] for k in ok if wanted[k])
|
|
duration = round(time.monotonic() - t0, 2)
|
|
if success:
|
|
state.last_success = time.time()
|
|
missing = [k for k in ok if wanted[k] and not ok[k]]
|
|
state.last_result = {
|
|
"ts": ts, "success": success, "logged_in": logged_in, "ok": ok, "wanted": wanted,
|
|
"battery_mv": battery_mv, "duration_s": duration, "path_len": contact.get("out_path_len", -1),
|
|
"pubkey": key[:12], "firmware": state.firmware, "history": history,
|
|
"error": "" if success else ("login failed (wrong password or no response)" if not logged_in
|
|
else "no response for: " + ", ".join(missing)),
|
|
}
|
|
|
|
# Poll health + power-management state: for alerting and dashboards.
|
|
p = (Point("meshcore_poll")
|
|
.tag("repeater", rpt.name).tag("pubkey", key[:12])
|
|
.field("success", int(success))
|
|
.field("login_ok", int(logged_in))
|
|
.field("status_ok", int(ok["status"]))
|
|
.field("telemetry_ok", int(ok["telemetry"]))
|
|
.field("neighbours_ok", int(ok["neighbours"]))
|
|
.field("path_len", int(contact.get("out_path_len", -1)))
|
|
.field("duration_s", float(duration))
|
|
.field("low_power_mode", int(bool(state.low_power)))
|
|
.field("interval_s", int(state.interval(rpt))))
|
|
if history is not None:
|
|
p = p.field("history_ok", int(history.startswith("backfilled")))
|
|
if rpt.battery.enabled:
|
|
p = (p.field("low_battery_v", rpt.battery.low_mv / 1000.0)
|
|
.field("recover_battery_v", rpt.battery.recover_mv / 1000.0))
|
|
pts.append(p.time(ts, WritePrecision.S))
|
|
log.info("%s: polled in %.1fs, battery=%s mV, mode=%s, next in %ds %s", rpt.name, duration,
|
|
battery_mv, "low-power" if state.low_power else "normal", state.interval(rpt), ok)
|
|
return pts
|
|
|
|
|
|
async def req_telemetry(mc: MeshCore, contact: dict, want: int, timeout: float) -> Optional[list]:
|
|
"""Telemetry request asking only for the permission categories in `want`.
|
|
The first request byte is an inverse mask the node applies to what it allows."""
|
|
res = await mc.commands.send_binary_req(contact, BinaryReqType.TELEMETRY,
|
|
data=bytes([~want & 0xFF, 0, 0]), timeout=timeout)
|
|
if res is None or res.type == EventType.ERROR:
|
|
return None
|
|
wait = timeout or res.payload["suggested_timeout"] / 800
|
|
ev = await mc.dispatcher.wait_for_event(EventType.TELEMETRY_RESPONSE,
|
|
attribute_filters={"tag": res.payload["expected_ack"].hex()},
|
|
timeout=wait)
|
|
return ev.payload["lpp"] if ev else None
|
|
|
|
|
|
async def poll_companion(mc: MeshCore, node: NodeCfg, cfg: Config, state: RepeaterState) -> list[Point]:
|
|
"""Remote companion: telemetry only, no login. Only answers if its owner allows
|
|
telemetry (for everyone, or for contacts they've flagged) and it knows us."""
|
|
t0 = time.monotonic()
|
|
ts = int(time.time())
|
|
state.last_poll = time.time()
|
|
|
|
await mc.ensure_contacts(follow=True)
|
|
contact = find_contact(mc, node)
|
|
if contact is None:
|
|
msg = "not in companion's contact list (wait for its advert, or add it)"
|
|
log.error("%s: %s", node.name, msg)
|
|
state.last_result = {"ts": ts, "success": False, "error": msg}
|
|
return [Point("meshcore_companion_poll").tag("companion", node.name).field("success", 0)
|
|
.time(ts, WritePrecision.S)]
|
|
key = contact["public_key"]
|
|
|
|
want = TELEM_PERM_BASE | TELEM_PERM_ENVIRONMENT | (TELEM_PERM_LOCATION if node.location else 0)
|
|
lpp = await retry(f"{node.name} telemetry", cfg.retries,
|
|
lambda: req_telemetry(mc, contact, want, cfg.timeout))
|
|
pts: list[Point] = []
|
|
battery_mv = None
|
|
if lpp is not None:
|
|
pts += telemetry_points(node, key, lpp, ts, "meshcore_companion_telemetry", "companion")
|
|
if p := gps_location_point(node, contact, lpp, ts):
|
|
pts.append(p)
|
|
battery_mv = battery_from_telemetry(lpp)
|
|
state.update_battery(node, battery_mv)
|
|
success = lpp is not None
|
|
duration = round(time.monotonic() - t0, 2)
|
|
if success:
|
|
state.last_success = time.time()
|
|
state.last_result = {
|
|
"ts": ts, "success": success, "battery_mv": battery_mv, "duration_s": duration,
|
|
"path_len": contact.get("out_path_len", -1), "pubkey": key[:12],
|
|
"error": "" if success else
|
|
"no telemetry reply (owner may not share telemetry, may not have us as a contact, or is out of range)",
|
|
}
|
|
pts.append(Point("meshcore_companion_poll")
|
|
.tag("companion", node.name).tag("pubkey", key[:12])
|
|
.field("success", int(success))
|
|
.field("path_len", int(contact.get("out_path_len", -1)))
|
|
.field("duration_s", float(duration))
|
|
.field("low_power_mode", int(bool(state.low_power)))
|
|
.field("interval_s", int(state.interval(node)))
|
|
.time(ts, WritePrecision.S))
|
|
log.info("%s: companion telemetry %s in %.1fs, battery=%s mV, next in %ds", node.name,
|
|
"ok" if success else "FAILED", duration, battery_mv, state.interval(node))
|
|
return pts
|
|
|
|
|
|
async def poll_node(mc: MeshCore, node: NodeCfg, cfg: Config, state: RepeaterState) -> list[Point]:
|
|
if node.kind == "companion":
|
|
return await poll_companion(mc, node, cfg, state)
|
|
return await poll_repeater(mc, node, cfg, state)
|
|
|
|
|
|
# Local companion stats (CMD_GET_STATS): response key -> InfluxDB field
|
|
LOCAL_STATS_FIELDS = {
|
|
"battery_mv": "battery_mv", "uptime_secs": "uptime_s", "errors": "err_flags", "queue_len": "tx_queue_len",
|
|
"noise_floor": "noise_floor_dbm", "last_rssi": "last_rssi_dbm", "last_snr": "last_snr_db",
|
|
"tx_air_secs": "tx_airtime_s", "rx_air_secs": "rx_airtime_s",
|
|
"recv": "packets_recv", "sent": "packets_sent", "flood_tx": "sent_flood", "direct_tx": "sent_direct",
|
|
"flood_rx": "recv_flood", "direct_rx": "recv_direct", "recv_errors": "recv_errors",
|
|
}
|
|
CONTACT_TYPES = {1: "chat", 2: "repeater", 3: "room", 4: "sensor"}
|
|
|
|
|
|
async def collect_local(mc: MeshCore) -> tuple[dict, Optional[list], list[str]]:
|
|
"""Stats from the attached companion over USB/TCP/BLE: no mesh airtime.
|
|
Returns (stats, self-telemetry LPP, unsupported request names)."""
|
|
stats: dict = {}
|
|
unsupported: list[str] = []
|
|
for name, fn in (("core", mc.commands.get_stats_core), ("radio", mc.commands.get_stats_radio),
|
|
("packets", mc.commands.get_stats_packets)):
|
|
try:
|
|
ev = await fn()
|
|
except Exception as e: # noqa: BLE001 - older firmware may just not answer
|
|
log.debug("local stats %s failed: %s", name, e)
|
|
ev = None
|
|
if ev is None or ev.type == EventType.ERROR:
|
|
unsupported.append(name)
|
|
else:
|
|
stats.update({k: v for k, v in ev.payload.items() if v is not None})
|
|
lpp = None
|
|
try:
|
|
ev = await mc.commands.get_self_telemetry()
|
|
if ev is not None and ev.type != EventType.ERROR:
|
|
lpp = ev.payload.get("lpp")
|
|
except Exception as e: # noqa: BLE001
|
|
log.debug("self telemetry failed: %s", e)
|
|
if lpp is None:
|
|
unsupported.append("telemetry")
|
|
# What the companion hears: contact counts by type, and how many advertised in the last hour.
|
|
now = time.time()
|
|
contacts = list((mc.contacts or {}).values())
|
|
stats["contacts"] = len(contacts)
|
|
for t, label in CONTACT_TYPES.items():
|
|
stats[f"contacts_{label}"] = sum(1 for c in contacts if c.get("type") == t)
|
|
stats["heard_1h"] = sum(1 for c in contacts if now - c.get("last_advert", 0) < 3600)
|
|
if "battery_mv" not in stats and (mv := battery_from_telemetry(lpp)):
|
|
stats["battery_mv"] = mv
|
|
return stats, lpp, unsupported
|
|
|
|
|
|
def local_points(name: str, key: str, stats: dict, lpp: Optional[list], ts: int) -> list[Point]:
|
|
p = Point("meshcore_local_stats").tag("companion", name).tag("pubkey", key[:12])
|
|
for src, dst in LOCAL_STATS_FIELDS.items():
|
|
if src in stats:
|
|
v = stats[src]
|
|
p = p.field(dst, float(v) if src == "last_snr" else int(v))
|
|
if "battery_mv" in stats:
|
|
p = p.field("battery_v", stats["battery_mv"] / 1000.0)
|
|
for k in ("contacts", "heard_1h", *(f"contacts_{t}" for t in CONTACT_TYPES.values())):
|
|
p = p.field(k, int(stats[k]))
|
|
pts = [p.time(ts, WritePrecision.S)]
|
|
pts += telemetry_points(NodeCfg(name=name), key, lpp or [], ts, "meshcore_local_telemetry", "companion")
|
|
return pts
|
|
|
|
|
|
# A node's position can come from its adverts or from GPS telemetry. Each is written to
|
|
# meshcore_node_location with the time the position was *observed*, so the newest point
|
|
# per node is the most current position whichever source it came from.
|
|
SANE_EPOCH = 1672531200 # 2023-01-01: advert timestamps before this are a node's unset clock
|
|
|
|
|
|
def location_point(name: str, contact: dict, lat: float, lon: float, observed: float,
|
|
source: str, polled: bool) -> Point:
|
|
key = contact.get("public_key", "")
|
|
return (Point("meshcore_node_location")
|
|
.tag("name", name).tag("pubkey", key[:12])
|
|
.tag("type", CONTACT_TYPES.get(contact.get("type"), "other"))
|
|
.tag("source", source)
|
|
.field("lat", float(lat)).field("lon", float(lon))
|
|
.field("hops", int(contact.get("out_path_len", -1)))
|
|
.field("last_advert", int(contact.get("last_advert", 0)))
|
|
.field("seen", int(observed))
|
|
.field("from_gps", int(source == "gps"))
|
|
.field("polled", int(polled))
|
|
.time(int(observed), WritePrecision.S))
|
|
|
|
|
|
def gps_location_point(node: NodeCfg, contact: dict, lpp: Optional[list], ts: int) -> Optional[Point]:
|
|
"""Position from a companion's shared GPS telemetry, if it sent one."""
|
|
for item in lpp or []:
|
|
v = item.get("value")
|
|
if item.get("type") == "gps" and isinstance(v, dict):
|
|
lat, lon = v.get("latitude"), v.get("longitude")
|
|
if lat and lon:
|
|
return location_point(node.name, contact, lat, lon, ts, "gps", True)
|
|
return None
|
|
|
|
|
|
def advert_location_points(mc: MeshCore, nodes: list[NodeCfg], seen: dict, now: float) -> list[Point]:
|
|
"""Positions from adverts in the companion's contact list (public broadcasts), plus
|
|
the companion's own advertised position. `seen` maps pubkey -> (last_advert, observed)
|
|
and is updated in place: a changed advert timestamp means a new advert just arrived."""
|
|
polled = {}
|
|
for n in nodes:
|
|
c = find_contact(mc, n)
|
|
if c:
|
|
polled[c["public_key"]] = n.name
|
|
pts = []
|
|
for c in (mc.contacts or {}).values():
|
|
lat, lon = c.get("adv_lat", 0), c.get("adv_lon", 0)
|
|
if not lat and not lon:
|
|
continue
|
|
key, adv = c.get("public_key", ""), int(c.get("last_advert", 0))
|
|
prev = seen.get(key)
|
|
if prev and prev[0] == adv:
|
|
observed = prev[1] # same advert as last scan
|
|
elif prev is None and SANE_EPOCH < adv <= now + 60:
|
|
observed = adv # first scan: trust a plausible advert clock
|
|
else:
|
|
observed = now # new advert since last scan, or bogus clock
|
|
seen[key] = (adv, observed)
|
|
name = polled.get(key) or c.get("adv_name") or key[:12]
|
|
pts.append(location_point(name, c, lat, lon, observed, "advert", key in polled))
|
|
info = mc.self_info or {}
|
|
if info.get("adv_lat") or info.get("adv_lon"):
|
|
me = {"public_key": info.get("public_key", ""), "out_path_len": 0, "last_advert": int(now)}
|
|
p = location_point(info.get("name") or "companion", me, info["adv_lat"], info["adv_lon"],
|
|
now, "advert", False)
|
|
pts.append(p.tag("type", "self"))
|
|
return pts
|
|
|
|
|
|
def _influx_client(ic: dict) -> InfluxDBClientAsync:
|
|
return InfluxDBClientAsync(url=ic["url"], token=ic.get("token", ""), org=ic.get("org") or "-",
|
|
timeout=10_000)
|
|
|
|
|
|
async def influx_test(ic: dict) -> dict:
|
|
"""Check reachability and write access by writing one marker point."""
|
|
if not (ic.get("url") and ic.get("bucket")):
|
|
return {"ok": False, "error": "URL and bucket are required"}
|
|
try:
|
|
async with _influx_client(ic) as client:
|
|
if not await client.ping():
|
|
return {"ok": False, "error": f"no response from {ic['url']}"}
|
|
try:
|
|
version = await client.version()
|
|
except Exception: # noqa: BLE001 - 1.x may not report it
|
|
version = "unknown"
|
|
await client.write_api().write(
|
|
bucket=ic["bucket"],
|
|
record=Point("meshcore_collector").field("test_write", 1).time(int(time.time()), WritePrecision.S))
|
|
return {"ok": True, "version": version}
|
|
except Exception as e: # noqa: BLE001 - report anything to the UI
|
|
return {"ok": False, "error": str(e).strip().splitlines()[0][:300]}
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Web UI password (scrypt, stdlib only)
|
|
# --------------------------------------------------------------------------
|
|
|
|
def hash_password(pw: str) -> str:
|
|
import hashlib
|
|
salt = os.urandom(16)
|
|
h = hashlib.scrypt(pw.encode(), salt=salt, n=2**14, r=8, p=1)
|
|
return f"scrypt$16384$8$1${salt.hex()}${h.hex()}"
|
|
|
|
|
|
def verify_password(pw: str, stored: str) -> bool:
|
|
import hashlib
|
|
import hmac
|
|
try:
|
|
_, n, r, p, salt, h = stored.split("$")
|
|
calc = hashlib.scrypt(pw.encode(), salt=bytes.fromhex(salt), n=int(n), r=int(r), p=int(p))
|
|
return hmac.compare_digest(calc.hex(), h)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
class LogBuffer(logging.Handler):
|
|
"""Keeps recent log lines for the web UI."""
|
|
|
|
def __init__(self, size: int = 500):
|
|
super().__init__()
|
|
self.lines: collections.deque = collections.deque(maxlen=size)
|
|
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
try:
|
|
self.lines.append({"ts": record.created, "level": record.levelname, "msg": record.getMessage()})
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Collector: owns config, radio connection and schedule; the web UI drives it
|
|
# --------------------------------------------------------------------------
|
|
|
|
class Collector:
|
|
RECONNECT_DELAY = 60
|
|
|
|
def __init__(self, cfg_path: str, dry_run: bool = False):
|
|
self.cfg_path = cfg_path
|
|
self.dry_run = dry_run
|
|
self.raw = read_raw(cfg_path)
|
|
self.cfg = Config.from_raw(self.raw)
|
|
self.states = load_state(self.cfg.state_file)
|
|
for r in self.cfg.nodes:
|
|
self.states.setdefault(r.name, RepeaterState())
|
|
self.mc: Optional[MeshCore] = None
|
|
self.companion = {"connected": False, "name": None, "error": None, "since": None}
|
|
self.influx_status: dict = {"ok": None, "error": None, "last_write": None, "points": 0}
|
|
self.local: dict = {"ts": None, "stats": {}, "unsupported": [], "error": None}
|
|
self._local_last = 0.0
|
|
self._location_last = 0.0
|
|
self._advert_seen: dict = {} # pubkey -> (last_advert, observed time)
|
|
self.advert: dict = {"ts": None, "flood": None, "ok": None, "error": None, "auto": False, "pending": None}
|
|
self._advert_file = os.path.join(os.path.dirname(self.cfg.state_file) or ".", "advert.json")
|
|
self._advert_last_auto = self._load_advert_time()
|
|
self.polling: Optional[str] = None
|
|
self.force: set[str] = set()
|
|
self.wake = asyncio.Event()
|
|
self.stopping = False
|
|
self._reconnect = False
|
|
self._next_connect = 0.0
|
|
self._mtime = self._config_mtime()
|
|
|
|
# ---- config changes (from the web UI) ----
|
|
async def update_config(self, raw: dict, renamed: Optional[dict] = None) -> None:
|
|
"""Validate, persist and apply a new raw config. Raises ValueError if invalid."""
|
|
self.reload_if_changed() # don't clobber edits made on disk meanwhile
|
|
cfg = Config.from_raw(raw)
|
|
write_raw(self.cfg_path, raw)
|
|
self._mtime = self._config_mtime()
|
|
self._apply(raw, cfg, renamed)
|
|
|
|
def reload_if_changed(self) -> None:
|
|
"""Pick up edits to the config file made outside the web UI (e.g. --set-password)."""
|
|
m = self._config_mtime()
|
|
if m is None or m == self._mtime:
|
|
return
|
|
self._mtime = m
|
|
try:
|
|
raw = read_raw(self.cfg_path)
|
|
cfg = Config.from_raw(raw)
|
|
except Exception as e: # noqa: BLE001 - keep running on the old config
|
|
log.error("config file changed but is invalid, keeping current settings: %s", e)
|
|
return
|
|
log.info("config file changed on disk; reloaded")
|
|
self._apply(raw, cfg)
|
|
|
|
def _config_mtime(self) -> Optional[int]:
|
|
try:
|
|
return os.stat(self.cfg_path).st_mtime_ns
|
|
except OSError:
|
|
return None
|
|
|
|
def _apply(self, raw: dict, cfg: Config, renamed: Optional[dict] = None) -> None:
|
|
old = self.cfg
|
|
self.raw, self.cfg = raw, cfg
|
|
for old_name, new_name in (renamed or {}).items():
|
|
if old_name in self.states:
|
|
self.states[new_name] = self.states.pop(old_name)
|
|
names = {r.name for r in cfg.nodes}
|
|
for n in list(self.states):
|
|
if n not in names:
|
|
del self.states[n]
|
|
for n in names:
|
|
self.states.setdefault(n, RepeaterState())
|
|
if cfg.connection != old.connection:
|
|
self._reconnect = True
|
|
self._next_connect = 0.0
|
|
self._save_state()
|
|
log.info("configuration updated")
|
|
self.wake.set()
|
|
|
|
def request_poll(self, name: str) -> None:
|
|
self.force.add(name)
|
|
self.wake.set()
|
|
|
|
def stop(self) -> None:
|
|
self.stopping = True
|
|
self.wake.set()
|
|
|
|
# ---- self adverts (so other companions can add us) ----
|
|
def request_advert(self, flood: bool) -> None:
|
|
self.advert["pending"] = bool(flood)
|
|
self.wake.set()
|
|
|
|
def _load_advert_time(self) -> float:
|
|
try:
|
|
with open(self._advert_file) as f:
|
|
return float(json.load(f).get("last_auto", 0))
|
|
except (OSError, ValueError):
|
|
return 0.0
|
|
|
|
def advert_next_due(self) -> Optional[float]:
|
|
if not self.cfg.advert_hours or self.mc is None:
|
|
return None
|
|
return self._advert_last_auto + self.cfg.advert_hours * 3600
|
|
|
|
async def maybe_advert(self) -> None:
|
|
"""Send a requested advert, or an automatic one when due."""
|
|
if self.mc is None:
|
|
return
|
|
flood = self.advert["pending"]
|
|
auto = False
|
|
if flood is None:
|
|
nd = self.advert_next_due()
|
|
if nd is None or nd > time.time():
|
|
return
|
|
flood, auto = self.cfg.advert_flood, True
|
|
self.advert["pending"] = None
|
|
try:
|
|
ev = await self.mc.commands.send_advert(flood=flood)
|
|
ok = ev is not None and ev.type != EventType.ERROR
|
|
err = None if ok else str((ev.payload if ev else None) or "no response")
|
|
except Exception as e: # noqa: BLE001
|
|
ok, err = False, str(e) or type(e).__name__
|
|
self.advert.update(ts=time.time(), flood=flood, ok=ok, error=err, auto=auto)
|
|
log.log(logging.INFO if ok else logging.ERROR, "%s %s advert %s%s",
|
|
"automatic" if auto else "manual", "flood" if flood else "zero-hop",
|
|
"sent" if ok else "FAILED", f": {err}" if err else "")
|
|
# Restart the automatic timer after an automatic attempt (even a failed one, so it
|
|
# can't retry in a tight loop) or after a successful manual advert of the same kind.
|
|
if auto or (ok and flood == self.cfg.advert_flood):
|
|
self._advert_last_auto = time.time()
|
|
if not self.dry_run:
|
|
try:
|
|
with open(self._advert_file, "w") as f:
|
|
json.dump({"last_auto": self._advert_last_auto}, f)
|
|
except OSError as e:
|
|
log.error("could not save advert time: %s", e)
|
|
|
|
# ---- radio ----
|
|
async def _disconnect(self) -> None:
|
|
if self.mc is not None:
|
|
try:
|
|
await self.mc.disconnect()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
self.mc = None
|
|
self.companion.update(connected=False, since=None)
|
|
|
|
async def _ensure_companion(self) -> bool:
|
|
if self._reconnect:
|
|
self._reconnect = False
|
|
await self._disconnect()
|
|
self.companion.update(name=None, error=None, uri=None)
|
|
if self.mc is not None and self.mc.is_connected:
|
|
return True
|
|
if self.mc is not None:
|
|
log.warning("companion radio disconnected")
|
|
await self._disconnect()
|
|
if not self.cfg.companion_configured:
|
|
self.companion.update(error="not configured")
|
|
return False
|
|
if time.time() < self._next_connect:
|
|
return False
|
|
try:
|
|
self.mc = await connect_companion(self.cfg.connection)
|
|
self.mc.auto_update_contacts = True
|
|
await self.mc.ensure_contacts()
|
|
self.companion.update(connected=True, name=self.mc.self_info.get("name"),
|
|
error=None, since=time.time(), uri=None)
|
|
try: # our own contact card, so people can add us by QR/link
|
|
ev = await self.mc.commands.export_contact()
|
|
if ev is not None and ev.type == EventType.CONTACT_URI:
|
|
self.companion["uri"] = ev.payload.get("uri")
|
|
except Exception as e: # noqa: BLE001 - optional
|
|
log.debug("export_contact failed: %s", e)
|
|
return True
|
|
except Exception as e: # noqa: BLE001
|
|
self.mc = None
|
|
self._next_connect = time.time() + self.RECONNECT_DELAY
|
|
self.companion.update(connected=False, error=str(e) or type(e).__name__)
|
|
log.error("companion connection failed: %s (retrying in %ds)", e, self.RECONNECT_DELAY)
|
|
return False
|
|
|
|
def contacts(self) -> list[dict]:
|
|
if self.mc is None or not self.mc.contacts:
|
|
return []
|
|
return [{"name": c.get("adv_name", ""), "pubkey": c.get("public_key", ""), "type": c.get("type"),
|
|
"path_len": c.get("out_path_len", -1), "last_advert": c.get("last_advert", 0)}
|
|
for c in self.mc.contacts.values()]
|
|
|
|
# ---- influx ----
|
|
async def _write(self, pts: list[Point]) -> None:
|
|
if not pts:
|
|
return
|
|
if self.dry_run:
|
|
for p in pts:
|
|
print(p.to_line_protocol())
|
|
return
|
|
if not self.cfg.influx_configured:
|
|
log.warning("InfluxDB not configured; dropping %d points", len(pts))
|
|
self.influx_status.update(ok=False, error="not configured")
|
|
return
|
|
try:
|
|
async with _influx_client(self.cfg.influx) as client:
|
|
await client.write_api().write(bucket=self.cfg.influx["bucket"], record=pts)
|
|
self.influx_status.update(ok=True, error=None, last_write=time.time(),
|
|
points=self.influx_status["points"] + len(pts))
|
|
log.info("wrote %d points to InfluxDB", len(pts))
|
|
except Exception as e: # noqa: BLE001 - keep running
|
|
msg = str(e).strip().splitlines()[0][:300] if str(e).strip() else type(e).__name__
|
|
self.influx_status.update(ok=False, error=msg)
|
|
log.error("InfluxDB write failed: %s", msg)
|
|
|
|
def _save_state(self) -> None:
|
|
if self.dry_run:
|
|
return
|
|
try:
|
|
save_state(self.cfg.state_file, self.states)
|
|
except OSError as e:
|
|
log.error("could not save state: %s", e)
|
|
|
|
# ---- scheduling ----
|
|
def local_next_due(self) -> Optional[float]:
|
|
if not self.cfg.local_interval or self.mc is None:
|
|
return None
|
|
return self._local_last + self.cfg.local_interval
|
|
|
|
async def poll_local(self, force: bool = False) -> None:
|
|
"""Stats from the attached companion itself (no mesh airtime)."""
|
|
nd = self.local_next_due()
|
|
if nd is None or (not force and nd > time.time()):
|
|
return
|
|
self._local_last = time.time()
|
|
ts = int(self._local_last)
|
|
try:
|
|
stats, lpp, unsupported = await collect_local(self.mc)
|
|
except Exception as e: # noqa: BLE001 - e.g. radio unplugged
|
|
self.local.update(error=str(e) or type(e).__name__)
|
|
log.error("local companion stats failed: %s", e)
|
|
await self._disconnect()
|
|
self._next_connect = time.time() + self.RECONNECT_DELAY
|
|
return
|
|
if set(unsupported) != set(self.local["unsupported"]) and unsupported:
|
|
log.info("local companion: not supported by its firmware: %s", ", ".join(unsupported))
|
|
self.local.update(ts=ts, stats=stats, unsupported=unsupported, error=None)
|
|
info = self.mc.self_info or {}
|
|
await self._write(local_points(info.get("name") or "companion", info.get("public_key", ""),
|
|
stats, lpp, ts))
|
|
|
|
def location_next_due(self) -> Optional[float]:
|
|
if not self.cfg.location_interval or self.mc is None:
|
|
return None
|
|
return self._location_last + self.cfg.location_interval
|
|
|
|
async def poll_locations(self, force: bool = False) -> None:
|
|
nd = self.location_next_due()
|
|
if nd is None or (not force and nd > time.time()):
|
|
return
|
|
self._location_last = time.time()
|
|
pts = advert_location_points(self.mc, self.cfg.nodes, self._advert_seen, self._location_last)
|
|
log.debug("writing %d node locations", len(pts))
|
|
await self._write(pts)
|
|
|
|
def due(self, now: float, force_all: bool = False) -> list[NodeCfg]:
|
|
return [r for r in self.cfg.nodes
|
|
if r.enabled and (force_all or r.name in self.force
|
|
or self.states[r.name].next_due(r) <= now)]
|
|
|
|
async def poll_due(self, force_all: bool = False) -> bool:
|
|
"""Poll the local companion and every node that is due.
|
|
Returns False if the radio was unavailable."""
|
|
due = self.due(time.time(), force_all)
|
|
now = time.time()
|
|
local_due = (self.cfg.local_interval > 0 and
|
|
(force_all or self._local_last + self.cfg.local_interval <= now))
|
|
loc_due = (self.cfg.location_interval > 0 and
|
|
(force_all or self._location_last + self.cfg.location_interval <= now))
|
|
if not due and not local_due and not loc_due:
|
|
return True
|
|
if not await self._ensure_companion():
|
|
return False
|
|
await self.poll_local(force=force_all)
|
|
if self.mc is not None:
|
|
await self.poll_locations(force=force_all)
|
|
if self.mc is None:
|
|
return False
|
|
for rpt in due:
|
|
if self.stopping or self._reconnect:
|
|
break
|
|
if rpt not in self.cfg.nodes: # removed/edited while we were busy
|
|
continue
|
|
self.force.discard(rpt.name)
|
|
state = self.states[rpt.name]
|
|
self.polling = rpt.name
|
|
try:
|
|
pts = await poll_node(self.mc, rpt, self.cfg, state)
|
|
except Exception as e: # noqa: BLE001 - radio trouble: reconnect later
|
|
log.error("%s: poll failed: %s", rpt.name, e)
|
|
state.last_result = {"ts": int(time.time()), "success": False, "error": f"poll failed: {e}"}
|
|
state.last_poll = time.time() - state.interval(rpt) + self.RECONNECT_DELAY
|
|
await self._disconnect()
|
|
self.companion["error"] = str(e)
|
|
self._next_connect = time.time() + self.RECONNECT_DELAY
|
|
return False
|
|
finally:
|
|
self.polling = None
|
|
await self._write(pts)
|
|
self._save_state()
|
|
return True
|
|
|
|
async def run(self) -> None:
|
|
while not self.stopping:
|
|
self.wake.clear()
|
|
self.reload_if_changed()
|
|
await self._ensure_companion()
|
|
await self.maybe_advert()
|
|
await self.poll_due()
|
|
if self.stopping:
|
|
break
|
|
now = time.time()
|
|
enabled = [r for r in self.cfg.nodes if r.enabled]
|
|
wake_at = min((self.states[r.name].next_due(r) for r in enabled), default=now + 3600)
|
|
for nd in (self.local_next_due(), self.location_next_due(), self.advert_next_due()):
|
|
if nd is not None:
|
|
wake_at = min(wake_at, nd)
|
|
if self.mc is None and self.cfg.companion_configured:
|
|
wake_at = max(wake_at, self._next_connect) if enabled else self._next_connect
|
|
try:
|
|
await asyncio.wait_for(self.wake.wait(), timeout=max(1.0, wake_at - now))
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
await self._disconnect()
|
|
|
|
|
|
async def main() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("-c", "--config", default="config.toml")
|
|
ap.add_argument("--once", action="store_true",
|
|
help="poll nodes that are due, then exit (for cron/systemd timers); no web UI")
|
|
ap.add_argument("--force", action="store_true", help="with --once: poll every node even if not due")
|
|
ap.add_argument("--dry-run", action="store_true", help="print line protocol instead of writing")
|
|
ap.add_argument("--set-password", action="store_true",
|
|
help="set the web UI password (prompts, or reads one line from stdin)")
|
|
ap.add_argument("--no-web", action="store_true", help="don't start the web UI")
|
|
ap.add_argument("-v", "--verbose", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
logbuf = LogBuffer()
|
|
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
|
logbuf.setLevel(logging.INFO)
|
|
logging.getLogger().addHandler(logbuf)
|
|
|
|
if args.set_password:
|
|
pw = getpass.getpass("New web UI password: ") if sys.stdin.isatty() else sys.stdin.readline().rstrip("\n")
|
|
if len(pw) < 8:
|
|
print("password must be at least 8 characters", file=sys.stderr)
|
|
return 1
|
|
raw = read_raw(args.config)
|
|
raw.setdefault("web", {})["password_hash"] = hash_password(pw)
|
|
write_raw(args.config, raw)
|
|
print("web UI password updated")
|
|
return 0
|
|
|
|
try:
|
|
col = Collector(args.config, dry_run=args.dry_run)
|
|
except (OSError, ValueError, tomllib.TOMLDecodeError) as e:
|
|
log.error("cannot load config %s: %s", args.config, e)
|
|
return 1
|
|
|
|
if args.once:
|
|
if not col.due(time.time(), args.force):
|
|
log.info("nothing due yet (use --force to poll anyway)")
|
|
ok = await col.poll_due(force_all=args.force)
|
|
if ok and col.cfg.advert_hours and await col._ensure_companion():
|
|
await col.maybe_advert()
|
|
await col._disconnect()
|
|
return 0 if ok else 1
|
|
|
|
runner = None
|
|
if not args.no_web and col.cfg.web.get("enabled", True):
|
|
import web
|
|
runner = await web.start(col, logbuf)
|
|
|
|
task = asyncio.create_task(col.run())
|
|
|
|
def on_signal():
|
|
col.stop()
|
|
# Give an in-flight poll a moment to finish, then cancel it.
|
|
asyncio.get_running_loop().call_later(5, task.cancel)
|
|
|
|
loop = asyncio.get_running_loop()
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
loop.add_signal_handler(sig, on_signal)
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
log.info("stopped while polling")
|
|
await col._disconnect()
|
|
finally:
|
|
if runner is not None:
|
|
await runner.cleanup()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(asyncio.run(main()))
|