diff --git a/.gitignore b/.gitignore index 82f9275..98f2d3b 100644 --- a/.gitignore +++ b/.gitignore @@ -160,3 +160,4 @@ cython_debug/ # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. #.idea/ +database/archive_data/ diff --git a/database/migrations/010_add_rain_data_table.sql b/database/migrations/010_add_rain_data_table.sql new file mode 100644 index 0000000..562f175 --- /dev/null +++ b/database/migrations/010_add_rain_data_table.sql @@ -0,0 +1,46 @@ +-- Migration 010: Add cml_rain_data table for processed rain rate estimates +-- This table stores the processed rain rate estimates and intermediate products from CML data + +-- Create the main table +CREATE TABLE cml_rain_data ( + time TIMESTAMPTZ NOT NULL, + cml_id TEXT NOT NULL, + sublink_id TEXT NOT NULL, + user_id TEXT NOT NULL, + tl REAL, -- Total loss (TSL - RSL) + wet BOOLEAN, -- Wet/dry classification + baseline REAL, -- Baseline attenuation + waa REAL, -- Wet antenna attenuation + a_rain REAL, -- Rain-induced path attenuation + r REAL, -- Rain rate estimate (mm/h) + PRIMARY KEY (time, cml_id, sublink_id, user_id) +); + +-- Convert to hypertable (TimescaleDB) +SELECT create_hypertable('cml_rain_data', 'time'); + +-- Enable compression (same strategy as cml_data) +ALTER TABLE cml_rain_data SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'user_id, cml_id' +); + +-- Add compression policy (compress chunks older than 7 days) +SELECT add_compression_policy('cml_rain_data', INTERVAL '7 days'); + +-- Add Row-Level Security (RLS) for multi-user isolation +ALTER TABLE cml_rain_data ENABLE ROW LEVEL SECURITY; + +-- Create policy ensuring users only see their own data +CREATE POLICY cml_rain_data_user_policy ON cml_rain_data + USING (user_id = current_user); + +-- Grant permissions to webserver_role for admin access +GRANT SELECT, INSERT ON cml_rain_data TO webserver_role; + +-- Create security-barrier view for safe access +CREATE VIEW cml_rain_data_secure WITH (security_barrier) AS + SELECT * FROM cml_rain_data + WHERE user_id = current_user; + +GRANT SELECT ON cml_rain_data_secure TO webserver_role; diff --git a/database/migrations/011_add_rain_stats_view.sql b/database/migrations/011_add_rain_stats_view.sql new file mode 100644 index 0000000..67161d6 --- /dev/null +++ b/database/migrations/011_add_rain_stats_view.sql @@ -0,0 +1,49 @@ +-- Migration 011: Add cml_rain_stats view for rain rate statistics +-- This view provides aggregated statistics for rain rate data, similar to cml_stats + +-- Create aggregated statistics view for rain rates +CREATE OR REPLACE VIEW cml_rain_stats AS +SELECT + r.cml_id::text, + r.user_id, + COUNT(*) AS total_records, + COUNT(CASE WHEN r.r IS NOT NULL AND r.r > 0 THEN 1 END) AS valid_records, + ROUND( + 100.0 * COUNT(CASE WHEN r.r IS NOT NULL AND r.r > 0 THEN 1 END) / COUNT(*), + 2 + ) AS completeness_percent, + ROUND(AVG(r.r)::numeric, 2) AS mean_rain_rate, + ROUND(STDDEV(r.r)::numeric, 2) AS stddev_rain_rate, + MAX(r.r) AS max_rain_rate, + -- Last rain rate (most recent non-zero value) + ( + SELECT r2.r FROM cml_rain_data r2 + WHERE r2.cml_id = r.cml_id + AND r2.user_id = r.user_id + AND r2.r IS NOT NULL AND r2.r > 0 + ORDER BY r2.time DESC LIMIT 1 + ) AS last_rain_rate, + -- 6-hour window statistics + ROUND( + 100.0 * COUNT(CASE WHEN r.r IS NOT NULL AND r.r > 0 AND r.time >= NOW() - INTERVAL '6 hours' THEN 1 END) + / NULLIF(COUNT(*) FILTER (WHERE r.time >= NOW() - INTERVAL '6 hours'), 0), + 2 + ) AS completeness_percent_6h, + COUNT(*) FILTER (WHERE r.time >= NOW() - INTERVAL '6 hours') AS total_records_6h, + COUNT(CASE WHEN r.r IS NOT NULL AND r.r > 0 AND r.time >= NOW() - INTERVAL '6 hours' THEN 1 END) AS valid_records_6h, + ROUND(AVG(r.r) FILTER (WHERE r.time >= NOW() - INTERVAL '6 hours')::numeric, 2) AS mean_rain_rate_6h, + ROUND(STDDEV(r.r) FILTER (WHERE r.time >= NOW() - INTERVAL '6 hours')::numeric, 2) AS stddev_rain_rate_6h, + -- 1-hour window statistics + ROUND( + 100.0 * COUNT(CASE WHEN r.r IS NOT NULL AND r.r > 0 AND r.time >= NOW() - INTERVAL '1 hour' THEN 1 END) + / NULLIF(COUNT(*) FILTER (WHERE r.time >= NOW() - INTERVAL '1 hour'), 0), + 2 + ) AS completeness_percent_1h, + ROUND(AVG(r.r) FILTER (WHERE r.time >= NOW() - INTERVAL '1 hour')::numeric, 2) AS mean_rain_rate_1h, + ROUND(STDDEV(r.r) FILTER (WHERE r.time >= NOW() - INTERVAL '1 hour')::numeric, 2) AS stddev_rain_rate_1h, + NOW() AS last_update +FROM cml_rain_data r +GROUP BY r.cml_id, r.user_id; + +-- Grant permissions to webserver_role for admin access +GRANT SELECT ON cml_rain_stats TO webserver_role; diff --git a/database/migrations/012_add_rain_data_1h.sql b/database/migrations/012_add_rain_data_1h.sql new file mode 100644 index 0000000..9e3c573 --- /dev/null +++ b/database/migrations/012_add_rain_data_1h.sql @@ -0,0 +1,56 @@ +-- Migration 012: Add continuous aggregate for rain data (1-hour buckets) +-- Similar to cml_data_1h, this provides efficient querying for Grafana dashboards + +-- Create 1-hour continuous aggregate for rain data +CREATE MATERIALIZED VIEW cml_rain_data_1h +WITH (timescaledb.continuous) AS +SELECT + time_bucket('1 hour', time) AS bucket, + user_id, + cml_id, + sublink_id, + MIN(r) AS r_min, + MAX(r) AS r_max, + AVG(r) AS r_avg, + MIN(tl) AS tl_min, + MAX(tl) AS tl_max, + AVG(tl) AS tl_avg, + MIN(a_rain) AS a_rain_min, + MAX(a_rain) AS a_rain_max, + AVG(a_rain) AS a_rain_avg +FROM cml_rain_data +GROUP BY bucket, user_id, cml_id, sublink_id +WITH NO DATA; + +-- Automatically refresh every hour, covering up to 2 days of history +SELECT add_continuous_aggregate_policy('cml_rain_data_1h', + start_offset => INTERVAL '2 days', + end_offset => INTERVAL '1 hour', + schedule_interval => INTERVAL '1 hour' +); + +-- Enable compression for chunks older than 7 days +ALTER TABLE cml_rain_data_1h SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'user_id, cml_id', + timescaledb.compress_orderby = 'bucket' +); + +SELECT add_compression_policy('cml_rain_data_1h', INTERVAL '7 days'); + +-- Enable Row-Level Security +ALTER TABLE cml_rain_data_1h ENABLE ROW LEVEL SECURITY; + +-- RLS policy for per-user isolation +CREATE POLICY cml_rain_data_1h_user_policy ON cml_rain_data_1h + USING (user_id = current_user); + +-- Grant permissions +GRANT SELECT ON cml_rain_data_1h TO webserver_role; + +-- Create security-barrier view for safe access +CREATE VIEW cml_rain_data_1h_secure WITH (security_barrier) AS + SELECT * FROM cml_rain_data_1h + WHERE user_id = current_user; + +GRANT SELECT ON cml_rain_data_1h_secure TO webserver_role; diff --git a/docker-compose.yml b/docker-compose.yml index 48c2b44..bea1262 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -138,17 +138,38 @@ services: processor: build: ./processor + restart: unless-stopped + depends_on: + database: + condition: service_healthy + volumes: + - ./processor/config:/app/config:ro + - processor_state:/app/data/state + environment: + - DATABASE_URL=postgresql://myuser:mypassword@database:5432/mydatabase + + # Jupyter notebook for rain processing development and validation + processor_notebook: + build: + context: ./processor + dockerfile: Dockerfile.jupyter ports: - - "5002:5002" + - "9888:8888" depends_on: - - database + database: + condition: service_healthy + volumes: + - ./processor:/app:ro + - processor_notebook_data:/app/data environment: - DATABASE_URL=postgresql://myuser:mypassword@database:5432/mydatabase + profiles: + - notebook webserver: build: ./webserver ports: - - "5000:5000" + - "5001:5000" depends_on: - database - sftp_receiver @@ -274,4 +295,6 @@ volumes: grafana_data: mno_openmrg_data_to_upload: mno_orange_cameroun_data_to_upload: + processor_state: + processor_notebook_data: # minio_data: # Uncomment if using MinIO \ No newline at end of file diff --git a/grafana/provisioning/dashboards/definitions/cml-rain-rates.json b/grafana/provisioning/dashboards/definitions/cml-rain-rates.json new file mode 100644 index 0000000..d3cc628 --- /dev/null +++ b/grafana/provisioning/dashboards/definitions/cml-rain-rates.json @@ -0,0 +1,278 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "editable": false, + "fiscalYearStartMonth": 0, + "graphTooltip": 0, + "id": null, + "links": [], + "liveNow": false, + "panels": [ + { + "datasource": { + "type": "grafana-postgresql-datasource", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "custom": { + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + } + } + }, + "overrides": [] + }, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 0 + }, + "id": 4, + "options": { + "text": "Select a CML ID using the dropdown below to view its rain rate data" + }, + "pluginVersion": "9.0.0", + "targets": [], + "type": "text" + }, + { + "datasource": { + "type": "grafana-postgresql-datasource", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Rain Rate (mm/h)", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "lineInterpolation": "linear", + "lineWidth": 2, + "pointSize": 4, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "mm/h" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sublink_1" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "blue", + "mode": "fixed" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sublink_2" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "semi-dark-orange", + "mode": "fixed" + } + } + ] + } + ] + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 1 + }, + "id": 2, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "grafana-postgresql-datasource", + "uid": "${datasource}" + }, + "format": "time_series", + "rawQuery": true, + "rawSql": "SELECT\n time_bucket('$__interval', time) AS \"time\",\n sublink_id AS metric,\n AVG(r) AS value\nFROM cml_rain_data_secure\nWHERE cml_id = '${cml_id}'\n AND time >= $__timeFrom()::timestamptz\n AND time <= $__timeTo()::timestamptz\nGROUP BY 1, 2\nORDER BY 1 ASC", + "refId": "A" + }, + { + "datasource": { + "type": "grafana-postgresql-datasource", + "uid": "${datasource}" + }, + "format": "time_series", + "rawQuery": true, + "rawSql": "SELECT\n time AS \"time\",\n sublink_id AS metric,\n r AS value\nFROM cml_rain_data_secure\nWHERE cml_id = '${cml_id}'\n AND '${interval}' = 'raw'\n AND time >= $__timeFrom()::timestamptz\n AND time <= $__timeTo()::timestamptz\nORDER BY 1 ASC", + "refId": "B" + } + ], + "title": "CML Time Series - Rain Rate", + "type": "timeseries" + } + ], + "refresh": "10s", + "schemaVersion": 38, + "style": "dark", + "tags": [ + "cml", + "rain", + "realtime" + ], + "templating": { + "list": [ + { + "current": {}, + "hide": 2, + "includeAll": false, + "multi": false, + "name": "datasource", + "options": [], + "query": "grafana-postgresql-datasource", + "refresh": 1, + "type": "datasource", + "label": "Datasource" + }, + { + "current": { + "selected": false, + "text": "10001", + "value": "10001" + }, + "datasource": { + "type": "grafana-postgresql-datasource", + "uid": "${datasource}" + }, + "definition": "SELECT DISTINCT cml_id::text FROM cml_metadata ORDER BY cml_id", + "description": null, + "error": null, + "hide": 0, + "includeAll": false, + "label": "CML ID", + "multi": false, + "name": "cml_id", + "options": [], + "query": "SELECT DISTINCT cml_id::text FROM cml_metadata ORDER BY cml_id", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "sort": 0, + "tagValuesQuery": "", + "tagsQuery": "", + "type": "query", + "useTags": false + }, + { + "current": { + "selected": true, + "text": "Auto", + "value": "auto" + }, + "description": "Time aggregation interval for downsampling data", + "hide": 0, + "includeAll": false, + "label": "Interval", + "multi": false, + "name": "interval", + "options": [ + { + "selected": true, + "text": "Auto", + "value": "auto" + }, + { + "selected": false, + "text": "Raw", + "value": "raw" + } + ], + "query": "Auto : auto,Raw : raw", + "queryValue": "", + "skipUrlSync": true, + "type": "custom" + } + ] + }, + "time": { + "from": "now-24h", + "to": "now" + }, + "timepicker": {}, + "timezone": "browser", + "title": "CML Rain Rates", + "uid": "cml-rain-rates", + "version": 1, + "weekStart": "" +} \ No newline at end of file diff --git a/processor/Dockerfile b/processor/Dockerfile index 459ca7c..6db3e28 100644 --- a/processor/Dockerfile +++ b/processor/Dockerfile @@ -2,9 +2,18 @@ FROM python:3.11-slim WORKDIR /app -COPY requirements.txt ./ +# Install dependencies +COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -COPY . . +# Copy source code +COPY *.py ./ +COPY workflows/ ./workflows/ + +# Create directories for config and state +RUN mkdir -p /app/config /app/data/state + +# Set environment variables +ENV PYTHONUNBUFFERED=1 CMD ["python", "main.py"] \ No newline at end of file diff --git a/processor/Dockerfile.jupyter b/processor/Dockerfile.jupyter new file mode 100644 index 0000000..f9d492c --- /dev/null +++ b/processor/Dockerfile.jupyter @@ -0,0 +1,35 @@ +# Build notebook image from scratch with all dependencies +FROM python:3.11-slim + +WORKDIR /app + +# Install all Python dependencies in one layer (with caching) +COPY requirements.txt . +RUN pip install --default-timeout=600 \ + psycopg2-binary \ + pandas \ + xarray \ + pyyaml \ + numpy \ + pycomlink \ + notebook \ + matplotlib + +# Copy source code +COPY *.py ./ +COPY workflows/ ./workflows/ + +# Create directories +RUN mkdir -p /app/config /app/data/state /notebooks + +# Copy notebooks into container +COPY notebooks/*.ipynb /notebooks/ + +# Set environment variables +ENV PYTHONUNBUFFERED=1 + +# Expose Jupyter port +EXPOSE 8888 + +# Start Jupyter notebook server +CMD ["jupyter", "notebook", "--ip=0.0.0.0", "--port=8888", "--no-browser", "--allow-root", "--NotebookApp.token=''", "--NotebookApp.password=''", "--notebook-dir=/notebooks"] diff --git a/processor/RAIN_PROCESSING_IMPLEMENTATION.md b/processor/RAIN_PROCESSING_IMPLEMENTATION.md new file mode 100644 index 0000000..3d1791a --- /dev/null +++ b/processor/RAIN_PROCESSING_IMPLEMENTATION.md @@ -0,0 +1,1656 @@ +# Rain Rate Processing Implementation Plan + +## Overview + +This document describes the implementation of continuous rain rate processing from CML attenuation data. The processor runs as a polling service that: + +1. Reads configuration from YAML file (which users can be enabled, processing frequency, data window) +2. Maintains state (last processed timestamp) in a JSON file +3. Polls the database at configurable intervals +4. Fetches raw CML data (RSL, TSL) and metadata for a sliding time window +5. Applies pluggable processing workflows (different algorithms for different MNO providers) +6. Writes results (rain rates and intermediate products) to a new database table + +**Key Design Decisions:** +- Configuration via YAML (not database) for simplicity and version control +- State tracking via JSON file for persistence across restarts +- Starts processing from "now" when first enabled (no automatic backfill) +- Pluggable workflow architecture to support MNO-specific variants +- Fixed interface between data fetching and processing algorithms +- Processing workflows use `xarray.Dataset` / `xarray.DataArray` as the main in-memory format, because this matches typical `pycomlink` and `poligrain` example workflows better than `pandas` + +**Feasibility:** +- Yes, this is feasible. +- The database read/write layer can still use SQL and optionally `pandas` internally for convenience, but the workflow interface should expose `xarray` objects. +- This is a good fit because CML processing is naturally multi-dimensional (`time`, `cml_id`, `sublink_id`) and many scientific processing examples already assume `xarray`. +- The main extra work is careful conversion between SQL table rows and a well-defined `xarray.Dataset` structure. + +--- + +## Short Step-by-Step Implementation Guide + +This section is intended as a quick guide for an implementing agent. Each step points to the section below where the details are defined. + +1. **Create the database target table for processed rain data** + See: **Database Schema** + +2. **Create the YAML configuration file for enabling/disabling processing and selecting workflow variants** + See: **Configuration Structure** + +3. **Create the JSON state file handling so the processor remembers the last processed timestamp per user** + See: **State File Structure** and **`state_manager.py`** + +4. **Implement the database read/write layer for raw CML data, metadata, and processed rain output** + See: **`data_interface.py`** + +5. **Implement one canonical conversion from database rows to `xarray.Dataset` and back** + See: **`dataset_builder.py`** + +6. **Implement the workflow interface and register workflow variants** + See: **`workflows/base.py`**, **`workflows/default.py`**, **`workflows/openmrg_basic.py`**, and **`registry.py`** + +7. **Implement the continuous polling service** + See: **`main.py`** + +8. **Add notebook-based validation that imports the exact same workflow code as the continuous processor** + See: **Notebook-Based Validation Workflow** + +9. **Prepare the website integration target so processed rain data can later be shown in the same layout as the real-time page** + See: **Planned Website Integration** + +10. **Install dependencies, update Docker configuration, and follow the implementation/testing checklist** + See: **Docker Configuration**, **Requirements**, **Implementation Checklist**, and **Testing Strategy** + +**Recommended implementation order:** +- first make the processor run end-to-end with a minimal workflow +- then validate it in the notebook +- then improve the scientific workflow +- only after that enable continuous processing for selected users + +--- + +## Database Schema + +### New Table: `cml_rain_data` + +This table stores the processed rain rate estimates and intermediate products. + +```sql +CREATE TABLE cml_rain_data ( + time TIMESTAMPTZ NOT NULL, + cml_id TEXT NOT NULL, + sublink_id TEXT NOT NULL, + user_id TEXT NOT NULL, + tl REAL, -- Total loss (TSL - RSL) + wet BOOLEAN, -- Wet/dry classification + baseline REAL, -- Baseline attenuation + waa REAL, -- Wet antenna attenuation + a_rain REAL, -- Rain-induced path attenuation + r REAL, -- Rain rate estimate (mm/h) + PRIMARY KEY (time, cml_id, sublink_id, user_id) +); + +-- Convert to hypertable (TimescaleDB) +SELECT create_hypertable('cml_rain_data', 'time'); + +-- Enable compression (same strategy as cml_data) +ALTER TABLE cml_rain_data SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'user_id, cml_id' +); + +-- Add compression policy (compress chunks older than 7 days) +SELECT add_compression_policy('cml_rain_data', INTERVAL '7 days'); + +-- Add Row-Level Security (RLS) for multi-user isolation +ALTER TABLE cml_rain_data ENABLE ROW LEVEL SECURITY; + +CREATE POLICY cml_rain_data_user_policy ON cml_rain_data + USING (user_id = current_user); + +-- Grant permissions to user roles +-- Note: Assumes user roles already exist (e.g., demo_openmrg, demo_orange_cameroun) +-- and webserver_role exists for admin access +GRANT SELECT, INSERT ON cml_rain_data TO webserver_role; +-- Individual user grants will be added per user (same pattern as cml_data) + +-- Create security-barrier view for safe access +CREATE VIEW cml_rain_data_secure WITH (security_barrier) AS + SELECT * FROM cml_rain_data + WHERE user_id = current_user; + +GRANT SELECT ON cml_rain_data_secure TO webserver_role; +``` + +**Migration File:** Create `database/migrations/010_add_rain_data_table.sql` with the above SQL. + +**Important Notes:** +- The table follows the same multi-user pattern as `cml_data` (user_id column + RLS) +- Compression is enabled to save space (rain data can be voluminous) +- `segmentby` includes `user_id, cml_id` for efficient compression and queries +- RLS ensures users only see their own data +- The security-barrier view provides safe access without bypassing RLS +- Even if workflows use `xarray`, the persisted format remains a normal relational table. The conversion back from `xarray` to rows must be explicit and validated. + +--- + +## Configuration Structure + +### YAML Configuration File + +**Location:** `processor/config/rain_processing.yml` + +**Structure:** +```yaml +# Global settings for the rain processing service +global: + # How often to reload this YAML file (seconds) + config_reload_interval_seconds: 60 + + # Default poll interval if not specified per user (seconds) + default_poll_interval_seconds: 900 # 15 minutes + + # Default data window if not specified per user (minutes) + default_data_window_minutes: 90 + +# Per-user configuration +users: + demo_openmrg: + enabled: true # Enable/disable processing for this user + processing_variant: default # Which workflow to use (maps to processor/workflows/.py) + poll_interval_seconds: 900 # Process every 15 minutes + data_window_minutes: 90 # Fetch 90 minutes of data for context + + demo_orange_cameroun: + enabled: false # Disabled by default + processing_variant: orange_cameroun_v1 + poll_interval_seconds: 60 # Process every 1 minute + data_window_minutes: 120 # Fetch 120 minutes of data + +# To add a new user, simply add an entry under 'users' +# The processor will automatically pick it up on next config reload +``` + +**Validation Rules:** +- `enabled` must be boolean +- `processing_variant` must exist in `processor/workflows/` directory +- `poll_interval_seconds` must be positive integer (recommend >= 60) +- `data_window_minutes` must be positive integer, typically >= `poll_interval_seconds/60` + +**Pitfall Warning:** +- YAML is whitespace-sensitive. Use consistent indentation (2 spaces recommended). +- User IDs in YAML must match `user_id` values in the database exactly. +- If a user is enabled but their workflow variant doesn't exist, processing will fail with error log. + +--- + +## State File Structure + +### JSON State File + +**Location:** `processor/data/state/rain_processing_state.json` + +**Structure:** +```json +{ + "demo_openmrg": { + "last_processed_time": "2026-06-19T10:30:00.000000Z" + }, + "demo_orange_cameroun": { + "last_processed_time": "2026-06-19T11:45:00.000000Z" + } +} +``` + +**Initialization:** +- On first run, if a user has no entry in the state file, initialize `last_processed_time` to `datetime.utcnow().isoformat() + 'Z'` +- This ensures no backfill happens on first enable (starts from "now") + +**Persistence:** +- File is written after each successful processing run +- Use atomic write pattern: write to temp file, then rename (to avoid corruption on crash) +- File must be stored in a Docker volume to persist across container restarts + +**Pitfall Warning:** +- Timestamps must be in ISO 8601 format with timezone (UTC recommended) +- Use file locking to prevent concurrent writes (though only one processor instance should run) +- If state file is deleted/corrupted, system resets to "now" for all users (safe default) + +--- + +## Python Module Structure + +### Directory Layout + +``` +processor/ +├── main.py # Entry point, main polling loop +├── config_loader.py # YAML configuration loading and validation +├── state_manager.py # JSON state file read/write with locking +├── data_interface.py # Database interface for reading/writing CML data +├── dataset_builder.py # Converts SQL query results into canonical xarray datasets +├── registry.py # Maps processing_variant names to workflow classes +├── requirements.txt # Python dependencies +├── Dockerfile # Container definition +├── config/ # Configuration files (mounted volume) +│ └── rain_processing.yml +├── data/ # State persistence (mounted volume) +│ └── state/ +│ └── rain_processing_state.json +└── workflows/ # Pluggable processing implementations + ├── __init__.py + ├── base.py # Abstract base class defining interface + ├── default.py # Default xarray-based implementation using pycomlink + ├── openmrg_basic.py # Basic OpenMRG-style workflow outline + └── orange_cameroun_v1.py # Example MNO-specific variant +``` + +--- + +## Module Specifications + +### 1. `config_loader.py` + +**Purpose:** Load and validate YAML configuration, with periodic reloading. + +**Functions:** + +```python +import yaml +from pathlib import Path +from typing import Dict, Any +from datetime import datetime + +class RainProcessingConfig: + """ + Loads and manages rain processing configuration from YAML. + Supports hot-reloading without service restart. + """ + + def __init__(self, config_path: str = "/app/config/rain_processing.yml"): + """ + Args: + config_path: Path to YAML configuration file + """ + self.config_path = Path(config_path) + self._config: Dict[str, Any] = {} + self._last_loaded: datetime = None + self.load() + + def load(self) -> None: + """ + Load configuration from YAML file. + Validates structure and raises ValueError if invalid. + + Raises: + FileNotFoundError: If config file doesn't exist + yaml.YAMLError: If YAML is malformed + ValueError: If required fields are missing or invalid + """ + # Implementation notes: + # 1. Read YAML file + # 2. Validate 'global' and 'users' sections exist + # 3. Validate each user entry has required fields + # 4. Set self._config and self._last_loaded + pass + + def should_reload(self, reload_interval_seconds: int) -> bool: + """ + Check if enough time has passed to reload config. + + Args: + reload_interval_seconds: Time between reloads + + Returns: + True if config should be reloaded + """ + # Implementation: Check if (now - _last_loaded) >= reload_interval_seconds + pass + + def get_global_config(self) -> Dict[str, Any]: + """Get global configuration section.""" + return self._config.get('global', {}) + + def get_user_config(self, user_id: str) -> Dict[str, Any]: + """ + Get configuration for a specific user. + + Args: + user_id: User identifier + + Returns: + User config dict, or None if user not in config + """ + return self._config.get('users', {}).get(user_id) + + def get_enabled_users(self) -> Dict[str, Dict[str, Any]]: + """ + Get all users where enabled=true. + + Returns: + Dict mapping user_id to user config + """ + users = self._config.get('users', {}) + return {uid: cfg for uid, cfg in users.items() if cfg.get('enabled', False)} +``` + +**Pitfall Warnings:** +- YAML parsing can raise various exceptions. Wrap in try/except and log clearly. +- Missing config file should be fatal error (service can't run without config). +- Invalid YAML should log error and keep previous valid config (don't crash mid-operation). +- File permissions: config file should be readable by the container user. + +--- + +### 2. `state_manager.py` + +**Purpose:** Manage persistent state (last processed timestamps) in JSON file. + +**Functions:** + +```python +import json +import fcntl +from pathlib import Path +from datetime import datetime +from typing import Optional, Dict + +class StateManager: + """ + Manages persistent state for rain processing service. + Uses file locking to ensure atomic updates. + """ + + def __init__(self, state_path: str = "/app/data/state/rain_processing_state.json"): + """ + Args: + state_path: Path to JSON state file + """ + self.state_path = Path(state_path) + self.state_path.parent.mkdir(parents=True, exist_ok=True) + + # Initialize empty state file if it doesn't exist + if not self.state_path.exists(): + self._write_state({}) + + def get_last_processed_time(self, user_id: str) -> Optional[datetime]: + """ + Get the last processed timestamp for a user. + + Args: + user_id: User identifier + + Returns: + Last processed datetime (UTC), or None if user has no state + """ + # Implementation notes: + # 1. Read state file with shared lock + # 2. Parse timestamp string to datetime + # 3. Return None if user not found + pass + + def update_last_processed_time(self, user_id: str, timestamp: datetime) -> None: + """ + Update the last processed timestamp for a user. + Uses atomic write pattern: write to temp file, then rename. + + Args: + user_id: User identifier + timestamp: New last processed time (should be UTC) + """ + # Implementation notes: + # 1. Read current state with exclusive lock + # 2. Update user's timestamp + # 3. Write to temp file + # 4. Rename temp file over original (atomic on POSIX) + # 5. Release lock + pass + + def initialize_user(self, user_id: str, timestamp: Optional[datetime] = None) -> None: + """ + Initialize state for a user if not already present. + Defaults to current time (UTC) to avoid backfill. + + Args: + user_id: User identifier + timestamp: Initial timestamp (defaults to now) + """ + if self.get_last_processed_time(user_id) is None: + if timestamp is None: + timestamp = datetime.utcnow() + self.update_last_processed_time(user_id, timestamp) + + def _read_state(self) -> Dict: + """Read state file with file locking.""" + # Use fcntl.flock(fd, fcntl.LOCK_SH) for shared lock + # Return empty dict if file doesn't exist or is empty + pass + + def _write_state(self, state: Dict) -> None: + """Write state file atomically with file locking.""" + # Use fcntl.flock(fd, fcntl.LOCK_EX) for exclusive lock + # Atomic write: write to .tmp file, then os.rename() + pass +``` + +**Pitfall Warnings:** +- **File locking is critical** if multiple processes might access the file (though only one processor should run). +- Always use UTC for timestamps to avoid timezone confusion. +- State file corruption: If JSON parsing fails, log error and consider it empty (start fresh). +- Atomic writes prevent partial writes on crash: always write to temp file then rename. +- The state directory must be a mounted Docker volume or state won't persist across restarts. + +--- + +### 3. `data_interface.py` + +**Purpose:** Fixed interface for reading raw CML data and metadata, and writing processed rain data. + +**Functions:** + +```python +import pandas as pd +import psycopg2 +from datetime import datetime +from typing import List, Optional + +class CMLDataInterface: + """ + Database interface for CML data operations. + Provides fixed API for workflows to read/write data. + """ + + def __init__(self, database_url: str): + """ + Args: + database_url: PostgreSQL connection string + """ + self.database_url = database_url + + def fetch_raw_cml_data_rows( + self, + user_id: str, + start_time: datetime, + end_time: datetime + ) -> pd.DataFrame: + """ + Fetch raw CML data (RSL, TSL) for a time window as tabular rows. + This is an internal helper for later conversion to xarray. + + Args: + user_id: User identifier + start_time: Window start (inclusive) + end_time: Window end (inclusive) + + Returns: + DataFrame with columns: time, cml_id, sublink_id, user_id, rsl, tsl + Sorted by time ascending + Empty DataFrame if no data found + """ + # Implementation notes: + # 1. Connect to database + # 2. Query: SELECT time, cml_id, sublink_id, user_id, rsl, tsl + # FROM cml_data + # WHERE user_id = %s AND time >= %s AND time <= %s + # ORDER BY time ASC + # 3. Return as pandas DataFrame + # 4. Handle connection errors gracefully (log and raise) + pass + + def fetch_cml_metadata_rows( + self, + user_id: str, + cml_ids: Optional[List[str]] = None + ) -> pd.DataFrame: + """ + Fetch CML metadata (coordinates, frequency, length, polarization) as tabular rows. + + Args: + user_id: User identifier + cml_ids: List of CML IDs to fetch (None = fetch all for user) + + Returns: + DataFrame with columns: cml_id, sublink_id, site_0_lon, site_0_lat, + site_1_lon, site_1_lat, frequency, polarization, length + Empty DataFrame if no metadata found + """ + # Implementation notes: + # 1. Query cml_metadata table + # 2. Filter by user_id and optionally cml_ids + # 3. Return all metadata fields + # 4. Handle case where metadata might be missing (return empty df, don't crash) + pass + + def write_rain_data(self, rain_df: pd.DataFrame) -> int: + """ + Write processed rain data to cml_rain_data table. + + Args: + rain_df: DataFrame with columns: time, cml_id, sublink_id, user_id, + tl, wet, baseline, waa, a_rain, r + + Returns: + Number of rows written + + Raises: + ValueError: If required columns are missing + psycopg2.Error: On database errors + """ + # Implementation notes: + # 1. Validate required columns exist + # 2. Use batch INSERT with ON CONFLICT DO UPDATE (upsert) + # to handle duplicate timestamps gracefully + # 3. Consider using COPY for performance if DataFrame is large (>1000 rows) + # 4. Commit transaction + # 5. Return row count + pass + + def close(self): + """Close database connection.""" + pass +``` + +**Pitfall Warnings:** +- **SQL injection**: Always use parameterized queries (`%s` placeholders), never string formatting. +- **Connection management**: Consider using connection pooling or context managers for clean resource handling. +- **Empty results**: Workflows must handle empty DataFrames gracefully (no data in time window). +- **Timezone awareness**: Database stores `TIMESTAMPTZ` (timezone-aware). Ensure datetime objects are UTC. +- **NULL handling**: RSL/TSL can be NULL in database. Workflows must handle NaN values in pandas. +- **ON CONFLICT**: Use `ON CONFLICT (time, cml_id, sublink_id, user_id) DO NOTHING` or `DO UPDATE` to avoid errors on re-processing same time window. + +### 4. `dataset_builder.py` + +**Purpose:** Convert SQL query results into a canonical `xarray.Dataset` structure for workflows. + +This module is important because the less capable agent may otherwise mix ad-hoc `xarray` layouts between workflows. The layout must be fixed early and reused everywhere. + +**Canonical dataset design:** +- Dimensions: `time`, `cml_id`, `sublink_id` +- Data variables from raw data: `rsl`, `tsl` +- Coordinate or data variables from metadata: `frequency`, `polarization`, `length`, `site_0_lon`, `site_0_lat`, `site_1_lon`, `site_1_lat` +- Scalar or attribute: `user_id` + +**Important design note:** +- Some providers may effectively have one `sublink_id` per `cml_id`, others may have multiple. +- To keep the interface stable, always keep `sublink_id` as an explicit dimension, even if it has length 1 for some providers. + +**Functions:** + +```python +import pandas as pd +import xarray as xr + +def build_cml_dataset( + raw_rows: pd.DataFrame, + metadata_rows: pd.DataFrame, +) -> xr.Dataset: + """ + Build the canonical xarray dataset used by all workflows. + + Args: + raw_rows: DataFrame with columns [time, cml_id, sublink_id, user_id, rsl, tsl] + metadata_rows: DataFrame with columns [cml_id, sublink_id, site_0_lon, site_0_lat, + site_1_lon, site_1_lat, frequency, polarization, length] + + Returns: + xr.Dataset with dimensions [time, cml_id, sublink_id] + and variables [rsl, tsl, frequency, polarization, length, ...] + """ + # Implementation notes: + # 1. Validate required columns exist + # 2. Ensure time is timezone-aware UTC and sorted + # 3. Pivot raw rows into a regular xarray structure + # 4. Attach metadata in a way that broadcasts correctly over time + # 5. Store user_id in attrs if it is constant for the dataset + pass + + +def flatten_rain_dataset(rain_ds: xr.Dataset) -> pd.DataFrame: + """ + Convert processed xarray dataset back to tabular rows for DB writing. + + Expected variables in rain_ds: + tl, wet, baseline, waa, a_rain, r + """ + # Implementation notes: + # 1. Convert dataset to DataFrame + # 2. Reset index to columns + # 3. Drop rows where all output variables are NaN + # 4. Re-add user_id from attrs if needed + pass +``` + +**Pitfall Warnings:** +- `xarray` does not magically solve irregular tabular data. The conversion step must define exactly how rows map to dimensions. +- Duplicate `(time, cml_id, sublink_id, user_id)` rows must be resolved before building the dataset. +- Metadata variables may be scalar per link, while raw variables vary over time. Broadcasting must be deliberate. +- String metadata like `polarization` may need special handling in `xarray`. +- If the agent uses inconsistent dimension order across workflows, later merging and flattening will become error-prone. + +--- + +### 5. `workflows/base.py` + +**Purpose:** Abstract base class defining the interface for all processing workflows. + +**Code:** + +```python +from abc import ABC, abstractmethod +import xarray as xr +from datetime import datetime + +class BaseRainWorkflow(ABC): + """ + Abstract base class for rain rate processing workflows. + All workflow variants must inherit from this class and implement process(). + """ + + @abstractmethod + def process( + self, + cml_ds: xr.Dataset, + window_start: datetime, + window_end: datetime + ) -> xr.Dataset: + """ + Process raw CML data to estimate rain rates. + + Args: + cml_ds: Canonical xarray dataset with dimensions [time, cml_id, sublink_id] + and variables including [rsl, tsl] plus metadata variables. + May contain NaN values and missing metadata. + + window_start: Start of processing window (for context) + window_end: End of processing window + + Returns: + xr.Dataset with dimensions [time, cml_id, sublink_id] and variables: + - tl: total loss (TSL - RSL) + - wet: boolean wet/dry classification + - baseline: baseline attenuation level + - waa: wet antenna attenuation estimate + - a_rain: rain-induced attenuation + - r: rain rate estimate (mm/h) + + The dataset should preserve coordinates and carry `user_id` in attrs. + All fields can be NULL/NaN if processing fails for a timestamp. + + Notes: + - Implementations should be robust to missing metadata + - Should handle gaps in time series gracefully + - Should not raise exceptions on bad data (log warnings, return partial results) + - Processing time window may be larger than output window (for temporal context) + - Output should usually be trimmed to the target interval that should be persisted, + even if a larger context window was used internally + """ + pass + + def get_name(self) -> str: + """Return workflow name for logging.""" + return self.__class__.__name__ +``` + +**Pitfall Warnings:** +- **Interface compliance**: All workflow implementations MUST accept the same input signature and return the same output structure. +- **User ID**: Workflows must preserve `user_id`, preferably in `cml_ds.attrs['user_id']` and later in flattened output rows. +- **Error handling**: Workflows should catch internal errors and return partial results rather than crashing the entire processor. +- **Missing metadata**: Not all CMLs may have metadata (e.g., new CMLs not yet in metadata table). Skip or use defaults. +- **Temporal context**: `raw_data` may span longer than `window_start` to `window_end` to provide context for algorithms (e.g., baseline estimation needs history). + +--- + +### 6. `workflows/default.py` + +**Purpose:** Default implementation using `xarray` plus `pycomlink`-style processing. + +**Skeleton:** + +```python +from .base import BaseRainWorkflow +import xarray as xr +from datetime import datetime +import logging + +logger = logging.getLogger(__name__) + +class DefaultRainWorkflow(BaseRainWorkflow): + """ + Default rain rate processing workflow using xarray and pycomlink. + + Processing steps: + 1. Calculate total loss (TL = TSL - RSL) + 2. Wet/dry classification + 3. Baseline estimation + 4. Wet antenna attenuation (WAA) correction + 5. Rain-induced attenuation estimation + 6. Rain rate retrieval using power-law relationship + """ + + def process( + self, + cml_ds: xr.Dataset, + window_start: datetime, + window_end: datetime + ) -> xr.Dataset: + """ + Process CML data to rain rates using pycomlink algorithms. + + See BaseRainWorkflow.process() for parameter/return documentation. + """ + logger.info("Processing CML dataset with default workflow") + + # TODO: Implementation details: + # 1. Work directly on cml_ds with dimensions [time, cml_id, sublink_id] + # 2. Calculate TL = TSL - RSL + # 3. Apply pycomlink wet/dry classification (e.g., Schleiss algorithm) + # 4. Estimate baseline (e.g., rolling window minimum during dry periods) + # 5. Estimate WAA (e.g., constant or time-variable) + # 6. Calculate A_rain = TL - baseline - WAA + # 7. Convert A_rain to rain rate using frequency, polarization, length + # R = a * (A_rain / L)^b where a, b depend on frequency/polarization + # 8. Trim output to the target persistence interval if needed + # 9. Return xr.Dataset with required variables + + # Placeholder return (implementation needed) + out = xr.Dataset(coords=cml_ds.coords) + out['tl'] = cml_ds['tsl'] - cml_ds['rsl'] + out['wet'] = xr.full_like(out['tl'], fill_value=False, dtype=bool) + out['baseline'] = xr.full_like(out['tl'], fill_value=float('nan')) + out['waa'] = xr.full_like(out['tl'], fill_value=float('nan')) + out['a_rain'] = xr.full_like(out['tl'], fill_value=float('nan')) + out['r'] = xr.full_like(out['tl'], fill_value=float('nan')) + out.attrs['user_id'] = cml_ds.attrs.get('user_id') + return out +``` + +**Pitfall Warnings:** +- **pycomlink API**: Ensure compatibility with installed pycomlink version. API may change between versions. +- **Parameter tuning**: Default parameters (e.g., wet/dry thresholds, power-law coefficients) may need tuning per MNO. +- **Frequency/polarization**: Power-law parameters (a, b) depend on frequency and polarization. Missing metadata means can't compute accurate rain rate. +- **Length units**: Ensure consistent units (typically km for length, GHz for frequency, mm/h for rain rate). +- **Time series alignment**: pycomlink may expect regularly-spaced time series. Irregular data needs interpolation or special handling. +- **xarray broadcasting**: Operations may silently broadcast over dimensions. The agent must verify shapes after each step. +- **Boolean arrays with NaN**: `wet` may need nullable handling if classification is unavailable for some timestamps. + +### 7. `workflows/openmrg_basic.py` + +**Purpose:** A basic workflow that follows the style of OpenMRG / pycomlink examples more closely. + +This workflow is not meant to be the final scientific implementation. It is a structured starting point for a less capable agent. + +**Suggested behavior for the first implementation:** +1. Build `tl = tsl - rsl` +2. Optionally resample or regularize time if required by the chosen pycomlink functions +3. Run a simple wet/dry classification on `tl` +4. Estimate a baseline from dry periods or a rolling dry reference +5. Estimate `waa` with a simple method or placeholder +6. Compute `a_rain = tl - baseline - waa` +7. Compute `r` from `a_rain`, `length`, `frequency`, and `polarization` +8. Return all intermediate variables in one `xarray.Dataset` + +**Skeleton:** + +```python +from .base import BaseRainWorkflow +import xarray as xr +import logging + +logger = logging.getLogger(__name__) + + +class OpenMRGBasicWorkflow(BaseRainWorkflow): + """ + Basic xarray-based workflow inspired by OpenMRG / pycomlink examples. + """ + + def process(self, cml_ds: xr.Dataset, window_start, window_end) -> xr.Dataset: + out = xr.Dataset(coords=cml_ds.coords) + + # Step 1: total loss + out['tl'] = cml_ds['tsl'] - cml_ds['rsl'] + + # Step 2: wet/dry classification + # Replace with actual pycomlink-based method once selected. + out['wet'] = xr.full_like(out['tl'], False, dtype=bool) + + # Step 3: baseline estimation + out['baseline'] = out['tl'].rolling(time=12, min_periods=1).min() + + # Step 4: wet antenna attenuation + out['waa'] = xr.zeros_like(out['tl']) + + # Step 5: rain-induced attenuation + out['a_rain'] = out['tl'] - out['baseline'] - out['waa'] + + # Step 6: rain rate retrieval + # Placeholder only. Real implementation must use frequency/polarization-dependent coefficients. + out['r'] = xr.where(out['a_rain'] > 0, out['a_rain'], 0.0) + + out = out.sel(time=slice(window_start, window_end)) + out.attrs['user_id'] = cml_ds.attrs.get('user_id') + return out +``` + +**Important note:** +- The above `r` computation is only a placeholder and is not scientifically correct. +- It is included only as a minimal end-to-end implementation target for the first coding pass. +- The real implementation should replace the placeholder steps with actual `pycomlink` / `poligrain` functions once the exact method is chosen. + +**Pitfall Warnings:** +- A rolling minimum baseline is only a crude starting point. +- Some pycomlink examples assume a specific sampling interval. If DB data are irregular, resampling may be required before processing. +- If the agent resamples, it must document the chosen interval and aggregation rule. + +--- + +### 8. `registry.py` + +**Purpose:** Map processing variant names to workflow classes. + +**Code:** + +```python +from typing import Dict, Type +from .workflows.base import BaseRainWorkflow +from .workflows.default import DefaultRainWorkflow +from .workflows.openmrg_basic import OpenMRGBasicWorkflow +# Import other workflows as they're added +# from .workflows.orange_cameroun_v1 import OrangeCamerounV1Workflow + +class WorkflowRegistry: + """ + Registry mapping processing variant names to workflow classes. + """ + + _registry: Dict[str, Type[BaseRainWorkflow]] = { + 'default': DefaultRainWorkflow, + 'openmrg_basic': OpenMRGBasicWorkflow, + # 'orange_cameroun_v1': OrangeCamerounV1Workflow, + } + + @classmethod + def get_workflow(cls, variant_name: str) -> BaseRainWorkflow: + """ + Get a workflow instance by variant name. + + Args: + variant_name: Name from config (e.g., 'default') + + Returns: + Instance of the workflow class + + Raises: + ValueError: If variant_name not in registry + """ + workflow_class = cls._registry.get(variant_name) + if workflow_class is None: + raise ValueError( + f"Unknown processing variant: {variant_name}. " + f"Available: {list(cls._registry.keys())}" + ) + return workflow_class() + + @classmethod + def register(cls, variant_name: str, workflow_class: Type[BaseRainWorkflow]): + """ + Register a new workflow variant. + + Args: + variant_name: Name to use in config + workflow_class: Class inheriting from BaseRainWorkflow + """ + cls._registry[variant_name] = workflow_class + + @classmethod + def list_variants(cls) -> list: + """Return list of registered variant names.""" + return list(cls._registry.keys()) +``` + +**Pitfall Warnings:** +- **Import errors**: If a workflow file has syntax errors or missing dependencies, the import will fail and crash the service. + Consider dynamic imports with try/except if you want graceful degradation. +- **Name conflicts**: Variant names must be unique. Use descriptive names (e.g., `openmrg_v1`, not just `openmrg`). + +--- + +### 9. `main.py` + +**Purpose:** Main entry point with polling loop. + +**Skeleton:** + +```python +import os +import sys +import time +import logging +from datetime import datetime, timedelta +from config_loader import RainProcessingConfig +from state_manager import StateManager +from data_interface import CMLDataInterface +from dataset_builder import build_cml_dataset, flatten_rain_dataset +from registry import WorkflowRegistry + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +def main(): + """ + Main polling loop for rain rate processing. + + Flow: + 1. Load configuration (with periodic reload) + 2. For each enabled user: + a. Check if time to process (based on poll_interval and last_processed_time) + b. If yes: + - Fetch raw data for time window + - Fetch metadata + - Run workflow + - Write results + - Update state + 3. Sleep and repeat + """ + + # Get database URL from environment + database_url = os.getenv('DATABASE_URL') + if not database_url: + logger.error("DATABASE_URL environment variable not set") + sys.exit(1) + + # Initialize components + config = RainProcessingConfig() + state_manager = StateManager() + data_interface = CMLDataInterface(database_url) + + logger.info("Rain processing service started") + logger.info(f"Available workflows: {WorkflowRegistry.list_variants()}") + + # Main loop + while True: + try: + # Reload config if needed + global_config = config.get_global_config() + reload_interval = global_config.get('config_reload_interval_seconds', 60) + if config.should_reload(reload_interval): + logger.info("Reloading configuration") + config.load() + + # Process enabled users + enabled_users = config.get_enabled_users() + logger.debug(f"Enabled users: {list(enabled_users.keys())}") + + for user_id, user_config in enabled_users.items(): + try: + process_user_if_ready( + user_id, + user_config, + state_manager, + data_interface + ) + except Exception as e: + logger.error(f"Error processing user {user_id}: {e}", exc_info=True) + # Continue to next user (don't let one user's error stop others) + + # Sleep before next iteration + # Use a short sleep to allow responsive config reloading + time.sleep(10) + + except KeyboardInterrupt: + logger.info("Shutting down gracefully") + break + except Exception as e: + logger.error(f"Unexpected error in main loop: {e}", exc_info=True) + time.sleep(30) # Back off on unexpected errors + +def process_user_if_ready( + user_id: str, + user_config: dict, + state_manager: StateManager, + data_interface: CMLDataInterface +): + """ + Check if user is ready for processing and run if so. + + Args: + user_id: User identifier + user_config: User configuration dict from YAML + state_manager: State manager instance + data_interface: Data interface instance + """ + # Get last processed time from state + last_processed = state_manager.get_last_processed_time(user_id) + + # Initialize if first run + if last_processed is None: + logger.info(f"Initializing state for new user: {user_id}") + state_manager.initialize_user(user_id) + return # Don't process on first initialization (starts from now) + + # Check if enough time has passed + now = datetime.utcnow() + poll_interval_seconds = user_config.get('poll_interval_seconds', 900) + time_since_last = (now - last_processed).total_seconds() + + if time_since_last < poll_interval_seconds: + logger.debug( + f"User {user_id}: {time_since_last:.0f}s since last processing " + f"(need {poll_interval_seconds}s)" + ) + return + + # Ready to process + logger.info(f"Processing user: {user_id}") + + # Define time window + window_end = now + window_minutes = user_config.get('data_window_minutes', 90) + window_start = window_end - timedelta(minutes=window_minutes) + + logger.info( + f"User {user_id}: fetching data from {window_start} to {window_end} " + f"({window_minutes} minutes)" + ) + + # Fetch data + raw_rows = data_interface.fetch_raw_cml_data_rows(user_id, window_start, window_end) + if raw_rows.empty: + logger.warning(f"User {user_id}: no raw data in time window, skipping") + # Still update state to avoid repeated processing attempts + state_manager.update_last_processed_time(user_id, window_end) + return + + cml_ids = raw_rows['cml_id'].unique().tolist() + metadata_rows = data_interface.fetch_cml_metadata_rows(user_id, cml_ids) + cml_ds = build_cml_dataset(raw_rows, metadata_rows) + + logger.info( + f"User {user_id}: fetched {len(raw_rows)} raw data points, " + f"{len(metadata_rows)} metadata records" + ) + + # Run workflow + variant_name = user_config.get('processing_variant', 'default') + try: + workflow = WorkflowRegistry.get_workflow(variant_name) + logger.info(f"User {user_id}: running workflow '{variant_name}'") + + rain_ds = workflow.process(cml_ds, window_start, window_end) + rain_data = flatten_rain_dataset(rain_ds) + + if rain_data.empty: + logger.warning(f"User {user_id}: workflow produced no output") + else: + logger.info(f"User {user_id}: workflow produced {len(rain_data)} results") + rows_written = data_interface.write_rain_data(rain_data) + logger.info(f"User {user_id}: wrote {rows_written} rows to database") + + except ValueError as e: + logger.error(f"User {user_id}: invalid workflow variant '{variant_name}': {e}") + return # Don't update state on config error + except Exception as e: + logger.error(f"User {user_id}: workflow processing failed: {e}", exc_info=True) + # Consider whether to update state on processing failure + # For now, don't update to retry on next iteration + return + + # Update state on success + state_manager.update_last_processed_time(user_id, window_end) + logger.info(f"User {user_id}: processing complete, state updated") + +if __name__ == "__main__": + main() +``` + +**Pitfall Warnings:** +- **Infinite loops**: Ensure the main loop has proper error handling to avoid crash-restart cycles. +- **Database connections**: Consider connection pooling or reconnection logic for long-running processes. +- **Memory leaks**: If processing large volumes, ensure DataFrames are freed after processing (Python GC usually handles this). +- **xarray size growth**: Converting sparse tabular data into dense arrays can increase memory usage. The agent should test realistic window sizes. +- **Clock drift**: Use UTC consistently. Container clock should sync with host. +- **Graceful shutdown**: Handle SIGTERM/SIGINT to close database connections cleanly. +- **Logging levels**: Use appropriate levels (DEBUG, INFO, WARNING, ERROR) for operational visibility. + +--- + +## Docker Configuration + +### Dockerfile + +Update `processor/Dockerfile`: + +```dockerfile +FROM python:3.10-slim + +WORKDIR /app + +# Install dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy source code +COPY *.py ./ +COPY workflows/ ./workflows/ + +# Create directories for config and state +RUN mkdir -p /app/config /app/data/state + +# Set environment variables +ENV PYTHONUNBUFFERED=1 + +CMD ["python", "main.py"] +``` + +### docker-compose.yml + +Update the processor service: + +```yaml +processor: + build: ./processor + restart: unless-stopped # Changed from default (restarts on failure) + depends_on: + database: + condition: service_healthy + environment: + - DATABASE_URL=postgresql://myuser:mypassword@database:5432/mydatabase + volumes: + - ./processor/config:/app/config:ro # Config file (read-only) + - processor_state:/app/data/state # State persistence + # Remove ports if not needed (no HTTP API currently) +``` + +Add volume at bottom of file: + +```yaml +volumes: + # ... existing volumes ... + processor_state: +``` + +**Pitfall Warnings:** +- **Volume permissions**: Ensure the container user has write access to `/app/data/state`. +- **Read-only config**: Mount config as `:ro` to prevent accidental modification from container. +- **Restart policy**: Use `unless-stopped` to auto-restart on crashes, but not if manually stopped. +- **Health checks**: Consider adding health check (e.g., timestamp of last successful processing). + +--- + +## Requirements + +Update `processor/requirements.txt`: + +``` +psycopg2-binary>=2.9.0 +pandas>=1.5.0 +xarray>=2023.0.0 +pyyaml>=6.0 +pycomlink>=0.4.0 +poligrain +``` + +**Pitfall Warnings:** +- **Version pinning**: Consider pinning exact versions for reproducibility (e.g., `pandas==1.5.3`). +- **pycomlink dependencies**: May pull in scipy, numpy (large images). Consider Alpine base image if size matters. +- **Compatibility**: Test that pycomlink works with Python 3.10+ (check their docs). +- **xarray backend assumptions**: The agent should not assume NetCDF I/O is needed in the processor. `xarray` is used as an in-memory data model here. + +--- + +## Implementation Checklist + +### Phase 1: Database and Infrastructure +- [ ] Create migration file `database/migrations/010_add_rain_data_table.sql` +- [ ] Apply migration to database +- [ ] Create `processor/config/rain_processing.yml` with initial config +- [ ] Update `processor/requirements.txt` +- [ ] Update `processor/Dockerfile` +- [ ] Update `docker-compose.yml` (processor service + volume) + +### Phase 2: Core Modules +- [ ] Implement `config_loader.py` with validation +- [ ] Implement `state_manager.py` with file locking +- [ ] Implement `data_interface.py` (read/write methods) +- [ ] Implement `dataset_builder.py` with one canonical xarray layout +- [ ] Create `workflows/base.py` abstract class +- [ ] Implement `registry.py` + +### Phase 3: Processing Workflow +- [ ] Implement `workflows/default.py` (can start with placeholder that just calculates TL in xarray) +- [ ] Implement `workflows/openmrg_basic.py` as the first end-to-end xarray workflow +- [ ] Enhance `default.py` / `openmrg_basic.py` with actual pycomlink and poligrain algorithms (iterative) + +### Phase 4: Main Service +- [ ] Implement `main.py` polling loop +- [ ] Add comprehensive logging +- [ ] Test end-to-end with one user + +### Phase 5: Testing and Deployment +- [ ] Test config hot-reload +- [ ] Test state persistence across restarts +- [ ] Test with multiple users +- [ ] Test error handling (missing data, bad workflow name, etc.) +- [ ] Deploy and monitor + +--- + +## Testing Strategy + +### Unit Tests +- `config_loader.py`: Test YAML parsing, validation, reload logic +- `state_manager.py`: Test file locking, atomic writes, initialization +- `data_interface.py`: Test with mock database or test database +- Workflow classes: Test with synthetic input data + +### Integration Tests +- End-to-end: Enable user in config, verify data written to `cml_rain_data` +- Test with empty database (no raw data) +- Test with missing metadata +- Test state file corruption recovery + +### Operational Tests +- Config hot-reload (edit YAML while running) +- Container restart (verify state persists) +- Database connection loss (verify reconnection) +- Invalid workflow variant (verify error handling) + +--- + +## Operational Notes + +### Monitoring + +**Key metrics to track:** +- Processing lag: `now() - last_processed_time` per user +- Processing duration per user +- Rows processed per run +- Error rate (failed processing attempts) + +**Logging strategy:** +- INFO: Normal operations (processing started/completed, rows written) +- WARNING: Recoverable issues (no data, missing metadata) +- ERROR: Failures (workflow errors, database errors) + +### Troubleshooting + +**Processor not running:** +- Check logs: `docker logs gmdi_prototype-processor-1` +- Verify DATABASE_URL is set correctly +- Check config file exists and is valid YAML + +**User not processing:** +- Check `enabled: true` in config +- Check state file for `last_processed_time` +- Verify `poll_interval_seconds` has passed +- Check for errors in logs for that specific user + +**No output data:** +- Verify raw data exists in `cml_data` for the user +- Check time window (might be too narrow or offset) +- Check workflow logs for errors +- Verify metadata exists + +**Reset processing for a user:** +```bash +# Edit state file (in volume or container) +# Remove user's entry or set last_processed_time to desired timestamp +# Or delete entire state file to reset all users to "now" +``` + +### Performance Tuning + +**If processing is slow:** +- Reduce `data_window_minutes` (less data to fetch) +- Increase `poll_interval_seconds` (process less frequently) +- Optimize workflow algorithms (profiling needed) +- Add database indexes on `cml_data(user_id, time)` + +**If processing lags behind real-time:** +- Decrease `poll_interval_seconds` +- Increase resources (CPU, memory) for processor container +- Consider parallel processing (one processor per user) + +--- + +## Future Enhancements + +**Potential additions (not in current scope):** + +1. **Backfill support**: Add flag in config to process historical data +2. **Parallel processing**: Run multiple workflow instances concurrently +3. **Real-time streaming**: Switch from polling to event-driven (LISTEN/NOTIFY) +4. **Quality control**: Add QC flags to output data +5. **Aggregated views**: Create materialized view for hourly/daily rain rates +6. **Web API**: Add HTTP endpoint to trigger/configure processing +7. **Metrics endpoint**: Expose Prometheus metrics for monitoring +8. **Workflow versioning**: Track which workflow version processed each record + +--- + +## Notebook-Based Validation Workflow + +Before enabling continuous processing for a user, it should be possible to test whether the selected workflow produces sensible results for a chosen day or time window. + +The recommended approach is a **Jupyter notebook working example** that imports and runs the **exact same workflow code** used by the continuous processor. + +This is important: the notebook must **not** reimplement the processing logic separately. It should call the same modules as the service, otherwise the notebook and the service may diverge over time. + +### Goal + +Provide a reproducible, inspectable workflow for: +- selecting one user +- selecting one day or arbitrary time window +- loading raw data and metadata from the database +- building the canonical `xarray.Dataset` +- running one configured workflow variant +- plotting intermediate and final outputs +- visually checking whether the processing looks plausible before enabling continuous mode + +### Recommended Notebook File + +Create a notebook such as: + +`notebooks/rain_processing_validation.ipynb` + +This notebook should be treated as a **working example** and a **manual validation tool**. + +### Required Reuse of Production Code + +The notebook should import the same modules used by the processor service: + +- `processor.config_loader` +- `processor.data_interface` +- `processor.dataset_builder` +- `processor.registry` +- workflow modules from `processor.workflows` + +The notebook should **not** contain copied versions of: +- SQL queries +- dataset conversion logic +- workflow logic + +Instead, it should call the production functions directly. + +### Suggested Notebook Flow + +#### 1. Setup and imports + +The notebook should: +- import `os`, `datetime`, `matplotlib`, `xarray` +- import the production processor modules +- read `DATABASE_URL` + +#### 2. User-selectable parameters + +The notebook should expose a small parameter section near the top, for example: + +```python +USER_ID = "demo_openmrg" +WORKFLOW_VARIANT = "openmrg_basic" +START_TIME = "2026-01-20T00:00:00Z" +END_TIME = "2026-01-21T00:00:00Z" +PLOT_CML_ID = None +PLOT_SUBLINK_ID = None +``` + +This keeps the notebook easy to reuse. + +#### 3. Load raw rows and metadata + +Use `CMLDataInterface` methods to fetch: +- raw CML rows from `cml_data` +- metadata rows from `cml_metadata` + +#### 4. Build canonical xarray dataset + +Use `build_cml_dataset()` from `processor.dataset_builder`. + +This ensures the notebook uses the same in-memory structure as the service. + +#### 5. Load workflow from registry + +Use `WorkflowRegistry.get_workflow(WORKFLOW_VARIANT)`. + +This is important because it guarantees the notebook runs the same workflow variant that continuous processing would run. + +#### 6. Run processing + +Call: + +```python +rain_ds = workflow.process(cml_ds, window_start, window_end) +``` + +#### 7. Plot results + +At minimum, the notebook should plot for one selected link: +- `tl` +- `wet` +- `baseline` +- `waa` +- `a_rain` +- `r` + +Recommended plotting style: +- one figure with multiple aligned subplots sharing the same time axis +- optional overlay of `tl`, `baseline`, and `a_rain` +- optional highlighting of wet periods + +#### 8. Optional export + +The notebook may optionally: +- convert `rain_ds` to a DataFrame +- save results to CSV for inspection +- save plots to PNG + +It should **not** write to `cml_rain_data` by default. + +### Minimal Plotting Expectations + +The notebook should include at least: + +1. **Time series plot of TL and baseline** +2. **Wet/dry indicator plot** +3. **Time series plot of A_rain** +4. **Time series plot of R** + +If possible, also include: +- a quick summary table of missing values +- number of wet timestamps +- min/max/mean of `r` + +### Why Notebook Validation Is Useful + +This helps catch issues before continuous processing is enabled, for example: +- wrong sign convention in `tl = tsl - rsl` +- unrealistic baseline behavior +- `waa` correction too large or too small +- rain rates always zero or unrealistically high +- metadata problems such as missing length or frequency +- irregular time sampling causing unstable results + +### Pitfall Warnings for the Implementing Agent + +- The notebook must import the production workflow code, not a notebook-only copy. +- If the notebook duplicates logic, later changes in the service may not be reflected in validation. +- The notebook should use the same canonical `xarray` layout as the service. +- Plotting all links at once may be too heavy; default to one selected `cml_id` / `sublink_id`. +- Some workflows may require a context window larger than the plotted interval. The notebook should allow this. +- If the workflow trims output to the target interval, the notebook should make that explicit. +- The notebook should fail clearly if the selected workflow variant is not registered. +- The notebook should not silently write to production tables. + +### Suggested Supporting Helper Functions + +To keep the notebook simple, it is useful to add small reusable helper functions in normal Python modules, for example in: + +`processor/plotting.py` + +Possible helper functions: + +```python +def select_single_link(ds, cml_id=None, sublink_id=None): + """Return a dataset slice for one link for plotting.""" + +def plot_rain_workflow_overview(ds): + """Create a standard multi-panel plot for tl, wet, baseline, waa, a_rain, r.""" + +def summarize_rain_dataset(ds): + """Return simple summary statistics for validation.""" +``` + +These helpers should also be importable by future scripts or tests. + +### Recommended Scope for First Implementation + +For the first implementation, keep the notebook simple: +- one user +- one selected day +- one workflow variant +- one selected link plotted +- no DB writes + +That is enough to validate whether the workflow behaves plausibly. + +### Relationship to Continuous Processing + +The intended workflow is: + +1. implement or update a workflow in `processor/workflows/` +2. test it in the notebook for a selected day +3. inspect plots and summary statistics +4. only then enable the workflow in `processor/config/rain_processing.yml` + +This should be the recommended operational path. + +--- + +## Planned Website Integration + +The website presentation of processed rain data has now been decided at a high level. + +### New Subsite + +Add a new subsite named: + +- `Rain rates` + +This subsite should follow the **same general layout as the existing real-time subsite**. + +This is recommended because: +- users already know that layout +- the existing map interaction model can be reused +- implementation effort is lower than creating a completely new page design + +### Map Behavior + +The map should reuse the same CML geometry and general interaction pattern as the real-time page. + +The main difference is the coloring mode. + +The map should support at least these coloring options: + +1. **Last rain rate value** +2. **Rainfall sum of last hour** +3. **Rainfall sum of last day** + +### Definition of Rainfall Sum + +For this project, the rainfall sum over a time window should be defined as: + +- the **average of the rain-rate values** over that window + +This is the chosen definition because it also works when: +- the time step `delta_t` is not constant +- different processing variants produce different sampling intervals + +So for a selected interval, such as the last hour or last day: + +$$ + ext{rainfall sum proxy} = \frac{1}{N} \sum_{i=1}^{N} R_i +$$ + +where $R_i$ are the rain-rate values available in that interval. + +**Important note for the implementing agent:** +- This is not a physical time integral of rainfall depth. +- It is a project-specific aggregation rule chosen for robustness and simplicity. +- The implementation must not silently replace this with a weighted sum over time unless the plan is updated later. + +### Grafana / Detail Panel Behavior + +When a user clicks a CML on the map: + +- the detail panel should show the **processed rain-rate time series** for that CML +- the Grafana panel should update based on the selected CML +- the interaction should mirror the current real-time page behavior as closely as possible + +The intended behavior is: + +1. user opens `Rain rates` +2. user selects a map coloring mode +3. user clicks a CML +4. Grafana panel updates to show processed rain-rate data for that CML + +### Backend / Query Implications + +To support the website efficiently, the backend will likely need query support for: + +- latest rain-rate value per CML +- average rain-rate over the last hour per CML +- average rain-rate over the last day per CML + +These may later be implemented as: +- SQL views +- materialized views +- Timescale continuous aggregates + +The exact implementation is not fixed yet, but the frontend target behavior is fixed. + +### Recommended Data Products for the Website + +The following derived products will likely be needed: + +1. **Latest rain-rate per CML** +2. **1-hour average rain-rate per CML** +3. **24-hour average rain-rate per CML** + +These products should be user-scoped in the same way as the rest of the application. + +### Pitfall Warnings for the Implementing Agent + +- Do not assume “rainfall sum” means a time integral. In this plan it explicitly means the average of available rain-rate values in the interval. +- If there are no values in the interval, the result should be `NULL` / missing, not zero. +- The map geometry should be reused from the existing real-time page rather than rebuilt separately. +- The Grafana panel should show processed rain-rate data, not raw RSL/TSL. +- The selected CML identifier used by the map must match the identifier used in the processed rain-rate queries. +- If the real-time page uses a specific route, template, or API structure, the new `Rain rates` page should follow the same pattern where possible. + +### Scope Note + +This section defines the intended UI behavior, but it does **not** yet fully specify: +- exact webserver routes +- exact SQL/API endpoints +- exact Grafana dashboard/panel configuration + +Those can be designed later, but the target behavior is now fixed enough for planning. + +--- + +## Summary + +This implementation provides a robust, maintainable foundation for continuous rain rate processing: + +✅ **Simple configuration** via YAML (version-controlled, easy to edit) +✅ **Persistent state** via JSON (survives restarts) +✅ **Pluggable workflows** (easy to add MNO-specific variants) +✅ **Fixed interface** between data layer and processing logic +✅ **No backfill** by default (starts from "now" when enabled) +✅ **Per-user control** of frequency and data window +✅ **Robust error handling** (one user's failure doesn't affect others) + +The modular design allows incremental development: start with a simple workflow that just calculates TL, then gradually add wet/dry classification, baseline estimation, and full rain rate retrieval as the algorithms are refined. diff --git a/processor/backfill.py b/processor/backfill.py new file mode 100644 index 0000000..61e054b --- /dev/null +++ b/processor/backfill.py @@ -0,0 +1,108 @@ +""" +One-shot backfill script: process historical CML data in daily batches. + +Usage: + python backfill.py --user demo_openmrg --days 3 --variant openmrg_basic + +Splits the requested number of past days into 1-day batches (oldest first), +runs the workflow for each batch, and writes results to the database. +""" + +import os +import sys +import logging +import argparse +from datetime import datetime, timedelta, timezone + +from data_interface import CMLDataInterface +from dataset_builder import build_cml_dataset, flatten_rain_dataset +from registry import WorkflowRegistry + +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + + +def run_batch(data_interface, user_id, variant_name, window_start, window_end): + logger.info( + f"Batch {window_start.strftime('%Y-%m-%d %H:%M')} → " + f"{window_end.strftime('%Y-%m-%d %H:%M')}" + ) + + raw_data = data_interface.fetch_raw_cml_data_rows(user_id, window_start, window_end) + if raw_data.empty: + logger.warning("No raw data found for this window, skipping.") + return 0 + + metadata = data_interface.fetch_cml_metadata_rows(user_id) + logger.info( + f"Fetched {len(raw_data)} raw data points, {len(metadata)} metadata records" + ) + + cml_ds = build_cml_dataset(raw_data, metadata) + workflow = WorkflowRegistry.get_workflow(variant_name) + rain_ds = workflow.process(cml_ds, window_start, window_end) + rain_data = flatten_rain_dataset(rain_ds) + + if rain_data.empty: + logger.warning("Workflow produced no output for this window.") + return 0 + + logger.info(f"Workflow produced {len(rain_data)} results") + rows_written = data_interface.write_rain_data(rain_data) + logger.info(f"Wrote {rows_written} rows to database") + return rows_written + + +def main(): + parser = argparse.ArgumentParser( + description="Backfill rain rate data in daily batches" + ) + parser.add_argument("--user", default="demo_openmrg", help="User ID to process") + parser.add_argument( + "--days", type=int, default=3, help="Number of past days to backfill" + ) + parser.add_argument( + "--batch-hours", type=int, default=6, help="Hours per batch (default 6)" + ) + parser.add_argument("--variant", default="openmrg_basic", help="Workflow variant") + args = parser.parse_args() + + database_url = os.getenv("DATABASE_URL") + if not database_url: + logger.error("DATABASE_URL environment variable not set") + sys.exit(1) + + data_interface = CMLDataInterface(database_url) + now = datetime.now(timezone.utc) + + # Build list of (start, end) windows in batch_hours chunks, oldest first + batch_td = timedelta(hours=args.batch_hours) + backfill_start = now - timedelta(days=args.days) + batches = [] + t = backfill_start + while t < now: + batch_end = min(t + batch_td, now) + batches.append((t, batch_end)) + t = batch_end + + logger.info(f"Starting backfill: {args.days} batches for user '{args.user}'") + total_rows = 0 + + for idx, (batch_start, batch_end) in enumerate(batches, 1): + logger.info(f"=== Batch {idx}/{len(batches)} ===") + try: + rows = run_batch( + data_interface, args.user, args.variant, batch_start, batch_end + ) + total_rows += rows + except Exception as e: + logger.error(f"Batch {idx} failed: {e}", exc_info=True) + logger.info("Continuing with next batch...") + + logger.info(f"Backfill complete. Total rows written: {total_rows}") + + +if __name__ == "__main__": + main() diff --git a/processor/config/rain_processing.yml b/processor/config/rain_processing.yml new file mode 100644 index 0000000..0a2d85d --- /dev/null +++ b/processor/config/rain_processing.yml @@ -0,0 +1,27 @@ +# Global settings for the rain processing service +global: + # How often to reload this YAML file (seconds) + config_reload_interval_seconds: 60 + + # Default poll interval if not specified per user (seconds) + default_poll_interval_seconds: 300 # 5 minutes + + # Default data window if not specified per user (minutes) + default_data_window_minutes: 120 + +# Per-user configuration +users: + demo_openmrg: + enabled: true # Enable/disable processing for this user + processing_variant: openmrg_basic # Which workflow to use (maps to processor/workflows/.py) + poll_interval_seconds: 300 # Process every 5 minutes + data_window_minutes: 120 # Fetch 120 minutes of data for context + + demo_orange_cameroun: + enabled: false # Disabled by default + processing_variant: openmrg_basic + poll_interval_seconds: 60 # Process every 1 minute + data_window_minutes: 120 # Fetch 120 minutes of data + +# To add a new user, simply add an entry under 'users' +# The processor will automatically pick it up on next config reload diff --git a/processor/config_loader.py b/processor/config_loader.py new file mode 100644 index 0000000..e34c0b4 --- /dev/null +++ b/processor/config_loader.py @@ -0,0 +1,166 @@ +""" +Configuration loader for rain processing service. +Loads and validates YAML configuration with support for hot-reloading. +""" + +import yaml +from pathlib import Path +from datetime import datetime, timezone +from typing import Dict, Any, Optional +import logging + +logger = logging.getLogger(__name__) + + +class RainProcessingConfig: + """ + Loads and manages rain processing configuration from YAML. + Supports hot-reloading without service restart. + """ + + def __init__(self, config_path: str = "/app/config/rain_processing.yml"): + """ + Args: + config_path: Path to YAML configuration file + """ + self.config_path = Path(config_path) + self._config: Dict[str, Any] = {} + self._last_loaded: Optional[datetime] = None + self.load() + + def load(self) -> None: + """ + Load configuration from YAML file. + Validates structure and raises ValueError if invalid. + + Raises: + FileNotFoundError: If config file doesn't exist + yaml.YAMLError: If YAML is malformed + ValueError: If required fields are missing or invalid + """ + try: + with open(self.config_path, "r") as f: + config = yaml.safe_load(f) + + # Validate structure + self._validate_config(config) + + self._config = config + self._last_loaded = datetime.now(timezone.utc) + logger.info(f"Configuration loaded from {self.config_path}") + + except FileNotFoundError: + logger.error(f"Configuration file not found: {self.config_path}") + raise + except yaml.YAMLError as e: + logger.error(f"YAML parsing error: {e}") + raise + except ValueError as e: + logger.error(f"Configuration validation error: {e}") + raise + + def _validate_config(self, config: Dict[str, Any]) -> None: + """ + Validate configuration structure. + + Args: + config: Configuration dictionary to validate + + Raises: + ValueError: If configuration is invalid + """ + if not isinstance(config, dict): + raise ValueError("Configuration must be a dictionary") + + # Validate global section + if "global" not in config: + raise ValueError("Missing 'global' section in configuration") + + global_cfg = config["global"] + if not isinstance(global_cfg, dict): + raise ValueError("'global' section must be a dictionary") + + # Validate users section + if "users" not in config: + raise ValueError("Missing 'users' section in configuration") + + users_cfg = config["users"] + if not isinstance(users_cfg, dict): + raise ValueError("'users' section must be a dictionary") + + # Validate each user entry + for user_id, user_config in users_cfg.items(): + if not isinstance(user_config, dict): + raise ValueError(f"User '{user_id}' configuration must be a dictionary") + + # Check required fields + if "enabled" not in user_config: + raise ValueError(f"User '{user_id}' missing 'enabled' field") + + if not isinstance(user_config["enabled"], bool): + raise ValueError(f"User '{user_id}' 'enabled' field must be boolean") + + if "processing_variant" not in user_config: + raise ValueError(f"User '{user_id}' missing 'processing_variant' field") + + # Validate optional fields if present + if "poll_interval_seconds" in user_config: + if ( + not isinstance(user_config["poll_interval_seconds"], int) + or user_config["poll_interval_seconds"] <= 0 + ): + raise ValueError( + f"User '{user_id}' 'poll_interval_seconds' must be a positive integer" + ) + + if "data_window_minutes" in user_config: + if ( + not isinstance(user_config["data_window_minutes"], int) + or user_config["data_window_minutes"] <= 0 + ): + raise ValueError( + f"User '{user_id}' 'data_window_minutes' must be a positive integer" + ) + + def should_reload(self, reload_interval_seconds: int) -> bool: + """ + Check if enough time has passed to reload config. + + Args: + reload_interval_seconds: Time between reloads + + Returns: + True if config should be reloaded + """ + if self._last_loaded is None: + return True + + now = datetime.now(timezone.utc) + elapsed = (now - self._last_loaded).total_seconds() + return elapsed >= reload_interval_seconds + + def get_global_config(self) -> Dict[str, Any]: + """Get global configuration section.""" + return self._config.get("global", {}) + + def get_user_config(self, user_id: str) -> Optional[Dict[str, Any]]: + """ + Get configuration for a specific user. + + Args: + user_id: User identifier + + Returns: + User config dict, or None if user not in config + """ + return self._config.get("users", {}).get(user_id) + + def get_enabled_users(self) -> Dict[str, Dict[str, Any]]: + """ + Get all users where enabled=true. + + Returns: + Dict mapping user_id to user config + """ + users = self._config.get("users", {}) + return {uid: cfg for uid, cfg in users.items() if cfg.get("enabled", False)} diff --git a/processor/data_interface.py b/processor/data_interface.py new file mode 100644 index 0000000..541110f --- /dev/null +++ b/processor/data_interface.py @@ -0,0 +1,230 @@ +""" +Database interface for CML data operations. +Provides fixed API for workflows to read/write data. +""" + +import pandas as pd +import psycopg2 +from psycopg2.extras import execute_values +from datetime import datetime, timezone +from typing import List, Optional +from contextlib import contextmanager +import logging + +logger = logging.getLogger(__name__) + + +class CMLDataInterface: + """ + Database interface for CML data operations. + Provides fixed API for workflows to read/write data. + """ + + def __init__(self, database_url: str): + """ + Args: + database_url: PostgreSQL connection string + """ + self.database_url = database_url + self._connection: Optional[psycopg2.extensions.connection] = None + + @contextmanager + def _get_connection(self): + """Context manager for database connections.""" + conn = None + try: + conn = psycopg2.connect(self.database_url) + yield conn + finally: + if conn is not None: + conn.close() + + def fetch_raw_cml_data_rows( + self, user_id: str, start_time: datetime, end_time: datetime + ) -> pd.DataFrame: + """ + Fetch raw CML data (RSL, TSL) for a time window as tabular rows. + + Args: + user_id: User identifier + start_time: Window start (inclusive) + end_time: Window end (inclusive) + + Returns: + DataFrame with columns: time, cml_id, sublink_id, user_id, rsl, tsl + Sorted by time ascending + Empty DataFrame if no data found + """ + # Ensure timestamps are timezone-aware + if start_time.tzinfo is None: + start_time = start_time.replace(tzinfo=timezone.utc) + if end_time.tzinfo is None: + end_time = end_time.replace(tzinfo=timezone.utc) + + query = """ + SELECT time, cml_id, sublink_id, user_id, rsl, tsl + FROM cml_data + WHERE user_id = %s + AND time >= %s + AND time <= %s + ORDER BY time ASC + """ + + try: + with self._get_connection() as conn: + df = pd.read_sql_query( + query, conn, params=(user_id, start_time, end_time), index_col=None + ) + + if df.empty: + logger.debug(f"No raw data found for user {user_id} in time window") + return df + + # Ensure time column is timezone-aware + if df["time"].dt.tz is None: + df["time"] = df["time"].dt.tz_localize("UTC") + + logger.debug(f"Fetched {len(df)} raw data rows for user {user_id}") + return df + + except psycopg2.Error as e: + logger.error(f"Database error fetching raw data: {e}") + raise + + def fetch_cml_metadata_rows( + self, user_id: str, cml_ids: Optional[List[str]] = None + ) -> pd.DataFrame: + """ + Fetch CML metadata (coordinates, frequency, length, polarization) as tabular rows. + + Args: + user_id: User identifier + cml_ids: List of CML IDs to fetch (None = fetch all for user) + + Returns: + DataFrame with columns: cml_id, sublink_id, site_0_lon, site_0_lat, + site_1_lon, site_1_lat, frequency, polarization, length + Empty DataFrame if no metadata found + """ + query = """ + SELECT cml_id, sublink_id, site_0_lon, site_0_lat, + site_1_lon, site_1_lat, frequency, polarization, length + FROM cml_metadata + WHERE user_id = %s + """ + + params = [user_id] + + if cml_ids is not None: + placeholders = ",".join(["%s"] * len(cml_ids)) + query += f" AND cml_id IN ({placeholders})" + params.extend(cml_ids) + + try: + with self._get_connection() as conn: + df = pd.read_sql_query(query, conn, params=params, index_col=None) + + if df.empty: + logger.debug(f"No metadata found for user {user_id}") + return df + + logger.debug(f"Fetched {len(df)} metadata rows for user {user_id}") + return df + + except psycopg2.Error as e: + logger.error(f"Database error fetching metadata: {e}") + raise + + def write_rain_data(self, rain_df: pd.DataFrame) -> int: + """ + Write processed rain data to cml_rain_data table. + + Args: + rain_df: DataFrame with columns: time, cml_id, sublink_id, user_id, + tl, wet, baseline, waa, a_rain, r + + Returns: + Number of rows written + + Raises: + ValueError: If required columns are missing + psycopg2.Error: On database errors + """ + # Validate required columns + required_columns = [ + "time", + "cml_id", + "sublink_id", + "user_id", + "tl", + "wet", + "baseline", + "waa", + "a_rain", + "r", + ] + + missing_columns = [ + col for col in required_columns if col not in rain_df.columns + ] + if missing_columns: + raise ValueError(f"Missing required columns: {missing_columns}") + + if rain_df.empty: + logger.debug("No rain data to write") + return 0 + + # Prepare data for insertion + # Convert wet column to boolean-compatible format + df_copy = rain_df.copy() + + # Ensure time is timezone-aware and convert to UTC + if df_copy["time"].dt.tz is None: + df_copy["time"] = df_copy["time"].dt.tz_localize("UTC") + else: + df_copy["time"] = df_copy["time"].dt.tz_convert("UTC") + + # Convert to list of tuples for execute_values + records = [] + for _, row in df_copy.iterrows(): + record = ( + row["time"], + row["cml_id"], + row["sublink_id"], + row["user_id"], + float(row["tl"]) if pd.notna(row["tl"]) else None, + bool(row["wet"]) if pd.notna(row["wet"]) else None, + float(row["baseline"]) if pd.notna(row["baseline"]) else None, + float(row["waa"]) if pd.notna(row["waa"]) else None, + float(row["a_rain"]) if pd.notna(row["a_rain"]) else None, + float(row["r"]) if pd.notna(row["r"]) else None, + ) + records.append(record) + + query = """ + INSERT INTO cml_rain_data + (time, cml_id, sublink_id, user_id, tl, wet, baseline, waa, a_rain, r) + VALUES %s + ON CONFLICT (time, cml_id, sublink_id, user_id) DO NOTHING + """ + + try: + with self._get_connection() as conn: + with conn.cursor() as cur: + execute_values(cur, query, records) + conn.commit() + + rows_written = cur.rowcount + logger.info(f"Wrote {rows_written} rain data rows to database") + return rows_written + + except psycopg2.Error as e: + logger.error(f"Database error writing rain data: {e}") + raise + + def close(self): + """Close database connection.""" + if self._connection is not None: + self._connection.close() + self._connection = None + logger.debug("Database connection closed") diff --git a/processor/dataset_builder.py b/processor/dataset_builder.py new file mode 100644 index 0000000..cd06ae6 --- /dev/null +++ b/processor/dataset_builder.py @@ -0,0 +1,246 @@ +""" +Dataset builder for CML data. +Converts SQL query results into canonical xarray.Dataset structure for workflows. +""" + +import pandas as pd +import xarray as xr +import numpy as np +from datetime import datetime, timezone +from typing import Optional +import logging + +logger = logging.getLogger(__name__) + + +def build_cml_dataset( + raw_rows: pd.DataFrame, + metadata_rows: pd.DataFrame, + round_to_seconds: Optional[int] = None, +) -> xr.Dataset: + """ + Build the canonical xarray dataset used by all workflows. + + Uses xr.Dataset.from_dataframe() for memory efficiency with large datasets. + This avoids creating a dense 3D grid and instead keeps data in sparse format. + + Args: + raw_rows: DataFrame with columns [time, cml_id, sublink_id, user_id, rsl, tsl] + metadata_rows: DataFrame with columns [cml_id, sublink_id, site_0_lon, site_0_lat, + site_1_lon, site_1_lat, frequency, polarization, length] + round_to_seconds: Round timestamps to this interval to reduce memory usage from + irregular sampling. Useful when CMLs have slightly different + observation times. Set to None to keep original timestamps. + Examples: 60 for 1-min data, 900 for 15-min data, None for irregular. + + Returns: + xr.Dataset with dimensions including time and link identifiers + and variables [rsl, tsl, frequency, polarization, length, ...] + """ + # Validate required columns in raw data + required_raw_cols = ["time", "cml_id", "sublink_id", "user_id", "rsl", "tsl"] + missing_raw = [col for col in required_raw_cols if col not in raw_rows.columns] + if missing_raw: + raise ValueError(f"Missing required columns in raw data: {missing_raw}") + + if raw_rows.empty: + logger.warning("Empty raw data provided, returning empty dataset") + return xr.Dataset() + + # Ensure time is timezone-aware and sorted + raw_rows = raw_rows.copy() + if raw_rows["time"].dt.tz is None: + raw_rows["time"] = raw_rows["time"].dt.tz_localize("UTC") + else: + raw_rows["time"] = raw_rows["time"].dt.tz_convert("UTC") + + # Round timestamps to reduce memory footprint from irregular sampling + # This is especially important when CMLs have slightly different observation times + if round_to_seconds is not None: + original_count = len(raw_rows) + raw_rows["time"] = raw_rows["time"].dt.round(f"{round_to_seconds}s") + # Remove duplicates that may result from rounding + raw_rows = raw_rows.drop_duplicates(subset=["time", "cml_id", "sublink_id"]) + rounded_count = len(raw_rows) + if original_count != rounded_count: + logger.info( + f"Rounded {original_count:,} timestamps to {round_to_seconds}s intervals, " + f"reduced to {rounded_count:,} rows (removed {original_count - rounded_count:,} duplicates)" + ) + + raw_rows = raw_rows.sort_values("time") + + # Get unique user_id (should be constant for this dataset) + user_ids = raw_rows["user_id"].unique() + if len(user_ids) > 1: + logger.warning(f"Multiple user_ids in dataset: {user_ids}, using first") + user_id = user_ids[0] + + # Create MultiIndex from relevant columns + raw_rows_indexed = raw_rows.set_index(["time", "cml_id", "sublink_id"]) + + # Convert to xarray using from_dataframe - much more memory efficient! + # This preserves the sparse structure instead of creating a dense 3D grid + ds = xr.Dataset.from_dataframe(raw_rows_indexed) + + # Add metadata as coordinates following poligrain/OpenMRG convention + # See: https://poligrain.readthedocs.io/en/latest/notebooks/Explore_example_data.html + if not metadata_rows.empty: + metadata_indexed = metadata_rows.set_index(["cml_id", "sublink_id"]) + + # Get unique coordinate values + cml_ids_meta = sorted( + metadata_indexed.index.get_level_values("cml_id").unique().tolist() + ) + sublink_ids_meta = sorted( + metadata_indexed.index.get_level_values("sublink_id").unique().tolist() + ) + + # Add site coordinates (indexed by cml_id only) + for col in ["site_0_lon", "site_0_lat", "site_1_lon", "site_1_lat"]: + if col in metadata_indexed.columns: + # Create mapping from cml_id to value (average across sublinks if needed) + site_values = {} + for cml_id in cml_ids_meta: + cml_data = metadata_indexed.xs( + cml_id, level="cml_id", drop_level=False + ) + if len(cml_data) > 0: + site_values[cml_id] = cml_data[ + col + ].mean() # Average if multiple sublinks + else: + site_values[cml_id] = np.nan + + ds.coords[col] = ( + "cml_id", + [site_values.get(cid, np.nan) for cid in cml_ids_meta], + ) + + # Add length coordinate (indexed by cml_id only) + if "length" in metadata_indexed.columns: + length_values = {} + for cml_id in cml_ids_meta: + cml_data = metadata_indexed.xs(cml_id, level="cml_id", drop_level=False) + if len(cml_data) > 0: + length_values[cml_id] = cml_data["length"].mean() + else: + length_values[cml_id] = np.nan + + ds.coords["length"] = ( + "cml_id", + [length_values.get(cid, np.nan) for cid in cml_ids_meta], + ) + + # Add frequency and polarization (indexed by both sublink_id and cml_id) + for col in ["frequency", "polarization"]: + if col in metadata_indexed.columns: + # Create 2D array with dimensions (sublink_id, cml_id) + meta_array = np.full( + (len(sublink_ids_meta), len(cml_ids_meta)), + np.nan if col == "frequency" else None, + dtype=object, + ) + + for (cml_id, sublink_id), row in metadata_indexed.iterrows(): + try: + i = sublink_ids_meta.index(sublink_id) + j = cml_ids_meta.index(cml_id) + val = row[col] + if col == "frequency": + val = float(val) if pd.notna(val) else np.nan + meta_array[i, j] = val + except (ValueError, IndexError): + pass + + ds.coords[col] = (["sublink_id", "cml_id"], meta_array) + + # Store user_id in attributes + ds.attrs["user_id"] = user_id + + logger.info(f"Built xarray dataset from {len(raw_rows)} rows") + logger.debug(f"Dataset dimensions: {dict(ds.dims)}") + logger.debug(f"Dataset variables: {list(ds.data_vars)}") + logger.debug(f"Dataset coordinates: {list(ds.coords)}") + + return ds + + +def flatten_rain_dataset(rain_ds: xr.Dataset) -> pd.DataFrame: + """ + Convert processed xarray dataset back to tabular rows for DB writing. + + Expected variables in rain_ds: + tl, wet, baseline, waa, a_rain, r + + Args: + rain_ds: Processed xarray dataset with dimensions [time, cml_id, sublink_id] + + Returns: + DataFrame with columns: time, cml_id, sublink_id, user_id, + tl, wet, baseline, waa, a_rain, r + """ + # Validate required variables + required_vars = ["tl", "wet", "baseline", "waa", "a_rain", "r"] + missing_vars = [var for var in required_vars if var not in rain_ds.variables] + + if missing_vars: + logger.warning(f"Missing variables in rain dataset: {missing_vars}") + + if rain_ds.dims.get("time", 0) == 0: + logger.debug("Empty rain dataset, returning empty DataFrame") + return pd.DataFrame( + columns=[ + "time", + "cml_id", + "sublink_id", + "user_id", + "tl", + "wet", + "baseline", + "waa", + "a_rain", + "r", + ] + ) + + # Convert to DataFrame + df = rain_ds.to_dataframe() + + # Reset index to get coordinates as columns + df = df.reset_index() + + # Filter out rows where all output variables are NaN + output_cols = [col for col in required_vars if col in df.columns] + if output_cols: + df = df.dropna(subset=output_cols, how="all") + + # Add user_id from attrs if not already present + if "user_id" not in df.columns: + df["user_id"] = rain_ds.attrs.get("user_id", "") + + # Ensure proper column order + column_order = [ + "time", + "cml_id", + "sublink_id", + "user_id", + "tl", + "wet", + "baseline", + "waa", + "a_rain", + "r", + ] + + # Only include columns that exist + existing_columns = [col for col in column_order if col in df.columns] + df = df[existing_columns] + + # Ensure time is timezone-aware + if df["time"].dt.tz is None: + df["time"] = df["time"].dt.tz_localize("UTC") + + logger.debug(f"Flattened rain dataset to {len(df)} rows") + + return df diff --git a/processor/main.py b/processor/main.py index eee9190..23f74d6 100644 --- a/processor/main.py +++ b/processor/main.py @@ -1,30 +1,188 @@ +""" +Main entry point for rain rate processing service. +Implements continuous polling loop for processing CML data. +""" + import os -import psycopg2 -import pandas as pd +import sys +import time +import logging +from datetime import datetime, timedelta, timezone +from config_loader import RainProcessingConfig +from state_manager import StateManager +from data_interface import CMLDataInterface +from dataset_builder import build_cml_dataset, flatten_rain_dataset +from registry import WorkflowRegistry -# Get the DATABASE_URL environment variable -database_url = os.getenv('DATABASE_URL') +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) -import sys +def main(): + """ + Main polling loop for rain rate processing. -def read_timescaledb_data(): - print(database_url, file=sys.stdout) - conn = psycopg2.connect(database_url) - query = 'SELECT * FROM cml_data' - df = pd.read_sql_query(query, conn) - conn.close() - return df + Flow: + 1. Load configuration (with periodic reload) + 2. For each enabled user: + a. Check if time to process (based on poll_interval and last_processed_time) + b. If yes: + - Fetch raw data for time window + - Fetch metadata + - Run workflow + - Write results + - Update state + 3. Sleep and repeat + """ + # Get database URL from environment + database_url = os.getenv("DATABASE_URL") + if not database_url: + logger.error("DATABASE_URL environment variable not set") + sys.exit(1) -def process_data(): - # Your implementation here - pass + # Initialize components + config = RainProcessingConfig() + state_manager = StateManager() + data_interface = CMLDataInterface(database_url) -if __name__ == "__main__": - import time - time.sleep(10) + logger.info("Rain processing service started") + logger.info(f"Available workflows: {WorkflowRegistry.list_variants()}") + + # Main loop + while True: + try: + # Reload config if needed + global_config = config.get_global_config() + reload_interval = global_config.get("config_reload_interval_seconds", 60) + if config.should_reload(reload_interval): + logger.info("Reloading configuration") + config.load() + + # Process enabled users + enabled_users = config.get_enabled_users() + logger.debug(f"Enabled users: {list(enabled_users.keys())}") + + for user_id, user_config in enabled_users.items(): + try: + process_user_if_ready( + user_id, user_config, state_manager, data_interface + ) + except Exception as e: + logger.error(f"Error processing user {user_id}: {e}", exc_info=True) + # Continue to next user (don't let one user's error stop others) + + # Sleep before next iteration + # Use a short sleep to allow responsive config reloading + time.sleep(10) + + except KeyboardInterrupt: + logger.info("Shutting down gracefully") + break + except Exception as e: + logger.error(f"Unexpected error in main loop: {e}", exc_info=True) + time.sleep(30) # Back off on unexpected errors + + +def process_user_if_ready( + user_id: str, + user_config: dict, + state_manager: StateManager, + data_interface: CMLDataInterface, +): + """ + Check if user is ready for processing and run if so. + + Args: + user_id: User identifier + user_config: User configuration dict from YAML + state_manager: State manager instance + data_interface: Data interface instance + """ + # Get last processed time from state + last_processed = state_manager.get_last_processed_time(user_id) + + # Initialize if first run + if last_processed is None: + logger.info(f"Initializing state for new user: {user_id}") + state_manager.initialize_user(user_id) + return # Don't process on first initialization (starts from now) + + # Check if enough time has passed + now = datetime.now(timezone.utc) + poll_interval_seconds = user_config.get("poll_interval_seconds", 900) + time_since_last = (now - last_processed).total_seconds() + + if time_since_last < poll_interval_seconds: + logger.debug( + f"User {user_id}: {time_since_last:.0f}s since last processing " + f"(need {poll_interval_seconds}s)" + ) + return - df = read_timescaledb_data() - print(df) - process_data() \ No newline at end of file + # Ready to process + logger.info(f"Processing user: {user_id}") + + # Define time window + window_end = now + window_minutes = user_config.get("data_window_minutes", 90) + window_start = window_end - timedelta(minutes=window_minutes) + + logger.info( + f"User {user_id}: fetching data from {window_start} to {window_end} " + f"({window_minutes} minutes)" + ) + + # Fetch data + raw_rows = data_interface.fetch_raw_cml_data_rows(user_id, window_start, window_end) + if raw_rows.empty: + logger.warning(f"User {user_id}: no raw data in time window, skipping") + # Still update state to avoid repeated processing attempts + state_manager.update_last_processed_time(user_id, window_end) + return + + cml_ids = raw_rows["cml_id"].unique().tolist() + metadata_rows = data_interface.fetch_cml_metadata_rows(user_id, cml_ids) + + # Build xarray dataset + cml_ds = build_cml_dataset(raw_rows, metadata_rows) + + logger.info( + f"User {user_id}: fetched {len(raw_rows)} raw data points, " + f"{len(metadata_rows)} metadata records" + ) + + # Run workflow + variant_name = user_config.get("processing_variant", "default") + try: + workflow = WorkflowRegistry.get_workflow(variant_name) + logger.info(f"User {user_id}: running workflow '{variant_name}'") + + rain_ds = workflow.process(cml_ds, window_start, window_end) + rain_data = flatten_rain_dataset(rain_ds) + + if rain_data.empty: + logger.warning(f"User {user_id}: workflow produced no output") + else: + logger.info(f"User {user_id}: workflow produced {len(rain_data)} results") + rows_written = data_interface.write_rain_data(rain_data) + logger.info(f"User {user_id}: wrote {rows_written} rows to database") + + except ValueError as e: + logger.error(f"User {user_id}: invalid workflow variant '{variant_name}': {e}") + return # Don't update state on config error + except Exception as e: + logger.error(f"User {user_id}: workflow processing failed: {e}", exc_info=True) + # Consider whether to update state on processing failure + # For now, don't update to retry on next iteration + return + + # Update state on success + state_manager.update_last_processed_time(user_id, window_end) + logger.info(f"User {user_id}: processing complete, state updated") + + +if __name__ == "__main__": + main() diff --git a/processor/notebooks/rain_processing_validation.ipynb b/processor/notebooks/rain_processing_validation.ipynb new file mode 100644 index 0000000..bad34dc --- /dev/null +++ b/processor/notebooks/rain_processing_validation.ipynb @@ -0,0 +1,454 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "5bdbab9d", + "metadata": {}, + "source": [ + "# CML Rain Rate Processing Validation\n", + "\n", + "This notebook validates rain rate processing workflows by running the **exact same production code** used by the continuous processor service.\n", + "\n", + "**Purpose:**\n", + "- Test workflow behavior on selected time windows before enabling continuous processing\n", + "- Visualize intermediate and final products (TL, wet/dry, baseline, WAA, A_rain, R)\n", + "- Check for issues in processing logic before deployment\n", + "\n", + "**Important:** This notebook imports production modules from `processor/` - it does NOT duplicate processing logic." + ] + }, + { + "cell_type": "markdown", + "id": "b79dd5cd", + "metadata": {}, + "source": [ + "## 1. Setup and Imports" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "e36dc40d", + "metadata": {}, + "outputs": [], + "source": [ + "import os\n", + "import sys\n", + "from datetime import datetime, timezone\n", + "import matplotlib.pyplot as plt\n", + "import xarray as xr\n", + "import pandas as pd\n", + "import numpy as np\n", + "\n", + "# Add processor directory to path\n", + "sys.path.insert(0, '/app')\n", + "\n", + "# Import production modules\n", + "from data_interface import CMLDataInterface\n", + "from dataset_builder import build_cml_dataset, flatten_rain_dataset\n", + "from registry import WorkflowRegistry\n", + "\n", + "%matplotlib inline" + ] + }, + { + "cell_type": "markdown", + "id": "0139baf6", + "metadata": {}, + "source": [ + "## 2. User-Selectable Parameters\n", + "\n", + "Modify these parameters to test different scenarios:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "300f181b", + "metadata": {}, + "outputs": [], + "source": [ + "# === CONFIGURATION PARAMETERS ===\n", + "\n", + "# Option 1: Use last N days of data (DEFAULT - uncomment to use)\n", + "DAYS_OF_DATA = 1 # Number of days to look back from now\n", + "USE_LATEST_DATA = True # Set to True to use DAYS_OF_DATA, False to use custom times below\n", + "\n", + "# Option 2: Use custom time window (uncomment to use, set USE_LATEST_DATA=False above)\n", + "# START_TIME = \"2026-06-20T00:00:00Z\" # Start of validation window\n", + "# END_TIME = \"2026-06-21T00:00:00Z\" # End of validation window\n", + "\n", + "USER_ID = \"demo_openmrg\" # User to validate\n", + "WORKFLOW_VARIANT = \"openmrg_basic\" # Workflow variant: 'default' or 'openmrg_basic'\n", + "PLOT_CML_ID = None # Specific CML to plot (None = auto-select first)\n", + "PLOT_SUBLINK_ID = None # Specific sublink to plot (None = auto-select first)\n", + "\n", + "# Parse timestamps based on mode\n", + "if USE_LATEST_DATA:\n", + " from datetime import timedelta\n", + " window_end = datetime.now(timezone.utc)\n", + " window_start = window_end - timedelta(days=DAYS_OF_DATA)\n", + " print(f\"Using latest {DAYS_OF_DATA} day(s) of data\")\n", + "else:\n", + " window_start = datetime.fromisoformat(START_TIME.replace('Z', '+00:00'))\n", + " window_end = datetime.fromisoformat(END_TIME.replace('Z', '+00:00'))\n", + " print(f\"Using custom time window: {START_TIME} to {END_TIME}\")\n", + "\n", + "print(f\"Validating workflow '{WORKFLOW_VARIANT}' for user '{USER_ID}'\")\n", + "print(f\"Time window: {window_start} to {window_end}\")" + ] + }, + { + "cell_type": "markdown", + "id": "143b279d", + "metadata": {}, + "source": [ + "## 3. Load Data from Database" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "def5ffdc", + "metadata": {}, + "outputs": [], + "source": [ + "# Get database URL from environment\n", + "database_url = os.getenv('DATABASE_URL', 'postgresql://myuser:mypassword@localhost:5432/mydatabase')\n", + "print(f\"Database URL: {database_url.split('@')[1].split('/')[0]}...[REDACTED]\")\n", + "\n", + "# Initialize data interface\n", + "data_interface = CMLDataInterface(database_url)\n", + "\n", + "# Fetch raw CML data\n", + "print(f\"\\nFetching raw CML data...\")\n", + "raw_rows = data_interface.fetch_raw_cml_data_rows(USER_ID, window_start, window_end)\n", + "print(f\"Retrieved {len(raw_rows)} raw data rows\")\n", + "\n", + "# TIMESTAMP ROUNDING FOR IRREGULAR SAMPLING\n", + "# Adjust based on your data's temporal resolution:\n", + "# - 60 for 1-minute data\n", + "# - 900 for 15-minute data \n", + "# - None to keep original timestamps\n", + "ROUND_SECONDS = 10 # Change this based on your data!\n", + "\n", + "print(f\"\\nRounding timestamps to {ROUND_SECONDS}s intervals...\")\n", + "metadata_rows = pd.DataFrame() # Will be fetched below after rounding\n", + "\n", + "# Fetch metadata BEFORE rounding so we have all CMLs\n", + "cml_ids_all = raw_rows['cml_id'].unique().tolist()\n", + "metadata_rows = data_interface.fetch_cml_metadata_rows(USER_ID, cml_ids_all)\n", + "print(f\"Retrieved {len(metadata_rows)} metadata records\")\n", + "\n", + "# Now apply timestamp rounding\n", + "if ROUND_SECONDS is not None and not raw_rows.empty:\n", + " original_count = len(raw_rows)\n", + " raw_rows['time'] = raw_rows['time'].dt.round(f'{ROUND_SECONDS}s')\n", + " raw_rows = raw_rows.drop_duplicates(subset=['time', 'cml_id', 'sublink_id'])\n", + " rounded_count = len(raw_rows)\n", + " if original_count != rounded_count:\n", + " print(f\"✓ Rounded to {ROUND_SECONDS}s: {original_count:,} → {rounded_count:,} rows (removed {original_count - rounded_count:,} duplicates)\")\n", + "\n", + "# Show final CML IDs after sampling/rounding\n", + "if not raw_rows.empty:\n", + " cml_ids = raw_rows['cml_id'].unique().tolist()\n", + " print(f\"CML IDs in dataset: {cml_ids[:10]}{'...' if len(cml_ids) > 10 else ''}\")\n", + "else:\n", + " print(\"No raw data found - cannot proceed\")" + ] + }, + { + "cell_type": "markdown", + "id": "b3317be6", + "metadata": {}, + "source": [ + "## 4. Build Canonical XArray Dataset" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9b754c85", + "metadata": {}, + "outputs": [], + "source": [ + "# Build xarray dataset using production function\n", + "if not raw_rows.empty:\n", + " cml_ds = build_cml_dataset(raw_rows, metadata_rows)\n", + " print(f\"\\nBuilt xarray dataset:\")\n", + " print(cml_ds)\n", + " print(f\"\\nDimensions: {dict(cml_ds.dims)}\")\n", + " print(f\"Variables: {list(cml_ds.data_vars)}\")\n", + " print(f\"User ID: {cml_ds.attrs.get('user_id')}\")\n", + "else:\n", + " print(\"Skipping dataset building - no data available\")\n", + " cml_ds = xr.Dataset()" + ] + }, + { + "cell_type": "markdown", + "id": "0d77562a", + "metadata": {}, + "source": [ + "## 5. Load and Run Workflow" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a0053689", + "metadata": {}, + "outputs": [], + "source": [ + "# Load workflow from registry (production code)\n", + "if cml_ds.dims.get('time', 0) > 0:\n", + " workflow = WorkflowRegistry.get_workflow(WORKFLOW_VARIANT)\n", + " print(f\"Loaded workflow: {workflow.get_name()}\")\n", + " \n", + " # Run processing\n", + " print(f\"\\nProcessing {cml_ds.dims.get('time', 0)} timestamps...\")\n", + " rain_ds = workflow.process(cml_ds, window_start, window_end)\n", + " \n", + " print(f\"\\nOutput dataset:\")\n", + " print(rain_ds)\n", + " print(f\"\\nOutput dimensions: {dict(rain_ds.dims)}\")\n", + " print(f\"Output variables: {list(rain_ds.data_vars)}\")\n", + "else:\n", + " print(\"Skipping workflow execution - empty input dataset\")\n", + " rain_ds = xr.Dataset()" + ] + }, + { + "cell_type": "markdown", + "id": "0f669122", + "metadata": {}, + "source": [ + "## 6. Summary Statistics" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4219e1fb", + "metadata": {}, + "outputs": [], + "source": [ + "# Calculate summary statistics\n", + "if rain_ds.dims.get('time', 0) > 0:\n", + " wet_count = int(rain_ds['wet'].sum().values)\n", + " total_count = int(rain_ds['wet'].size)\n", + " \n", + " rain_values = rain_ds['r'].values\n", + " rain_non_nan = rain_values[~np.isnan(rain_values)]\n", + " \n", + " print(\"=== RAIN RATE SUMMARY STATISTICS ===\")\n", + " print(f\"Total observations: {total_count}\")\n", + " print(f\"Wet observations: {wet_count} ({100*wet_count/total_count:.1f}%)\")\n", + " print(f\"Dry observations: {total_count - wet_count} ({100*(total_count-wet_count)/total_count:.1f}%)\")\n", + " print(f\"\\nRain rate (mm/h):\")\n", + " print(f\" Min: {np.min(rain_non_nan):.3f}\" if len(rain_non_nan) > 0 else \" Min: N/A\")\n", + " print(f\" Max: {np.max(rain_non_nan):.3f}\" if len(rain_non_nan) > 0 else \" Max: N/A\")\n", + " print(f\" Mean: {np.mean(rain_non_nan):.3f}\" if len(rain_non_nan) > 0 else \" Mean: N/A\")\n", + " print(f\" Median: {np.median(rain_non_nan):.3f}\" if len(rain_non_nan) > 0 else \" Median: N/A\")\n", + " print(f\" Std Dev: {np.std(rain_non_nan):.3f}\" if len(rain_non_nan) > 0 else \" Std Dev: N/A\")\n", + "else:\n", + " print(\"No data available for summary statistics\")" + ] + }, + { + "cell_type": "markdown", + "id": "c20ba1c4", + "metadata": {}, + "source": [ + "## 7. Select CML for Plotting" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "dbfcc8f3", + "metadata": {}, + "outputs": [], + "source": [ + "# Select CML and sublink for plotting\n", + "if rain_ds.dims.get('time', 0) > 0:\n", + " cml_ids_available = rain_ds.coords['cml_id'].values.tolist()\n", + " sublink_ids_available = rain_ds.coords['sublink_id'].values.tolist()\n", + " \n", + " # Auto-select if not specified\n", + " if PLOT_CML_ID is None:\n", + " PLOT_CML_ID = cml_ids_available[0]\n", + " if PLOT_SUBLINK_ID is None:\n", + " PLOT_SUBLINK_ID = sublink_ids_available[0]\n", + " \n", + " print(f\"Plotting CML: {PLOT_CML_ID}, Sublink: {PLOT_SUBLINK_ID}\")\n", + " \n", + " # Slice dataset for selected link\n", + " try:\n", + " link_ds = rain_ds.sel(cml_id=PLOT_CML_ID, sublink_id=PLOT_SUBLINK_ID)\n", + " print(f\"Selected {len(link_ds.time)} timestamps for plotting\")\n", + " except Exception as e:\n", + " print(f\"Error selecting link: {e}\")\n", + " link_ds = None\n", + "else:\n", + " link_ds = None\n", + " print(\"No data available for plotting\")" + ] + }, + { + "cell_type": "markdown", + "id": "c679daae", + "metadata": {}, + "source": [ + "## 8. Plot Results" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "300c670d", + "metadata": {}, + "outputs": [], + "source": [ + "# Create multi-panel plot\n", + "if link_ds is not None and len(link_ds.time) > 0:\n", + " fig, axes = plt.subplots(6, 1, figsize=(14, 18), sharex=True)\n", + " fig.suptitle(f'Rain Rate Processing Results - CML {PLOT_CML_ID}, Sublink {PLOT_SUBLINK_ID}\\nWorkflow: {WORKFLOW_VARIANT}', fontsize=14, fontweight='bold')\n", + " \n", + " time = link_ds.time.values\n", + " \n", + " # Panel 1: Total Loss (TL)\n", + " axes[0].plot(time, link_ds['tl'].values, 'b-', linewidth=1, label='TL')\n", + " axes[0].set_ylabel('TL (dB)')\n", + " axes[0].set_title('Total Loss (TSL - RSL)')\n", + " axes[0].grid(True, alpha=0.3)\n", + " axes[0].legend(loc='upper right')\n", + " \n", + " # Panel 2: Wet/Dry Classification\n", + " wet_values = link_ds['wet'].values.astype(int)\n", + " axes[1].fill_between(time, 0, 1, where=wet_values, alpha=0.5, color='blue', label='Wet')\n", + " axes[1].fill_between(time, 0, 1, where=~wet_values, alpha=0.5, color='gray', label='Dry')\n", + " axes[1].set_yticks([0, 1])\n", + " axes[1].set_yticklabels(['Dry', 'Wet'])\n", + " axes[1].set_ylabel('Classification')\n", + " axes[1].set_title('Wet/Dry Classification')\n", + " axes[1].grid(True, alpha=0.3)\n", + " axes[1].legend(loc='upper right')\n", + " \n", + " # Panel 3: Baseline\n", + " axes[2].plot(time, link_ds['tl'].values, 'b-', linewidth=0.5, alpha=0.5, label='TL')\n", + " axes[2].plot(time, link_ds['baseline'].values, 'r-', linewidth=2, label='Baseline')\n", + " axes[2].set_ylabel('Attenuation (dB)')\n", + " axes[2].set_title('Baseline Estimation')\n", + " axes[2].grid(True, alpha=0.3)\n", + " axes[2].legend(loc='upper right')\n", + " \n", + " # Panel 4: Wet Antenna Attenuation (WAA)\n", + " axes[3].plot(time, link_ds['waa'].values, 'g-', linewidth=1, label='WAA')\n", + " axes[3].set_ylabel('WAA (dB)')\n", + " axes[3].set_title('Wet Antenna Attenuation')\n", + " axes[3].grid(True, alpha=0.3)\n", + " axes[3].legend(loc='upper right')\n", + " \n", + " # Panel 5: Rain-Induced Attenuation (A_rain)\n", + " axes[4].plot(time, link_ds['a_rain'].values, 'm-', linewidth=1, label='A_rain')\n", + " axes[4].set_ylabel('A_rain (dB)')\n", + " axes[4].set_title('Rain-Induced Attenuation')\n", + " axes[4].grid(True, alpha=0.3)\n", + " axes[4].legend(loc='upper right')\n", + " \n", + " # Panel 6: Rain Rate (R)\n", + " axes[5].plot(time, link_ds['r'].values, 'r-', linewidth=1.5, label='R')\n", + " axes[5].set_ylabel('R (mm/h)')\n", + " axes[5].set_xlabel('Time')\n", + " axes[5].set_title('Rain Rate Estimate')\n", + " axes[5].grid(True, alpha=0.3)\n", + " axes[5].legend(loc='upper right')\n", + " \n", + " # Rotate x-axis labels\n", + " plt.xticks(rotation=45, ha='right')\n", + " \n", + " plt.tight_layout()\n", + " plt.show()\n", + " \n", + " # Additional plot: TL, Baseline, and A_rain overlay\n", + " fig2, ax2 = plt.subplots(1, 1, figsize=(14, 6))\n", + " ax2.plot(time, link_ds['tl'].values, 'b-', linewidth=1, alpha=0.7, label='TL')\n", + " ax2.plot(time, link_ds['baseline'].values, 'g--', linewidth=2, label='Baseline')\n", + " ax2.plot(time, link_ds['a_rain'].values, 'r-', linewidth=1.5, alpha=0.8, label='A_rain')\n", + " ax2.set_ylabel('Attenuation (dB)')\n", + " ax2.set_xlabel('Time')\n", + " ax2.set_title(f'TL, Baseline, and Rain-Induced Attenuation Overlay - CML {PLOT_CML_ID}')\n", + " ax2.grid(True, alpha=0.3)\n", + " ax2.legend(loc='upper right')\n", + " plt.xticks(rotation=45, ha='right')\n", + " plt.tight_layout()\n", + " plt.show()\n", + " \n", + "else:\n", + " print(\"No data available for plotting\")" + ] + }, + { + "cell_type": "markdown", + "id": "a6b19639", + "metadata": {}, + "source": [ + "## 9. Optional: Export Results\n", + "\n", + "Uncomment the cell below to export results to CSV for further analysis." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "e63c5a6c", + "metadata": {}, + "outputs": [], + "source": [ + "# # Flatten and export results (optional)\n", + "# if rain_ds.dims.get('time', 0) > 0:\n", + "# rain_df = flatten_rain_dataset(rain_ds)\n", + "# \n", + "# # Save to CSV\n", + "# output_file = f\"rain_processing_{USER_ID}_{WORKFLOW_VARIANT}.csv\"\n", + "# rain_df.to_csv(output_file, index=False)\n", + "# print(f\"Exported {len(rain_df)} rows to {output_file}\")\n", + "# \n", + "# # Display first few rows\n", + "# print(\"\\nFirst 10 rows:\")\n", + "# display(rain_df.head(10))\n", + "# else:\n", + "# print(\"No data to export\")" + ] + }, + { + "cell_type": "markdown", + "id": "dace4e5b", + "metadata": {}, + "source": [ + "## 10. Next Steps\n", + "\n", + "After validating the workflow in this notebook:\n", + "\n", + "1. **Review the plots and statistics** above\n", + "2. **Check for issues:**\n", + " - Are TL values reasonable (typically 0-20 dB)?\n", + " - Does wet/dry classification match expected patterns?\n", + " - Is baseline tracking dry periods correctly?\n", + " - Are rain rates realistic (0-100+ mm/h during events)?\n", + "3. **If results look good:** Enable the workflow in `processor/config/rain_processing.yml`\n", + "4. **If issues found:** Adjust workflow parameters or algorithm in `processor/workflows/`\n", + "\n", + "**Note:** This notebook does NOT write to the database. It's for validation only." + ] + } + ], + "metadata": { + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/processor/plotting.py b/processor/plotting.py new file mode 100644 index 0000000..86e239c --- /dev/null +++ b/processor/plotting.py @@ -0,0 +1,176 @@ +""" +Plotting utilities for rain rate processing validation. +Provides standard visualization functions for workflow results. +""" + +import matplotlib.pyplot as plt +import xarray as xr +import numpy as np +from typing import Optional + + +def select_single_link( + ds: xr.Dataset, cml_id: Optional[str] = None, sublink_id: Optional[str] = None +) -> xr.Dataset: + """ + Return a dataset slice for one link for plotting. + + Args: + ds: Input xarray dataset with dimensions [time, cml_id, sublink_id] + cml_id: CML ID to select (None = first available) + sublink_id: Sublink ID to select (None = first available) + + Returns: + Sliced dataset with only time dimension + """ + if ds.dims.get("time", 0) == 0: + return ds + + # Auto-select if not specified + if cml_id is None: + cml_id = ds.coords["cml_id"].values[0] + if sublink_id is None: + sublink_id = ds.coords["sublink_id"].values[0] + + try: + return ds.sel(cml_id=cml_id, sublink_id=sublink_id) + except Exception as e: + raise ValueError( + f"Could not select link (cml_id={cml_id}, sublink_id={sublink_id}): {e}" + ) + + +def plot_rain_workflow_overview( + ds: xr.Dataset, + cml_id: Optional[str] = None, + sublink_id: Optional[str] = None, + title: Optional[str] = None, + figsize: tuple = (14, 18), +) -> plt.Figure: + """ + Create a standard multi-panel plot for tl, wet, baseline, waa, a_rain, r. + + Args: + ds: Processed rain dataset with variables [tl, wet, baseline, waa, a_rain, r] + cml_id: CML ID to plot (None = first available) + sublink_id: Sublink ID to plot (None = first available) + title: Plot title (None = auto-generated) + figsize: Figure size in inches + + Returns: + Matplotlib figure object + """ + # Select single link + link_ds = select_single_link(ds, cml_id, sublink_id) + + if link_ds.dims.get("time", 0) == 0: + raise ValueError("No data available for plotting") + + # Get selected coordinates for title + plot_cml_id = link_ds.attrs.get("selected_cml_id", str(cml_id)) + plot_sublink_id = link_ds.attrs.get("selected_sublink_id", str(sublink_id)) + + if title is None: + title = f"Rain Rate Processing Results - CML {plot_cml_id}, Sublink {plot_sublink_id}" + + # Create figure with 6 panels + fig, axes = plt.subplots(6, 1, figsize=figsize, sharex=True) + fig.suptitle(title, fontsize=14, fontweight="bold") + + time = link_ds.time.values + + # Panel 1: Total Loss (TL) + axes[0].plot(time, link_ds["tl"].values, "b-", linewidth=1, label="TL") + axes[0].set_ylabel("TL (dB)") + axes[0].set_title("Total Loss (TSL - RSL)") + axes[0].grid(True, alpha=0.3) + axes[0].legend(loc="upper right") + + # Panel 2: Wet/Dry Classification + wet_values = link_ds["wet"].values.astype(int) + axes[1].fill_between( + time, 0, 1, where=wet_values, alpha=0.5, color="blue", label="Wet" + ) + axes[1].fill_between( + time, 0, 1, where=~wet_values, alpha=0.5, color="gray", label="Dry" + ) + axes[1].set_yticks([0, 1]) + axes[1].set_yticklabels(["Dry", "Wet"]) + axes[1].set_ylabel("Classification") + axes[1].set_title("Wet/Dry Classification") + axes[1].grid(True, alpha=0.3) + axes[1].legend(loc="upper right") + + # Panel 3: Baseline + axes[2].plot(time, link_ds["tl"].values, "b-", linewidth=0.5, alpha=0.5, label="TL") + axes[2].plot(time, link_ds["baseline"].values, "r-", linewidth=2, label="Baseline") + axes[2].set_ylabel("Attenuation (dB)") + axes[2].set_title("Baseline Estimation") + axes[2].grid(True, alpha=0.3) + axes[2].legend(loc="upper right") + + # Panel 4: Wet Antenna Attenuation (WAA) + axes[3].plot(time, link_ds["waa"].values, "g-", linewidth=1, label="WAA") + axes[3].set_ylabel("WAA (dB)") + axes[3].set_title("Wet Antenna Attenuation") + axes[3].grid(True, alpha=0.3) + axes[3].legend(loc="upper right") + + # Panel 5: Rain-Induced Attenuation (A_rain) + axes[4].plot(time, link_ds["a_rain"].values, "m-", linewidth=1, label="A_rain") + axes[4].set_ylabel("A_rain (dB)") + axes[4].set_title("Rain-Induced Attenuation") + axes[4].grid(True, alpha=0.3) + axes[4].legend(loc="upper right") + + # Panel 6: Rain Rate (R) + axes[5].plot(time, link_ds["r"].values, "r-", linewidth=1.5, label="R") + axes[5].set_ylabel("R (mm/h)") + axes[5].set_xlabel("Time") + axes[5].set_title("Rain Rate Estimate") + axes[5].grid(True, alpha=0.3) + axes[5].legend(loc="upper right") + + # Rotate x-axis labels + plt.xticks(rotation=45, ha="right") + plt.tight_layout() + + return fig + + +def summarize_rain_dataset(ds: xr.Dataset) -> dict: + """ + Return simple summary statistics for validation. + + Args: + ds: Processed rain dataset + + Returns: + Dictionary with summary statistics + """ + if ds.dims.get("time", 0) == 0: + return {"error": "No data available"} + + wet = ds["wet"].values + rain = ds["r"].values + + # Filter out NaN values + rain_non_nan = rain[~np.isnan(rain)] + wet_count = int(np.nansum(wet)) + total_count = int(wet.size) + + stats = { + "total_observations": total_count, + "wet_observations": wet_count, + "dry_observations": total_count - wet_count, + "wet_percentage": 100.0 * wet_count / total_count if total_count > 0 else 0.0, + "rain_min": float(np.min(rain_non_nan)) if len(rain_non_nan) > 0 else None, + "rain_max": float(np.max(rain_non_nan)) if len(rain_non_nan) > 0 else None, + "rain_mean": float(np.mean(rain_non_nan)) if len(rain_non_nan) > 0 else None, + "rain_median": ( + float(np.median(rain_non_nan)) if len(rain_non_nan) > 0 else None + ), + "rain_std": float(np.std(rain_non_nan)) if len(rain_non_nan) > 0 else None, + } + + return stats diff --git a/processor/registry.py b/processor/registry.py new file mode 100644 index 0000000..e7dc5d6 --- /dev/null +++ b/processor/registry.py @@ -0,0 +1,57 @@ +""" +Registry for mapping processing variant names to workflow classes. +""" + +from typing import Dict, Type +from workflows.base import BaseRainWorkflow +from workflows.default import DefaultRainWorkflow +from workflows.openmrg_basic import OpenMRGBasicWorkflow + + +class WorkflowRegistry: + """ + Registry mapping processing variant names to workflow classes. + """ + + _registry: Dict[str, Type[BaseRainWorkflow]] = { + "default": DefaultRainWorkflow, + "openmrg_basic": OpenMRGBasicWorkflow, + } + + @classmethod + def get_workflow(cls, variant_name: str) -> BaseRainWorkflow: + """ + Get a workflow instance by variant name. + + Args: + variant_name: Name from config (e.g., 'default') + + Returns: + Instance of the workflow class + + Raises: + ValueError: If variant_name not in registry + """ + workflow_class = cls._registry.get(variant_name) + if workflow_class is None: + raise ValueError( + f"Unknown processing variant: {variant_name}. " + f"Available: {list(cls._registry.keys())}" + ) + return workflow_class() + + @classmethod + def register(cls, variant_name: str, workflow_class: Type[BaseRainWorkflow]): + """ + Register a new workflow variant. + + Args: + variant_name: Name to use in config + workflow_class: Class inheriting from BaseRainWorkflow + """ + cls._registry[variant_name] = workflow_class + + @classmethod + def list_variants(cls) -> list: + """Return list of registered variant names.""" + return list(cls._registry.keys()) diff --git a/processor/requirements.txt b/processor/requirements.txt index 5dbbe6c..c7596d6 100644 --- a/processor/requirements.txt +++ b/processor/requirements.txt @@ -1,2 +1,6 @@ -psycopg2-binary -pandas \ No newline at end of file +psycopg2-binary>=2.9.0 +pandas>=1.5.0 +xarray>=2023.0.0 +pyyaml>=6.0 +numpy>=1.20.0 +pycomlink>=0.6 \ No newline at end of file diff --git a/processor/state_manager.py b/processor/state_manager.py new file mode 100644 index 0000000..87550ee --- /dev/null +++ b/processor/state_manager.py @@ -0,0 +1,159 @@ +""" +State manager for rain processing service. +Manages persistent state (last processed timestamps) in JSON file with file locking. +""" + +import json +import fcntl +import os +from pathlib import Path +from datetime import datetime, timezone +from typing import Optional, Dict +import logging + +logger = logging.getLogger(__name__) + + +class StateManager: + """ + Manages persistent state for rain processing service. + Uses file locking to ensure atomic updates. + """ + + def __init__(self, state_path: str = "/app/data/state/rain_processing_state.json"): + """ + Args: + state_path: Path to JSON state file + """ + self.state_path = Path(state_path) + self.state_path.parent.mkdir(parents=True, exist_ok=True) + + # Initialize empty state file if it doesn't exist + if not self.state_path.exists(): + self._write_state({}) + logger.info(f"Initialized empty state file at {self.state_path}") + + def get_last_processed_time(self, user_id: str) -> Optional[datetime]: + """ + Get the last processed timestamp for a user. + + Args: + user_id: User identifier + + Returns: + Last processed datetime (UTC), or None if user has no state + """ + try: + state = self._read_state() + user_state = state.get(user_id) + + if user_state is None: + return None + + timestamp_str = user_state.get("last_processed_time") + if timestamp_str is None: + return None + + # Parse ISO 8601 timestamp + # Handle both 'Z' suffix and '+00:00' format + timestamp_str = timestamp_str.replace("Z", "+00:00") + return datetime.fromisoformat(timestamp_str) + + except (json.JSONDecodeError, KeyError, ValueError) as e: + logger.error(f"Error reading state for user {user_id}: {e}") + return None + + def update_last_processed_time(self, user_id: str, timestamp: datetime) -> None: + """ + Update the last processed timestamp for a user. + Uses atomic write pattern: write to temp file, then rename. + + Args: + user_id: User identifier + timestamp: New last processed time (should be UTC) + """ + try: + # Ensure timestamp is timezone-aware + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=timezone.utc) + + # Read current state with exclusive lock + state = self._read_state() + + # Update user's timestamp + state[user_id] = { + "last_processed_time": timestamp.isoformat().replace("+00:00", "Z") + } + + # Write atomically + self._write_state(state) + logger.debug(f"Updated state for user {user_id}: {timestamp}") + + except Exception as e: + logger.error(f"Error updating state for user {user_id}: {e}") + raise + + def initialize_user( + self, user_id: str, timestamp: Optional[datetime] = None + ) -> None: + """ + Initialize state for a user if not already present. + Defaults to current time (UTC) to avoid backfill. + + Args: + user_id: User identifier + timestamp: Initial timestamp (defaults to now) + """ + if self.get_last_processed_time(user_id) is None: + if timestamp is None: + timestamp = datetime.now(timezone.utc) + + self.update_last_processed_time(user_id, timestamp) + logger.info(f"Initialized state for new user: {user_id}") + + def _read_state(self) -> Dict: + """Read state file with file locking.""" + if not self.state_path.exists(): + return {} + + try: + with open(self.state_path, "r") as f: + # Use shared lock for reading + fcntl.flock(f.fileno(), fcntl.LOCK_SH) + try: + content = f.read() + if not content.strip(): + return {} + return json.loads(content) + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + except (json.JSONDecodeError, IOError) as e: + logger.error(f"Error reading state file: {e}") + return {} + + def _write_state(self, state: Dict) -> None: + """Write state file atomically with file locking.""" + try: + # Create temp file path + temp_path = self.state_path.with_suffix(".tmp") + + # Write to temp file with exclusive lock + with open(temp_path, "w") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + json.dump(state, f, indent=2) + f.flush() + os.fsync(f.fileno()) + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + + # Atomic rename + os.rename(temp_path, self.state_path) + + except Exception as e: + logger.error(f"Error writing state file: {e}") + # Clean up temp file if it exists + temp_path = self.state_path.with_suffix(".tmp") + if temp_path.exists(): + temp_path.unlink() + raise diff --git a/processor/workflows/__init__.py b/processor/workflows/__init__.py new file mode 100644 index 0000000..4a5d5db --- /dev/null +++ b/processor/workflows/__init__.py @@ -0,0 +1,13 @@ +""" +Workflows package for rain rate processing. +""" + +from .base import BaseRainWorkflow +from .default import DefaultRainWorkflow +from .openmrg_basic import OpenMRGBasicWorkflow + +__all__ = [ + "BaseRainWorkflow", + "DefaultRainWorkflow", + "OpenMRGBasicWorkflow", +] diff --git a/processor/workflows/base.py b/processor/workflows/base.py new file mode 100644 index 0000000..c94d058 --- /dev/null +++ b/processor/workflows/base.py @@ -0,0 +1,56 @@ +""" +Abstract base class for rain rate processing workflows. +All workflow variants must inherit from this class and implement process(). +""" + +from abc import ABC, abstractmethod +import xarray as xr +from datetime import datetime + + +class BaseRainWorkflow(ABC): + """ + Abstract base class for rain rate processing workflows. + All workflow variants must inherit from this class and implement process(). + """ + + @abstractmethod + def process( + self, cml_ds: xr.Dataset, window_start: datetime, window_end: datetime + ) -> xr.Dataset: + """ + Process raw CML data to estimate rain rates. + + Args: + cml_ds: Canonical xarray dataset with dimensions [time, cml_id, sublink_id] + and variables including [rsl, tsl] plus metadata variables. + May contain NaN values and missing metadata. + + window_start: Start of processing window (for context) + window_end: End of processing window + + Returns: + xr.Dataset with dimensions [time, cml_id, sublink_id] and variables: + - tl: total loss (TSL - RSL) + - wet: boolean wet/dry classification + - baseline: baseline attenuation level + - waa: wet antenna attenuation estimate + - a_rain: rain-induced attenuation + - r: rain rate estimate (mm/h) + + The dataset should preserve coordinates and carry `user_id` in attrs. + All fields can be NULL/NaN if processing fails for a timestamp. + + Notes: + - Implementations should be robust to missing metadata + - Should handle gaps in time series gracefully + - Should not raise exceptions on bad data (log warnings, return partial results) + - Processing time window may be larger than output window (for temporal context) + - Output should usually be trimmed to the target interval that should be persisted, + even if a larger context window was used internally + """ + pass + + def get_name(self) -> str: + """Return workflow name for logging.""" + return self.__class__.__name__ diff --git a/processor/workflows/default.py b/processor/workflows/default.py new file mode 100644 index 0000000..97742cb --- /dev/null +++ b/processor/workflows/default.py @@ -0,0 +1,126 @@ +""" +Default rain rate processing workflow using xarray and pycomlink-style algorithms. +""" + +from .base import BaseRainWorkflow +import xarray as xr +from datetime import datetime, timezone +import logging +import numpy as np + +logger = logging.getLogger(__name__) + + +class DefaultRainWorkflow(BaseRainWorkflow): + """ + Default rain rate processing workflow using xarray and pycomlink-style algorithms. + + Processing steps: + 1. Calculate total loss (TL = TSL - RSL) + 2. Wet/dry classification + 3. Baseline estimation + 4. Wet antenna attenuation (WAA) correction + 5. Rain-induced attenuation estimation + 6. Rain rate retrieval using power-law relationship + """ + + def process( + self, cml_ds: xr.Dataset, window_start: datetime, window_end: datetime + ) -> xr.Dataset: + """ + Process CML data to rain rates using pycomlink-style algorithms. + + See BaseRainWorkflow.process() for parameter/return documentation. + """ + logger.info("Processing CML dataset with default workflow") + + if cml_ds.dims.get("time", 0) == 0: + logger.warning("Empty dataset provided, returning empty output") + return xr.Dataset() + + # Ensure timestamps are timezone-aware + if window_start.tzinfo is None: + window_start = window_start.replace(tzinfo=timezone.utc) + if window_end.tzinfo is None: + window_end = window_end.replace(tzinfo=timezone.utc) + + # Step 1: Calculate total loss (TL = TSL - RSL) + tl = cml_ds["tsl"] - cml_ds["rsl"] + + # Step 2: Wet/dry classification + # Simple threshold-based classification: TL > 0.5 dB indicates wet + # This is a placeholder - real implementation should use Schleiss algorithm + wet = tl > 0.5 + + # Step 3: Baseline estimation + # Use rolling window minimum during dry periods + # This is a simplified approach - real implementation needs more sophisticated method + baseline = tl.rolling(time=12, min_periods=1).min() + + # Step 4: Wet antenna attenuation (WAA) + # Simple constant WAA estimate (placeholder) + # Real implementation should use frequency-dependent model + waa = xr.where(wet, 1.0, 0.0) # 1 dB when wet, 0 when dry + + # Step 5: Rain-induced attenuation + a_rain = tl - baseline - waa + + # Ensure a_rain is non-negative + a_rain = xr.where(a_rain < 0, 0.0, a_rain) + + # Step 6: Rain rate retrieval + # Use power-law relationship: R = a * (A_rain / L)^b + # where a, b depend on frequency and polarization + # This is a simplified version - real implementation should use ITU-R P.838 coefficients + + # Get frequency and length from metadata (if available) + if "frequency" in cml_ds and "length" in cml_ds: + freq = cml_ds["frequency"] + length = cml_ds["length"] + + # Simplified power-law coefficients (should be frequency-dependent) + a_coeff = 0.01 # Placeholder + b_coeff = 1.0 # Placeholder + + # Calculate rain rate + # Avoid division by zero + rain_rate = xr.where( + (length > 0) & (a_rain > 0), a_coeff * (a_rain / length) ** b_coeff, 0.0 + ) + else: + logger.warning( + "Missing frequency or length metadata, using placeholder rain rates" + ) + rain_rate = xr.where(a_rain > 0, a_rain, 0.0) + + # Create output dataset + out = xr.Dataset( + coords={ + "time": cml_ds.coords["time"], + "cml_id": cml_ds.coords["cml_id"], + "sublink_id": cml_ds.coords["sublink_id"], + } + ) + + out["tl"] = tl + out["wet"] = wet + out["baseline"] = baseline + out["waa"] = waa + out["a_rain"] = a_rain + out["r"] = rain_rate + + # Preserve user_id + out.attrs["user_id"] = cml_ds.attrs.get("user_id") + + # Trim output to target window if specified + if window_start is not None and window_end is not None: + try: + out = out.sel(time=slice(window_start, window_end)) + except Exception as e: + logger.warning(f"Could not trim to time window: {e}") + + logger.info( + f"Processing complete: {out.dims.get('time', 0)} timestamps in output" + ) + + return out diff --git a/processor/workflows/openmrg_basic.py b/processor/workflows/openmrg_basic.py new file mode 100644 index 0000000..d20ded3 --- /dev/null +++ b/processor/workflows/openmrg_basic.py @@ -0,0 +1,80 @@ +""" +Basic xarray-based workflow using pycomlink algorithms. +Follows the pattern from pycomlink example notebooks. +""" + +from .base import BaseRainWorkflow +import xarray as xr +from datetime import datetime, timezone +import logging +import numpy as np + +logger = logging.getLogger(__name__) + + +class OpenMRGBasicWorkflow(BaseRainWorkflow): + """ + Basic xarray-based workflow using pycomlink algorithms. + + Processing steps follow pycomlink example: + 1. Calculate total loss (TL = TSL - RSL) + 2. Wet/dry classification using rolling std > 0.8 + 3. Baseline estimation using pycomlink baseline_constant + 4. WAA using pycomlink waa_schleiss_2013 + 5. Rain-induced attenuation: A = TL - baseline - WAA + 6. Rain rate using ITU-R power-law with frequency/polarization + """ + + def process( + self, ds_cmls: xr.Dataset, window_start: datetime, window_end: datetime + ) -> xr.Dataset: + """ + Process CML data to rain rates. + + See BaseRainWorkflow.process() for parameter/return documentation. + """ + import pycomlink as pycml + + # calculate total loss + ds_cmls["tl"] = ds_cmls.tsl - ds_cmls.rsl + # seperate periods of rain from dry time steps + ds_cmls["wet"] = ( + ds_cmls.tl.rolling(time=60 * 6, center=True).std(skipna=False) > 1.0 + ) + # estiamte the baseline during rain events + ds_cmls["baseline"] = pycml.processing.baseline.baseline_constant( + trsl=ds_cmls.tl, + wet=ds_cmls.wet, + n_average_last_dry=5, + ) + # compenmsate for wet antenna attenuation + ds_cmls["waa"] = pycml.processing.wet_antenna.waa_schleiss_2013( + rsl=ds_cmls.tl, + baseline=ds_cmls.baseline, + wet=ds_cmls.wet, + waa_max=2.2, + delta_t=1, + tau=15, + ) + # calculate attenuation caused by rain and remove negative attenuation + ds_cmls["a_rain"] = ds_cmls.tl - ds_cmls.baseline - ds_cmls.waa + ds_cmls["a_rain"].values[ds_cmls.a_rain < 0] = 0 + # derive rain rate via the k-R relation + ds_cmls["r"] = pycml.processing.k_R_relation.calc_R_from_A( + A=ds_cmls.a_rain, + L_km=ds_cmls.length.astype(float) / 1000, # convert to km + f_GHz=ds_cmls.frequency / 1000, # convert to GHz + pol=ds_cmls.polarization, + ) + ds_cmls["r"].values[ds_cmls.r < 0.1] = 0 + + # Trim output to target persistence window + if window_start is not None and window_end is not None: + try: + ds_cmls = ds_cmls.sel(time=slice(window_start, window_end)) + except Exception as e: + logger.warning( + f"Could not trim to time window [{window_start}, {window_end}]: {e}" + ) + + return ds_cmls[["tl", "wet", "baseline", "waa", "a_rain", "r"]] diff --git a/webserver/main.py b/webserver/main.py index 5792b4b..359952c 100644 --- a/webserver/main.py +++ b/webserver/main.py @@ -404,6 +404,23 @@ def realtime(): ) +@app.route("/rain-rates") +@login_required +def rain_rates(): + """Rain rates page""" + map_html = generate_cml_map(current_user.id) + cmls = get_available_cmls(current_user.id) + default_cml = cmls[0] if cmls else None + + return render_template( + "rain_rates.html", + map_html=map_html, + cmls=cmls, + selected_cml=default_cml, + grafana_org_id=current_user.grafana_org_id, + ) + + @app.route("/grafana") @login_required def grafana_root_redirect(): @@ -546,15 +563,15 @@ def api_cml_stats(): 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]), + "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 ] @@ -564,6 +581,120 @@ def api_cml_stats(): return jsonify([]) +@app.route("/api/rain-stats") +@login_required +def api_rain_stats(): + """API endpoint for fetching per-CML rain rate statistics for rain rates visualization""" + try: + # Get optional time offset token. + # Supported values: + # "0" -> latest single measurement + # "-1h" -> mean over NOW-1h .. NOW + # "-2h" -> mean over NOW-2h .. NOW-1h + # "-1d" -> mean over NOW-24h .. NOW + # "-2d" -> mean over NOW-48h .. NOW-24h + offset = request.args.get("offset", default="0", type=str) + + with user_db_scope(current_user.id) as conn: + cur = conn.cursor() + + if offset == "0": + # Fetch latest available rain data for each CML + cur.execute( + """ + SELECT DISTINCT ON (cml_id) + cml_id::text, + r AS last_rain_rate + FROM cml_rain_data_secure + ORDER BY cml_id, time DESC + """ + ) + else: + start_expr = None + end_expr = None + use_hourly_accumulation = False + + if offset.endswith("h"): + hours_offset = int(offset[:-1]) + if hours_offset >= 0: + raise ValueError("Hour offset must be negative (e.g. -1h)") + # -1h means now-1h..now, -2h means now-2h..now-1h, etc. + start_expr = f"NOW() + ({hours_offset} || ' hours')::INTERVAL" + end_expr = ( + "NOW()" + if hours_offset == -1 + else f"NOW() + ({hours_offset + 1} || ' hours')::INTERVAL" + ) + elif offset == "-1d": + start_expr = "NOW() - INTERVAL '1 day'" + end_expr = "NOW()" + use_hourly_accumulation = True + elif offset == "-2d": + start_expr = "NOW() - INTERVAL '2 day'" + end_expr = "NOW() - INTERVAL '1 day'" + use_hourly_accumulation = True + else: + raise ValueError(f"Unsupported offset: {offset}") + + if use_hourly_accumulation: + # Daily windows: accumulate hourly means + # 1) mean(r) per CML per hour, 2) sum those hourly means + cur.execute( + f""" + WITH hourly_means AS ( + SELECT + cml_id, + time_bucket('1 hour', time) AS hour_bucket, + AVG(r) AS hour_mean_r + FROM cml_rain_data_secure + WHERE time >= {start_expr} + AND time < {end_expr} + GROUP BY cml_id, hour_bucket + ) + SELECT + cml_id::text, + SUM(hour_mean_r) AS last_rain_rate + FROM hourly_means + GROUP BY cml_id + ORDER BY cml_id + """ + ) + else: + cur.execute( + f""" + SELECT + cml_id::text, + AVG(r) as last_rain_rate + FROM cml_rain_data_secure + WHERE time >= {start_expr} + AND time < {end_expr} + GROUP BY cml_id + ORDER BY cml_id + """ + ) + + data = cur.fetchall() + cur.close() + + stats = [ + { + "cml_id": str(row[0]), + "total_records": 0, + "valid_records": 0, + "last_rain_rate": safe_float(row[1]), + "mean_rain_rate_60min": None, + } + for row in data + ] + return jsonify(stats) + except Exception as e: + print(f"Error fetching rain stats: {e}") + import traceback + + traceback.print_exc() + return jsonify([]) + + @app.route("/api/data-time-range") @login_required def api_data_time_range(): diff --git a/webserver/templates/base.html b/webserver/templates/base.html index f12e26c..93736b4 100644 --- a/webserver/templates/base.html +++ b/webserver/templates/base.html @@ -31,6 +31,10 @@ Real-Time Data +
  • + Rain Rates +
  • Archive diff --git a/webserver/templates/rain_rates.html b/webserver/templates/rain_rates.html new file mode 100644 index 0000000..dec610b --- /dev/null +++ b/webserver/templates/rain_rates.html @@ -0,0 +1,497 @@ +{% extends "base.html" %} + +{% block title %}Rain Rates - GMDI Platform{% endblock %} + +{% block extra_css %} + + +{% endblock %} + +{% block content %} +
    +
    +
    +
    +
    +
    + +
    +
    +{% endblock %} + +{% block extra_js %} + + +{% endblock %} \ No newline at end of file