diff --git a/data/mesh_ingestor/handlers.py b/data/mesh_ingestor/handlers.py index 9b4a080..88daa45 100644 --- a/data/mesh_ingestor/handlers.py +++ b/data/mesh_ingestor/handlers.py @@ -38,6 +38,11 @@ _IGNORED_PACKET_LOG_PATH = ( _IGNORED_PACKET_LOCK = threading.Lock() """Lock guarding writes to :data:`_IGNORED_PACKET_LOG_PATH`.""" +_VALID_TELEMETRY_TYPES: frozenset[str] = frozenset( + {"device", "environment", "power", "air_quality"} +) +"""Allowed values for the ``telemetry_type`` discriminator field.""" + _HOST_TELEMETRY_INTERVAL_SECS = 60 * 60 """Minimum interval between accepted host telemetry packets.""" @@ -654,6 +659,10 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None: _aq = telemetry_section.get("airQualityMetrics") or telemetry_section.get( "air_quality_metrics" ) + # 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 + # packet; the elif chain handles any hypothetical overlap from future providers. if isinstance(_dm, Mapping): telemetry_type: str | None = "device" elif isinstance(_em, Mapping): @@ -665,6 +674,16 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None: else: telemetry_type = None + if telemetry_type is not None and telemetry_type not in _VALID_TELEMETRY_TYPES: + config._debug_log( + "Unexpected telemetry_type value; dropping field", + context="handlers.store_telemetry", + severity="warning", + always=True, + telemetry_type=telemetry_type, + ) + telemetry_type = None + channel = _coerce_int(_first(decoded, "channel", default=None)) if channel is None: channel = _coerce_int(_first(packet, "channel", default=None)) diff --git a/data/mesh_ingestor/provider.py b/data/mesh_ingestor/provider.py index f6e11dc..e2f5d2a 100644 --- a/data/mesh_ingestor/provider.py +++ b/data/mesh_ingestor/provider.py @@ -21,19 +21,10 @@ web app ingest contract. from __future__ import annotations -import enum from collections.abc import Iterable from typing import Protocol, runtime_checkable -class ProviderCapability(enum.Flag): - """Feature flags describing what a provider can supply.""" - - NONE = 0 - NODE_SNAPSHOT = enum.auto() - HEARTBEATS = enum.auto() - - @runtime_checkable class Provider(Protocol): """Abstract source of mesh observations.""" @@ -61,5 +52,4 @@ class Provider(Protocol): __all__ = [ "Provider", - "ProviderCapability", ] diff --git a/data/migrations/20260328_add_protocol_column.sql b/data/migrations/20260328_add_protocol_column.sql index f1e4211..285bffa 100644 --- a/data/migrations/20260328_add_protocol_column.sql +++ b/data/migrations/20260328_add_protocol_column.sql @@ -26,4 +26,14 @@ ALTER TABLE positions ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; ALTER TABLE telemetry ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; ALTER TABLE traces ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; ALTER TABLE neighbors ADD COLUMN protocol TEXT NOT NULL DEFAULT 'meshtastic'; + +-- Indices to support ?protocol= filtering on every entity endpoint without +-- full table scans as multi-protocol traffic grows. +CREATE INDEX IF NOT EXISTS idx_ingestors_protocol ON ingestors(protocol); +CREATE INDEX IF NOT EXISTS idx_nodes_protocol ON nodes(protocol); +CREATE INDEX IF NOT EXISTS idx_messages_protocol ON messages(protocol); +CREATE INDEX IF NOT EXISTS idx_positions_protocol ON positions(protocol); +CREATE INDEX IF NOT EXISTS idx_telemetry_protocol ON telemetry(protocol); +CREATE INDEX IF NOT EXISTS idx_traces_protocol ON traces(protocol); +CREATE INDEX IF NOT EXISTS idx_neighbors_protocol ON neighbors(protocol); COMMIT; diff --git a/data/migrations/20260328_add_telemetry_type.sql b/data/migrations/20260328_add_telemetry_type.sql index bb6eace..3e3c44c 100644 --- a/data/migrations/20260328_add_telemetry_type.sql +++ b/data/migrations/20260328_add_telemetry_type.sql @@ -26,15 +26,16 @@ UPDATE telemetry SET telemetry_type = 'device' AND (battery_level IS NOT NULL OR channel_utilization IS NOT NULL OR air_util_tx IS NOT NULL OR uptime_seconds IS NOT NULL); --- Power sensor: voltage/current without any device field. --- Note: device_metrics also stores a `voltage` reading (~4.2 V for battery). --- A device row that has voltage but lacks all four device-discriminator fields --- (battery_level, channel_utilization, air_util_tx, uptime_seconds) would be --- classified as 'power' here. In practice firmware always sends at least one --- of those alongside voltage, so the ambiguity is negligible for historical data. +-- Power sensor: current is the unambiguous power-sensor discriminator. +-- voltage is intentionally excluded here: device_metrics also stores a voltage +-- reading (~4.2 V for battery), so using voltage alone would misclassify device +-- rows whose four device-discriminator fields (battery_level, channel_utilization, +-- air_util_tx, uptime_seconds) happen to be NULL. Rows that have only voltage +-- and no other classifiable fields are left as NULL (unclassified), which is +-- more accurate than a wrong classification. UPDATE telemetry SET telemetry_type = 'power' WHERE telemetry_type IS NULL - AND (current IS NOT NULL OR voltage IS NOT NULL); + AND current IS NOT NULL; -- Environment: temperature/humidity/pressure UPDATE telemetry SET telemetry_type = 'environment' diff --git a/tests/test_mesh.py b/tests/test_mesh.py index 2495c4c..f9f6070 100644 --- a/tests/test_mesh.py +++ b/tests/test_mesh.py @@ -2527,6 +2527,41 @@ def test_store_packet_dict_telemetry_type_absent_for_unknown_subtype( assert "telemetry_type" not in payload +def test_store_packet_dict_invalid_telemetry_type_is_dropped(mesh_module, monkeypatch): + """A telemetry_type value that isn't in _VALID_TELEMETRY_TYPES is omitted from the payload.""" + mesh = mesh_module + captured = [] + monkeypatch.setattr( + mesh, + "_queue_post_json", + lambda path, payload, *, priority: captured.append((path, payload, priority)), + ) + + # Inject a bad type by monkey-patching the validator constant so we can + # verify the drop path without needing a real packet with an impossible type. + monkeypatch.setattr(mesh.handlers, "_VALID_TELEMETRY_TYPES", frozenset()) + + packet = { + "id": 3_000_000_010, + "rxTime": 1_758_040_000, + "fromId": "!aabbccdd", + "toId": "^all", + "decoded": { + "portnum": "TELEMETRY_APP", + "telemetry": { + "time": 1_758_040_000, + "deviceMetrics": {"batteryLevel": 80}, + }, + }, + } + + mesh.store_packet_dict(packet) + + assert captured + _, payload, _ = captured[0] + assert "telemetry_type" not in payload + + def test_store_packet_dict_throttles_host_telemetry(mesh_module, monkeypatch): mesh = mesh_module captured = [] diff --git a/web/lib/potato_mesh/application/data_processing.rb b/web/lib/potato_mesh/application/data_processing.rb index 4ed8755..053a214 100644 --- a/web/lib/potato_mesh/application/data_processing.rb +++ b/web/lib/potato_mesh/application/data_processing.rb @@ -17,6 +17,9 @@ module PotatoMesh module App module DataProcessing + # Allowed values for the +telemetry_type+ discriminator column. + VALID_TELEMETRY_TYPES = %w[device environment power air_quality].freeze + # Coerce a Ruby boolean into a SQLite integer (1/0) while passing through # any other value unchanged. Used when writing boolean node fields. # @@ -1071,6 +1074,7 @@ module PotatoMesh air_quality_metrics ||= normalize_json_object(telemetry_section["airQualityMetrics"]) if telemetry_section&.key?("airQualityMetrics") telemetry_type = string_or_nil(payload["telemetry_type"]) + telemetry_type = nil unless VALID_TELEMETRY_TYPES.include?(telemetry_type) telemetry_type ||= if device_metrics&.any? "device" elsif environment_metrics&.any? diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index 2db7326..d9c06dc 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -4066,6 +4066,24 @@ RSpec.describe "Potato Mesh Sinatra app" do expect_stored_telemetry_type(24_005, "air_quality") end + it "rejects an invalid telemetry_type and falls back to metric inference" do + payload = [ + { + "id" => 24_006, + "node_id" => "!teltype06", + "rx_time" => reference_time.to_i - 50, + "telemetry_type" => "bogus_value", + "device_metrics" => { "battery_level" => 55, "channel_utilization" => 20 }, + }, + ] + + post "/api/telemetry", payload.to_json, auth_headers + + expect(last_response).to be_ok + # Invalid explicit type must be discarded; device_metrics inference takes over. + expect_stored_telemetry_type(24_006, "device") + end + it "returns 400 when more than 1000 telemetry packets are provided" do payload = Array.new(1001) { |i| { "id" => i + 1, "rx_time" => reference_time.to_i - i } }