From 7e9fdb8e00ebb91406dda1ab7ad1b798dba5972b Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 12:16:29 +0100 Subject: [PATCH 1/5] feat: add meteosix wrf observation --- README.md | 110 ++-- config/config.exs | 57 ++ docker/docker-compose.tel.yml | 8 +- docker/docker-compose.yml | 2 +- etc/grafana/dashboards/meteosix_wrf.json | 529 ++++++++++++++++++ etc/grafana/datasource.yml | 13 + lib/chronodash/application.ex | 7 +- lib/chronodash/datasource/meteosix/wrf.ex | 6 +- lib/chronodash/metrics/location.ex | 19 +- lib/chronodash/metrics/observation.ex | 27 +- lib/chronodash/metrics/observation_data.ex | 28 + .../datasource/meteosix/wrf/forecast.ex | 150 +++-- lib/chronodash/polling/scheduler.ex | 39 ++ lib/chronodash/polling/supervisor.ex | 23 + lib/chronodash/polling/worker.ex | 138 +++++ lib/chronodash/prom_ex.ex | 8 +- lib/chronodash/prom_ex/meteosix_plugin.ex | 28 + lib/meteosix/wrf.ex | 2 +- .../20260228053251_create_metrics_final.exs | 86 +++ ...0260228072419_update_metrics_pk_upsert.exs | 60 ++ .../repo/locations/20260228053251.json | 104 ++++ .../repo/locations/20260228072419.json | 119 ++++ .../repo/observations/20260228053251.json | 159 ++++++ .../repo/observations/20260228072419.json | 147 +++++ priv/specs/chronodash/schema.dbml | 53 ++ priv/specs/schema.dbml | 53 ++ .../datasource/meteosix/wrf/forecast_test.exs | 114 ++++ test/support/http_client_mock.ex | 10 +- 28 files changed, 1957 insertions(+), 142 deletions(-) create mode 100644 etc/grafana/dashboards/meteosix_wrf.json create mode 100644 lib/chronodash/metrics/observation_data.ex create mode 100644 lib/chronodash/polling/scheduler.ex create mode 100644 lib/chronodash/polling/supervisor.ex create mode 100644 lib/chronodash/polling/worker.ex create mode 100644 lib/chronodash/prom_ex/meteosix_plugin.ex create mode 100644 priv/repo/migrations/20260228053251_create_metrics_final.exs create mode 100644 priv/repo/migrations/20260228072419_update_metrics_pk_upsert.exs create mode 100644 priv/resource_snapshots/repo/locations/20260228053251.json create mode 100644 priv/resource_snapshots/repo/locations/20260228072419.json create mode 100644 priv/resource_snapshots/repo/observations/20260228053251.json create mode 100644 priv/resource_snapshots/repo/observations/20260228072419.json create mode 100644 priv/specs/chronodash/schema.dbml create mode 100644 priv/specs/schema.dbml create mode 100644 test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs diff --git a/README.md b/README.md index e66208b..1cb59e9 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,18 @@ # Chronodash -## Project Flow & Structure Info +## Project Overview -Chronodash is an Elixir/Phoenix application designed to provide web-based functionality and business logic handling. -The project follows a clear separation of concerns: +Chronodash is a multi-datasource monitoring and monitoring application built with **Elixir**, **Phoenix**, and the **Ash Framework**. It is designed to collect time-series metrics from various providers (starting with MeteoSIX) and visualize them in **Grafana** using **TimescaleDB**. -1. **Configuration** – All environment-specific and runtime settings are under `config/`. -2. **Codebase** – Core Elixir modules live in `lib/`, split between backend logic (`chronodash`) and web interface (`chronodash_web`). -3. **Privileged Resources** – `priv/` holds static assets, database migrations, seeds, and translation files. -4. **Dockerization** – All container setup is located in `docker/` for easy deployment. -5. **Tests** – Test files are isolated under `test/` to validate modules and controllers (not detailed here). +--- + +## Key Architecture Concepts + +1. **Generic Polling Engine** – Orchestrates data collection from multiple sources using a standardized pipeline. +2. **Standardized Data Contract** – All DataSources return a unified `ObservationData` struct, decoupling fetching logic from persistence. +3. **Ash Framework** – Manages the domain logic and persistence layer with high-performance bulk operations. +4. **TimescaleDB** – Optimized storage for time-series observations, enabling efficient long-term data retention and fast queries. +5. **Observability** – Integrated with **PromEx** for system metrics and **Grafana** for business/weather metrics. --- @@ -17,57 +20,42 @@ The project follows a clear separation of concerns: ``` . -├── AGENTS.md # Documentation or definition of system agents -├── Makefile # Make script to automate project tasks -├── README.md # Main project documentation -├── config # Application configuration files -│ ├── config.exs # General project configuration -│ ├── dev.exs # Development-specific configuration -│ ├── prod.exs # Production-specific configuration -│ ├── runtime.exs # Runtime-loaded configuration -│ └── test.exs # Test environment configuration -├── docker # Docker containerization files -│ ├── Dockerfile -│ ├── docker-compose.tel.yml -│ └── docker-compose.yml -├── lib # Main Elixir source code -│ ├── chronodash # Core business logic modules -│ │ ├── application.ex -│ │ ├── mailer.ex -│ │ └── repo.ex -│ ├── chronodash.ex # Main project entry point -│ ├── chronodash_web # Web (Phoenix) layer of the project -│ │ ├── controllers # Web controllers handling routes and requests -│ │ │ ├── error_json.ex -│ │ │ └── health_controller.ex -│ │ ├── endpoint.ex -│ │ ├── gettext.ex -│ │ ├── router.ex -│ │ └── telemetry.ex -│ └── chronodash_web.ex # Web module entry point -├── mix.exs # Mix configuration file (Elixir build tool) -├── mix.lock # Dependency lockfile -├── priv # Private application resources -│ ├── gettext # Translation files -│ │ ├── en -│ │ │ └── LC_MESSAGES -│ │ │ └── errors.po -│ │ └── errors.pot # Translation template file -│ ├── repo # Database-related files -│ │ ├── migrations -│ │ └── seeds.exs -│ ├── specs # Specification files -│ │ └── cronodash -│ │ └── openapi.json -│ └── static # Static assets served by Phoenix -│ ├── favicon.ico -│ └── robots.txt -└── test # Project test files - ├── chronodash_web - │ └── controllers - │ └── error_json_test.exs - ├── support - │ ├── conn_case.ex - │ └── data_case.ex - └── test_helper.exs +├── .env # Environment variables (API keys, DB credentials, etc.) +├── .gitignore # Git ignore rules +├── Makefile # Automation scripts (deploy, attach, db-up, etc.) +├── config/ # Configuration files +├── docker/ # Deployment infrastructure +├── etc/ # Provisioning and configuration +│ ├── grafana/ # Dashboards and Datasource provisioning +│ └── prometheus/ # Scraping configuration +├── lib # Main Elixir source code +│ ├── chronodash/ # Core Application +│ │ ├── accounts/ # Ash Domain: User management +│ │ ├── datasource/ # High-level data orchestration +│ │ ├── metrics/ # Ash Domain: Locations and Observations +│ │ ├── models/ # Standardized DTOs and internal contracts +│ │ ├── polling/ # Generic Polling Engine (Supervisors/Workers) +│ │ └── prom_ex/ # Custom metrics plugins +│ ├── chronodash_web/ # Phoenix Web Layer +│ ├── http_client/ # Centralized HTTP client (Finch + Req) +│ └── meteosix/ # MeteoSIX API v5 Client +├── priv # Database and static resources +│ ├── repo/migrations/ # TimescaleDB hypertable migrations +│ └── specs/schema.dbml # Database design documentation +└── test # Unit and integration tests + ├── chronodash/ # Core logic tests + └── support/ # Mock clients and test cases ``` + + +## Getting Started + +### Prerequisites +- Docker & Docker Compose +- Elixir 1.15+ (for local development) + +### Quick Start +1. Get your MeteoSIX API Key and add it to `.env`. +2. Configure your DB credentials in `.env`. +3. Deploy the app with telemetry: `make deploy_with_tel` +4. Access Grafana: `http://localhost:3000` (admin/admin) diff --git a/config/config.exs b/config/config.exs index 2cdf9bb..6c073a5 100644 --- a/config/config.exs +++ b/config/config.exs @@ -54,6 +54,63 @@ config :logger, :default_formatter, # Use Jason for JSON parsing in Phoenix config :phoenix, :json_library, Jason +# Polling configuration +config :chronodash, :polling_jobs, [ + %{ + id: :meteosix_wrf_sky_state, + mfa: {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, [{43.37, -8.42}, :sky_state]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + }, + %{ + id: :meteosix_wrf_temperature, + mfa: {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, [{43.37, -8.42}, :temperature]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + }, + %{ + id: :meteosix_wrf_precipitation, + mfa: + {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, [{43.37, -8.42}, :precipitation_amount]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + }, + %{ + id: :meteosix_wrf_wind, + mfa: {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, [{43.37, -8.42}, :wind]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + }, + %{ + id: :meteosix_wrf_humidity, + mfa: + {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, [{43.37, -8.42}, :relative_humidity]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + }, + %{ + id: :meteosix_wrf_clouds, + mfa: + {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, [{43.37, -8.42}, :cloud_area_fraction]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + }, + %{ + id: :meteosix_wrf_pressure, + mfa: + {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, + [{43.37, -8.42}, :air_pressure_at_sea_level]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + }, + %{ + id: :meteosix_wrf_snow, + mfa: {Chronodash.DataSource.MeteoSIX.WRF, :get_forecast, [{43.37, -8.42}, :snow_level]}, + rate: :timer.hours(6), + metadata: %{source: "meteosix"} + } +] + # Finch configuration config :chronodash, :http_client, Chronodash.HttpClient.Finch diff --git a/docker/docker-compose.tel.yml b/docker/docker-compose.tel.yml index 61d2668..57689b7 100644 --- a/docker/docker-compose.tel.yml +++ b/docker/docker-compose.tel.yml @@ -7,7 +7,7 @@ services: ports: - "9090:9090" networks: - - cronodash + - chronodash grafana: image: grafana/grafana:latest @@ -23,9 +23,9 @@ services: depends_on: - prometheus networks: - - cronodash + - chronodash networks: - cronodash: + chronodash: external: true - name: ash-postgres-network + name: chronodash diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 760ac67..80fc8c3 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -53,5 +53,5 @@ services: networks: chronodash: - name: cronodash + name: chronodash driver: bridge diff --git a/etc/grafana/dashboards/meteosix_wrf.json b/etc/grafana/dashboards/meteosix_wrf.json new file mode 100644 index 0000000..f2078ea --- /dev/null +++ b/etc/grafana/dashboards/meteosix_wrf.json @@ -0,0 +1,529 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Graphics --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 0, + "id": null, + "links": [], + "liveNow": false, + "panels": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "celsius" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 0 + }, + "id": 1, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "group": [], + "metricColumn": "none", + "rawQuery": true, +"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Temperature\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'temperature'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A", + "select": [ + [ + { + "params": [ + "value" + ], + "type": "column" + } + ] + ], + "timeColumn": "timestamp", + "where": [] + } + ], + "title": "Temperature Forecast", + "type": "timeseries" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "bars", + "fillOpacity": 80, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "lengthmm" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 0 + }, + "id": 2, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawQuery": true, +"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n SUM(value) as \"Precipitation\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'precipitation_amount'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" + } + ], + "title": "Precipitation Forecast", + "type": "timeseries" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "mappings": [ + { + "options": { + "1": { + "color": "yellow", + "index": 0, + "text": "SUNNY" + }, + "2": { + "color": "light-blue", + "index": 1, + "text": "PARTLY_CLOUDY" + }, + "3": { + "color": "blue", + "index": 2, + "text": "HIGH_CLOUDS" + }, + "4": { + "color": "grey", + "index": 3, + "text": "CLOUDY" + }, + "5": { + "color": "dark-grey", + "index": 4, + "text": "OVERCAST" + } + }, + "type": "value" + } + ], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 4, + "w": 24, + "x": 0, + "y": 8 + }, + "id": 3, + "options": { + "alignValue": "left", + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "mergeValues": true, + "rowHeight": 0.9, + "showValue": "always", + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawQuery": true, +"rawSql": "SELECT\n timestamp as time,\n value as \"Sky State\",\n locations.name as location\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'sky_state'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nORDER BY 1", + "refId": "A" + } + ], + "title": "Sky State Timeline", + "type": "state-timeline" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "unit": "percent" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 12 + }, + "id": 4, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawQuery": true, +"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Humidity\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'relative_humidity'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" + } + ], + "title": "Relative Humidity", + "type": "timeseries" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "unit": "percent" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 12 + }, + "id": 5, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawQuery": true, +"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Clouds\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'cloud_area_fraction'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" + } + ], + "title": "Cloud Coverage", + "type": "timeseries" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "unit": "velocitykmh" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 12 + }, + "id": 6, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawQuery": true, +"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Wind Speed\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'wind'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" + } + ], + "title": "Wind Speed", + "type": "timeseries" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "unit": "pressurehpa" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 20 + }, + "id": 7, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawQuery": true, +"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Pressure\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'air_pressure_at_sea_level'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" + } + ], + "title": "Air Pressure", + "type": "timeseries" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "unit": "lengthm" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 20 + }, + "id": 8, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawQuery": true, +"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Snow Level\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'snow_level'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" + } + ], + "title": "Snow Level", + "type": "timeseries" + } + ], + "schemaVersion": 38, + "style": "dark", + "tags": [ + "meteosix", + "weather" + ], + "templating": { + "list": [ + { + "current": {}, + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "definition": "SELECT name FROM locations", + "hide": 0, + "includeAll": true, + "multi": true, + "name": "location", + "options": [], + "query": "SELECT name FROM locations", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "sort": 1, + "type": "query" + } + ] + }, + "time": { + "from": "now", + "to": "now+24h" + }, + "timepicker": {}, + "timezone": "", + "title": "MeteoSIX WRF Forecast", + "uid": "meteosix_wrf", + "version": 1 +} diff --git a/etc/grafana/datasource.yml b/etc/grafana/datasource.yml index a221c3c..c7eefcb 100644 --- a/etc/grafana/datasource.yml +++ b/etc/grafana/datasource.yml @@ -7,3 +7,16 @@ datasources: url: http://prometheus:9090 isDefault: true editable: true + + - name: PostgreSQL + type: postgres + access: proxy + url: postgres:5432 + user: postgres + secureJsonData: + password: postgres + jsonData: + database: chronodash_dev + sslmode: 'disable' + postgresVersion: 1600 + timescaledb: true diff --git a/lib/chronodash/application.ex b/lib/chronodash/application.ex index 52040ab..6ed0c84 100644 --- a/lib/chronodash/application.ex +++ b/lib/chronodash/application.ex @@ -10,6 +10,7 @@ defmodule Chronodash.Application do children = [ Chronodash.PromEx, ChronodashWeb.Telemetry, + {Registry, keys: :unique, name: Chronodash.Registry}, Chronodash.Repo, {DNSCluster, query: Application.get_env(:chronodash, :dns_cluster_query) || :ignore}, {Phoenix.PubSub, name: Chronodash.PubSub}, @@ -17,8 +18,10 @@ defmodule Chronodash.Application do # {Chronodash.Worker, arg}, # Start to serve requests, typically the last entry ChronodashWeb.Endpoint, - {Finch, - Application.get_env(:chronodash, :default_http_client_config, name: Chronodash.Finch)} + {Finch, name: Chronodash.Finch}, + # Polling Services + Chronodash.Polling.Supervisor, + Chronodash.Polling.Scheduler ] Chronodash.Release.create_and_migrate() diff --git a/lib/chronodash/datasource/meteosix/wrf.ex b/lib/chronodash/datasource/meteosix/wrf.ex index f56b3b4..656521a 100644 --- a/lib/chronodash/datasource/meteosix/wrf.ex +++ b/lib/chronodash/datasource/meteosix/wrf.ex @@ -2,16 +2,16 @@ defmodule Chronodash.DataSource.MeteoSIX.WRF do @moduledoc """ High-level service for fetching WRF Forecast from MeteoSIX. """ - alias MeteoSIX.WRF alias Chronodash.Models.DataSource.MeteoSIX.WRF.Forecast + alias MeteoSIX.WRF @doc """ - Fetches forecast from MeteoSIX and parses it into a DTO. + Fetches forecast from MeteoSIX and returns a list of standardized ObservationData structs. """ def get_forecast(id_or_coords, var_atom, opts \\ []) do case WRF.get_forecast(id_or_coords, var_atom, opts) do {:ok, response} -> - {:ok, Forecast.new(response, var_atom)} + {:ok, Forecast.to_observation_data(response, var_atom)} {:error, reason} -> {:error, reason} diff --git a/lib/chronodash/metrics/location.ex b/lib/chronodash/metrics/location.ex index 16f492d..e07a40d 100644 --- a/lib/chronodash/metrics/location.ex +++ b/lib/chronodash/metrics/location.ex @@ -26,7 +26,6 @@ defmodule Chronodash.Metrics.Location do end attribute :external_id, :string do - description("ID used by external providers (e.g. MeteoSIX locationId)") public?(true) end @@ -34,12 +33,18 @@ defmodule Chronodash.Metrics.Location do end actions do - defaults([ - :read, - :destroy, - create: [:name, :latitude, :longitude, :external_id], - update: [:name, :latitude, :longitude, :external_id] - ]) + defaults([:read, :destroy]) + + create :create do + accept([:name, :latitude, :longitude, :external_id]) + upsert?(true) + upsert_identity(:name) + upsert_fields([:latitude, :longitude, :external_id]) + end + end + + identities do + identity(:name, [:name]) end postgres do diff --git a/lib/chronodash/metrics/observation.ex b/lib/chronodash/metrics/observation.ex index 9650988..a7d03f5 100644 --- a/lib/chronodash/metrics/observation.ex +++ b/lib/chronodash/metrics/observation.ex @@ -1,7 +1,7 @@ defmodule Chronodash.Metrics.Observation do @moduledoc """ Unified table for all time-series metrics. - Designed to be a TimescaleDB hypertable. + Designed to be a TimescaleDB hypertable with composite PK for upserts. """ use Ash.Resource, otp_app: :chronodash, @@ -9,22 +9,22 @@ defmodule Chronodash.Metrics.Observation do data_layer: AshPostgres.DataLayer attributes do - # TimescaleDB hypertables with unique indexes (including PK) - # MUST include the partitioning column (timestamp). - uuid_primary_key :id do - default(&Ash.UUID.generate/0) + attribute :location_id, :uuid do + primary_key?(true) + allow_nil?(false) public?(true) end - attribute :timestamp, :utc_datetime_usec do + attribute :metric_type, :string do primary_key?(true) allow_nil?(false) + description("e.g., temperature, wind_speed, sky_state") public?(true) end - attribute :metric_type, :string do + attribute :timestamp, :utc_datetime_usec do + primary_key?(true) allow_nil?(false) - description("e.g., temperature, wind_speed, sky_state") public?(true) end @@ -54,6 +54,8 @@ defmodule Chronodash.Metrics.Observation do relationships do belongs_to :location, Chronodash.Metrics.Location do public?(true) + source_attribute(:location_id) + allow_nil?(false) end end @@ -61,10 +63,17 @@ defmodule Chronodash.Metrics.Observation do defaults([:read, :destroy]) create :create do - accept([:metric_type, :value, :unit, :timestamp, :source, :metadata, :location_id]) + accept([:location_id, :metric_type, :value, :unit, :timestamp, :source, :metadata]) + upsert?(true) + upsert_identity(:unique_observation) + upsert_fields([:value, :unit, :source, :metadata]) end end + identities do + identity(:unique_observation, [:location_id, :metric_type, :timestamp]) + end + postgres do table "observations" repo Chronodash.Repo diff --git a/lib/chronodash/metrics/observation_data.ex b/lib/chronodash/metrics/observation_data.ex new file mode 100644 index 0000000..e1b326b --- /dev/null +++ b/lib/chronodash/metrics/observation_data.ex @@ -0,0 +1,28 @@ +defmodule Chronodash.Metrics.ObservationData do + @moduledoc """ + A standardized internal struct representing a single metric observation. + """ + defstruct [ + :location_name, + :latitude, + :longitude, + :metric_type, + :value, + :unit, + :timestamp, + :source, + :metadata + ] + + @type t :: %__MODULE__{ + location_name: String.t(), + latitude: float() | nil, + longitude: float() | nil, + metric_type: String.t(), + value: float(), + unit: String.t() | nil, + timestamp: DateTime.t(), + source: String.t(), + metadata: map() + } +end diff --git a/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex b/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex index 8681281..92d2b93 100644 --- a/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex +++ b/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex @@ -1,62 +1,126 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.Forecast do @moduledoc """ - DTO for MeteoSIX WRF Forecast data. + Parser for MeteoSIX WRF Forecast data. + Converts raw MeteoSIX JSON into standardized ObservationData structs. """ - defstruct [ - :location_name, - :coords, - :metric_type, - # List of %{timestamp: DateTime.t(), value: any()} - :values - ] - - @type t :: %__MODULE__{ - location_name: String.t(), - coords: {float(), float()}, - metric_type: atom(), - values: list(%{timestamp: DateTime.t(), value: any()}) - } + alias Chronodash.Metrics.ObservationData @doc """ - Parses a MeteoSIX response into a Forecast DTO. + Parses a MeteoSIX response into a list of ObservationData. """ - def new(response, metric_type) do + def to_observation_data(response, metric_type) do case response["features"] do [feature | _] -> - %__MODULE__{ - location_name: feature["properties"]["name"], - coords: extract_coords(feature["geometry"]), - metric_type: metric_type, - values: parse_values(feature["properties"]["days"], metric_type) - } + coords = extract_coords(feature["geometry"]) + name = feature["properties"]["name"] || default_name(coords) + days = feature["properties"]["days"] || [] + + parse_days(days, name, coords, metric_type) _ -> - {:error, :no_data} + [] end end - defp extract_coords(%{"coordinates" => [lon, lat]}), do: {lat, lon} - defp extract_coords(_), do: nil - - defp parse_values(days, metric_type) when is_list(days) do - Enum.flat_map(days, fn day -> - # Find the variable in this day - variable = Enum.find(day["variables"], fn v -> v["name"] == to_string(metric_type) end) - - if variable do - Enum.map(variable["values"], fn val -> - %{ - timestamp: parse_timestamp(val["timeInstant"]), - value: val["value"] - } - end) - else - [] - end + defp parse_days(days, location_name, coords, metric_type) do + Enum.flat_map(days, &process_day(&1, location_name, coords, metric_type)) + end + + defp process_day(day, location_name, coords, metric_type) do + case Enum.find(day["variables"], fn v -> v["name"] == to_string(metric_type) end) do + nil -> [] + variable -> process_variable(variable, location_name, coords, metric_type) + end + end + + defp process_variable(variable, location_name, coords, metric_type) do + Enum.flat_map(variable["values"], fn val -> + timestamp = parse_timestamp(val["timeInstant"]) + extract_metrics(val, variable, metric_type, location_name, coords, timestamp) end) end - defp parse_values(_, _), do: [] + # Special handling for composite variables (Wind) + defp extract_metrics(val, variable, :wind, name, coords, ts) do + [ + build_base( + name, + coords, + "wind_module", + val["moduleValue"], + variable["moduleUnits"], + ts, + val + ), + build_base( + name, + coords, + "wind_direction", + val["directionValue"], + variable["directionUnits"], + ts, + val + ) + ] + end + + # Default handling for simple variables + defp extract_metrics(val, variable, metric_type, name, coords, ts) do + [ + build_base(name, coords, to_string(metric_type), val["value"], variable["units"], ts, val) + ] + end + + defp build_base(name, {lat, lon}, type, val, unit, ts, raw) do + %ObservationData{ + location_name: name, + latitude: lat, + longitude: lon, + metric_type: type, + value: normalize_value(val), + unit: unit, + timestamp: ts, + source: "meteosix", + metadata: %{original_value: val, raw_payload: raw} + } + end + + defp normalize_value(v) when is_number(v), do: v * 1.0 + + defp normalize_value(v) when is_binary(v) do + cond do + Regex.match?(~r/^-?\d+(\.\d+)?$/, v) -> + {f, _} = Float.parse(v) + f + + # Sky States + v == "SUNNY" -> + 1.0 + + v == "PARTLY_CLOUDY" -> + 2.0 + + v == "HIGH_CLOUDS" -> + 3.0 + + v == "CLOUDY" -> + 4.0 + + v == "OVERCAST" -> + 5.0 + + true -> + 0.0 + end + end + + defp normalize_value(_), do: 0.0 + + defp extract_coords(%{"coordinates" => [lon, lat]}), do: {lat, lon} + defp extract_coords(_), do: {0.0, 0.0} + + defp default_name({lat, lon}), do: "#{lat}, #{lon}" + defp default_name(_), do: "Unknown Location" defp parse_timestamp(nil), do: nil diff --git a/lib/chronodash/polling/scheduler.ex b/lib/chronodash/polling/scheduler.ex new file mode 100644 index 0000000..757b0a0 --- /dev/null +++ b/lib/chronodash/polling/scheduler.ex @@ -0,0 +1,39 @@ +defmodule Chronodash.Polling.Scheduler do + @moduledoc """ + Reads configured polling jobs and starts them in the Polling Supervisor. + """ + use GenServer + require Logger + + alias Chronodash.Polling.Supervisor, as: PollingSupervisor + + def start_link(init_arg) do + GenServer.start_link(__MODULE__, init_arg, name: __MODULE__) + end + + @impl true + def init(_init_arg) do + send(self(), :start_jobs) + {:ok, []} + end + + @impl true + def handle_info(:start_jobs, state) do + jobs = Application.get_env(:chronodash, :polling_jobs, []) + + Enum.each(jobs, fn job -> + case PollingSupervisor.start_child(job) do + {:ok, _pid} -> + Logger.info("Successfully started polling job: #{job.id}") + + {:error, {:already_started, _pid}} -> + Logger.debug("Polling job #{job.id} already started") + + {:error, reason} -> + Logger.error("Failed to start polling job #{job.id}: #{inspect(reason)}") + end + end) + + {:noreply, state} + end +end diff --git a/lib/chronodash/polling/supervisor.ex b/lib/chronodash/polling/supervisor.ex new file mode 100644 index 0000000..3971de6 --- /dev/null +++ b/lib/chronodash/polling/supervisor.ex @@ -0,0 +1,23 @@ +defmodule Chronodash.Polling.Supervisor do + @moduledoc """ + A dynamic supervisor to manage individual polling workers. + """ + use DynamicSupervisor + + def start_link(init_arg) do + DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__) + end + + @impl true + def init(_init_arg) do + DynamicSupervisor.init(strategy: :one_for_one) + end + + @doc """ + Adds a new polling job to the supervisor. + """ + def start_child(opts) do + spec = {Chronodash.Polling.Worker, opts} + DynamicSupervisor.start_child(__MODULE__, spec) + end +end diff --git a/lib/chronodash/polling/worker.ex b/lib/chronodash/polling/worker.ex new file mode 100644 index 0000000..d701083 --- /dev/null +++ b/lib/chronodash/polling/worker.ex @@ -0,0 +1,138 @@ +defmodule Chronodash.Polling.Worker do + @moduledoc """ + A generic worker that polls a datasource and saves results in bulk. + Decoupled from specific datasource logic via standardize ObservationData. + """ + use GenServer + require Logger + require Ash.Query + + @doc """ + Starts a polling worker. + """ + def start_link(opts) do + id = Map.fetch!(opts, :id) + GenServer.start_link(__MODULE__, opts, name: via_tuple(id)) + end + + defp via_tuple(id), do: {:via, Registry, {Chronodash.Registry, {:polling_worker, id}}} + + @impl true + def init(opts) do + state = %{ + id: Map.fetch!(opts, :id), + mfa: Map.fetch!(opts, :mfa), + rate: Map.fetch!(opts, :rate), + metadata: Map.get(opts, :metadata, %{}) + } + + send(self(), :poll) + {:ok, state} + end + + @impl true + def handle_info(:poll, state) do + {mod, fun, args} = state.mfa + + case apply(mod, fun, args) do + {:ok, observations} when is_list(observations) -> + handle_observations(observations, state) + + {:error, reason} -> + Logger.error("Polling job #{state.id} failed: #{inspect(reason)}") + emit_run_telemetry(state, :error) + end + + schedule_next(state.rate) + {:noreply, state} + end + + defp handle_observations([], state) do + Logger.debug("Polling job #{state.id} returned no data.") + emit_run_telemetry(state, :success) + end + + defp handle_observations(observations, state) do + # 1. Ensure location exists (all observations in a poll usually share a location) + first = List.first(observations) + + case get_or_create_location(first) do + {:ok, location} -> + # 2. Enrich observations with location_id and convert to maps for bulk create + inputs = + Enum.map(observations, fn obs -> + emit_observation_telemetry(obs, state) + + %{ + location_id: location.id, + metric_type: obs.metric_type, + value: obs.value, + unit: obs.unit, + timestamp: obs.timestamp, + source: obs.source, + metadata: obs.metadata + } + end) + + # 3. Bulk Upsert + case Ash.bulk_create(inputs, Chronodash.Metrics.Observation, :create, + domain: Chronodash.Metrics + ) do + %Ash.BulkResult{status: :success} -> + Logger.info( + "Polling job #{state.id}: successfully saved #{length(inputs)} observations." + ) + + emit_run_telemetry(state, :success) + + %Ash.BulkResult{errors: errors} -> + Logger.error( + "Polling job #{state.id}: failed to bulk save observations: #{inspect(errors)}" + ) + + emit_run_telemetry(state, :error) + end + + {:error, error} -> + Logger.error("Polling job #{state.id}: could not resolve location: #{inspect(error)}") + emit_run_telemetry(state, :error) + end + end + + defp get_or_create_location(obs) do + params = %{ + name: obs.location_name, + latitude: obs.latitude, + longitude: obs.longitude + } + + case Chronodash.Metrics.Location + |> Ash.Changeset.for_create(:create, params) + |> Ash.create(domain: Chronodash.Metrics) do + {:ok, loc} -> {:ok, loc} + {:error, error} -> {:error, error} + end + end + + defp emit_observation_telemetry(obs, state) do + measurements = %{value: obs.value} + + metadata = + Map.merge(state.metadata, %{ + location: obs.location_name, + variable: obs.metric_type, + timestamp: obs.timestamp + }) + + :telemetry.execute([:chronodash, :polling, :observation], measurements, metadata) + end + + defp emit_run_telemetry(state, status) do + metadata = Map.merge(state.metadata, %{status: status, job_id: state.id}) + :telemetry.execute([:chronodash, :polling, :run], %{count: 1}, metadata) + end + + defp schedule_next(rate) do + Process.send_after(self(), :poll, rate) + end +end diff --git a/lib/chronodash/prom_ex.ex b/lib/chronodash/prom_ex.ex index c94da93..a2476c6 100644 --- a/lib/chronodash/prom_ex.ex +++ b/lib/chronodash/prom_ex.ex @@ -63,14 +63,10 @@ defmodule Chronodash.PromEx do Plugins.Application, Plugins.Beam, {Plugins.Phoenix, router: ChronodashWeb.Router, endpoint: ChronodashWeb.Endpoint}, - {Plugins.Ecto, otp_app: :chronodash, repos: [Chronodash.Repo]} - # Plugins.Oban, - # Plugins.PhoenixLiveView, - # Plugins.Absinthe, - # Plugins.Broadway, + {Plugins.Ecto, otp_app: :chronodash, repos: [Chronodash.Repo]}, # Add your own PromEx metrics plugins - # Chronodash.Users.PromExPlugin + Chronodash.PromEx.MeteoSIXPlugin ] end diff --git a/lib/chronodash/prom_ex/meteosix_plugin.ex b/lib/chronodash/prom_ex/meteosix_plugin.ex new file mode 100644 index 0000000..200f8aa --- /dev/null +++ b/lib/chronodash/prom_ex/meteosix_plugin.ex @@ -0,0 +1,28 @@ +defmodule Chronodash.PromEx.MeteoSIXPlugin do + @moduledoc """ + PromEx plugin for MeteoSIX metrics collected via the generic polling engine. + """ + use PromEx.Plugin + + @impl true + def event_metrics(_opts) do + [ + meteosix_wrf_metrics() + ] + end + + defp meteosix_wrf_metrics do + Event.build( + :chronodash_meteosix_wrf_event, + [ + last_value( + [:chronodash, :meteosix, :wrf, :observation, :value], + event_name: [:chronodash, :polling, :observation], + description: "MeteoSIX WRF Forecast observation value.", + measurement: :value, + tags: [:variable, :location, :source] + ) + ] + ) + end +end diff --git a/lib/meteosix/wrf.ex b/lib/meteosix/wrf.ex index a70d3e8..196b197 100644 --- a/lib/meteosix/wrf.ex +++ b/lib/meteosix/wrf.ex @@ -4,7 +4,7 @@ defmodule MeteoSIX.WRF do """ alias MeteoSIX - @type vars :: + @type variable :: :sky_state | :temperature | :precipitation_amount diff --git a/priv/repo/migrations/20260228053251_create_metrics_final.exs b/priv/repo/migrations/20260228053251_create_metrics_final.exs new file mode 100644 index 0000000..6155957 --- /dev/null +++ b/priv/repo/migrations/20260228053251_create_metrics_final.exs @@ -0,0 +1,86 @@ +defmodule Chronodash.Repo.Migrations.CreateMetricsFinal do + @moduledoc """ + Updates resources based on their most recent snapshots. + + This file was autogenerated with `mix ash_postgres.generate_migrations` + """ + + use Ecto.Migration + + def up do + execute "CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;" + + create table(:observations, primary_key: false) do + add :id, :uuid, null: false, default: fragment("gen_random_uuid()"), primary_key: true + add :timestamp, :utc_datetime_usec, null: false, primary_key: true + add :metric_type, :text, null: false + add :value, :float, null: false + add :unit, :text + add :source, :text, null: false + add :metadata, :map, default: %{} + + add :inserted_at, :timestamptz, + null: false, + default: fragment("(now() AT TIME ZONE 'utc')") + + add :updated_at, :timestamptz, + null: false, + default: fragment("(now() AT TIME ZONE 'utc')") + + add :location_id, :uuid + end + + execute "SELECT create_hypertable('observations', 'timestamp');" + + create table(:locations, primary_key: false) do + add :id, :uuid, null: false, default: fragment("gen_random_uuid()"), primary_key: true + end + + alter table(:observations) do + modify :location_id, + references(:locations, + column: :id, + name: "observations_location_id_fkey", + type: :uuid, + prefix: "public", + on_delete: :delete_all + ) + end + + alter table(:locations) do + add :name, :text, null: false + add :latitude, :float, null: false + add :longitude, :float, null: false + add :external_id, :text + + add :inserted_at, :timestamptz, + null: false, + default: fragment("(now() AT TIME ZONE 'utc')") + + add :updated_at, :timestamptz, + null: false, + default: fragment("(now() AT TIME ZONE 'utc')") + end + end + + def down do + alter table(:locations) do + remove :updated_at + remove :inserted_at + remove :external_id + remove :longitude + remove :latitude + remove :name + end + + drop constraint(:observations, "observations_location_id_fkey") + + alter table(:observations) do + modify :location_id, :uuid + end + + drop table(:locations) + + drop table(:observations) + end +end diff --git a/priv/repo/migrations/20260228072419_update_metrics_pk_upsert.exs b/priv/repo/migrations/20260228072419_update_metrics_pk_upsert.exs new file mode 100644 index 0000000..529fb94 --- /dev/null +++ b/priv/repo/migrations/20260228072419_update_metrics_pk_upsert.exs @@ -0,0 +1,60 @@ +defmodule Chronodash.Repo.Migrations.UpdateMetricsPkUpsert do + @moduledoc """ + Updates resources based on their most recent snapshots. + + This file was autogenerated with `mix ash_postgres.generate_migrations` + """ + + use Ecto.Migration + + def up do + drop table(:observations) + + create table(:observations, primary_key: false) do + add :location_id, :uuid, null: false, primary_key: true + add :metric_type, :text, null: false, primary_key: true + add :timestamp, :utc_datetime_usec, null: false, primary_key: true + add :value, :float, null: false + add :unit, :text + add :source, :text, null: false + add :metadata, :map, default: %{} + + add :inserted_at, :timestamptz, + null: false, + default: fragment("(now() AT TIME ZONE 'utc')") + + add :updated_at, :timestamptz, + null: false, + default: fragment("(now() AT TIME ZONE 'utc')") + end + + execute "SELECT create_hypertable('observations', 'timestamp');" + + alter table(:observations) do + modify :location_id, + references(:locations, + column: :id, + name: "observations_location_id_fkey", + type: :uuid, + prefix: "public", + on_delete: :delete_all + ) + end + + create unique_index(:locations, [:name], name: "locations_name_index") + end + + def down do + drop constraint("observations", "observations_pkey") + + drop_if_exists unique_index(:locations, [:name], name: "locations_name_index") + + alter table(:observations) do + modify :location_id, :uuid, null: true + modify :metric_type, :text + add :id, :uuid, null: false, default: fragment("gen_random_uuid()") + end + + execute("ALTER TABLE \"observations\" ADD PRIMARY KEY (id, timestamp)") + end +end diff --git a/priv/resource_snapshots/repo/locations/20260228053251.json b/priv/resource_snapshots/repo/locations/20260228053251.json new file mode 100644 index 0000000..a6f3879 --- /dev/null +++ b/priv/resource_snapshots/repo/locations/20260228053251.json @@ -0,0 +1,104 @@ +{ + "attributes": [ + { + "allow_nil?": false, + "default": "fragment(\"gen_random_uuid()\")", + "generated?": false, + "precision": null, + "primary_key?": true, + "references": null, + "scale": null, + "size": null, + "source": "id", + "type": "uuid" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "name", + "type": "text" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "latitude", + "type": "float" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "longitude", + "type": "float" + }, + { + "allow_nil?": true, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "external_id", + "type": "text" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "inserted_at", + "type": "utc_datetime_usec" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "updated_at", + "type": "utc_datetime_usec" + } + ], + "base_filter": null, + "check_constraints": [], + "create_table_options": null, + "custom_indexes": [], + "custom_statements": [], + "has_create_action": true, + "hash": "B8B321A5D277CBCCF27686F8B5A6B9B8913BF9862FE3649926EF0F5375BC11D7", + "identities": [], + "multitenancy": { + "attribute": null, + "global": null, + "strategy": null + }, + "repo": "Elixir.Chronodash.Repo", + "schema": null, + "table": "locations" +} \ No newline at end of file diff --git a/priv/resource_snapshots/repo/locations/20260228072419.json b/priv/resource_snapshots/repo/locations/20260228072419.json new file mode 100644 index 0000000..2b0a1c1 --- /dev/null +++ b/priv/resource_snapshots/repo/locations/20260228072419.json @@ -0,0 +1,119 @@ +{ + "attributes": [ + { + "allow_nil?": false, + "default": "fragment(\"gen_random_uuid()\")", + "generated?": false, + "precision": null, + "primary_key?": true, + "references": null, + "scale": null, + "size": null, + "source": "id", + "type": "uuid" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "name", + "type": "text" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "latitude", + "type": "float" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "longitude", + "type": "float" + }, + { + "allow_nil?": true, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "external_id", + "type": "text" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "inserted_at", + "type": "utc_datetime_usec" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "updated_at", + "type": "utc_datetime_usec" + } + ], + "base_filter": null, + "check_constraints": [], + "create_table_options": null, + "custom_indexes": [], + "custom_statements": [], + "has_create_action": true, + "hash": "013B7CAC7BE307EDC0FA391C5D40AE1E0B4E16E8600D612031C2B54C3E23B1A1", + "identities": [ + { + "all_tenants?": false, + "base_filter": null, + "index_name": "locations_name_index", + "keys": [ + { + "type": "atom", + "value": "name" + } + ], + "name": "name", + "nils_distinct?": true, + "where": null + } + ], + "multitenancy": { + "attribute": null, + "global": null, + "strategy": null + }, + "repo": "Elixir.Chronodash.Repo", + "schema": null, + "table": "locations" +} \ No newline at end of file diff --git a/priv/resource_snapshots/repo/observations/20260228053251.json b/priv/resource_snapshots/repo/observations/20260228053251.json new file mode 100644 index 0000000..496188f --- /dev/null +++ b/priv/resource_snapshots/repo/observations/20260228053251.json @@ -0,0 +1,159 @@ +{ + "attributes": [ + { + "allow_nil?": false, + "default": "fragment(\"gen_random_uuid()\")", + "generated?": false, + "precision": null, + "primary_key?": true, + "references": null, + "scale": null, + "size": null, + "source": "id", + "type": "uuid" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": true, + "references": null, + "scale": null, + "size": null, + "source": "timestamp", + "type": "utc_datetime_usec" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "metric_type", + "type": "text" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "value", + "type": "float" + }, + { + "allow_nil?": true, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "unit", + "type": "text" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "source", + "type": "text" + }, + { + "allow_nil?": true, + "default": "%{}", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "metadata", + "type": "map" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "inserted_at", + "type": "utc_datetime_usec" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "updated_at", + "type": "utc_datetime_usec" + }, + { + "allow_nil?": true, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": { + "deferrable": false, + "destination_attribute": "id", + "destination_attribute_default": null, + "destination_attribute_generated": null, + "index?": false, + "match_type": null, + "match_with": null, + "multitenancy": { + "attribute": null, + "global": null, + "strategy": null + }, + "name": "observations_location_id_fkey", + "on_delete": "delete", + "on_update": null, + "primary_key?": true, + "schema": "public", + "table": "locations" + }, + "scale": null, + "size": null, + "source": "location_id", + "type": "uuid" + } + ], + "base_filter": null, + "check_constraints": [], + "create_table_options": null, + "custom_indexes": [], + "custom_statements": [], + "has_create_action": true, + "hash": "DCBC9E90E6D1089B3814C54D7E43C5CAAD2C9D621BB1BC9B0B8730F0ED11845D", + "identities": [], + "multitenancy": { + "attribute": null, + "global": null, + "strategy": null + }, + "repo": "Elixir.Chronodash.Repo", + "schema": null, + "table": "observations" +} \ No newline at end of file diff --git a/priv/resource_snapshots/repo/observations/20260228072419.json b/priv/resource_snapshots/repo/observations/20260228072419.json new file mode 100644 index 0000000..fd349cb --- /dev/null +++ b/priv/resource_snapshots/repo/observations/20260228072419.json @@ -0,0 +1,147 @@ +{ + "attributes": [ + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": true, + "references": { + "deferrable": false, + "destination_attribute": "id", + "destination_attribute_default": null, + "destination_attribute_generated": null, + "index?": false, + "match_type": null, + "match_with": null, + "multitenancy": { + "attribute": null, + "global": null, + "strategy": null + }, + "name": "observations_location_id_fkey", + "on_delete": "delete", + "on_update": null, + "primary_key?": true, + "schema": "public", + "table": "locations" + }, + "scale": null, + "size": null, + "source": "location_id", + "type": "uuid" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": true, + "references": null, + "scale": null, + "size": null, + "source": "metric_type", + "type": "text" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": true, + "references": null, + "scale": null, + "size": null, + "source": "timestamp", + "type": "utc_datetime_usec" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "value", + "type": "float" + }, + { + "allow_nil?": true, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "unit", + "type": "text" + }, + { + "allow_nil?": false, + "default": "nil", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "source", + "type": "text" + }, + { + "allow_nil?": true, + "default": "%{}", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "metadata", + "type": "map" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "inserted_at", + "type": "utc_datetime_usec" + }, + { + "allow_nil?": false, + "default": "fragment(\"(now() AT TIME ZONE 'utc')\")", + "generated?": false, + "precision": null, + "primary_key?": false, + "references": null, + "scale": null, + "size": null, + "source": "updated_at", + "type": "utc_datetime_usec" + } + ], + "base_filter": null, + "check_constraints": [], + "create_table_options": null, + "custom_indexes": [], + "custom_statements": [], + "has_create_action": true, + "hash": "0E228D9F1F7DBFF93F0EFF42DD4AF11EF0968D3B2741C852C6AC0C816F11DF53", + "identities": [], + "multitenancy": { + "attribute": null, + "global": null, + "strategy": null + }, + "repo": "Elixir.Chronodash.Repo", + "schema": null, + "table": "observations" +} \ No newline at end of file diff --git a/priv/specs/chronodash/schema.dbml b/priv/specs/chronodash/schema.dbml new file mode 100644 index 0000000..114786b --- /dev/null +++ b/priv/specs/chronodash/schema.dbml @@ -0,0 +1,53 @@ +// Use DBML to define your database schema +// https://www.dbml.org/ + +Project chronodash { + database_type: 'PostgreSQL + TimescaleDB' + Note: 'Chronodash: A multi-datasource metrics monitoring application' +} + +Table users { + id uuid [primary key] + name varchar [not null] + email varchar [not null, unique] + city varchar + alert_hours varchar + inserted_at timestamptz + updated_at timestamptz +} + +Table locations { + id uuid [primary key] + name varchar [not null, unique] + latitude float [not null] + longitude float [not null] + external_id varchar [note: 'ID for external providers like MeteoSIX locationIds'] + inserted_at timestamptz + updated_at timestamptz +} + +Table observations { + location_id uuid [ref: > locations.id] + metric_type varchar [note: 'temperature, wind_speed, sky_state, etc.'] + timestamp timestamptz [not null] + value float [not null] + unit varchar + source varchar [not null, note: 'meteosix, openweather, etc.'] + metadata jsonb [note: 'Extra provider-specific data'] + inserted_at timestamptz + updated_at timestamptz + + indexes { + (location_id, metric_type, timestamp) [pk] + } +} + +// Many-to-Many for Users following Locations +Table user_locations { + user_id uuid [ref: > users.id] + location_id uuid [ref: > locations.id] + + indexes { + (user_id, location_id) [pk] + } +} diff --git a/priv/specs/schema.dbml b/priv/specs/schema.dbml new file mode 100644 index 0000000..86352c0 --- /dev/null +++ b/priv/specs/schema.dbml @@ -0,0 +1,53 @@ +// Use DBML to define your database schema +// https://www.dbml.org/ + +Project chronodash { + database_type: 'PostgreSQL + TimescaleDB' + Note: 'Chronodash: A multi-datasource metrics monitoring application' +} + +Table users { + id uuid [primary key] + name varchar [not null] + email varchar [not null, unique] + city varchar + alert_hours varchar + inserted_at timestamp + updated_at timestamp +} + +Table locations { + id uuid [primary key] + name varchar [not null, unique] + latitude float [not null] + longitude float [not null] + external_id varchar [note: 'ID for external providers like MeteoSIX locationIds'] + inserted_at timestamp + updated_at timestamp +} + +Table observations { + location_id uuid [ref: > locations.id] + metric_type varchar [note: 'temperature, wind_speed, sky_state, etc.'] + timestamp timestamp [not null] + value float [not null] + unit varchar + source varchar [not null, note: 'meteosix, openweather, etc.'] + metadata jsonb [note: 'Extra provider-specific data'] + inserted_at timestamp + updated_at timestamp + + indexes { + (location_id, metric_type, timestamp) [pk] + } +} + +// Many-to-Many for Users following Locations +Table user_locations { + user_id uuid [ref: > users.id] + location_id uuid [ref: > locations.id] + + indexes { + (user_id, location_id) [pk] + } +} diff --git a/test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs b/test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs new file mode 100644 index 0000000..dd3f55c --- /dev/null +++ b/test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs @@ -0,0 +1,114 @@ +defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.ForecastTest do + use ExUnit.Case, async: true + alias Chronodash.Metrics.ObservationData + alias Chronodash.Models.DataSource.MeteoSIX.WRF.Forecast + + describe "to_observation_data/2" do + test "correctly parses and normalizes simple metric (temperature)" do + response = %{ + "features" => [ + %{ + "geometry" => %{"coordinates" => [-8.42, 43.37], "type" => "Point"}, + "properties" => %{ + "name" => "Test City", + "days" => [ + %{ + "variables" => [ + %{ + "name" => "temperature", + "units" => "degC", + "values" => [ + %{"timeInstant" => "2026-02-28T10:00:00+01", "value" => "15.5"} + ] + } + ] + } + ] + } + } + ] + } + + observations = Forecast.to_observation_data(response, :temperature) + assert is_list(observations) + [obs] = observations + + assert %ObservationData{} = obs + assert obs.location_name == "Test City" + assert obs.latitude == 43.37 + assert obs.metric_type == "temperature" + assert obs.value == 15.5 + assert obs.unit == "degC" + assert obs.timestamp == ~U[2026-02-28 09:00:00Z] + end + + test "correctly parses and splits composite metric (wind)" do + response = %{ + "features" => [ + %{ + "geometry" => %{"coordinates" => [-8.42, 43.37], "type" => "Point"}, + "properties" => %{ + "name" => "Port Area", + "days" => [ + %{ + "variables" => [ + %{ + "name" => "wind", + "moduleUnits" => "kmh", + "directionUnits" => "deg", + "values" => [ + %{ + "timeInstant" => "2026-02-28T12:00:00+01", + "moduleValue" => 20.0, + "directionValue" => 180.0 + } + ] + } + ] + } + ] + } + } + ] + } + + observations = Forecast.to_observation_data(response, :wind) + assert length(observations) == 2 + + module = Enum.find(observations, &(&1.metric_type == "wind_module")) + direction = Enum.find(observations, &(&1.metric_type == "wind_direction")) + + assert module.value == 20.0 + assert module.unit == "kmh" + assert direction.value == 180.0 + assert direction.unit == "deg" + end + + test "normalizes sky state strings to numeric values" do + response = %{ + "features" => [ + %{ + "geometry" => %{"coordinates" => [-8.42, 43.37], "type" => "Point"}, + "properties" => %{ + "days" => [ + %{ + "variables" => [ + %{ + "name" => "sky_state", + "values" => [ + %{"timeInstant" => "2026-02-28T10:00:00+01", "value" => "PARTLY_CLOUDY"} + ] + } + ] + } + ] + } + } + ] + } + + [obs] = Forecast.to_observation_data(response, :sky_state) + assert obs.value == 2.0 + end + end +end diff --git a/test/support/http_client_mock.ex b/test/support/http_client_mock.ex index 7a79e6d..b33f156 100644 --- a/test/support/http_client_mock.ex +++ b/test/support/http_client_mock.ex @@ -2,18 +2,18 @@ defmodule Chronodash.HttpClient.Mock do @behaviour Chronodash.HttpClient def get(_url, _headers, _opts) do - {:ok, %{status: 200, body: "ok"}} + {:ok, %{status: 200, body: %{"features" => []}}} end def post(_url, _body, _headers, _opts) do - {:ok, %{status: 201, body: "created"}} + {:ok, %{status: 201, body: %{"status" => "created"}}} end def request(method, _url, _headers, _body, _opts) do case method do - :get -> {:ok, %{status: 200, body: "ok"}} - :post -> {:ok, %{status: 201, body: "created"}} - _ -> {:ok, %{status: 200, body: "ok"}} + :get -> {:ok, %{status: 200, body: %{"features" => []}}} + :post -> {:ok, %{status: 201, body: %{"status" => "created"}}} + _ -> {:ok, %{status: 200, body: %{"features" => []}}} end end end From d70d76a7fc51fc5a9f7542f69945234db27d4e2a Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 12:16:51 +0100 Subject: [PATCH 2/5] docs: add a TODO comment --- lib/chronodash/polling/worker.ex | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/chronodash/polling/worker.ex b/lib/chronodash/polling/worker.ex index d701083..6814fd2 100644 --- a/lib/chronodash/polling/worker.ex +++ b/lib/chronodash/polling/worker.ex @@ -75,6 +75,8 @@ defmodule Chronodash.Polling.Worker do end) # 3. Bulk Upsert + # TODO: If you eventually poll thousands of locations, you could batch the Ash.bulk_create calls + # (e.g., save in groups of 500 using Enum.chunk_every/2) to avoid long-running transactions. case Ash.bulk_create(inputs, Chronodash.Metrics.Observation, :create, domain: Chronodash.Metrics ) do From eba7dfce2359b884d218a256624d99ebe44ba0e6 Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 12:33:48 +0100 Subject: [PATCH 3/5] refactor: add multiple metrics from one resource --- etc/grafana/dashboards/meteosix_wrf.json | 240 ++++-------------- lib/chronodash/datasource/meteosix/wrf.ex | 7 +- .../datasource/meteosix/wrf/forecast.ex | 160 +++++------- lib/chronodash/polling/worker.ex | 187 ++++++++------ .../datasource/meteosix/wrf/forecast_test.exs | 46 ++-- 5 files changed, 263 insertions(+), 377 deletions(-) diff --git a/etc/grafana/dashboards/meteosix_wrf.json b/etc/grafana/dashboards/meteosix_wrf.json index f2078ea..539ef15 100644 --- a/etc/grafana/dashboards/meteosix_wrf.json +++ b/etc/grafana/dashboards/meteosix_wrf.json @@ -32,48 +32,6 @@ "color": { "mode": "palette-classic" }, - "custom": { - "axisBorderShow": false, - "axisCenteredZero": false, - "axisColorMode": "text", - "axisLabel": "", - "axisPlacement": "auto", - "barAlignment": 0, - "drawStyle": "line", - "fillOpacity": 10, - "gradientMode": "none", - "hideFrom": { - "legend": false, - "tooltip": false, - "viz": false - }, - "insertNulls": false, - "lineInterpolation": "linear", - "lineWidth": 1, - "pointSize": 5, - "scaleDistribution": { - "type": "linear" - }, - "showPoints": "auto", - "spanNulls": false, - "stacking": { - "group": "A", - "mode": "none" - }, - "thresholdsStyle": { - "mode": "off" - } - }, - "mappings": [], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - } - ] - }, "unit": "celsius" }, "overrides": [] @@ -87,7 +45,6 @@ "id": 1, "options": { "legend": { - "calcs": [], "displayMode": "list", "placement": "bottom", "showLegend": true @@ -104,23 +61,8 @@ "uid": "PostgreSQL" }, "format": "time_series", - "group": [], - "metricColumn": "none", - "rawQuery": true, -"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Temperature\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'temperature'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", - "refId": "A", - "select": [ - [ - { - "params": [ - "value" - ], - "type": "column" - } - ] - ], - "timeColumn": "timestamp", - "where": [] + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Temperature\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'temperature'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" } ], "title": "Temperature Forecast", @@ -136,48 +78,6 @@ "color": { "mode": "palette-classic" }, - "custom": { - "axisBorderShow": false, - "axisCenteredZero": false, - "axisColorMode": "text", - "axisLabel": "", - "axisPlacement": "auto", - "barAlignment": 0, - "drawStyle": "bars", - "fillOpacity": 80, - "gradientMode": "none", - "hideFrom": { - "legend": false, - "tooltip": false, - "viz": false - }, - "insertNulls": false, - "lineInterpolation": "linear", - "lineWidth": 1, - "pointSize": 5, - "scaleDistribution": { - "type": "linear" - }, - "showPoints": "auto", - "spanNulls": false, - "stacking": { - "group": "A", - "mode": "none" - }, - "thresholdsStyle": { - "mode": "off" - } - }, - "mappings": [], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - } - ] - }, "unit": "lengthmm" }, "overrides": [] @@ -191,7 +91,6 @@ "id": 2, "options": { "legend": { - "calcs": [], "displayMode": "list", "placement": "bottom", "showLegend": true @@ -208,8 +107,7 @@ "uid": "PostgreSQL" }, "format": "time_series", - "rawQuery": true, -"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n SUM(value) as \"Precipitation\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'precipitation_amount'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n SUM(value) as \"Precipitation\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'precipitation_amount'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", "refId": "A" } ], @@ -223,50 +121,18 @@ }, "fieldConfig": { "defaults": { - "color": { - "mode": "palette-classic" - }, "mappings": [ { "options": { - "1": { - "color": "yellow", - "index": 0, - "text": "SUNNY" - }, - "2": { - "color": "light-blue", - "index": 1, - "text": "PARTLY_CLOUDY" - }, - "3": { - "color": "blue", - "index": 2, - "text": "HIGH_CLOUDS" - }, - "4": { - "color": "grey", - "index": 3, - "text": "CLOUDY" - }, - "5": { - "color": "dark-grey", - "index": 4, - "text": "OVERCAST" - } + "1": { "color": "yellow", "text": "SUNNY" }, + "2": { "color": "light-blue", "text": "PARTLY_CLOUDY" }, + "3": { "color": "blue", "text": "HIGH_CLOUDS" }, + "4": { "color": "grey", "text": "CLOUDY" }, + "5": { "color": "dark-grey", "text": "OVERCAST" } }, "type": "value" } - ], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - } - ] - } + ] }, "overrides": [] }, @@ -277,21 +143,6 @@ "y": 8 }, "id": 3, - "options": { - "alignValue": "left", - "legend": { - "displayMode": "list", - "placement": "bottom", - "showLegend": true - }, - "mergeValues": true, - "rowHeight": 0.9, - "showValue": "always", - "tooltip": { - "mode": "single", - "sort": "none" - } - }, "targets": [ { "datasource": { @@ -299,8 +150,7 @@ "uid": "PostgreSQL" }, "format": "time_series", - "rawQuery": true, -"rawSql": "SELECT\n timestamp as time,\n value as \"Sky State\",\n locations.name as location\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'sky_state'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nORDER BY 1", + "rawSql": "SELECT\n timestamp as time,\n value as \"Sky State\",\n locations.name as location\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'sky_state'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nORDER BY 1", "refId": "A" } ], @@ -314,9 +164,6 @@ }, "fieldConfig": { "defaults": { - "color": { - "mode": "palette-classic" - }, "unit": "percent" }, "overrides": [] @@ -335,8 +182,7 @@ "uid": "PostgreSQL" }, "format": "time_series", - "rawQuery": true, -"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Humidity\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'relative_humidity'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Humidity\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'relative_humidity'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", "refId": "A" } ], @@ -350,9 +196,6 @@ }, "fieldConfig": { "defaults": { - "color": { - "mode": "palette-classic" - }, "unit": "percent" }, "overrides": [] @@ -371,8 +214,7 @@ "uid": "PostgreSQL" }, "format": "time_series", - "rawQuery": true, -"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Clouds\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'cloud_area_fraction'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Clouds\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'cloud_area_fraction'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", "refId": "A" } ], @@ -386,9 +228,6 @@ }, "fieldConfig": { "defaults": { - "color": { - "mode": "palette-classic" - }, "unit": "velocitykmh" }, "overrides": [] @@ -407,8 +246,7 @@ "uid": "PostgreSQL" }, "format": "time_series", - "rawQuery": true, -"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Wind Speed\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'wind'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Wind Speed\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'wind_module'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", "refId": "A" } ], @@ -422,9 +260,6 @@ }, "fieldConfig": { "defaults": { - "color": { - "mode": "palette-classic" - }, "unit": "pressurehpa" }, "overrides": [] @@ -443,8 +278,7 @@ "uid": "PostgreSQL" }, "format": "time_series", - "rawQuery": true, -"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Pressure\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'air_pressure_at_sea_level'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Pressure\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'air_pressure_at_sea_level'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", "refId": "A" } ], @@ -458,9 +292,6 @@ }, "fieldConfig": { "defaults": { - "color": { - "mode": "palette-classic" - }, "unit": "lengthm" }, "overrides": [] @@ -479,13 +310,44 @@ "uid": "PostgreSQL" }, "format": "time_series", - "rawQuery": true, -"rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Snow Level\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'snow_level'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Snow Level\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'snow_level'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", "refId": "A" } ], "title": "Snow Level", "type": "timeseries" + }, + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "fieldConfig": { + "defaults": { + "unit": "degree" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 28 + }, + "id": 9, + "targets": [ + { + "datasource": { + "type": "postgres", + "uid": "PostgreSQL" + }, + "format": "time_series", + "rawSql": "SELECT\n $__timeGroupAlias(timestamp, 1h),\n AVG(value) as \"Wind Direction\"\nFROM\n observations\nJOIN\n locations ON observations.location_id = locations.id\nWHERE\n metric_type = 'wind_direction'\n AND locations.name IN ($location)\n AND $__timeFilter(timestamp)\nGROUP BY 1\nORDER BY 1", + "refId": "A" + } + ], + "title": "Wind Direction", + "type": "timeseries" } ], "schemaVersion": 38, @@ -497,7 +359,11 @@ "templating": { "list": [ { - "current": {}, + "current": { + "selected": true, + "text": "All", + "value": "$__all" + }, "datasource": { "type": "postgres", "uid": "PostgreSQL" @@ -525,5 +391,5 @@ "timezone": "", "title": "MeteoSIX WRF Forecast", "uid": "meteosix_wrf", - "version": 1 + "version": 6 } diff --git a/lib/chronodash/datasource/meteosix/wrf.ex b/lib/chronodash/datasource/meteosix/wrf.ex index 656521a..11706f0 100644 --- a/lib/chronodash/datasource/meteosix/wrf.ex +++ b/lib/chronodash/datasource/meteosix/wrf.ex @@ -6,12 +6,15 @@ defmodule Chronodash.DataSource.MeteoSIX.WRF do alias MeteoSIX.WRF @doc """ - Fetches forecast from MeteoSIX and returns a list of standardized ObservationData structs. + Fetches forecast from MeteoSIX and parses it into a DTO. """ def get_forecast(id_or_coords, var_atom, opts \\ []) do case WRF.get_forecast(id_or_coords, var_atom, opts) do {:ok, response} -> - {:ok, Forecast.to_observation_data(response, var_atom)} + case Forecast.new(response, var_atom) do + %Forecast{} = forecast -> {:ok, forecast} + {:error, _} = error -> error + end {:error, reason} -> {:error, reason} diff --git a/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex b/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex index 92d2b93..ffd21b7 100644 --- a/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex +++ b/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex @@ -1,127 +1,99 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.Forecast do @moduledoc """ - Parser for MeteoSIX WRF Forecast data. - Converts raw MeteoSIX JSON into standardized ObservationData structs. + DTO for MeteoSIX WRF Forecast data. + Handles both simple and composite metrics (like wind). """ - alias Chronodash.Metrics.ObservationData + defstruct [ + :location_name, + :coords, + :metric_type, + # List of %{timestamp: DateTime.t(), metrics: [%{name: String.t(), value: any(), unit: String.t()}]} + :points + ] + + @type metric_entry :: %{name: String.t(), value: any(), unit: String.t() | nil} + @type point :: %{timestamp: DateTime.t(), metrics: list(metric_entry())} + + @type t :: %__MODULE__{ + location_name: String.t(), + coords: {float(), float()}, + metric_type: atom(), + points: list(point()) + } @doc """ - Parses a MeteoSIX response into a list of ObservationData. + Parses a MeteoSIX response into a Forecast DTO. """ - def to_observation_data(response, metric_type) do + def new(response, metric_type) do case response["features"] do [feature | _] -> coords = extract_coords(feature["geometry"]) name = feature["properties"]["name"] || default_name(coords) - days = feature["properties"]["days"] || [] - parse_days(days, name, coords, metric_type) + %__MODULE__{ + location_name: name, + coords: coords, + metric_type: metric_type, + points: parse_points(feature["properties"]["days"], metric_type) + } _ -> - [] + {:error, :no_data} end end - defp parse_days(days, location_name, coords, metric_type) do - Enum.flat_map(days, &process_day(&1, location_name, coords, metric_type)) - end + defp default_name({lat, lon}), do: "#{lat}, #{lon}" + defp default_name(_), do: "Unknown Location" - defp process_day(day, location_name, coords, metric_type) do - case Enum.find(day["variables"], fn v -> v["name"] == to_string(metric_type) end) do - nil -> [] - variable -> process_variable(variable, location_name, coords, metric_type) - end - end + defp extract_coords(%{"coordinates" => [lon, lat]}), do: {lat, lon} + defp extract_coords(_), do: {0.0, 0.0} - defp process_variable(variable, location_name, coords, metric_type) do - Enum.flat_map(variable["values"], fn val -> - timestamp = parse_timestamp(val["timeInstant"]) - extract_metrics(val, variable, metric_type, location_name, coords, timestamp) + defp parse_points(days, metric_type) when is_list(days) do + Enum.flat_map(days, fn day -> + variable = Enum.find(day["variables"], fn v -> v["name"] == to_string(metric_type) end) + + if variable do + Enum.map(variable["values"], fn val -> + %{ + timestamp: parse_timestamp(val["timeInstant"]), + metrics: extract_metrics(val, variable, metric_type) + } + end) + else + [] + end end) end - # Special handling for composite variables (Wind) - defp extract_metrics(val, variable, :wind, name, coords, ts) do + defp parse_points(_, _), do: [] + + # Special handling for composite variables + defp extract_metrics(val, variable, :wind) do [ - build_base( - name, - coords, - "wind_module", - val["moduleValue"], - variable["moduleUnits"], - ts, - val - ), - build_base( - name, - coords, - "wind_direction", - val["directionValue"], - variable["directionUnits"], - ts, - val - ) + %{ + name: "wind_module", + value: val["moduleValue"], + unit: variable["moduleUnits"] + }, + %{ + name: "wind_direction", + value: val["directionValue"], + unit: variable["directionUnits"] + } ] end # Default handling for simple variables - defp extract_metrics(val, variable, metric_type, name, coords, ts) do + defp extract_metrics(val, variable, metric_type) do [ - build_base(name, coords, to_string(metric_type), val["value"], variable["units"], ts, val) + %{ + name: to_string(metric_type), + value: val["value"], + unit: variable["units"] + } ] end - defp build_base(name, {lat, lon}, type, val, unit, ts, raw) do - %ObservationData{ - location_name: name, - latitude: lat, - longitude: lon, - metric_type: type, - value: normalize_value(val), - unit: unit, - timestamp: ts, - source: "meteosix", - metadata: %{original_value: val, raw_payload: raw} - } - end - - defp normalize_value(v) when is_number(v), do: v * 1.0 - - defp normalize_value(v) when is_binary(v) do - cond do - Regex.match?(~r/^-?\d+(\.\d+)?$/, v) -> - {f, _} = Float.parse(v) - f - - # Sky States - v == "SUNNY" -> - 1.0 - - v == "PARTLY_CLOUDY" -> - 2.0 - - v == "HIGH_CLOUDS" -> - 3.0 - - v == "CLOUDY" -> - 4.0 - - v == "OVERCAST" -> - 5.0 - - true -> - 0.0 - end - end - - defp normalize_value(_), do: 0.0 - - defp extract_coords(%{"coordinates" => [lon, lat]}), do: {lat, lon} - defp extract_coords(_), do: {0.0, 0.0} - - defp default_name({lat, lon}), do: "#{lat}, #{lon}" - defp default_name(_), do: "Unknown Location" - defp parse_timestamp(nil), do: nil defp parse_timestamp(time_str) do diff --git a/lib/chronodash/polling/worker.ex b/lib/chronodash/polling/worker.ex index 6814fd2..a29a64c 100644 --- a/lib/chronodash/polling/worker.ex +++ b/lib/chronodash/polling/worker.ex @@ -1,7 +1,6 @@ defmodule Chronodash.Polling.Worker do @moduledoc """ - A generic worker that polls a datasource and saves results in bulk. - Decoupled from specific datasource logic via standardize ObservationData. + A generic worker that polls a datasource and saves results. """ use GenServer require Logger @@ -35,105 +34,147 @@ defmodule Chronodash.Polling.Worker do {mod, fun, args} = state.mfa case apply(mod, fun, args) do - {:ok, observations} when is_list(observations) -> - handle_observations(observations, state) + {:ok, forecast} -> + handle_success(forecast, state) {:error, reason} -> Logger.error("Polling job #{state.id} failed: #{inspect(reason)}") - emit_run_telemetry(state, :error) + emit_telemetry(state, :error, %{reason: inspect(reason)}) end schedule_next(state.rate) {:noreply, state} end - defp handle_observations([], state) do - Logger.debug("Polling job #{state.id} returned no data.") - emit_run_telemetry(state, :success) + defp handle_success(forecast, state) do + # 1. Find or create the location in our DB + location = get_or_create_location(forecast) + + # 2. Save all forecast points to the DB + # TODO: If you eventually poll thousands of locations, you could batch the Ash.bulk_create calls + # (e.g., save in groups of 500 using Enum.chunk_every/2) to avoid long-running transactions. + if Map.has_key?(forecast, :points) do + Enum.each(forecast.points, fn point -> + Enum.each(point.metrics, fn metric -> + save_observation(metric, point.timestamp, location, state) + emit_observation(metric, point.timestamp, forecast.location_name, state) + end) + end) + end + + emit_telemetry(state, :success, %{}) end - defp handle_observations(observations, state) do - # 1. Ensure location exists (all observations in a poll usually share a location) - first = List.first(observations) - - case get_or_create_location(first) do - {:ok, location} -> - # 2. Enrich observations with location_id and convert to maps for bulk create - inputs = - Enum.map(observations, fn obs -> - emit_observation_telemetry(obs, state) - - %{ - location_id: location.id, - metric_type: obs.metric_type, - value: obs.value, - unit: obs.unit, - timestamp: obs.timestamp, - source: obs.source, - metadata: obs.metadata - } - end) - - # 3. Bulk Upsert - # TODO: If you eventually poll thousands of locations, you could batch the Ash.bulk_create calls - # (e.g., save in groups of 500 using Enum.chunk_every/2) to avoid long-running transactions. - case Ash.bulk_create(inputs, Chronodash.Metrics.Observation, :create, - domain: Chronodash.Metrics - ) do - %Ash.BulkResult{status: :success} -> - Logger.info( - "Polling job #{state.id}: successfully saved #{length(inputs)} observations." - ) - - emit_run_telemetry(state, :success) - - %Ash.BulkResult{errors: errors} -> - Logger.error( - "Polling job #{state.id}: failed to bulk save observations: #{inspect(errors)}" - ) - - emit_run_telemetry(state, :error) - end - - {:error, error} -> - Logger.error("Polling job #{state.id}: could not resolve location: #{inspect(error)}") - emit_run_telemetry(state, :error) + defp get_or_create_location(forecast) do + if is_nil(forecast.location_name) or forecast.location_name == "" do + Logger.error("Cannot get/create location: location_name is missing in forecast DTO.") + nil + else + # 1. Use name + coordinates to ensure uniqueness + {lat, lon} = forecast.coords || {0.0, 0.0} + + params = %{ + name: forecast.location_name, + latitude: lat, + longitude: lon + } + + # 2. Use create action with upsert? true + case Chronodash.Metrics.Location + |> Ash.Changeset.for_create(:create, params) + |> Ash.create(domain: Chronodash.Metrics) do + {:ok, loc} -> + loc + + {:error, error} -> + Logger.error( + "Failed to get/create location '#{forecast.location_name}': #{inspect(error)}" + ) + + nil + end end end - defp get_or_create_location(obs) do - params = %{ - name: obs.location_name, - latitude: obs.latitude, - longitude: obs.longitude - } - - case Chronodash.Metrics.Location - |> Ash.Changeset.for_create(:create, params) - |> Ash.create(domain: Chronodash.Metrics) do - {:ok, loc} -> {:ok, loc} - {:error, error} -> {:error, error} + defp save_observation(metric, timestamp, location, state) do + if is_nil(location) do + Logger.error("Cannot save observation for #{metric.name} because location is nil.") + else + source = Map.get(state.metadata, :source, "unknown") + + params = %{ + location_id: location.id, + metric_type: metric.name, + value: normalize_value(metric.value), + unit: metric.unit, + timestamp: timestamp, + source: source, + metadata: %{original_value: metric.value} + } + + case Chronodash.Metrics.Observation + |> Ash.Changeset.for_create(:create, params) + |> Ash.create(domain: Chronodash.Metrics) do + {:ok, _obs} -> + :ok + + {:error, error} -> + Logger.error("Failed to upsert observation #{metric.name}: #{inspect(error)}") + end end end - defp emit_observation_telemetry(obs, state) do - measurements = %{value: obs.value} + defp emit_observation(metric, timestamp, location_name, state) do + # We emit a metric for each data point + measurements = %{value: normalize_value(metric.value)} metadata = Map.merge(state.metadata, %{ - location: obs.location_name, - variable: obs.metric_type, - timestamp: obs.timestamp + location: location_name, + variable: metric.name, + timestamp: timestamp }) :telemetry.execute([:chronodash, :polling, :observation], measurements, metadata) end - defp emit_run_telemetry(state, status) do - metadata = Map.merge(state.metadata, %{status: status, job_id: state.id}) + defp emit_telemetry(state, status, extra_metadata) do + metadata = Map.merge(state.metadata, extra_metadata) |> Map.put(:status, status) :telemetry.execute([:chronodash, :polling, :run], %{count: 1}, metadata) end + defp normalize_value(v) when is_number(v), do: v * 1.0 + + defp normalize_value(v) when is_binary(v) do + cond do + # Handle numeric strings + Regex.match?(~r/^-?\d+(\.\d+)?$/, v) -> + {f, _} = Float.parse(v) + f + + # Sky States + v == "SUNNY" -> + 1.0 + + v == "PARTLY_CLOUDY" -> + 2.0 + + v == "HIGH_CLOUDS" -> + 3.0 + + v == "CLOUDY" -> + 4.0 + + v == "OVERCAST" -> + 5.0 + + true -> + 0.0 + end + end + + defp normalize_value(_), do: 0.0 + defp schedule_next(rate) do Process.send_after(self(), :poll, rate) end diff --git a/test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs b/test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs index dd3f55c..2908e41 100644 --- a/test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs +++ b/test/chronodash/models/datasource/meteosix/wrf/forecast_test.exs @@ -1,10 +1,9 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.ForecastTest do use ExUnit.Case, async: true - alias Chronodash.Metrics.ObservationData alias Chronodash.Models.DataSource.MeteoSIX.WRF.Forecast - describe "to_observation_data/2" do - test "correctly parses and normalizes simple metric (temperature)" do + describe "new/2" do + test "correctly parses simple metric (temperature)" do response = %{ "features" => [ %{ @@ -29,20 +28,20 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.ForecastTest do ] } - observations = Forecast.to_observation_data(response, :temperature) - assert is_list(observations) - [obs] = observations + assert %Forecast{} = forecast = Forecast.new(response, :temperature) + assert forecast.location_name == "Test City" + assert forecast.coords == {43.37, -8.42} + assert forecast.metric_type == :temperature - assert %ObservationData{} = obs - assert obs.location_name == "Test City" - assert obs.latitude == 43.37 - assert obs.metric_type == "temperature" - assert obs.value == 15.5 - assert obs.unit == "degC" - assert obs.timestamp == ~U[2026-02-28 09:00:00Z] + [point] = forecast.points + assert point.timestamp == ~U[2026-02-28 09:00:00Z] + [metric] = point.metrics + assert metric.name == "temperature" + assert metric.value == "15.5" + assert metric.unit == "degC" end - test "correctly parses and splits composite metric (wind)" do + test "correctly parses composite metric (wind)" do response = %{ "features" => [ %{ @@ -72,11 +71,14 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.ForecastTest do ] } - observations = Forecast.to_observation_data(response, :wind) - assert length(observations) == 2 + assert %Forecast{} = forecast = Forecast.new(response, :wind) + assert forecast.metric_type == :wind - module = Enum.find(observations, &(&1.metric_type == "wind_module")) - direction = Enum.find(observations, &(&1.metric_type == "wind_direction")) + [point] = forecast.points + assert length(point.metrics) == 2 + + module = Enum.find(point.metrics, &(&1.name == "wind_module")) + direction = Enum.find(point.metrics, &(&1.name == "wind_direction")) assert module.value == 20.0 assert module.unit == "kmh" @@ -84,7 +86,7 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.ForecastTest do assert direction.unit == "deg" end - test "normalizes sky state strings to numeric values" do + test "returns raw sky state strings" do response = %{ "features" => [ %{ @@ -107,8 +109,10 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.ForecastTest do ] } - [obs] = Forecast.to_observation_data(response, :sky_state) - assert obs.value == 2.0 + forecast = Forecast.new(response, :sky_state) + [point] = forecast.points + [metric] = point.metrics + assert metric.value == "PARTLY_CLOUDY" end end end From f2955fc00e3a48b8dd22e71c2b3b65ba4cb9adc1 Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 12:40:35 +0100 Subject: [PATCH 4/5] fix: use correct dbml --- priv/specs/chronodash/schema.dbml | 14 ++++---- priv/specs/schema.dbml | 53 ------------------------------- 2 files changed, 7 insertions(+), 60 deletions(-) delete mode 100644 priv/specs/schema.dbml diff --git a/priv/specs/chronodash/schema.dbml b/priv/specs/chronodash/schema.dbml index 114786b..86352c0 100644 --- a/priv/specs/chronodash/schema.dbml +++ b/priv/specs/chronodash/schema.dbml @@ -12,8 +12,8 @@ Table users { email varchar [not null, unique] city varchar alert_hours varchar - inserted_at timestamptz - updated_at timestamptz + inserted_at timestamp + updated_at timestamp } Table locations { @@ -22,20 +22,20 @@ Table locations { latitude float [not null] longitude float [not null] external_id varchar [note: 'ID for external providers like MeteoSIX locationIds'] - inserted_at timestamptz - updated_at timestamptz + inserted_at timestamp + updated_at timestamp } Table observations { location_id uuid [ref: > locations.id] metric_type varchar [note: 'temperature, wind_speed, sky_state, etc.'] - timestamp timestamptz [not null] + timestamp timestamp [not null] value float [not null] unit varchar source varchar [not null, note: 'meteosix, openweather, etc.'] metadata jsonb [note: 'Extra provider-specific data'] - inserted_at timestamptz - updated_at timestamptz + inserted_at timestamp + updated_at timestamp indexes { (location_id, metric_type, timestamp) [pk] diff --git a/priv/specs/schema.dbml b/priv/specs/schema.dbml deleted file mode 100644 index 86352c0..0000000 --- a/priv/specs/schema.dbml +++ /dev/null @@ -1,53 +0,0 @@ -// Use DBML to define your database schema -// https://www.dbml.org/ - -Project chronodash { - database_type: 'PostgreSQL + TimescaleDB' - Note: 'Chronodash: A multi-datasource metrics monitoring application' -} - -Table users { - id uuid [primary key] - name varchar [not null] - email varchar [not null, unique] - city varchar - alert_hours varchar - inserted_at timestamp - updated_at timestamp -} - -Table locations { - id uuid [primary key] - name varchar [not null, unique] - latitude float [not null] - longitude float [not null] - external_id varchar [note: 'ID for external providers like MeteoSIX locationIds'] - inserted_at timestamp - updated_at timestamp -} - -Table observations { - location_id uuid [ref: > locations.id] - metric_type varchar [note: 'temperature, wind_speed, sky_state, etc.'] - timestamp timestamp [not null] - value float [not null] - unit varchar - source varchar [not null, note: 'meteosix, openweather, etc.'] - metadata jsonb [note: 'Extra provider-specific data'] - inserted_at timestamp - updated_at timestamp - - indexes { - (location_id, metric_type, timestamp) [pk] - } -} - -// Many-to-Many for Users following Locations -Table user_locations { - user_id uuid [ref: > users.id] - location_id uuid [ref: > locations.id] - - indexes { - (user_id, location_id) [pk] - } -} From 117370b6d32f0bc6b843310fd6fc0caf63e9b5a1 Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 12:58:11 +0100 Subject: [PATCH 5/5] refactor: use more auxiliar funs --- .../datasource/meteosix/wrf/forecast.ex | 31 ++++++++++--------- lib/chronodash/polling/worker.ex | 25 ++++++++++----- 2 files changed, 34 insertions(+), 22 deletions(-) diff --git a/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex b/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex index ffd21b7..5e158c4 100644 --- a/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex +++ b/lib/chronodash/models/datasource/meteosix/wrf/forecast.ex @@ -49,24 +49,27 @@ defmodule Chronodash.Models.DataSource.MeteoSIX.WRF.Forecast do defp extract_coords(_), do: {0.0, 0.0} defp parse_points(days, metric_type) when is_list(days) do - Enum.flat_map(days, fn day -> - variable = Enum.find(day["variables"], fn v -> v["name"] == to_string(metric_type) end) - - if variable do - Enum.map(variable["values"], fn val -> - %{ - timestamp: parse_timestamp(val["timeInstant"]), - metrics: extract_metrics(val, variable, metric_type) - } - end) - else - [] - end - end) + Enum.flat_map(days, &process_forecast_day(&1, metric_type)) end defp parse_points(_, _), do: [] + defp process_forecast_day(day, metric_type) do + case Enum.find(day["variables"], fn v -> v["name"] == to_string(metric_type) end) do + nil -> [] + variable -> map_values_to_points(variable, metric_type) + end + end + + defp map_values_to_points(variable, metric_type) do + Enum.map(variable["values"], fn val -> + %{ + timestamp: parse_timestamp(val["timeInstant"]), + metrics: extract_metrics(val, variable, metric_type) + } + end) + end + # Special handling for composite variables defp extract_metrics(val, variable, :wind) do [ diff --git a/lib/chronodash/polling/worker.ex b/lib/chronodash/polling/worker.ex index a29a64c..a47cbbc 100644 --- a/lib/chronodash/polling/worker.ex +++ b/lib/chronodash/polling/worker.ex @@ -53,18 +53,27 @@ defmodule Chronodash.Polling.Worker do # 2. Save all forecast points to the DB # TODO: If you eventually poll thousands of locations, you could batch the Ash.bulk_create calls # (e.g., save in groups of 500 using Enum.chunk_every/2) to avoid long-running transactions. - if Map.has_key?(forecast, :points) do - Enum.each(forecast.points, fn point -> - Enum.each(point.metrics, fn metric -> - save_observation(metric, point.timestamp, location, state) - emit_observation(metric, point.timestamp, forecast.location_name, state) - end) - end) - end + process_forecast_points(forecast, location, state) emit_telemetry(state, :success, %{}) end + defp process_forecast_points(%{points: points} = forecast, location, state) + when is_list(points) do + Enum.each(points, fn point -> + process_metrics(point.metrics, point.timestamp, location, forecast.location_name, state) + end) + end + + defp process_forecast_points(_forecast, _location, _state), do: :ok + + defp process_metrics(metrics, timestamp, location, location_name, state) do + Enum.each(metrics, fn metric -> + save_observation(metric, timestamp, location, state) + emit_observation(metric, timestamp, location_name, state) + end) + end + defp get_or_create_location(forecast) do if is_nil(forecast.location_name) or forecast.location_name == "" do Logger.error("Cannot get/create location: location_name is missing in forecast DTO.")