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

@@ -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},

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

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

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

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

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

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

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

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