mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-08-10 19:03:28 +02:00
data,web: ingest every telemetry family from both protocols (#850)
* data,web: ingest every telemetry family from both protocols * data: 24h per-node telemetry poll cooldown; RX_ONLY ingestor switch
This commit is contained in:
l5y
@@ -3038,3 +3038,94 @@ GH-A1** (MeshCore message/contact machinery — naming, `last_heard`, and
|
||||
stale-contact behavior unchanged), and **B1/B4/B5** (all suites, headers,
|
||||
formatters). The JS suite is exercised for regression only — RF7 adds no
|
||||
frontend behavior.
|
||||
|
||||
---
|
||||
|
||||
## Bugfix: Missing telemetry at ingest (all families, both protocols)
|
||||
|
||||
Two ingest-time data losses. **Meshtastic:** the telemetry protobuf `oneof` has
|
||||
eight variants, but extraction targeted only `deviceMetrics.*` /
|
||||
`environmentMetrics.*` paths — PowerMetrics (16 fields), AirQualityMetrics (25,
|
||||
incl. PM series, particle counts, CO2, formaldehyde, VOC/NOx), HealthMetrics
|
||||
(3), LocalStats (15), HostMetrics (9), TrafficManagementStats (7), and the
|
||||
repeated `oneWireTemperature` were dropped; the last four families were not
|
||||
even recognised by the discriminator, landing as rows with no `telemetry_type`
|
||||
and no metrics. The web app mirrored the drop (no columns, no metric
|
||||
definitions, `power_metrics`/`air_quality_metrics` consulted only for type
|
||||
inference). **MeshCore:** telemetry was structurally unreachable — no
|
||||
subscription to `TELEMETRY_RESPONSE`/`STATUS_RESPONSE`/`BATTERY`, no telemetry
|
||||
commands issued, no CayenneLPP mapping — although the `meshcore` library
|
||||
(≥2.3.5) exposes self battery/sensors and per-contact pulls, violating
|
||||
Invariant IV (protocol parity; the web/DB side was already protocol-ready).
|
||||
Fix: the ingestor extracts **every** field of all eight Meshtastic families
|
||||
(`telemetry_type` gains `local_stats`/`health`/`host`/`traffic`; body
|
||||
temperature stays distinct as `health_temperature`; `one_wire_temperature` is
|
||||
a JSON float list), the web app stores and serves all new columns (schema +
|
||||
boot auto-migration + insert/upsert; `GET /api/telemetry` is `SELECT *`), and
|
||||
the MeshCore provider collects host self-telemetry (no airtime) plus
|
||||
round-robin contact telemetry/status polls (conservative, env-tunable,
|
||||
disableable). Frontend intentionally untouched. `CONTRACTS.md` amended
|
||||
additively (D8); apex (I) and privacy (II) untouched.
|
||||
|
||||
### TI-A1 — Meshtastic ingestor extracts every telemetry family
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_handlers_unit.py -k "ExtendedTelemetry" )
|
||||
```
|
||||
**Expected:** pass. For each `oneof` family the queued `/api/telemetry`
|
||||
payload carries the family's snake_case metric keys and the correct
|
||||
`telemetry_type`: power (`ch1_voltage`…`ch8_current`), air_quality
|
||||
(`pm*_standard/environmental`, `particles_*`, `co2*`, `form_*`, `pm_voc_idx`,
|
||||
`pm_nox_idx`, `particles_tps`), health (`heart_bpm`, `spo2`,
|
||||
`health_temperature` — never the ambient `temperature` key), local_stats
|
||||
(counters + reuse of `uptime_seconds`/`channel_utilization`/`air_util_tx`),
|
||||
host (`freemem_bytes`, `diskfree*_bytes`, `load*`, `user_string`), traffic
|
||||
(`packets_inspected`, …), and environment's `one_wire_temperature` list.
|
||||
|
||||
### TI-A2 — Web app stores and serves the extended metrics
|
||||
```bash
|
||||
( cd web && bundle exec rspec spec/data_processing_spec.rb -e "extended metric families" )
|
||||
```
|
||||
**Expected:** pass. `insert_telemetry` persists values from the
|
||||
`power_metrics` / `air_quality_metrics` / `health_metrics` / `local_stats` /
|
||||
`host_metrics` / `traffic_management_stats` sub-objects (and their flat
|
||||
snake_case keys) into real columns; the diagnostics `telemetry_type` values
|
||||
are accepted; `one_wire_temperature` round-trips as a JSON array;
|
||||
`user_string` stores text. Existing databases gain the columns via the boot
|
||||
auto-migrator (`ensure_schema_upgrades`), fresh installs via
|
||||
`data/telemetry.sql`.
|
||||
|
||||
### TI-A3 — MeshCore provider collects telemetry
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_provider_unit.py -k "telemetry" )
|
||||
```
|
||||
**Expected:** pass. The MeshCore event-handler map subscribes
|
||||
`TELEMETRY_RESPONSE`, `STATUS_RESPONSE`, and `BATTERY`; CayenneLPP entries map
|
||||
to the canonical metric keys (temperature, `relative_humidity`,
|
||||
`barometric_pressure`, voltage, current, lux, `battery_level`); status
|
||||
responses map `bat` (mV) → voltage (V) and uptime; events resolve
|
||||
`pubkey_pre` to the contact's canonical node id (host prefix → host node);
|
||||
resulting packets flow through `store_packet_dict` → `store_telemetry_packet`
|
||||
with `protocol="meshcore"`. The poll loop honours
|
||||
`MESHCORE_TELEMETRY_POLL_SECONDS` (0 disables contact polling) and
|
||||
`MESHCORE_SELF_TELEMETRY_SECONDS`, one on-air request at a time (local LoRa
|
||||
only — no broker, Invariant I). Each contact is additionally capped by a
|
||||
fixed 24 h per-node cooldown (stamped at the poll attempt; an all-fresh
|
||||
roster tick transmits nothing, and departed contacts are pruned from the
|
||||
stamp table). `RX_ONLY=1` forbids every ingestor-initiated transmission:
|
||||
contact polls stop entirely while the airtime-free companion-link self reads
|
||||
continue.
|
||||
|
||||
### TI-R1 — Regression: prior acceptance still holds
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/ ) && ( cd web && bundle exec rspec ) && ( cd web && npm test )
|
||||
```
|
||||
**Expected:** every prior check still passes. At risk and explicitly required
|
||||
to remain green: **C2** (canonical POST shapes — the metric additions are
|
||||
additive, existing keys unchanged), **A4b/A4e** (MeshCore provider conformance
|
||||
and advert handling — new subscriptions must not disturb existing handlers),
|
||||
**A2/A2a** (privacy — telemetry remains ungated by `PRIVATE`, unchanged),
|
||||
**D2** (channel filters unaffected), and the host-telemetry suppression window
|
||||
(self-poll responses are throttled by the existing
|
||||
`store_telemetry_packet` host gate). The frontend is intentionally untouched
|
||||
(TM-A1 unchanged); `tests/` fixtures are unmodified so CI replay (C2) is
|
||||
unaffected.
|
||||
|
||||
@@ -110,6 +110,9 @@ The web app can be configured with environment variables (defaults shown):
|
||||
| `MESH_UDP_GROUP` | `224.0.0.69` | Multicast group joined in UDP transport. |
|
||||
| `MESH_UDP_PORT` | `4403` | Multicast port joined in UDP transport. |
|
||||
| `INGESTOR_NODE_ID` | _unset_ | `!xxxxxxxx` id used for the ingestor heartbeat in UDP transport (which cannot auto-detect "self"). |
|
||||
| `MESHCORE_TELEMETRY_POLL_SECONDS` | `300` | Seconds between MeshCore contact telemetry polls (one on-air request per interval, round-robin over the roster; each contact is additionally polled at most once per 24 h — when every contact is fresh the tick sends nothing). Set `0` to disable on-air polling. |
|
||||
| `MESHCORE_SELF_TELEMETRY_SECONDS` | `3600` | Seconds between MeshCore host self-telemetry reads (battery/sensors over the companion link, no airtime). Set `0` to disable. |
|
||||
| `RX_ONLY` | `0` | Set to `1` to forbid every ingestor-initiated mesh transmission (receive-only listening post). Currently disables the MeshCore contact telemetry/status polls; local companion-link reads (self telemetry, roster, channels) continue. |
|
||||
| `FEDERATION` | `1` | Set to `1` to announce your instance and crawl peers, or `0` to disable federation. Private mode overrides this. |
|
||||
| `PRIVATE` | `0` | Set to `1` to hide the chat UI, disable message APIs, and exclude hidden clients from public listings. |
|
||||
| `EVENTS` | `1` | Set to `0` to disable the live-update SSE stream (`GET /api/events`); clients then fall back to polling at the refresh interval. |
|
||||
|
||||
@@ -145,11 +145,45 @@ Single telemetry payload:
|
||||
- Packet: `channel` (int), `portnum` (string|nil), `bitfield` (int|nil), `hop_limit` (int|nil)
|
||||
- RF: `snr` (float|nil), `rssi` (int|nil)
|
||||
- Raw: `payload_b64` (string; may be empty string when unknown)
|
||||
- Metrics: many optional snake_case keys (`battery_level`, `voltage`, `temperature`, etc.)
|
||||
- Subtype: `telemetry_type` (string|nil) — optional discriminator identifying which Meshtastic protobuf oneof was set; one of `"device"`, `"environment"`, `"power"`, or `"air_quality"`. Ingestors that detect the subtype SHOULD include this field; omit rather than send `null` when unknown. The web app infers the type from metric-field presence when absent, so old ingestors remain compatible.
|
||||
- Metrics: many optional snake_case keys, one per stored column. Device:
|
||||
`battery_level`, `voltage`, `channel_utilization`, `air_util_tx`,
|
||||
`uptime_seconds`. Environment: `temperature`, `relative_humidity`,
|
||||
`barometric_pressure`, `gas_resistance`, `current`, `iaq`, `distance`,
|
||||
`lux`/`white_lux`/`ir_lux`/`uv_lux`, `wind_direction`/`wind_speed`/
|
||||
`wind_gust`/`wind_lull`, `weight`, `radiation`, `rainfall_1h`/`rainfall_24h`,
|
||||
`soil_moisture`/`soil_temperature`, and `one_wire_temperature`
|
||||
(list[float], stored as a JSON array). Power (TI-A1/A2): `ch1_voltage` …
|
||||
`ch8_voltage`, `ch1_current` … `ch8_current`. Air quality:
|
||||
`pm10_standard`/`pm25_standard`/`pm100_standard`/`pm40_standard`,
|
||||
`pm10_environmental`/`pm25_environmental`/`pm100_environmental`,
|
||||
`particles_03um`/`particles_05um`/`particles_10um`/`particles_25um`/
|
||||
`particles_40um`/`particles_50um`/`particles_100um`, `particles_tps`,
|
||||
`co2`/`co2_temperature`/`co2_humidity`,
|
||||
`form_formaldehyde`/`form_humidity`/`form_temperature`,
|
||||
`pm_temperature`/`pm_humidity`/`pm_voc_idx`/`pm_nox_idx`. Health:
|
||||
`heart_bpm`, `spo2`, `health_temperature` (body temperature — deliberately
|
||||
distinct from the ambient `temperature`). Local stats: `num_packets_tx`,
|
||||
`num_packets_rx`, `num_packets_rx_bad`, `num_online_nodes`,
|
||||
`num_total_nodes`, `num_rx_dupe`, `num_tx_relay`, `num_tx_relay_canceled`,
|
||||
`heap_total_bytes`, `heap_free_bytes`, `num_tx_dropped`, `noise_floor`
|
||||
(plus the shared `uptime_seconds`/`channel_utilization`/`air_util_tx`).
|
||||
Host: `freemem_bytes`, `diskfree1_bytes`/`diskfree2_bytes`/
|
||||
`diskfree3_bytes`, `load1`/`load5`/`load15`, `user_string` (string).
|
||||
Traffic: `packets_inspected`, `position_dedup_drops`,
|
||||
`nodeinfo_cache_hits`, `rate_limit_drops`, `unknown_packet_drops`,
|
||||
`hop_exhausted_packets`, `router_hops_preserved`. The web app also accepts
|
||||
each family nested as a sub-object (`device_metrics`, `environment_metrics`,
|
||||
`power_metrics`, `air_quality_metrics`, `local_stats`, `health_metrics`,
|
||||
`host_metrics`, `traffic_management_stats`) with camelCase or snake_case
|
||||
field names; nested family objects are consulted for **values**, not only
|
||||
for type inference. All metric additions are additive (D8) — absent keys
|
||||
are simply omitted, never sent as `null`.
|
||||
- Subtype: `telemetry_type` (string|nil) — optional discriminator identifying which Meshtastic protobuf oneof was set; one of `"device"`, `"environment"`, `"power"`, `"air_quality"`, `"local_stats"`, `"health"`, `"host"`, or `"traffic"` (the last four added additively for the LocalStats / HealthMetrics / HostMetrics / TrafficManagementStats variants, TI-A1). Ingestors that detect the subtype SHOULD include this field; omit rather than send `null` when unknown. The web app infers the type from metric-field presence when absent, so old ingestors remain compatible.
|
||||
- Meta: `ingestor`, `lora_freq`, `modem_preset`
|
||||
- `protocol` (optional string; `"meshtastic"` or `"meshcore"`) — explicit per-record protocol stamp; same semantics as on `POST /api/messages`.
|
||||
|
||||
**MeshCore telemetry sourcing (TI-A3).** MeshCore exposes other nodes' telemetry only as on-air *pull* requests (there is no unsolicited telemetry broadcast the companion library surfaces), so the MeshCore provider collects it three ways and normalises every reading into this same payload shape with `protocol="meshcore"`: (1) **host self-telemetry** over the local companion link (`get_bat` → battery millivolts as `voltage`; `get_self_telemetry` → the host's CayenneLPP sensor list), no LoRa airtime, cadence `MESHCORE_SELF_TELEMETRY_SECONDS` (default 3600 s, matching the host-telemetry suppression window; `<= 0` disables); (2) **round-robin contact polling** (`req_telemetry_sync`, falling back to `req_status_sync` when a node reports no sensors) at one on-air request per `MESHCORE_TELEMETRY_POLL_SECONDS` (default 300 s; `<= 0` disables) regardless of roster size, with each contact additionally capped at **one poll per 24 h** (a fixed per-node cooldown, stamped at the poll attempt so unreachable nodes are not hammered; when every contact is fresh the tick transmits nothing) — and `RX_ONLY=1` forbids these on-air polls entirely (receive-only ingestors; the local self reads in (1) are unaffected); (3) **unsolicited/tag-matched events** (`TELEMETRY_RESPONSE`, `STATUS_RESPONSE`, `BATTERY`) whenever the radio surfaces them. CayenneLPP types map to canonical keys (`temperature`, `humidity`→`relative_humidity`, `barometer`→`barometric_pressure`, `voltage`, `current` — scaled A→mA to match the Meshtastic column convention, `illuminance`→`lux`, `percentage`→`battery_level`); status `bat`/`level` millivolt gauges map to `voltage` (V). MeshCore assigns no firmware packet id, so the record `id` is the deterministic 53-bit fingerprint of *(node id, receive second, source kind)* — re-reads of the same source in the same second collapse into one row via the `telemetry.id` upsert.
|
||||
|
||||
#### `POST /api/neighbors`
|
||||
|
||||
Neighbors snapshot payload:
|
||||
|
||||
@@ -140,6 +140,39 @@ raise ``ValueError`` at import and prevent the service from starting."""
|
||||
INGESTOR_NODE_ID = os.environ.get("INGESTOR_NODE_ID", "").strip() or None
|
||||
"""Optional ``!xxxxxxxx`` host node id used for the ingestor heartbeat in UDP mode."""
|
||||
|
||||
RX_ONLY = os.environ.get("RX_ONLY") == "1"
|
||||
"""Receive-only mode: forbid every ingestor-initiated mesh transmission.
|
||||
|
||||
Some operators run listening posts where any TX is undesired. When set, the
|
||||
ingestor never transmits on the mesh: currently this disables the MeshCore
|
||||
contact telemetry/status polls (the only ingestor-initiated RF traffic), and
|
||||
any future TX feature must honour it too. Local companion-link reads (host
|
||||
self-telemetry, contact roster, channel queries) are not transmissions and
|
||||
continue to work."""
|
||||
|
||||
MESHCORE_TELEMETRY_POLL_SECONDS = int(
|
||||
os.environ.get("MESHCORE_TELEMETRY_POLL_SECONDS", "300").strip() or "300"
|
||||
)
|
||||
"""Seconds between successive MeshCore contact telemetry polls (TI-A3).
|
||||
|
||||
MeshCore exposes other nodes' telemetry only via on-air pull requests, so the
|
||||
provider round-robins the contact roster issuing one request per interval —
|
||||
airtime is bounded to one request per ``MESHCORE_TELEMETRY_POLL_SECONDS``
|
||||
regardless of roster size. Values ``<= 0`` disable contact polling entirely
|
||||
(host self-telemetry is governed separately by
|
||||
``MESHCORE_SELF_TELEMETRY_SECONDS``). Stripped with a default fallback like
|
||||
``MESH_UDP_PORT`` so a blank value in a ``.env`` file cannot break startup."""
|
||||
|
||||
MESHCORE_SELF_TELEMETRY_SECONDS = int(
|
||||
os.environ.get("MESHCORE_SELF_TELEMETRY_SECONDS", "3600").strip() or "3600"
|
||||
)
|
||||
"""Seconds between MeshCore host self-telemetry reads (battery + sensors).
|
||||
|
||||
Self reads are local companion-link commands (no LoRa airtime). The default
|
||||
matches the host-telemetry suppression window in
|
||||
``handlers._state._HOST_TELEMETRY_INTERVAL_SECS`` (one hour) so more frequent
|
||||
reads would only be suppressed anyway. Values ``<= 0`` disable self polling."""
|
||||
|
||||
|
||||
def _parse_lora_freq_env(raw: str | None) -> float | int | None:
|
||||
"""Parse the ``FREQUENCY`` environment variable into a numeric LoRa frequency.
|
||||
|
||||
@@ -34,17 +34,203 @@ from .position import base64_payload
|
||||
from .radio import _apply_radio_metadata, _apply_radio_metadata_to_nodes
|
||||
|
||||
_VALID_TELEMETRY_TYPES: frozenset[str] = frozenset(
|
||||
{"device", "environment", "power", "air_quality"}
|
||||
{
|
||||
"device",
|
||||
"environment",
|
||||
"power",
|
||||
"air_quality",
|
||||
"local_stats",
|
||||
"health",
|
||||
"host",
|
||||
"traffic",
|
||||
}
|
||||
)
|
||||
"""Allowed discriminator values for the ``telemetry_type`` field.
|
||||
|
||||
Meshtastic uses a protobuf ``oneof`` so only one metric sub-object can be
|
||||
populated per packet. Values outside this set indicate a firmware version
|
||||
that added a new type not yet handled here; those are logged and dropped to
|
||||
avoid persisting unexpected data shapes.
|
||||
populated per packet. One value per ``Telemetry.variant`` member (TI-A1).
|
||||
Values outside this set indicate a firmware version that added a new type not
|
||||
yet handled here; those are logged and dropped to avoid persisting unexpected
|
||||
data shapes.
|
||||
"""
|
||||
|
||||
|
||||
def _coerce_str(value) -> str | None:
|
||||
"""Coerce a free-text metric (e.g. ``HostMetrics.userString``) to a
|
||||
trimmed string, or ``None`` when blank or not text."""
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
trimmed = value.strip()
|
||||
return trimmed or None
|
||||
|
||||
|
||||
def _coerce_float_list(value) -> list[float] | None:
|
||||
"""Coerce a repeated float metric (``EnvironmentMetrics.oneWireTemperature``)
|
||||
to a list of finite floats, or ``None`` when nothing usable remains."""
|
||||
if not isinstance(value, (list, tuple)):
|
||||
return None
|
||||
floats = [f for f in (_coerce_float(item) for item in value) if f is not None]
|
||||
return floats or None
|
||||
|
||||
|
||||
def _family_fields(family_camel: str, family_snake: str, fields) -> tuple:
|
||||
"""Expand ``(payload_key, coercer, camel_name)`` triples into full metric
|
||||
definitions probing the family sub-object under both key spellings.
|
||||
|
||||
Parameters:
|
||||
family_camel: camelCase sub-object key as decoded from protobuf JSON
|
||||
(e.g. ``"powerMetrics"``).
|
||||
family_snake: snake_case twin accepted for defensive compatibility.
|
||||
fields: Iterable of ``(payload_key, coercer, camel_name)`` triples.
|
||||
|
||||
Returns:
|
||||
Tuple of ``(payload_key, coercer, candidate_paths)`` definitions.
|
||||
"""
|
||||
return tuple(
|
||||
(
|
||||
key,
|
||||
coercer,
|
||||
(
|
||||
f"{family_camel}.{camel}",
|
||||
f"{family_camel}.{key}",
|
||||
f"{family_snake}.{key}",
|
||||
),
|
||||
)
|
||||
for key, coercer, camel in fields
|
||||
)
|
||||
|
||||
|
||||
def _power_channel_fields() -> tuple:
|
||||
"""Build the 16 ``PowerMetrics`` channel definitions (``ch1``–``ch8``,
|
||||
voltage + current per channel)."""
|
||||
fields = []
|
||||
for ch in range(1, 9):
|
||||
for kind, camel_kind in (("voltage", "Voltage"), ("current", "Current")):
|
||||
fields.append((f"ch{ch}_{kind}", _coerce_float, f"ch{ch}{camel_kind}"))
|
||||
return _family_fields("powerMetrics", "power_metrics", fields)
|
||||
|
||||
|
||||
_EXTENDED_METRIC_FIELDS: tuple = (
|
||||
_power_channel_fields()
|
||||
+ _family_fields(
|
||||
"airQualityMetrics",
|
||||
"air_quality_metrics",
|
||||
(
|
||||
("pm10_standard", _coerce_int, "pm10Standard"),
|
||||
("pm25_standard", _coerce_int, "pm25Standard"),
|
||||
("pm100_standard", _coerce_int, "pm100Standard"),
|
||||
("pm40_standard", _coerce_int, "pm40Standard"),
|
||||
("pm10_environmental", _coerce_int, "pm10Environmental"),
|
||||
("pm25_environmental", _coerce_int, "pm25Environmental"),
|
||||
("pm100_environmental", _coerce_int, "pm100Environmental"),
|
||||
("particles_03um", _coerce_int, "particles03um"),
|
||||
("particles_05um", _coerce_int, "particles05um"),
|
||||
("particles_10um", _coerce_int, "particles10um"),
|
||||
("particles_25um", _coerce_int, "particles25um"),
|
||||
("particles_40um", _coerce_int, "particles40um"),
|
||||
("particles_50um", _coerce_int, "particles50um"),
|
||||
("particles_100um", _coerce_int, "particles100um"),
|
||||
("particles_tps", _coerce_float, "particlesTps"),
|
||||
("co2", _coerce_int, "co2"),
|
||||
("co2_temperature", _coerce_float, "co2Temperature"),
|
||||
("co2_humidity", _coerce_float, "co2Humidity"),
|
||||
("form_formaldehyde", _coerce_float, "formFormaldehyde"),
|
||||
("form_humidity", _coerce_float, "formHumidity"),
|
||||
("form_temperature", _coerce_float, "formTemperature"),
|
||||
("pm_temperature", _coerce_float, "pmTemperature"),
|
||||
("pm_humidity", _coerce_float, "pmHumidity"),
|
||||
("pm_voc_idx", _coerce_float, "pmVocIdx"),
|
||||
("pm_nox_idx", _coerce_float, "pmNoxIdx"),
|
||||
),
|
||||
)
|
||||
+ _family_fields(
|
||||
"healthMetrics",
|
||||
"health_metrics",
|
||||
(
|
||||
("heart_bpm", _coerce_int, "heartBpm"),
|
||||
("spo2", _coerce_int, "spO2"),
|
||||
# Body temperature is deliberately kept apart from the ambient
|
||||
# ``temperature`` column so a chest strap never reads as weather.
|
||||
("health_temperature", _coerce_float, "temperature"),
|
||||
),
|
||||
)
|
||||
+ _family_fields(
|
||||
"localStats",
|
||||
"local_stats",
|
||||
(
|
||||
("num_packets_tx", _coerce_int, "numPacketsTx"),
|
||||
("num_packets_rx", _coerce_int, "numPacketsRx"),
|
||||
("num_packets_rx_bad", _coerce_int, "numPacketsRxBad"),
|
||||
("num_online_nodes", _coerce_int, "numOnlineNodes"),
|
||||
("num_total_nodes", _coerce_int, "numTotalNodes"),
|
||||
("num_rx_dupe", _coerce_int, "numRxDupe"),
|
||||
("num_tx_relay", _coerce_int, "numTxRelay"),
|
||||
("num_tx_relay_canceled", _coerce_int, "numTxRelayCanceled"),
|
||||
("heap_total_bytes", _coerce_int, "heapTotalBytes"),
|
||||
("heap_free_bytes", _coerce_int, "heapFreeBytes"),
|
||||
("num_tx_dropped", _coerce_int, "numTxDropped"),
|
||||
("noise_floor", _coerce_int, "noiseFloor"),
|
||||
),
|
||||
)
|
||||
+ _family_fields(
|
||||
"hostMetrics",
|
||||
"host_metrics",
|
||||
(
|
||||
("freemem_bytes", _coerce_int, "freememBytes"),
|
||||
("diskfree1_bytes", _coerce_int, "diskfree1Bytes"),
|
||||
("diskfree2_bytes", _coerce_int, "diskfree2Bytes"),
|
||||
("diskfree3_bytes", _coerce_int, "diskfree3Bytes"),
|
||||
("load1", _coerce_int, "load1"),
|
||||
("load5", _coerce_int, "load5"),
|
||||
("load15", _coerce_int, "load15"),
|
||||
("user_string", _coerce_str, "userString"),
|
||||
),
|
||||
)
|
||||
+ _family_fields(
|
||||
"trafficManagementStats",
|
||||
"traffic_management_stats",
|
||||
(
|
||||
("packets_inspected", _coerce_int, "packetsInspected"),
|
||||
("position_dedup_drops", _coerce_int, "positionDedupDrops"),
|
||||
("nodeinfo_cache_hits", _coerce_int, "nodeinfoCacheHits"),
|
||||
("rate_limit_drops", _coerce_int, "rateLimitDrops"),
|
||||
("unknown_packet_drops", _coerce_int, "unknownPacketDrops"),
|
||||
("hop_exhausted_packets", _coerce_int, "hopExhaustedPackets"),
|
||||
("router_hops_preserved", _coerce_int, "routerHopsPreserved"),
|
||||
),
|
||||
)
|
||||
+ _family_fields(
|
||||
"environmentMetrics",
|
||||
"environment_metrics",
|
||||
(("one_wire_temperature", _coerce_float_list, "oneWireTemperature"),),
|
||||
)
|
||||
)
|
||||
"""Extended metric definitions for the telemetry families beyond the original
|
||||
device/environment pair (TI-A1): ``(payload_key, coercer, candidate_paths)``
|
||||
per field. Candidate paths are dotted ``_first`` lookups into the decoded
|
||||
telemetry section."""
|
||||
|
||||
|
||||
def _extract_extended_metrics(telemetry_section: Mapping) -> dict:
|
||||
"""Extract every non-None extended-family metric from *telemetry_section*.
|
||||
|
||||
Parameters:
|
||||
telemetry_section: Decoded ``Telemetry`` dict (the packet's
|
||||
``decoded["telemetry"]`` mapping).
|
||||
|
||||
Returns:
|
||||
Mapping of snake_case payload key → coerced value, containing only the
|
||||
fields actually present in the packet, so absent fields are omitted
|
||||
from the POST body rather than sent as null.
|
||||
"""
|
||||
metrics: dict = {}
|
||||
for payload_key, coercer, candidates in _EXTENDED_METRIC_FIELDS:
|
||||
value = coercer(_first(telemetry_section, *candidates, default=None))
|
||||
if value is not None:
|
||||
metrics[payload_key] = value
|
||||
return metrics
|
||||
|
||||
|
||||
def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
|
||||
"""Persist telemetry metrics extracted from a packet.
|
||||
|
||||
@@ -120,6 +306,14 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
|
||||
_aq = telemetry_section.get("airQualityMetrics") or telemetry_section.get(
|
||||
"air_quality_metrics"
|
||||
)
|
||||
_ls = telemetry_section.get("localStats") or telemetry_section.get("local_stats")
|
||||
_hm = telemetry_section.get("healthMetrics") or telemetry_section.get(
|
||||
"health_metrics"
|
||||
)
|
||||
_ho = telemetry_section.get("hostMetrics") or telemetry_section.get("host_metrics")
|
||||
_tm = telemetry_section.get("trafficManagementStats") or telemetry_section.get(
|
||||
"traffic_management_stats"
|
||||
)
|
||||
# Priority order matters: deviceMetrics is checked first because the device
|
||||
# sub-object also carries a voltage field that overlaps with powerMetrics.
|
||||
# Meshtastic uses a protobuf oneof so only one sub-object can be populated per
|
||||
@@ -132,6 +326,14 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
|
||||
telemetry_type = "power"
|
||||
elif isinstance(_aq, Mapping):
|
||||
telemetry_type = "air_quality"
|
||||
elif isinstance(_ls, Mapping):
|
||||
telemetry_type = "local_stats"
|
||||
elif isinstance(_hm, Mapping):
|
||||
telemetry_type = "health"
|
||||
elif isinstance(_ho, Mapping):
|
||||
telemetry_type = "host"
|
||||
elif isinstance(_tm, Mapping):
|
||||
telemetry_type = "traffic"
|
||||
else:
|
||||
telemetry_type = None
|
||||
|
||||
@@ -190,6 +392,9 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
|
||||
"channel_utilization",
|
||||
"deviceMetrics.channelUtilization",
|
||||
"deviceMetrics.channel_utilization",
|
||||
# LocalStats repeats the utilisation gauges; reuse the column.
|
||||
"localStats.channelUtilization",
|
||||
"local_stats.channel_utilization",
|
||||
default=None,
|
||||
)
|
||||
)
|
||||
@@ -200,6 +405,8 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
|
||||
"air_util_tx",
|
||||
"deviceMetrics.airUtilTx",
|
||||
"deviceMetrics.air_util_tx",
|
||||
"localStats.airUtilTx",
|
||||
"local_stats.air_util_tx",
|
||||
default=None,
|
||||
)
|
||||
)
|
||||
@@ -210,6 +417,11 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
|
||||
"uptime_seconds",
|
||||
"deviceMetrics.uptimeSeconds",
|
||||
"deviceMetrics.uptime_seconds",
|
||||
# LocalStats and HostMetrics both report an uptime gauge.
|
||||
"localStats.uptimeSeconds",
|
||||
"local_stats.uptime_seconds",
|
||||
"hostMetrics.uptimeSeconds",
|
||||
"host_metrics.uptime_seconds",
|
||||
default=None,
|
||||
)
|
||||
)
|
||||
@@ -503,6 +715,10 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None:
|
||||
telemetry_payload["soil_moisture"] = soil_moisture
|
||||
if soil_temperature is not None:
|
||||
telemetry_payload["soil_temperature"] = soil_temperature
|
||||
# Extended families (power / air-quality / health / local / host / traffic
|
||||
# stats and the one-wire probe list) are table-driven; only present fields
|
||||
# are added, matching the conditional style above (TI-A1).
|
||||
telemetry_payload.update(_extract_extended_metrics(telemetry_section))
|
||||
if telemetry_type is not None:
|
||||
telemetry_payload["telemetry_type"] = telemetry_type
|
||||
|
||||
|
||||
@@ -38,6 +38,7 @@ from .messages import (
|
||||
_synthetic_node_dict,
|
||||
)
|
||||
from .position import _store_meshcore_position
|
||||
from .telemetry import _make_telemetry_handlers
|
||||
|
||||
|
||||
def _process_self_info(
|
||||
@@ -420,4 +421,7 @@ def _make_event_handlers(iface: _MeshcoreInterface, target: str | None) -> dict:
|
||||
"CONTACT_DELETED": on_contact_deleted,
|
||||
"RX_LOG_DATA": on_rx_log_data,
|
||||
"DISCONNECTED": on_disconnected,
|
||||
# Telemetry surfaces (TI-A3): contact/self telemetry pulls, status
|
||||
# responses, and the host battery event.
|
||||
**_make_telemetry_handlers(iface, _handlers),
|
||||
}
|
||||
|
||||
@@ -26,6 +26,7 @@ from .channels import _ensure_channel_names
|
||||
from .connection import _make_connection
|
||||
from .handlers import _make_event_handlers
|
||||
from .interface import ClosedBeforeConnectedError, _MeshcoreInterface
|
||||
from .telemetry import _telemetry_poll_loop
|
||||
|
||||
|
||||
async def _ensure_autoadd_eviction(mc) -> None:
|
||||
@@ -238,7 +239,17 @@ async def _run_meshcore(
|
||||
|
||||
await mc.start_auto_message_fetching()
|
||||
|
||||
await stop_event.wait()
|
||||
# Telemetry collection (TI-A3): host self reads over the companion
|
||||
# link plus round-robin contact pulls, cadence-bounded by config.
|
||||
poll_task = asyncio.create_task(_telemetry_poll_loop(mc, iface))
|
||||
try:
|
||||
await stop_event.wait()
|
||||
finally:
|
||||
poll_task.cancel()
|
||||
try:
|
||||
await poll_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
|
||||
except Exception as exc:
|
||||
if not connected_event.is_set():
|
||||
|
||||
@@ -0,0 +1,485 @@
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""MeshCore telemetry collection (TI-A3).
|
||||
|
||||
MeshCore surfaces telemetry three ways: unsolicited/tag-matched
|
||||
``TELEMETRY_RESPONSE`` events carrying a decoded CayenneLPP list,
|
||||
``STATUS_RESPONSE`` events carrying battery/uptime gauges, and the host-only
|
||||
``BATTERY`` event. Other nodes' telemetry is **pull-only** (there is no
|
||||
broadcast the library surfaces), so this module also provides the poll loop
|
||||
that round-robins the contact roster with ``req_telemetry_sync`` — one on-air
|
||||
request per :data:`~data.mesh_ingestor.config.MESHCORE_TELEMETRY_POLL_SECONDS`
|
||||
— and reads the host radio's own battery/sensors over the local companion
|
||||
link (no airtime). Every reading is normalised into the canonical telemetry
|
||||
packet shape and flows through the shared
|
||||
:func:`~data.mesh_ingestor.handlers.store_telemetry_packet` pipeline with
|
||||
``protocol="meshcore"``, preserving protocol parity (SPEC Invariant IV) and
|
||||
the local-LoRa apex (Invariant I).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from collections.abc import Mapping
|
||||
|
||||
from ... import config
|
||||
from .interface import _MeshcoreInterface
|
||||
from .messages import _derive_message_id
|
||||
|
||||
_TELEMETRY_NODE_COOLDOWN_SECONDS: int = 24 * 60 * 60
|
||||
"""Minimum seconds between telemetry polls of the same contact.
|
||||
|
||||
The 300-second poll tick bounds *total* airtime, but on a small roster the
|
||||
round-robin would revisit each node every ``roster_size × interval`` — far
|
||||
more often than telemetry freshness needs. This per-node cooldown caps every
|
||||
contact at one poll per 24 h (counted from the poll attempt, so unreachable
|
||||
nodes are not hammered either); when every contact is fresh the tick sends
|
||||
nothing. Deliberately a constant, not an environment knob."""
|
||||
|
||||
|
||||
_LPP_TYPE_NAMES: dict[int, str] = {
|
||||
101: "illuminance",
|
||||
103: "temperature",
|
||||
104: "humidity",
|
||||
115: "barometer",
|
||||
116: "voltage",
|
||||
117: "current",
|
||||
120: "percentage",
|
||||
}
|
||||
"""CayenneLPP numeric type codes mapped to the library's canonical names.
|
||||
|
||||
Only the types with a matching telemetry column are listed; entries the
|
||||
library already name-encodes (via its ``my_lpp_types`` JSON encoder) arrive as
|
||||
strings and bypass this table."""
|
||||
|
||||
_LPP_DEVICE_FIELDS: dict[str, str] = {
|
||||
"voltage": "voltage",
|
||||
"percentage": "batteryLevel",
|
||||
}
|
||||
"""LPP type name → ``deviceMetrics`` field for battery-style readings."""
|
||||
|
||||
_LPP_ENVIRONMENT_FIELDS: dict[str, str] = {
|
||||
"temperature": "temperature",
|
||||
"humidity": "relativeHumidity",
|
||||
"barometer": "barometricPressure",
|
||||
"current": "current",
|
||||
"illuminance": "lux",
|
||||
}
|
||||
"""LPP type name → ``environmentMetrics`` field for sensor readings."""
|
||||
|
||||
_LPP_VALUE_SCALERS: dict[str, float] = {
|
||||
# CayenneLPP current is in amps; Meshtastic's EnvironmentMetrics current
|
||||
# (and therefore the shared ``current`` column) is in milliamps. Scale so
|
||||
# one column never mixes units across protocols.
|
||||
"current": 1000.0,
|
||||
}
|
||||
"""LPP type name → multiplier applied before storing the value."""
|
||||
|
||||
|
||||
def _lpp_entry_parts(entry: Mapping) -> tuple[str | None, float | None]:
|
||||
"""Extract the canonical type name and numeric value from an LPP entry.
|
||||
|
||||
Parameters:
|
||||
entry: One ``{"channel", "type", "value"}`` mapping from the decoded
|
||||
``lpp`` list. ``type`` may be the library's name string or a raw
|
||||
numeric LPP type code; ``value`` must be numeric to be usable.
|
||||
|
||||
Returns:
|
||||
Tuple of ``(type_name, value)``; either element is ``None`` when the
|
||||
entry is malformed or the type is not one we map.
|
||||
"""
|
||||
raw_type = entry.get("type")
|
||||
if isinstance(raw_type, str):
|
||||
type_name: str | None = raw_type.strip().lower() or None
|
||||
elif isinstance(raw_type, (int, float)) and not isinstance(raw_type, bool):
|
||||
type_name = _LPP_TYPE_NAMES.get(int(raw_type))
|
||||
else:
|
||||
type_name = None
|
||||
|
||||
value = entry.get("value")
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
return type_name, None
|
||||
return type_name, float(value)
|
||||
|
||||
|
||||
def _lpp_to_telemetry_section(lpp) -> dict | None:
|
||||
"""Convert a decoded CayenneLPP list into a telemetry section.
|
||||
|
||||
Battery-style readings land under ``deviceMetrics`` and sensor readings
|
||||
under ``environmentMetrics``; both sub-objects may be present when a node
|
||||
reports mixed sensors (the flat extraction in ``store_telemetry_packet``
|
||||
persists every field regardless of the single ``telemetry_type`` stamp).
|
||||
|
||||
Parameters:
|
||||
lpp: Decoded LPP entry list from a ``TELEMETRY_RESPONSE`` payload.
|
||||
|
||||
Returns:
|
||||
Telemetry section dict, or ``None`` when nothing usable was mapped.
|
||||
"""
|
||||
if not isinstance(lpp, (list, tuple)):
|
||||
return None
|
||||
device: dict = {}
|
||||
environment: dict = {}
|
||||
for entry in lpp:
|
||||
if not isinstance(entry, Mapping):
|
||||
continue
|
||||
type_name, value = _lpp_entry_parts(entry)
|
||||
if type_name is None or value is None:
|
||||
continue
|
||||
value *= _LPP_VALUE_SCALERS.get(type_name, 1.0)
|
||||
if type_name in _LPP_DEVICE_FIELDS:
|
||||
device.setdefault(_LPP_DEVICE_FIELDS[type_name], value)
|
||||
elif type_name in _LPP_ENVIRONMENT_FIELDS:
|
||||
environment.setdefault(_LPP_ENVIRONMENT_FIELDS[type_name], value)
|
||||
else:
|
||||
config._debug_log(
|
||||
"Unmapped MeshCore LPP entry",
|
||||
context="meshcore.telemetry.lpp",
|
||||
lpp_type=entry.get("type"),
|
||||
)
|
||||
section: dict = {}
|
||||
if device:
|
||||
section["deviceMetrics"] = device
|
||||
if environment:
|
||||
section["environmentMetrics"] = environment
|
||||
return section or None
|
||||
|
||||
|
||||
def _millivolts_to_volts(value) -> float | None:
|
||||
"""Convert a millivolt gauge (``bat`` / ``level``) to volts.
|
||||
|
||||
Parameters:
|
||||
value: Millivolt reading as reported by MeshCore firmware.
|
||||
|
||||
Returns:
|
||||
Voltage in volts rounded to millivolt precision, or ``None`` when the
|
||||
input is not a positive number.
|
||||
"""
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
return None
|
||||
if value <= 0:
|
||||
return None
|
||||
return round(float(value) / 1000.0, 3)
|
||||
|
||||
|
||||
def _status_to_telemetry_section(status: Mapping) -> dict | None:
|
||||
"""Convert a ``STATUS_RESPONSE`` payload into a telemetry section.
|
||||
|
||||
Parameters:
|
||||
status: Parsed status dict (``parse_status``) with ``bat`` in
|
||||
millivolts and ``uptime`` in seconds.
|
||||
|
||||
Returns:
|
||||
Telemetry section with a ``deviceMetrics`` sub-object, or ``None``
|
||||
when the status carries no usable gauge.
|
||||
"""
|
||||
if not isinstance(status, Mapping):
|
||||
return None
|
||||
device: dict = {}
|
||||
voltage = _millivolts_to_volts(status.get("bat"))
|
||||
if voltage is not None:
|
||||
device["voltage"] = voltage
|
||||
uptime = status.get("uptime")
|
||||
if not isinstance(uptime, bool) and isinstance(uptime, (int, float)) and uptime > 0:
|
||||
device["uptimeSeconds"] = int(uptime)
|
||||
return {"deviceMetrics": device} if device else None
|
||||
|
||||
|
||||
def _resolve_event_node_id(iface: _MeshcoreInterface, pubkey_pre) -> str | None:
|
||||
"""Resolve an event's ``pubkey_pre`` to a canonical node id.
|
||||
|
||||
Contacts resolve through the roster snapshot; the host radio's own prefix
|
||||
(self-telemetry responses are tagged with it) resolves to the host node.
|
||||
|
||||
Parameters:
|
||||
iface: Active MeshCore interface.
|
||||
pubkey_pre: Six-byte (12 hex char) public-key prefix from the event.
|
||||
|
||||
Returns:
|
||||
Canonical ``!xxxxxxxx`` node id, or ``None`` when unknown.
|
||||
"""
|
||||
if not isinstance(pubkey_pre, str) or not pubkey_pre:
|
||||
return None
|
||||
node_id = iface.lookup_node_id(pubkey_pre)
|
||||
if node_id:
|
||||
return node_id
|
||||
self_info = getattr(iface, "_self_info_payload", None) or {}
|
||||
own_key = self_info.get("public_key", "")
|
||||
if isinstance(own_key, str) and own_key.lower().startswith(pubkey_pre.lower()):
|
||||
return iface.host_node_id
|
||||
return None
|
||||
|
||||
|
||||
def _queue_meshcore_telemetry(
|
||||
handlers: object, node_id: str | None, section: Mapping | None, kind: str
|
||||
) -> bool:
|
||||
"""Queue one normalised MeshCore telemetry packet.
|
||||
|
||||
Parameters:
|
||||
handlers: The ``data.mesh_ingestor.handlers`` module (passed in to
|
||||
avoid circular imports, matching the other MeshCore handlers).
|
||||
node_id: Canonical node the reading belongs to.
|
||||
section: Telemetry section (``deviceMetrics``/``environmentMetrics``).
|
||||
kind: Discriminator for the packet-id fingerprint (``"lpp"``,
|
||||
``"status"``, ``"battery"``) so distinct sources heard in the same
|
||||
second cannot collide, while re-reads of the same source collapse
|
||||
into one row via the web app's ``telemetry.id`` upsert.
|
||||
|
||||
Returns:
|
||||
``True`` when a packet was queued, ``False`` when skipped.
|
||||
"""
|
||||
if not node_id or not section:
|
||||
return False
|
||||
rx_time = int(time.time())
|
||||
packet = {
|
||||
"id": _derive_message_id(node_id, rx_time, f"tel-{kind}", ""),
|
||||
"rxTime": rx_time,
|
||||
"rx_time": rx_time,
|
||||
"fromId": node_id,
|
||||
"from_id": node_id,
|
||||
"protocol": "meshcore",
|
||||
"decoded": {
|
||||
"portnum": "TELEMETRY_APP",
|
||||
"telemetry": {**section, "time": rx_time},
|
||||
},
|
||||
}
|
||||
handlers._mark_packet_seen()
|
||||
handlers.store_packet_dict(packet)
|
||||
config._debug_log(
|
||||
"MeshCore telemetry queued",
|
||||
context="meshcore.telemetry",
|
||||
node_id=node_id,
|
||||
kind=kind,
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def _make_telemetry_handlers(iface: _MeshcoreInterface, handlers: object) -> dict:
|
||||
"""Build the telemetry-related MeshCore event callbacks.
|
||||
|
||||
Parameters:
|
||||
iface: Active MeshCore interface (node-id resolution + host id).
|
||||
handlers: The ``data.mesh_ingestor.handlers`` module.
|
||||
|
||||
Returns:
|
||||
Mapping of ``EventType`` member name → async callback, merged into the
|
||||
provider's main handler map by ``_make_event_handlers``.
|
||||
"""
|
||||
|
||||
async def on_telemetry_response(evt) -> None:
|
||||
payload = evt.payload or {}
|
||||
node_id = _resolve_event_node_id(iface, payload.get("pubkey_pre"))
|
||||
_queue_meshcore_telemetry(
|
||||
handlers, node_id, _lpp_to_telemetry_section(payload.get("lpp")), "lpp"
|
||||
)
|
||||
|
||||
async def on_status_response(evt) -> None:
|
||||
payload = evt.payload or {}
|
||||
node_id = _resolve_event_node_id(iface, payload.get("pubkey_pre"))
|
||||
_queue_meshcore_telemetry(
|
||||
handlers, node_id, _status_to_telemetry_section(payload), "status"
|
||||
)
|
||||
|
||||
async def on_battery(evt) -> None:
|
||||
# BATTERY is host-only: the response to get_bat() on the companion link.
|
||||
payload = evt.payload or {}
|
||||
voltage = _millivolts_to_volts(payload.get("level"))
|
||||
section = {"deviceMetrics": {"voltage": voltage}} if voltage else None
|
||||
_queue_meshcore_telemetry(handlers, iface.host_node_id, section, "battery")
|
||||
|
||||
return {
|
||||
"TELEMETRY_RESPONSE": on_telemetry_response,
|
||||
"STATUS_RESPONSE": on_status_response,
|
||||
"BATTERY": on_battery,
|
||||
}
|
||||
|
||||
|
||||
def _next_poll_contact(iface: _MeshcoreInterface, state: dict) -> dict | None:
|
||||
"""Pick the next roster contact due for a telemetry poll, round-robin.
|
||||
|
||||
Contacts polled within :data:`_TELEMETRY_NODE_COOLDOWN_SECONDS` are
|
||||
skipped; the returned contact is stamped as polled immediately (before the
|
||||
request is sent), so failed or timed-out polls honour the cooldown too.
|
||||
Departed roster entries are pruned from the stamp table so a long-running
|
||||
process cannot accumulate stale state.
|
||||
|
||||
Parameters:
|
||||
iface: Active MeshCore interface holding the contact snapshot.
|
||||
state: Mutable poll-loop state carrying the ``cursor`` position and
|
||||
the per-contact ``last_polled`` monotonic stamps.
|
||||
|
||||
Returns:
|
||||
The next due contact dict, or ``None`` when the roster is empty or
|
||||
every contact is still inside its cooldown window.
|
||||
"""
|
||||
with iface._contacts_lock:
|
||||
contacts = [iface._contacts[key] for key in sorted(iface._contacts)]
|
||||
if not contacts:
|
||||
return None
|
||||
last_polled = state.setdefault("last_polled", {})
|
||||
roster_keys = {contact.get("public_key") for contact in contacts}
|
||||
for key in list(last_polled):
|
||||
if key not in roster_keys:
|
||||
del last_polled[key]
|
||||
now = time.monotonic()
|
||||
cursor = state.get("cursor", 0)
|
||||
for offset in range(len(contacts)):
|
||||
index = (cursor + offset) % len(contacts)
|
||||
contact = contacts[index]
|
||||
key = contact.get("public_key")
|
||||
stamp = last_polled.get(key)
|
||||
if stamp is not None and now - stamp < _TELEMETRY_NODE_COOLDOWN_SECONDS:
|
||||
continue
|
||||
state["cursor"] = index + 1
|
||||
last_polled[key] = now
|
||||
return contact
|
||||
return None
|
||||
|
||||
|
||||
async def _poll_self_telemetry(mc, iface: _MeshcoreInterface, handlers: object) -> None:
|
||||
"""Read the host radio's battery and sensors over the companion link.
|
||||
|
||||
Both commands return their matching event object directly; the payload is
|
||||
processed inline (subscriptions also fire for these events — the derived
|
||||
packet id collapses the duplicate). Errors are logged and swallowed so a
|
||||
firmware without a command never kills the poll loop.
|
||||
|
||||
Parameters:
|
||||
mc: Connected MeshCore instance.
|
||||
iface: Active interface (host node id).
|
||||
handlers: The ``data.mesh_ingestor.handlers`` module.
|
||||
"""
|
||||
try:
|
||||
evt = await mc.commands.get_bat()
|
||||
payload = getattr(evt, "payload", None) or {}
|
||||
voltage = _millivolts_to_volts(payload.get("level"))
|
||||
section = {"deviceMetrics": {"voltage": voltage}} if voltage else None
|
||||
_queue_meshcore_telemetry(handlers, iface.host_node_id, section, "battery")
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
"MeshCore self battery read failed",
|
||||
context="meshcore.telemetry.self",
|
||||
severity="warning",
|
||||
error=str(exc),
|
||||
)
|
||||
try:
|
||||
evt = await mc.commands.get_self_telemetry()
|
||||
payload = getattr(evt, "payload", None) or {}
|
||||
_queue_meshcore_telemetry(
|
||||
handlers,
|
||||
iface.host_node_id,
|
||||
_lpp_to_telemetry_section(payload.get("lpp")),
|
||||
"lpp",
|
||||
)
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
"MeshCore self telemetry read failed",
|
||||
context="meshcore.telemetry.self",
|
||||
severity="warning",
|
||||
error=str(exc),
|
||||
)
|
||||
|
||||
|
||||
async def _poll_contact_telemetry(
|
||||
mc, iface: _MeshcoreInterface, handlers: object, state: dict
|
||||
) -> None:
|
||||
"""Send one on-air telemetry pull to the next roster contact.
|
||||
|
||||
Falls back to a status request when the telemetry pull yields nothing, so
|
||||
sensor-less nodes still report battery/uptime. One contact per call keeps
|
||||
airtime bounded to a single request per poll interval regardless of roster
|
||||
size; the meshcore library serialises mesh requests internally.
|
||||
|
||||
Parameters:
|
||||
mc: Connected MeshCore instance.
|
||||
iface: Active interface (roster + node-id resolution).
|
||||
handlers: The ``data.mesh_ingestor.handlers`` module.
|
||||
state: Mutable poll-loop state (round-robin cursor).
|
||||
"""
|
||||
contact = _next_poll_contact(iface, state)
|
||||
if contact is None:
|
||||
return
|
||||
node_id = iface.lookup_node_id((contact.get("public_key") or "")[:12])
|
||||
if node_id is None:
|
||||
return
|
||||
try:
|
||||
lpp = await mc.commands.req_telemetry_sync(contact)
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
"MeshCore contact telemetry poll failed",
|
||||
context="meshcore.telemetry.poll",
|
||||
node_id=node_id,
|
||||
error=str(exc),
|
||||
)
|
||||
return
|
||||
if _queue_meshcore_telemetry(
|
||||
handlers, node_id, _lpp_to_telemetry_section(lpp), "lpp"
|
||||
):
|
||||
return
|
||||
try:
|
||||
status = await mc.commands.req_status_sync(contact)
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
"MeshCore contact status poll failed",
|
||||
context="meshcore.telemetry.poll",
|
||||
node_id=node_id,
|
||||
error=str(exc),
|
||||
)
|
||||
return
|
||||
_queue_meshcore_telemetry(
|
||||
handlers, node_id, _status_to_telemetry_section(status), "status"
|
||||
)
|
||||
|
||||
|
||||
async def _telemetry_poll_loop(mc, iface: _MeshcoreInterface) -> None:
|
||||
"""Drive periodic self and contact telemetry collection until cancelled.
|
||||
|
||||
Cadence comes from :data:`~data.mesh_ingestor.config` —
|
||||
``MESHCORE_SELF_TELEMETRY_SECONDS`` (local, no airtime; ``<= 0`` disables)
|
||||
and ``MESHCORE_TELEMETRY_POLL_SECONDS`` (one on-air request per interval;
|
||||
``<= 0`` disables), with each contact additionally capped by the fixed
|
||||
per-node cooldown (:data:`_TELEMETRY_NODE_COOLDOWN_SECONDS`). ``RX_ONLY``
|
||||
forbids every ingestor-initiated transmission, so it disables the on-air
|
||||
contact polls regardless of the poll interval; the self reads are local
|
||||
companion-link commands and stay active. The loop wakes once per
|
||||
second-granularity deadline rather than busy-polling.
|
||||
|
||||
Parameters:
|
||||
mc: Connected MeshCore instance.
|
||||
iface: Active interface.
|
||||
"""
|
||||
from ... import handlers as _handlers
|
||||
|
||||
self_interval = config.MESHCORE_SELF_TELEMETRY_SECONDS
|
||||
poll_interval = 0 if config.RX_ONLY else config.MESHCORE_TELEMETRY_POLL_SECONDS
|
||||
if self_interval <= 0 and poll_interval <= 0:
|
||||
return
|
||||
|
||||
state: dict = {}
|
||||
next_self = time.monotonic() if self_interval > 0 else None
|
||||
# Delay the first on-air poll by one full interval so a restart storm
|
||||
# cannot burst-request the roster.
|
||||
next_poll = time.monotonic() + poll_interval if poll_interval > 0 else None
|
||||
while True:
|
||||
now = time.monotonic()
|
||||
if next_self is not None and now >= next_self:
|
||||
await _poll_self_telemetry(mc, iface, _handlers)
|
||||
next_self = now + self_interval
|
||||
if next_poll is not None and now >= next_poll:
|
||||
await _poll_contact_telemetry(mc, iface, _handlers, state)
|
||||
next_poll = now + poll_interval
|
||||
deadlines = [d for d in (next_self, next_poll) if d is not None]
|
||||
await asyncio.sleep(max(1.0, min(deadlines) - time.monotonic()))
|
||||
@@ -0,0 +1,82 @@
|
||||
-- Copyright © 2025-26 l5yth & contributors
|
||||
-- Licensed under the Apache License, Version 2.0 (see LICENSE)
|
||||
--
|
||||
-- Reference migration for pre-existing databases (TI-A2): adds the extended
|
||||
-- telemetry metric columns (PowerMetrics, AirQualityMetrics, HealthMetrics,
|
||||
-- LocalStats, HostMetrics, TrafficManagementStats, one-wire probe list).
|
||||
-- The web app applies the same additions automatically at boot via
|
||||
-- ensure_schema_upgrades (web/lib/potato_mesh/application/database.rb); this
|
||||
-- script exists for operators migrating a database by hand.
|
||||
|
||||
ALTER TABLE telemetry ADD COLUMN ch1_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch1_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch2_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch2_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch3_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch3_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch4_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch4_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch5_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch5_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch6_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch6_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch7_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch7_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch8_voltage REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN ch8_current REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN pm10_standard INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN pm25_standard INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN pm100_standard INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN pm40_standard INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN pm10_environmental INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN pm25_environmental INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN pm100_environmental INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_03um INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_05um INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_10um INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_25um INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_40um INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_50um INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_100um INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN particles_tps REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN co2 INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN co2_temperature REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN co2_humidity REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN form_formaldehyde REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN form_humidity REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN form_temperature REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN pm_temperature REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN pm_humidity REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN pm_voc_idx REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN pm_nox_idx REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN heart_bpm INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN spo2 INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN health_temperature REAL;
|
||||
ALTER TABLE telemetry ADD COLUMN num_packets_tx INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_packets_rx INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_packets_rx_bad INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_online_nodes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_total_nodes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_rx_dupe INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_tx_relay INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_tx_relay_canceled INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN heap_total_bytes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN heap_free_bytes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN num_tx_dropped INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN noise_floor INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN freemem_bytes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN diskfree1_bytes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN diskfree2_bytes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN diskfree3_bytes INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN load1 INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN load5 INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN load15 INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN user_string TEXT;
|
||||
ALTER TABLE telemetry ADD COLUMN packets_inspected INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN position_dedup_drops INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN nodeinfo_cache_hits INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN rate_limit_drops INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN unknown_packet_drops INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN hop_exhausted_packets INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN router_hops_preserved INTEGER;
|
||||
ALTER TABLE telemetry ADD COLUMN one_wire_temperature TEXT;
|
||||
+80
-1
@@ -56,7 +56,86 @@ CREATE TABLE IF NOT EXISTS telemetry (
|
||||
soil_temperature REAL,
|
||||
ingestor TEXT,
|
||||
protocol TEXT NOT NULL DEFAULT 'meshtastic',
|
||||
telemetry_type TEXT
|
||||
telemetry_type TEXT,
|
||||
-- PowerMetrics (TI-A2): per-channel INA voltage/current readings.
|
||||
ch1_voltage REAL,
|
||||
ch1_current REAL,
|
||||
ch2_voltage REAL,
|
||||
ch2_current REAL,
|
||||
ch3_voltage REAL,
|
||||
ch3_current REAL,
|
||||
ch4_voltage REAL,
|
||||
ch4_current REAL,
|
||||
ch5_voltage REAL,
|
||||
ch5_current REAL,
|
||||
ch6_voltage REAL,
|
||||
ch6_current REAL,
|
||||
ch7_voltage REAL,
|
||||
ch7_current REAL,
|
||||
ch8_voltage REAL,
|
||||
ch8_current REAL,
|
||||
-- AirQualityMetrics (TI-A2): PM mass/particle series, CO2, formaldehyde, VOC/NOx.
|
||||
pm10_standard INTEGER,
|
||||
pm25_standard INTEGER,
|
||||
pm100_standard INTEGER,
|
||||
pm40_standard INTEGER,
|
||||
pm10_environmental INTEGER,
|
||||
pm25_environmental INTEGER,
|
||||
pm100_environmental INTEGER,
|
||||
particles_03um INTEGER,
|
||||
particles_05um INTEGER,
|
||||
particles_10um INTEGER,
|
||||
particles_25um INTEGER,
|
||||
particles_40um INTEGER,
|
||||
particles_50um INTEGER,
|
||||
particles_100um INTEGER,
|
||||
particles_tps REAL,
|
||||
co2 INTEGER,
|
||||
co2_temperature REAL,
|
||||
co2_humidity REAL,
|
||||
form_formaldehyde REAL,
|
||||
form_humidity REAL,
|
||||
form_temperature REAL,
|
||||
pm_temperature REAL,
|
||||
pm_humidity REAL,
|
||||
pm_voc_idx REAL,
|
||||
pm_nox_idx REAL,
|
||||
-- HealthMetrics (TI-A2): body sensors; kept apart from ambient temperature.
|
||||
heart_bpm INTEGER,
|
||||
spo2 INTEGER,
|
||||
health_temperature REAL,
|
||||
-- LocalStats (TI-A2): radio/network counters.
|
||||
num_packets_tx INTEGER,
|
||||
num_packets_rx INTEGER,
|
||||
num_packets_rx_bad INTEGER,
|
||||
num_online_nodes INTEGER,
|
||||
num_total_nodes INTEGER,
|
||||
num_rx_dupe INTEGER,
|
||||
num_tx_relay INTEGER,
|
||||
num_tx_relay_canceled INTEGER,
|
||||
heap_total_bytes INTEGER,
|
||||
heap_free_bytes INTEGER,
|
||||
num_tx_dropped INTEGER,
|
||||
noise_floor INTEGER,
|
||||
-- HostMetrics (TI-A2): companion-host gauges.
|
||||
freemem_bytes INTEGER,
|
||||
diskfree1_bytes INTEGER,
|
||||
diskfree2_bytes INTEGER,
|
||||
diskfree3_bytes INTEGER,
|
||||
load1 INTEGER,
|
||||
load5 INTEGER,
|
||||
load15 INTEGER,
|
||||
user_string TEXT,
|
||||
-- TrafficManagementStats (TI-A2): router traffic counters.
|
||||
packets_inspected INTEGER,
|
||||
position_dedup_drops INTEGER,
|
||||
nodeinfo_cache_hits INTEGER,
|
||||
rate_limit_drops INTEGER,
|
||||
unknown_packet_drops INTEGER,
|
||||
hop_exhausted_packets INTEGER,
|
||||
router_hops_preserved INTEGER,
|
||||
-- EnvironmentMetrics repeated one-wire probe list (JSON float array).
|
||||
one_wire_temperature TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_telemetry_rx_time ON telemetry(rx_time);
|
||||
|
||||
@@ -811,6 +811,193 @@ class TestStoreTelemetryPacket:
|
||||
assert "telemetry_type" not in payload
|
||||
|
||||
|
||||
class TestExtendedTelemetryExtraction:
|
||||
"""Regression guards for TI-A1: every Meshtastic telemetry family is
|
||||
extracted at ingest instead of being dropped (power, air-quality, health,
|
||||
local/host/traffic stats, and the repeated one-wire probe list)."""
|
||||
|
||||
def _queued_payload(self, telemetry, pkt_id=3001):
|
||||
"""Run ``store_telemetry_packet`` for *telemetry* and return the queued
|
||||
POST payload dict.
|
||||
|
||||
Parameters:
|
||||
telemetry: The decoded ``telemetry`` section for the packet.
|
||||
pkt_id: Packet id to stamp on the synthetic packet.
|
||||
|
||||
Returns:
|
||||
The payload queued to ``/api/telemetry``.
|
||||
"""
|
||||
import data.mesh_ingestor.queue as q
|
||||
|
||||
sent = []
|
||||
original = q._queue_post_json
|
||||
q._queue_post_json = lambda path, payload, *, priority, **kw: sent.append(
|
||||
(path, payload)
|
||||
)
|
||||
try:
|
||||
pkt = {
|
||||
"id": pkt_id,
|
||||
"rxTime": 1_700_000_000,
|
||||
"fromId": "!aabbccdd",
|
||||
"decoded": {"portnum": "TELEMETRY_APP", "telemetry": telemetry},
|
||||
}
|
||||
handlers.store_telemetry_packet(pkt, pkt["decoded"])
|
||||
finally:
|
||||
q._queue_post_json = original
|
||||
assert sent, "telemetry packet was not queued at all"
|
||||
return sent[0][1]
|
||||
|
||||
def test_power_metrics_extracted(self):
|
||||
"""PowerMetrics channel readings survive into the queued payload."""
|
||||
payload = self._queued_payload(
|
||||
{
|
||||
"time": 1_700_000_000,
|
||||
"powerMetrics": {
|
||||
"ch1Voltage": 3.94,
|
||||
"ch1Current": 121.5,
|
||||
"ch3Voltage": 11.9,
|
||||
"ch8Current": 0.25,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert payload.get("telemetry_type") == "power"
|
||||
assert payload.get("ch1_voltage") == 3.94
|
||||
assert payload.get("ch1_current") == 121.5
|
||||
assert payload.get("ch3_voltage") == 11.9
|
||||
assert payload.get("ch8_current") == 0.25
|
||||
|
||||
def test_air_quality_metrics_extracted(self):
|
||||
"""AirQualityMetrics PM / particle / CO2 / VOC readings are extracted."""
|
||||
payload = self._queued_payload(
|
||||
{
|
||||
"time": 1_700_000_000,
|
||||
"airQualityMetrics": {
|
||||
"pm25Standard": 8,
|
||||
"pm10Environmental": 5,
|
||||
"pm40Standard": 3,
|
||||
"particles03um": 1200,
|
||||
"particles40um": 40,
|
||||
"co2": 700,
|
||||
"co2Temperature": 24.5,
|
||||
"formFormaldehyde": 0.03,
|
||||
"pmVocIdx": 110.0,
|
||||
"particlesTps": 1.5,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert payload.get("telemetry_type") == "air_quality"
|
||||
assert payload.get("pm25_standard") == 8
|
||||
assert payload.get("pm10_environmental") == 5
|
||||
assert payload.get("pm40_standard") == 3
|
||||
assert payload.get("particles_03um") == 1200
|
||||
assert payload.get("particles_40um") == 40
|
||||
assert payload.get("co2") == 700
|
||||
assert payload.get("co2_temperature") == 24.5
|
||||
assert payload.get("form_formaldehyde") == 0.03
|
||||
assert payload.get("pm_voc_idx") == 110.0
|
||||
assert payload.get("particles_tps") == 1.5
|
||||
|
||||
def test_health_metrics_extracted(self):
|
||||
"""HealthMetrics values map to dedicated columns; body temperature
|
||||
never masquerades as the ambient ``temperature`` metric."""
|
||||
payload = self._queued_payload(
|
||||
{
|
||||
"time": 1_700_000_000,
|
||||
"healthMetrics": {"heartBpm": 72, "spO2": 97, "temperature": 36.6},
|
||||
}
|
||||
)
|
||||
assert payload.get("telemetry_type") == "health"
|
||||
assert payload.get("heart_bpm") == 72
|
||||
assert payload.get("spo2") == 97
|
||||
assert payload.get("health_temperature") == 36.6
|
||||
assert "temperature" not in payload
|
||||
|
||||
def test_local_stats_extracted(self):
|
||||
"""LocalStats counters are extracted; the shared uptime / utilisation
|
||||
fields reuse the existing columns."""
|
||||
payload = self._queued_payload(
|
||||
{
|
||||
"time": 1_700_000_000,
|
||||
"localStats": {
|
||||
"uptimeSeconds": 3600,
|
||||
"channelUtilization": 12.5,
|
||||
"airUtilTx": 3.2,
|
||||
"numPacketsTx": 10,
|
||||
"numPacketsRx": 42,
|
||||
"numRxDupe": 2,
|
||||
"heapTotalBytes": 200_000,
|
||||
"heapFreeBytes": 80_000,
|
||||
"noiseFloor": -95,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert payload.get("telemetry_type") == "local_stats"
|
||||
assert payload.get("uptime_seconds") == 3600
|
||||
assert payload.get("channel_utilization") == 12.5
|
||||
assert payload.get("air_util_tx") == 3.2
|
||||
assert payload.get("num_packets_tx") == 10
|
||||
assert payload.get("num_packets_rx") == 42
|
||||
assert payload.get("num_rx_dupe") == 2
|
||||
assert payload.get("heap_total_bytes") == 200_000
|
||||
assert payload.get("heap_free_bytes") == 80_000
|
||||
assert payload.get("noise_floor") == -95
|
||||
|
||||
def test_host_metrics_extracted(self):
|
||||
"""HostMetrics gauges (including the free-text user string) survive."""
|
||||
payload = self._queued_payload(
|
||||
{
|
||||
"time": 1_700_000_000,
|
||||
"hostMetrics": {
|
||||
"uptimeSeconds": 86_400,
|
||||
"freememBytes": 1_048_576,
|
||||
"diskfree1Bytes": 2**33,
|
||||
"load1": 35,
|
||||
"load15": 12,
|
||||
"userString": "potato",
|
||||
},
|
||||
}
|
||||
)
|
||||
assert payload.get("telemetry_type") == "host"
|
||||
assert payload.get("uptime_seconds") == 86_400
|
||||
assert payload.get("freemem_bytes") == 1_048_576
|
||||
assert payload.get("diskfree1_bytes") == 2**33
|
||||
assert payload.get("load1") == 35
|
||||
assert payload.get("load15") == 12
|
||||
assert payload.get("user_string") == "potato"
|
||||
|
||||
def test_traffic_stats_extracted(self):
|
||||
"""TrafficManagementStats counters are extracted."""
|
||||
payload = self._queued_payload(
|
||||
{
|
||||
"time": 1_700_000_000,
|
||||
"trafficManagementStats": {
|
||||
"packetsInspected": 100,
|
||||
"rateLimitDrops": 3,
|
||||
"routerHopsPreserved": 7,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert payload.get("telemetry_type") == "traffic"
|
||||
assert payload.get("packets_inspected") == 100
|
||||
assert payload.get("rate_limit_drops") == 3
|
||||
assert payload.get("router_hops_preserved") == 7
|
||||
|
||||
def test_one_wire_temperature_extracted(self):
|
||||
"""The repeated one-wire probe list is preserved as a float list."""
|
||||
payload = self._queued_payload(
|
||||
{
|
||||
"time": 1_700_000_000,
|
||||
"environmentMetrics": {
|
||||
"temperature": 21.5,
|
||||
"oneWireTemperature": [20.0, 21.25],
|
||||
},
|
||||
}
|
||||
)
|
||||
assert payload.get("telemetry_type") == "environment"
|
||||
assert payload.get("temperature") == 21.5
|
||||
assert payload.get("one_wire_temperature") == [20.0, 21.25]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# store_nodeinfo_packet
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -1492,6 +1492,541 @@ def _setup_channel_msg_handlers(monkeypatch, *, contacts=None):
|
||||
return captured, upserted, iface, hmap
|
||||
|
||||
|
||||
def test_event_handlers_cover_telemetry_events(monkeypatch):
|
||||
"""Regression guard for TI-A3: the MeshCore event-handler map subscribes
|
||||
the telemetry surfaces (contact telemetry pulls, status responses, and the
|
||||
host radio's battery event) instead of dropping them unhandled."""
|
||||
_, _, _, hmap = _setup_channel_msg_handlers(monkeypatch)
|
||||
missing = {"TELEMETRY_RESPONSE", "STATUS_RESPONSE", "BATTERY"} - set(hmap)
|
||||
assert not missing, f"telemetry events not subscribed: {sorted(missing)}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MeshCore telemetry collection (TI-A3)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_TEST_CONTACT_KEY = "aabbccddeeff" + "00" * 26
|
||||
"""Full 32-byte public key (hex) for the telemetry test contact."""
|
||||
|
||||
|
||||
def _telemetry_module():
|
||||
"""Return the MeshCore telemetry module under test."""
|
||||
import data.mesh_ingestor.protocols.meshcore.telemetry as mc_tel
|
||||
|
||||
return mc_tel
|
||||
|
||||
|
||||
def _telemetry_env(monkeypatch, *, contacts=None, frozen_time=1_700_000_000):
|
||||
"""Build the patched environment for MeshCore telemetry tests.
|
||||
|
||||
Parameters:
|
||||
monkeypatch: pytest monkeypatch fixture.
|
||||
contacts: Optional contact dicts pre-registered on the interface.
|
||||
frozen_time: Wall-clock second ``time.time`` is pinned to.
|
||||
|
||||
Returns:
|
||||
Tuple ``(mc_tel, iface, stub, captured)`` — module under test, the
|
||||
interface, the stubbed handlers module, and the captured packet list.
|
||||
"""
|
||||
import time as _time
|
||||
|
||||
mc_tel = _telemetry_module()
|
||||
captured: list = []
|
||||
stub = _make_stub_handlers_module()
|
||||
stub.store_packet_dict = lambda pkt: captured.append(pkt)
|
||||
monkeypatch.setattr(mc_tel.config, "_debug_log", lambda *_a, **_k: None)
|
||||
monkeypatch.setattr(_time, "time", lambda: frozen_time)
|
||||
iface = _MeshcoreInterface(target=None)
|
||||
for contact in contacts or []:
|
||||
iface._update_contact(contact)
|
||||
return mc_tel, iface, stub, captured
|
||||
|
||||
|
||||
def test_lpp_entry_parts_accepts_names_codes_and_rejects_junk():
|
||||
"""LPP entries resolve by name string or numeric code; junk is rejected."""
|
||||
mc_tel = _telemetry_module()
|
||||
assert mc_tel._lpp_entry_parts({"type": "Temperature", "value": 21.5}) == (
|
||||
"temperature",
|
||||
21.5,
|
||||
)
|
||||
assert mc_tel._lpp_entry_parts({"type": 116, "value": 3.9}) == ("voltage", 3.9)
|
||||
assert mc_tel._lpp_entry_parts({"type": 999, "value": 1.0}) == (None, 1.0)
|
||||
assert mc_tel._lpp_entry_parts({"type": " ", "value": 1.0}) == (None, 1.0)
|
||||
assert mc_tel._lpp_entry_parts({"type": None, "value": 1.0}) == (None, 1.0)
|
||||
assert mc_tel._lpp_entry_parts({"type": "voltage", "value": True}) == (
|
||||
"voltage",
|
||||
None,
|
||||
)
|
||||
assert mc_tel._lpp_entry_parts({"type": "voltage", "value": "n/a"}) == (
|
||||
"voltage",
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
def test_lpp_to_telemetry_section_maps_device_and_environment(monkeypatch):
|
||||
"""Battery-style readings map to deviceMetrics, sensors to environmentMetrics."""
|
||||
mc_tel = _telemetry_module()
|
||||
monkeypatch.setattr(mc_tel.config, "_debug_log", lambda *_a, **_k: None)
|
||||
section = mc_tel._lpp_to_telemetry_section(
|
||||
[
|
||||
{"channel": 1, "type": "voltage", "value": 4.05},
|
||||
{"channel": 1, "type": "percentage", "value": 87},
|
||||
{"channel": 2, "type": "temperature", "value": 21.5},
|
||||
{"channel": 2, "type": "humidity", "value": 40.2},
|
||||
{"channel": 2, "type": "barometer", "value": 1013.2},
|
||||
{"channel": 3, "type": "illuminance", "value": 120.0},
|
||||
{"channel": 3, "type": "current", "value": 0.12},
|
||||
{"channel": 9, "type": "gps", "value": {"lat": 1}},
|
||||
"not-a-mapping",
|
||||
]
|
||||
)
|
||||
assert section == {
|
||||
"deviceMetrics": {"voltage": 4.05, "batteryLevel": 87.0},
|
||||
"environmentMetrics": {
|
||||
"temperature": 21.5,
|
||||
"relativeHumidity": 40.2,
|
||||
"barometricPressure": 1013.2,
|
||||
"lux": 120.0,
|
||||
# LPP current arrives in amps and is scaled to the column's
|
||||
# milliamp convention (Meshtastic EnvironmentMetrics unit).
|
||||
"current": 120.0,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_lpp_to_telemetry_section_rejects_empty_and_non_lists(monkeypatch):
|
||||
"""Unusable LPP inputs yield None (nothing queued downstream)."""
|
||||
mc_tel = _telemetry_module()
|
||||
monkeypatch.setattr(mc_tel.config, "_debug_log", lambda *_a, **_k: None)
|
||||
assert mc_tel._lpp_to_telemetry_section(None) is None
|
||||
assert mc_tel._lpp_to_telemetry_section({"type": "temperature"}) is None
|
||||
assert mc_tel._lpp_to_telemetry_section([]) is None
|
||||
assert mc_tel._lpp_to_telemetry_section([{"type": "gps", "value": 1.0}]) is None
|
||||
|
||||
|
||||
def test_lpp_duplicate_types_keep_first_reading(monkeypatch):
|
||||
"""setdefault semantics: the first reading of a type wins within a packet."""
|
||||
mc_tel = _telemetry_module()
|
||||
monkeypatch.setattr(mc_tel.config, "_debug_log", lambda *_a, **_k: None)
|
||||
section = mc_tel._lpp_to_telemetry_section(
|
||||
[
|
||||
{"type": "temperature", "value": 21.5},
|
||||
{"type": "temperature", "value": 99.0},
|
||||
]
|
||||
)
|
||||
assert section == {"environmentMetrics": {"temperature": 21.5}}
|
||||
|
||||
|
||||
def test_millivolts_to_volts_bounds():
|
||||
"""mV gauges convert to volts; non-positive and junk values are rejected."""
|
||||
mc_tel = _telemetry_module()
|
||||
assert mc_tel._millivolts_to_volts(4056) == 4.056
|
||||
assert mc_tel._millivolts_to_volts(0) is None
|
||||
assert mc_tel._millivolts_to_volts(-5) is None
|
||||
assert mc_tel._millivolts_to_volts(True) is None
|
||||
assert mc_tel._millivolts_to_volts("4056") is None
|
||||
|
||||
|
||||
def test_status_to_telemetry_section_maps_battery_and_uptime():
|
||||
"""STATUS_RESPONSE bat/uptime map to deviceMetrics voltage/uptimeSeconds."""
|
||||
mc_tel = _telemetry_module()
|
||||
assert mc_tel._status_to_telemetry_section({"bat": 4056, "uptime": 3600}) == {
|
||||
"deviceMetrics": {"voltage": 4.056, "uptimeSeconds": 3600}
|
||||
}
|
||||
assert mc_tel._status_to_telemetry_section({"bat": 4056}) == {
|
||||
"deviceMetrics": {"voltage": 4.056}
|
||||
}
|
||||
assert mc_tel._status_to_telemetry_section({"uptime": 0, "bat": 0}) is None
|
||||
assert mc_tel._status_to_telemetry_section({"uptime": True}) is None
|
||||
assert mc_tel._status_to_telemetry_section("junk") is None
|
||||
|
||||
|
||||
def test_resolve_event_node_id_roster_host_and_unknown(monkeypatch):
|
||||
"""pubkey_pre resolves via roster, then host prefix, else None."""
|
||||
mc_tel, iface, _stub, _captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
roster_id = iface.lookup_node_id(_TEST_CONTACT_KEY[:12])
|
||||
assert mc_tel._resolve_event_node_id(iface, _TEST_CONTACT_KEY[:12]) == roster_id
|
||||
|
||||
iface.host_node_id = "!deadbeef"
|
||||
iface._self_info_payload = {"public_key": "FFEE" + "11" * 30}
|
||||
assert mc_tel._resolve_event_node_id(iface, "ffee1111") == "!deadbeef"
|
||||
|
||||
assert mc_tel._resolve_event_node_id(iface, "0123456789ab") is None
|
||||
assert mc_tel._resolve_event_node_id(iface, "") is None
|
||||
assert mc_tel._resolve_event_node_id(iface, None) is None
|
||||
|
||||
|
||||
def test_queue_meshcore_telemetry_packet_shape(monkeypatch):
|
||||
"""Queued packets carry the canonical shape, protocol stamp, and stable id."""
|
||||
mc_tel, iface, stub, captured = _telemetry_env(monkeypatch)
|
||||
seen = []
|
||||
stub._mark_packet_seen = lambda: seen.append(True)
|
||||
|
||||
queued = mc_tel._queue_meshcore_telemetry(
|
||||
stub, "!11223344", {"deviceMetrics": {"voltage": 4.05}}, "battery"
|
||||
)
|
||||
assert queued is True
|
||||
assert seen == [True]
|
||||
packet = captured[0]
|
||||
assert packet["protocol"] == "meshcore"
|
||||
assert packet["from_id"] == "!11223344"
|
||||
assert packet["decoded"]["portnum"] == "TELEMETRY_APP"
|
||||
assert packet["decoded"]["telemetry"]["deviceMetrics"] == {"voltage": 4.05}
|
||||
assert packet["decoded"]["telemetry"]["time"] == 1_700_000_000
|
||||
assert isinstance(packet["id"], int) and 0 <= packet["id"] < (1 << 53)
|
||||
|
||||
# Same node/kind/second → identical id (web-side PRIMARY KEY collapse).
|
||||
mc_tel._queue_meshcore_telemetry(
|
||||
stub, "!11223344", {"deviceMetrics": {"voltage": 4.06}}, "battery"
|
||||
)
|
||||
assert captured[1]["id"] == packet["id"]
|
||||
# A different kind in the same second must not collide.
|
||||
mc_tel._queue_meshcore_telemetry(
|
||||
stub, "!11223344", {"deviceMetrics": {"voltage": 4.06}}, "status"
|
||||
)
|
||||
assert captured[2]["id"] != packet["id"]
|
||||
|
||||
|
||||
def test_queue_meshcore_telemetry_skips_incomplete(monkeypatch):
|
||||
"""Missing node id or empty section queues nothing."""
|
||||
mc_tel, _iface, stub, captured = _telemetry_env(monkeypatch)
|
||||
assert (
|
||||
mc_tel._queue_meshcore_telemetry(stub, None, {"deviceMetrics": {}}, "x")
|
||||
is False
|
||||
)
|
||||
assert mc_tel._queue_meshcore_telemetry(stub, "!11223344", None, "x") is False
|
||||
assert captured == []
|
||||
|
||||
|
||||
def test_telemetry_event_callbacks_ingest(monkeypatch):
|
||||
"""The three event callbacks resolve the node and queue telemetry."""
|
||||
mc_tel, iface, stub, captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
iface.host_node_id = "!deadbeef"
|
||||
handlers_map = mc_tel._make_telemetry_handlers(iface, stub)
|
||||
|
||||
asyncio.run(
|
||||
handlers_map["TELEMETRY_RESPONSE"](
|
||||
_FakeEvt(
|
||||
{
|
||||
"pubkey_pre": _TEST_CONTACT_KEY[:12],
|
||||
"lpp": [{"type": "temperature", "value": 21.5}],
|
||||
}
|
||||
)
|
||||
)
|
||||
)
|
||||
asyncio.run(
|
||||
handlers_map["STATUS_RESPONSE"](
|
||||
_FakeEvt({"pubkey_pre": _TEST_CONTACT_KEY[:12], "bat": 4056})
|
||||
)
|
||||
)
|
||||
asyncio.run(handlers_map["BATTERY"](_FakeEvt({"level": 3900})))
|
||||
asyncio.run(handlers_map["BATTERY"](_FakeEvt({}))) # no gauge → skipped
|
||||
|
||||
assert len(captured) == 3
|
||||
assert captured[0]["decoded"]["telemetry"]["environmentMetrics"] == {
|
||||
"temperature": 21.5
|
||||
}
|
||||
assert captured[1]["decoded"]["telemetry"]["deviceMetrics"] == {"voltage": 4.056}
|
||||
assert captured[2]["from_id"] == "!deadbeef"
|
||||
assert captured[2]["decoded"]["telemetry"]["deviceMetrics"] == {"voltage": 3.9}
|
||||
|
||||
|
||||
def test_next_poll_contact_round_robin_with_cooldown(monkeypatch):
|
||||
"""Contact polling walks the roster in stable order, then idles until a
|
||||
contact's 24 h cooldown expires instead of wrapping immediately."""
|
||||
second_key = "bbccddeeff00" + "11" * 26
|
||||
mc_tel, iface, _stub, _captured = _telemetry_env(
|
||||
monkeypatch,
|
||||
contacts=[
|
||||
{"public_key": _TEST_CONTACT_KEY, "adv_name": "A"},
|
||||
{"public_key": second_key, "adv_name": "B"},
|
||||
],
|
||||
)
|
||||
state: dict = {}
|
||||
first = mc_tel._next_poll_contact(iface, state)
|
||||
second = mc_tel._next_poll_contact(iface, state)
|
||||
assert [first["public_key"], second["public_key"]] == [
|
||||
_TEST_CONTACT_KEY,
|
||||
second_key,
|
||||
]
|
||||
# Both contacts were just stamped — the roster is fully fresh, so the
|
||||
# tick has nothing to send.
|
||||
assert mc_tel._next_poll_contact(iface, state) is None
|
||||
|
||||
# Aging one contact past the cooldown makes it eligible again.
|
||||
state["last_polled"][second_key] -= mc_tel._TELEMETRY_NODE_COOLDOWN_SECONDS + 1
|
||||
assert mc_tel._next_poll_contact(iface, state)["public_key"] == second_key
|
||||
|
||||
empty = _MeshcoreInterface(target=None)
|
||||
assert mc_tel._next_poll_contact(empty, {}) is None
|
||||
|
||||
|
||||
def test_next_poll_contact_prunes_departed_roster_entries(monkeypatch):
|
||||
"""Cooldown stamps for contacts no longer in the roster are dropped."""
|
||||
mc_tel, iface, _stub, _captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "A"}]
|
||||
)
|
||||
state = {"last_polled": {"departed" + "00" * 26: 123.0}}
|
||||
picked = mc_tel._next_poll_contact(iface, state)
|
||||
assert picked["public_key"] == _TEST_CONTACT_KEY
|
||||
assert set(state["last_polled"]) == {_TEST_CONTACT_KEY}
|
||||
|
||||
|
||||
def test_poll_contact_telemetry_honours_cooldown_across_ticks(monkeypatch):
|
||||
"""With shared state, a second tick inside the cooldown sends nothing."""
|
||||
import types
|
||||
|
||||
mc_tel, iface, stub, captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
requests = {"count": 0}
|
||||
|
||||
class _Commands:
|
||||
async def req_telemetry_sync(self, contact):
|
||||
requests["count"] += 1
|
||||
return [{"type": "temperature", "value": 21.5}]
|
||||
|
||||
async def req_status_sync(self, contact):
|
||||
return None
|
||||
|
||||
mc = types.SimpleNamespace(commands=_Commands())
|
||||
state: dict = {}
|
||||
asyncio.run(mc_tel._poll_contact_telemetry(mc, iface, stub, state))
|
||||
asyncio.run(mc_tel._poll_contact_telemetry(mc, iface, stub, state))
|
||||
assert requests["count"] == 1
|
||||
assert len(captured) == 1
|
||||
|
||||
|
||||
def test_poll_self_telemetry_ingests_and_swallows_errors(monkeypatch):
|
||||
"""Self polling queues battery + sensor packets; command errors are logged."""
|
||||
import types
|
||||
|
||||
mc_tel, iface, stub, captured = _telemetry_env(monkeypatch)
|
||||
iface.host_node_id = "!deadbeef"
|
||||
|
||||
class _Commands:
|
||||
async def get_bat(self):
|
||||
return types.SimpleNamespace(payload={"level": 4100})
|
||||
|
||||
async def get_self_telemetry(self):
|
||||
return types.SimpleNamespace(
|
||||
payload={"lpp": [{"type": "temperature", "value": 22.0}]}
|
||||
)
|
||||
|
||||
mc = types.SimpleNamespace(commands=_Commands())
|
||||
asyncio.run(mc_tel._poll_self_telemetry(mc, iface, stub))
|
||||
assert len(captured) == 2
|
||||
|
||||
class _BrokenCommands:
|
||||
async def get_bat(self):
|
||||
raise RuntimeError("no battery command")
|
||||
|
||||
async def get_self_telemetry(self):
|
||||
raise RuntimeError("no telemetry command")
|
||||
|
||||
captured.clear()
|
||||
asyncio.run(
|
||||
mc_tel._poll_self_telemetry(
|
||||
types.SimpleNamespace(commands=_BrokenCommands()), iface, stub
|
||||
)
|
||||
)
|
||||
assert captured == []
|
||||
|
||||
|
||||
def test_poll_contact_telemetry_paths(monkeypatch):
|
||||
"""Contact polling ingests LPP, falls back to status, and survives errors."""
|
||||
import types
|
||||
|
||||
mc_tel, iface, stub, captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
|
||||
def _mc(
|
||||
telemetry_result=None,
|
||||
telemetry_error=None,
|
||||
status_result=None,
|
||||
status_error=None,
|
||||
):
|
||||
class _Commands:
|
||||
async def req_telemetry_sync(self, contact):
|
||||
if telemetry_error:
|
||||
raise telemetry_error
|
||||
return telemetry_result
|
||||
|
||||
async def req_status_sync(self, contact):
|
||||
if status_error:
|
||||
raise status_error
|
||||
return status_result
|
||||
|
||||
return types.SimpleNamespace(commands=_Commands())
|
||||
|
||||
# LPP success → one packet, no status fallback needed.
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
_mc(telemetry_result=[{"type": "temperature", "value": 21.5}]),
|
||||
iface,
|
||||
stub,
|
||||
{},
|
||||
)
|
||||
)
|
||||
assert len(captured) == 1
|
||||
|
||||
# Empty LPP → status fallback ingests battery.
|
||||
captured.clear()
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
_mc(telemetry_result=None, status_result={"bat": 4056}), iface, stub, {}
|
||||
)
|
||||
)
|
||||
assert len(captured) == 1
|
||||
assert captured[0]["decoded"]["telemetry"]["deviceMetrics"]["voltage"] == 4.056
|
||||
|
||||
# Telemetry request error → logged, no status attempt, nothing queued.
|
||||
captured.clear()
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
_mc(telemetry_error=RuntimeError("timeout")), iface, stub, {}
|
||||
)
|
||||
)
|
||||
assert captured == []
|
||||
|
||||
# Status fallback error → logged, nothing queued.
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
_mc(status_error=RuntimeError("timeout")), iface, stub, {}
|
||||
)
|
||||
)
|
||||
assert captured == []
|
||||
|
||||
# Empty roster → no-op.
|
||||
empty = _MeshcoreInterface(target=None)
|
||||
asyncio.run(mc_tel._poll_contact_telemetry(_mc(), empty, stub, {}))
|
||||
assert captured == []
|
||||
|
||||
# A roster entry whose public key is too short to derive a node id is
|
||||
# skipped before any on-air request.
|
||||
unresolvable = _MeshcoreInterface(target=None)
|
||||
unresolvable._update_contact({"public_key": "abcd", "adv_name": "Ghost"})
|
||||
asyncio.run(mc_tel._poll_contact_telemetry(_mc(), unresolvable, stub, {}))
|
||||
assert captured == []
|
||||
|
||||
|
||||
def test_telemetry_poll_loop_disabled_and_ticking(monkeypatch):
|
||||
"""The poll loop exits when disabled and fires both poll kinds when enabled."""
|
||||
import types
|
||||
|
||||
mc_tel, iface, stub, captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
iface.host_node_id = "!deadbeef"
|
||||
import data.mesh_ingestor as _mesh_pkg
|
||||
|
||||
monkeypatch.setattr(_mesh_pkg, "handlers", stub)
|
||||
|
||||
# Both cadences disabled → the loop returns immediately.
|
||||
monkeypatch.setattr(mc_tel.config, "MESHCORE_SELF_TELEMETRY_SECONDS", 0)
|
||||
monkeypatch.setattr(mc_tel.config, "MESHCORE_TELEMETRY_POLL_SECONDS", 0)
|
||||
asyncio.run(mc_tel._telemetry_poll_loop(types.SimpleNamespace(), iface))
|
||||
|
||||
# Enabled: run the loop as a task, let the immediate self tick and the
|
||||
# (shortened) contact tick fire, then cancel.
|
||||
calls = {"self": 0, "contact": 0}
|
||||
|
||||
class _Commands:
|
||||
async def get_bat(self):
|
||||
calls["self"] += 1
|
||||
return types.SimpleNamespace(payload={"level": 4100})
|
||||
|
||||
async def get_self_telemetry(self):
|
||||
return types.SimpleNamespace(payload={"lpp": []})
|
||||
|
||||
async def req_telemetry_sync(self, contact):
|
||||
calls["contact"] += 1
|
||||
return [{"type": "temperature", "value": 21.5}]
|
||||
|
||||
async def req_status_sync(self, contact):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(mc_tel.config, "MESHCORE_SELF_TELEMETRY_SECONDS", 3600)
|
||||
monkeypatch.setattr(mc_tel.config, "MESHCORE_TELEMETRY_POLL_SECONDS", 1)
|
||||
|
||||
async def _drive():
|
||||
task = asyncio.create_task(
|
||||
mc_tel._telemetry_poll_loop(
|
||||
types.SimpleNamespace(commands=_Commands()), iface
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(1.3)
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
asyncio.run(_drive())
|
||||
assert calls["self"] == 1 # immediate first self tick
|
||||
assert calls["contact"] >= 1 # first on-air poll after one interval
|
||||
|
||||
|
||||
def test_telemetry_poll_loop_rx_only_disables_on_air_polls(monkeypatch):
|
||||
"""RX_ONLY forbids ingestor TX: contact polls stop, local self reads stay."""
|
||||
import types
|
||||
|
||||
mc_tel, iface, stub, _captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
iface.host_node_id = "!deadbeef"
|
||||
import data.mesh_ingestor as _mesh_pkg
|
||||
|
||||
monkeypatch.setattr(_mesh_pkg, "handlers", stub)
|
||||
monkeypatch.setattr(mc_tel.config, "RX_ONLY", True)
|
||||
|
||||
calls = {"self": 0, "contact": 0}
|
||||
|
||||
class _Commands:
|
||||
async def get_bat(self):
|
||||
calls["self"] += 1
|
||||
return types.SimpleNamespace(payload={"level": 4100})
|
||||
|
||||
async def get_self_telemetry(self):
|
||||
return types.SimpleNamespace(payload={"lpp": []})
|
||||
|
||||
async def req_telemetry_sync(self, contact):
|
||||
calls["contact"] += 1
|
||||
return None
|
||||
|
||||
async def req_status_sync(self, contact):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(mc_tel.config, "MESHCORE_SELF_TELEMETRY_SECONDS", 3600)
|
||||
monkeypatch.setattr(mc_tel.config, "MESHCORE_TELEMETRY_POLL_SECONDS", 1)
|
||||
|
||||
async def _drive():
|
||||
task = asyncio.create_task(
|
||||
mc_tel._telemetry_poll_loop(
|
||||
types.SimpleNamespace(commands=_Commands()), iface
|
||||
)
|
||||
)
|
||||
await asyncio.sleep(1.3)
|
||||
task.cancel()
|
||||
try:
|
||||
await task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
asyncio.run(_drive())
|
||||
assert calls["self"] == 1 # companion-link reads are not transmissions
|
||||
assert calls["contact"] == 0 # no on-air request under RX_ONLY
|
||||
|
||||
# RX_ONLY with self polling also disabled → the loop exits immediately.
|
||||
monkeypatch.setattr(mc_tel.config, "MESHCORE_SELF_TELEMETRY_SECONDS", 0)
|
||||
asyncio.run(mc_tel._telemetry_poll_loop(types.SimpleNamespace(), iface))
|
||||
|
||||
|
||||
def test_on_channel_msg_queues_packet(monkeypatch):
|
||||
"""on_channel_msg must call store_packet_dict with the correct packet fields."""
|
||||
import asyncio
|
||||
@@ -3331,6 +3866,10 @@ def _make_fake_meshcore_mod(
|
||||
"CONTACT_DELETED",
|
||||
"RX_LOG_DATA",
|
||||
"DISCONNECTED",
|
||||
# Telemetry surfaces subscribed since TI-A3.
|
||||
"TELEMETRY_RESPONSE",
|
||||
"STATUS_RESPONSE",
|
||||
"BATTERY",
|
||||
"CONNECTED",
|
||||
"ACK",
|
||||
"OK",
|
||||
|
||||
@@ -31,6 +31,7 @@ require_relative "data_processing/node_writes"
|
||||
require_relative "data_processing/positions"
|
||||
require_relative "data_processing/neighbors"
|
||||
require_relative "data_processing/traces"
|
||||
require_relative "data_processing/telemetry_metrics"
|
||||
require_relative "data_processing/telemetry"
|
||||
require_relative "data_processing/decrypted_payloads"
|
||||
require_relative "data_processing/meshcore_chat"
|
||||
|
||||
@@ -18,7 +18,16 @@ module PotatoMesh
|
||||
module App
|
||||
module DataProcessing
|
||||
# Allowed values for the +telemetry_type+ discriminator column.
|
||||
VALID_TELEMETRY_TYPES = %w[device environment power air_quality].freeze
|
||||
VALID_TELEMETRY_TYPES = %w[
|
||||
device
|
||||
environment
|
||||
power
|
||||
air_quality
|
||||
local_stats
|
||||
health
|
||||
host
|
||||
traffic
|
||||
].freeze
|
||||
|
||||
# Half-window (seconds) for the meshcore content-level message dedup
|
||||
# in +insert_message+ and the matching one-shot backfill. Two
|
||||
|
||||
@@ -267,8 +267,11 @@ module PotatoMesh
|
||||
#
|
||||
# @param key_map [Hash{Symbol=>Array<String>}] ordered mapping of source names to candidate keys.
|
||||
# @param sources [Hash{Symbol=>Hash}] data structures to search for metric values.
|
||||
# @param type [Symbol] coercion strategy, ``:float`` or ``:integer``.
|
||||
# @return [Numeric, nil] coerced metric value or nil when no candidates exist.
|
||||
# @param type [Symbol] coercion strategy: ``:float``, ``:integer``,
|
||||
# ``:string`` (trimmed text, e.g. ``user_string``), or ``:float_array``
|
||||
# (list of floats serialised to a JSON string, e.g.
|
||||
# ``one_wire_temperature``).
|
||||
# @return [Numeric, String, nil] coerced metric value or nil when no candidates exist.
|
||||
def resolve_numeric_metric(key_map, sources, type)
|
||||
key_map.each do |source, keys|
|
||||
next if keys.nil? || keys.empty?
|
||||
@@ -294,6 +297,10 @@ module PotatoMesh
|
||||
coerce_float(value)
|
||||
when :integer
|
||||
coerce_integer(value)
|
||||
when :string
|
||||
string_or_nil(value)
|
||||
when :float_array
|
||||
coerce_float_array_json(value)
|
||||
else
|
||||
value
|
||||
end
|
||||
@@ -307,6 +314,22 @@ module PotatoMesh
|
||||
|
||||
private :resolve_numeric_metric
|
||||
|
||||
# Coerce a repeated float metric into its JSON storage form.
|
||||
#
|
||||
# @param value [Object] candidate value; only arrays are accepted.
|
||||
# @return [String, nil] JSON array of finite floats, or nil when nothing
|
||||
# usable remains (empty arrays are treated as absent, never stored).
|
||||
def coerce_float_array_json(value)
|
||||
return nil unless value.is_a?(Array)
|
||||
|
||||
floats = value.map { |item| coerce_float(item) }.compact
|
||||
return nil if floats.empty?
|
||||
|
||||
JSON.generate(floats)
|
||||
end
|
||||
|
||||
private :coerce_float_array_json
|
||||
|
||||
# Persist a telemetry packet and refresh the related node row.
|
||||
#
|
||||
# @param db [SQLite3::Database] open database handle.
|
||||
@@ -371,6 +394,14 @@ module PotatoMesh
|
||||
power_metrics ||= normalize_json_object(telemetry_section["powerMetrics"]) if telemetry_section&.key?("powerMetrics")
|
||||
air_quality_metrics = normalize_json_object(payload["air_quality_metrics"] || payload["airQualityMetrics"])
|
||||
air_quality_metrics ||= normalize_json_object(telemetry_section["airQualityMetrics"]) if telemetry_section&.key?("airQualityMetrics")
|
||||
local_stats = normalize_json_object(payload["local_stats"] || payload["localStats"])
|
||||
local_stats ||= normalize_json_object(telemetry_section["localStats"]) if telemetry_section&.key?("localStats")
|
||||
health_metrics = normalize_json_object(payload["health_metrics"] || payload["healthMetrics"])
|
||||
health_metrics ||= normalize_json_object(telemetry_section["healthMetrics"]) if telemetry_section&.key?("healthMetrics")
|
||||
host_metrics = normalize_json_object(payload["host_metrics"] || payload["hostMetrics"])
|
||||
host_metrics ||= normalize_json_object(telemetry_section["hostMetrics"]) if telemetry_section&.key?("hostMetrics")
|
||||
traffic_stats = normalize_json_object(payload["traffic_management_stats"] || payload["trafficManagementStats"])
|
||||
traffic_stats ||= normalize_json_object(telemetry_section["trafficManagementStats"]) if telemetry_section&.key?("trafficManagementStats")
|
||||
|
||||
telemetry_type = string_or_nil(payload["telemetry_type"])
|
||||
telemetry_type = nil unless VALID_TELEMETRY_TYPES.include?(telemetry_type)
|
||||
@@ -382,6 +413,14 @@ module PotatoMesh
|
||||
"power"
|
||||
elsif air_quality_metrics&.any?
|
||||
"air_quality"
|
||||
elsif local_stats&.any?
|
||||
"local_stats"
|
||||
elsif health_metrics&.any?
|
||||
"health"
|
||||
elsif host_metrics&.any?
|
||||
"host"
|
||||
elsif traffic_stats&.any?
|
||||
"traffic"
|
||||
end
|
||||
|
||||
sources = {
|
||||
@@ -389,10 +428,16 @@ module PotatoMesh
|
||||
telemetry: telemetry_section,
|
||||
device: device_metrics,
|
||||
environment: environment_metrics,
|
||||
power: power_metrics,
|
||||
air_quality: air_quality_metrics,
|
||||
local_stats: local_stats,
|
||||
health: health_metrics,
|
||||
host: host_metrics,
|
||||
traffic: traffic_stats,
|
||||
}
|
||||
|
||||
metric_values = {}
|
||||
TELEMETRY_METRIC_DEFINITIONS.each do |column, type, key_map|
|
||||
(TELEMETRY_METRIC_DEFINITIONS + EXTENDED_TELEMETRY_METRIC_DEFINITIONS).each do |column, type, key_map|
|
||||
value = resolve_numeric_metric(key_map, sources, type)
|
||||
metric_values[column] = value unless value.nil?
|
||||
end
|
||||
@@ -470,13 +515,16 @@ module PotatoMesh
|
||||
protocol,
|
||||
telemetry_type,
|
||||
]
|
||||
# Extended metric columns bind after telemetry_type, in the canonical
|
||||
# EXTENDED_TELEMETRY_COLUMN_NAMES order the SQL fragments share.
|
||||
row += EXTENDED_TELEMETRY_COLUMN_NAMES.map { |column| metric_values[column] }
|
||||
|
||||
placeholders = Array.new(row.length, "?").join(",")
|
||||
|
||||
with_busy_retry do
|
||||
db.execute <<~SQL, row
|
||||
INSERT INTO telemetry(id,node_id,node_num,from_id,to_id,rx_time,rx_iso,telemetry_time,channel,portnum,hop_limit,snr,rssi,bitfield,payload_b64,
|
||||
battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds,temperature,relative_humidity,barometric_pressure,gas_resistance,current,iaq,distance,lux,white_lux,ir_lux,uv_lux,wind_direction,wind_speed,weight,wind_gust,wind_lull,radiation,rainfall_1h,rainfall_24h,soil_moisture,soil_temperature,ingestor,protocol,telemetry_type)
|
||||
battery_level,voltage,channel_utilization,air_util_tx,uptime_seconds,temperature,relative_humidity,barometric_pressure,gas_resistance,current,iaq,distance,lux,white_lux,ir_lux,uv_lux,wind_direction,wind_speed,weight,wind_gust,wind_lull,radiation,rainfall_1h,rainfall_24h,soil_moisture,soil_temperature,ingestor,protocol,telemetry_type#{EXTENDED_TELEMETRY_INSERT_COLUMNS_SQL})
|
||||
VALUES (#{placeholders})
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
node_id=COALESCE(excluded.node_id,telemetry.node_id),
|
||||
@@ -521,7 +569,7 @@ module PotatoMesh
|
||||
soil_temperature=COALESCE(excluded.soil_temperature,telemetry.soil_temperature),
|
||||
ingestor=COALESCE(NULLIF(telemetry.ingestor,''), excluded.ingestor),
|
||||
protocol=COALESCE(NULLIF(telemetry.protocol,'meshtastic'), excluded.protocol),
|
||||
telemetry_type=COALESCE(excluded.telemetry_type,telemetry.telemetry_type)
|
||||
telemetry_type=COALESCE(excluded.telemetry_type,telemetry.telemetry_type)#{EXTENDED_TELEMETRY_UPSERT_SQL}
|
||||
SQL
|
||||
end
|
||||
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
# frozen_string_literal: true
|
||||
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
module PotatoMesh
|
||||
module App
|
||||
module DataProcessing
|
||||
# Build metric definitions for one telemetry family (TI-A2).
|
||||
#
|
||||
# Each triple expands to a +[column, type, key_map]+ definition. The
|
||||
# flat payload is probed with the snake_case column name **only**; the
|
||||
# protobuf-JSON camelCase twin is accepted solely inside the family
|
||||
# sub-object (for third-party ingestors that post nested shapes).
|
||||
# Camel twins are not unique across families — +healthMetrics.temperature+
|
||||
# shares its camel name with the ambient +temperature+ metric — so
|
||||
# probing them against the flat payload would let one family's reading
|
||||
# corrupt another's column (the ambient-into-+health_temperature+ leak).
|
||||
#
|
||||
# @param source [Symbol] source-layer key of the family sub-object
|
||||
# (e.g. +:power+) as registered in +insert_telemetry+'s sources hash.
|
||||
# @param triples [Array<Array>] list of +[column, type, camel_name]+.
|
||||
# @return [Array<Array>] metric definitions for the family.
|
||||
def self.build_family_metric_definitions(source, triples)
|
||||
triples.map do |column, type, camel|
|
||||
[column, type, { payload: [column], source => [column, camel].uniq }]
|
||||
end
|
||||
end
|
||||
|
||||
# PowerMetrics channel triples: +ch1+–+ch8+, voltage + current each.
|
||||
POWER_CHANNEL_METRIC_TRIPLES = (1..8).flat_map do |ch|
|
||||
[
|
||||
["ch#{ch}_voltage", :float, "ch#{ch}Voltage"],
|
||||
["ch#{ch}_current", :float, "ch#{ch}Current"],
|
||||
]
|
||||
end.freeze
|
||||
|
||||
# Ordered metric definitions for the telemetry families beyond the
|
||||
# original device/environment pair: PowerMetrics, AirQualityMetrics,
|
||||
# HealthMetrics, LocalStats, HostMetrics, TrafficManagementStats, and
|
||||
# the repeated one-wire probe list. Consumed by +insert_telemetry+
|
||||
# alongside +TELEMETRY_METRIC_DEFINITIONS+; the column order here is the
|
||||
# canonical order used by the INSERT/upsert SQL and the schema
|
||||
# auto-migration.
|
||||
EXTENDED_TELEMETRY_METRIC_DEFINITIONS = (build_family_metric_definitions(:power, POWER_CHANNEL_METRIC_TRIPLES) +
|
||||
build_family_metric_definitions(:air_quality, [
|
||||
["pm10_standard", :integer, "pm10Standard"],
|
||||
["pm25_standard", :integer, "pm25Standard"],
|
||||
["pm100_standard", :integer, "pm100Standard"],
|
||||
["pm40_standard", :integer, "pm40Standard"],
|
||||
["pm10_environmental", :integer, "pm10Environmental"],
|
||||
["pm25_environmental", :integer, "pm25Environmental"],
|
||||
["pm100_environmental", :integer, "pm100Environmental"],
|
||||
["particles_03um", :integer, "particles03um"],
|
||||
["particles_05um", :integer, "particles05um"],
|
||||
["particles_10um", :integer, "particles10um"],
|
||||
["particles_25um", :integer, "particles25um"],
|
||||
["particles_40um", :integer, "particles40um"],
|
||||
["particles_50um", :integer, "particles50um"],
|
||||
["particles_100um", :integer, "particles100um"],
|
||||
["particles_tps", :float, "particlesTps"],
|
||||
["co2", :integer, "co2"],
|
||||
["co2_temperature", :float, "co2Temperature"],
|
||||
["co2_humidity", :float, "co2Humidity"],
|
||||
["form_formaldehyde", :float, "formFormaldehyde"],
|
||||
["form_humidity", :float, "formHumidity"],
|
||||
["form_temperature", :float, "formTemperature"],
|
||||
["pm_temperature", :float, "pmTemperature"],
|
||||
["pm_humidity", :float, "pmHumidity"],
|
||||
["pm_voc_idx", :float, "pmVocIdx"],
|
||||
["pm_nox_idx", :float, "pmNoxIdx"],
|
||||
]) +
|
||||
build_family_metric_definitions(:health, [
|
||||
["heart_bpm", :integer, "heartBpm"],
|
||||
["spo2", :integer, "spO2"],
|
||||
# Body temperature stays apart from the ambient temperature column.
|
||||
["health_temperature", :float, "temperature"],
|
||||
]) +
|
||||
build_family_metric_definitions(:local_stats, [
|
||||
["num_packets_tx", :integer, "numPacketsTx"],
|
||||
["num_packets_rx", :integer, "numPacketsRx"],
|
||||
["num_packets_rx_bad", :integer, "numPacketsRxBad"],
|
||||
["num_online_nodes", :integer, "numOnlineNodes"],
|
||||
["num_total_nodes", :integer, "numTotalNodes"],
|
||||
["num_rx_dupe", :integer, "numRxDupe"],
|
||||
["num_tx_relay", :integer, "numTxRelay"],
|
||||
["num_tx_relay_canceled", :integer, "numTxRelayCanceled"],
|
||||
["heap_total_bytes", :integer, "heapTotalBytes"],
|
||||
["heap_free_bytes", :integer, "heapFreeBytes"],
|
||||
["num_tx_dropped", :integer, "numTxDropped"],
|
||||
["noise_floor", :integer, "noiseFloor"],
|
||||
]) +
|
||||
build_family_metric_definitions(:host, [
|
||||
["freemem_bytes", :integer, "freememBytes"],
|
||||
["diskfree1_bytes", :integer, "diskfree1Bytes"],
|
||||
["diskfree2_bytes", :integer, "diskfree2Bytes"],
|
||||
["diskfree3_bytes", :integer, "diskfree3Bytes"],
|
||||
["load1", :integer, "load1"],
|
||||
["load5", :integer, "load5"],
|
||||
["load15", :integer, "load15"],
|
||||
["user_string", :string, "userString"],
|
||||
]) +
|
||||
build_family_metric_definitions(:traffic, [
|
||||
["packets_inspected", :integer, "packetsInspected"],
|
||||
["position_dedup_drops", :integer, "positionDedupDrops"],
|
||||
["nodeinfo_cache_hits", :integer, "nodeinfoCacheHits"],
|
||||
["rate_limit_drops", :integer, "rateLimitDrops"],
|
||||
["unknown_packet_drops", :integer, "unknownPacketDrops"],
|
||||
["hop_exhausted_packets", :integer, "hopExhaustedPackets"],
|
||||
["router_hops_preserved", :integer, "routerHopsPreserved"],
|
||||
]) +
|
||||
build_family_metric_definitions(:environment, [
|
||||
["one_wire_temperature", :float_array, "oneWireTemperature"],
|
||||
])).freeze
|
||||
|
||||
# Ordered extended column names (canonical SQL order).
|
||||
EXTENDED_TELEMETRY_COLUMN_NAMES =
|
||||
EXTENDED_TELEMETRY_METRIC_DEFINITIONS.map(&:first).freeze
|
||||
|
||||
# SQLite column type per coercion strategy.
|
||||
EXTENDED_TELEMETRY_SQL_TYPES = {
|
||||
float: "REAL",
|
||||
integer: "INTEGER",
|
||||
string: "TEXT",
|
||||
float_array: "TEXT",
|
||||
}.freeze
|
||||
|
||||
# +[name, sqlite_type]+ pairs consumed by the boot-time schema
|
||||
# auto-migration (+ensure_schema_upgrades+) so existing databases gain
|
||||
# the extended columns without operator action.
|
||||
EXTENDED_TELEMETRY_COLUMN_TYPES =
|
||||
EXTENDED_TELEMETRY_METRIC_DEFINITIONS.map do |column, type, _|
|
||||
[column, EXTENDED_TELEMETRY_SQL_TYPES.fetch(type)]
|
||||
end.freeze
|
||||
|
||||
# SQL fragment appended to the INSERT column list (leading comma form).
|
||||
EXTENDED_TELEMETRY_INSERT_COLUMNS_SQL =
|
||||
EXTENDED_TELEMETRY_COLUMN_NAMES.map { |column| ",#{column}" }.join.freeze
|
||||
|
||||
# SQL fragment appended to the upsert SET list: keep the stored value
|
||||
# whenever the new row carries NULL, matching the base metric columns.
|
||||
EXTENDED_TELEMETRY_UPSERT_SQL =
|
||||
EXTENDED_TELEMETRY_COLUMN_NAMES.map do |column|
|
||||
",\n #{column}=COALESCE(excluded.#{column},telemetry.#{column})"
|
||||
end.join.freeze
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -395,7 +395,11 @@ module PotatoMesh
|
||||
end
|
||||
|
||||
telemetry_columns = db.execute("PRAGMA table_info(telemetry)").map { |row| row[1] }
|
||||
TELEMETRY_COLUMN_DEFINITIONS.each do |name, type|
|
||||
# The environment expansion plus the extended metric families (TI-A2:
|
||||
# power / air-quality / health / local / host / traffic stats and the
|
||||
# one-wire probe list) share one idempotent backfill loop; the extended
|
||||
# pairs derive from the same definitions insert_telemetry writes with.
|
||||
(TELEMETRY_COLUMN_DEFINITIONS + DataProcessing::EXTENDED_TELEMETRY_COLUMN_TYPES).each do |name, type|
|
||||
next if telemetry_columns.include?(name)
|
||||
|
||||
db.execute("ALTER TABLE telemetry ADD COLUMN #{name} #{type}")
|
||||
|
||||
@@ -316,6 +316,230 @@ RSpec.describe PotatoMesh::App::DataProcessing do
|
||||
end
|
||||
end
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# insert_telemetry (extended metric families — TI-A2)
|
||||
# ---------------------------------------------------------------------------
|
||||
describe "#insert_telemetry — extended metric families" do
|
||||
include_context "with isolated db"
|
||||
|
||||
# Insert +payload+ and return the stored telemetry row as a Hash.
|
||||
#
|
||||
# @param payload [Hash] inbound telemetry payload.
|
||||
# @return [Hash] stored row for the payload's id.
|
||||
def stored_row(payload)
|
||||
db = open_db
|
||||
dp.insert_telemetry(db, payload)
|
||||
row = db.execute("SELECT * FROM telemetry WHERE id = ?", [payload["id"]]).first
|
||||
db.close
|
||||
row
|
||||
end
|
||||
|
||||
it "stores power metric values from power_metrics" do
|
||||
row = stored_row(
|
||||
"id" => 91_001,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"power_metrics" => { "ch1Voltage" => 3.94, "ch2Current" => 121.5 },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("power")
|
||||
expect(row["ch1_voltage"]).to eq(3.94)
|
||||
expect(row["ch2_current"]).to eq(121.5)
|
||||
end
|
||||
|
||||
it "stores air quality metric values from air_quality_metrics" do
|
||||
row = stored_row(
|
||||
"id" => 91_002,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"air_quality_metrics" => { "pm25Standard" => 8, "co2" => 700 },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("air_quality")
|
||||
expect(row["pm25_standard"]).to eq(8)
|
||||
expect(row["co2"]).to eq(700)
|
||||
end
|
||||
|
||||
it "stores health metric values under dedicated columns" do
|
||||
row = stored_row(
|
||||
"id" => 91_003,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"health_metrics" => { "heartBpm" => 72, "spO2" => 97, "temperature" => 36.6 },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("health")
|
||||
expect(row["heart_bpm"]).to eq(72)
|
||||
expect(row["spo2"]).to eq(97)
|
||||
expect(row["health_temperature"]).to eq(36.6)
|
||||
expect(row["temperature"]).to be_nil
|
||||
end
|
||||
|
||||
it "accepts the diagnostics telemetry_type values and counters" do
|
||||
row = stored_row(
|
||||
"id" => 91_004,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"telemetry_type" => "local_stats",
|
||||
"local_stats" => { "numPacketsRx" => 42, "noiseFloor" => -95 },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("local_stats")
|
||||
expect(row["num_packets_rx"]).to eq(42)
|
||||
expect(row["noise_floor"]).to eq(-95)
|
||||
end
|
||||
|
||||
it "stores host metrics including the user string" do
|
||||
row = stored_row(
|
||||
"id" => 91_005,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"host_metrics" => { "freememBytes" => 1_048_576, "load1" => 35, "userString" => "potato" },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("host")
|
||||
expect(row["freemem_bytes"]).to eq(1_048_576)
|
||||
expect(row["load1"]).to eq(35)
|
||||
expect(row["user_string"]).to eq("potato")
|
||||
end
|
||||
|
||||
it "stores one_wire_temperature as a JSON array" do
|
||||
row = stored_row(
|
||||
"id" => 91_006,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"environment_metrics" => { "temperature" => 21.5, "oneWireTemperature" => [20.0, 21.25] },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("environment")
|
||||
expect(row["temperature"]).to eq(21.5)
|
||||
expect(JSON.parse(row["one_wire_temperature"])).to eq([20.0, 21.25])
|
||||
end
|
||||
|
||||
it "never leaks a flat ambient temperature into health_temperature" do
|
||||
# The exact shape the Python ingestor posts for every EnvironmentMetrics
|
||||
# packet (flat snake_case keys): ambient temperature must stay out of
|
||||
# the body-temperature column even though their camelCase twins collide.
|
||||
row = stored_row(
|
||||
"id" => 91_009,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"temperature" => 21.5,
|
||||
"relative_humidity" => 40.2,
|
||||
"telemetry_type" => "environment",
|
||||
)
|
||||
expect(row["temperature"]).to eq(21.5)
|
||||
expect(row["health_temperature"]).to be_nil
|
||||
end
|
||||
|
||||
it "still reads body temperature from the nested health_metrics object" do
|
||||
row = stored_row(
|
||||
"id" => 91_010,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"health_metrics" => { "temperature" => 36.6 },
|
||||
)
|
||||
expect(row["health_temperature"]).to eq(36.6)
|
||||
expect(row["temperature"]).to be_nil
|
||||
end
|
||||
|
||||
it "infers the local_stats type from the sub-object when no type is sent" do
|
||||
row = stored_row(
|
||||
"id" => 91_008,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"local_stats" => { "numPacketsTx" => 10 },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("local_stats")
|
||||
expect(row["num_packets_tx"]).to eq(10)
|
||||
end
|
||||
|
||||
it "stores the traffic management counters" do
|
||||
row = stored_row(
|
||||
"id" => 91_007,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"traffic_management_stats" => { "packetsInspected" => 100, "rateLimitDrops" => 3 },
|
||||
)
|
||||
expect(row["telemetry_type"]).to eq("traffic")
|
||||
expect(row["packets_inspected"]).to eq(100)
|
||||
expect(row["rate_limit_drops"]).to eq(3)
|
||||
end
|
||||
|
||||
it "round-trips every extended metric column from flat snake_case keys" do
|
||||
# Data-driven over the full definition list: one synthetic value per
|
||||
# column, typed to its coercion strategy, exactly as the Python ingestor
|
||||
# posts them (flat snake_case). Guards every INSERT/upsert list entry.
|
||||
defs = PotatoMesh::App::DataProcessing::EXTENDED_TELEMETRY_METRIC_DEFINITIONS
|
||||
payload = { "id" => 91_100, "node_id" => "!11223344", "rx_time" => now }
|
||||
expected = {}
|
||||
defs.each_with_index do |(column, type, _key_map), index|
|
||||
value = case type
|
||||
when :float then 1.5 + index
|
||||
when :integer then 100 + index
|
||||
when :string then "value-#{index}"
|
||||
when :float_array then [1.0 + index, 2.0 + index]
|
||||
end
|
||||
payload[column] = value
|
||||
expected[column] = type == :float_array ? JSON.generate(value) : value
|
||||
end
|
||||
row = stored_row(payload)
|
||||
expected.each do |column, value|
|
||||
expect(row[column]).to eq(value), "column #{column} did not round-trip"
|
||||
end
|
||||
end
|
||||
|
||||
it "keeps stored extended values when an upsert carries NULL for them" do
|
||||
db = open_db
|
||||
dp.insert_telemetry(
|
||||
db,
|
||||
{
|
||||
"id" => 91_200, "node_id" => "!11223344", "rx_time" => now,
|
||||
"power_metrics" => { "ch1Voltage" => 3.94 },
|
||||
},
|
||||
)
|
||||
dp.insert_telemetry(
|
||||
db,
|
||||
{
|
||||
"id" => 91_200, "node_id" => "!11223344", "rx_time" => now + 1,
|
||||
"device_metrics" => { "batteryLevel" => 80 },
|
||||
},
|
||||
)
|
||||
row = db.execute("SELECT ch1_voltage, battery_level FROM telemetry WHERE id = 91200").first
|
||||
db.close
|
||||
expect(row["ch1_voltage"]).to eq(3.94)
|
||||
expect(row["battery_level"]).to eq(80.0)
|
||||
end
|
||||
|
||||
it "treats blank user strings and junk one-wire lists as absent" do
|
||||
row = stored_row(
|
||||
"id" => 91_300,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"host_metrics" => { "userString" => " ", "load1" => 12 },
|
||||
"environment_metrics" => { "oneWireTemperature" => ["junk"] },
|
||||
)
|
||||
expect(row["user_string"]).to be_nil
|
||||
expect(row["one_wire_temperature"]).to be_nil
|
||||
expect(row["load1"]).to eq(12)
|
||||
end
|
||||
|
||||
it "filters junk entries out of a mixed one-wire list" do
|
||||
row = stored_row(
|
||||
"id" => 91_301,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"environment_metrics" => { "oneWireTemperature" => ["junk", 20.0, nil, 21.25] },
|
||||
)
|
||||
expect(JSON.parse(row["one_wire_temperature"])).to eq([20.0, 21.25])
|
||||
end
|
||||
|
||||
it "ignores a non-array one_wire_temperature value" do
|
||||
row = stored_row(
|
||||
"id" => 91_302,
|
||||
"node_id" => "!11223344",
|
||||
"rx_time" => now,
|
||||
"environment_metrics" => { "temperature" => 21.5, "oneWireTemperature" => "20.0" },
|
||||
)
|
||||
expect(row["one_wire_temperature"]).to be_nil
|
||||
expect(row["temperature"]).to eq(21.5)
|
||||
end
|
||||
end
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# upsert_node — Bug 1: lastHeard = 0 must not be stored as 0
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -151,6 +151,35 @@ RSpec.describe PotatoMesh::App::Database do
|
||||
expect { harness_class.ensure_schema_upgrades }.not_to raise_error
|
||||
end
|
||||
|
||||
it "backfills every extended telemetry metric column on an existing schema (TI-A2)" do
|
||||
SQLite3::Database.new(PotatoMesh::Config.db_path) do |db|
|
||||
db.execute("CREATE TABLE nodes(node_id TEXT)")
|
||||
db.execute("CREATE TABLE messages(id INTEGER PRIMARY KEY)")
|
||||
db.execute("CREATE TABLE telemetry(id INTEGER PRIMARY KEY, rx_time INTEGER NOT NULL, rx_iso TEXT NOT NULL)")
|
||||
end
|
||||
|
||||
harness_class.ensure_schema_upgrades
|
||||
|
||||
telemetry_columns = column_names_for("telemetry")
|
||||
# Data-driven over the same definitions insert_telemetry writes with, so
|
||||
# the auto-migration can never drift from the write path.
|
||||
PotatoMesh::App::DataProcessing::EXTENDED_TELEMETRY_COLUMN_TYPES.each do |name, _type|
|
||||
expect(telemetry_columns).to include(name)
|
||||
end
|
||||
|
||||
expect { harness_class.ensure_schema_upgrades }.not_to raise_error
|
||||
end
|
||||
|
||||
it "ships every extended telemetry metric column in the fresh-install schema (TI-A2)" do
|
||||
# Assert against the bundled DDL file itself (not a migrated database) so
|
||||
# a column added to the write path can never be forgotten in
|
||||
# data/telemetry.sql — fresh installs execute that file verbatim.
|
||||
schema_sql = File.read(File.expand_path("../../data/telemetry.sql", __dir__))
|
||||
PotatoMesh::App::DataProcessing::EXTENDED_TELEMETRY_COLUMN_TYPES.each do |name, type|
|
||||
expect(schema_sql).to match(/^\s+#{Regexp.escape(name)}\s+#{type}\b/i)
|
||||
end
|
||||
end
|
||||
|
||||
it "initialises the telemetry table when it is missing" do
|
||||
SQLite3::Database.new(PotatoMesh::Config.db_path) do |db|
|
||||
db.execute("CREATE TABLE nodes(node_id TEXT)")
|
||||
|
||||
Reference in New Issue
Block a user