adam/potato-mesh
1
0
Fork 0
mirror of https://github.com/l5yth/potato-mesh.git synced 2026-08-07 17:33:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add POSITION_APP ingestion and API support (#160)

Browse Source 
* Add POSITION_APP ingestion and API support

* Adjust mesh receive subscriptions and priorities

* run linters
This commit is contained in:
l5y
2025-09-23 16:42:51 +02:00
committed by GitHub
parent 084c5ae158
commit 98a8203591
5 changed files with 985 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/mesh.py
+274 -3
View File
@@ -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)
data/positions.sql
+41
View File
@@ -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);
tests/test_mesh.py
+79 -3
View File
@@ -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):
web/app.rb
+399 -3
View File
@@ -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<Hash>] 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.
web/spec/app_spec.rb
+192
View File
@@ -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