From db670fbb7c734f6788a4a55acc6f706065341b92 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Wed, 19 Nov 2025 11:31:42 +0100 Subject: [PATCH] aggregate telemetry over the last 7 days (#470) * aggregate telemetry over the last 7 days * cover missing unit test vectors --- web/lib/potato_mesh/application/queries.rb | 106 ++++++++++++++ web/lib/potato_mesh/application/routes/api.rb | 31 +++++ .../js/app/__tests__/charts-page.test.js | 50 +++++-- web/public/assets/js/app/charts-page.js | 130 +++++++++++++++--- web/spec/app_spec.rb | 48 +++++++ 5 files changed, 338 insertions(+), 27 deletions(-) diff --git a/web/lib/potato_mesh/application/queries.rb b/web/lib/potato_mesh/application/queries.rb index 35ae042..322d149 100644 --- a/web/lib/potato_mesh/application/queries.rb +++ b/web/lib/potato_mesh/application/queries.rb @@ -18,6 +18,36 @@ module PotatoMesh module App module Queries MAX_QUERY_LIMIT = 1000 + DEFAULT_TELEMETRY_WINDOW_SECONDS = 86_400 + DEFAULT_TELEMETRY_BUCKET_SECONDS = 300 + TELEMETRY_AGGREGATE_COLUMNS = + %w[ + battery_level + voltage + channel_utilization + air_util_tx + temperature + relative_humidity + barometric_pressure + gas_resistance + current + iaq + distance + lux + white_lux + ir_lux + uv_lux + wind_direction + wind_speed + wind_gust + wind_lull + weight + radiation + rainfall_1h + rainfall_24h + soil_moisture + soil_temperature + ].freeze # Remove nil or empty values from an API response hash to reduce payload size # while preserving legitimate zero-valued measurements. @@ -465,6 +495,82 @@ module PotatoMesh db&.close end + def query_telemetry_buckets(window_seconds:, bucket_seconds:) + window = coerce_integer(window_seconds) || DEFAULT_TELEMETRY_WINDOW_SECONDS + window = DEFAULT_TELEMETRY_WINDOW_SECONDS if window <= 0 + bucket = coerce_integer(bucket_seconds) || DEFAULT_TELEMETRY_BUCKET_SECONDS + bucket = DEFAULT_TELEMETRY_BUCKET_SECONDS if bucket <= 0 + + db = open_database(readonly: true) + db.results_as_hash = true + now = Time.now.to_i + min_timestamp = now - window + bucket_expression = "((COALESCE(rx_time, telemetry_time) / ?) * ?)" + select_clauses = [ + "#{bucket_expression} AS bucket_start", + "COUNT(*) AS sample_count", + "MIN(COALESCE(rx_time, telemetry_time)) AS first_timestamp", + "MAX(COALESCE(rx_time, telemetry_time)) AS last_timestamp", + ] + + TELEMETRY_AGGREGATE_COLUMNS.each do |column| + select_clauses << "AVG(#{column}) AS #{column}_avg" + select_clauses << "MIN(#{column}) AS #{column}_min" + select_clauses << "MAX(#{column}) AS #{column}_max" + end + + sql = <<~SQL + SELECT + #{select_clauses.join(",\n ")} + FROM telemetry + WHERE COALESCE(rx_time, telemetry_time) IS NOT NULL + AND COALESCE(rx_time, telemetry_time, 0) >= ? + GROUP BY bucket_start + ORDER BY bucket_start ASC + LIMIT ? + SQL + params = [bucket, bucket, min_timestamp, MAX_QUERY_LIMIT] + rows = db.execute(sql, params) + rows.map do |row| + bucket_start = coerce_integer(row["bucket_start"]) + bucket_end = bucket_start ? bucket_start + bucket : nil + first_timestamp = coerce_integer(row["first_timestamp"]) + last_timestamp = coerce_integer(row["last_timestamp"]) + + aggregates = {} + TELEMETRY_AGGREGATE_COLUMNS.each do |column| + avg = coerce_float(row["#{column}_avg"]) + min_value = coerce_float(row["#{column}_min"]) + max_value = coerce_float(row["#{column}_max"]) + + metrics = {} + metrics["avg"] = avg unless avg.nil? + metrics["min"] = min_value unless min_value.nil? + metrics["max"] = max_value unless max_value.nil? + aggregates[column] = metrics unless metrics.empty? + end + + bucket_response = { + "bucket_start" => bucket_start, + "bucket_start_iso" => bucket_start ? Time.at(bucket_start).utc.iso8601 : nil, + "bucket_end" => bucket_end, + "bucket_end_iso" => bucket_end ? Time.at(bucket_end).utc.iso8601 : nil, + "bucket_seconds" => bucket, + "sample_count" => coerce_integer(row["sample_count"]), + "first_timestamp" => first_timestamp, + "first_timestamp_iso" => first_timestamp ? Time.at(first_timestamp).utc.iso8601 : nil, + "last_timestamp" => last_timestamp, + "last_timestamp_iso" => last_timestamp ? Time.at(last_timestamp).utc.iso8601 : nil, + "aggregates" => aggregates, + } + bucket_response["timestamp"] = bucket_start if bucket_start + bucket_response["timestamp_iso"] = bucket_response["bucket_start_iso"] if bucket_response["bucket_start_iso"] + compact_api_row(bucket_response) + end + ensure + db&.close + end + def query_traces(limit, node_ref: nil) limit = coerce_query_limit(limit) db = open_database(readonly: true) diff --git a/web/lib/potato_mesh/application/routes/api.rb b/web/lib/potato_mesh/application/routes/api.rb index 0e83db2..d54e27e 100644 --- a/web/lib/potato_mesh/application/routes/api.rb +++ b/web/lib/potato_mesh/application/routes/api.rb @@ -127,6 +127,37 @@ module PotatoMesh query_telemetry(limit).to_json end + app.get "/api/telemetry/aggregated" do + content_type :json + default_window = PotatoMesh::App::Queries::DEFAULT_TELEMETRY_WINDOW_SECONDS + default_bucket = PotatoMesh::App::Queries::DEFAULT_TELEMETRY_BUCKET_SECONDS + + window_seconds = if params.key?("windowSeconds") + coerce_integer(params["windowSeconds"]) + else + default_window + end + bucket_seconds = if params.key?("bucketSeconds") + coerce_integer(params["bucketSeconds"]) + else + default_bucket + end + + if window_seconds.nil? || window_seconds <= 0 + halt 400, { error: "windowSeconds must be positive" }.to_json + end + if bucket_seconds.nil? || bucket_seconds <= 0 + halt 400, { error: "bucketSeconds must be positive" }.to_json + end + + bucket_count = (window_seconds.to_f / bucket_seconds).ceil + if bucket_count > PotatoMesh::App::Queries::MAX_QUERY_LIMIT + halt 400, { error: "bucketSeconds too small for requested window" }.to_json + end + + query_telemetry_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds).to_json + end + app.get "/api/telemetry/:id" do content_type :json node_ref = string_or_nil(params["id"]) diff --git a/web/public/assets/js/app/__tests__/charts-page.test.js b/web/public/assets/js/app/__tests__/charts-page.test.js index 3b9ecbc..d5e676c 100644 --- a/web/public/assets/js/app/__tests__/charts-page.test.js +++ b/web/public/assets/js/app/__tests__/charts-page.test.js @@ -33,23 +33,35 @@ function createResponse(status, body) { }; } -test('fetchAggregatedTelemetry requests the latest 1000 telemetry entries', async () => { +test('fetchAggregatedTelemetry requests aggregated telemetry buckets and normalizes stats', async () => { const requests = []; const fetchImpl = async url => { requests.push(url); - return createResponse(200, [{ rx_time: 1_700_000_000, node_id: '!demo' }]); + return createResponse(200, [ + { + bucket_start: 1_700_000_000, + bucket_seconds: 300, + sample_count: 4, + aggregates: { + battery_level: { avg: 85.2, min: 80, max: 90 }, + temperature: { avg: 22.5 }, + }, + }, + ]); }; - const snapshots = await fetchAggregatedTelemetry({ fetchImpl }); + const snapshots = await fetchAggregatedTelemetry({ fetchImpl, windowMs: 604_800_000, bucketSeconds: 900 }); assert.equal(requests.length, 1); - assert.equal(requests[0], '/api/telemetry?limit=1000'); + assert.equal(requests[0], '/api/telemetry/aggregated?windowSeconds=604800&bucketSeconds=900'); assert.equal(Array.isArray(snapshots), true); - assert.equal(snapshots[0].node_id, '!demo'); + assert.equal(snapshots[0].battery_level, 85.2); + assert.equal(snapshots[0].battery_level_max, 90); + assert.equal(snapshots[0].temperature, 22.5); }); test('fetchAggregatedTelemetry validates fetch availability and response codes', async () => { await assert.rejects(() => fetchAggregatedTelemetry({ fetchImpl: null }), /fetch implementation/i); const fetchImpl = async () => createResponse(503, []); - await assert.rejects(() => fetchAggregatedTelemetry({ fetchImpl }), /Failed to fetch telemetry/); + await assert.rejects(() => fetchAggregatedTelemetry({ fetchImpl }), /Failed to fetch aggregated telemetry/); }); test('initializeChartsPage renders the telemetry charts when snapshots are available', async () => { @@ -59,7 +71,14 @@ test('initializeChartsPage renders the telemetry charts when snapshots are avail return id === 'chartsPage' ? container : null; }, }; - const fetchImpl = async () => createResponse(200, [{ rx_time: 1_700_000_000, temperature: 22.5 }]); + const fetchImpl = async () => createResponse(200, [ + { + bucket_start: 1_700_000_000, + bucket_seconds: 300, + sample_count: 3, + aggregates: { temperature: { avg: 22.5 } }, + }, + ]); let receivedOptions = null; const renderCharts = (node, options) => { receivedOptions = options; @@ -69,8 +88,10 @@ test('initializeChartsPage renders the telemetry charts when snapshots are avail assert.equal(result, true); assert.equal(container.innerHTML.includes('node-detail__charts'), true); assert.ok(receivedOptions); - assert.equal(receivedOptions.chartOptions.windowMs, 86_400_000); + assert.equal(receivedOptions.chartOptions.windowMs, 604_800_000); assert.equal(typeof receivedOptions.chartOptions.lineReducer, 'function'); + assert.equal(typeof receivedOptions.chartOptions.xAxisTickBuilder, 'function'); + assert.equal(typeof receivedOptions.chartOptions.xAxisTickFormatter, 'function'); const average = receivedOptions.chartOptions.lineReducer( [ { timestamp: 0, value: 0 }, @@ -79,6 +100,12 @@ test('initializeChartsPage renders the telemetry charts when snapshots are avail ], ); assert.equal(Array.isArray(average), true); + const nowMs = Date.UTC(2025, 8, 16); // September 16, 2025 + const ticks = receivedOptions.chartOptions.xAxisTickBuilder(nowMs, 604_800_000); + assert.equal(Array.isArray(ticks), true); + assert.equal(new Date(ticks[0]).getHours(), 0); + const label = receivedOptions.chartOptions.xAxisTickFormatter(ticks[0]); + assert.equal(/^\d{2}$/.test(label), true); }); test('initializeChartsPage shows an error message when fetching fails', async () => { @@ -122,7 +149,12 @@ test('initializeChartsPage shows a status when rendering produces no markup', as return container; }, }; - const fetchImpl = async () => createResponse(200, [{ rx_time: 1_700_000_000 }]); + const fetchImpl = async () => createResponse(200, [ + { + bucket_start: 1_700_000_000, + aggregates: { voltage: { avg: 3.9 } }, + }, + ]); const renderCharts = () => ''; const result = await initializeChartsPage({ document: documentStub, fetchImpl, renderCharts }); assert.equal(result, true); diff --git a/web/public/assets/js/app/charts-page.js b/web/public/assets/js/app/charts-page.js index 76f6d67..954016a 100644 --- a/web/public/assets/js/app/charts-page.js +++ b/web/public/assets/js/app/charts-page.js @@ -16,9 +16,10 @@ import { renderTelemetryCharts } from './node-page.js'; -const TELEMETRY_AGGREGATE_LIMIT = 1000; +const TELEMETRY_BUCKET_SECONDS = 60 * 60; const HOUR_MS = 60 * 60 * 1000; const DAY_MS = 24 * HOUR_MS; +const CHART_WINDOW_MS = 7 * DAY_MS; function escapeHtml(value) { return String(value ?? '') @@ -40,22 +41,98 @@ function padTwo(value) { return num < 10 ? `0${Math.trunc(num)}` : String(Math.trunc(num)); } -function buildHourlyTickList(nowMs, windowMs = DAY_MS) { +function buildMidnightTickList(nowMs, windowMs = CHART_WINDOW_MS) { const ticks = []; - const safeWindow = Number.isFinite(windowMs) && windowMs > 0 ? windowMs : DAY_MS; + const safeWindow = Number.isFinite(windowMs) && windowMs > 0 ? windowMs : CHART_WINDOW_MS; const domainStart = nowMs - safeWindow; const cursor = new Date(nowMs); - cursor.setMinutes(0, 0, 0); - for (let ts = cursor.getTime(); ts >= domainStart; ts -= HOUR_MS) { + cursor.setHours(0, 0, 0, 0); + for (let ts = cursor.getTime(); ts >= domainStart; ts -= DAY_MS) { ticks.push(ts); } return ticks.reverse(); } -function formatHourLabel(timestampMs) { +function formatDayOfMonthLabel(timestampMs) { const date = new Date(timestampMs); if (Number.isNaN(date.getTime())) return ''; - return padTwo(date.getHours()); + return padTwo(date.getDate()); +} + +function normalizeAggregatedSnapshot(bucket) { + if (!bucket || typeof bucket !== 'object') { + return null; + } + const timestamp = + Number.isFinite(bucket.timestamp) ? bucket.timestamp + : Number.isFinite(bucket.bucket_start) ? bucket.bucket_start + : Number.isFinite(bucket.bucketStart) ? bucket.bucketStart + : null; + if (!Number.isFinite(timestamp)) { + return null; + } + const bucketSecondsCandidate = + Number.isFinite(bucket.bucket_seconds) ? bucket.bucket_seconds + : Number.isFinite(bucket.bucketSeconds) ? bucket.bucketSeconds + : TELEMETRY_BUCKET_SECONDS; + const bucketSeconds = bucketSecondsCandidate > 0 ? bucketSecondsCandidate : TELEMETRY_BUCKET_SECONDS; + const timestampIso = + typeof bucket.timestamp_iso === 'string' ? bucket.timestamp_iso + : typeof bucket.timestampIso === 'string' ? bucket.timestampIso + : typeof bucket.bucket_start_iso === 'string' ? bucket.bucket_start_iso + : typeof bucket.bucketStartIso === 'string' ? bucket.bucketStartIso + : null; + const snapshot = { + rx_time: Number.isFinite(bucket.rx_time) ? bucket.rx_time : timestamp, + telemetry_time: Number.isFinite(bucket.telemetry_time) ? bucket.telemetry_time : timestamp, + rx_iso: typeof bucket.rx_iso === 'string' ? bucket.rx_iso : timestampIso, + telemetry_time_iso: + typeof bucket.telemetry_time_iso === 'string' + ? bucket.telemetry_time_iso + : timestampIso, + timestamp, + timestampIso, + bucket_seconds: bucketSeconds, + bucket_start: Number.isFinite(bucket.bucket_start) ? bucket.bucket_start : timestamp, + bucket_start_iso: + typeof bucket.bucket_start_iso === 'string' + ? bucket.bucket_start_iso + : timestampIso, + bucket_end: Number.isFinite(bucket.bucket_end) ? bucket.bucket_end : timestamp + bucketSeconds, + bucket_end_iso: typeof bucket.bucket_end_iso === 'string' ? bucket.bucket_end_iso : null, + sample_count: + Number.isFinite(bucket.sample_count) + ? bucket.sample_count + : Number.isFinite(bucket.sampleCount) + ? bucket.sampleCount + : null, + }; + const aggregates = bucket.aggregates && typeof bucket.aggregates === 'object' ? bucket.aggregates : null; + if (aggregates) { + snapshot.aggregates = aggregates; + for (const [field, stats] of Object.entries(aggregates)) { + if (!stats || typeof stats !== 'object') { + continue; + } + const avg = Number.isFinite(stats.avg) ? stats.avg : null; + const min = Number.isFinite(stats.min) ? stats.min : null; + const max = Number.isFinite(stats.max) ? stats.max : null; + if (avg != null) { + snapshot[field] = avg; + snapshot[`${field}_avg`] = avg; + } + if (min != null) { + snapshot[`${field}_min`] = min; + } + if (max != null) { + snapshot[`${field}_max`] = max; + } + } + } + if (!snapshot.bucket_end_iso && Number.isFinite(snapshot.bucket_end)) { + snapshot.bucket_end_iso = new Date(snapshot.bucket_end * 1000).toISOString(); + } + return snapshot; } export function buildMovingAverageSeries(points, windowMs = HOUR_MS) { @@ -86,18 +163,34 @@ export function buildMovingAverageSeries(points, windowMs = HOUR_MS) { return averages; } -export async function fetchAggregatedTelemetry({ fetchImpl = globalThis.fetch, limit = TELEMETRY_AGGREGATE_LIMIT } = {}) { +export async function fetchAggregatedTelemetry({ + fetchImpl = globalThis.fetch, + windowMs = CHART_WINDOW_MS, + bucketSeconds = TELEMETRY_BUCKET_SECONDS, +} = {}) { const fetchFn = typeof fetchImpl === 'function' ? fetchImpl : null; if (!fetchFn) { throw new TypeError('A fetch implementation is required to load telemetry'); } - const effectiveLimit = Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : TELEMETRY_AGGREGATE_LIMIT; - const response = await fetchFn(`/api/telemetry?limit=${effectiveLimit}`, { cache: 'no-store' }); + const windowSecondsCandidate = + Number.isFinite(windowMs) ? Math.floor(windowMs / 1000) : Math.floor(CHART_WINDOW_MS / 1000); + const windowSeconds = windowSecondsCandidate > 0 ? windowSecondsCandidate : Math.floor(CHART_WINDOW_MS / 1000); + const bucketSecondsCandidate = Number.isFinite(bucketSeconds) ? Math.floor(bucketSeconds) : TELEMETRY_BUCKET_SECONDS; + const bucketSecondsSafe = bucketSecondsCandidate > 0 ? bucketSecondsCandidate : TELEMETRY_BUCKET_SECONDS; + const response = await fetchFn( + `/api/telemetry/aggregated?windowSeconds=${windowSeconds}&bucketSeconds=${bucketSecondsSafe}`, + { cache: 'no-store' }, + ); if (!response.ok) { - throw new Error(`Failed to fetch telemetry (HTTP ${response.status})`); + throw new Error(`Failed to fetch aggregated telemetry (HTTP ${response.status})`); } const payload = await response.json(); - return Array.isArray(payload) ? payload : []; + if (!Array.isArray(payload)) { + return []; + } + return payload + .map(bucket => normalizeAggregatedSnapshot(bucket)) + .filter(snapshot => snapshot != null); } export async function initializeChartsPage(options = {}) { @@ -113,12 +206,13 @@ export async function initializeChartsPage(options = {}) { const renderCharts = typeof options.renderCharts === 'function' ? options.renderCharts : renderTelemetryCharts; const fetchImpl = options.fetchImpl ?? globalThis.fetch; - const limit = options.limit ?? TELEMETRY_AGGREGATE_LIMIT; + const bucketSeconds = options.bucketSeconds ?? TELEMETRY_BUCKET_SECONDS; + const windowMs = options.windowMs ?? CHART_WINDOW_MS; container.innerHTML = renderStatus('Loading aggregated telemetry charts…'); try { - const snapshots = await fetchAggregatedTelemetry({ fetchImpl, limit }); + const snapshots = await fetchAggregatedTelemetry({ fetchImpl, bucketSeconds, windowMs }); if (!Array.isArray(snapshots) || snapshots.length === 0) { container.innerHTML = renderStatus('Telemetry snapshots are unavailable.'); return true; @@ -127,10 +221,10 @@ export async function initializeChartsPage(options = {}) { const chartsHtml = renderCharts(node, { nowMs: Date.now(), chartOptions: { - windowMs: DAY_MS, - timeRangeLabel: 'Last 24 hours', - xAxisTickBuilder: buildHourlyTickList, - xAxisTickFormatter: formatHourLabel, + windowMs, + timeRangeLabel: 'Last 7 days', + xAxisTickBuilder: buildMidnightTickList, + xAxisTickFormatter: formatDayOfMonthLabel, lineReducer: points => buildMovingAverageSeries(points, HOUR_MS), }, }); diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb index d10413d..7d750c1 100644 --- a/web/spec/app_spec.rb +++ b/web/spec/app_spec.rb @@ -4350,6 +4350,54 @@ RSpec.describe "Potato Mesh Sinatra app" do end end + describe "GET /api/telemetry/aggregated" do + it "returns aggregated telemetry buckets for the requested interval" do + post "/api/telemetry", telemetry_fixture.to_json, auth_headers + expect(last_response).to be_ok + + get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300" + + expect(last_response).to be_ok + buckets = JSON.parse(last_response.body) + expect(buckets).not_to be_empty + a_bucket = buckets.first + expect(a_bucket["bucket_seconds"]).to eq(300) + expect(a_bucket["sample_count"]).to be >= 1 + expect(a_bucket["bucket_start"]).to be_a(Integer) + expect(a_bucket["bucket_end"]).to be_a(Integer) + expect(a_bucket["aggregates"]).to be_a(Hash) + expect(a_bucket["aggregates"]).to have_key("battery_level") + expect(a_bucket["aggregates"]["battery_level"]).to include("avg") + expect(a_bucket).not_to have_key("device_metrics") + end + + it "applies default window and bucket sizes when parameters are omitted" do + post "/api/telemetry", telemetry_fixture.to_json, auth_headers + expect(last_response).to be_ok + + get "/api/telemetry/aggregated" + + expect(last_response).to be_ok + buckets = JSON.parse(last_response.body) + expect(buckets.length).to be >= 1 + expect(buckets.first["bucket_seconds"]).to eq(PotatoMesh::App::Queries::DEFAULT_TELEMETRY_BUCKET_SECONDS) + end + + it "rejects invalid bucket and window parameters" do + get "/api/telemetry/aggregated?windowSeconds=0&bucketSeconds=300" + expect(last_response.status).to eq(400) + expect(JSON.parse(last_response.body)).to eq("error" => "windowSeconds must be positive") + + get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=0" + expect(last_response.status).to eq(400) + expect(JSON.parse(last_response.body)).to eq("error" => "bucketSeconds must be positive") + + get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=1" + expect(last_response.status).to eq(400) + expect(JSON.parse(last_response.body)).to eq("error" => "bucketSeconds too small for requested window") + end + end + describe "GET /api/traces" do it "returns stored traces ordered by receive time" do clear_database