diff --git a/data/mesh.py b/data/mesh.py index eed2f68..e8d685e 100644 --- a/data/mesh.py +++ b/data/mesh.py @@ -27,6 +27,7 @@ import dataclasses import heapq import itertools import json, os, time, threading, signal, urllib.request, urllib.error +import math from collections.abc import Mapping from meshtastic.serial_interface import SerialInterface @@ -87,10 +88,20 @@ _POST_QUEUE = [] _POST_QUEUE_COUNTER = itertools.count() _POST_QUEUE_ACTIVE = False -_NODE_POST_PRIORITY = 0 -_MESSAGE_POST_PRIORITY = 10 +_MESSAGE_POST_PRIORITY = 0 +_POSITION_POST_PRIORITY = 10 +_NODE_POST_PRIORITY = 20 _DEFAULT_POST_PRIORITY = 50 +_RECEIVE_TOPICS = ( + "meshtastic.receive", + "meshtastic.receive.text", + "meshtastic.receive.position", + "meshtastic.receive.POSITION_APP", + "meshtastic.receive.user", + "meshtastic.receive.NODEINFO_APP", +) + def _get(obj, key, default=None): """Return a key or attribute value from ``obj``. @@ -311,6 +322,64 @@ def _first(d, *names, default=None): return default +def _coerce_int(value): + """Return ``value`` converted to ``int`` when possible.""" + + if value is None: + return None + if isinstance(value, bool): + return int(value) + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) if math.isfinite(value) else None + if isinstance(value, (str, bytes, bytearray)): + text = value.decode() if isinstance(value, (bytes, bytearray)) else value + stripped = text.strip() + if not stripped: + return None + try: + if stripped.lower().startswith("0x"): + return int(stripped, 16) + return int(stripped, 10) + except ValueError: + try: + return int(float(stripped)) + except ValueError: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _coerce_float(value): + """Return ``value`` converted to ``float`` when possible.""" + + if value is None: + return None + if isinstance(value, bool): + return float(value) + if isinstance(value, (int, float)): + result = float(value) + return result if math.isfinite(result) else None + if isinstance(value, (str, bytes, bytearray)): + text = value.decode() if isinstance(value, (bytes, bytearray)) else value + stripped = text.strip() + if not stripped: + return None + try: + result = float(stripped) + except ValueError: + return None + return result if math.isfinite(result) else None + try: + result = float(value) + except (TypeError, ValueError): + return None + return result if math.isfinite(result) else None + + def _pkt_to_dict(packet) -> dict: """Normalise a received packet into a JSON-friendly dictionary. @@ -558,6 +627,183 @@ def _nodeinfo_user_dict(node_info, decoded_user) -> dict | None: return user_dict +def store_position_packet(packet: dict, decoded: Mapping): + """Handle ``POSITION_APP`` packets and forward them to ``/api/positions``.""" + + node_ref = _first(packet, "fromId", "from_id", "from", default=None) + if node_ref is None: + node_ref = _first(decoded, "num", default=None) + node_id = _canonical_node_id(node_ref) + if node_id is None: + return + + node_num = _coerce_int(_first(decoded, "num", default=None)) + if node_num is None: + node_num = _node_num_from_id(node_id) + + pkt_id = _coerce_int(_first(packet, "id", "packet_id", "packetId", default=None)) + if pkt_id is None: + return + + rx_time = _coerce_int(_first(packet, "rxTime", "rx_time", default=time.time())) + if rx_time is None: + rx_time = int(time.time()) + + to_id = _first(packet, "toId", "to_id", "to", default=None) + to_id = to_id if to_id not in {"", None} else None + + position_section = decoded.get("position") if isinstance(decoded, Mapping) else None + if not isinstance(position_section, Mapping): + position_section = {} + + latitude = _coerce_float( + _first(position_section, "latitude", "raw.latitude", default=None) + ) + if latitude is None: + lat_i = _coerce_int( + _first( + position_section, + "latitudeI", + "latitude_i", + "raw.latitude_i", + default=None, + ) + ) + if lat_i is not None: + latitude = lat_i / 1e7 + + longitude = _coerce_float( + _first(position_section, "longitude", "raw.longitude", default=None) + ) + if longitude is None: + lon_i = _coerce_int( + _first( + position_section, + "longitudeI", + "longitude_i", + "raw.longitude_i", + default=None, + ) + ) + if lon_i is not None: + longitude = lon_i / 1e7 + + altitude = _coerce_float( + _first(position_section, "altitude", "raw.altitude", default=None) + ) + position_time = _coerce_int( + _first(position_section, "time", "raw.time", default=None) + ) + location_source = _first( + position_section, + "locationSource", + "location_source", + "raw.location_source", + default=None, + ) + location_source = ( + str(location_source).strip() if location_source not in {None, ""} else None + ) + + precision_bits = _coerce_int( + _first( + position_section, + "precisionBits", + "precision_bits", + "raw.precision_bits", + default=None, + ) + ) + sats_in_view = _coerce_int( + _first( + position_section, + "satsInView", + "sats_in_view", + "raw.sats_in_view", + default=None, + ) + ) + pdop = _coerce_float( + _first(position_section, "PDOP", "pdop", "raw.PDOP", "raw.pdop", default=None) + ) + ground_speed = _coerce_float( + _first( + position_section, + "groundSpeed", + "ground_speed", + "raw.ground_speed", + default=None, + ) + ) + ground_track = _coerce_float( + _first( + position_section, + "groundTrack", + "ground_track", + "raw.ground_track", + 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)) + bitfield = _coerce_int(_first(decoded, "bitfield", default=None)) + + payload_bytes = _extract_payload_bytes(decoded) + payload_b64 = ( + base64.b64encode(payload_bytes).decode("ascii") if payload_bytes else None + ) + + raw_section = decoded.get("raw") if isinstance(decoded, Mapping) else None + raw_payload = _node_to_dict(raw_section) if raw_section else None + if raw_payload is None and position_section: + raw_position = ( + position_section.get("raw") + if isinstance(position_section, Mapping) + else None + ) + if raw_position: + raw_payload = _node_to_dict(raw_position) + + position_payload = { + "id": pkt_id, + "node_id": node_id, + "node_num": node_num, + "num": node_num, + "from_id": node_id, + "to_id": to_id, + "rx_time": rx_time, + "rx_iso": _iso(rx_time), + "latitude": latitude, + "longitude": longitude, + "altitude": altitude, + "position_time": position_time, + "location_source": location_source, + "precision_bits": precision_bits, + "sats_in_view": sats_in_view, + "pdop": pdop, + "ground_speed": ground_speed, + "ground_track": ground_track, + "snr": snr, + "rssi": rssi, + "hop_limit": hop_limit, + "bitfield": bitfield, + "payload_b64": payload_b64, + } + if raw_payload: + position_payload["raw"] = raw_payload + + _queue_post_json( + "/api/positions", position_payload, priority=_POSITION_POST_PRIORITY + ) + + if DEBUG: + print( + f"[debug] stored position for {node_id} lat={latitude!r} lon={longitude!r} rx_time={rx_time}" + ) + + def store_nodeinfo_packet(packet: dict, decoded: Mapping): """Handle ``NODEINFO_APP`` packets and forward them to ``/api/nodes``.""" @@ -728,6 +974,10 @@ def store_packet_dict(p: dict): store_nodeinfo_packet(p, dec) return + if portnum in {"4", "POSITION_APP"}: + store_position_packet(p, dec) + return + text = _first(dec, "payload.text", "text", default=None) if not text: return # ignore non-text packets @@ -795,6 +1045,11 @@ def on_receive(packet, interface): interface: Serial interface instance (unused). """ + if isinstance(packet, dict): + if packet.get("_potatomesh_seen"): + return + packet["_potatomesh_seen"] = True + p = None try: p = _pkt_to_dict(packet) @@ -804,6 +1059,20 @@ def on_receive(packet, interface): print(f"[warn] failed to store packet: {e} | info: {info}") +def _subscribe_receive_topics() -> list[str]: + """Subscribe ``on_receive`` to relevant PubSub topics.""" + + subscribed = [] + for topic in _RECEIVE_TOPICS: + try: + pub.subscribe(on_receive, topic) + subscribed.append(topic) + except Exception as exc: # pragma: no cover - pub may raise in prod only + if DEBUG: + print(f"[debug] failed to subscribe to {topic!r}: {exc}") + return subscribed + + # --- Main --------------------------------------------------------------------- def _node_items_snapshot(nodes_obj, retries: int = 3): """Return a snapshot list of ``(node_id, node)`` pairs. @@ -854,7 +1123,9 @@ def main(): """Run the mesh synchronisation daemon.""" # Subscribe to PubSub topics (reliable in current meshtastic) - pub.subscribe(on_receive, "meshtastic.receive") + subscribed = _subscribe_receive_topics() + if DEBUG and subscribed: + print(f"[debug] subscribed to receive topics: {', '.join(subscribed)}") iface = _create_serial_interface(PORT) diff --git a/data/positions.sql b/data/positions.sql new file mode 100644 index 0000000..545568b --- /dev/null +++ b/data/positions.sql @@ -0,0 +1,41 @@ +-- 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 positions ( + id INTEGER PRIMARY KEY, + node_id TEXT, + node_num INTEGER, + rx_time INTEGER NOT NULL, + rx_iso TEXT NOT NULL, + position_time INTEGER, + to_id TEXT, + latitude REAL, + longitude REAL, + altitude REAL, + location_source TEXT, + precision_bits INTEGER, + sats_in_view INTEGER, + pdop REAL, + ground_speed REAL, + ground_track REAL, + snr REAL, + rssi INTEGER, + hop_limit INTEGER, + bitfield INTEGER, + payload_b64 TEXT, + raw_json TEXT +); + +CREATE INDEX IF NOT EXISTS idx_positions_rx_time ON positions(rx_time); +CREATE INDEX IF NOT EXISTS idx_positions_node_id ON positions(node_id); diff --git a/tests/test_mesh.py b/tests/test_mesh.py index 0e31fbc..fe87ec8 100644 --- a/tests/test_mesh.py +++ b/tests/test_mesh.py @@ -228,6 +228,82 @@ def test_store_packet_dict_posts_text_message(mesh_module, monkeypatch): assert priority == mesh._MESSAGE_POST_PRIORITY +def test_store_packet_dict_posts_position(mesh_module, monkeypatch): + mesh = mesh_module + captured = [] + monkeypatch.setattr( + mesh, + "_queue_post_json", + lambda path, payload, *, priority: captured.append((path, payload, priority)), + ) + + packet = { + "id": 200498337, + "rxTime": 1_758_624_186, + "fromId": "!b1fa2b07", + "toId": "^all", + "rxSnr": -9.5, + "rxRssi": -104, + "decoded": { + "portnum": "POSITION_APP", + "bitfield": 1, + "position": { + "latitudeI": int(52.518912 * 1e7), + "longitudeI": int(13.5512064 * 1e7), + "altitude": -16, + "time": 1_758_624_189, + "locationSource": "LOC_INTERNAL", + "precisionBits": 17, + "satsInView": 7, + "PDOP": 211, + "groundSpeed": 2, + "groundTrack": 0, + "raw": { + "latitude_i": int(52.518912 * 1e7), + "longitude_i": int(13.5512064 * 1e7), + "altitude": -16, + "time": 1_758_624_189, + }, + }, + "payload": { + "__bytes_b64__": "DQDATR8VAMATCBjw//////////8BJb150mgoAljTAXgCgAEAmAEHuAER", + }, + }, + } + + mesh.store_packet_dict(packet) + + assert captured, "Expected POST to be triggered for position packet" + path, payload, priority = captured[0] + assert path == "/api/positions" + assert priority == mesh._POSITION_POST_PRIORITY + assert payload["id"] == 200498337 + assert payload["node_id"] == "!b1fa2b07" + assert payload["node_num"] == int("b1fa2b07", 16) + assert payload["num"] == payload["node_num"] + assert payload["rx_time"] == 1_758_624_186 + assert payload["rx_iso"] == mesh._iso(1_758_624_186) + assert payload["latitude"] == pytest.approx(52.518912) + assert payload["longitude"] == pytest.approx(13.5512064) + assert payload["altitude"] == pytest.approx(-16) + assert payload["position_time"] == 1_758_624_189 + assert payload["location_source"] == "LOC_INTERNAL" + assert payload["precision_bits"] == 17 + assert payload["sats_in_view"] == 7 + assert payload["pdop"] == pytest.approx(211.0) + assert payload["ground_speed"] == pytest.approx(2.0) + assert payload["ground_track"] == pytest.approx(0.0) + assert payload["snr"] == pytest.approx(-9.5) + assert payload["rssi"] == -104 + assert payload["hop_limit"] is None + assert payload["bitfield"] == 1 + assert ( + payload["payload_b64"] + == "DQDATR8VAMATCBjw//////////8BJb150mgoAljTAXgCgAEAmAEHuAER" + ) + assert payload["raw"]["time"] == 1_758_624_189 + + def test_store_packet_dict_handles_nodeinfo_packet(mesh_module, monkeypatch): mesh = mesh_module captured = [] @@ -393,7 +469,7 @@ def test_store_packet_dict_ignores_non_text(mesh_module, monkeypatch): "toId": "!def", "decoded": { "payload": {"text": "ignored"}, - "portnum": "POSITION_APP", + "portnum": "ENVIRONMENTAL_MEASUREMENT", }, } @@ -612,7 +688,7 @@ def test_store_packet_dict_handles_invalid_channel(mesh_module, monkeypatch): assert priority == mesh._MESSAGE_POST_PRIORITY -def test_post_queue_prioritises_nodes(mesh_module, monkeypatch): +def test_post_queue_prioritises_messages(mesh_module, monkeypatch): mesh = mesh_module mesh._clear_post_queue() calls = [] @@ -629,7 +705,7 @@ def test_post_queue_prioritises_nodes(mesh_module, monkeypatch): mesh._drain_post_queue() - assert [path for path, _ in calls] == ["/api/nodes", "/api/messages"] + assert [path for path, _ in calls] == ["/api/messages", "/api/nodes"] def test_store_packet_dict_requires_id(mesh_module, monkeypatch): diff --git a/web/app.rb b/web/app.rb index b2ca1ef..6707a91 100644 --- a/web/app.rb +++ b/web/app.rb @@ -109,6 +109,60 @@ def sanitized_matrix_room value.empty? ? nil : value end +def string_or_nil(value) + return nil if value.nil? + + str = value.is_a?(String) ? value : value.to_s + trimmed = str.strip + trimmed.empty? ? nil : trimmed +end + +def coerce_integer(value) + case value + when Integer + value + when Float + value.finite? ? value.to_i : nil + when Numeric + value.to_i + when String + trimmed = value.strip + return nil if trimmed.empty? + return trimmed.to_i(16) if trimmed.match?(/\A0[xX][0-9A-Fa-f]+\z/) + return trimmed.to_i(10) if trimmed.match?(/\A-?\d+\z/) + begin + float_val = Float(trimmed) + float_val.finite? ? float_val.to_i : nil + rescue ArgumentError + nil + end + else + nil + end +end + +def coerce_float(value) + case value + when Float + value.finite? ? value : nil + when Integer + value.to_f + when Numeric + value.to_f + when String + trimmed = value.strip + return nil if trimmed.empty? + begin + float_val = Float(trimmed) + float_val.finite? ? float_val : nil + rescue ArgumentError + nil + end + else + nil + end +end + def sanitized_max_distance_km return nil unless defined?(MAX_NODE_DISTANCE_KM) @@ -210,8 +264,9 @@ end def db_schema_present? return false unless File.exist?(DB_PATH) db = open_database(readonly: true) - tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages')").flatten - tables.include?("nodes") && tables.include?("messages") + required = %w[nodes messages positions] + tables = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name IN ('nodes','messages','positions')").flatten + (required - tables).empty? rescue SQLite3::Exception false ensure @@ -224,7 +279,7 @@ end def init_db FileUtils.mkdir_p(File.dirname(DB_PATH)) db = open_database - %w[nodes messages].each do |schema| + %w[nodes messages positions].each do |schema| sql_file = File.expand_path("../data/#{schema}.sql", __dir__) db.execute_batch(File.read(sql_file)) end @@ -343,6 +398,40 @@ ensure db&.close end +# Retrieve recorded position packets ordered by receive time. +# +# @param limit [Integer] maximum number of rows returned. +# @return [Array] collection of position rows formatted for the API. +def query_positions(limit) + db = open_database(readonly: true) + db.results_as_hash = true + rows = db.execute <<~SQL, [limit] + SELECT id, node_id, node_num, rx_time, rx_iso, position_time, + to_id, latitude, longitude, altitude, location_source, + precision_bits, sats_in_view, pdop, ground_speed, + ground_track, snr, rssi, hop_limit, bitfield, + payload_b64, raw_json + FROM positions + ORDER BY rx_time DESC + LIMIT ? + SQL + rows.each do |r| + pt = r["position_time"] + if pt + begin + r["position_time"] = Integer(pt, 10) + rescue ArgumentError, TypeError + r["position_time"] = coerce_integer(pt) + end + end + pt_val = r["position_time"] + r["position_time_iso"] = Time.at(pt_val).utc.iso8601 if pt_val + end + rows +ensure + db&.close +end + # GET /api/messages # # Returns a JSON array of stored text messages including node metadata. @@ -352,6 +441,15 @@ get "/api/messages" do query_messages(limit).to_json end +# GET /api/positions +# +# Returns a JSON array of recorded position packets. +get "/api/positions" do + content_type :json + limit = [params["limit"]&.to_i || 200, 1000].min + query_positions(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 @@ -518,6 +616,282 @@ def prefer_canonical_sender?(message) message.is_a?(Hash) && message.key?("packet_id") && !message.key?("id") end +# Update or create a node entry using information from a position payload. +# +# @param db [SQLite3::Database] open database handle. +# @param node_id [String, nil] canonical node identifier when available. +# @param node_num [Integer, nil] numeric node reference if known. +# @param rx_time [Integer] time the packet was received by the gateway. +# @param position_time [Integer, nil] timestamp reported by the device. +# @param location_source [String, nil] location source flag from the packet. +# @param latitude [Float, nil] reported latitude. +# @param longitude [Float, nil] reported longitude. +# @param altitude [Float, nil] reported altitude. +# @param snr [Float, nil] link SNR for the packet. +def update_node_from_position(db, node_id, node_num, rx_time, position_time, location_source, latitude, longitude, altitude, snr) + 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 + rx = coerce_integer(rx_time) || now + rx = now if rx && rx > now + pos_time = coerce_integer(position_time) + pos_time = nil if pos_time && pos_time > now + last_heard = [rx, pos_time].compact.max || rx + last_heard = now if last_heard && last_heard > now + + loc = string_or_nil(location_source) + lat = coerce_float(latitude) + lon = coerce_float(longitude) + alt = coerce_float(altitude) + snr_val = coerce_float(snr) + + row = [ + id, + num, + last_heard, + last_heard, + pos_time, + loc, + lat, + lon, + alt, + snr_val, + ] + with_busy_retry do + db.execute <<~SQL, row + INSERT INTO nodes(node_id,num,last_heard,first_heard,position_time,location_source,latitude,longitude,altitude,snr) + VALUES (?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(node_id) DO UPDATE SET + num=COALESCE(excluded.num,nodes.num), + snr=COALESCE(excluded.snr,nodes.snr), + last_heard=MAX(COALESCE(nodes.last_heard,0),COALESCE(excluded.last_heard,0)), + position_time=CASE + WHEN COALESCE(excluded.position_time,0) >= COALESCE(nodes.position_time,0) + THEN excluded.position_time + ELSE nodes.position_time + END, + location_source=CASE + WHEN COALESCE(excluded.position_time,0) >= COALESCE(nodes.position_time,0) + AND excluded.location_source IS NOT NULL + THEN excluded.location_source + ELSE nodes.location_source + END, + latitude=CASE + WHEN COALESCE(excluded.position_time,0) >= COALESCE(nodes.position_time,0) + AND excluded.latitude IS NOT NULL + THEN excluded.latitude + ELSE nodes.latitude + END, + longitude=CASE + WHEN COALESCE(excluded.position_time,0) >= COALESCE(nodes.position_time,0) + AND excluded.longitude IS NOT NULL + THEN excluded.longitude + ELSE nodes.longitude + END, + altitude=CASE + WHEN COALESCE(excluded.position_time,0) >= COALESCE(nodes.position_time,0) + AND excluded.altitude IS NOT NULL + THEN excluded.altitude + ELSE nodes.altitude + END + SQL + end +end + +# Insert a position packet into the history table and refresh node metadata. +# +# @param db [SQLite3::Database] open database handle. +# @param payload [Hash] position payload provided by the data daemon. +def insert_position(db, payload) + pos_id = coerce_integer(payload["id"] || payload["packet_id"]) + return unless pos_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"]) + node_id ||= format("!%08x", raw_node_num & 0xFFFFFFFF) if node_id.nil? && raw_node_num + + payload_for_num = payload.is_a?(Hash) ? 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 + + to_id = string_or_nil(payload["to_id"] || payload["to"]) + + position_section = payload["position"].is_a?(Hash) ? payload["position"] : {} + + lat = coerce_float(payload["latitude"]) || coerce_float(position_section["latitude"]) + lon = coerce_float(payload["longitude"]) || coerce_float(position_section["longitude"]) + alt = coerce_float(payload["altitude"]) || coerce_float(position_section["altitude"]) + + lat ||= begin + lat_i = coerce_integer(position_section["latitudeI"] || position_section["latitude_i"] || position_section.dig("raw", "latitude_i")) + lat_i ? lat_i / 1e7 : nil + end + lon ||= begin + lon_i = coerce_integer(position_section["longitudeI"] || position_section["longitude_i"] || position_section.dig("raw", "longitude_i")) + lon_i ? lon_i / 1e7 : nil + end + alt ||= coerce_float(position_section.dig("raw", "altitude")) + + position_time = coerce_integer( + payload["position_time"] || + position_section["time"] || + position_section.dig("raw", "time"), + ) + + location_source = string_or_nil( + payload["location_source"] || + payload["locationSource"] || + position_section["location_source"] || + position_section["locationSource"] || + position_section.dig("raw", "location_source"), + ) + + precision_bits = coerce_integer( + payload["precision_bits"] || + payload["precisionBits"] || + position_section["precision_bits"] || + position_section["precisionBits"] || + position_section.dig("raw", "precision_bits"), + ) + + sats_in_view = coerce_integer( + payload["sats_in_view"] || + payload["satsInView"] || + position_section["sats_in_view"] || + position_section["satsInView"] || + position_section.dig("raw", "sats_in_view"), + ) + + pdop = coerce_float( + payload["pdop"] || + payload["PDOP"] || + position_section["pdop"] || + position_section["PDOP"] || + position_section.dig("raw", "PDOP") || + position_section.dig("raw", "pdop"), + ) + + ground_speed = coerce_float( + payload["ground_speed"] || + payload["groundSpeed"] || + position_section["ground_speed"] || + position_section["groundSpeed"] || + position_section.dig("raw", "ground_speed"), + ) + + ground_track = coerce_float( + payload["ground_track"] || + payload["groundTrack"] || + position_section["ground_track"] || + position_section["groundTrack"] || + position_section.dig("raw", "ground_track"), + ) + + snr = coerce_float(payload["snr"] || payload["rx_snr"] || payload["rxSnr"]) + rssi = coerce_integer(payload["rssi"] || payload["rx_rssi"] || payload["rxRssi"]) + hop_limit = coerce_integer(payload["hop_limit"] || payload["hopLimit"]) + bitfield = coerce_integer(payload["bitfield"]) + + payload_b64 = string_or_nil(payload["payload_b64"] || payload["payload"]) + payload_b64 ||= string_or_nil(position_section.dig("payload", "__bytes_b64__")) + + raw_value = payload["raw_json"] || payload["raw"] || position_section["raw"] + raw_json = if raw_value.is_a?(String) + raw_value + elsif raw_value + begin + JSON.dump(raw_value) + rescue StandardError + raw_value.to_s + end + end + + row = [ + pos_id, + node_id, + node_num, + rx_time, + rx_iso, + position_time, + to_id, + lat, + lon, + alt, + location_source, + precision_bits, + sats_in_view, + pdop, + ground_speed, + ground_track, + snr, + rssi, + hop_limit, + bitfield, + payload_b64, + raw_json, + ] + + with_busy_retry do + db.execute <<~SQL, row + INSERT INTO positions(id,node_id,node_num,rx_time,rx_iso,position_time,to_id,latitude,longitude,altitude,location_source, + precision_bits,sats_in_view,pdop,ground_speed,ground_track,snr,rssi,hop_limit,bitfield,payload_b64,raw_json) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(id) DO UPDATE SET + node_id=COALESCE(excluded.node_id,positions.node_id), + node_num=COALESCE(excluded.node_num,positions.node_num), + rx_time=excluded.rx_time, + rx_iso=excluded.rx_iso, + position_time=COALESCE(excluded.position_time,positions.position_time), + to_id=COALESCE(excluded.to_id,positions.to_id), + latitude=COALESCE(excluded.latitude,positions.latitude), + longitude=COALESCE(excluded.longitude,positions.longitude), + altitude=COALESCE(excluded.altitude,positions.altitude), + location_source=COALESCE(excluded.location_source,positions.location_source), + precision_bits=COALESCE(excluded.precision_bits,positions.precision_bits), + sats_in_view=COALESCE(excluded.sats_in_view,positions.sats_in_view), + pdop=COALESCE(excluded.pdop,positions.pdop), + ground_speed=COALESCE(excluded.ground_speed,positions.ground_speed), + ground_track=COALESCE(excluded.ground_track,positions.ground_track), + snr=COALESCE(excluded.snr,positions.snr), + rssi=COALESCE(excluded.rssi,positions.rssi), + hop_limit=COALESCE(excluded.hop_limit,positions.hop_limit), + bitfield=COALESCE(excluded.bitfield,positions.bitfield), + payload_b64=COALESCE(excluded.payload_b64,positions.payload_b64), + raw_json=COALESCE(excluded.raw_json,positions.raw_json) + SQL + end + + update_node_from_position( + db, + node_id, + node_num, + rx_time, + position_time, + location_source, + lat, + lon, + alt, + snr, + ) +end + # Insert a text message if it does not already exist. # # @param db [SQLite3::Database] open database handle. @@ -643,6 +1017,28 @@ ensure db&.close end +# POST /api/positions +# +# Accepts an array or object describing position packets and stores each entry. +post "/api/positions" do + require_token! + content_type :json + begin + data = JSON.parse(read_json_body) + rescue JSON::ParserError + halt 400, { error: "invalid JSON" }.to_json + end + positions = data.is_a?(Array) ? data : [data] + halt 400, { error: "too many positions" }.to_json if positions.size > 1000 + db = open_database + positions.each do |pos| + insert_position(db, pos) + end + { status: "ok" }.to_json +ensure + db&.close +end + # GET / # # Renders the main site with configuration-driven defaults for the template. diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index c451a40..d95dbaf 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -38,6 +38,7 @@ RSpec.describe "Potato Mesh Sinatra app" do with_db do |db| db.execute("DELETE FROM messages") db.execute("DELETE FROM nodes") + db.execute("DELETE FROM positions") end end @@ -473,6 +474,160 @@ RSpec.describe "Potato Mesh Sinatra app" do end end + describe "POST /api/positions" do + it "stores position packets and updates node metadata" do + node_id = "!specpos01" + node_num = 0x1234_5678 + initial_last_heard = reference_time.to_i - 600 + node_payload = { + node_id => { + "num" => node_num, + "user" => { "shortName" => "SpecPos" }, + "lastHeard" => initial_last_heard, + "position" => { + "time" => initial_last_heard - 60, + "latitude" => 52.0, + "longitude" => 13.0, + }, + }, + } + + post "/api/nodes", node_payload.to_json, auth_headers + expect(last_response).to be_ok + + rx_time = reference_time.to_i - 120 + position_time = rx_time - 30 + raw_payload = { "time" => position_time, "latitude_i" => (52.5 * 1e7).to_i } + position_payload = { + "id" => 9_001, + "node_id" => node_id, + "node_num" => node_num, + "rx_time" => rx_time, + "rx_iso" => Time.at(rx_time).utc.iso8601, + "to_id" => "^all", + "latitude" => 52.5, + "longitude" => 13.4, + "altitude" => 42.0, + "position_time" => position_time, + "location_source" => "LOC_INTERNAL", + "precision_bits" => 15, + "sats_in_view" => 6, + "pdop" => 2.5, + "ground_speed" => 3.2, + "ground_track" => 180.0, + "snr" => -8.5, + "rssi" => -90, + "hop_limit" => 3, + "bitfield" => 1, + "payload_b64" => "AQI=", + "raw" => raw_payload, + } + + post "/api/positions", position_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 + row = db.get_first_row("SELECT * FROM positions WHERE id = ?", [9_001]) + expect(row["node_id"]).to eq(node_id) + expect(row["node_num"]).to eq(node_num) + expect(row["rx_time"]).to eq(rx_time) + expect(row["rx_iso"]).to eq(Time.at(rx_time).utc.iso8601) + expect(row["position_time"]).to eq(position_time) + expect_same_value(row["latitude"], 52.5) + expect_same_value(row["longitude"], 13.4) + expect_same_value(row["altitude"], 42.0) + expect(row["location_source"]).to eq("LOC_INTERNAL") + expect(row["precision_bits"]).to eq(15) + expect(row["sats_in_view"]).to eq(6) + expect_same_value(row["pdop"], 2.5) + expect_same_value(row["ground_speed"], 3.2) + expect_same_value(row["ground_track"], 180.0) + expect_same_value(row["snr"], -8.5) + expect(row["rssi"]).to eq(-90) + expect(row["hop_limit"]).to eq(3) + expect(row["bitfield"]).to eq(1) + expect(row["payload_b64"]).to eq("AQI=") + expect(JSON.parse(row["raw_json"])).to eq(raw_payload.transform_keys(&:to_s)) + end + + with_db(readonly: true) do |db| + db.results_as_hash = true + node_row = db.get_first_row( + "SELECT last_heard, position_time, latitude, longitude, altitude, location_source, snr FROM nodes WHERE node_id = ?", + [node_id], + ) + expect(node_row["last_heard"]).to eq(rx_time) + expect(node_row["position_time"]).to eq(position_time) + expect_same_value(node_row["latitude"], 52.5) + expect_same_value(node_row["longitude"], 13.4) + expect_same_value(node_row["altitude"], 42.0) + expect(node_row["location_source"]).to eq("LOC_INTERNAL") + expect_same_value(node_row["snr"], -8.5) + end + end + + it "creates node records when none exist" do + node_id = "!specnew01" + node_num = 0xfeed_cafe + rx_time = reference_time.to_i - 60 + position_time = rx_time - 10 + payload = { + "id" => 9_002, + "node_id" => node_id, + "node_num" => node_num, + "rx_time" => rx_time, + "rx_iso" => Time.at(rx_time).utc.iso8601, + "latitude" => 52.1, + "longitude" => 13.1, + "altitude" => 33.0, + "position_time" => position_time, + "location_source" => "LOC_EXTERNAL", + } + + post "/api/positions", payload.to_json, auth_headers + + expect(last_response).to be_ok + + with_db(readonly: true) do |db| + db.results_as_hash = true + node_row = db.get_first_row("SELECT * FROM nodes WHERE node_id = ?", [node_id]) + expect(node_row).not_to be_nil + expect(node_row["num"]).to eq(node_num) + expect(node_row["last_heard"]).to eq(rx_time) + expect(node_row["first_heard"]).to eq(rx_time) + expect(node_row["position_time"]).to eq(position_time) + expect_same_value(node_row["latitude"], 52.1) + expect_same_value(node_row["longitude"], 13.1) + expect_same_value(node_row["altitude"], 33.0) + expect(node_row["location_source"]).to eq("LOC_EXTERNAL") + end + end + + it "returns 400 when the payload is not valid JSON" do + post "/api/positions", "{", 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 positions are provided" do + payload = Array.new(1001) { |i| { "id" => i + 1, "rx_time" => reference_time.to_i - i } } + + post "/api/positions", payload.to_json, auth_headers + + expect(last_response.status).to eq(400) + expect(JSON.parse(last_response.body)).to eq("error" => "too many positions") + + with_db(readonly: true) do |db| + count = db.get_first_value("SELECT COUNT(*) FROM positions") + 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 } } @@ -823,4 +978,41 @@ RSpec.describe "Potato Mesh Sinatra app" do end end end + + describe "GET /api/positions" do + it "returns stored positions ordered by receive time" do + node_id = "!specfetch" + rx_times = [reference_time.to_i - 50, reference_time.to_i - 10] + rx_times.each_with_index do |rx_time, idx| + payload = { + "id" => 20_000 + idx, + "node_id" => node_id, + "rx_time" => rx_time, + "rx_iso" => Time.at(rx_time).utc.iso8601, + "position_time" => rx_time - 5, + "latitude" => 52.0 + idx, + "longitude" => 13.0 + idx, + "payload_b64" => "AQI=", + } + post "/api/positions", payload.to_json, auth_headers + expect(last_response).to be_ok + end + + get "/api/positions?limit=1" + + expect(last_response).to be_ok + data = JSON.parse(last_response.body) + expect(data.length).to eq(1) + entry = data.first + expect(entry["id"]).to eq(20_001) + expect(entry["node_id"]).to eq(node_id) + expect(entry["rx_time"]).to eq(rx_times.last) + expect(entry["rx_iso"]).to eq(Time.at(rx_times.last).utc.iso8601) + expect(entry["position_time"]).to eq(rx_times.last - 5) + expect(entry["position_time_iso"]).to eq(Time.at(rx_times.last - 5).utc.iso8601) + expect(entry["latitude"]).to eq(53.0) + expect(entry["longitude"]).to eq(14.0) + expect(entry["payload_b64"]).to eq("AQI=") + end + end end