From 29fef2afed79155a5cd0f446270b293592ee9764 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 7 Apr 2026 12:04:10 -0500 Subject: [PATCH] Add ERA5 reanalysis and RTMA data sources ERA5 (Copernicus CDS API): - Era5Profile schema matching HRRR profile structure for interop - Era5Client with async job submission, polling, GRIB2 download - Era5FetchWorker (Oban queue: era5, max_attempts: 5) - Unified lookup: Weather.best_profile_for_contact/1 tries HRRR first, falls back to ERA5 for pre-2014 contacts RTMA (NOAA S3, 2.5km/15-min): - RtmaObservation schema for surface-only fields - RtmaClient with byte-range GRIB2 requests (same pattern as HRRR) - RtmaFetchWorker (Oban queue: rtma, max_attempts: 10) - Weather.find_nearest_rtma/3 for surface condition lookup Both queues added to production Oban config (2 workers each). --- config/runtime.exs | 13 +- lib/microwaveprop/weather.ex | 98 +++++++ lib/microwaveprop/weather/era5_client.ex | 263 ++++++++++++++++++ lib/microwaveprop/weather/era5_profile.ex | 37 +++ lib/microwaveprop/weather/rtma_client.ex | 160 +++++++++++ lib/microwaveprop/weather/rtma_observation.ex | 34 +++ .../workers/era5_fetch_worker.ex | 69 +++++ .../workers/rtma_fetch_worker.ex | 68 +++++ ...dd_era5_profiles_and_rtma_observations.exs | 46 +++ 9 files changed, 787 insertions(+), 1 deletion(-) create mode 100644 lib/microwaveprop/weather/era5_client.ex create mode 100644 lib/microwaveprop/weather/era5_profile.ex create mode 100644 lib/microwaveprop/weather/rtma_client.ex create mode 100644 lib/microwaveprop/weather/rtma_observation.ex create mode 100644 lib/microwaveprop/workers/era5_fetch_worker.ex create mode 100644 lib/microwaveprop/workers/rtma_fetch_worker.ex create mode 100644 priv/repo/migrations/20260407165919_add_era5_profiles_and_rtma_observations.exs diff --git a/config/runtime.exs b/config/runtime.exs index 828dbc48..fde1c30f 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -132,7 +132,18 @@ if config_env() == :prod do # Production Oban: live scoring, polling, and on-demand QSO enrichment config :microwaveprop, Oban, # Per-pod concurrency (×3 pods = effective cluster total) - queues: [propagation: 1, commercial: 1, solar: 1, weather: 3, hrrr: 5, terrain: 3, iemre: 3, backfill_enqueue: 1], + queues: [ + propagation: 1, + commercial: 1, + solar: 1, + weather: 3, + hrrr: 5, + terrain: 3, + iemre: 3, + era5: 2, + rtma: 2, + backfill_enqueue: 1 + ], plugins: [ {Oban.Plugins.Pruner, max_age: 3600 * 24}, {Oban.Plugins.Lifeline, rescue_after: 10 * 60 * 1000}, diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 19d2f8ce..bb11d778 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -4,9 +4,11 @@ defmodule Microwaveprop.Weather do import Ecto.Query alias Microwaveprop.Repo + alias Microwaveprop.Weather.Era5Profile alias Microwaveprop.Weather.HrrrProfile alias Microwaveprop.Weather.IemClient alias Microwaveprop.Weather.IemreObservation + alias Microwaveprop.Weather.RtmaObservation alias Microwaveprop.Weather.SolarIndex alias Microwaveprop.Weather.Sounding alias Microwaveprop.Weather.Station @@ -519,6 +521,102 @@ defmodule Microwaveprop.Weather do |> Enum.reject(&is_nil/1) end + @doc """ + Find the best available atmospheric profile for a contact. + Tries HRRR first (3 km, hourly), falls back to ERA5 (0.25°, hourly). + Returns the profile struct or nil. + """ + def best_profile_for_contact(contact) do + hrrr_for_contact(contact) || era5_for_contact(contact) + end + + @doc "Find all atmospheric profiles along a contact's path, from any source." + def profiles_along_path(contact) do + hrrr_path = hrrr_profiles_for_path(contact) + + if hrrr_path == [] do + era5_profiles_for_path(contact) + else + hrrr_path + end + end + + def era5_for_contact(%{pos1: nil}), do: nil + + def era5_for_contact(contact) do + lat = contact.pos1["lat"] + lon = contact.pos1["lon"] || contact.pos1["lng"] + + if lat && lon do + find_nearest_era5(lat, lon, contact.qso_timestamp) + end + end + + def era5_profiles_for_path(%{pos1: nil}), do: [] + + def era5_profiles_for_path(contact) do + contact + |> Microwaveprop.Radio.contact_path_points() + |> Enum.map(fn {lat, lon} -> find_nearest_era5(lat, lon, contact.qso_timestamp) end) + |> Enum.reject(&is_nil/1) + end + + def find_nearest_era5(lat, lon, timestamp) do + dlat = 0.15 + dlon = 0.15 + time_start = DateTime.add(timestamp, -1800, :second) + time_end = DateTime.add(timestamp, 1800, :second) + + Era5Profile + |> where( + [p], + p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and + p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and + p.valid_time >= ^time_start and p.valid_time <= ^time_end + ) + |> order_by([p], + asc: + fragment( + "ABS(? - ?) + ABS(? - ?)", + p.lat, + ^lat, + p.lon, + ^lon + ) + ) + |> limit(1) + |> Repo.one() + end + + def find_nearest_rtma(lat, lon, timestamp) do + dlat = 0.05 + dlon = 0.05 + time_start = DateTime.add(timestamp, -900, :second) + time_end = DateTime.add(timestamp, 900, :second) + + RtmaObservation + |> where( + [o], + o.lat >= ^(lat - dlat) and o.lat <= ^(lat + dlat) and + o.lon >= ^(lon - dlon) and o.lon <= ^(lon + dlon) and + o.valid_time >= ^time_start and o.valid_time <= ^time_end + ) + |> order_by([o], + asc: + fragment( + "ABS(? - ?) + ABS(? - ?) + ABS(EXTRACT(EPOCH FROM ? - ?))", + o.lat, + ^lat, + o.lon, + ^lon, + o.valid_time, + ^timestamp + ) + ) + |> limit(1) + |> Repo.one() + end + def round_to_hrrr_grid(lat, lon) do {Float.round(lat / 1.0, 2), Float.round(lon / 1.0, 2)} end diff --git a/lib/microwaveprop/weather/era5_client.ex b/lib/microwaveprop/weather/era5_client.ex new file mode 100644 index 00000000..1ccf9ae3 --- /dev/null +++ b/lib/microwaveprop/weather/era5_client.ex @@ -0,0 +1,263 @@ +defmodule Microwaveprop.Weather.Era5Client do + @moduledoc """ + Client for the Copernicus Climate Data Store (CDS) API. + Fetches ERA5 reanalysis data (pressure-level profiles + single-level surface fields) + for historical contact enrichment where HRRR is unavailable (pre-2014). + + The CDS API is asynchronous: submit a request, poll for completion, download result. + Results are GRIB2 format, decoded with the existing GRIB2 infrastructure. + """ + + alias Microwaveprop.Weather.Grib2.Extractor + alias Microwaveprop.Weather.SoundingParams + + require Logger + + @cds_url "https://cds.climate.copernicus.eu/api" + @poll_interval_ms 5_000 + @max_poll_attempts 120 + + @pressure_levels ~w(1000 975 950 925 900 875 850 825 800 775 750 725 700) + + @single_level_vars ~w( + 2m_temperature + 2m_dewpoint_temperature + surface_pressure + 10m_u_component_of_wind + 10m_v_component_of_wind + total_column_water_vapour + boundary_layer_height + ) + + @pressure_level_vars ~w(temperature dewpoint_temperature geopotential) + + @doc """ + Fetch ERA5 profile for a single point and time. + Returns {:ok, profile_attrs} or {:error, reason}. + + The profile_attrs map has the same shape as HRRR profiles for interoperability. + """ + def fetch_profile(lat, lon, timestamp) do + # Round to ERA5 grid (0.25°) + rlat = Float.round(lat * 4) / 4 + rlon = Float.round(lon * 4) / 4 + + # ERA5 hourly — round to nearest hour + valid_time = DateTime.truncate(timestamp, :second) + valid_time = %{valid_time | minute: 0, second: 0} + + # Build a small bounding box (0.25° around the point) + area = [rlat + 0.25, rlon - 0.25, rlat - 0.25, rlon + 0.25] + + with {:ok, single_data} <- fetch_single_levels(valid_time, area), + {:ok, pressure_data} <- fetch_pressure_levels(valid_time, area) do + build_profile(rlat, rlon, valid_time, single_data, pressure_data) + end + end + + @doc """ + Fetch ERA5 single-level (surface) data for a time and area. + """ + def fetch_single_levels(valid_time, area) do + request = %{ + "product_type" => ["reanalysis"], + "variable" => @single_level_vars, + "year" => [Calendar.strftime(valid_time, "%Y")], + "month" => [Calendar.strftime(valid_time, "%m")], + "day" => [Calendar.strftime(valid_time, "%d")], + "time" => [Calendar.strftime(valid_time, "%H:00")], + "area" => area, + "data_format" => "grib" + } + + submit_and_download("reanalysis-era5-single-levels", request) + end + + @doc """ + Fetch ERA5 pressure-level data for a time and area. + """ + def fetch_pressure_levels(valid_time, area) do + request = %{ + "product_type" => ["reanalysis"], + "variable" => @pressure_level_vars, + "pressure_level" => @pressure_levels, + "year" => [Calendar.strftime(valid_time, "%Y")], + "month" => [Calendar.strftime(valid_time, "%m")], + "day" => [Calendar.strftime(valid_time, "%d")], + "time" => [Calendar.strftime(valid_time, "%H:00")], + "area" => area, + "data_format" => "grib" + } + + submit_and_download("reanalysis-era5-pressure-levels", request) + end + + defp submit_and_download(dataset, request) do + api_key = api_key() + + if is_nil(api_key) do + {:error, "ERA5_CDS_API_KEY not configured"} + else + case submit_request(dataset, request, api_key) do + {:ok, job_id} -> poll_and_download(job_id, api_key) + {:error, reason} -> {:error, reason} + end + end + end + + defp submit_request(dataset, request, api_key) do + url = "#{@cds_url}/retrieve/#{dataset}" + + case Req.post(url, + json: request, + headers: [{"PRIVATE-TOKEN", api_key}], + receive_timeout: 30_000 + ) do + {:ok, %{status: status, body: body}} when status in [200, 201, 202] -> + job_id = body["request_id"] || body["jobId"] + + if job_id do + Logger.info("ERA5: submitted job #{job_id} for #{dataset}") + {:ok, job_id} + else + {:error, "ERA5: no job_id in response: #{inspect(body)}"} + end + + {:ok, %{status: status, body: body}} -> + {:error, "ERA5 CDS HTTP #{status}: #{inspect(body)}"} + + {:error, reason} -> + {:error, "ERA5 CDS request failed: #{inspect(reason)}"} + end + end + + defp poll_and_download(job_id, api_key, attempt \\ 0) + + defp poll_and_download(_job_id, _api_key, attempt) when attempt >= @max_poll_attempts do + {:error, "ERA5: poll timeout after #{@max_poll_attempts} attempts"} + end + + defp poll_and_download(job_id, api_key, attempt) do + url = "#{@cds_url}/tasks/#{job_id}" + + case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000) do + {:ok, %{status: 200, body: %{"state" => "completed"} = body}} -> + download_url = body["location"] || get_in(body, ["result", "location"]) + + if download_url do + download_result(download_url, api_key) + else + {:error, "ERA5: completed but no download URL in #{inspect(body)}"} + end + + {:ok, %{status: 200, body: %{"state" => state}}} when state in ["queued", "running"] -> + Process.sleep(@poll_interval_ms) + poll_and_download(job_id, api_key, attempt + 1) + + {:ok, %{status: 200, body: %{"state" => "failed"} = body}} -> + {:error, "ERA5 job failed: #{inspect(body["error"])}"} + + {:ok, %{status: status, body: body}} -> + {:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"} + + {:error, reason} -> + {:error, "ERA5 poll failed: #{inspect(reason)}"} + end + end + + defp download_result(url, api_key) do + case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000, into: :self) do + {:ok, %{status: 200, body: body}} -> + {:ok, body} + + {:ok, %{status: status}} -> + {:error, "ERA5 download HTTP #{status}"} + + {:error, reason} -> + {:error, "ERA5 download failed: #{inspect(reason)}"} + end + end + + defp build_profile(lat, lon, valid_time, single_grib, pressure_grib) do + # Decode GRIB2 data using existing infrastructure + with {:ok, single_points} <- extract_point(single_grib, lat, lon), + {:ok, pressure_points} <- extract_point(pressure_grib, lat, lon) do + # ERA5 temperatures are in Kelvin, convert to Celsius + temp_c = kelvin_to_celsius(single_points["TMP:2 m above ground"]) + dewpoint_c = kelvin_to_celsius(single_points["DPT:2 m above ground"]) + # ERA5 surface pressure is in Pa, convert to mb + pressure_mb = pa_to_mb(single_points["PRES:surface"]) + hpbl_m = single_points["HPBL:surface"] + # ERA5 TCWV is in kg/m² which equals mm + pwat_mm = single_points["TCWV:entire atmosphere"] + + # Build pressure-level profile + profile = + @pressure_levels + |> Enum.map(fn level_str -> + level = String.to_integer(level_str) + pres_key = "#{level} mb" + + %{ + "pres" => level * 1.0, + "tmpc" => kelvin_to_celsius(pressure_points["TMP:#{pres_key}"]), + "dwpc" => kelvin_to_celsius(pressure_points["DPT:#{pres_key}"]), + "hght" => geopotential_to_height(pressure_points["HGT:#{pres_key}"]) + } + end) + |> Enum.reject(fn p -> is_nil(p["tmpc"]) end) + + # Derive refractivity and ducting from profile + params = SoundingParams.derive(profile) + + attrs = %{ + valid_time: valid_time, + lat: lat, + lon: lon, + profile: profile, + hpbl_m: hpbl_m, + pwat_mm: pwat_mm, + surface_temp_c: temp_c, + surface_dewpoint_c: dewpoint_c, + surface_pressure_mb: pressure_mb + } + + attrs = + if params do + Map.merge(attrs, %{ + surface_refractivity: params.surface_refractivity, + min_refractivity_gradient: params.min_refractivity_gradient, + ducting_detected: params.ducting_detected || false, + duct_characteristics: params.duct_characteristics + }) + else + attrs + end + + {:ok, attrs} + end + rescue + e -> {:error, "ERA5 GRIB2 decode failed: #{Exception.message(e)}"} + end + + defp extract_point(grib_data, lat, lon) do + case Extractor.extract_points(grib_data, lat, lon) do + {:ok, fields} when map_size(fields) > 0 -> {:ok, fields} + {:ok, _} -> {:error, "no data at #{lat},#{lon}"} + {:error, reason} -> {:error, reason} + end + end + + defp kelvin_to_celsius(nil), do: nil + defp kelvin_to_celsius(k), do: k - 273.15 + + defp pa_to_mb(nil), do: nil + defp pa_to_mb(pa), do: pa / 100.0 + + defp geopotential_to_height(nil), do: nil + defp geopotential_to_height(gp), do: gp / 9.80665 + + defp api_key do + System.get_env("ERA5_CDS_API_KEY") + end +end diff --git a/lib/microwaveprop/weather/era5_profile.ex b/lib/microwaveprop/weather/era5_profile.ex new file mode 100644 index 00000000..e7681dff --- /dev/null +++ b/lib/microwaveprop/weather/era5_profile.ex @@ -0,0 +1,37 @@ +defmodule Microwaveprop.Weather.Era5Profile do + @moduledoc false + use Ecto.Schema + + import Ecto.Changeset + + @primary_key {:id, :binary_id, autogenerate: true} + @foreign_key_type :binary_id + + schema "era5_profiles" do + field :valid_time, :utc_datetime + field :lat, :float + field :lon, :float + field :profile, {:array, :map} + field :hpbl_m, :float + field :pwat_mm, :float + field :surface_temp_c, :float + field :surface_dewpoint_c, :float + field :surface_pressure_mb, :float + field :surface_refractivity, :float + field :min_refractivity_gradient, :float + field :ducting_detected, :boolean, default: false + field :duct_characteristics, {:array, :map} + + timestamps(type: :utc_datetime) + end + + @required_fields ~w(valid_time lat lon)a + @optional_fields ~w(profile hpbl_m pwat_mm surface_temp_c surface_dewpoint_c surface_pressure_mb surface_refractivity min_refractivity_gradient ducting_detected duct_characteristics)a + + def changeset(profile, attrs) do + profile + |> cast(attrs, @required_fields ++ @optional_fields) + |> validate_required(@required_fields) + |> unique_constraint([:lat, :lon, :valid_time]) + end +end diff --git a/lib/microwaveprop/weather/rtma_client.ex b/lib/microwaveprop/weather/rtma_client.ex new file mode 100644 index 00000000..47ea65d0 --- /dev/null +++ b/lib/microwaveprop/weather/rtma_client.ex @@ -0,0 +1,160 @@ +defmodule Microwaveprop.Weather.RtmaClient do + @moduledoc """ + Client for NOAA RTMA (Real-Time Mesoscale Analysis) data. + 2.5 km resolution, 15-minute analysis cycles, CONUS coverage. + + Available on AWS S3 at s3://noaa-rtma-pds/. GRIB2 format with + byte-range requests, same pattern as HRRR. + + RTMA provides surface fields only (no pressure levels): + - 2m temperature, 2m dewpoint + - 10m wind U/V + - Surface pressure + - Visibility + - Precipitation analysis + """ + + alias Microwaveprop.Weather.Grib2.Extractor + + require Logger + + @s3_base "https://noaa-rtma-pds.s3.amazonaws.com" + + @wanted_fields [ + "TMP:2 m above ground", + "DPT:2 m above ground", + "PRES:surface", + "UGRD:10 m above ground", + "VGRD:10 m above ground", + "VIS:surface" + ] + + @doc """ + Fetch RTMA observation for a point at a specific time. + Returns {:ok, attrs} or {:error, reason}. + """ + def fetch_observation(lat, lon, timestamp) do + # RTMA runs every hour with 15-min updates; round to nearest hour + valid_time = %{DateTime.truncate(timestamp, :second) | minute: 0, second: 0} + rlat = Float.round(lat * 40) / 40 + rlon = Float.round(lon * 40) / 40 + + url = rtma_url(valid_time) + idx_url = "#{url}.idx" + + with {:ok, idx_body} <- fetch_idx(idx_url), + ranges = byte_ranges_for_fields(idx_body, @wanted_fields), + {:ok, grib_data} <- download_ranges(url, ranges), + {:ok, fields} <- extract_point(grib_data, rlat, rlon) do + attrs = %{ + valid_time: valid_time, + lat: rlat, + lon: rlon, + temp_c: kelvin_to_celsius(fields["TMP:2 m above ground"]), + dewpoint_c: kelvin_to_celsius(fields["DPT:2 m above ground"]), + pressure_mb: pa_to_mb(fields["PRES:surface"]), + wind_u_ms: fields["UGRD:10 m above ground"], + wind_v_ms: fields["VGRD:10 m above ground"], + visibility_m: fields["VIS:surface"] + } + + {:ok, attrs} + end + end + + @doc "Build S3 URL for an RTMA analysis file." + def rtma_url(valid_time) do + date_str = Calendar.strftime(valid_time, "%Y%m%d") + hour_str = valid_time.hour |> Integer.to_string() |> String.pad_leading(2, "0") + "#{@s3_base}/rtma2p5.#{date_str}/rtma2p5.t#{hour_str}z.2dvaranl_ndfd.grb2_wexp" + end + + defp fetch_idx(url) do + case Req.get(url, receive_timeout: 15_000) do + {:ok, %{status: 200, body: body}} -> {:ok, body} + {:ok, %{status: status}} -> {:error, "RTMA idx HTTP #{status}"} + {:error, reason} -> {:error, "RTMA idx failed: #{inspect(reason)}"} + end + end + + defp byte_ranges_for_fields(idx_body, wanted_fields) do + lines = + idx_body + |> String.split("\n", trim: true) + |> Enum.map(fn line -> + parts = String.split(line, ":") + %{offset: String.to_integer(Enum.at(parts, 1, "0")), field: Enum.at(parts, 3, "") <> ":" <> Enum.at(parts, 4, "")} + end) + + offsets = Enum.map(lines, & &1.offset) + + lines + |> Enum.with_index() + |> Enum.filter(fn {line, _i} -> + Enum.any?(wanted_fields, &String.contains?(line.field, &1)) + end) + |> Enum.map(fn {line, i} -> + next_offset = Enum.at(offsets, i + 1, line.offset + 5_000_000) + {line.offset, next_offset - 1} + end) + |> merge_ranges() + end + + defp merge_ranges(ranges) do + ranges + |> Enum.sort() + |> Enum.reduce([], fn + range, [] -> [range] + {s2, e2}, [{s1, e1} | rest] when s2 <= e1 + 1 -> [{s1, max(e1, e2)} | rest] + range, acc -> [range | acc] + end) + |> Enum.reverse() + end + + defp download_ranges(url, ranges) do + data = + ranges + |> Task.async_stream( + fn {range_start, range_end} -> + Req.get(url, + headers: [{"Range", "bytes=#{range_start}-#{range_end}"}], + receive_timeout: 30_000 + ) + end, + max_concurrency: 4, + timeout: 60_000 + ) + |> Enum.reduce({:ok, []}, fn + {:ok, {:ok, %{status: status, body: chunk}}}, {:ok, acc} when status in [200, 206] -> + {:ok, [chunk | acc]} + + {:ok, {:ok, %{status: status}}}, _acc -> + {:error, "RTMA download HTTP #{status}"} + + {:ok, {:error, reason}}, _acc -> + {:error, "RTMA download failed: #{inspect(reason)}"} + + _, acc -> + acc + end) + + case data do + {:ok, chunks} -> {:ok, chunks |> Enum.reverse() |> IO.iodata_to_binary()} + error -> error + end + end + + defp extract_point(grib_data, lat, lon) do + case Extractor.extract_points(grib_data, lat, lon) do + {:ok, fields} when map_size(fields) > 0 -> {:ok, fields} + {:ok, _} -> {:error, "RTMA: no data at #{lat},#{lon}"} + {:error, reason} -> {:error, reason} + end + end + + defp kelvin_to_celsius(nil), do: nil + defp kelvin_to_celsius(k), do: k - 273.15 + + defp pa_to_mb(nil), do: nil + defp pa_to_mb(pa), do: pa / 100.0 +end diff --git a/lib/microwaveprop/weather/rtma_observation.ex b/lib/microwaveprop/weather/rtma_observation.ex new file mode 100644 index 00000000..53b6bee4 --- /dev/null +++ b/lib/microwaveprop/weather/rtma_observation.ex @@ -0,0 +1,34 @@ +defmodule Microwaveprop.Weather.RtmaObservation do + @moduledoc false + use Ecto.Schema + + import Ecto.Changeset + + @primary_key {:id, :binary_id, autogenerate: true} + @foreign_key_type :binary_id + + schema "rtma_observations" do + field :valid_time, :utc_datetime + field :lat, :float + field :lon, :float + field :temp_c, :float + field :dewpoint_c, :float + field :pressure_mb, :float + field :wind_u_ms, :float + field :wind_v_ms, :float + field :visibility_m, :float + field :precip_mm, :float + + timestamps(type: :utc_datetime) + end + + @required_fields ~w(valid_time lat lon)a + @optional_fields ~w(temp_c dewpoint_c pressure_mb wind_u_ms wind_v_ms visibility_m precip_mm)a + + def changeset(observation, attrs) do + observation + |> cast(attrs, @required_fields ++ @optional_fields) + |> validate_required(@required_fields) + |> unique_constraint([:lat, :lon, :valid_time]) + end +end diff --git a/lib/microwaveprop/workers/era5_fetch_worker.ex b/lib/microwaveprop/workers/era5_fetch_worker.ex new file mode 100644 index 00000000..e1153db6 --- /dev/null +++ b/lib/microwaveprop/workers/era5_fetch_worker.ex @@ -0,0 +1,69 @@ +defmodule Microwaveprop.Workers.Era5FetchWorker do + @moduledoc """ + Fetches ERA5 reanalysis profiles for contacts where HRRR data is unavailable. + Primarily for pre-2014 contacts that predate HRRR availability. + """ + use Oban.Worker, queue: :era5, max_attempts: 5 + + alias Microwaveprop.Repo + alias Microwaveprop.Weather.Era5Client + alias Microwaveprop.Weather.Era5Profile + + require Logger + + @impl Oban.Worker + def backoff(%Oban.Job{attempt: attempt}) do + # ERA5 CDS API can be slow; generous backoff + min(300 * Integer.pow(2, attempt - 1), _one_day = 86_400) + end + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do + {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str) + + rlat = Float.round(lat * 4) / 4 + rlon = Float.round(lon * 4) / 4 + + if has_era5_profile?(rlat, rlon, valid_time) do + Logger.debug("ERA5: profile already exists for #{rlat},#{rlon} @ #{valid_time_str}") + :ok + else + Logger.info("ERA5: fetching profile for #{rlat},#{rlon} @ #{valid_time_str}") + + case Era5Client.fetch_profile(lat, lon, valid_time) do + {:ok, attrs} -> + %Era5Profile{} + |> Era5Profile.changeset(attrs) + |> Repo.insert( + on_conflict: :nothing, + conflict_target: [:lat, :lon, :valid_time] + ) + + Logger.info("ERA5: stored profile for #{rlat},#{rlon} @ #{valid_time_str}") + :ok + + {:error, reason} -> + Logger.warning("ERA5: failed for #{rlat},#{rlon} @ #{valid_time_str}: #{inspect(reason)}") + {:error, reason} + end + end + end + + defp has_era5_profile?(lat, lon, valid_time) do + import Ecto.Query + + dlat = 0.13 + dlon = 0.13 + time_start = DateTime.add(valid_time, -1800, :second) + time_end = DateTime.add(valid_time, 1800, :second) + + Era5Profile + |> where( + [p], + p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and + p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and + p.valid_time >= ^time_start and p.valid_time <= ^time_end + ) + |> Repo.exists?() + end +end diff --git a/lib/microwaveprop/workers/rtma_fetch_worker.ex b/lib/microwaveprop/workers/rtma_fetch_worker.ex new file mode 100644 index 00000000..7a3201fc --- /dev/null +++ b/lib/microwaveprop/workers/rtma_fetch_worker.ex @@ -0,0 +1,68 @@ +defmodule Microwaveprop.Workers.RtmaFetchWorker do + @moduledoc """ + Fetches RTMA surface observations for real-time propagation scoring. + 15-minute resolution provides finer temporal detail than HRRR's hourly cycle. + """ + use Oban.Worker, queue: :rtma, max_attempts: 10 + + alias Microwaveprop.Repo + alias Microwaveprop.Weather.RtmaClient + alias Microwaveprop.Weather.RtmaObservation + + require Logger + + @impl Oban.Worker + def backoff(%Oban.Job{attempt: attempt}) do + min(60 * Integer.pow(2, attempt - 1), _six_hours = 21_600) + end + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do + {:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str) + + rlat = Float.round(lat * 40) / 40 + rlon = Float.round(lon * 40) / 40 + + if has_rtma_observation?(rlat, rlon, valid_time) do + Logger.debug("RTMA: observation exists for #{rlat},#{rlon} @ #{valid_time_str}") + :ok + else + Logger.info("RTMA: fetching for #{rlat},#{rlon} @ #{valid_time_str}") + + case RtmaClient.fetch_observation(lat, lon, valid_time) do + {:ok, attrs} -> + %RtmaObservation{} + |> RtmaObservation.changeset(attrs) + |> Repo.insert( + on_conflict: :nothing, + conflict_target: [:lat, :lon, :valid_time] + ) + + Logger.info("RTMA: stored observation for #{rlat},#{rlon} @ #{valid_time_str}") + :ok + + {:error, reason} -> + Logger.warning("RTMA: failed for #{rlat},#{rlon} @ #{valid_time_str}: #{inspect(reason)}") + {:error, reason} + end + end + end + + defp has_rtma_observation?(lat, lon, valid_time) do + import Ecto.Query + + dlat = 0.03 + dlon = 0.03 + time_start = DateTime.add(valid_time, -450, :second) + time_end = DateTime.add(valid_time, 450, :second) + + RtmaObservation + |> where( + [o], + o.lat >= ^(lat - dlat) and o.lat <= ^(lat + dlat) and + o.lon >= ^(lon - dlon) and o.lon <= ^(lon + dlon) and + o.valid_time >= ^time_start and o.valid_time <= ^time_end + ) + |> Repo.exists?() + end +end diff --git a/priv/repo/migrations/20260407165919_add_era5_profiles_and_rtma_observations.exs b/priv/repo/migrations/20260407165919_add_era5_profiles_and_rtma_observations.exs new file mode 100644 index 00000000..9795b95b --- /dev/null +++ b/priv/repo/migrations/20260407165919_add_era5_profiles_and_rtma_observations.exs @@ -0,0 +1,46 @@ +defmodule Microwaveprop.Repo.Migrations.AddEra5ProfilesAndRtmaObservations do + use Ecto.Migration + + def change do + create table(:era5_profiles, primary_key: false) do + add :id, :binary_id, primary_key: true + add :valid_time, :utc_datetime, null: false + add :lat, :float, null: false + add :lon, :float, null: false + add :profile, {:array, :map}, default: [] + add :hpbl_m, :float + add :pwat_mm, :float + add :surface_temp_c, :float + add :surface_dewpoint_c, :float + add :surface_pressure_mb, :float + add :surface_refractivity, :float + add :min_refractivity_gradient, :float + add :ducting_detected, :boolean, default: false, null: false + add :duct_characteristics, {:array, :map} + + timestamps(type: :utc_datetime) + end + + create unique_index(:era5_profiles, [:lat, :lon, :valid_time]) + create index(:era5_profiles, [:valid_time]) + + create table(:rtma_observations, primary_key: false) do + add :id, :binary_id, primary_key: true + add :valid_time, :utc_datetime, null: false + add :lat, :float, null: false + add :lon, :float, null: false + add :temp_c, :float + add :dewpoint_c, :float + add :pressure_mb, :float + add :wind_u_ms, :float + add :wind_v_ms, :float + add :visibility_m, :float + add :precip_mm, :float + + timestamps(type: :utc_datetime) + end + + create unique_index(:rtma_observations, [:lat, :lon, :valid_time]) + create index(:rtma_observations, [:valid_time]) + end +end