diff --git a/ACCEPTANCE.md b/ACCEPTANCE.md index 83e31bd..e84adfb 100644 --- a/ACCEPTANCE.md +++ b/ACCEPTANCE.md @@ -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. diff --git a/README.md b/README.md index ab77629..e67da66 100644 --- a/README.md +++ b/README.md @@ -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. | diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 0978686..04e251f 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -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: diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index d837fdf..2d7ba73 100644 --- a/data/mesh_ingestor/config.py +++ b/data/mesh_ingestor/config.py @@ -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. diff --git a/data/mesh_ingestor/handlers/telemetry.py b/data/mesh_ingestor/handlers/telemetry.py index d02435a..3734b51 100644 --- a/data/mesh_ingestor/handlers/telemetry.py +++ b/data/mesh_ingestor/handlers/telemetry.py @@ -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 diff --git a/data/mesh_ingestor/protocols/meshcore/handlers.py b/data/mesh_ingestor/protocols/meshcore/handlers.py index 12dfad3..078ea4c 100644 --- a/data/mesh_ingestor/protocols/meshcore/handlers.py +++ b/data/mesh_ingestor/protocols/meshcore/handlers.py @@ -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), } diff --git a/data/mesh_ingestor/protocols/meshcore/runner.py b/data/mesh_ingestor/protocols/meshcore/runner.py index f7ab568..4637ce4 100644 --- a/data/mesh_ingestor/protocols/meshcore/runner.py +++ b/data/mesh_ingestor/protocols/meshcore/runner.py @@ -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(): diff --git a/data/mesh_ingestor/protocols/meshcore/telemetry.py b/data/mesh_ingestor/protocols/meshcore/telemetry.py new file mode 100644 index 0000000..42722d6 --- /dev/null +++ b/data/mesh_ingestor/protocols/meshcore/telemetry.py @@ -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())) diff --git a/data/migrations/20260722_add_extended_telemetry_metrics.sql b/data/migrations/20260722_add_extended_telemetry_metrics.sql new file mode 100644 index 0000000..e0539ab --- /dev/null +++ b/data/migrations/20260722_add_extended_telemetry_metrics.sql @@ -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; diff --git a/data/telemetry.sql b/data/telemetry.sql index d270e58..e0dd8ab 100644 --- a/data/telemetry.sql +++ b/data/telemetry.sql @@ -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); diff --git a/tests/test_handlers_unit.py b/tests/test_handlers_unit.py index 4e292f9..a54f291 100644 --- a/tests/test_handlers_unit.py +++ b/tests/test_handlers_unit.py @@ -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 # --------------------------------------------------------------------------- diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py index a1c23e0..aabb8b5 100644 --- a/tests/test_provider_unit.py +++ b/tests/test_provider_unit.py @@ -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", diff --git a/web/lib/potato_mesh/application/data_processing.rb b/web/lib/potato_mesh/application/data_processing.rb index 3b8779a..dc37ded 100644 --- a/web/lib/potato_mesh/application/data_processing.rb +++ b/web/lib/potato_mesh/application/data_processing.rb @@ -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" diff --git a/web/lib/potato_mesh/application/data_processing/coercions.rb b/web/lib/potato_mesh/application/data_processing/coercions.rb index 8c9e03f..9057134 100644 --- a/web/lib/potato_mesh/application/data_processing/coercions.rb +++ b/web/lib/potato_mesh/application/data_processing/coercions.rb @@ -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 diff --git a/web/lib/potato_mesh/application/data_processing/telemetry.rb b/web/lib/potato_mesh/application/data_processing/telemetry.rb index 67eefd9..954468f 100644 --- a/web/lib/potato_mesh/application/data_processing/telemetry.rb +++ b/web/lib/potato_mesh/application/data_processing/telemetry.rb @@ -267,8 +267,11 @@ module PotatoMesh # # @param key_map [Hash{Symbol=>Array}] 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 diff --git a/web/lib/potato_mesh/application/data_processing/telemetry_metrics.rb b/web/lib/potato_mesh/application/data_processing/telemetry_metrics.rb new file mode 100644 index 0000000..38a6e5e --- /dev/null +++ b/web/lib/potato_mesh/application/data_processing/telemetry_metrics.rb @@ -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] list of +[column, type, camel_name]+. + # @return [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 diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb index 48d9086..c4474fd 100644 --- a/web/lib/potato_mesh/application/database.rb +++ b/web/lib/potato_mesh/application/database.rb @@ -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}") diff --git a/web/spec/data_processing_spec.rb b/web/spec/data_processing_spec.rb index fcd248e..d9eb64f 100644 --- a/web/spec/data_processing_spec.rb +++ b/web/spec/data_processing_spec.rb @@ -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 # --------------------------------------------------------------------------- diff --git a/web/spec/database_spec.rb b/web/spec/database_spec.rb index e52f176..8f09572 100644 --- a/web/spec/database_spec.rb +++ b/web/spec/database_spec.rb @@ -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)")