Add MeshCore repeater telemetry collector for InfluxDB with Grafana dashboard
Polls MeshCore repeaters over the mesh via a local companion radio (serial/TCP/BLE): status counters, CayenneLPP telemetry and the neighbour table, written to InfluxDB along with per-poll health. Includes a low-battery back-off for solar nodes (hysteresis between low and recover thresholds, slower interval, optional request skips), persisted state so --once works from cron, and an InfluxQL Grafana dashboard. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
richandClaude Opus 5.5 committed
commit
f2c285f939
6 files changed
+2840
No files matched your search
@@ -0,0 +1,4 @@
|
||||
config.toml
|
||||
state.json
|
||||
.venv/
|
||||
__pycache__/
|
||||
@@ -0,0 +1,183 @@
|
||||
# meshcore-repeater-stat-influx
|
||||
|
||||
Polls MeshCore repeaters **over the mesh** through a companion radio attached to
|
||||
this machine (USB serial, TCP/WiFi or BLE), and writes the results to InfluxDB
|
||||
for Grafana.
|
||||
|
||||
```
|
||||
[this script] --USB/TCP/BLE--> [companion radio] ~~LoRa mesh~~> [repeater(s)]
|
||||
|
|
||||
+--> InfluxDB --> Grafana
|
||||
```
|
||||
|
||||
Each poll cycle, for each repeater: **login → status → telemetry → neighbours → logout**.
|
||||
|
||||
## Setup
|
||||
|
||||
```sh
|
||||
python3 -m venv .venv && . .venv/bin/activate
|
||||
pip install -r requirements.txt
|
||||
cp config.example.toml config.toml # edit it
|
||||
python meshcore_influx.py -c config.toml --once --dry-run -v # test, prints line protocol
|
||||
python meshcore_influx.py -c config.toml # run forever
|
||||
```
|
||||
|
||||
Requires Python 3.11+. The companion must run **companion radio firmware**, and the
|
||||
repeater must already be in its contact list (it gets added when the companion hears
|
||||
the repeater's advert). A guest password gets you status, base telemetry and the
|
||||
neighbour table. Use the admin password if you want the extra sensor channels.
|
||||
|
||||
> While this script is connected over USB/BLE, the phone app can't use the same
|
||||
> companion at the same time.
|
||||
|
||||
## What gets collected
|
||||
|
||||
| Measurement | Source | Contents |
|
||||
|---|---|---|
|
||||
| `meshcore_repeater_status` | `REQ_TYPE_GET_STATUS` | battery mV/V, TX queue, noise floor, last RSSI/SNR, packets rx/tx, flood/direct rx/tx, dup counts, TX/RX airtime, uptime, recv errors, error flags |
|
||||
| `meshcore_repeater_telemetry` | `REQ_TYPE_GET_TELEMETRY_DATA` (CayenneLPP) | one point per sensor, tagged `channel` + `sensor` (voltage, temperature, humidity, gps…). Usually at least battery voltage and MCU temperature |
|
||||
| `meshcore_repeater_neighbours` | `REQ_TYPE_GET_NEIGHBOURS` | `count` |
|
||||
| `meshcore_repeater_neighbour` | same | per zero-hop neighbouring repeater: `snr_db`, `secs_ago`, `last_heard` |
|
||||
| `meshcore_poll` | this script | per-request success flags, path length, poll duration, `low_power_mode`, `interval_s`, configured battery thresholds |
|
||||
|
||||
All points are tagged `repeater` (configured name) and `pubkey` (12 hex chars).
|
||||
|
||||
## Does the repeater store telemetry history? No.
|
||||
|
||||
I checked this against the firmware source (`examples/simple_repeater`, v1.17.x):
|
||||
|
||||
- **Nothing is buffered for later.** Every reply is built from live values when the
|
||||
request arrives. The min/max/avg history request (`REQ_TYPE_GET_AVG_MIN_MAX`, backed by
|
||||
`TimeSeriesData`) is only in the **sensor** firmware (`simple_sensor`). The
|
||||
repeater firmware doesn't implement it.
|
||||
- **The status counters are cumulative since boot**, and they live in RAM. A reboot,
|
||||
power loss, or `clear stats` resets them to zero.
|
||||
- **Gauges are instantaneous.** Battery, noise floor, TX queue, last RSSI/SNR and
|
||||
temperature are only valid when sampled. Anything between two polls is never seen.
|
||||
- **The neighbour table** holds the last-heard time and SNR for up to `MAX_NEIGHBOURS`
|
||||
(50 on nearly every board) zero-hop repeaters. It lives in RAM, and the oldest entry
|
||||
gets overwritten.
|
||||
- **The on-flash packet log** (`log start`) can only be dumped over the local serial
|
||||
console (`sender_timestamp == 0`), not over the mesh.
|
||||
|
||||
### So how often should you poll?
|
||||
|
||||
Because the counters are cumulative, a missed poll costs resolution, not data. The
|
||||
next successful poll still gives correct totals and rates. What you lose between polls:
|
||||
|
||||
1. **Gauges** (battery, noise floor, temperature): you only see what you sample.
|
||||
Battery and temperature change slowly, so 15 min is plenty.
|
||||
2. **Activity across a reboot**: everything since the last poll before the reboot.
|
||||
Watch `uptime_s` dropping to spot reboots.
|
||||
3. **`direct_dups` / `flood_dups`** are sent as uint16 and wrap at 65,536. A busy
|
||||
repeater could wrap these between infrequent polls.
|
||||
|
||||
The trade-off is airtime. One cycle is about 4–6 request/response exchanges, and each
|
||||
one is repeated on every hop between your companion and the repeater. That airtime is
|
||||
shared with everyone else on the mesh.
|
||||
|
||||
**Recommendation:** `interval_seconds = 900` (15 min). Go to 300 s if the repeater is
|
||||
1 hop away and you want finer traffic graphs. Go to 1800–3600 s if it's several hops
|
||||
out or the mesh is busy. You can also turn `neighbours = false` to cut the largest
|
||||
reply.
|
||||
|
||||
## Low-battery back-off (solar nodes)
|
||||
|
||||
Every poll costs the repeater some power: it has to receive and transmit the
|
||||
replies. So when its battery is low, the collector polls less often:
|
||||
|
||||
```
|
||||
battery < low_battery_mv -> low-power mode: poll every low_battery_interval_seconds,
|
||||
skip the requests in low_battery_skip
|
||||
battery >= recover_battery_mv -> back to normal interval
|
||||
in between -> stay in whichever mode you're in (hysteresis)
|
||||
```
|
||||
|
||||
The defaults are 3500 / 3750 mV, 1 hour, and skip neighbours. They suit a single
|
||||
Li-ion cell. LiFePO4 has a much flatter curve, so use something like 3200 / 3300.
|
||||
|
||||
- **Where the battery reading comes from:** the status reply. If status is disabled
|
||||
or fails, it falls back to the voltage reading in telemetry. A board that reports
|
||||
0 mV can't measure its battery, so the back-off never triggers for it.
|
||||
- **Per-repeater settings:** any `[poll]` setting except `state_file` can be overridden in a
|
||||
`[[repeater]]` block. For example, use `low_battery_mv = 0` for a mains-powered repeater.
|
||||
- **First run:** with no history, the collector starts in low-power mode if the
|
||||
battery is below the *recover* level. This is the safe choice for solar nodes.
|
||||
- **State file:** `state.json` keeps the mode and last poll time across restarts.
|
||||
This also means `--once` from cron works: run it every few minutes and it only
|
||||
polls repeaters that are due. `--force` polls everything regardless.
|
||||
- **Unreachable repeater:** a repeater with a flat battery usually doesn't answer.
|
||||
The collector stays in its current mode and tries again at the current interval.
|
||||
|
||||
## Grafana dashboard
|
||||
|
||||
Import `grafana/meshcore-repeater-dashboard.json` (Dashboards → New → Import). Then
|
||||
pick your InfluxDB data source and repeater from the dropdowns at the top. Sections:
|
||||
|
||||
- **Overview stats:** battery, polling mode, poll interval, poll success %,
|
||||
uptime, noise floor, temperature, neighbour count.
|
||||
- **Power:** battery voltage with the low/recover thresholds drawn as dashed lines.
|
||||
The thresholds are read from the data, so they always match your config. There's also
|
||||
a polling-mode / reachability timeline, temperature, and poll interval.
|
||||
- **Radio:** noise floor and RSSI, SNR, TX/RX duty cycle %, packets per hour,
|
||||
flood vs direct, duplicates and receive errors, TX queue and error flags.
|
||||
- **Neighbours:** SNR history and latest SNR for each neighbouring repeater.
|
||||
- **Collector & link health:** poll duration, hop count, uptime (reboots show as
|
||||
drops), and any extra telemetry sensors.
|
||||
|
||||
The queries are **InfluxQL**, which works with InfluxDB 1.x, 2.x and 3.x:
|
||||
|
||||
- **InfluxDB 1.8:** add an InfluxDB data source with query language *InfluxQL*.
|
||||
- **InfluxDB 2.x:** create a DBRP mapping for the bucket first:
|
||||
`influx v1 dbrp create --db meshcore --rp autogen --bucket-id <id> --default`.
|
||||
Then add the data source as InfluxQL, with the token in an `Authorization: Token <token>` header.
|
||||
- **InfluxDB 3:** use the InfluxQL query language option.
|
||||
|
||||
The **Battery** stat's colour thresholds are static, set to 3.50 / 3.75 V. Edit them if
|
||||
your config differs.
|
||||
|
||||
I tested the dashboard on Grafana 13.2 against InfluxDB 1.8, using simulated
|
||||
data written by this collector.
|
||||
|
||||
## Grafana queries
|
||||
|
||||
Counters reset on reboot, so always use a non-negative derivative.
|
||||
|
||||
**InfluxQL** (packets/min received):
|
||||
```sql
|
||||
SELECT non_negative_derivative(last("packets_recv"), 1m) FROM "meshcore_repeater_status"
|
||||
WHERE $timeFilter AND "repeater" = 'My Repeater' GROUP BY time($__interval)
|
||||
```
|
||||
|
||||
**Flux** (TX duty cycle %):
|
||||
```flux
|
||||
from(bucket: "meshcore")
|
||||
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|
||||
|> filter(fn: (r) => r._measurement == "meshcore_repeater_status" and r._field == "tx_airtime_s")
|
||||
|> derivative(unit: 1s, nonNegative: true)
|
||||
|> map(fn: (r) => ({r with _value: r._value * 100.0}))
|
||||
```
|
||||
|
||||
Other useful panels:
|
||||
- battery_v
|
||||
- noise_floor_dbm
|
||||
- neighbour `snr_db` grouped by `neighbour_name`
|
||||
- `meshcore_poll.success` as a state timeline, for "is the repeater reachable"
|
||||
|
||||
## Running as a service (Linux, systemd)
|
||||
|
||||
```ini
|
||||
# /etc/systemd/system/meshcore-influx.service
|
||||
[Unit]
|
||||
Description=MeshCore repeater -> InfluxDB
|
||||
After=network-online.target
|
||||
|
||||
[Service]
|
||||
WorkingDirectory=/opt/meshcore-repeater-stat-influx
|
||||
ExecStart=/opt/meshcore-repeater-stat-influx/.venv/bin/python meshcore_influx.py -c config.toml
|
||||
Restart=always
|
||||
RestartSec=30
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
```
|
||||
@@ -0,0 +1,50 @@
|
||||
# Copy to config.toml and edit.
|
||||
|
||||
[connection]
|
||||
# The companion radio attached to the machine running this script.
|
||||
type = "serial" # "serial", "tcp" or "ble"
|
||||
port = "/dev/ttyUSB0" # serial: device path (macOS: /dev/cu.usbserial-XXXX or /dev/cu.usbmodemXXXX)
|
||||
baudrate = 115200
|
||||
# type = "tcp"
|
||||
# host = "192.168.1.50"
|
||||
# port = 5000
|
||||
# type = "ble"
|
||||
# address = "AA:BB:CC:DD:EE:FF"
|
||||
# pin = "123456"
|
||||
|
||||
[influxdb]
|
||||
url = "http://localhost:8086"
|
||||
token = "my-token" # InfluxDB 1.8: "username:password"
|
||||
org = "home" # InfluxDB 1.8: "-"
|
||||
bucket = "meshcore" # InfluxDB 1.8: "database/retention_policy"
|
||||
|
||||
[poll]
|
||||
interval_seconds = 900 # see README: 15 min is a sensible default
|
||||
retries = 3 # per request; mesh packets get lost
|
||||
timeout_seconds = 0 # 0 = use the radio's suggested timeout (scales with hop count)
|
||||
state_file = "state.json" # remembers last poll time + low-power mode across restarts
|
||||
|
||||
# Low-battery back-off (solar nodes). Below low_battery_mv, poll every
|
||||
# low_battery_interval_seconds and skip the requests in low_battery_skip;
|
||||
# stay there until the battery reaches recover_battery_mv. 0 disables.
|
||||
# Values below suit a single Li-ion cell; LiFePO4 is flatter (e.g. 3200 / 3300).
|
||||
low_battery_mv = 3500
|
||||
recover_battery_mv = 3750
|
||||
low_battery_interval_seconds = 3600
|
||||
low_battery_skip = ["neighbours"] # any of "status", "telemetry", "neighbours"
|
||||
# Every key in this [poll] block from interval_seconds down (except state_file)
|
||||
# can be overridden per [[repeater]].
|
||||
|
||||
# One block per repeater. It must already be in the companion's contact list.
|
||||
[[repeater]]
|
||||
name = "My Repeater" # advert name, case-insensitive
|
||||
password = "guest-or-admin-pw" # guest is enough for status + basic telemetry + neighbours
|
||||
# pubkey_prefix = "a1b2c3d4" # use instead of name if names collide
|
||||
status = true
|
||||
telemetry = true
|
||||
neighbours = true
|
||||
|
||||
# [[repeater]]
|
||||
# name = "Hilltop"
|
||||
# password = "..."
|
||||
# low_battery_mv = 0 # mains powered: never back off
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,472 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Poll MeshCore repeaters over the mesh (via a locally attached companion radio)
|
||||
and write their stats, sensor telemetry and neighbour table to InfluxDB.
|
||||
|
||||
companion radio (USB / TCP / BLE) -> mesh -> repeater(s)
|
||||
|
|
||||
this script -> InfluxDB -> Grafana
|
||||
|
||||
Usage:
|
||||
python meshcore_influx.py -c config.toml # run forever
|
||||
python meshcore_influx.py -c config.toml --once # single poll (cron)
|
||||
python meshcore_influx.py -c config.toml --once --dry-run # print, don't write
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
import tomllib
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
from meshcore import MeshCore, EventType
|
||||
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().
|
||||
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"}
|
||||
|
||||
|
||||
@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")
|
||||
|
||||
|
||||
@dataclass
|
||||
class RepeaterCfg:
|
||||
name: str # contact name as seen by the companion
|
||||
password: str = "" # admin or guest password ("" = rely on existing ACL entry)
|
||||
pubkey_prefix: str = "" # optional, used instead of name if set
|
||||
status: bool = True
|
||||
telemetry: bool = True
|
||||
neighbours: bool = True
|
||||
interval: int = 900
|
||||
battery: BatteryPolicy = field(default_factory=BatteryPolicy)
|
||||
|
||||
|
||||
def build_repeater(raw: dict, poll: dict) -> RepeaterCfg:
|
||||
eff = {k: raw.get(k, poll.get(k)) for k in POLICY_KEYS}
|
||||
low = int(eff["low_battery_mv"] or 0)
|
||||
recover = int(eff["recover_battery_mv"] or low)
|
||||
if recover < low:
|
||||
raise ValueError(f"{raw['name']}: recover_battery_mv ({recover}) must be >= low_battery_mv ({low})")
|
||||
pol = BatteryPolicy(
|
||||
low_mv=low,
|
||||
recover_mv=recover,
|
||||
interval=int(eff["low_battery_interval_seconds"] or 3600),
|
||||
skip=tuple(eff["low_battery_skip"] if eff["low_battery_skip"] is not None else ("neighbours",)),
|
||||
)
|
||||
rest = {k: v for k, v in raw.items() if k not in POLICY_KEYS}
|
||||
return RepeaterCfg(**rest, interval=int(eff["interval_seconds"] or 900), battery=pol)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
connection: dict
|
||||
influx: dict
|
||||
retries: int = 3
|
||||
timeout: float = 0 # 0 = use the radio's suggested timeout
|
||||
state_file: str = "state.json"
|
||||
repeaters: list[RepeaterCfg] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def load(cls, path: str) -> "Config":
|
||||
with open(path, "rb") as f:
|
||||
raw = tomllib.load(f)
|
||||
poll = raw.get("poll", {})
|
||||
return cls(
|
||||
connection=raw["connection"],
|
||||
influx=raw["influxdb"],
|
||||
retries=int(poll.get("retries", 3)),
|
||||
timeout=float(poll.get("timeout_seconds", 0)),
|
||||
state_file=poll.get("state_file", "state.json"),
|
||||
repeaters=[build_repeater(r, poll) for r in raw["repeater"]],
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 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
|
||||
|
||||
def interval(self, rpt: RepeaterCfg) -> int:
|
||||
return rpt.battery.interval if self.low_power else rpt.interval
|
||||
|
||||
def next_due(self, rpt: RepeaterCfg) -> float:
|
||||
return self.last_poll + self.interval(rpt)
|
||||
|
||||
def update_battery(self, rpt: RepeaterCfg, 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]:
|
||||
try:
|
||||
with open(path) as f:
|
||||
return {k: RepeaterState(**v) 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: RepeaterCfg) -> 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: RepeaterCfg, cfg: Config) -> bool:
|
||||
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:
|
||||
log.info("%s: logged in (admin=%s)", rpt.name, ev.payload.get("is_admin"))
|
||||
return ev is not None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Conversion to InfluxDB points
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def status_point(rpt: RepeaterCfg, 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: RepeaterCfg, key: str, lpp: list[dict], ts: int) -> list[Point]:
|
||||
"""One point per LPP entry: tags channel + sensor type, value(s) as float fields."""
|
||||
pts = []
|
||||
for item in lpp or []:
|
||||
p = (Point("meshcore_repeater_telemetry")
|
||||
.tag("repeater", 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: RepeaterCfg, 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: RepeaterCfg, 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 ("status", "telemetry", "neighbours")}
|
||||
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:
|
||||
log.error("%s: not in companion's contact list (wait for its advert, or add it)", rpt.name)
|
||||
return [Point("meshcore_poll").tag("repeater", rpt.name).field("success", 0)
|
||||
.field("error", "contact not found").time(ts, WritePrecision.S)]
|
||||
key = contact["public_key"]
|
||||
|
||||
logged_in = await login(mc, contact, rpt, cfg)
|
||||
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 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
|
||||
|
||||
try:
|
||||
await mc.commands.send_logout(contact)
|
||||
except Exception: # noqa: BLE001 - best effort
|
||||
pass
|
||||
|
||||
state.update_battery(rpt, battery_mv)
|
||||
|
||||
# Poll health + power-management state: for alerting and dashboards.
|
||||
p = (Point("meshcore_poll")
|
||||
.tag("repeater", rpt.name).tag("pubkey", key[:12])
|
||||
.field("success", int(logged_in and all(ok[k] for k in ok if wanted[k])))
|
||||
.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", round(time.monotonic() - t0, 2))
|
||||
.field("low_power_mode", int(bool(state.low_power)))
|
||||
.field("interval_s", int(state.interval(rpt))))
|
||||
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,
|
||||
time.monotonic() - t0, battery_mv, "low-power" if state.low_power else "normal",
|
||||
state.interval(rpt), ok)
|
||||
return pts
|
||||
|
||||
|
||||
async def write_points(cfg: Config, pts: list[Point], dry_run: bool) -> None:
|
||||
if dry_run:
|
||||
for p in pts:
|
||||
print(p.to_line_protocol())
|
||||
return
|
||||
ic = cfg.influx
|
||||
async with InfluxDBClientAsync(url=ic["url"], token=ic["token"], org=ic.get("org", "-")) as client:
|
||||
await client.write_api().write(bucket=ic["bucket"], record=pts)
|
||||
log.info("wrote %d points to InfluxDB", len(pts))
|
||||
|
||||
|
||||
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 repeaters that are due, then exit (for cron/systemd timers)")
|
||||
ap.add_argument("--force", action="store_true", help="with --once: poll every repeater even if not due")
|
||||
ap.add_argument("--dry-run", action="store_true", help="print line protocol instead of writing")
|
||||
ap.add_argument("-v", "--verbose", action="store_true")
|
||||
args = ap.parse_args()
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||||
cfg = Config.load(args.config)
|
||||
states = load_state(cfg.state_file)
|
||||
for r in cfg.repeaters:
|
||||
states.setdefault(r.name, RepeaterState())
|
||||
|
||||
stop = asyncio.Event()
|
||||
loop = asyncio.get_running_loop()
|
||||
for sig in (signal.SIGINT, signal.SIGTERM):
|
||||
loop.add_signal_handler(sig, stop.set)
|
||||
|
||||
mc: Optional[MeshCore] = None
|
||||
rc = 0
|
||||
try:
|
||||
while not stop.is_set():
|
||||
now = time.time()
|
||||
due = [r for r in cfg.repeaters
|
||||
if args.force or states[r.name].next_due(r) <= now]
|
||||
if due:
|
||||
pts: list[Point] = []
|
||||
try:
|
||||
if mc is None or not mc.is_connected:
|
||||
mc = await connect_companion(cfg.connection)
|
||||
for rpt in due: # sequential: one radio, one request in flight
|
||||
pts += await poll_repeater(mc, rpt, cfg, states[rpt.name])
|
||||
except Exception as e: # noqa: BLE001 - reconnect on next cycle
|
||||
log.error("poll cycle failed: %s", e)
|
||||
rc = 1
|
||||
if mc is not None:
|
||||
try:
|
||||
await mc.disconnect()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
mc = None
|
||||
# Don't spin on a dead companion: retry unpolled repeaters in a minute.
|
||||
for rpt in due:
|
||||
st = states[rpt.name]
|
||||
if st.next_due(rpt) <= time.time():
|
||||
st.last_poll = time.time() - st.interval(rpt) + 60
|
||||
if pts:
|
||||
try:
|
||||
await write_points(cfg, pts, args.dry_run)
|
||||
except Exception as e: # noqa: BLE001 - keep the daemon alive
|
||||
log.error("InfluxDB write failed: %s", e)
|
||||
rc = 1
|
||||
save_state(cfg.state_file, states)
|
||||
elif args.once:
|
||||
log.info("no repeater due yet (use --force to poll anyway)")
|
||||
|
||||
if args.once:
|
||||
break
|
||||
next_wake = min(states[r.name].next_due(r) for r in cfg.repeaters)
|
||||
try:
|
||||
await asyncio.wait_for(stop.wait(), timeout=max(1.0, next_wake - time.time()))
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
finally:
|
||||
if mc is not None:
|
||||
await mc.disconnect()
|
||||
return rc if args.once else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(asyncio.run(main()))
|
||||
@@ -0,0 +1,2 @@
|
||||
meshcore>=2.3
|
||||
influxdb-client[async]>=1.40
|
||||
Reference in new issue
Block a user