diff --git a/database/init.sql b/database/init.sql index 847e1fe..e1b695c 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,180 @@ 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 (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 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: aggregate across all sublinks for the bucket immediately before p_at_time + LEFT JOIN LATERAL ( + 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 + 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 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/015_add_cml_stats_history.sql b/database/migrations/015_add_cml_stats_history.sql new file mode 100644 index 0000000..a270033 --- /dev/null +++ b/database/migrations/015_add_cml_stats_history.sql @@ -0,0 +1,185 @@ +-- 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. +-- +-- 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 +-- +-- 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/015_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 + -- (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; +$$; + +-- 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; 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..a39e40e --- /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/015_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 015 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/grafana/provisioning/dashboards/definitions/cml-realtime.json b/grafana/provisioning/dashboards/definitions/cml-realtime.json index 27a06e6..bea7728 100644 --- a/grafana/provisioning/dashboards/definitions/cml-realtime.json +++ b/grafana/provisioning/dashboards/definitions/cml-realtime.json @@ -9,7 +9,7 @@ }, "enable": true, "hide": true, - "iconColor": "rgba(0, 211, 255, 1)", + "iconColor": "rgba(120, 120, 120, 0.6)", "name": "Annotations & Alerts", "type": "dashboard" } @@ -1002,6 +1002,15 @@ "queryValue": "", "skipUrlSync": true, "type": "custom" + }, + { + "name": "marker_time", + "type": "textbox", + "label": "Marker time (epoch ms)", + "hide": 2, + "current": { + "value": "0" + } } ] }, @@ -1010,7 +1019,7 @@ "to": "now" }, "timepicker": {}, - "timezone": "", + "timezone": "utc", "title": "CML Real-Time Data", "uid": "cml-realtime", "version": 1, diff --git a/parser/backfill_stats_history.py b/parser/backfill_stats_history.py new file mode 100644 index 0000000..1dbead2 --- /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 015 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..0701caa 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,130 @@ 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: + # 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( + "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: + # cml_stats is a regular table (not a CAGG), so no refresh needed. + # It's updated continuously by update_cml_stats(). + + 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( diff --git a/parser/tests/test_stats_snapshot.py b/parser/tests/test_stats_snapshot.py new file mode 100644 index 0000000..28159ff --- /dev/null +++ b/parser/tests/test_stats_snapshot.py @@ -0,0 +1,157 @@ +"""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 + cursor = mock_connection.cursor.return_value + + # Should call execute twice: CAGG refresh + materialize + assert cursor.execute.call_count == 2 + + # 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 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 during materialize.""" + writer = DBWriter("postgresql://test", user_id="demo_openmrg") + writer.conn = mock_connection + + # 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() + + +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_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") + 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/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..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, @@ -27,7 +28,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 @@ -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") @@ -522,48 +528,182 @@ 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_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() - 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: query cml_stats_history via the validated timestamp + cur.execute( + "SELECT * FROM get_cml_stats_at(%s::timestamptz, %s)", + (at_ts, current_user.id), + ) + else: + # 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, + mean_rsl_1h, + stddev_rsl_1h, + last_rsl, + FALSE as is_provisional + FROM cml_stats + ORDER BY cml_id + """ + ) + data = cur.fetchall() cur.close() - stats = [ - { - "cml_id": str(row[0]), - "completeness_percent": safe_float(row[1]), # 6h window - "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 - "last_rsl": safe_float(row[8]), - } - 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}") 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/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(): diff --git a/webserver/templates/realtime.html b/webserver/templates/realtime.html index 4cbcf33..929ab6c 100644 --- a/webserver/templates/realtime.html +++ b/webserver/templates/realtime.html @@ -63,6 +63,8 @@ background: white; } + + .resizer { position: absolute; bottom: 400px; @@ -115,7 +117,19 @@ 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; + let sliderDebounce = null; // debounce timer for slider input + let sliderFetchCtrl = null; // AbortController for in-flight slider requests + + // Grafana annotation state + var coloringAnnotationId = null; + var annotationAbortCtrl = null; + var annotationRefreshTimer = null; // Color functions for each option const colorFunctions = { @@ -137,13 +151,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) } }; @@ -226,12 +242,36 @@ var select = L.DomUtil.create('select', '', container); select.id = 'coloringOption'; select.innerHTML = ` - + - + `; - // 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); @@ -274,6 +314,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 +332,81 @@ updateMapColors(); }); } + + // Time slider event handlers + var liveBtn = document.getElementById('timeLiveBtn'); + 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); + + 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); + }); + } + + 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 () { + var epochSec = parseInt(this.value, 10); + var dt = new Date(epochSec * 1000); + + 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 >= nowHour) label += ' (partial)'; + document.getElementById('timeDisplay').textContent = label; + + 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'; + }); + } } // Load and draw CML data @@ -421,6 +541,191 @@ } } + // Fetch and apply CML stats with optional historical time + function fetchAndApplyStats(atIso, signal) { + 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) { + 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'); + 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); + }); + console.log('[fetchAndApplyStats] Applied stats to all lines'); + }) + .catch(function (err) { + if (err.name !== 'AbortError') console.error('[fetchAndApplyStats] Error:', err); + }); + } + + // Sync Grafana iframe time with selected time + function syncGrafanaTime(epochSec) { + var grafanaPanel = document.getElementById('grafana-panel'); + var newUrl; + try { + // 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 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 = 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')) { + 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; + } + } + + 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) { + // 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) { + 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; @@ -438,6 +743,7 @@ document.addEventListener('DOMContentLoaded', function () { initializeMap(); initializeResizer(); + cleanupColoringAnnotations(); }); // Initialize resizer for adjustable dashboard height 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 new file mode 100644 index 0000000..9fad241 --- /dev/null +++ b/webserver/tests/test_coloring_annotation_api.py @@ -0,0 +1,155 @@ +"""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.""" + 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_post_resp = Mock() + mock_post_resp.json.return_value = {"id": 42, "message": "Annotation added"} + mock_post_resp.status_code = 200 + + 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.""" + 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.""" + # 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() + 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 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