diff --git a/data/mesh.py b/data/mesh.py index 6b00ee3..eaa0091 100644 --- a/data/mesh.py +++ b/data/mesh.py @@ -159,6 +159,7 @@ _POST_QUEUE_ACTIVE = False _MESSAGE_POST_PRIORITY = 0 _POSITION_POST_PRIORITY = 10 +_TELEMETRY_POST_PRIORITY = 15 _NODE_POST_PRIORITY = 20 _DEFAULT_POST_PRIORITY = 50 @@ -859,7 +860,7 @@ def store_position_packet(packet: dict, decoded: Mapping): position_payload = { "id": pkt_id, - "node_id": node_id, + "node_id": node_id or raw_from, "node_num": node_num, "num": node_num, "from_id": node_id, @@ -895,6 +896,160 @@ def store_position_packet(packet: dict, decoded: Mapping): ) +def store_telemetry_packet(packet: dict, decoded: Mapping): + """Handle ``TELEMETRY_APP`` packets and forward them to ``/api/telemetry``.""" + + telemetry_section = ( + decoded.get("telemetry") if isinstance(decoded, Mapping) else None + ) + if not isinstance(telemetry_section, Mapping): + return + + pkt_id = _coerce_int(_first(packet, "id", "packet_id", "packetId", default=None)) + if pkt_id is None: + return + + raw_from = _first(packet, "fromId", "from_id", "from", default=None) + node_id = _canonical_node_id(raw_from) + node_num = _coerce_int(_first(decoded, "num", "node_num", default=None)) + if node_num is None: + node_num = _node_num_from_id(node_id or raw_from) + + to_id = _first(packet, "toId", "to_id", "to", default=None) + + raw_rx_time = _first(packet, "rxTime", "rx_time", default=time.time()) + try: + rx_time = int(raw_rx_time) + except (TypeError, ValueError): + rx_time = int(time.time()) + rx_iso = _iso(rx_time) + + telemetry_time = _coerce_int(_first(telemetry_section, "time", default=None)) + + channel = _coerce_int(_first(decoded, "channel", default=None)) + if channel is None: + channel = _coerce_int(_first(packet, "channel", default=None)) + if channel is None: + channel = 0 + + bitfield = _coerce_int(_first(decoded, "bitfield", default=None)) + snr = _coerce_float(_first(packet, "snr", "rx_snr", "rxSnr", default=None)) + rssi = _coerce_int(_first(packet, "rssi", "rx_rssi", "rxRssi", default=None)) + hop_limit = _coerce_int(_first(packet, "hopLimit", "hop_limit", default=None)) + + portnum_raw = _first(decoded, "portnum", default=None) + portnum = str(portnum_raw) if portnum_raw is not None else None + + payload_bytes = _extract_payload_bytes(decoded) + payload_b64 = ( + base64.b64encode(payload_bytes).decode("ascii") if payload_bytes else None + ) + + device_metrics_section = telemetry_section.get( + "deviceMetrics" + ) or telemetry_section.get("device_metrics") + device_metrics = ( + _node_to_dict(device_metrics_section) + if isinstance(device_metrics_section, Mapping) + else None + ) + + environment_section = telemetry_section.get( + "environmentMetrics" + ) or telemetry_section.get("environment_metrics") + environment_metrics = ( + _node_to_dict(environment_section) + if isinstance(environment_section, Mapping) + else None + ) + + metrics_lookup = lambda mapping, *names: _first(mapping or {}, *names, default=None) + + battery_level = _coerce_float( + metrics_lookup(device_metrics, "batteryLevel", "battery_level") + ) + voltage = _coerce_float(metrics_lookup(device_metrics, "voltage")) + channel_utilization = _coerce_float( + metrics_lookup(device_metrics, "channelUtilization", "channel_utilization") + ) + air_util_tx = _coerce_float( + metrics_lookup(device_metrics, "airUtilTx", "air_util_tx") + ) + uptime_seconds = _coerce_int( + metrics_lookup(device_metrics, "uptimeSeconds", "uptime_seconds") + ) + + temperature = _coerce_float( + metrics_lookup( + environment_metrics, + "temperature", + "temperatureC", + "temperature_c", + "tempC", + ) + ) + relative_humidity = _coerce_float( + metrics_lookup( + environment_metrics, + "relativeHumidity", + "relative_humidity", + "humidity", + ) + ) + barometric_pressure = _coerce_float( + metrics_lookup( + environment_metrics, + "barometricPressure", + "barometric_pressure", + "pressure", + ) + ) + + telemetry_payload = { + "id": pkt_id, + "node_id": node_id, + "node_num": node_num, + "from_id": node_id or raw_from, + "to_id": to_id, + "rx_time": rx_time, + "rx_iso": rx_iso, + "telemetry_time": telemetry_time, + "channel": channel, + "portnum": portnum, + "bitfield": bitfield, + "snr": snr, + "rssi": rssi, + "hop_limit": hop_limit, + "payload_b64": payload_b64, + } + + if battery_level is not None: + telemetry_payload["battery_level"] = battery_level + if voltage is not None: + telemetry_payload["voltage"] = voltage + if channel_utilization is not None: + telemetry_payload["channel_utilization"] = channel_utilization + if air_util_tx is not None: + telemetry_payload["air_util_tx"] = air_util_tx + if uptime_seconds is not None: + telemetry_payload["uptime_seconds"] = uptime_seconds + if temperature is not None: + telemetry_payload["temperature"] = temperature + if relative_humidity is not None: + telemetry_payload["relative_humidity"] = relative_humidity + if barometric_pressure is not None: + telemetry_payload["barometric_pressure"] = barometric_pressure + + _queue_post_json( + "/api/telemetry", telemetry_payload, priority=_TELEMETRY_POST_PRIORITY + ) + + if DEBUG: + print( + f"[debug] stored telemetry for {node_id!r} rx_time={rx_time} battery={battery_level!r}" + ) + + def store_nodeinfo_packet(packet: dict, decoded: Mapping): """Handle ``NODEINFO_APP`` packets and forward them to ``/api/nodes``.""" @@ -1060,6 +1215,16 @@ def store_packet_dict(p: dict): portnum_raw = _first(dec, "portnum", default=None) portnum = str(portnum_raw).upper() if portnum_raw is not None else None + portnum_int = _coerce_int(portnum_raw) + + telemetry_section = dec.get("telemetry") if isinstance(dec, Mapping) else None + if ( + portnum == "TELEMETRY_APP" + or portnum_int == 65 + or isinstance(telemetry_section, Mapping) + ): + store_telemetry_packet(p, dec) + return if portnum in {"5", "NODEINFO_APP"}: store_nodeinfo_packet(p, dec) diff --git a/data/telemetry.sql b/data/telemetry.sql new file mode 100644 index 0000000..37c2ecc --- /dev/null +++ b/data/telemetry.sql @@ -0,0 +1,43 @@ +-- Copyright (C) 2025 l5yth +-- +-- 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. + +CREATE TABLE IF NOT EXISTS telemetry ( + id INTEGER PRIMARY KEY, + node_id TEXT, + node_num INTEGER, + from_id TEXT, + to_id TEXT, + rx_time INTEGER NOT NULL, + rx_iso TEXT NOT NULL, + telemetry_time INTEGER, + channel INTEGER, + portnum TEXT, + hop_limit INTEGER, + snr REAL, + rssi INTEGER, + bitfield INTEGER, + payload_b64 TEXT, + battery_level REAL, + voltage REAL, + channel_utilization REAL, + air_util_tx REAL, + uptime_seconds INTEGER, + temperature REAL, + relative_humidity REAL, + barometric_pressure REAL +); + +CREATE INDEX IF NOT EXISTS idx_telemetry_rx_time ON telemetry(rx_time); +CREATE INDEX IF NOT EXISTS idx_telemetry_node_id ON telemetry(node_id); +CREATE INDEX IF NOT EXISTS idx_telemetry_time ON telemetry(telemetry_time); diff --git a/tests/telemetry.json b/tests/telemetry.json new file mode 100644 index 0000000..bc84727 --- /dev/null +++ b/tests/telemetry.json @@ -0,0 +1,84 @@ +[ + { + "id": 1256091342, + "node_id": "!9e95cf60", + "from_id": "!9e95cf60", + "to_id": "^all", + "rx_time": 1758024300, + "rx_iso": "2025-09-16T12:05:00Z", + "telemetry_time": 1758024300, + "channel": 0, + "portnum": "TELEMETRY_APP", + "battery_level": 101, + "bitfield": 1, + "payload_b64": "DTVr0mgSFQhlFQIrh0AdJb8YPyXYFSA9KJTPEg==", + "device_metrics": { + "batteryLevel": 101, + "voltage": 4.224, + "channelUtilization": 0.59666663, + "airUtilTx": 0.03908333, + "uptimeSeconds": 305044 + }, + "raw": { + "device_metrics": { + "battery_level": 101, + "voltage": 4.224, + "channel_utilization": 0.59666663, + "air_util_tx": 0.03908333, + "uptime_seconds": 305044 + } + } + }, + { + "id": 2817720548, + "node_id": "!2a2a2a2a", + "from_id": "!2a2a2a2a", + "to_id": "^all", + "rx_time": 1758024400, + "rx_iso": "2025-09-16T12:06:40Z", + "telemetry_time": 1758024390, + "channel": 0, + "portnum": "TELEMETRY_APP", + "bitfield": 1, + "environment_metrics": { + "temperature": 21.98, + "relativeHumidity": 39.475586, + "barometricPressure": 1017.8353 + }, + "raw": { + "environment_metrics": { + "temperature": 21.98, + "relative_humidity": 39.475586, + "barometric_pressure": 1017.8353 + } + } + }, + { + "id": 345678901, + "node_id": "!1234abcd", + "from_id": "!1234abcd", + "node_num": 305441741, + "to_id": "^all", + "rx_time": 1758024500, + "rx_iso": "2025-09-16T12:08:20Z", + "telemetry_time": 1758024450, + "channel": 1, + "portnum": "TELEMETRY_APP", + "payload_b64": "AAEC", + "device_metrics": { + "battery_level": 58.5, + "voltage": 3.92, + "channel_utilization": 0.284, + "air_util_tx": 0.051, + "uptime_seconds": 86400 + }, + "local_stats": { + "numPacketsTx": 1280, + "numPacketsRx": 1425, + "numClients": 6, + "numNodes": 18, + "freeHeap": 21344, + "heapLowWater": 19876 + } + } +] diff --git a/tests/test_mesh.py b/tests/test_mesh.py index e25f8fe..36c0692 100644 --- a/tests/test_mesh.py +++ b/tests/test_mesh.py @@ -979,6 +979,104 @@ def test_store_packet_dict_includes_encrypted_payload(mesh_module, monkeypatch): assert priority == mesh._MESSAGE_POST_PRIORITY +def test_store_packet_dict_handles_telemetry_packet(mesh_module, monkeypatch): + mesh = mesh_module + captured = [] + monkeypatch.setattr( + mesh, + "_queue_post_json", + lambda path, payload, *, priority: captured.append((path, payload, priority)), + ) + + packet = { + "id": 1_256_091_342, + "rxTime": 1_758_024_300, + "fromId": "!9e95cf60", + "toId": "^all", + "decoded": { + "portnum": "TELEMETRY_APP", + "bitfield": 1, + "telemetry": { + "time": 1_758_024_300, + "deviceMetrics": { + "batteryLevel": 101, + "voltage": 4.224, + "channelUtilization": 0.59666663, + "airUtilTx": 0.03908333, + "uptimeSeconds": 305044, + }, + "localStats": { + "numPacketsTx": 1280, + "numPacketsRx": 1425, + }, + }, + "payload": { + "__bytes_b64__": "DTVr0mgSFQhlFQIrh0AdJb8YPyXYFSA9KJTPEg==", + }, + }, + } + + mesh.store_packet_dict(packet) + + assert captured + path, payload, priority = captured[0] + assert path == "/api/telemetry" + assert priority == mesh._TELEMETRY_POST_PRIORITY + assert payload["id"] == 1_256_091_342 + assert payload["node_id"] == "!9e95cf60" + assert payload["from_id"] == "!9e95cf60" + assert payload["rx_time"] == 1_758_024_300 + assert payload["telemetry_time"] == 1_758_024_300 + assert payload["channel"] == 0 + assert payload["bitfield"] == 1 + assert payload["payload_b64"] == "DTVr0mgSFQhlFQIrh0AdJb8YPyXYFSA9KJTPEg==" + assert payload["battery_level"] == pytest.approx(101.0) + assert payload["voltage"] == pytest.approx(4.224) + assert payload["channel_utilization"] == pytest.approx(0.59666663) + assert payload["air_util_tx"] == pytest.approx(0.03908333) + assert payload["uptime_seconds"] == 305044 + + +def test_store_packet_dict_handles_environment_telemetry(mesh_module, monkeypatch): + mesh = mesh_module + captured = [] + monkeypatch.setattr( + mesh, + "_queue_post_json", + lambda path, payload, *, priority: captured.append((path, payload, priority)), + ) + + packet = { + "id": 2_817_720_548, + "rxTime": 1_758_024_400, + "from": 3_698_627_780, + "decoded": { + "portnum": "TELEMETRY_APP", + "telemetry": { + "time": 1_758_024_390, + "environmentMetrics": { + "temperature": 21.98, + "relativeHumidity": 39.475586, + "barometricPressure": 1017.8353, + }, + }, + }, + } + + mesh.store_packet_dict(packet) + + assert captured + path, payload, priority = captured[0] + assert path == "/api/telemetry" + assert payload["id"] == 2_817_720_548 + assert payload["node_id"] == "!dc7494c4" + assert payload["from_id"] == "!dc7494c4" + assert payload["telemetry_time"] == 1_758_024_390 + assert payload["temperature"] == pytest.approx(21.98) + assert payload["relative_humidity"] == pytest.approx(39.475586) + assert payload["barometric_pressure"] == pytest.approx(1017.8353) + + def test_post_queue_prioritises_messages(mesh_module, monkeypatch): mesh = mesh_module mesh._clear_post_queue() diff --git a/web/app.rb b/web/app.rb index fbb818d..87e73f1 100644 --- a/web/app.rb +++ b/web/app.rb @@ -177,6 +177,37 @@ def coerce_float(value) end end +def normalize_json_value(value) + case value + when Hash + value.each_with_object({}) do |(key, val), memo| + memo[key.to_s] = normalize_json_value(val) + end + when Array + value.map { |element| normalize_json_value(element) } + else + value + end +end + +def normalize_json_object(value) + case value + when Hash + normalize_json_value(value) + when String + trimmed = value.strip + return nil if trimmed.empty? + begin + parsed = JSON.parse(trimmed) + rescue JSON::ParserError + return nil + end + parsed.is_a?(Hash) ? normalize_json_value(parsed) : nil + else + nil + end +end + def sanitized_max_distance_km return nil unless defined?(MAX_NODE_DISTANCE_KM) @@ -284,8 +315,8 @@ end def db_schema_present? return false unless File.exist?(DB_PATH) db = open_database(readonly: true) - required = %w[nodes messages positions] - tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions')").flatten + required = %w[nodes messages positions telemetry] + tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions','telemetry')").flatten (required - tables).empty? rescue SQLite3::Exception false @@ -299,7 +330,7 @@ end def init_db FileUtils.mkdir_p(File.dirname(DB_PATH)) db = open_database - %w[nodes messages positions].each do |schema| + %w[nodes messages positions telemetry].each do |schema| sql_file = File.expand_path("../data/#{schema}.sql", __dir__) db.execute_batch(File.read(sql_file)) end @@ -473,6 +504,52 @@ ensure db&.close end +def query_telemetry(limit) + db = open_database(readonly: true) + db.results_as_hash = true + rows = db.execute <<~SQL, [limit] + SELECT 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 + FROM telemetry + ORDER BY rx_time DESC + LIMIT ? + SQL + now = Time.now.to_i + rows.each do |r| + rx_time = coerce_integer(r["rx_time"]) + r["rx_time"] = rx_time if rx_time + r["rx_iso"] = Time.at(rx_time).utc.iso8601 if rx_time && string_or_nil(r["rx_iso"]).nil? + + node_num = coerce_integer(r["node_num"]) + r["node_num"] = node_num if node_num + + telemetry_time = coerce_integer(r["telemetry_time"]) + telemetry_time = nil if telemetry_time && telemetry_time > now + r["telemetry_time"] = telemetry_time + r["telemetry_time_iso"] = Time.at(telemetry_time).utc.iso8601 if telemetry_time + + r["channel"] = coerce_integer(r["channel"]) + r["hop_limit"] = coerce_integer(r["hop_limit"]) + r["rssi"] = coerce_integer(r["rssi"]) + r["bitfield"] = coerce_integer(r["bitfield"]) + r["snr"] = coerce_float(r["snr"]) + r["battery_level"] = coerce_float(r["battery_level"]) + r["voltage"] = coerce_float(r["voltage"]) + r["channel_utilization"] = coerce_float(r["channel_utilization"]) + r["air_util_tx"] = coerce_float(r["air_util_tx"]) + r["uptime_seconds"] = coerce_integer(r["uptime_seconds"]) + r["temperature"] = coerce_float(r["temperature"]) + r["relative_humidity"] = coerce_float(r["relative_humidity"]) + r["barometric_pressure"] = coerce_float(r["barometric_pressure"]) + end + rows +ensure + db&.close +end + # GET /api/messages # # Returns a JSON array of stored text messages including node metadata. @@ -492,6 +569,15 @@ get "/api/positions" do query_positions(limit).to_json end +# GET /api/telemetry +# +# Returns a JSON array of recorded telemetry packets. +get "/api/telemetry" do + content_type :json + limit = [params["limit"]&.to_i || 200, 1000].min + query_telemetry(limit).to_json +end + # Determine the numeric node reference for a canonical node identifier. # # The Meshtastic protobuf encodes the node ID as a hexadecimal string prefixed @@ -1037,6 +1123,228 @@ def insert_position(db, payload) ) end +def update_node_from_telemetry(db, node_id, node_num, _rx_time, metrics = {}) + num = coerce_integer(node_num) + id = string_or_nil(node_id) + if id&.start_with?("!") + id = "!#{id.delete_prefix("!").downcase}" + end + id ||= format("!%08x", num & 0xFFFFFFFF) if num + return unless id + + now = Time.now.to_i + + ensure_unknown_node(db, id, num, heard_time: now) + + battery = coerce_float(metrics[:battery_level] || metrics["battery_level"]) + voltage = coerce_float(metrics[:voltage] || metrics["voltage"]) + channel_util = coerce_float(metrics[:channel_utilization] || metrics["channel_utilization"]) + air_util_tx = coerce_float(metrics[:air_util_tx] || metrics["air_util_tx"]) + uptime = coerce_integer(metrics[:uptime_seconds] || metrics["uptime_seconds"]) + + assignments = [] + params = [] + + if num + assignments << "num = ?" + params << num + end + + assignments << "last_heard = ?" + params << now + + assignments << "first_heard = COALESCE(first_heard, ?)" + params << now + + metric_updates = { + "battery_level" => battery, + "voltage" => voltage, + "channel_utilization" => channel_util, + "air_util_tx" => air_util_tx, + "uptime_seconds" => uptime, + } + + metric_updates.each do |column, value| + next if value.nil? + + assignments << "#{column} = ?" + params << value + end + + return if assignments.empty? + + assignments_sql = assignments.join(", ") + params << id + + with_busy_retry do + db.execute("UPDATE nodes SET #{assignments_sql} WHERE node_id = ?", params) + end +end + +def insert_telemetry(db, payload) + return unless payload.is_a?(Hash) + + telemetry_id = coerce_integer(payload["id"] || payload["packet_id"]) + return unless telemetry_id + + now = Time.now.to_i + rx_time = coerce_integer(payload["rx_time"]) + rx_time = now if rx_time.nil? || rx_time > now + rx_iso = string_or_nil(payload["rx_iso"]) + rx_iso ||= Time.at(rx_time).utc.iso8601 + + raw_node_id = payload["node_id"] || payload["from_id"] || payload["from"] + node_id = string_or_nil(raw_node_id) + node_id = "!#{node_id.delete_prefix("!").downcase}" if node_id&.start_with?("!") + raw_node_num = coerce_integer(payload["node_num"]) || coerce_integer(payload["num"]) + + payload_for_num = payload.dup + payload_for_num["num"] ||= raw_node_num if raw_node_num + node_num = resolve_node_num(node_id, payload_for_num) + node_num ||= raw_node_num + + canonical = normalize_node_id(db, node_id || node_num) + node_id = canonical if canonical + + from_id = string_or_nil(payload["from_id"]) || node_id + to_id = string_or_nil(payload["to_id"] || payload["to"]) + + telemetry_time = coerce_integer(payload["telemetry_time"] || payload["time"] || payload.dig("telemetry", "time")) + telemetry_time = nil if telemetry_time && telemetry_time > now + + channel = coerce_integer(payload["channel"]) + portnum = string_or_nil(payload["portnum"]) + hop_limit = coerce_integer(payload["hop_limit"] || payload["hopLimit"]) + snr = coerce_float(payload["snr"]) + rssi = coerce_integer(payload["rssi"]) + bitfield = coerce_integer(payload["bitfield"]) + payload_b64 = string_or_nil(payload["payload_b64"] || payload["payload"]) + + telemetry_section = normalize_json_object(payload["telemetry"]) + device_metrics = normalize_json_object(payload["device_metrics"] || payload["deviceMetrics"]) + device_metrics ||= normalize_json_object(telemetry_section["deviceMetrics"]) if telemetry_section&.key?("deviceMetrics") + environment_metrics = normalize_json_object(payload["environment_metrics"] || payload["environmentMetrics"]) + environment_metrics ||= normalize_json_object(telemetry_section["environmentMetrics"]) if telemetry_section&.key?("environmentMetrics") + + fetch_metric = lambda do |map, *names| + next nil unless map.is_a?(Hash) + names.each do |name| + next unless name + key = name.to_s + return map[key] if map.key?(key) + end + nil + end + + battery_level = payload.key?("battery_level") ? payload["battery_level"] : nil + battery_level = coerce_float(battery_level) + battery_level ||= coerce_float(fetch_metric.call(device_metrics, :battery_level, :batteryLevel)) + + voltage = payload.key?("voltage") ? payload["voltage"] : nil + voltage = coerce_float(voltage) + voltage ||= coerce_float(fetch_metric.call(device_metrics, :voltage)) + + channel_utilization = payload.key?("channel_utilization") ? payload["channel_utilization"] : nil + channel_utilization ||= payload["channelUtilization"] if payload.key?("channelUtilization") + channel_utilization = coerce_float(channel_utilization) + channel_utilization ||= coerce_float(fetch_metric.call(device_metrics, :channel_utilization, :channelUtilization)) + + air_util_tx = payload.key?("air_util_tx") ? payload["air_util_tx"] : nil + air_util_tx ||= payload["airUtilTx"] if payload.key?("airUtilTx") + air_util_tx = coerce_float(air_util_tx) + air_util_tx ||= coerce_float(fetch_metric.call(device_metrics, :air_util_tx, :airUtilTx)) + + uptime_seconds = payload.key?("uptime_seconds") ? payload["uptime_seconds"] : nil + uptime_seconds ||= payload["uptimeSeconds"] if payload.key?("uptimeSeconds") + uptime_seconds = coerce_integer(uptime_seconds) + uptime_seconds ||= coerce_integer(fetch_metric.call(device_metrics, :uptime_seconds, :uptimeSeconds)) + + temperature = payload.key?("temperature") ? payload["temperature"] : nil + temperature = coerce_float(temperature) + temperature ||= coerce_float(fetch_metric.call(environment_metrics, :temperature, :temperatureC, :temperature_c, :tempC)) + + relative_humidity = payload.key?("relative_humidity") ? payload["relative_humidity"] : nil + relative_humidity ||= payload["relativeHumidity"] if payload.key?("relativeHumidity") + relative_humidity ||= payload["humidity"] if payload.key?("humidity") + relative_humidity = coerce_float(relative_humidity) + relative_humidity ||= coerce_float(fetch_metric.call(environment_metrics, :relative_humidity, :relativeHumidity, :humidity)) + + barometric_pressure = payload.key?("barometric_pressure") ? payload["barometric_pressure"] : nil + barometric_pressure ||= payload["barometricPressure"] if payload.key?("barometricPressure") + barometric_pressure ||= payload["pressure"] if payload.key?("pressure") + barometric_pressure = coerce_float(barometric_pressure) + barometric_pressure ||= coerce_float(fetch_metric.call(environment_metrics, :barometric_pressure, :barometricPressure, :pressure)) + + row = [ + 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, + ] + + 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) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(id) DO UPDATE SET + node_id=COALESCE(excluded.node_id,telemetry.node_id), + node_num=COALESCE(excluded.node_num,telemetry.node_num), + from_id=COALESCE(excluded.from_id,telemetry.from_id), + to_id=COALESCE(excluded.to_id,telemetry.to_id), + rx_time=excluded.rx_time, + rx_iso=excluded.rx_iso, + telemetry_time=COALESCE(excluded.telemetry_time,telemetry.telemetry_time), + channel=COALESCE(excluded.channel,telemetry.channel), + portnum=COALESCE(excluded.portnum,telemetry.portnum), + hop_limit=COALESCE(excluded.hop_limit,telemetry.hop_limit), + snr=COALESCE(excluded.snr,telemetry.snr), + rssi=COALESCE(excluded.rssi,telemetry.rssi), + bitfield=COALESCE(excluded.bitfield,telemetry.bitfield), + payload_b64=COALESCE(excluded.payload_b64,telemetry.payload_b64), + battery_level=COALESCE(excluded.battery_level,telemetry.battery_level), + voltage=COALESCE(excluded.voltage,telemetry.voltage), + channel_utilization=COALESCE(excluded.channel_utilization,telemetry.channel_utilization), + air_util_tx=COALESCE(excluded.air_util_tx,telemetry.air_util_tx), + uptime_seconds=COALESCE(excluded.uptime_seconds,telemetry.uptime_seconds), + temperature=COALESCE(excluded.temperature,telemetry.temperature), + relative_humidity=COALESCE(excluded.relative_humidity,telemetry.relative_humidity), + barometric_pressure=COALESCE(excluded.barometric_pressure,telemetry.barometric_pressure) + SQL + end + + update_node_from_telemetry( + db, + node_id || from_id, + node_num, + rx_time, + battery_level: battery_level, + voltage: voltage, + channel_utilization: channel_utilization, + air_util_tx: air_util_tx, + uptime_seconds: uptime_seconds, + ) +end + # Insert a text message if it does not already exist. # # @param db [SQLite3::Database] open database handle. @@ -1258,6 +1566,28 @@ ensure db&.close end +# POST /api/telemetry +# +# Accepts an array or object describing telemetry packets and stores each entry. +post "/api/telemetry" do + require_token! + content_type :json + begin + data = JSON.parse(read_json_body) + rescue JSON::ParserError + halt 400, { error: "invalid JSON" }.to_json + end + telemetry_packets = data.is_a?(Array) ? data : [data] + halt 400, { error: "too many telemetry packets" }.to_json if telemetry_packets.size > 1000 + db = open_database + telemetry_packets.each do |packet| + insert_telemetry(db, packet) + end + { status: "ok" }.to_json +ensure + db&.close +end + get "/potatomesh-logo.svg" do # Sinatra знает корень через settings.root (обычно это каталог app.rb) path = File.expand_path("potatomesh-logo.svg", settings.public_folder) diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index 965267f..99a63b7 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -39,6 +39,7 @@ RSpec.describe "Potato Mesh Sinatra app" do db.execute("DELETE FROM messages") db.execute("DELETE FROM nodes") db.execute("DELETE FROM positions") + db.execute("DELETE FROM telemetry") end end @@ -143,6 +144,7 @@ RSpec.describe "Potato Mesh Sinatra app" do end let(:nodes_fixture) { JSON.parse(File.read(fixture_path("nodes.json"))) } let(:messages_fixture) { JSON.parse(File.read(fixture_path("messages.json"))) } + let(:telemetry_fixture) { JSON.parse(File.read(fixture_path("telemetry.json"))) } let(:reference_time) do latest = nodes_fixture.map { |node| node["last_heard"] }.compact.max Time.at((latest || Time.now.to_i) + 1000) @@ -850,6 +852,94 @@ RSpec.describe "Potato Mesh Sinatra app" do end end + describe "POST /api/telemetry" do + it "stores telemetry packets and updates node metrics" do + payload = telemetry_fixture + + post "/api/telemetry", payload.to_json, auth_headers + + expect(last_response).to be_ok + expect(JSON.parse(last_response.body)).to eq("status" => "ok") + + with_db(readonly: true) do |db| + db.results_as_hash = true + rows = db.execute( + "SELECT * FROM telemetry ORDER BY id", + ) + + expect(rows.size).to eq(payload.size) + + first = rows.find { |row| row["id"] == payload[0]["id"] } + expect(first).not_to be_nil + expect(first["node_id"]).to eq(payload[0]["node_id"]) + expect(first["rx_time"]).to eq(payload[0]["rx_time"]) + expect_same_value(first["battery_level"], payload[0]["battery_level"]) + expect_same_value(first["voltage"], payload[0].dig("device_metrics", "voltage")) + expect_same_value(first["channel_utilization"], payload[0].dig("device_metrics", "channelUtilization")) + expect_same_value(first["air_util_tx"], payload[0].dig("device_metrics", "airUtilTx")) + expect(first["uptime_seconds"]).to eq(payload[0].dig("device_metrics", "uptimeSeconds")) + + environment_row = rows.find { |row| row["id"] == payload[1]["id"] } + expect(environment_row["temperature"]).to be_within(1e-6).of(payload[1].dig("environment_metrics", "temperature")) + expect(environment_row["relative_humidity"]).to be_within(1e-6).of(payload[1].dig("environment_metrics", "relativeHumidity")) + expect(environment_row["barometric_pressure"]).to be_within(1e-6).of(payload[1].dig("environment_metrics", "barometricPressure")) + end + + with_db(readonly: true) do |db| + db.results_as_hash = true + + metrics_node = db.get_first_row( + "SELECT battery_level, voltage, channel_utilization, air_util_tx, uptime_seconds, last_heard, first_heard FROM nodes WHERE node_id = ?", + [payload[0]["node_id"]], + ) + expect_same_value(metrics_node["battery_level"], payload[0]["battery_level"]) + expect_same_value(metrics_node["voltage"], payload[0]["device_metrics"]["voltage"]) + expect_same_value(metrics_node["channel_utilization"], payload[0]["device_metrics"]["channelUtilization"]) + expect_same_value(metrics_node["air_util_tx"], payload[0]["device_metrics"]["airUtilTx"]) + expect(metrics_node["uptime_seconds"]).to eq(payload[0]["device_metrics"]["uptimeSeconds"]) + expect(metrics_node["last_heard"]).to eq(reference_time.to_i) + expect(metrics_node["first_heard"]).to eq(reference_time.to_i) + + env_node = db.get_first_row( + "SELECT last_heard, battery_level, voltage FROM nodes WHERE node_id = ?", + [payload[1]["node_id"]], + ) + expect(env_node["last_heard"]).to eq(reference_time.to_i) + expect(env_node["battery_level"]).to be_nil + expect(env_node["voltage"]).to be_nil + + local_node = db.get_first_row( + "SELECT battery_level, uptime_seconds, last_heard FROM nodes WHERE node_id = ?", + [payload[2]["node_id"]], + ) + expect_same_value(local_node["battery_level"], payload[2]["device_metrics"]["battery_level"]) + expect(local_node["uptime_seconds"]).to eq(payload[2]["device_metrics"]["uptime_seconds"]) + expect(local_node["last_heard"]).to eq(reference_time.to_i) + end + end + + it "returns 400 when the payload is not valid JSON" do + post "/api/telemetry", "{", auth_headers + + expect(last_response.status).to eq(400) + expect(JSON.parse(last_response.body)).to eq("error" => "invalid JSON") + 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 } } + + post "/api/telemetry", payload.to_json, auth_headers + + expect(last_response.status).to eq(400) + expect(JSON.parse(last_response.body)).to eq("error" => "too many telemetry packets") + + with_db(readonly: true) do |db| + count = db.get_first_value("SELECT COUNT(*) FROM telemetry") + expect(count).to eq(0) + end + end + end + it "returns 400 when more than 1000 messages are provided" do payload = Array.new(1001) { |i| { "packet_id" => i + 1 } } @@ -1410,4 +1500,36 @@ RSpec.describe "Potato Mesh Sinatra app" do expect(entry["payload_b64"]).to eq("AQI=") end end + + describe "GET /api/telemetry" do + it "returns stored telemetry ordered by receive time" do + post "/api/telemetry", telemetry_fixture.to_json, auth_headers + expect(last_response).to be_ok + + get "/api/telemetry?limit=2" + + expect(last_response).to be_ok + data = JSON.parse(last_response.body) + expect(data.length).to eq(2) + + latest = telemetry_fixture.max_by { |entry| entry["rx_time"] } + second_latest = telemetry_fixture.sort_by { |entry| entry["rx_time"] }[-2] + + first_entry = data.first + expect(first_entry["id"]).to eq(latest["id"]) + expect(first_entry["node_id"]).to eq(latest["node_id"]) + expect(first_entry["rx_time"]).to eq(latest["rx_time"]) + expect(first_entry["telemetry_time"]).to eq(latest["telemetry_time"]) + expect(first_entry["telemetry_time_iso"]).to eq(Time.at(latest["telemetry_time"]).utc.iso8601) + expect(first_entry).not_to have_key("device_metrics") + expect_same_value(first_entry["battery_level"], latest.dig("device_metrics", "battery_level") || latest.dig("device_metrics", "batteryLevel")) + + second_entry = data.last + expect(second_entry["id"]).to eq(second_latest["id"]) + expect(second_entry).not_to have_key("environment_metrics") + expect(second_entry["temperature"]).to be_within(1e-6).of(second_latest["environment_metrics"]["temperature"]) + expect(second_entry["relative_humidity"]).to be_within(1e-6).of(second_latest["environment_metrics"]["relativeHumidity"]) + expect(second_entry["barometric_pressure"]).to be_within(1e-6).of(second_latest["environment_metrics"]["barometricPressure"]) + end + end end