mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-08-07 17:33:26 +02:00
Add telemetry API and ingestion support (#205)
* Add telemetry ingestion and API support * Flatten telemetry storage and API responses * Fix telemetry insert placeholder count * Adjust telemetry node updates * run black * run rufo
This commit is contained in:
l5y
+166
-1
@@ -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)
|
||||
|
||||
@@ -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);
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
]
|
||||
@@ -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()
|
||||
|
||||
+333
-3
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user