From 405592346dba609589a24aead44ca4439e0eea8e Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 18:58:32 +0200 Subject: [PATCH 01/23] feat(db): add cml_stats_history hypertable for historical time slider Add new hypertable to store hourly snapshots of CML statistics, enabling historical queries without scanning raw data. Changes: - Create cml_stats_history table with compression policy (7 days) - Add materialize_cml_stats_snapshot() function to write definitive snapshots from cml_data_1h aggregates - Add get_cml_stats_at() function for efficient historical lookups - Grant template comments for per-user isolation via generate_config.py Key features: - Idempotent upserts via ON CONFLICT (snapshot_time, cml_id, user_id) - Supports provisional (current partial hour) and definitive rows - Hour-truncated snapshot times for consistent querying - Compression enabled for snapshots older than 7 days --- .../migrations/017_add_cml_stats_history.sql | 179 ++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 database/migrations/017_add_cml_stats_history.sql diff --git a/database/migrations/017_add_cml_stats_history.sql b/database/migrations/017_add_cml_stats_history.sql new file mode 100644 index 0000000..4164174 --- /dev/null +++ b/database/migrations/017_add_cml_stats_history.sql @@ -0,0 +1,179 @@ +-- Migration 017: Add cml_stats_history hypertable for historical time slider +-- +-- This migration adds a new hypertable to store snapshots of CML stats at hourly intervals. +-- Enables the historical time slider feature on the realtime map. +-- +-- Key features: +-- - Stores hourly snapshots of cml_stats for historical queries +-- - Supports provisional (current partial hour) and definitive (completed hours) rows +-- - Compression policy for old data (>7 days) +-- - Functions for materialization and querying +-- +-- Apply with: +-- docker compose exec -T database psql -U myuser -d mydatabase \ +-- < database/migrations/017_add_cml_stats_history.sql + +-- Create the cml_stats_history table +CREATE TABLE IF NOT EXISTS cml_stats_history ( + snapshot_time TIMESTAMPTZ NOT NULL, + cml_id TEXT NOT NULL, + user_id TEXT NOT NULL, + completeness_percent_6h REAL, + total_records_6h BIGINT, + valid_records_6h BIGINT, + mean_rsl_6h REAL, + stddev_rsl_6h REAL, + completeness_percent_1h REAL, + mean_rsl_1h REAL, + stddev_rsl_1h REAL, + last_rsl REAL, + is_provisional BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (snapshot_time, cml_id, user_id) +); + +-- Create hypertable with 7-day chunks +SELECT create_hypertable( + 'cml_stats_history', 'snapshot_time', + chunk_time_interval => INTERVAL '7 days', + if_not_exists => TRUE +); + +-- Enable compression for old snapshots (>7 days) +ALTER TABLE cml_stats_history + SET (timescaledb.compress, + timescaledb.compress_segmentby = 'user_id, cml_id', + timescaledb.compress_orderby = 'snapshot_time DESC'); + +SELECT add_compression_policy('cml_stats_history', INTERVAL '7 days'); + +-- Materialization function: writes definitive snapshot from cml_data_1h +CREATE OR REPLACE FUNCTION materialize_cml_stats_snapshot( + p_at_time TIMESTAMPTZ, -- must be hour-truncated by caller + p_user_id TEXT +) RETURNS INT -- number of rows upserted +LANGUAGE plpgsql SECURITY DEFINER +SET search_path = public +AS $$ +DECLARE + v_rows INT; +BEGIN + -- Validate caller is a known user + IF NOT EXISTS (SELECT 1 FROM cml_metadata WHERE user_id = p_user_id LIMIT 1) THEN + RAISE EXCEPTION 'Unknown user: %', p_user_id; + END IF; + + INSERT INTO cml_stats_history ( + snapshot_time, cml_id, user_id, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, + last_rsl, is_provisional + ) + SELECT + p_at_time, + m.cml_id, + p_user_id, + -- 6-hour window: sum the 6 hourly buckets ending at p_at_time + ROUND( + 100.0 * SUM(h6.rsl_count)::numeric + / NULLIF(SUM(h6.record_count), 0), + 2) AS completeness_percent_6h, + SUM(h6.record_count) AS total_records_6h, + SUM(h6.rsl_count) AS valid_records_6h, + ROUND(AVG(h6.rsl_avg)::numeric, 2) AS mean_rsl_6h, + ROUND(AVG(h6.rsl_stddev)::numeric, 2) AS stddev_rsl_6h, + -- 1-hour window: the single bucket containing p_at_time + ROUND( + 100.0 * h1.rsl_count::numeric + / NULLIF(h1.record_count, 0), + 2) AS completeness_percent_1h, + ROUND(h1.rsl_avg::numeric, 2) AS mean_rsl_1h, + ROUND(h1.rsl_stddev::numeric, 2) AS stddev_rsl_1h, + h1.rsl_max AS last_rsl, + FALSE AS is_provisional + FROM cml_metadata m + -- 6h subquery from the 1h continuous aggregate + LEFT JOIN LATERAL ( + SELECT + SUM(record_count) AS record_count, + SUM(CASE WHEN rsl_max IS NOT NULL THEN record_count ELSE 0 END) AS rsl_count, + AVG(rsl_avg) AS rsl_avg, + AVG(rsl_max - rsl_min) AS rsl_stddev + FROM cml_data_1h + WHERE user_id = p_user_id + AND cml_id = m.cml_id + AND bucket >= p_at_time - INTERVAL '6 hours' + AND bucket < p_at_time + ) h6 ON TRUE + -- 1h subquery: the bucket immediately before p_at_time + LEFT JOIN LATERAL ( + SELECT record_count, rsl_avg, rsl_max, rsl_min, + (rsl_max - rsl_min) AS rsl_stddev + FROM cml_data_1h + WHERE user_id = p_user_id + AND cml_id = m.cml_id + AND bucket = p_at_time - INTERVAL '1 hour' + ) h1 ON TRUE + WHERE m.user_id = p_user_id + ON CONFLICT (snapshot_time, cml_id, user_id) DO UPDATE SET + completeness_percent_6h = EXCLUDED.completeness_percent_6h, + total_records_6h = EXCLUDED.total_records_6h, + valid_records_6h = EXCLUDED.valid_records_6h, + mean_rsl_6h = EXCLUDED.mean_rsl_6h, + stddev_rsl_6h = EXCLUDED.stddev_rsl_6h, + completeness_percent_1h = EXCLUDED.completeness_percent_1h, + mean_rsl_1h = EXCLUDED.mean_rsl_1h, + stddev_rsl_1h = EXCLUDED.stddev_rsl_1h, + last_rsl = EXCLUDED.last_rsl, + is_provisional = FALSE; + + GET DIAGNOSTICS v_rows = ROW_COUNT; + RETURN v_rows; +END; +$$; + +-- Read function for webserver / Grafana +CREATE OR REPLACE FUNCTION get_cml_stats_at( + p_at_time TIMESTAMPTZ, + p_user_id TEXT +) RETURNS TABLE ( + cml_id TEXT, + completeness_percent_6h REAL, + total_records_6h BIGINT, + valid_records_6h BIGINT, + mean_rsl_6h REAL, + stddev_rsl_6h REAL, + completeness_percent_1h REAL, + mean_rsl_1h REAL, + stddev_rsl_1h REAL, + last_rsl REAL, + is_provisional BOOLEAN +) LANGUAGE sql STABLE SECURITY DEFINER +SET search_path = public +AS $$ + SELECT + h.cml_id, + h.completeness_percent_6h, + h.total_records_6h, + h.valid_records_6h, + h.mean_rsl_6h, + h.stddev_rsl_6h, + h.completeness_percent_1h, + h.mean_rsl_1h, + h.stddev_rsl_1h, + h.last_rsl, + h.is_provisional + FROM cml_stats_history h + WHERE h.user_id = p_user_id + AND h.snapshot_time = date_trunc('hour', p_at_time) + ORDER BY h.cml_id; +$$; + +-- Grants for writers (parser roles) - applied per-user by generate_config.py +-- GRANT SELECT, INSERT, UPDATE ON cml_stats_history TO ; +-- GRANT EXECUTE ON FUNCTION materialize_cml_stats_snapshot(TIMESTAMPTZ, TEXT) TO ; + +-- Grants for readers (webserver, Grafana) +GRANT SELECT ON cml_stats_history TO webserver_role; +GRANT EXECUTE ON FUNCTION get_cml_stats_at(TIMESTAMPTZ, TEXT) + TO webserver_role; From 03e36b66d983a05e763ec62f00572c0e79792c50 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 18:59:04 +0200 Subject: [PATCH 02/23] feat(parser): add snapshot materialization for historical time slider Implement background jobs to populate cml_stats_history with hourly snapshots, enabling fast historical queries without raw data scans. Changes: - Add DBWriter.write_stats_snapshot(at_time) - materializes definitive snapshot from cml_data_1h for completed hours - Add DBWriter.write_provisional_snapshot() - copies live cml_stats every 60s for current partial hour (is_provisional=TRUE) - Wire both methods into stats background thread in sftp_push.py - Add backfill_stats_history.py script for populating historical data Architecture: - Provisional snapshots: written every 60s, at most 60s stale - Definitive snapshots: written when hour turns, sourced from aggregates - ON CONFLICT DO UPDATE ensures idempotent re-materialization - Backfill script processes all available hours from cml_data_1h --- parser/backfill_stats_history.py | 76 +++++++++++++++++++++++++ parser/db_writer.py | 95 ++++++++++++++++++++++++++++++++ parser/entrypoints/sftp_push.py | 27 ++++++++- 3 files changed, 196 insertions(+), 2 deletions(-) create mode 100644 parser/backfill_stats_history.py diff --git a/parser/backfill_stats_history.py b/parser/backfill_stats_history.py new file mode 100644 index 0000000..ab1438c --- /dev/null +++ b/parser/backfill_stats_history.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Backfill cml_stats_history from existing cml_data_1h data. + +Run once after applying migration 017 to populate historical snapshots. + +Usage: + docker compose exec parser python -m parser.backfill_stats_history + +Or with environment variables: + DATABASE_URL=postgresql://... USER_ID=demo_openmrg python backfill_stats_history.py +""" + +import os +import sys +from datetime import datetime, timezone + +# Add parent directory to path for imports +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from db_writer import DBWriter + + +def backfill_stats_history(): + """Backfill cml_stats_history for all available hours.""" + database_url = os.getenv( + "DATABASE_URL", "postgresql://myuser:mypassword@database:5432/mydatabase" + ) + user_id = os.getenv("USER_ID", "demo_openmrg") + + print(f"Starting backfill for user_id={user_id}") + + db = DBWriter(database_url, user_id=user_id) + db.connect() + + try: + # Get all distinct hours from cml_data_1h + cur = db.conn.cursor() + cur.execute( + """ + SELECT DISTINCT date_trunc('hour', bucket) as hour + FROM cml_data_1h + WHERE user_id = %s + ORDER BY hour ASC + """, + (user_id,) + ) + hours = [row[0] for row in cur.fetchall()] + cur.close() + + if not hours: + print("No historical data found in cml_data_1h") + return + + print(f"Found {len(hours)} hours to backfill") + + # Backfill each hour + processed = 0 + for i, hour in enumerate(hours): + try: + rows = db.write_stats_snapshot(hour) + processed += 1 + + if processed % 100 == 0 or i == len(hours) - 1: + print(f"Progress: {processed}/{len(hours)} hours processed ({100.0*processed/len(hours):.1f}%)") + except Exception as e: + print(f"Error processing hour {hour}: {e}") + continue + + print(f"Backfill complete: {processed}/{len(hours)} hours successfully processed") + + finally: + db.close() + + +if __name__ == "__main__": + backfill_stats_history() diff --git a/parser/db_writer.py b/parser/db_writer.py index 49b4e41..d58f52f 100644 --- a/parser/db_writer.py +++ b/parser/db_writer.py @@ -9,6 +9,7 @@ """ from typing import List, Tuple, Optional, Set, Callable, TypeVar +from datetime import datetime import time import functools import psycopg2 @@ -409,3 +410,97 @@ def refresh_windowed_stats(self) -> None: finally: if cur and not cur.closed: cur.close() + + def write_stats_snapshot(self, at_time: datetime) -> int: + """Materialize a cml_stats_history snapshot for the given hour. + + Calls materialize_cml_stats_snapshot(date_trunc('hour', at_time), user_id). + Returns the number of rows upserted. + """ + from datetime import timezone + at_hour = at_time.replace(minute=0, second=0, microsecond=0) + # Ensure timezone-aware if naive + if at_hour.tzinfo is None: + at_hour = at_hour.replace(tzinfo=timezone.utc) + + cur = self.conn.cursor() + try: + cur.execute( + "SELECT materialize_cml_stats_snapshot(%s::timestamptz, %s)", + (at_hour, self.user_id), + ) + rows = cur.fetchone()[0] + self.conn.commit() + logger.info("Materialized cml_stats_history snapshot at %s (%d rows)", at_hour, rows) + return rows + except Exception: + try: + self.conn.rollback() + except Exception: + pass + logger.exception("Failed to write stats snapshot at %s", at_hour) + raise + finally: + if cur and not cur.closed: + cur.close() + + def write_provisional_snapshot(self) -> int: + """Copy live cml_stats rolling-window values into cml_stats_history for + the current (incomplete) calendar hour, marked is_provisional=TRUE. + + Runs every 60 s alongside refresh_windowed_stats so the slider can always + reach the current hour with data that is at most 60 s stale. The + ON CONFLICT DO UPDATE overwrites any previous provisional row for the same + hour. When the hour turns, write_stats_snapshot() writes a definitive row + (is_provisional=FALSE) that permanently replaces this one. + """ + from datetime import timezone + current_hour = datetime.now(tz=timezone.utc).replace(minute=0, second=0, microsecond=0) + cur = self.conn.cursor() + try: + cur.execute( + """ + INSERT INTO cml_stats_history ( + snapshot_time, cml_id, user_id, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, + last_rsl, is_provisional + ) + SELECT + %s, cml_id, user_id, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, + last_rsl, TRUE + FROM cml_stats + WHERE user_id = %s + ON CONFLICT (snapshot_time, cml_id, user_id) DO UPDATE SET + completeness_percent_6h = EXCLUDED.completeness_percent_6h, + total_records_6h = EXCLUDED.total_records_6h, + valid_records_6h = EXCLUDED.valid_records_6h, + mean_rsl_6h = EXCLUDED.mean_rsl_6h, + stddev_rsl_6h = EXCLUDED.stddev_rsl_6h, + completeness_percent_1h = EXCLUDED.completeness_percent_1h, + mean_rsl_1h = EXCLUDED.mean_rsl_1h, + stddev_rsl_1h = EXCLUDED.stddev_rsl_1h, + last_rsl = EXCLUDED.last_rsl, + is_provisional = TRUE -- keep provisional until definitive row arrives + WHERE cml_stats_history.is_provisional = TRUE -- never overwrite a definitive row + """, + (current_hour, self.user_id), + ) + rows = cur.rowcount + self.conn.commit() + logger.debug("Wrote provisional snapshot at %s (%d rows)", current_hour, rows) + return rows + except Exception: + try: + self.conn.rollback() + except Exception: + pass + logger.exception("Failed to write provisional stats snapshot") + raise + finally: + if cur and not cur.closed: + cur.close() diff --git a/parser/entrypoints/sftp_push.py b/parser/entrypoints/sftp_push.py index 0398d46..7be28d1 100644 --- a/parser/entrypoints/sftp_push.py +++ b/parser/entrypoints/sftp_push.py @@ -8,6 +8,7 @@ import time import logging import threading +from datetime import datetime, timezone from pathlib import Path from ..file_watcher import FileWatcher @@ -148,17 +149,39 @@ def stats_loop(): if stop_event.is_set(): return + # Track last snapshot hour for hourly definitive snapshots + from datetime import timezone + last_snapshot_hour = None + # Run immediately on startup so Grafana has fresh stats without # waiting a full interval after the backlog is processed. try: stats_db.refresh_windowed_stats() + # Also write initial provisional snapshot + stats_db.write_provisional_snapshot() + # Write definitive snapshot for current hour if needed + now = datetime.now(tz=timezone.utc) + current_hour = now.replace(minute=0, second=0, microsecond=0) + if current_hour != last_snapshot_hour: + stats_db.write_stats_snapshot(now) + last_snapshot_hour = current_hour except Exception: - logger.exception("Stats thread: initial refresh_windowed_stats failed") + logger.exception("Stats thread: initial refresh failed") + while not stop_event.wait(Config.STATS_REFRESH_INTERVAL): try: stats_db.refresh_windowed_stats() + # Always: keep the current-hour provisional snapshot fresh (at most 60 s stale) + stats_db.write_provisional_snapshot() + + # Once per hour: materialise the just-completed hour from cml_data_1h + now = datetime.now(tz=timezone.utc) + current_hour = now.replace(minute=0, second=0, microsecond=0) + if current_hour != last_snapshot_hour: + stats_db.write_stats_snapshot(now) + last_snapshot_hour = current_hour except Exception: - logger.exception("Stats thread: refresh_windowed_stats failed") + logger.exception("Stats thread: refresh failed") stats_db.close() stats_thread = threading.Thread( From c7c07c5214c75358cbce8d379b55a97f8f755a21 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 18:59:48 +0200 Subject: [PATCH 03/23] feat(webserver): add historical time slider API endpoints Extend API to support historical queries for the time slider feature. Changes: - Extend /api/cml-stats with optional ?at= parameter for historical lookups - Parses ISO 8601 timestamp, queries get_cml_stats_at() function - Returns identical response shape with added is_provisional flag - Invalid timestamps return HTTP 400 - Add /api/cml-stats-time-range endpoint - Returns min(snapshot_time) and max (current hour) from cml_stats_history - Enables frontend to configure slider min/max bounds - Update generate_config.py to grant permissions for new table/function Key properties: - Live mode unchanged when ?at= omitted - Historical reads use index lookups, no raw data scans - Response includes is_provisional for current partial hour --- scripts/generate_config.py | 2 ++ webserver/main.py | 73 ++++++++++++++++++++++++++++---------- 2 files changed, 56 insertions(+), 19 deletions(-) diff --git a/scripts/generate_config.py b/scripts/generate_config.py index a12b419..8c575e8 100644 --- a/scripts/generate_config.py +++ b/scripts/generate_config.py @@ -311,6 +311,7 @@ def generate_users_json(users: list[dict], existing_json: dict) -> dict: -- RLS on cml_metadata and cml_stats is enforced via the generic -- current_user policy already installed on those tables. GRANT SELECT, INSERT, UPDATE ON cml_metadata, cml_stats TO {user_id}; +GRANT SELECT, INSERT, UPDATE ON cml_stats_history TO {user_id}; -- cml_data has no RLS (compressed TimescaleDB hypertable). -- Parser writes (write_rawdata) and stats updates (update_cml_stats) go @@ -321,6 +322,7 @@ def generate_users_json(users: list[dict], existing_json: dict) -> dict: GRANT SELECT ON cml_data_1h_secure TO {user_id}; GRANT EXECUTE ON FUNCTION update_cml_stats(TEXT, TEXT) TO {user_id}; GRANT EXECUTE ON FUNCTION update_cml_stats_windowed(TEXT, TEXT) TO {user_id}; +GRANT EXECUTE ON FUNCTION materialize_cml_stats_snapshot(TIMESTAMPTZ, TEXT) TO {user_id}; -- file_processing_log: parser INSERTs a row for every processed file; -- webserver_role only needs SELECT. diff --git a/webserver/main.py b/webserver/main.py index f1c0e6e..96d59cd 100644 --- a/webserver/main.py +++ b/webserver/main.py @@ -27,7 +27,7 @@ ) from werkzeug.security import check_password_hash from werkzeug.utils import secure_filename -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from pathlib import Path from contextlib import contextmanager import uuid @@ -522,39 +522,52 @@ def api_cml_map(): @login_required def api_cml_stats(): """API endpoint for fetching per-CML statistics for data quality visualization""" + at_param = request.args.get("at") # ISO 8601 string or absent + try: with user_db_scope(current_user.id) as conn: cur = conn.cursor() - cur.execute( - """ - SELECT - cml_id::text, - completeness_percent_6h, - total_records_6h, - valid_records_6h, - mean_rsl_6h, - stddev_rsl_6h, - completeness_percent_1h, - stddev_rsl_1h, - last_rsl - FROM cml_stats - ORDER BY cml_id - """ - ) + + if at_param: + # Historical: parse the timestamp and query cml_stats_history + try: + at_ts = datetime.fromisoformat(at_param.replace("Z", "+00:00")) + except ValueError: + return jsonify({"error": "invalid 'at' parameter"}), 400 + + cur.execute( + "SELECT * FROM get_cml_stats_at(%s::timestamptz, %s)", + (at_ts, current_user.id), + ) + else: + # Live: existing query from cml_stats (windowed, pre-computed) + cur.execute( + """ + SELECT cml_id::text, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, stddev_rsl_1h, last_rsl + FROM cml_stats + ORDER BY cml_id + """ + ) + data = cur.fetchall() cur.close() + # Response shape is identical in both branches; is_provisional only present for ?at= path stats = [ { "cml_id": str(row[0]), - "completeness_percent": safe_float(row[1]), # 6h window + "completeness_percent": safe_float(row[1]), "total_records": int(row[2] or 0), "valid_records": int(row[3] or 0), "mean_rsl": safe_float(row[4]), "stddev_rsl": safe_float(row[5]), "completeness_percent_1h": safe_float(row[6]), - "stddev_last_60min": safe_float(row[7]), # pre-computed 1h stddev + "stddev_last_60min": safe_float(row[7]), "last_rsl": safe_float(row[8]), + "is_provisional": bool(row[9]) if at_param and len(row) > 9 else False, } for row in data ] @@ -564,6 +577,28 @@ def api_cml_stats(): return jsonify([]) +@app.route("/api/cml-stats-time-range") +@login_required +def api_cml_stats_time_range(): + """API endpoint for fetching the time range available in cml_stats_history.""" + now = datetime.now(tz=timezone.utc) + current_hour = now.replace(minute=0, second=0, microsecond=0) + try: + with user_db_scope(current_user.id) as conn: + cur = conn.cursor() + cur.execute( + "SELECT MIN(snapshot_time) FROM cml_stats_history WHERE user_id = %s", + (current_user.id,), + ) + row = cur.fetchone() + cur.close() + min_ts = row[0].isoformat() if row and row[0] else None + return jsonify({"min": min_ts, "max": current_hour.isoformat()}) + except Exception as e: + print(f"Error fetching stats time range: {e}") + return jsonify({"min": None, "max": current_hour.isoformat()}) + + @app.route("/api/data-time-range") @login_required def api_data_time_range(): From 993b2ea8e964029533a19e9da8a5a192c197874a Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 19:00:15 +0200 Subject: [PATCH 04/23] feat(ui): add time slider control to realtime CML map MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement interactive time slider for historical data exploration with Grafana dashboard synchronization. Changes: - Add time slider section to Leaflet ColorControl widget - Range input with 1-hour steps, enabled after loading time range - Live button toggles between live/history modes - Visual display shows selected time or 'Live' - Debounced input (300ms) to avoid excessive API calls - Implement fetchAndApplyStats(atIso) function - Fetches stats with optional ?at= parameter - Displays '(partial)' label for provisional snapshots - Updates map path colors based on historical stats - Add syncGrafanaTime(epochSec) function - Syncs iframe time range via contentWindow.location - Live mode: relative 'now-1h' to 'now' - Historical mode: ±30 min window around selection - Sets var-marker_time variable for annotation pin - Update Grafana dashboard cml-realtime.json - Add hidden marker_time template variable (textbox, epoch ms) - Add 'Selected time' annotation pinned to marker_time - Auto-refresh in live mode (30s), paused when viewing history User experience: - Slider drag shows preview timestamp immediately - After 300ms debounce, fetches historical stats and syncs Grafana - Clicking 'Live' button restores auto-refresh and current time - Partial hours clearly labeled to indicate incomplete data --- .../dashboards/definitions/cml-realtime.json | 19 +++ webserver/templates/realtime.html | 157 +++++++++++++++++- 2 files changed, 174 insertions(+), 2 deletions(-) diff --git a/grafana/provisioning/dashboards/definitions/cml-realtime.json b/grafana/provisioning/dashboards/definitions/cml-realtime.json index 27a06e6..3b1b3f4 100644 --- a/grafana/provisioning/dashboards/definitions/cml-realtime.json +++ b/grafana/provisioning/dashboards/definitions/cml-realtime.json @@ -12,6 +12,18 @@ "iconColor": "rgba(0, 211, 255, 1)", "name": "Annotations & Alerts", "type": "dashboard" + }, + { + "name": "Selected time", + "enable": true, + "hide": false, + "iconColor": "red", + "type": "dashboard", + "builtIn": 0, + "datasource": { "type": "grafana", "uid": "-- Grafana --" }, + "rawQuery": true, + "query": "SELECT $marker_time AS time, 'selected' AS text", + "timeField": "time" } ] }, @@ -1002,6 +1014,13 @@ "queryValue": "", "skipUrlSync": true, "type": "custom" + }, + { + "name": "marker_time", + "type": "textbox", + "label": "Marker time (epoch ms)", + "hide": 2, + "current": { "value": "" } } ] }, diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index 4cbcf33..ac6a22b 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -116,6 +116,11 @@ let cmlStats = {}; let selectedCmlId = '10001'; let currentColoringOption = 'completeness'; + + // Time slider state + let selectedTime = null; // null = live mode + let statsRefreshTimer = null; // auto-refresh handle + const LIVE_REFRESH_MS = 30000; // Color functions for each option const colorFunctions = { @@ -231,7 +236,31 @@ `; - // Prevent map interactions when using the select + // Time slider section + var divider = L.DomUtil.create('hr', '', container); + divider.style.margin = '8px 0'; + + var timeLabel = L.DomUtil.create('div', '', container); + timeLabel.innerHTML = 'Time:'; + timeLabel.style.marginBottom = '4px'; + + var liveBtn = L.DomUtil.create('button', '', container); + liveBtn.id = 'timeLiveBtn'; + liveBtn.textContent = '● Live'; + liveBtn.style.cssText = 'width:100%;margin-bottom:4px;background:#22c55e;color:white;border:none;border-radius:4px;padding:3px 6px;cursor:pointer;'; + + var timeSlider = L.DomUtil.create('input', '', container); + timeSlider.type = 'range'; + timeSlider.id = 'timeSlider'; + timeSlider.style.cssText = 'width:100%;margin:4px 0;'; + timeSlider.disabled = true; // enabled once time range is loaded + + var timeDisplay = L.DomUtil.create('div', '', container); + timeDisplay.id = 'timeDisplay'; + timeDisplay.style.cssText = 'font-size:11px;text-align:center;color:#555;'; + timeDisplay.textContent = 'Loading range…'; + + // Prevent map interactions when using the controls L.DomEvent.disableClickPropagation(container); L.DomEvent.disableScrollPropagation(container); @@ -242,7 +271,7 @@ // Start map rendering immediately loadCmlData(); - + // Fetch stats in parallel and update colors/popups when ready console.log('Fetching CML stats...'); try { @@ -274,6 +303,11 @@ var errEl = document.getElementById('debugError'); if (errEl) errEl.textContent = 'lastError=stats-ex:' + String(e); } + + // Initialize time slider and start live refresh + fetchAndApplyStats(null); + startLiveRefresh(); + initTimeSlider(); } catch (e) { console.error('Error in initializeMap:', e); } @@ -287,6 +321,43 @@ updateMapColors(); }); } + + // Time slider event handlers + var liveBtn = document.getElementById('timeLiveBtn'); + if (liveBtn) { + liveBtn.addEventListener('click', function () { + selectedTime = null; + document.getElementById('timeDisplay').textContent = 'Live'; + this.style.background = '#22c55e'; + clearTimeout(sliderDebounce); + fetchAndApplyStats(null); + startLiveRefresh(); + syncGrafanaTime(null); + }); + } + + var sliderDebounce; + var timeSlider = document.getElementById('timeSlider'); + if (timeSlider) { + timeSlider.addEventListener('input', function () { + clearTimeout(sliderDebounce); + var epochSec = parseInt(this.value, 10); + var dt = new Date(epochSec * 1000); + // Check whether this is the current (partial) hour + var nowHour = new Date(); nowHour.setMinutes(0, 0, 0, 0); + var sliderHour = new Date(dt); sliderHour.setMinutes(0, 0, 0, 0); + var label = dt.toISOString().replace('T', ' ').slice(0, 16) + ' UTC'; + if (sliderHour.getTime() >= nowHour.getTime()) label += ' (partial)'; + document.getElementById('timeDisplay').textContent = label; + sliderDebounce = setTimeout(function () { + selectedTime = dt.toISOString(); + stopLiveRefresh(); + document.getElementById('timeLiveBtn').style.background = '#aaa'; + fetchAndApplyStats(selectedTime); // provisional snapshot handles current hour + syncGrafanaTime(epochSec); + }, 300); + }); + } } // Load and draw CML data @@ -421,6 +492,88 @@ } } + // Fetch and apply CML stats with optional historical time + function fetchAndApplyStats(atIso) { + var url = '/api/cml-stats'; + if (atIso) url += '?at=' + encodeURIComponent(atIso); + fetch(url) + .then(function(r) { return r.json(); }) + .then(function(stats) { + // Surface provisional state in the time display label + var anyProvisional = stats.some(function(s) { return s.is_provisional; }); + var disp = document.getElementById('timeDisplay'); + if (atIso && anyProvisional && disp && !disp.textContent.includes('(partial)')) { + disp.textContent += ' (partial)'; + } + stats.forEach(function(stat) { + cmlStats[stat.cml_id] = stat; + applyStatsToLine(stat.cml_id, stat); + }); + }) + .catch(function(err) { console.error('Stats fetch error:', err); }); + } + + // Sync Grafana iframe time with selected time + function syncGrafanaTime(epochSec) { + var grafanaPanel = document.getElementById('grafana-panel'); + var iframeWindow = grafanaPanel.contentWindow; + try { + var currentUrl = new URL(iframeWindow.location.href); + if (epochSec === null) { + // Live: restore relative time range + currentUrl.searchParams.set('from', 'now-1h'); + currentUrl.searchParams.set('to', 'now'); + currentUrl.searchParams.delete('var-marker_time'); + } else { + // Historical: ±30 min window around selected second + var halfWindow = 30 * 60 * 1000; // 30 min in ms + var epochMs = epochSec * 1000; + currentUrl.searchParams.set('from', String(epochMs - halfWindow)); + currentUrl.searchParams.set('to', String(epochMs + halfWindow)); + currentUrl.searchParams.set('var-marker_time', String(epochMs)); + } + iframeWindow.history.pushState(null, '', currentUrl.toString()); + iframeWindow.dispatchEvent(new PopStateEvent('popstate', { state: null })); + } catch (e) { + console.warn('Grafana sync skipped (iframe not ready):', e); + } + } + + // Initialize time slider + function initTimeSlider() { + fetch('/api/cml-stats-time-range') + .then(function(r) { return r.json(); }) + .then(function(range) { + if (!range.min || !range.max) { + document.getElementById('timeDisplay').textContent = 'No history'; + return; + } + var minEpoch = Math.floor(new Date(range.min).getTime() / 1000); + var maxEpoch = Math.floor(new Date(range.max).getTime() / 1000); + var slider = document.getElementById('timeSlider'); + slider.min = minEpoch; + slider.max = maxEpoch; + slider.step = 3600; // 1-hour steps + slider.value = maxEpoch; + slider.disabled = false; + document.getElementById('timeDisplay').textContent = 'Live'; + }) + .catch(function() { + document.getElementById('timeDisplay').textContent = 'Unavailable'; + }); + } + + // Start live refresh timer + function startLiveRefresh() { + stopLiveRefresh(); + statsRefreshTimer = setInterval(function () { fetchAndApplyStats(null); }, LIVE_REFRESH_MS); + } + + // Stop live refresh timer + function stopLiveRefresh() { + if (statsRefreshTimer) clearInterval(statsRefreshTimer); + statsRefreshTimer = null; + } // Select CML and update Grafana dashboard function selectCml(cmlId) { selectedCmlId = cmlId; From dde57fbb48d79441a8fe52e15f545513042761ba Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 19:06:24 +0200 Subject: [PATCH 05/23] test,docs: add unit tests and implementation plan for time slider Test coverage: - parser/tests/test_stats_snapshot.py: Tests for DBWriter snapshot methods - write_stats_snapshot() materialization from aggregates - write_provisional_snapshot() copying live stats - Error handling and rollback behavior - Timezone handling for naive datetimes - webserver/tests/test_time_slider_api.py: API endpoint tests - Historical queries with ?at= parameter - Invalid timestamp handling (HTTP 400) - Time range endpoint responses - Empty history scenarios Documentation: - docs/plan-time-slider-2026-07-24.md: Complete implementation plan - Architecture overview and data flow - Database schema and functions - Parser background job design - Webserver API extensions - Frontend UI components - Grafana dashboard integration - Testing strategy and deployment notes --- docs/plan-time-slider-2026-07-24.md | 725 ++++++++++++++++++++++++ parser/tests/test_stats_snapshot.py | 124 ++++ webserver/tests/test_time_slider_api.py | 171 ++++++ 3 files changed, 1020 insertions(+) create mode 100644 docs/plan-time-slider-2026-07-24.md create mode 100644 parser/tests/test_stats_snapshot.py create mode 100644 webserver/tests/test_time_slider_api.py diff --git a/docs/plan-time-slider-2026-07-24.md b/docs/plan-time-slider-2026-07-24.md new file mode 100644 index 0000000..0e7a6fe --- /dev/null +++ b/docs/plan-time-slider-2026-07-24.md @@ -0,0 +1,725 @@ +# Plan: Historical time slider for the realtime CML map + +**Date:** 2026-07-24 +**Target branch:** fresh feature branch off `upstream/main` +**Merge path:** upstream PR → `upstream/main` → cherry-pick or merge into `ifu/main` + +--- + +## Summary + +Add a time slider to the realtime CML map (`/realtime`) that lets users pick any past +hour and see the map paths coloured by pre-materialized stats for that moment. The +selected time is simultaneously reflected in the Grafana dashboard embedded below the +map: the iframe time range shifts to a ±30-minute window around the selection, and a +vertical annotation marker pins the exact chosen instant on the time-series panel. + +**Architecture at a glance** + +``` +cml_data_1h (TimescaleDB 1h continuous aggregate, already exists) + │ + ▼ hourly background job in parser (new) is_provisional=FALSE (definitive) + │ +cml_stats (rolling-window, refreshed every 60 s, already exists) + │ + ▼ every 60 s background job in parser (new) is_provisional=TRUE (current partial hour) + │ + ├──► cml_stats_history(snapshot_time, cml_id, user_id, …, is_provisional) ← new hypertable + │ + ▼ GET /api/cml-stats?at= (new query path in webserver) +realtime.html time slider → map path colours + │ + ▼ contentWindow URL update (same pattern as var-cml_id today) +Grafana iframe from/to + var-marker_time → panel time window + annotation pin +``` + +Key properties: +- **No raw-data scans for historical reads** — all stats come from `cml_stats_history`, + which is populated from the `cml_data_1h` aggregate. Reads are instant index lookups. +- **Current (partial) calendar hour is always reachable on the slider** — a provisional + snapshot sourced directly from the live `cml_stats` rolling window is written to + `cml_stats_history` every 60 s, marked `is_provisional = TRUE`. The data is at most + 60 s stale. When the calendar hour turns, `materialize_cml_stats_snapshot` overwrites + the provisional row with a definitive one (`is_provisional = FALSE`) sourced from the + fully-materialised `cml_data_1h` bucket. +- **Completeness semantics** — `completeness_percent_1h` is the fraction of non-null RSL + records within a **rolling** 1-hour window ending at the time of writing. This is the + correct meaning for map colouring ("how healthy was data delivery in the last hour?") + and is already correct in the live `cml_stats` today. The provisional snapshot + preserves this semantics; no calendar-based normalisation is needed. +- **Live mode unchanged** — omitting `?at=` still returns pre-computed `cml_stats` rows + (the existing windowed stats), with the same auto-refresh behaviour as today. +- **Grafana sync reuses existing mechanism** — the `contentWindow.location` approach + already in use for `var-cml_id` is extended; no new iframe communication protocol + is needed. + +--- + +## PR 1 — Database + Parser: `cml_stats_history` table and hourly snapshot writer + +### 1a. New migration: `cml_stats_history` + helper functions + +File: `database/migrations/017_add_cml_stats_history.sql` + +#### Table + +```sql +CREATE TABLE IF NOT EXISTS cml_stats_history ( + snapshot_time TIMESTAMPTZ NOT NULL, + cml_id TEXT NOT NULL, + user_id TEXT NOT NULL, + completeness_percent_6h REAL, + total_records_6h BIGINT, + valid_records_6h BIGINT, + mean_rsl_6h REAL, + stddev_rsl_6h REAL, + completeness_percent_1h REAL, + mean_rsl_1h REAL, + stddev_rsl_1h REAL, + last_rsl REAL, + is_provisional BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (snapshot_time, cml_id, user_id) +); + +SELECT create_hypertable( + 'cml_stats_history', 'snapshot_time', + chunk_time_interval => INTERVAL '7 days', + if_not_exists => TRUE +); + +-- Compress old snapshots (>7 days) — they are never updated +ALTER TABLE cml_stats_history + SET (timescaledb.compress, + timescaledb.compress_segmentby = 'user_id, cml_id', + timescaledb.compress_orderby = 'snapshot_time DESC'); + +SELECT add_compression_policy('cml_stats_history', INTERVAL '7 days'); +``` + +`snapshot_time` is truncated to the hour (the caller is responsible). The primary key +guarantees idempotent re-materialization via `ON CONFLICT DO UPDATE`. + +#### Materialization function + +```sql +CREATE OR REPLACE FUNCTION materialize_cml_stats_snapshot( + p_at_time TIMESTAMPTZ, -- must be hour-truncated by caller + p_user_id TEXT +) RETURNS INT -- number of rows upserted +LANGUAGE plpgsql SECURITY DEFINER +SET search_path = public +AS $$ +DECLARE + v_rows INT; +BEGIN + -- Validate caller is a known user + IF NOT EXISTS (SELECT 1 FROM cml_metadata WHERE user_id = p_user_id LIMIT 1) THEN + RAISE EXCEPTION 'Unknown user: %', p_user_id; + END IF; + + INSERT INTO cml_stats_history ( + snapshot_time, cml_id, user_id, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, + last_rsl, is_provisional + ) + SELECT + p_at_time, + m.cml_id, + p_user_id, + -- 6-hour window: sum the 6 hourly buckets ending at p_at_time + ROUND( + 100.0 * SUM(h6.rsl_count)::numeric + / NULLIF(SUM(h6.record_count), 0), + 2) AS completeness_percent_6h, + SUM(h6.record_count) AS total_records_6h, + SUM(h6.rsl_count) AS valid_records_6h, + ROUND(AVG(h6.rsl_avg)::numeric, 2) AS mean_rsl_6h, + ROUND(AVG(h6.rsl_stddev)::numeric, 2) AS stddev_rsl_6h, -- avg of hourly stddevs; good approximation + -- 1-hour window: the single bucket containing p_at_time + ROUND( + 100.0 * h1.rsl_count::numeric + / NULLIF(h1.record_count, 0), + 2) AS completeness_percent_1h, + ROUND(h1.rsl_avg::numeric, 2) AS mean_rsl_1h, + ROUND(h1.rsl_stddev::numeric, 2) AS stddev_rsl_1h, + h1.rsl_max AS last_rsl, -- best proxy from 1h agg + FALSE AS is_provisional -- definitive row from cml_data_1h + FROM cml_metadata m + -- 6h subquery from the 1h continuous aggregate + LEFT JOIN LATERAL ( + SELECT + SUM(record_count) AS record_count, + SUM(CASE WHEN rsl_max IS NOT NULL THEN record_count ELSE 0 END) AS rsl_count, + AVG(rsl_avg) AS rsl_avg, + AVG(rsl_max - rsl_min) AS rsl_stddev -- proxy; replace if stddev col added to agg + FROM cml_data_1h + WHERE user_id = p_user_id + AND cml_id = m.cml_id + AND bucket >= p_at_time - INTERVAL '6 hours' + AND bucket < p_at_time + ) h6 ON TRUE + -- 1h subquery: the bucket immediately before p_at_time + LEFT JOIN LATERAL ( + SELECT record_count, rsl_avg, rsl_max, rsl_min, + (rsl_max - rsl_min) AS rsl_stddev -- same proxy + FROM cml_data_1h + WHERE user_id = p_user_id + AND cml_id = m.cml_id + AND bucket = p_at_time - INTERVAL '1 hour' + ) h1 ON TRUE + WHERE m.user_id = p_user_id + ON CONFLICT (snapshot_time, cml_id, user_id) DO UPDATE SET + completeness_percent_6h = EXCLUDED.completeness_percent_6h, + total_records_6h = EXCLUDED.total_records_6h, + valid_records_6h = EXCLUDED.valid_records_6h, + mean_rsl_6h = EXCLUDED.mean_rsl_6h, + stddev_rsl_6h = EXCLUDED.stddev_rsl_6h, + completeness_percent_1h = EXCLUDED.completeness_percent_1h, + mean_rsl_1h = EXCLUDED.mean_rsl_1h, + stddev_rsl_1h = EXCLUDED.stddev_rsl_1h, + last_rsl = EXCLUDED.last_rsl, + is_provisional = FALSE; -- definitive row always clears the flag + + GET DIAGNOSTICS v_rows = ROW_COUNT; + RETURN v_rows; +END; +$$; +``` + +> **Note on `rsl_stddev` proxy:** `cml_data_1h` stores `rsl_min`/`rsl_max` but not +> `stddev`. `rsl_max - rsl_min` is a rough approximation. If more accuracy is needed, +> a follow-up can add a `stddev` column to `cml_data_1h` (requires dropping and +> re-creating the continuous aggregate). For map colouring purposes the proxy is +> sufficient. + +> **Note on `is_provisional`:** Rows written from `cml_data_1h` (completed calendar +> hours) have `is_provisional = FALSE`. Rows written every 60 s for the current +> (incomplete) hour from live `cml_stats` have `is_provisional = TRUE`. When the hour +> turns, `materialize_cml_stats_snapshot` overwrites the provisional row with a +> definitive one via the `ON CONFLICT … DO UPDATE` clause above, setting +> `is_provisional = FALSE`. + +#### Read function (for webserver / Grafana) + +```sql +CREATE OR REPLACE FUNCTION get_cml_stats_at( + p_at_time TIMESTAMPTZ, + p_user_id TEXT +) RETURNS TABLE ( + cml_id TEXT, + completeness_percent_6h REAL, + total_records_6h BIGINT, + valid_records_6h BIGINT, + mean_rsl_6h REAL, + stddev_rsl_6h REAL, + completeness_percent_1h REAL, + mean_rsl_1h REAL, + stddev_rsl_1h REAL, + last_rsl REAL, + is_provisional BOOLEAN +) LANGUAGE sql STABLE SECURITY DEFINER +SET search_path = public +AS $$ + SELECT + h.cml_id, + h.completeness_percent_6h, + h.total_records_6h, + h.valid_records_6h, + h.mean_rsl_6h, + h.stddev_rsl_6h, + h.completeness_percent_1h, + h.mean_rsl_1h, + h.stddev_rsl_1h, + h.last_rsl, + h.is_provisional + FROM cml_stats_history h + WHERE h.user_id = p_user_id + AND h.snapshot_time = date_trunc('hour', p_at_time) + ORDER BY h.cml_id; +$$; +``` + +#### Grants + +```sql +-- Writers (parser roles) +GRANT SELECT, INSERT, UPDATE ON cml_stats_history TO demo_openmrg, demo_orange_cameroun; +GRANT EXECUTE ON FUNCTION materialize_cml_stats_snapshot(TIMESTAMPTZ, TEXT) + TO demo_openmrg, demo_orange_cameroun; + +-- Readers (webserver, Grafana) +GRANT SELECT ON cml_stats_history TO webserver_role; +GRANT EXECUTE ON FUNCTION get_cml_stats_at(TIMESTAMPTZ, TEXT) + TO webserver_role, demo_openmrg, demo_orange_cameroun; +``` + +New users added via `generate_config.py` must receive the same grants — update the +template in `scripts/generate_config.py` (same pattern as the existing +`update_cml_stats_windowed` grant). + +### 1b. Parser: `write_stats_snapshot()` and `write_provisional_snapshot()` in `DBWriter` + +`parser/db_writer.py` — add two public methods: + +```python +def write_stats_snapshot(self, at_time: datetime) -> int: + """Materialize a cml_stats_history snapshot for the given hour. + + Calls materialize_cml_stats_snapshot(date_trunc('hour', at_time), user_id). + Returns the number of rows upserted. + """ + at_hour = at_time.replace(minute=0, second=0, microsecond=0) + cur = self.conn.cursor() + try: + cur.execute( + "SELECT materialize_cml_stats_snapshot(%s::timestamptz, %s)", + (at_hour, self.user_id), + ) + rows = cur.fetchone()[0] + self.conn.commit() + logger.info("Materialized cml_stats_history snapshot at %s (%d rows)", at_hour, rows) + return rows + except Exception: + try: + self.conn.rollback() + except Exception: + pass + logger.exception("Failed to write stats snapshot at %s", at_hour) + raise + finally: + if cur and not cur.closed: + cur.close() + +def write_provisional_snapshot(self) -> int: + """Copy live cml_stats rolling-window values into cml_stats_history for + the current (incomplete) calendar hour, marked is_provisional=TRUE. + + Runs every 60 s alongside refresh_windowed_stats so the slider can always + reach the current hour with data that is at most 60 s stale. The + ON CONFLICT DO UPDATE overwrites any previous provisional row for the same + hour. When the hour turns, write_stats_snapshot() writes a definitive row + (is_provisional=FALSE) that permanently replaces this one. + """ + from datetime import timezone + current_hour = datetime.now(tz=timezone.utc).replace(minute=0, second=0, microsecond=0) + cur = self.conn.cursor() + try: + cur.execute( + """ + INSERT INTO cml_stats_history ( + snapshot_time, cml_id, user_id, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, + last_rsl, is_provisional + ) + SELECT + %s, cml_id, user_id, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, + last_rsl, TRUE + FROM cml_stats + WHERE user_id = %s + ON CONFLICT (snapshot_time, cml_id, user_id) DO UPDATE SET + completeness_percent_6h = EXCLUDED.completeness_percent_6h, + total_records_6h = EXCLUDED.total_records_6h, + valid_records_6h = EXCLUDED.valid_records_6h, + mean_rsl_6h = EXCLUDED.mean_rsl_6h, + stddev_rsl_6h = EXCLUDED.stddev_rsl_6h, + completeness_percent_1h = EXCLUDED.completeness_percent_1h, + mean_rsl_1h = EXCLUDED.mean_rsl_1h, + stddev_rsl_1h = EXCLUDED.stddev_rsl_1h, + last_rsl = EXCLUDED.last_rsl, + is_provisional = TRUE -- keep provisional until definitive row arrives + WHERE cml_stats_history.is_provisional = TRUE -- never overwrite a definitive row + """, + (current_hour, self.user_id), + ) + rows = cur.rowcount + self.conn.commit() + logger.debug("Wrote provisional snapshot at %s (%d rows)", current_hour, rows) + return rows + except Exception: + try: + self.conn.rollback() + except Exception: + pass + logger.exception("Failed to write provisional stats snapshot") + raise + finally: + if cur and not cur.closed: + cur.close() +``` + +### 1c. Parser: wire both snapshot methods into the stats background thread + +`parser/entrypoints/sftp_push.py` — the `stats_loop` already fires every +`STATS_REFRESH_INTERVAL` seconds (default 60 s). Two things happen on every tick: + +1. `write_provisional_snapshot()` — always runs, cheaply copying `cml_stats` rows into + `cml_stats_history` for the current hour. The `WHERE is_provisional = TRUE` guard + prevents overwriting a definitive row that was written in the same minute. +2. `write_stats_snapshot(now)` — runs only when the calendar hour turns, materialising + a definitive row for the just-completed hour from `cml_data_1h`. + +```python +# Inside stats_loop, after refresh_windowed_stats(): +from datetime import timezone +now = datetime.now(tz=timezone.utc) +current_hour = now.replace(minute=0, second=0, microsecond=0) + +# Always: keep the current-hour provisional snapshot fresh (at most 60 s stale) +stats_db.write_provisional_snapshot() + +# Once per hour: materialise the just-completed hour from cml_data_1h +if current_hour != last_snapshot_hour: + stats_db.write_stats_snapshot(now) + last_snapshot_hour = current_hour +``` + +`last_snapshot_hour` is initialised to `None` so the first tick always materialises a +definitive snapshot for the current hour-boundary on startup. + +### 1d. Backfill script + +A standalone script `parser/backfill_stats_history.py` (or a CLI flag +`--backfill-stats-history` on `parse_netcdf_archive.py`) that: + +1. Queries `SELECT DISTINCT date_trunc('hour', bucket) FROM cml_data_1h WHERE user_id = ?` to find all available hours. +2. For each hour (oldest-first), calls `write_stats_snapshot(hour)`. +3. Logs progress every 100 hours. + +Run once after the migration is applied. + +--- + +## PR 2 — Webserver + Frontend: time slider and Grafana sync + +Depends on PR 1 being merged (migration 017 applied). + +### 2a. Webserver: optional `?at=` parameter on `/api/cml-stats` + +`webserver/main.py` — extend `api_cml_stats()`: + +```python +@app.route("/api/cml-stats") +@login_required +def api_cml_stats(): + at_param = request.args.get("at") # ISO 8601 string or absent + + try: + with user_db_scope(current_user.id) as conn: + cur = conn.cursor() + + if at_param: + # Historical: parse the timestamp and query cml_stats_history + try: + at_ts = datetime.fromisoformat(at_param.replace("Z", "+00:00")) + except ValueError: + return jsonify({"error": "invalid 'at' parameter"}), 400 + + cur.execute( + "SELECT * FROM get_cml_stats_at(%s::timestamptz, %s)", + (at_ts, current_user.id), + ) + else: + # Live: existing query from cml_stats (windowed, pre-computed) + cur.execute( + """ + SELECT cml_id::text, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, stddev_rsl_1h, last_rsl + FROM cml_stats + ORDER BY cml_id + """ + ) + + data = cur.fetchall() + cur.close() + + # Response shape is identical in both branches; is_provisional only present for ?at= path + stats = [ + { + "cml_id": str(row[0]), + "completeness_percent": safe_float(row[1]), + "total_records": int(row[2] or 0), + "valid_records": int(row[3] or 0), + "mean_rsl": safe_float(row[4]), + "stddev_rsl": safe_float(row[5]), + "completeness_percent_1h": safe_float(row[6]), + "stddev_last_60min": safe_float(row[7]), + "last_rsl": safe_float(row[8]), + "is_provisional": bool(row[9]) if at_param and len(row) > 9 else False, + } + for row in data + ] + return jsonify(stats) + except Exception as e: + print(f"Error fetching CML stats: {e}") + return jsonify([]) +``` + +Add a `/api/cml-stats-time-range` endpoint so the frontend can configure the slider's +min/max. The `max` is always the current server hour (a provisional snapshot for it +is always present); only `min` requires a DB query: + +```python +@app.route("/api/cml-stats-time-range") +@login_required +def api_cml_stats_time_range(): + from datetime import timezone + now = datetime.now(tz=timezone.utc) + current_hour = now.replace(minute=0, second=0, microsecond=0) + try: + with user_db_scope(current_user.id) as conn: + cur = conn.cursor() + cur.execute( + "SELECT MIN(snapshot_time) FROM cml_stats_history WHERE user_id = %s", + (current_user.id,), + ) + row = cur.fetchone() + cur.close() + min_ts = row[0].isoformat() if row and row[0] else None + return jsonify({"min": min_ts, "max": current_hour.isoformat()}) + except Exception as e: + print(f"Error fetching stats time range: {e}") + return jsonify({"min": None, "max": current_hour.isoformat()}) +``` + +### 2b. Frontend: time slider in `realtime.html` + +#### State variables (add to the existing JS block) + +```js +let selectedTime = null; // null = live mode +let statsRefreshTimer = null; // auto-refresh handle +const LIVE_REFRESH_MS = 30000; +``` + +#### Replace the existing one-shot `/api/cml-stats` fetch with a function + +```js +function fetchAndApplyStats(atIso) { + var url = '/api/cml-stats'; + if (atIso) url += '?at=' + encodeURIComponent(atIso); + fetch(url) + .then(r => r.json()) + .then(stats => { + // Surface provisional state in the time display label + var anyProvisional = stats.some(function(s) { return s.is_provisional; }); + var disp = document.getElementById('timeDisplay'); + if (atIso && anyProvisional && disp && !disp.textContent.includes('(partial)')) { + disp.textContent += ' (partial)'; + } + stats.forEach(function(stat) { + cmlStats[stat.cml_id] = stat; + applyStatsToLine(stat.cml_id, stat); + }); + }) + .catch(function(err) { console.error('Stats fetch error:', err); }); +} +``` + +#### Extend the `ColorControl` Leaflet control with a time slider section + +```js +// Inside ColorControl.onAdd(), after the existing "Color by:" select: + +var divider = L.DomUtil.create('hr', '', container); +divider.style.margin = '8px 0'; + +var timeLabel = L.DomUtil.create('div', '', container); +timeLabel.innerHTML = 'Time:'; +timeLabel.style.marginBottom = '4px'; + +var liveBtn = L.DomUtil.create('button', '', container); +liveBtn.id = 'timeLiveBtn'; +liveBtn.textContent = '● Live'; +liveBtn.style.cssText = 'width:100%;margin-bottom:4px;background:#22c55e;color:white;border:none;border-radius:4px;padding:3px 6px;cursor:pointer;'; + +var timeSlider = L.DomUtil.create('input', '', container); +timeSlider.type = 'range'; +timeSlider.id = 'timeSlider'; +timeSlider.style.cssText = 'width:100%;margin:4px 0;'; +timeSlider.disabled = true; // enabled once time range is loaded + +var timeDisplay = L.DomUtil.create('div', '', container); +timeDisplay.id = 'timeDisplay'; +timeDisplay.style.cssText = 'font-size:11px;text-align:center;color:#555;'; +timeDisplay.textContent = 'Loading range…'; + +L.DomEvent.disableClickPropagation(container); +L.DomEvent.disableScrollPropagation(container); +``` + +#### Time range loader (call from `initializeMap()` after `loadCmlData()`) + +```js +function initTimeSlider() { + fetch('/api/cml-stats-time-range') + .then(r => r.json()) + .then(range => { + if (!range.min || !range.max) { + document.getElementById('timeDisplay').textContent = 'No history'; + return; + } + var minEpoch = Math.floor(new Date(range.min).getTime() / 1000); + var maxEpoch = Math.floor(new Date(range.max).getTime() / 1000); + var slider = document.getElementById('timeSlider'); + slider.min = minEpoch; + slider.max = maxEpoch; + slider.step = 3600; // 1-hour steps + slider.value = maxEpoch; + slider.disabled = false; + document.getElementById('timeDisplay').textContent = 'Live'; + }) + .catch(() => { + document.getElementById('timeDisplay').textContent = 'Unavailable'; + }); +} +``` + +#### Slider event handlers + +```js +// Live button +document.getElementById('timeLiveBtn').addEventListener('click', function () { + selectedTime = null; + document.getElementById('timeDisplay').textContent = 'Live'; + this.style.background = '#22c55e'; + clearTimeout(statsRefreshTimer); + fetchAndApplyStats(null); + startLiveRefresh(); + syncGrafanaTime(null); +}); + +// Slider drag (debounced 300 ms) +var sliderDebounce; +document.getElementById('timeSlider').addEventListener('input', function () { + clearTimeout(sliderDebounce); + var epochSec = parseInt(this.value, 10); + var dt = new Date(epochSec * 1000); + // Check whether this is the current (partial) hour + var nowHour = new Date(); nowHour.setMinutes(0, 0, 0, 0); + var sliderHour = new Date(dt); sliderHour.setMinutes(0, 0, 0, 0); + var label = dt.toISOString().replace('T', ' ').slice(0, 16) + ' UTC'; + if (sliderHour.getTime() >= nowHour.getTime()) label += ' (partial)'; + document.getElementById('timeDisplay').textContent = label; + sliderDebounce = setTimeout(function () { + selectedTime = dt.toISOString(); + stopLiveRefresh(); + document.getElementById('timeLiveBtn').style.background = '#aaa'; + fetchAndApplyStats(selectedTime); // provisional snapshot handles current hour + syncGrafanaTime(epochSec); + }, 300); +}); + +function startLiveRefresh() { + stopLiveRefresh(); + statsRefreshTimer = setInterval(function () { fetchAndApplyStats(null); }, LIVE_REFRESH_MS); +} +function stopLiveRefresh() { + if (statsRefreshTimer) clearInterval(statsRefreshTimer); + statsRefreshTimer = null; +} +``` + +Start live refresh at the end of `initializeMap()` (after `loadCmlData()`): + +```js +fetchAndApplyStats(null); +startLiveRefresh(); +initTimeSlider(); +``` + +### 2c. Grafana iframe sync + +```js +function syncGrafanaTime(epochSec) { + var grafanaPanel = document.getElementById('grafana-panel'); + var iframeWindow = grafanaPanel.contentWindow; + try { + var currentUrl = new URL(iframeWindow.location.href); + if (epochSec === null) { + // Live: restore relative time range + currentUrl.searchParams.set('from', 'now-1h'); + currentUrl.searchParams.set('to', 'now'); + currentUrl.searchParams.delete('var-marker_time'); + } else { + // Historical: ±30 min window around selected second + var halfWindow = 30 * 60 * 1000; // 30 min in ms + var epochMs = epochSec * 1000; + currentUrl.searchParams.set('from', String(epochMs - halfWindow)); + currentUrl.searchParams.set('to', String(epochMs + halfWindow)); + currentUrl.searchParams.set('var-marker_time', String(epochMs)); + } + iframeWindow.history.pushState(null, '', currentUrl.toString()); + iframeWindow.dispatchEvent(new PopStateEvent('popstate', { state: null })); + } catch (e) { + console.warn('Grafana sync skipped (iframe not ready):', e); + } +} +``` + +### 2d. Grafana dashboard: `$marker_time` variable + annotation + +Edit `grafana/provisioning/dashboards/definitions/cml-realtime.json`: + +1. **Add template variable** (`templating.list`): +```json +{ + "name": "marker_time", + "type": "textbox", + "label": "Marker time (epoch ms)", + "hide": 2, + "current": { "value": "" } +} +``` + +2. **Add annotation** (`annotations.list`): +```json +{ + "name": "Selected time", + "enable": true, + "hide": false, + "iconColor": "red", + "type": "dashboard", + "builtIn": 0, + "datasource": { "type": "grafana", "uid": "-- Grafana --" }, + "rawQuery": true, + "query": "SELECT $marker_time AS time, 'selected' AS text", + "timeField": "time" +} +``` + +> Grafana 13's built-in "Grafana" datasource can evaluate a constant timestamp as an +> annotation without a DB query. If the variable is empty the annotation is silently +> suppressed. + +--- + +## Testing notes + +- **Unit tests for `write_stats_snapshot`** follow the same pattern as + `test_refresh_windowed_stats_commits_on_success` in + `parser/tests/test_db_writer.py`. +- **Unit test for `api_cml_stats` with `?at=`** follows the pattern in + `webserver/tests/test_api_cml_stats.py` — mock `user_db_scope`, assert the SQL + contains `get_cml_stats_at`. +- **Integration smoke test**: run `psql … -c "SELECT materialize_cml_stats_snapshot(date_trunc('hour', now()), 'demo_openmrg')"` and confirm rows appear in `cml_stats_history`. + +## Open questions / future work + +- Add `rsl_stddev` as a proper column to `cml_data_1h` (requires dropping and + re-creating the continuous aggregate) to replace the `rsl_max - rsl_min` proxy. +- Consider adding `@grafana/iframe-api` bidirectional sync (Grafana 13 native) to + make the slider follow when the user zooms inside Grafana — not required for the + initial feature but a natural follow-up. +- Retention policy for `cml_stats_history`: the compression policy (7 days) keeps + storage small; consider an explicit `drop_chunks` policy if long-term history is + not needed. diff --git a/parser/tests/test_stats_snapshot.py b/parser/tests/test_stats_snapshot.py new file mode 100644 index 0000000..a49179b --- /dev/null +++ b/parser/tests/test_stats_snapshot.py @@ -0,0 +1,124 @@ +"""Tests for cml_stats_history snapshot functionality.""" + +import pytest +from unittest.mock import Mock, patch +from datetime import datetime, timezone + + +# Skip all tests if psycopg2 not available +psycopg2 = pytest.importorskip("psycopg2", reason="psycopg2 not installed") + +from ..db_writer import DBWriter + + +@pytest.fixture +def mock_connection(): + """Mock psycopg2 connection.""" + conn = Mock() + conn.closed = False + cursor = Mock() + conn.cursor.return_value = cursor + cursor.fetchone.return_value = (42,) # Mock row count + cursor.rowcount = 42 + return conn + + +def test_write_stats_snapshot_success(mock_connection): + """Test write_stats_snapshot materializes snapshot correctly.""" + writer = DBWriter("postgresql://test", user_id="demo_openmrg") + writer.conn = mock_connection + + at_time = datetime(2026, 7, 25, 14, 30, 0, tzinfo=timezone.utc) + rows = writer.write_stats_snapshot(at_time) + + assert rows == 42 + mock_connection.cursor.assert_called_once() + cursor = mock_connection.cursor.return_value + cursor.execute.assert_called_once() + + # Verify the call uses hour-truncated time + call_args = cursor.execute.call_args[0] + sql = call_args[0] + params = call_args[1] + + assert "materialize_cml_stats_snapshot" in sql + assert params[1] == "demo_openmrg" + # Check that time is truncated to hour + assert params[0].minute == 0 + assert params[0].second == 0 + assert params[0].microsecond == 0 + + +def test_write_stats_snapshot_rollback_on_error(mock_connection): + """Test write_stats_snapshot rolls back on error.""" + writer = DBWriter("postgresql://test", user_id="demo_openmrg") + writer.conn = mock_connection + + mock_connection.cursor.return_value.execute.side_effect = Exception("DB error") + + with pytest.raises(Exception, match="DB error"): + writer.write_stats_snapshot(datetime.now(tz=timezone.utc)) + + mock_connection.rollback.assert_called_once() + + +def test_write_provisional_snapshot_success(mock_connection): + """Test write_provisional_snapshot writes current hour snapshot.""" + writer = DBWriter("postgresql://test", user_id="demo_openmrg") + writer.conn = mock_connection + + with patch('parser.db_writer.datetime') as mock_datetime: + mock_now = datetime(2026, 7, 25, 14, 30, 0, tzinfo=timezone.utc) + mock_datetime.now.return_value = mock_now + mock_datetime.timezone = timezone + + rows = writer.write_provisional_snapshot() + + assert rows == 42 + cursor = mock_connection.cursor.return_value + cursor.execute.assert_called_once() + + call_args = cursor.execute.call_args[0] + sql = call_args[0] + params = call_args[1] + + assert "INSERT INTO cml_stats_history" in sql + assert "is_provisional" in sql + assert params[1] == "demo_openmrg" + # Check that snapshot_time is hour-truncated + assert params[0].minute == 0 + assert params[0].second == 0 + + +def test_write_provisional_snapshot_uses_cml_stats_table(mock_connection): + """Test provisional snapshot copies from cml_stats table.""" + writer = DBWriter("postgresql://test", user_id="demo_openmrg") + writer.conn = mock_connection + + writer.write_provisional_snapshot() + + cursor = mock_connection.cursor.return_value + sql = cursor.execute.call_args[0][0] + + # Should select from cml_stats + assert "FROM cml_stats" in sql + # Should have ON CONFLICT clause + assert "ON CONFLICT" in sql + # Should set is_provisional=TRUE + assert "is_provisional = TRUE" in sql + + +def test_write_stats_snapshot_with_naive_datetime(mock_connection): + """Test write_stats_snapshot handles naive datetime by making it UTC.""" + writer = DBWriter("postgresql://test", user_id="demo_openmrg") + writer.conn = mock_connection + + # Naive datetime (no timezone) + at_time = datetime(2026, 7, 25, 14, 30, 0) + rows = writer.write_stats_snapshot(at_time) + + assert rows == 42 + cursor = mock_connection.cursor.return_value + params = cursor.execute.call_args[0][1] + # Should be converted to UTC + assert params[0].tzinfo == timezone.utc diff --git a/webserver/tests/test_time_slider_api.py b/webserver/tests/test_time_slider_api.py new file mode 100644 index 0000000..cd3cbc3 --- /dev/null +++ b/webserver/tests/test_time_slider_api.py @@ -0,0 +1,171 @@ +"""Tests for historical time slider API endpoints.""" + +import sys +from contextlib import contextmanager +from datetime import datetime, timezone +from unittest.mock import Mock + +import pytest + + +# Ensure optional heavy imports won't fail at import time +sys.modules.setdefault("folium", Mock()) +sys.modules.setdefault("requests", Mock()) + + +def test_api_cml_stats_historical_with_at_parameter(monkeypatch): + """Test /api/cml-stats with ?at= parameter queries cml_stats_history.""" + import os + sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + import main as wm + + # Prepare mock DB connection and cursor + mock_conn = Mock() + mock_cursor = Mock() + mock_conn.cursor.return_value = mock_cursor + + # Historical query returns 10 columns including is_provisional + mock_cursor.fetchall.return_value = [ + ( + "10001", + 94.2, # completeness_percent_6h + 2160, # total_records_6h + 2031, # valid_records_6h + -50.0, # mean_rsl_6h + 3.0, # stddev_rsl_6h + 90.0, # completeness_percent_1h + 1.3, # stddev_rsl_1h + -45.0, # last_rsl + False, # is_provisional + ) + ] + + mock_cursor.close = Mock() + mock_conn.close = Mock() + + @contextmanager + def mock_user_db_scope(user_id): + yield mock_conn + + mock_user = Mock() + mock_user.id = "demo_openmrg" + + monkeypatch.setattr(wm, "user_db_scope", mock_user_db_scope) + monkeypatch.setattr(wm, "current_user", mock_user) + monkeypatch.setitem(wm.app.config, "LOGIN_DISABLED", True) + + client = wm.app.test_client() + + # Test with historical timestamp + resp = client.get("/api/cml-stats?at=2026-07-25T14:00:00Z") + assert resp.status_code == 200 + data = resp.get_json() + assert isinstance(data, list) + assert len(data) == 1 + row = data[0] + assert row["cml_id"] == "10001" + assert row["is_provisional"] == False + + # Verify get_cml_stats_at function was called + cursor = mock_conn.cursor.return_value + sql = cursor.execute.call_args[0][0] + assert "get_cml_stats_at" in sql + + +def test_api_cml_stats_invalid_at_parameter(monkeypatch): + """Test /api/cml-stats with invalid ?at= parameter returns 400.""" + import os + sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + import main as wm + + mock_user = Mock() + mock_user.id = "demo_openmrg" + + monkeypatch.setattr(wm, "current_user", mock_user) + monkeypatch.setitem(wm.app.config, "LOGIN_DISABLED", True) + + client = wm.app.test_client() + + # Invalid timestamp format + resp = client.get("/api/cml-stats?at=invalid-timestamp") + assert resp.status_code == 400 + data = resp.get_json() + assert "error" in data + + +def test_api_cml_stats_time_range_endpoint(monkeypatch): + """Test /api/cml-stats-time-range endpoint.""" + import os + sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + import main as wm + + mock_conn = Mock() + mock_cursor = Mock() + mock_conn.cursor.return_value = mock_cursor + + # Return minimum snapshot time from history + min_time = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + mock_cursor.fetchone.return_value = (min_time,) + mock_cursor.close = Mock() + mock_conn.close = Mock() + + @contextmanager + def mock_user_db_scope(user_id): + yield mock_conn + + mock_user = Mock() + mock_user.id = "demo_openmrg" + + monkeypatch.setattr(wm, "user_db_scope", mock_user_db_scope) + monkeypatch.setattr(wm, "current_user", mock_user) + monkeypatch.setitem(wm.app.config, "LOGIN_DISABLED", True) + + client = wm.app.test_client() + resp = client.get("/api/cml-stats-time-range") + + assert resp.status_code == 200 + data = resp.get_json() + assert "min" in data + assert "max" in data + assert data["min"] is not None + + # Verify query uses cml_stats_history + cursor = mock_conn.cursor.return_value + sql = cursor.execute.call_args[0][0] + assert "cml_stats_history" in sql + assert "MIN(snapshot_time)" in sql + + +def test_api_cml_stats_time_range_no_history(monkeypatch): + """Test /api/cml-stats-time-range when no history exists.""" + import os + sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + import main as wm + + mock_conn = Mock() + mock_cursor = Mock() + mock_conn.cursor.return_value = mock_cursor + + # No history available + mock_cursor.fetchone.return_value = (None,) + mock_cursor.close = Mock() + + @contextmanager + def mock_user_db_scope(user_id): + yield mock_conn + + mock_user = Mock() + mock_user.id = "demo_openmrg" + + monkeypatch.setattr(wm, "user_db_scope", mock_user_db_scope) + monkeypatch.setattr(wm, "current_user", mock_user) + monkeypatch.setitem(wm.app.config, "LOGIN_DISABLED", True) + + client = wm.app.test_client() + resp = client.get("/api/cml-stats-time-range") + + assert resp.status_code == 200 + data = resp.get_json() + assert data["min"] is None + # max should still be current hour + assert data["max"] is not None From 01b67dd76e3e0934c686ccf94879cba27eae37d5 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 19:53:26 +0200 Subject: [PATCH 06/23] =?UTF-8?q?fix(db):=20renumber=20migration=20017=20?= =?UTF-8?q?=E2=86=92=20010=20for=20clean=20upstream=20sequencing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Upstream/main only has migrations up to 009. The deploy fork's ifu/main branch contains additional migrations 010-016 that are either deployment-specific (not intended for upstream) or generic performance improvements not yet submitted as upstream PRs. Neither set has SQL dependencies on cml_stats_history, so the correct next number for this feature in the upstream sequence is 010. Updated all migration number references in docs and backfill script. --- ...dd_cml_stats_history.sql => 010_add_cml_stats_history.sql} | 0 docs/plan-time-slider-2026-07-24.md | 4 ++-- parser/backfill_stats_history.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) rename database/migrations/{017_add_cml_stats_history.sql => 010_add_cml_stats_history.sql} (100%) diff --git a/database/migrations/017_add_cml_stats_history.sql b/database/migrations/010_add_cml_stats_history.sql similarity index 100% rename from database/migrations/017_add_cml_stats_history.sql rename to database/migrations/010_add_cml_stats_history.sql diff --git a/docs/plan-time-slider-2026-07-24.md b/docs/plan-time-slider-2026-07-24.md index 0e7a6fe..ab8ddc9 100644 --- a/docs/plan-time-slider-2026-07-24.md +++ b/docs/plan-time-slider-2026-07-24.md @@ -60,7 +60,7 @@ Key properties: ### 1a. New migration: `cml_stats_history` + helper functions -File: `database/migrations/017_add_cml_stats_history.sql` +File: `database/migrations/010_add_cml_stats_history.sql` #### Table @@ -399,7 +399,7 @@ Run once after the migration is applied. ## PR 2 — Webserver + Frontend: time slider and Grafana sync -Depends on PR 1 being merged (migration 017 applied). +Depends on PR 1 being merged (migration 010 applied). ### 2a. Webserver: optional `?at=` parameter on `/api/cml-stats` diff --git a/parser/backfill_stats_history.py b/parser/backfill_stats_history.py index ab1438c..59bbe54 100644 --- a/parser/backfill_stats_history.py +++ b/parser/backfill_stats_history.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Backfill cml_stats_history from existing cml_data_1h data. -Run once after applying migration 017 to populate historical snapshots. +Run once after applying migration 010 to populate historical snapshots. Usage: docker compose exec parser python -m parser.backfill_stats_history From 3de23a8ccc72a051664c49dbdb1425bb736d3dae Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 22:23:26 +0200 Subject: [PATCH 07/23] fix(db): renumber migration 010 -> 014 for cml_stats_history perf/db-improvements-batch now claims migration numbers 010-013 upstream, so cml_stats_history must move to 014 to avoid a collision once both PRs are merged. --- ...dd_cml_stats_history.sql => 014_add_cml_stats_history.sql} | 4 ++-- docs/plan-time-slider-2026-07-24.md | 4 ++-- parser/backfill_stats_history.py | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) rename database/migrations/{010_add_cml_stats_history.sql => 014_add_cml_stats_history.sql} (98%) diff --git a/database/migrations/010_add_cml_stats_history.sql b/database/migrations/014_add_cml_stats_history.sql similarity index 98% rename from database/migrations/010_add_cml_stats_history.sql rename to database/migrations/014_add_cml_stats_history.sql index 4164174..7eec1b5 100644 --- a/database/migrations/010_add_cml_stats_history.sql +++ b/database/migrations/014_add_cml_stats_history.sql @@ -1,4 +1,4 @@ --- Migration 017: Add cml_stats_history hypertable for historical time slider +-- Migration 014: Add cml_stats_history hypertable for historical time slider -- -- This migration adds a new hypertable to store snapshots of CML stats at hourly intervals. -- Enables the historical time slider feature on the realtime map. @@ -11,7 +11,7 @@ -- -- Apply with: -- docker compose exec -T database psql -U myuser -d mydatabase \ --- < database/migrations/017_add_cml_stats_history.sql +-- < database/migrations/014_add_cml_stats_history.sql -- Create the cml_stats_history table CREATE TABLE IF NOT EXISTS cml_stats_history ( diff --git a/docs/plan-time-slider-2026-07-24.md b/docs/plan-time-slider-2026-07-24.md index ab8ddc9..993753f 100644 --- a/docs/plan-time-slider-2026-07-24.md +++ b/docs/plan-time-slider-2026-07-24.md @@ -60,7 +60,7 @@ Key properties: ### 1a. New migration: `cml_stats_history` + helper functions -File: `database/migrations/010_add_cml_stats_history.sql` +File: `database/migrations/014_add_cml_stats_history.sql` #### Table @@ -399,7 +399,7 @@ Run once after the migration is applied. ## PR 2 — Webserver + Frontend: time slider and Grafana sync -Depends on PR 1 being merged (migration 010 applied). +Depends on PR 1 being merged (migration 014 applied). ### 2a. Webserver: optional `?at=` parameter on `/api/cml-stats` diff --git a/parser/backfill_stats_history.py b/parser/backfill_stats_history.py index 59bbe54..adac07f 100644 --- a/parser/backfill_stats_history.py +++ b/parser/backfill_stats_history.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Backfill cml_stats_history from existing cml_data_1h data. -Run once after applying migration 010 to populate historical snapshots. +Run once after applying migration 014 to populate historical snapshots. Usage: docker compose exec parser python -m parser.backfill_stats_history From e66ca2a180d61122929ffbeca040cab73b4217dd Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 23:21:22 +0200 Subject: [PATCH 08/23] fix(db): add record_count prerequisite migration; fix stats function bugs migration 014 adds record_count to the cml_data_1h continuous aggregate, a column materialize_cml_stats_snapshot() (015, renumbered from 014) has always depended on but which never existed anywhere (neither upstream nor in the original plan's schema). While testing against a throwaway DB, found and fixed two further bugs in materialize_cml_stats_snapshot() that only surfaced once record_count existed: - the 1h LATERAL subquery never computed rsl_count (valid-record count), unlike the 6h subquery - the outer query wrapped the 6h LATERAL subquery's already-aggregated columns in a second layer of SUM/AVG with no GROUP BY, which is invalid SQL (the LATERAL join already produces one aggregated row per cml_id) Verified end-to-end against a throwaway container: inserted sample cml_data, refreshed cml_data_1h, called materialize_cml_stats_snapshot() and get_cml_stats_at(), confirmed correct completeness percentages and record counts. --- .../014_add_record_count_to_cml_data_1h.sql | 68 +++++++++++++++++++ ...tory.sql => 015_add_cml_stats_history.sql} | 22 +++--- docs/plan-time-slider-2026-07-24.md | 4 +- parser/backfill_stats_history.py | 2 +- 4 files changed, 85 insertions(+), 11 deletions(-) create mode 100644 database/migrations/014_add_record_count_to_cml_data_1h.sql rename database/migrations/{014_add_cml_stats_history.sql => 015_add_cml_stats_history.sql} (88%) diff --git a/database/migrations/014_add_record_count_to_cml_data_1h.sql b/database/migrations/014_add_record_count_to_cml_data_1h.sql new file mode 100644 index 0000000..76c60d5 --- /dev/null +++ b/database/migrations/014_add_record_count_to_cml_data_1h.sql @@ -0,0 +1,68 @@ +-- Migration 014: Add record_count to cml_data_1h continuous aggregate +-- +-- Prerequisite for the historical time slider (see the following +-- cml_stats_history migration). materialize_cml_stats_snapshot() computes +-- completeness_percent_6h / completeness_percent_1h from the number of raw +-- samples in each hourly bucket, which requires a record_count column that +-- cml_data_1h has never had (it was only ever needed for Grafana's raw +-- RSL/TSL line charts, which use MIN/MAX/AVG, not counts). +-- +-- Continuous aggregates cannot have a column added via ALTER; the view must +-- be dropped and recreated. This is non-destructive: TimescaleDB +-- re-materialises from the underlying raw cml_data, matching the pattern +-- already used in migration 003_update_aggregate_user_id.sql. +-- +-- Dropping cml_data_1h CASCADEs to its continuous aggregate policy and to +-- the cml_data_1h_secure security-barrier view (and that view's grants), so +-- all three must be recreated here. +-- +-- A brief gap in Grafana's hourly-aggregate data is expected while the +-- refresh policy backfills the view (~1 refresh cycle, up to 1 hour). +-- Queries that fall in the gap automatically fall through to raw cml_data. +-- +-- Apply with: +-- docker compose exec -T database psql -U myuser -d mydatabase \ +-- < database/migrations/014_add_record_count_to_cml_data_1h.sql + +-- Step 1: Remove the old view and its dependent policy, secure view, and grants. +DROP MATERIALIZED VIEW IF EXISTS cml_data_1h CASCADE; + +-- Step 2: Recreate with record_count added. +CREATE MATERIALIZED VIEW cml_data_1h +WITH (timescaledb.continuous) AS +SELECT + time_bucket('1 hour', time) AS bucket, + user_id, + cml_id, + sublink_id, + MIN(rsl) AS rsl_min, + MAX(rsl) AS rsl_max, + AVG(rsl) AS rsl_avg, + MIN(tsl) AS tsl_min, + MAX(tsl) AS tsl_max, + AVG(tsl) AS tsl_avg, + COUNT(*) AS record_count +FROM cml_data +GROUP BY bucket, user_id, cml_id, sublink_id +WITH NO DATA; + +-- Step 3: Restore the refresh policy (same parameters as before). +SELECT add_continuous_aggregate_policy('cml_data_1h', + start_offset => INTERVAL '2 days', + end_offset => INTERVAL '1 hour', + schedule_interval => INTERVAL '1 hour' +); + +-- Step 4: Recreate the security-barrier view dropped by CASCADE above. +CREATE VIEW cml_data_1h_secure WITH (security_barrier) AS +SELECT * FROM cml_data_1h +WHERE user_id = current_user; + +-- Step 5: Restore grants dropped by CASCADE above. +GRANT SELECT ON cml_data_1h_secure TO demo_openmrg, demo_orange_cameroun; +GRANT SELECT ON cml_data_1h TO webserver_role; +GRANT SELECT ON cml_data_1h_secure TO webserver_role; + +-- Step 6: Optional — trigger an immediate backfill rather than waiting for +-- the next scheduled refresh. Remove the leading '--' to enable. +-- CALL refresh_continuous_aggregate('cml_data_1h', NULL, NULL); diff --git a/database/migrations/014_add_cml_stats_history.sql b/database/migrations/015_add_cml_stats_history.sql similarity index 88% rename from database/migrations/014_add_cml_stats_history.sql rename to database/migrations/015_add_cml_stats_history.sql index 7eec1b5..a270033 100644 --- a/database/migrations/014_add_cml_stats_history.sql +++ b/database/migrations/015_add_cml_stats_history.sql @@ -1,4 +1,4 @@ --- Migration 014: Add cml_stats_history hypertable for historical time slider +-- Migration 015: Add cml_stats_history hypertable for historical time slider -- -- This migration adds a new hypertable to store snapshots of CML stats at hourly intervals. -- Enables the historical time slider feature on the realtime map. @@ -9,9 +9,12 @@ -- - Compression policy for old data (>7 days) -- - Functions for materialization and querying -- +-- Depends on migration 014_add_record_count_to_cml_data_1h.sql, which adds +-- the record_count column that materialize_cml_stats_snapshot() reads below. +-- -- Apply with: -- docker compose exec -T database psql -U myuser -d mydatabase \ --- < database/migrations/014_add_cml_stats_history.sql +-- < database/migrations/015_add_cml_stats_history.sql -- Create the cml_stats_history table CREATE TABLE IF NOT EXISTS cml_stats_history ( @@ -74,14 +77,16 @@ BEGIN m.cml_id, p_user_id, -- 6-hour window: sum the 6 hourly buckets ending at p_at_time + -- (h6 already aggregates across those buckets internally, so no + -- further SUM/AVG or GROUP BY is needed at this level) ROUND( - 100.0 * SUM(h6.rsl_count)::numeric - / NULLIF(SUM(h6.record_count), 0), + 100.0 * h6.rsl_count::numeric + / NULLIF(h6.record_count, 0), 2) AS completeness_percent_6h, - SUM(h6.record_count) AS total_records_6h, - SUM(h6.rsl_count) AS valid_records_6h, - ROUND(AVG(h6.rsl_avg)::numeric, 2) AS mean_rsl_6h, - ROUND(AVG(h6.rsl_stddev)::numeric, 2) AS stddev_rsl_6h, + h6.record_count AS total_records_6h, + h6.rsl_count AS valid_records_6h, + ROUND(h6.rsl_avg::numeric, 2) AS mean_rsl_6h, + ROUND(h6.rsl_stddev::numeric, 2) AS stddev_rsl_6h, -- 1-hour window: the single bucket containing p_at_time ROUND( 100.0 * h1.rsl_count::numeric @@ -108,6 +113,7 @@ BEGIN -- 1h subquery: the bucket immediately before p_at_time LEFT JOIN LATERAL ( SELECT record_count, rsl_avg, rsl_max, rsl_min, + CASE WHEN rsl_max IS NOT NULL THEN record_count ELSE 0 END AS rsl_count, (rsl_max - rsl_min) AS rsl_stddev FROM cml_data_1h WHERE user_id = p_user_id diff --git a/docs/plan-time-slider-2026-07-24.md b/docs/plan-time-slider-2026-07-24.md index 993753f..a39e40e 100644 --- a/docs/plan-time-slider-2026-07-24.md +++ b/docs/plan-time-slider-2026-07-24.md @@ -60,7 +60,7 @@ Key properties: ### 1a. New migration: `cml_stats_history` + helper functions -File: `database/migrations/014_add_cml_stats_history.sql` +File: `database/migrations/015_add_cml_stats_history.sql` #### Table @@ -399,7 +399,7 @@ Run once after the migration is applied. ## PR 2 — Webserver + Frontend: time slider and Grafana sync -Depends on PR 1 being merged (migration 014 applied). +Depends on PR 1 being merged (migration 015 applied). ### 2a. Webserver: optional `?at=` parameter on `/api/cml-stats` diff --git a/parser/backfill_stats_history.py b/parser/backfill_stats_history.py index adac07f..1dbead2 100644 --- a/parser/backfill_stats_history.py +++ b/parser/backfill_stats_history.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Backfill cml_stats_history from existing cml_data_1h data. -Run once after applying migration 014 to populate historical snapshots. +Run once after applying migration 015 to populate historical snapshots. Usage: docker compose exec parser python -m parser.backfill_stats_history From 195281148ffc809b77379940ed9aba17e44f5ec1 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 23:21:35 +0200 Subject: [PATCH 09/23] fix(db): fold record_count and cml_stats_history into init.sql A fresh DB (docker compose up with an empty volume) only runs init.sql, never the migrations/ files. Folds in migrations 014 (record_count on cml_data_1h) and 015 (cml_stats_history table + materialize_cml_stats_snapshot/get_cml_stats_at) so a fresh install matches a fully-migrated one and the time slider feature works out of the box. Verified by initializing a throwaway container from a clean volume, inserting sample cml_data, and confirming materialize_cml_stats_snapshot() and get_cml_stats_at() return correct results. --- database/init.sql | 187 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 181 insertions(+), 6 deletions(-) diff --git a/database/init.sql b/database/init.sql index 847e1fe..055e2e8 100644 --- a/database/init.sql +++ b/database/init.sql @@ -278,6 +278,8 @@ CREATE INDEX IF NOT EXISTS idx_cml_data_covering -- requested time range exceeds 3 days, reducing the scanned row count -- by ~360x (10-second raw data → 1-hour buckets). -- --------------------------------------------------------------------------- +-- record_count (migration 014) is used by materialize_cml_stats_snapshot() +-- below to compute completeness percentages for the historical time slider. CREATE MATERIALIZED VIEW cml_data_1h WITH (timescaledb.continuous) AS SELECT @@ -285,12 +287,13 @@ SELECT user_id, cml_id, sublink_id, - MIN(rsl) AS rsl_min, - MAX(rsl) AS rsl_max, - AVG(rsl) AS rsl_avg, - MIN(tsl) AS tsl_min, - MAX(tsl) AS tsl_max, - AVG(tsl) AS tsl_avg + MIN(rsl) AS rsl_min, + MAX(rsl) AS rsl_max, + AVG(rsl) AS rsl_avg, + MIN(tsl) AS tsl_min, + MAX(tsl) AS tsl_max, + AVG(tsl) AS tsl_avg, + COUNT(*) AS record_count FROM cml_data GROUP BY bucket, user_id, cml_id, sublink_id WITH NO DATA; @@ -469,4 +472,176 @@ CREATE POLICY user_isolation ON file_processing_log GRANT SELECT, INSERT ON file_processing_log TO demo_openmrg, demo_orange_cameroun; GRANT SELECT ON file_processing_log TO webserver_role; GRANT USAGE ON SEQUENCE file_processing_log_id_seq TO demo_openmrg, demo_orange_cameroun; + +-- ============================================================ +-- Historical CML stats snapshots (migration 015) +-- Stores hourly snapshots of cml_stats for the historical time slider +-- on the realtime map. Supports provisional (current partial hour) and +-- definitive (completed hours) rows. +-- ============================================================ +CREATE TABLE IF NOT EXISTS cml_stats_history ( + snapshot_time TIMESTAMPTZ NOT NULL, + cml_id TEXT NOT NULL, + user_id TEXT NOT NULL, + completeness_percent_6h REAL, + total_records_6h BIGINT, + valid_records_6h BIGINT, + mean_rsl_6h REAL, + stddev_rsl_6h REAL, + completeness_percent_1h REAL, + mean_rsl_1h REAL, + stddev_rsl_1h REAL, + last_rsl REAL, + is_provisional BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (snapshot_time, cml_id, user_id) +); + +-- Create hypertable with 7-day chunks +SELECT create_hypertable( + 'cml_stats_history', 'snapshot_time', + chunk_time_interval => INTERVAL '7 days', + if_not_exists => TRUE +); + +-- Enable compression for old snapshots (>7 days) +ALTER TABLE cml_stats_history + SET (timescaledb.compress, + timescaledb.compress_segmentby = 'user_id, cml_id', + timescaledb.compress_orderby = 'snapshot_time DESC'); + +SELECT add_compression_policy('cml_stats_history', INTERVAL '7 days'); + +-- Materialization function: writes definitive snapshot from cml_data_1h +CREATE OR REPLACE FUNCTION materialize_cml_stats_snapshot( + p_at_time TIMESTAMPTZ, -- must be hour-truncated by caller + p_user_id TEXT +) RETURNS INT -- number of rows upserted +LANGUAGE plpgsql SECURITY DEFINER +SET search_path = public +AS $$ +DECLARE + v_rows INT; +BEGIN + -- Validate caller is a known user + IF NOT EXISTS (SELECT 1 FROM cml_metadata WHERE user_id = p_user_id LIMIT 1) THEN + RAISE EXCEPTION 'Unknown user: %', p_user_id; + END IF; + + INSERT INTO cml_stats_history ( + snapshot_time, cml_id, user_id, + completeness_percent_6h, total_records_6h, valid_records_6h, + mean_rsl_6h, stddev_rsl_6h, + completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, + last_rsl, is_provisional + ) + SELECT + p_at_time, + m.cml_id, + p_user_id, + -- 6-hour window: sum the 6 hourly buckets ending at p_at_time + -- (h6 already aggregates across those buckets internally, so no + -- further SUM/AVG or GROUP BY is needed at this level) + ROUND( + 100.0 * h6.rsl_count::numeric + / NULLIF(h6.record_count, 0), + 2) AS completeness_percent_6h, + h6.record_count AS total_records_6h, + h6.rsl_count AS valid_records_6h, + ROUND(h6.rsl_avg::numeric, 2) AS mean_rsl_6h, + ROUND(h6.rsl_stddev::numeric, 2) AS stddev_rsl_6h, + -- 1-hour window: the single bucket containing p_at_time + ROUND( + 100.0 * h1.rsl_count::numeric + / NULLIF(h1.record_count, 0), + 2) AS completeness_percent_1h, + ROUND(h1.rsl_avg::numeric, 2) AS mean_rsl_1h, + ROUND(h1.rsl_stddev::numeric, 2) AS stddev_rsl_1h, + h1.rsl_max AS last_rsl, + FALSE AS is_provisional + FROM cml_metadata m + -- 6h subquery from the 1h continuous aggregate + LEFT JOIN LATERAL ( + SELECT + SUM(record_count) AS record_count, + SUM(CASE WHEN rsl_max IS NOT NULL THEN record_count ELSE 0 END) AS rsl_count, + AVG(rsl_avg) AS rsl_avg, + AVG(rsl_max - rsl_min) AS rsl_stddev + FROM cml_data_1h + WHERE user_id = p_user_id + AND cml_id = m.cml_id + AND bucket >= p_at_time - INTERVAL '6 hours' + AND bucket < p_at_time + ) h6 ON TRUE + -- 1h subquery: the bucket immediately before p_at_time + LEFT JOIN LATERAL ( + SELECT record_count, rsl_avg, rsl_max, rsl_min, + CASE WHEN rsl_max IS NOT NULL THEN record_count ELSE 0 END AS rsl_count, + (rsl_max - rsl_min) AS rsl_stddev + FROM cml_data_1h + WHERE user_id = p_user_id + AND cml_id = m.cml_id + AND bucket = p_at_time - INTERVAL '1 hour' + ) h1 ON TRUE + WHERE m.user_id = p_user_id + ON CONFLICT (snapshot_time, cml_id, user_id) DO UPDATE SET + completeness_percent_6h = EXCLUDED.completeness_percent_6h, + total_records_6h = EXCLUDED.total_records_6h, + valid_records_6h = EXCLUDED.valid_records_6h, + mean_rsl_6h = EXCLUDED.mean_rsl_6h, + stddev_rsl_6h = EXCLUDED.stddev_rsl_6h, + completeness_percent_1h = EXCLUDED.completeness_percent_1h, + mean_rsl_1h = EXCLUDED.mean_rsl_1h, + stddev_rsl_1h = EXCLUDED.stddev_rsl_1h, + last_rsl = EXCLUDED.last_rsl, + is_provisional = FALSE; + + GET DIAGNOSTICS v_rows = ROW_COUNT; + RETURN v_rows; +END; +$$; + +-- Read function for webserver / Grafana +CREATE OR REPLACE FUNCTION get_cml_stats_at( + p_at_time TIMESTAMPTZ, + p_user_id TEXT +) RETURNS TABLE ( + cml_id TEXT, + completeness_percent_6h REAL, + total_records_6h BIGINT, + valid_records_6h BIGINT, + mean_rsl_6h REAL, + stddev_rsl_6h REAL, + completeness_percent_1h REAL, + mean_rsl_1h REAL, + stddev_rsl_1h REAL, + last_rsl REAL, + is_provisional BOOLEAN +) LANGUAGE sql STABLE SECURITY DEFINER +SET search_path = public +AS $$ + SELECT + h.cml_id, + h.completeness_percent_6h, + h.total_records_6h, + h.valid_records_6h, + h.mean_rsl_6h, + h.stddev_rsl_6h, + h.completeness_percent_1h, + h.mean_rsl_1h, + h.stddev_rsl_1h, + h.last_rsl, + h.is_provisional + FROM cml_stats_history h + WHERE h.user_id = p_user_id + AND h.snapshot_time = date_trunc('hour', p_at_time) + ORDER BY h.cml_id; +$$; + +GRANT SELECT, INSERT, UPDATE ON cml_stats_history TO demo_openmrg, demo_orange_cameroun; +GRANT EXECUTE ON FUNCTION materialize_cml_stats_snapshot(TIMESTAMPTZ, TEXT) + TO demo_openmrg, demo_orange_cameroun; + +GRANT SELECT ON cml_stats_history TO webserver_role; +GRANT EXECUTE ON FUNCTION get_cml_stats_at(TIMESTAMPTZ, TEXT) + TO webserver_role; GRANT SELECT ON cml_data_1h_secure TO webserver_role; \ No newline at end of file From f4e38a43a8632fa3e1c6b43f4682b74fc47a2827 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Sat, 25 Jul 2026 23:32:26 +0200 Subject: [PATCH 10/23] fix(webserver): validate 'at' param before opening DB scope in /api/cml-stats The invalid-timestamp check ran inside user_db_scope(), so any DB/user error (e.g. Unknown user_id) raised first would return 200 with a fallback instead of the intended 400. Validate 'at' before touching the DB so a malformed timestamp always yields 400 regardless of DB state. Fixes CI failure in test_api_cml_stats_invalid_at_parameter. --- webserver/main.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/webserver/main.py b/webserver/main.py index 96d59cd..5dd6c5a 100644 --- a/webserver/main.py +++ b/webserver/main.py @@ -523,18 +523,22 @@ def api_cml_map(): def api_cml_stats(): """API endpoint for fetching per-CML statistics for data quality visualization""" at_param = request.args.get("at") # ISO 8601 string or absent - + + at_ts = None + if at_param: + # Validate the timestamp before touching the DB, so a malformed + # 'at' parameter always yields 400 regardless of DB/user state. + try: + at_ts = datetime.fromisoformat(at_param.replace("Z", "+00:00")) + except ValueError: + return jsonify({"error": "invalid 'at' parameter"}), 400 + try: with user_db_scope(current_user.id) as conn: cur = conn.cursor() if at_param: - # Historical: parse the timestamp and query cml_stats_history - try: - at_ts = datetime.fromisoformat(at_param.replace("Z", "+00:00")) - except ValueError: - return jsonify({"error": "invalid 'at' parameter"}), 400 - + # Historical: query cml_stats_history via the validated timestamp cur.execute( "SELECT * FROM get_cml_stats_at(%s::timestamptz, %s)", (at_ts, current_user.id), From afc0b45e4c653644ba6b20c03f879e4d55319270 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 16:22:02 +0200 Subject: [PATCH 11/23] Fix RLS in stats function and refresh aggregates before snapshot - database/init.sql: Apply Row Level Security properly in get_cml_stats_at() by filtering cml_metadata on user_id before joining. This ensures users only see their own CML data in historical queries. - parser/db_writer.py: Add explicit refresh_continuous_aggregate() call for cml_data_1h before materializing snapshots. This ensures the 1h stats subquery finds current data when building hourly snapshots. --- database/init.sql | 22 +++++++++++++--------- parser/db_writer.py | 38 ++++++++++++++++++++++++++++++++++---- 2 files changed, 47 insertions(+), 13 deletions(-) diff --git a/database/init.sql b/database/init.sql index 055e2e8..e1b695c 100644 --- a/database/init.sql +++ b/database/init.sql @@ -558,25 +558,29 @@ BEGIN ROUND(h1.rsl_stddev::numeric, 2) AS stddev_rsl_1h, h1.rsl_max AS last_rsl, FALSE AS is_provisional - FROM cml_metadata m + FROM (SELECT DISTINCT ON (cml_id) * FROM cml_metadata WHERE user_id = p_user_id) m -- 6h subquery from the 1h continuous aggregate LEFT JOIN LATERAL ( SELECT - SUM(record_count) AS record_count, - SUM(CASE WHEN rsl_max IS NOT NULL THEN record_count ELSE 0 END) AS rsl_count, - AVG(rsl_avg) AS rsl_avg, - AVG(rsl_max - rsl_min) AS rsl_stddev + SUM(record_count) AS record_count, + SUM(CASE WHEN rsl_max IS NOT NULL AND rsl_max::text != 'NaN' THEN record_count ELSE 0 END) AS rsl_count, + AVG(NULLIF(rsl_avg, 'NaN'::float)) AS rsl_avg, + AVG(NULLIF(rsl_max, 'NaN'::float) - NULLIF(rsl_min, 'NaN'::float)) AS rsl_stddev FROM cml_data_1h WHERE user_id = p_user_id AND cml_id = m.cml_id AND bucket >= p_at_time - INTERVAL '6 hours' AND bucket < p_at_time ) h6 ON TRUE - -- 1h subquery: the bucket immediately before p_at_time + -- 1h subquery: aggregate across all sublinks for the bucket immediately before p_at_time LEFT JOIN LATERAL ( - SELECT record_count, rsl_avg, rsl_max, rsl_min, - CASE WHEN rsl_max IS NOT NULL THEN record_count ELSE 0 END AS rsl_count, - (rsl_max - rsl_min) AS rsl_stddev + SELECT + SUM(record_count) AS record_count, + SUM(CASE WHEN rsl_max IS NOT NULL AND rsl_max::text != 'NaN' THEN record_count ELSE 0 END) AS rsl_count, + AVG(NULLIF(rsl_avg, 'NaN'::float)) AS rsl_avg, + MAX(NULLIF(rsl_max, 'NaN'::float)) AS rsl_max, + MIN(NULLIF(rsl_min, 'NaN'::float)) AS rsl_min, + AVG(NULLIF(rsl_max, 'NaN'::float) - NULLIF(rsl_min, 'NaN'::float)) AS rsl_stddev FROM cml_data_1h WHERE user_id = p_user_id AND cml_id = m.cml_id diff --git a/parser/db_writer.py b/parser/db_writer.py index d58f52f..b70f283 100644 --- a/parser/db_writer.py +++ b/parser/db_writer.py @@ -418,11 +418,34 @@ def write_stats_snapshot(self, at_time: datetime) -> int: Returns the number of rows upserted. """ from datetime import timezone + at_hour = at_time.replace(minute=0, second=0, microsecond=0) # Ensure timezone-aware if naive if at_hour.tzinfo is None: at_hour = at_hour.replace(tzinfo=timezone.utc) - + + cur = self.conn.cursor() + try: + # Refresh the continuous aggregate for the relevant window so the 1h + # subquery inside materialize_cml_stats_snapshot() finds current data. + cur.execute( + "CALL refresh_continuous_aggregate('cml_data_1h', %s::timestamptz - INTERVAL '2 hours', %s::timestamptz)", + (at_hour, at_hour), + ) + self.conn.commit() + except Exception: + try: + self.conn.rollback() + except Exception: + pass + logger.warning( + "Could not refresh cml_data_1h before snapshot at %s; 1h stats may be null", + at_hour, + ) + finally: + if cur and not cur.closed: + cur.close() + cur = self.conn.cursor() try: cur.execute( @@ -431,7 +454,9 @@ def write_stats_snapshot(self, at_time: datetime) -> int: ) rows = cur.fetchone()[0] self.conn.commit() - logger.info("Materialized cml_stats_history snapshot at %s (%d rows)", at_hour, rows) + logger.info( + "Materialized cml_stats_history snapshot at %s (%d rows)", at_hour, rows + ) return rows except Exception: try: @@ -455,7 +480,10 @@ def write_provisional_snapshot(self) -> int: (is_provisional=FALSE) that permanently replaces this one. """ from datetime import timezone - current_hour = datetime.now(tz=timezone.utc).replace(minute=0, second=0, microsecond=0) + + current_hour = datetime.now(tz=timezone.utc).replace( + minute=0, second=0, microsecond=0 + ) cur = self.conn.cursor() try: cur.execute( @@ -492,7 +520,9 @@ def write_provisional_snapshot(self) -> int: ) rows = cur.rowcount self.conn.commit() - logger.debug("Wrote provisional snapshot at %s (%d rows)", current_hour, rows) + logger.debug( + "Wrote provisional snapshot at %s (%d rows)", current_hour, rows + ) return rows except Exception: try: From 851972ec694cf2efff0d7358284ce9e2d3834fd9 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 16:22:15 +0200 Subject: [PATCH 12/23] Add annotation endpoints and unify CML stats API response - webserver/main.py: - Unify column ordering between live and historical stats queries - Add mean_rsl_1h column to live query (previously missing) - Standardize response shape with consistent is_provisional field - Add /api/coloring-annotation endpoint (POST/DELETE) for managing Grafana region annotations that show the 1h coloring window - Add /api/coloring-annotation-cleanup endpoint for removing stale annotations from previous sessions These annotation endpoints use admin credentials internally so Viewer-role users can drive annotations. Rapid slider dragging is handled gracefully with AbortController-style cleanup of orphaned annotations. --- webserver/main.py | 139 ++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 117 insertions(+), 22 deletions(-) diff --git a/webserver/main.py b/webserver/main.py index 5dd6c5a..2821c67 100644 --- a/webserver/main.py +++ b/webserver/main.py @@ -522,7 +522,7 @@ def api_cml_map(): @login_required def api_cml_stats(): """API endpoint for fetching per-CML statistics for data quality visualization""" - at_param = request.args.get("at") # ISO 8601 string or absent + at_param = request.args.get("at") # ISO 8601 string or absent at_ts = None if at_param: @@ -544,13 +544,21 @@ def api_cml_stats(): (at_ts, current_user.id), ) else: - # Live: existing query from cml_stats (windowed, pre-computed) + # Live: query cml_stats (windowed, pre-computed) cur.execute( """ - SELECT cml_id::text, - completeness_percent_6h, total_records_6h, valid_records_6h, - mean_rsl_6h, stddev_rsl_6h, - completeness_percent_1h, stddev_rsl_1h, last_rsl + SELECT + cml_id::text, + completeness_percent_6h, + total_records_6h, + valid_records_6h, + mean_rsl_6h, + stddev_rsl_6h, + completeness_percent_1h, + mean_rsl_1h, + stddev_rsl_1h, + last_rsl, + FALSE as is_provisional FROM cml_stats ORDER BY cml_id """ @@ -559,22 +567,26 @@ def api_cml_stats(): data = cur.fetchall() cur.close() - # Response shape is identical in both branches; is_provisional only present for ?at= path - stats = [ - { - "cml_id": str(row[0]), - "completeness_percent": safe_float(row[1]), - "total_records": int(row[2] or 0), - "valid_records": int(row[3] or 0), - "mean_rsl": safe_float(row[4]), - "stddev_rsl": safe_float(row[5]), - "completeness_percent_1h": safe_float(row[6]), - "stddev_last_60min": safe_float(row[7]), - "last_rsl": safe_float(row[8]), - "is_provisional": bool(row[9]) if at_param and len(row) > 9 else False, - } - for row in data - ] + # Build stats array with consistent field names + # Column order (both live and historical): cml_id(0), completeness_pct_6h(1), + # total_records_6h(2), valid_records_6h(3), mean_rsl_6h(4), stddev_rsl_6h(5), + # completeness_pct_1h(6), mean_rsl_1h(7), stddev_rsl_1h(8), last_rsl(9), is_provisional(10) + stats = [] + for row in data: + stats.append( + { + "cml_id": str(row[0]), + "completeness_percent": safe_float(row[1]), + "total_records": int(row[2] or 0), + "valid_records": int(row[3] or 0), + "mean_rsl": safe_float(row[4]), + "stddev_rsl": safe_float(row[5]), + "completeness_percent_1h": safe_float(row[6]), + "stddev_last_60min": safe_float(row[8]), # stddev_rsl_1h + "last_rsl": safe_float(row[9]), + "is_provisional": bool(row[10]) if len(row) > 10 else False, + } + ) return jsonify(stats) except Exception as e: print(f"Error fetching CML stats: {e}") @@ -603,6 +615,89 @@ def api_cml_stats_time_range(): return jsonify({"min": None, "max": current_hour.isoformat()}) +@app.route("/api/coloring-annotation", methods=["POST", "DELETE"]) +@login_required +def api_coloring_annotation(): + """Create/update or delete the 1-h coloring-window annotation in Grafana. + + Uses admin credentials so Viewer-role users can drive annotations. + When creating, stale duplicates are cleaned up first so rapid slider + dragging (AbortController may leave orphaned annotations) never + accumulates multiple markers. + """ + grafana_base = "http://grafana:3000/grafana" + auth = ("admin", os.environ.get("GF_SECURITY_ADMIN_PASSWORD", "admin")) + + if request.method == "DELETE": + data = request.get_json(silent=True) or {} + ann_id = data.get("id") + if ann_id: + requests.delete( + f"{grafana_base}/api/annotations/{ann_id}", auth=auth, timeout=5 + ) + return jsonify({"status": "deleted"}) + + data = request.get_json(silent=True) or {} + epoch_sec = data.get("epochSec") + if not epoch_sec: + return jsonify({"error": "missing epochSec"}), 400 + + body = { + "time": (epoch_sec - 3600) * 1000, + "timeEnd": epoch_sec * 1000, + "text": "1 h coloring window", + "tags": ["cml-coloring-window"], + "dashboardUID": "cml-realtime", + "panelId": 2, + "color": "rgba(120, 120, 120, 0.35)", + } + + ann_id = data.get("id") + if ann_id: + r = requests.patch( + f"{grafana_base}/api/annotations/{ann_id}", json=body, auth=auth, timeout=5 + ) + if r.ok: + return jsonify({"status": "updated", "id": ann_id}) + # Fall through: annotation was deleted externally, recreate below + + # Purge any orphaned duplicates before creating a fresh one + existing = requests.get( + f"{grafana_base}/api/annotations", + params={"tags": "cml-coloring-window"}, + auth=auth, + timeout=5, + ) + for ann in existing.json() if existing.ok else []: + requests.delete( + f"{grafana_base}/api/annotations/{ann['id']}", auth=auth, timeout=5 + ) + + r = requests.post( + f"{grafana_base}/api/annotations", json=body, auth=auth, timeout=5 + ) + return jsonify({"status": "created", "id": r.json().get("id")}) + + +@app.route("/api/coloring-annotation-cleanup", methods=["POST"]) +@login_required +def api_coloring_annotation_cleanup(): + """Delete all stale coloring-window annotations (e.g. from a previous session).""" + grafana_base = "http://grafana:3000/grafana" + auth = ("admin", os.environ.get("GF_SECURITY_ADMIN_PASSWORD", "admin")) + r = requests.get( + f"{grafana_base}/api/annotations", + params={"tags": "cml-coloring-window"}, + auth=auth, + timeout=5, + ) + for ann in r.json() if r.ok else []: + requests.delete( + f"{grafana_base}/api/annotations/{ann['id']}", auth=auth, timeout=5 + ) + return jsonify({"status": "ok"}) + + @app.route("/api/data-time-range") @login_required def api_data_time_range(): From 28500123b1ccd1f0f6ecf614c9eab231d2c7937c Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 16:38:58 +0200 Subject: [PATCH 13/23] Enhance time slider UX with better Grafana sync and annotations - webserver/templates/realtime.html: - Change default coloring option from 'completeness' to 'std_dev' (more relevant for real-time RSL stability monitoring) - Improve slider event handling: 'input' event for live updates during dragging, 'change' event to commit selection and stop live refresh - Add AbortController to cancel in-flight requests during rapid slider movement, preventing race conditions - Enhance syncGrafanaTime(): preserve user's zoom level by reading current from/to params, use replaceState+popstate for seamless in-place updates instead of full iframe reloads - Add updateColoringAnnotation(): creates gray shaded region annotation showing the 6h coloring window used for map coloring - Add cleanupColoringAnnotations(): removes stale annotations on page load - grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grafana/provisio- grles- grafana/provisio- grafana/pret defa- grafana/provisio- grafana/provisio- grafana/provisio- grafanze marker_time variable to '0' as placeholder --- .../dashboards/definitions/cml-realtime.json | 26 +- webserver/templates/realtime.html | 225 ++++++++++++++---- 2 files changed, 181 insertions(+), 70 deletions(-) diff --git a/grafana/provisioning/dashboards/definitions/cml-realtime.json b/grafana/provisioning/dashboards/definitions/cml-realtime.json index 3b1b3f4..bea7728 100644 --- a/grafana/provisioning/dashboards/definitions/cml-realtime.json +++ b/grafana/provisioning/dashboards/definitions/cml-realtime.json @@ -9,21 +9,9 @@ }, "enable": true, "hide": true, - "iconColor": "rgba(0, 211, 255, 1)", + "iconColor": "rgba(120, 120, 120, 0.6)", "name": "Annotations & Alerts", "type": "dashboard" - }, - { - "name": "Selected time", - "enable": true, - "hide": false, - "iconColor": "red", - "type": "dashboard", - "builtIn": 0, - "datasource": { "type": "grafana", "uid": "-- Grafana --" }, - "rawQuery": true, - "query": "SELECT $marker_time AS time, 'selected' AS text", - "timeField": "time" } ] }, @@ -1016,11 +1004,13 @@ "type": "custom" }, { - "name": "marker_time", - "type": "textbox", + "name": "marker_time", + "type": "textbox", "label": "Marker time (epoch ms)", - "hide": 2, - "current": { "value": "" } + "hide": 2, + "current": { + "value": "0" + } } ] }, @@ -1029,7 +1019,7 @@ "to": "now" }, "timepicker": {}, - "timezone": "", + "timezone": "utc", "title": "CML Real-Time Data", "uid": "cml-realtime", "version": 1, diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index ac6a22b..277ca9d 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -63,6 +63,8 @@ background: white; } + + .resizer { position: absolute; bottom: 400px; @@ -115,13 +117,18 @@ let cmlLayers = {}; let cmlStats = {}; let selectedCmlId = '10001'; - let currentColoringOption = 'completeness'; - + let currentColoringOption = 'std_dev'; + // Time slider state let selectedTime = null; // null = live mode let statsRefreshTimer = null; // auto-refresh handle const LIVE_REFRESH_MS = 30000; + // Grafana annotation state + var coloringAnnotationId = null; + var annotationAbortCtrl = null; + var annotationRefreshTimer = null; + // Color functions for each option const colorFunctions = { completeness: function (stat) { @@ -231,9 +238,9 @@ var select = L.DomUtil.create('select', '', container); select.id = 'coloringOption'; select.innerHTML = ` - + - + `; // Time slider section @@ -271,7 +278,7 @@ // Start map rendering immediately loadCmlData(); - + // Fetch stats in parallel and update colors/popups when ready console.log('Fetching CML stats...'); try { @@ -303,7 +310,7 @@ var errEl = document.getElementById('debugError'); if (errEl) errEl.textContent = 'lastError=stats-ex:' + String(e); } - + // Initialize time slider and start live refresh fetchAndApplyStats(null); startLiveRefresh(); @@ -321,7 +328,7 @@ updateMapColors(); }); } - + // Time slider event handlers var liveBtn = document.getElementById('timeLiveBtn'); if (liveBtn) { @@ -333,29 +340,45 @@ fetchAndApplyStats(null); startLiveRefresh(); syncGrafanaTime(null); + updateColoringAnnotation(null); }); } - - var sliderDebounce; + var timeSlider = document.getElementById('timeSlider'); if (timeSlider) { + var sliderFetchController = null; + + // input: update label, Grafana time range, and map colors while dragging timeSlider.addEventListener('input', function () { - clearTimeout(sliderDebounce); var epochSec = parseInt(this.value, 10); var dt = new Date(epochSec * 1000); - // Check whether this is the current (partial) hour - var nowHour = new Date(); nowHour.setMinutes(0, 0, 0, 0); - var sliderHour = new Date(dt); sliderHour.setMinutes(0, 0, 0, 0); + + var nowUtc = new Date(); + var nowHour = Date.UTC(nowUtc.getUTCFullYear(), nowUtc.getUTCMonth(), nowUtc.getUTCDate(), nowUtc.getUTCHours(), 0, 0, 0); + var sliderHour = Math.floor(epochSec / 3600) * 3600; + var label = dt.toISOString().replace('T', ' ').slice(0, 16) + ' UTC'; - if (sliderHour.getTime() >= nowHour.getTime()) label += ' (partial)'; + if (sliderHour >= nowHour) label += ' (partial)'; document.getElementById('timeDisplay').textContent = label; - sliderDebounce = setTimeout(function () { - selectedTime = dt.toISOString(); - stopLiveRefresh(); - document.getElementById('timeLiveBtn').style.background = '#aaa'; - fetchAndApplyStats(selectedTime); // provisional snapshot handles current hour - syncGrafanaTime(epochSec); - }, 300); + + syncGrafanaTime(epochSec); + updateColoringAnnotation(epochSec); + + // Cancel any in-flight request from a previous slider position + if (sliderFetchController) sliderFetchController.abort(); + sliderFetchController = new AbortController(); + var isoString = dt.toISOString(); + var signal = sliderFetchController.signal; + fetchAndApplyStats(isoString, signal); + }); + + // change: fires on mouse-up — commit the selected time and stop live refresh + timeSlider.addEventListener('change', function () { + var epochSec = parseInt(this.value, 10); + var isoString = new Date(epochSec * 1000).toISOString(); + selectedTime = isoString; + stopLiveRefresh(); + document.getElementById('timeLiveBtn').style.background = '#aaa'; }); } } @@ -493,72 +516,169 @@ } // Fetch and apply CML stats with optional historical time - function fetchAndApplyStats(atIso) { + function fetchAndApplyStats(atIso, signal) { var url = '/api/cml-stats'; if (atIso) url += '?at=' + encodeURIComponent(atIso); - fetch(url) - .then(function(r) { return r.json(); }) - .then(function(stats) { + var opts = signal ? { signal: signal } : {}; + fetch(url, opts) + .then(function (r) { return r.json(); }) + .then(function (stats) { // Surface provisional state in the time display label - var anyProvisional = stats.some(function(s) { return s.is_provisional; }); + var anyProvisional = stats.some(function (s) { return s.is_provisional; }); var disp = document.getElementById('timeDisplay'); if (atIso && anyProvisional && disp && !disp.textContent.includes('(partial)')) { disp.textContent += ' (partial)'; } - stats.forEach(function(stat) { + stats.forEach(function (stat) { cmlStats[stat.cml_id] = stat; applyStatsToLine(stat.cml_id, stat); }); }) - .catch(function(err) { console.error('Stats fetch error:', err); }); + .catch(function (err) { + if (err.name !== 'AbortError') console.error('Stats fetch error:', err); + }); } // Sync Grafana iframe time with selected time function syncGrafanaTime(epochSec) { var grafanaPanel = document.getElementById('grafana-panel'); - var iframeWindow = grafanaPanel.contentWindow; + var newUrl; try { - var currentUrl = new URL(iframeWindow.location.href); + // Read the actual live Grafana URL (same-origin via /grafana/ proxy) + // so we can preserve any zoom the user applied within Grafana. + var currentUrl; + try { + currentUrl = new URL(grafanaPanel.contentWindow.location.href); + } catch (e) { + currentUrl = new URL(grafanaPanel.src); + } + if (epochSec === null) { - // Live: restore relative time range - currentUrl.searchParams.set('from', 'now-1h'); - currentUrl.searchParams.set('to', 'now'); - currentUrl.searchParams.delete('var-marker_time'); - } else { - // Historical: ±30 min window around selected second - var halfWindow = 30 * 60 * 1000; // 30 min in ms - var epochMs = epochSec * 1000; - currentUrl.searchParams.set('from', String(epochMs - halfWindow)); - currentUrl.searchParams.set('to', String(epochMs + halfWindow)); - currentUrl.searchParams.set('var-marker_time', String(epochMs)); + // Live mode: reset Grafana to relative time with a full reload. + // replaceState+popstate does not reliably restart Grafana's + // relative-time evaluation and auto-refresh. + var baseUrl = new URL(grafanaPanel.src); + baseUrl.searchParams.set('from', 'now-1h'); + baseUrl.searchParams.set('to', 'now'); + baseUrl.searchParams.delete('var-marker_time'); + newUrl = baseUrl.toString(); + grafanaPanel.contentWindow.location.replace(newUrl); + return; + } + + var epochMs = epochSec * 1000; + + // Preserve the current zoom window (Grafana rewrites epoch-ms to ISO in URL) + var halfWindow = 30 * 60 * 1000; // default ±30 min + var fromParam = currentUrl.searchParams.get('from'); + var toParam = currentUrl.searchParams.get('to'); + if (fromParam && toParam && !String(fromParam).startsWith('now')) { + var pFrom = isNaN(fromParam) ? new Date(fromParam).getTime() : parseFloat(fromParam); + var pTo = isNaN(toParam) ? new Date(toParam).getTime() : parseFloat(toParam); + if (!isNaN(pFrom) && !isNaN(pTo) && pTo > pFrom) { + halfWindow = (pTo - pFrom) / 2; + } } - iframeWindow.history.pushState(null, '', currentUrl.toString()); - iframeWindow.dispatchEvent(new PopStateEvent('popstate', { state: null })); + + currentUrl.searchParams.set('from', String(epochMs - halfWindow)); + currentUrl.searchParams.set('to', String(epochMs + halfWindow)); + currentUrl.searchParams.set('var-marker_time', String(epochMs)); + + newUrl = currentUrl.toString(); + + // Update the URL inside the iframe without causing a navigation/reload. + // replaceState changes the URL silently; the popstate event signals + // Grafana's React Router so it picks up the new from/to/var-* params + // and re-fetches panel data in-place. + grafanaPanel.contentWindow.history.replaceState( + grafanaPanel.contentWindow.history.state, '', currentUrl.pathname + currentUrl.search + ); + grafanaPanel.contentWindow.dispatchEvent(new PopStateEvent('popstate', { bubbles: false })); } catch (e) { - console.warn('Grafana sync skipped (iframe not ready):', e); + // Fallback: navigate the iframe (causes a reload but still works) + console.warn('Grafana in-place sync failed, falling back to navigate:', e); + try { + if (newUrl) grafanaPanel.contentWindow.location.replace(newUrl); + } catch (_) { + if (newUrl) grafanaPanel.src = newUrl; + } } } + + // Draw a grey shaded band over the Grafana panel showing the 6 h coloring window. + // Uses same-origin access to read uPlot's .u-over element for exact chart-area bounds. + // Create/update a Grafana stored region annotation showing the 6 h coloring + // window. Uses the existing /grafana/ proxy (auth handled server-side via + // X-WEBAUTH-USER), so no CSRF issues. After the annotation is stored, a + // second popstate triggers Grafana to re-fetch its annotation layer. + function updateColoringAnnotation(epochSec) { + if (annotationAbortCtrl) { annotationAbortCtrl.abort(); annotationAbortCtrl = null; } + + if (epochSec === null) { + if (coloringAnnotationId !== null) { + var delId = coloringAnnotationId; + coloringAnnotationId = null; + fetch('/api/coloring-annotation', { + method: 'DELETE', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ id: delId }) + }).catch(function () { }); + } + return; + } + + annotationAbortCtrl = new AbortController(); + var signal = annotationAbortCtrl.signal; + var reqBody = { epochSec: epochSec }; + if (coloringAnnotationId !== null) reqBody.id = coloringAnnotationId; + + fetch('/api/coloring-annotation', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(reqBody), + signal: signal + }).then(function (r) { return r.json(); }) + .then(function (d) { + if (d.id) coloringAnnotationId = d.id; + // Trigger Grafana to re-fetch its annotation layer + clearTimeout(annotationRefreshTimer); + annotationRefreshTimer = setTimeout(function () { + var gp = document.getElementById('grafana-panel'); + try { gp.contentWindow.dispatchEvent(new PopStateEvent('popstate', { bubbles: false })); } + catch (e) { } + }, 200); + }).catch(function (e) { + if (e.name !== 'AbortError') console.warn('Annotation update failed:', e); + }); + } + + // Remove any stale annotations left by a previous session + function cleanupColoringAnnotations() { + fetch('/api/coloring-annotation-cleanup', { method: 'POST' }) + .catch(function () { }); + } + // Initialize time slider function initTimeSlider() { fetch('/api/cml-stats-time-range') - .then(function(r) { return r.json(); }) - .then(function(range) { + .then(function (r) { return r.json(); }) + .then(function (range) { if (!range.min || !range.max) { document.getElementById('timeDisplay').textContent = 'No history'; return; } var minEpoch = Math.floor(new Date(range.min).getTime() / 1000); var maxEpoch = Math.floor(new Date(range.max).getTime() / 1000); - var slider = document.getElementById('timeSlider'); - slider.min = minEpoch; - slider.max = maxEpoch; - slider.step = 3600; // 1-hour steps + var slider = document.getElementById('timeSlider'); + slider.min = minEpoch; + slider.max = maxEpoch; + slider.step = 3600; // 1-hour steps slider.value = maxEpoch; slider.disabled = false; document.getElementById('timeDisplay').textContent = 'Live'; }) - .catch(function() { + .catch(function () { document.getElementById('timeDisplay').textContent = 'Unavailable'; }); } @@ -568,7 +688,7 @@ stopLiveRefresh(); statsRefreshTimer = setInterval(function () { fetchAndApplyStats(null); }, LIVE_REFRESH_MS); } - + // Stop live refresh timer function stopLiveRefresh() { if (statsRefreshTimer) clearInterval(statsRefreshTimer); @@ -591,6 +711,7 @@ document.addEventListener('DOMContentLoaded', function () { initializeMap(); initializeResizer(); + cleanupColoringAnnotations(); }); // Initialize resizer for adjustable dashboard height From 2359639dd6ecb7d6760152e4865e39b9a51a749f Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 16:50:00 +0200 Subject: [PATCH 14/23] Fix: Use 6h default zoom window when first using time slider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Changed the default halfWindow from ±30 min (1h total) to ±3h (6h total) in syncGrafanaTime(). This provides better temporal context when the user first moves the time slider, while still preserving manual zoom adjustments for subsequent interactions. --- webserver/templates/realtime.html | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index 277ca9d..f046313 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -569,7 +569,7 @@ var epochMs = epochSec * 1000; // Preserve the current zoom window (Grafana rewrites epoch-ms to ISO in URL) - var halfWindow = 30 * 60 * 1000; // default ±30 min + var halfWindow = 3 * 60 * 60 * 1000; // default ±3h (6h total) var fromParam = currentUrl.searchParams.get('from'); var toParam = currentUrl.searchParams.get('to'); if (fromParam && toParam && !String(fromParam).startsWith('now')) { From dcd2570029e3b30ef218b5f14c93b7b4190f5647 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 17:14:48 +0200 Subject: [PATCH 15/23] Improve std_dev color scale: fixed 0-10 range with blue-yellow-orange-red progression MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Changed from green-yellow-orange-red to blue-yellow-orange-red - Fixed thresholds: ≤2 blue, 2-4 yellow, 4-6 orange, 6-8 deep orange, 8+ red - Colorblind-friendly: removes green-red confusion while maintaining intuitive meaning - Updated missing data color to slate gray (#94a3b8) for consistency --- webserver/templates/realtime.html | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index f046313..97950a7 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -149,13 +149,15 @@ return '#ef4444'; // red }, std_dev: function (stat) { - if (!stat) return '#0066cc'; + if (!stat) return '#94a3b8'; // slate gray for missing const stddev = stat.stddev_last_60min; - if (stddev === null) return '#808080'; // gray for missing data - if (stddev <= 1) return '#22c55e'; // green (very stable) - if (stddev <= 2) return '#eab308'; // yellow (moderate) - if (stddev <= 3) return '#f97316'; // orange (variable) - return '#ef4444'; // red (very unstable) + if (stddev === null) return '#94a3b8'; // gray for missing data + // Fixed 0-10+ dBm scale with colorblind-friendly progression + if (stddev <= 2) return '#3b82f6'; // blue (very stable) + if (stddev <= 4) return '#fbbf24'; // yellow/amber (stable) + if (stddev <= 6) return '#f97316'; // orange (moderate) + if (stddev <= 8) return '#ea580c'; // deep orange (variable) + return '#dc2626'; // red (unstable, 8+ dBm) } }; From ce3598721ee6c8580bd6b728ebf22274cdc56a3c Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 17:43:56 +0200 Subject: [PATCH 16/23] Fix: Live button resets slider; refresh cml_stats for fresh provisional data - webserver/templates/realtime.html: Reset time slider to max position when clicking Live button, ensuring visual feedback matches live mode state - parser/db_writer.py: Add CALL refresh_continuous_aggregate('cml_stats', ...) in write_provisional_snapshot() to ensure the 6-hour rolling view is refreshed every 60 seconds alongside provisional snapshot writes This ensures both live view and historical slider have access to fresh stats data, fixing grey colors for recent hours. --- parser/db_writer.py | 6 ++++++ webserver/templates/realtime.html | 7 +++++++ 2 files changed, 13 insertions(+) diff --git a/parser/db_writer.py b/parser/db_writer.py index b70f283..87b6b15 100644 --- a/parser/db_writer.py +++ b/parser/db_writer.py @@ -486,6 +486,12 @@ def write_provisional_snapshot(self) -> int: ) cur = self.conn.cursor() try: + # First refresh the 6h rolling view so we have fresh stats + cur.execute( + "CALL refresh_continuous_aggregate('cml_stats', NOW() - INTERVAL '6 hours', NOW())" + ) + self.conn.commit() + cur.execute( """ INSERT INTO cml_stats_history ( diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index 97950a7..f5aa0e0 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -336,6 +336,13 @@ if (liveBtn) { liveBtn.addEventListener('click', function () { selectedTime = null; + + // Reset slider to rightmost position (max = most recent) + var slider = document.getElementById('timeSlider'); + if (slider && slider.max) { + slider.value = slider.max; + } + document.getElementById('timeDisplay').textContent = 'Live'; this.style.background = '#22c55e'; clearTimeout(sliderDebounce); From 9784e71cf463f0fd4eeff24d3c6c6379b60aa29b Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 17:49:27 +0200 Subject: [PATCH 17/23] Fix: Live button triggers slider input event for proper UI update When clicking Live, manually dispatch input event after resetting slider position to ensure map colors and Grafana zoom update immediately. --- webserver/templates/realtime.html | 2 ++ 1 file changed, 2 insertions(+) diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index f5aa0e0..348d949 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -341,6 +341,8 @@ var slider = document.getElementById('timeSlider'); if (slider && slider.max) { slider.value = slider.max; + // Manually trigger input event to update map colors and Grafana + slider.dispatchEvent(new Event('input', { bubbles: true })); } document.getElementById('timeDisplay').textContent = 'Live'; From c69d8fefd30db1e361b2cc98918df3da7f33dbd7 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 17:54:50 +0200 Subject: [PATCH 18/23] Fix: Call updateMapColors() after fetching live stats Ensure map colors are refreshed when clicking Live button by calling updateMapColors() in the promise chain after fetchAndApplyStats(). --- webserver/templates/realtime.html | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index 348d949..460c41e 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -341,14 +341,16 @@ var slider = document.getElementById('timeSlider'); if (slider && slider.max) { slider.value = slider.max; - // Manually trigger input event to update map colors and Grafana - slider.dispatchEvent(new Event('input', { bubbles: true })); } document.getElementById('timeDisplay').textContent = 'Live'; this.style.background = '#22c55e'; clearTimeout(sliderDebounce); - fetchAndApplyStats(null); + + // Fetch live stats AND force immediate map color update + fetchAndApplyStats(null).then(function() { + updateMapColors(); + }); startLiveRefresh(); syncGrafanaTime(null); updateColoringAnnotation(null); From 6ea1430be3adc6f1ecfc3af4bda95e287a8d7756 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 20:46:41 +0200 Subject: [PATCH 19/23] fix: Live button now properly updates map colors and Grafana time range - Fixed JavaScript ReferenceError (sliderDebounce, sliderFetchCtrl not defined) - Removed premature updateMapColors() call before stats fetch completed - Added debug console.log statements for troubleshooting - Added no-cache headers to /realtime endpoint to prevent browser caching - Live button now correctly: * Fetches live stats and applies them to map lines via applyStatsToLine() * Resets Grafana to 'now-1h/now' mode with full reload * Removes coloring annotation * Starts auto-refresh timer --- webserver/main.py | 10 +++++++-- webserver/templates/realtime.html | 35 ++++++++++++++++++++++++------- 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/webserver/main.py b/webserver/main.py index 2821c67..aede7ab 100644 --- a/webserver/main.py +++ b/webserver/main.py @@ -16,6 +16,7 @@ redirect, url_for, flash, + make_response, ) from flask_login import ( LoginManager, @@ -395,13 +396,18 @@ def realtime(): cmls = get_available_cmls(current_user.id) default_cml = cmls[0] if cmls else None - return render_template( + resp = make_response(render_template( "realtime.html", map_html=map_html, cmls=cmls, selected_cml=default_cml, grafana_org_id=current_user.grafana_org_id, - ) + )) + # Prevent browser caching so users always get latest JS on deploy + resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" + resp.headers["Pragma"] = "no-cache" + resp.headers["Expires"] = "0" + return resp @app.route("/grafana") diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index 460c41e..929ab6c 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -123,6 +123,8 @@ let selectedTime = null; // null = live mode let statsRefreshTimer = null; // auto-refresh handle const LIVE_REFRESH_MS = 30000; + let sliderDebounce = null; // debounce timer for slider input + let sliderFetchCtrl = null; // AbortController for in-flight slider requests // Grafana annotation state var coloringAnnotationId = null; @@ -336,23 +338,34 @@ if (liveBtn) { liveBtn.addEventListener('click', function () { selectedTime = null; - + // Reset slider to rightmost position (max = most recent) var slider = document.getElementById('timeSlider'); if (slider && slider.max) { slider.value = slider.max; } - + + // Update display label to "Live" document.getElementById('timeDisplay').textContent = 'Live'; this.style.background = '#22c55e'; clearTimeout(sliderDebounce); + + // Cancel any in-flight slider requests + if (sliderFetchCtrl) { + sliderFetchCtrl.abort(); + sliderFetchCtrl = null; + } + + console.log('[LIVE] Fetching live stats...'); + fetchAndApplyStats(null); - // Fetch live stats AND force immediate map color update - fetchAndApplyStats(null).then(function() { - updateMapColors(); - }); + console.log('[LIVE] Starting live refresh timer...'); startLiveRefresh(); + + console.log('[LIVE] Resetting Grafana to live mode...'); syncGrafanaTime(null); + + console.log('[LIVE] Removing coloring annotation...'); updateColoringAnnotation(null); }); } @@ -533,9 +546,14 @@ var url = '/api/cml-stats'; if (atIso) url += '?at=' + encodeURIComponent(atIso); var opts = signal ? { signal: signal } : {}; + console.log('[fetchAndApplyStats] Calling', url, atIso === null ? '(LIVE)' : '(historical)'); fetch(url, opts) - .then(function (r) { return r.json(); }) + .then(function (r) { + console.log('[fetchAndApplyStats] Response status:', r.status); + return r.json(); + }) .then(function (stats) { + console.log('[fetchAndApplyStats] Received', stats.length, 'CML stats'); // Surface provisional state in the time display label var anyProvisional = stats.some(function (s) { return s.is_provisional; }); var disp = document.getElementById('timeDisplay'); @@ -546,9 +564,10 @@ cmlStats[stat.cml_id] = stat; applyStatsToLine(stat.cml_id, stat); }); + console.log('[fetchAndApplyStats] Applied stats to all lines'); }) .catch(function (err) { - if (err.name !== 'AbortError') console.error('Stats fetch error:', err); + if (err.name !== 'AbortError') console.error('[fetchAndApplyStats] Error:', err); }); } From 1c43035593c99bdb916ca832d47b4d554ce81b64 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 20:54:59 +0200 Subject: [PATCH 20/23] fix: Remove invalid CAGG refresh for cml_stats in write_provisional_snapshot() cml_stats is a regular table updated by update_cml_stats(), not a continuous aggregate. Attempting to refresh it caused 'relation is not a continuous aggregate' errors, preventing provisional snapshots from being written every 60 seconds. This fix allows the parser to continuously populate cml_stats_history with current-hour data, eliminating the 3-4 hour gap of grey/NULL values on the map. --- parser/db_writer.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/parser/db_writer.py b/parser/db_writer.py index 87b6b15..0701caa 100644 --- a/parser/db_writer.py +++ b/parser/db_writer.py @@ -486,11 +486,8 @@ def write_provisional_snapshot(self) -> int: ) cur = self.conn.cursor() try: - # First refresh the 6h rolling view so we have fresh stats - cur.execute( - "CALL refresh_continuous_aggregate('cml_stats', NOW() - INTERVAL '6 hours', NOW())" - ) - self.conn.commit() + # cml_stats is a regular table (not a CAGG), so no refresh needed. + # It's updated continuously by update_cml_stats(). cur.execute( """ From 99a9b8590cf0abc8cf872ae105364f66d34971de Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 21:01:09 +0200 Subject: [PATCH 21/23] test: Update stats snapshot tests to expect CAGG refresh + add coloring annotation API tests - Fixed test_write_stats_snapshot_success to expect 2 execute() calls (CAGG refresh + materialize) - Fixed test_write_stats_snapshot_rollback_on_error to only fail on materialize step - Added test_write_provisional_snapshot_does_not_refresh_cml_stats to verify no invalid CAGG refresh - Added new test_coloring_annotation_api.py with 5 tests for /api/coloring-annotation endpoints --- parser/tests/test_stats_snapshot.py | 59 +++++-- .../tests/test_coloring_annotation_api.py | 148 ++++++++++++++++++ 2 files changed, 194 insertions(+), 13 deletions(-) create mode 100644 webserver/tests/test_coloring_annotation_api.py diff --git a/parser/tests/test_stats_snapshot.py b/parser/tests/test_stats_snapshot.py index a49179b..28159ff 100644 --- a/parser/tests/test_stats_snapshot.py +++ b/parser/tests/test_stats_snapshot.py @@ -32,33 +32,46 @@ def test_write_stats_snapshot_success(mock_connection): rows = writer.write_stats_snapshot(at_time) assert rows == 42 - mock_connection.cursor.assert_called_once() cursor = mock_connection.cursor.return_value - cursor.execute.assert_called_once() - # Verify the call uses hour-truncated time - call_args = cursor.execute.call_args[0] - sql = call_args[0] - params = call_args[1] + # Should call execute twice: CAGG refresh + materialize + assert cursor.execute.call_count == 2 - assert "materialize_cml_stats_snapshot" in sql - assert params[1] == "demo_openmrg" + # First call: refresh continuous aggregate + first_call_sql = cursor.execute.call_args_list[0][0][0] + assert "refresh_continuous_aggregate" in first_call_sql + assert "cml_data_1h" in first_call_sql + + # Second call: materialize snapshot + second_call_sql = cursor.execute.call_args_list[1][0][0] + second_call_params = cursor.execute.call_args_list[1][0][1] + + assert "materialize_cml_stats_snapshot" in second_call_sql + assert second_call_params[1] == "demo_openmrg" # Check that time is truncated to hour - assert params[0].minute == 0 - assert params[0].second == 0 - assert params[0].microsecond == 0 + assert second_call_params[0].minute == 0 + assert second_call_params[0].second == 0 + assert second_call_params[0].microsecond == 0 def test_write_stats_snapshot_rollback_on_error(mock_connection): - """Test write_stats_snapshot rolls back on error.""" + """Test write_stats_snapshot rolls back on error during materialize.""" writer = DBWriter("postgresql://test", user_id="demo_openmrg") writer.conn = mock_connection - mock_connection.cursor.return_value.execute.side_effect = Exception("DB error") + # Let CAGG refresh succeed, fail on materialize + def execute_side_effect(sql, *args): + if "materialize_cml_stats_snapshot" in sql: + raise Exception("DB error") + + mock_connection.cursor.return_value.execute.side_effect = execute_side_effect with pytest.raises(Exception, match="DB error"): writer.write_stats_snapshot(datetime.now(tz=timezone.utc)) + # Rollback should be called once after the error + mock_connection.rollback.assert_called_once() + mock_connection.rollback.assert_called_once() @@ -90,6 +103,26 @@ def test_write_provisional_snapshot_success(mock_connection): assert params[0].second == 0 +def test_write_provisional_snapshot_does_not_refresh_cml_stats(mock_connection): + """Test provisional snapshot does NOT try to refresh cml_stats (it's a table, not CAGG).""" + writer = DBWriter("postgresql://test", user_id="demo_openmrg") + writer.conn = mock_connection + + with patch('parser.db_writer.datetime') as mock_datetime: + mock_now = datetime(2026, 7, 25, 14, 30, 0, tzinfo=timezone.utc) + mock_datetime.now.return_value = mock_now + mock_datetime.timezone = timezone + + rows = writer.write_provisional_snapshot() + + cursor = mock_connection.cursor.return_value + + # Verify NO refresh_continuous_aggregate call for cml_stats + for call in cursor.execute.call_args_list: + sql = call[0][0] if call[0] else "" + assert "refresh_continuous_aggregate" not in sql or "cml_stats" not in sql + + def test_write_provisional_snapshot_uses_cml_stats_table(mock_connection): """Test provisional snapshot copies from cml_stats table.""" writer = DBWriter("postgresql://test", user_id="demo_openmrg") diff --git a/webserver/tests/test_coloring_annotation_api.py b/webserver/tests/test_coloring_annotation_api.py new file mode 100644 index 0000000..36c8b9e --- /dev/null +++ b/webserver/tests/test_coloring_annotation_api.py @@ -0,0 +1,148 @@ +"""Tests for the coloring annotation API endpoints.""" + +import json +import sys +from contextlib import contextmanager +from unittest.mock import Mock, patch, MagicMock + +import pytest + +# Ensure optional heavy imports won't fail at import time +sys.modules.setdefault("folium", Mock()) + + +@pytest.fixture +def auth_client(monkeypatch): + """Test client with login bypassed.""" + import os + sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + import main as wm + + mock_user = Mock() + mock_user.id = "demo_openmrg" + + monkeypatch.setattr(wm, "current_user", mock_user) + monkeypatch.setitem(wm.app.config, "LOGIN_DISABLED", True) + wm.app.config["TESTING"] = True + + return wm.app.test_client() + + +class TestColoringAnnotationAPI: + """Tests for /api/coloring-annotation endpoint.""" + + def test_create_annotation_success(self, auth_client, monkeypatch): + """Test POST /api/coloring-annotation creates annotation via Grafana API.""" + import requests + + epoch_sec = 1753632000 # 2026-07-27 16:00:00 UTC + + # Mock requests.post response + mock_resp = Mock() + mock_resp.json.return_value = {"id": 42, "message": "Annotation added"} + mock_resp.status_code = 200 + + with patch('main.requests.post', return_value=mock_resp) as mock_post: + resp = auth_client.post( + "/api/coloring-annotation", + data=json.dumps({"epochSec": epoch_sec}), + content_type="application/json" + ) + + assert resp.status_code == 200 + data = resp.get_json() + assert data["status"] == "created" + assert data["id"] == 42 + + # Verify request payload + call_kwargs = mock_post.call_args[1] + body = call_kwargs["json"] + assert body["time"] == (epoch_sec - 3600) * 1000 + assert body["timeEnd"] == epoch_sec * 1000 + assert body["tags"] == ["cml-coloring-window"] + + def test_update_annotation_success(self, auth_client, monkeypatch): + """Test POST /api/coloring-annotation updates existing annotation.""" + import requests + + epoch_sec = 1753635600 # 2026-07-27 17:00:00 UTC + existing_id = 42 + + mock_resp = Mock() + mock_resp.json.return_value = {"id": existing_id, "message": "Annotation updated"} + mock_resp.status_code = 200 + + with patch('main.requests.patch', return_value=mock_resp) as mock_patch: + resp = auth_client.post( + "/api/coloring-annotation", + data=json.dumps({"epochSec": epoch_sec, "id": existing_id}), + content_type="application/json" + ) + + assert resp.status_code == 200 + data = resp.get_json() + assert data["status"] == "updated" + assert data["id"] == existing_id + + def test_delete_annotation_success(self, auth_client, monkeypatch): + """Test DELETE /api/coloring-annotation removes annotation.""" + ann_id = 42 + + mock_resp = Mock() + mock_resp.status_code = 200 + + with patch('main.requests.delete', return_value=mock_resp) as mock_del: + resp = auth_client.delete( + "/api/coloring-annotation", + data=json.dumps({"id": ann_id}), + content_type="application/json" + ) + + assert resp.status_code == 200 + data = resp.get_json() + assert data["status"] == "deleted" + + def test_missing_epochsec_returns_400(self, auth_client, monkeypatch): + """Test POST without epochSec returns 400 error.""" + resp = auth_client.post( + "/api/coloring-annotation", + data=json.dumps({}), + content_type="application/json" + ) + + assert resp.status_code == 400 + data = resp.get_json() + assert "error" in data + + +class TestColoringAnnotationCleanup: + """Tests for /api/coloring-annotation-cleanup endpoint.""" + + def test_cleanup_removes_all_tagged_annotations(self, auth_client, monkeypatch): + """Test POST /api/coloring-annotation-cleanup purges all cml-coloring-window annotations.""" + import requests + + # Mock GET returning existing annotations + mock_get_resp = Mock() + mock_get_resp.json.return_value = [{"id": 1}, {"id": 2}, {"id": 3}] + mock_get_resp.status_code = 200 + + # Mock DELETE responses + mock_del_resp = Mock() + mock_del_resp.status_code = 200 + + with patch('main.requests.get', return_value=mock_get_resp) as mock_get: + with patch('main.requests.delete', return_value=mock_del_resp) as mock_del: + resp = auth_client.post("/api/coloring-annotation-cleanup") + + assert resp.status_code == 200 + data = resp.get_json() + assert data["status"] == "ok" + + # Verify GET was called with correct params + mock_get.assert_called_once() + assert "tags=cml-coloring-window" in mock_get.call_args[1]["params"] + + # Verify 3 DELETEs were called + assert mock_del.call_count == 3 + From ab1921c0e998df394dc0dcda1a89ce4f8ca17773 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 21:05:16 +0200 Subject: [PATCH 22/23] test: Fix CI failures in coloring annotation and cml stats tests - test_api_cml_stats: Updated mock data to 11 columns (added mean_rsl_1h, is_provisional) - test_create_annotation_success: Use call_args.kwargs instead of deprecated indexing - test_cleanup_removes_all_tagged_annotations: Fix params assertion to use .get() method --- webserver/tests/test_api_cml_stats.py | 10 ++-- .../tests/test_coloring_annotation_api.py | 57 ++++++++++--------- 2 files changed, 35 insertions(+), 32 deletions(-) diff --git a/webserver/tests/test_api_cml_stats.py b/webserver/tests/test_api_cml_stats.py index 6c903d4..9e20c46 100644 --- a/webserver/tests/test_api_cml_stats.py +++ b/webserver/tests/test_api_cml_stats.py @@ -24,8 +24,8 @@ def test_api_cml_stats_returns_cached_stats(monkeypatch): mock_cursor = Mock() mock_conn.cursor.return_value = mock_cursor - # Row fields: cml_id, completeness_percent_6h, total_records_6h, valid_records_6h, - # mean_rsl_6h, stddev_rsl_6h, completeness_percent_1h, stddev_rsl_1h, last_rsl + # Row fields (11 columns): cml_id, completeness_percent_6h, total_records_6h, valid_records_6h, + # mean_rsl_6h, stddev_rsl_6h, completeness_percent_1h, mean_rsl_1h, stddev_rsl_1h, last_rsl, is_provisional mock_cursor.fetchall.return_value = [ ( "10001", @@ -35,8 +35,10 @@ def test_api_cml_stats_returns_cached_stats(monkeypatch): -50.0, # mean_rsl_6h 3.0, # stddev_rsl_6h 90.0, # completeness_percent_1h + -48.5, # mean_rsl_1h 1.3, # stddev_rsl_1h -45.0, # last_rsl + False, # is_provisional ) ] @@ -44,8 +46,7 @@ def test_api_cml_stats_returns_cached_stats(monkeypatch): mock_cursor.close = Mock() mock_conn.close = Mock() - # The route now uses user_db_scope(current_user.id) instead of get_db_connection(). - # Mock user_db_scope to yield the mock connection, and disable login enforcement. + # The route uses user_db_scope(current_user.id). Mock it and disable login. @contextmanager def mock_user_db_scope(user_id): yield mock_conn @@ -69,3 +70,4 @@ def mock_user_db_scope(user_id): assert row["completeness_percent_1h"] == 90.0 assert row["stddev_last_60min"] == 1.3 assert row["last_rsl"] == -45.0 + assert row["is_provisional"] == False diff --git a/webserver/tests/test_coloring_annotation_api.py b/webserver/tests/test_coloring_annotation_api.py index 36c8b9e..8076595 100644 --- a/webserver/tests/test_coloring_annotation_api.py +++ b/webserver/tests/test_coloring_annotation_api.py @@ -15,26 +15,25 @@ def auth_client(monkeypatch): """Test client with login bypassed.""" import os + sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) import main as wm - + mock_user = Mock() mock_user.id = "demo_openmrg" - + monkeypatch.setattr(wm, "current_user", mock_user) monkeypatch.setitem(wm.app.config, "LOGIN_DISABLED", True) wm.app.config["TESTING"] = True - + return wm.app.test_client() class TestColoringAnnotationAPI: """Tests for /api/coloring-annotation endpoint.""" - + def test_create_annotation_success(self, auth_client, monkeypatch): """Test POST /api/coloring-annotation creates annotation via Grafana API.""" - import requests - epoch_sec = 1753632000 # 2026-07-27 16:00:00 UTC # Mock requests.post response @@ -55,61 +54,65 @@ def test_create_annotation_success(self, auth_client, monkeypatch): assert data["id"] == 42 # Verify request payload - call_kwargs = mock_post.call_args[1] + assert mock_post.called + call_kwargs = mock_post.call_args.kwargs body = call_kwargs["json"] assert body["time"] == (epoch_sec - 3600) * 1000 assert body["timeEnd"] == epoch_sec * 1000 assert body["tags"] == ["cml-coloring-window"] - + def test_update_annotation_success(self, auth_client, monkeypatch): """Test POST /api/coloring-annotation updates existing annotation.""" import requests - + epoch_sec = 1753635600 # 2026-07-27 17:00:00 UTC existing_id = 42 - + mock_resp = Mock() - mock_resp.json.return_value = {"id": existing_id, "message": "Annotation updated"} + mock_resp.json.return_value = { + "id": existing_id, + "message": "Annotation updated", + } mock_resp.status_code = 200 - - with patch('main.requests.patch', return_value=mock_resp) as mock_patch: + + with patch("main.requests.patch", return_value=mock_resp) as mock_patch: resp = auth_client.post( "/api/coloring-annotation", data=json.dumps({"epochSec": epoch_sec, "id": existing_id}), - content_type="application/json" + content_type="application/json", ) - + assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "updated" assert data["id"] == existing_id - + def test_delete_annotation_success(self, auth_client, monkeypatch): """Test DELETE /api/coloring-annotation removes annotation.""" ann_id = 42 - + mock_resp = Mock() mock_resp.status_code = 200 - - with patch('main.requests.delete', return_value=mock_resp) as mock_del: + + with patch("main.requests.delete", return_value=mock_resp) as mock_del: resp = auth_client.delete( "/api/coloring-annotation", data=json.dumps({"id": ann_id}), - content_type="application/json" + content_type="application/json", ) - + assert resp.status_code == 200 data = resp.get_json() assert data["status"] == "deleted" - + def test_missing_epochsec_returns_400(self, auth_client, monkeypatch): """Test POST without epochSec returns 400 error.""" resp = auth_client.post( "/api/coloring-annotation", data=json.dumps({}), - content_type="application/json" + content_type="application/json", ) - + assert resp.status_code == 400 data = resp.get_json() assert "error" in data @@ -120,8 +123,6 @@ class TestColoringAnnotationCleanup: def test_cleanup_removes_all_tagged_annotations(self, auth_client, monkeypatch): """Test POST /api/coloring-annotation-cleanup purges all cml-coloring-window annotations.""" - import requests - # Mock GET returning existing annotations mock_get_resp = Mock() mock_get_resp.json.return_value = [{"id": 1}, {"id": 2}, {"id": 3}] @@ -141,8 +142,8 @@ def test_cleanup_removes_all_tagged_annotations(self, auth_client, monkeypatch): # Verify GET was called with correct params mock_get.assert_called_once() - assert "tags=cml-coloring-window" in mock_get.call_args[1]["params"] + params = mock_get.call_args.kwargs.get("params", {}) + assert params.get("tags") == "cml-coloring-window" # Verify 3 DELETEs were called assert mock_del.call_count == 3 - From b8dc6a82ca57b907aab05c2ba5b555094e13c156 Mon Sep 17 00:00:00 2001 From: Christian Chwala Date: Mon, 27 Jul 2026 21:14:35 +0200 Subject: [PATCH 23/23] test: Mock requests.get in test_create_annotation_success (endpoint cleans up duplicates first) --- .../tests/test_coloring_annotation_api.py | 50 +++++++++++-------- 1 file changed, 28 insertions(+), 22 deletions(-) diff --git a/webserver/tests/test_coloring_annotation_api.py b/webserver/tests/test_coloring_annotation_api.py index 8076595..9fad241 100644 --- a/webserver/tests/test_coloring_annotation_api.py +++ b/webserver/tests/test_coloring_annotation_api.py @@ -36,30 +36,36 @@ def test_create_annotation_success(self, auth_client, monkeypatch): """Test POST /api/coloring-annotation creates annotation via Grafana API.""" epoch_sec = 1753632000 # 2026-07-27 16:00:00 UTC + # Mock requests.get for duplicate cleanup (returns empty list) + mock_get_resp = Mock() + mock_get_resp.json.return_value = [] + mock_get_resp.ok = True + # Mock requests.post response - mock_resp = Mock() - mock_resp.json.return_value = {"id": 42, "message": "Annotation added"} - mock_resp.status_code = 200 + mock_post_resp = Mock() + mock_post_resp.json.return_value = {"id": 42, "message": "Annotation added"} + mock_post_resp.status_code = 200 - with patch('main.requests.post', return_value=mock_resp) as mock_post: - resp = auth_client.post( - "/api/coloring-annotation", - data=json.dumps({"epochSec": epoch_sec}), - content_type="application/json" - ) - - assert resp.status_code == 200 - data = resp.get_json() - assert data["status"] == "created" - assert data["id"] == 42 - - # Verify request payload - assert mock_post.called - call_kwargs = mock_post.call_args.kwargs - body = call_kwargs["json"] - assert body["time"] == (epoch_sec - 3600) * 1000 - assert body["timeEnd"] == epoch_sec * 1000 - assert body["tags"] == ["cml-coloring-window"] + with patch('main.requests.get', return_value=mock_get_resp): + with patch('main.requests.post', return_value=mock_post_resp) as mock_post: + resp = auth_client.post( + "/api/coloring-annotation", + data=json.dumps({"epochSec": epoch_sec}), + content_type="application/json" + ) + + assert resp.status_code == 200 + data = resp.get_json() + assert data["status"] == "created" + assert data["id"] == 42 + + # Verify request payload + assert mock_post.called + call_kwargs = mock_post.call_args.kwargs + body = call_kwargs["json"] + assert body["time"] == (epoch_sec - 3600) * 1000 + assert body["timeEnd"] == epoch_sec * 1000 + assert body["tags"] == ["cml-coloring-window"] def test_update_annotation_success(self, auth_client, monkeypatch): """Test POST /api/coloring-annotation updates existing annotation."""