prop/lib/microwaveprop/weather/iem_client.ex

defmodule Microwaveprop.Weather.IemClient do
  @moduledoc false

  alias Microwaveprop.Weather.IemRateLimiter

  require Logger

  @iem_base "https://mesonet.agron.iastate.edu"

  # --- URL builders ---

  @spec network_url(String.t()) :: String.t()
  def network_url(network) do
    "#{@iem_base}/json/network.py?network=#{network}"
  end

  @spec asos_url(String.t() | [String.t()], DateTime.t(), DateTime.t()) :: String.t()
  def asos_url(station_id_or_codes, start_dt, end_dt) do
    codes = List.wrap(station_id_or_codes)
    station_params = Enum.map_join(codes, "&", &"station=#{&1}")

    "#{@iem_base}/cgi-bin/request/asos.py" <>
      "?#{station_params}" <>
      "&data=tmpf&data=dwpf&data=relh&data=sknt&data=drct" <>
      "&data=mslp&data=alti&data=skyc1&data=p01i&data=wxcodes" <>
      "&year1=#{start_dt.year}&month1=#{start_dt.month}&day1=#{start_dt.day}" <>
      "&hour1=#{start_dt.hour}&minute1=0" <>
      "&year2=#{end_dt.year}&month2=#{end_dt.month}&day2=#{end_dt.day}" <>
      "&hour2=#{end_dt.hour}&minute2=59" <>
      "&format=onlycomma&latlon=no&elev=no&missing=null&trace=null&direct=no&report_type=3"
  end

  @spec raob_url(String.t(), DateTime.t()) :: String.t()
  def raob_url(station_id, dt) do
    ts = format_iem_ts(dt)
    "#{@iem_base}/json/raob.py?ts=#{ts}&station=#{station_id}"
  end

  @spec iemre_url(float(), float(), Date.t()) :: String.t()
  def iemre_url(lat, lon, date) do
    "#{@iem_base}/iemre/hourly/#{date}/#{lat}/#{lon}/json"
  end

  # --- HTTP fetchers ---

  @spec fetch_network(String.t()) :: {:ok, [map()]} | {:error, term()}
  def fetch_network(network) do
    url = network_url(network)
    IemRateLimiter.acquire()

    url
    |> Req.get(req_options())
    |> handle_response(&parse_network_json/1, "IEM network")
  end

  @spec fetch_asos(String.t(), DateTime.t(), DateTime.t()) :: {:ok, [map()]} | {:error, term()}
  def fetch_asos(station_id, start_dt, end_dt) do
    url = asos_url(station_id, start_dt, end_dt)

    Microwaveprop.Instrument.span([:iem, :fetch_asos], %{station: station_id}, fn ->
      IemRateLimiter.acquire()

      url
      |> Req.get(req_options())
      |> handle_response(&parse_asos_csv/1, "IEM ASOS")
    end)
  end

  @doc """
  Batched variant of `fetch_asos/3` — one HTTP call for all `codes`,
  returning `{:ok, %{code => [rows]}}`. Stations absent from the
  response appear in the map with `[]` so callers can distinguish
  "IEM had no data" from "we never asked about this code".

  IEM's ASOS CSV endpoint accepts multiple `station=` params and
  keys the response rows on the first column. A ~15-station batch
  replaces ~15 individual requests, each of which pays the
  per-pod `IemRateLimiter` gap + 429 retry tail.
  """
  @spec fetch_asos_batch([String.t()], DateTime.t(), DateTime.t()) ::
          {:ok, %{String.t() => [map()]}} | {:error, term()}
  def fetch_asos_batch(codes, start_dt, end_dt) when is_list(codes) and codes != [] do
    url = asos_url(codes, start_dt, end_dt)

    Microwaveprop.Instrument.span([:iem, :fetch_asos_batch], %{count: length(codes)}, fn ->
      IemRateLimiter.acquire()

      parse = fn body ->
        rows = parse_asos_csv(body)
        grouped = Enum.group_by(rows, & &1.station_code)
        # Ensure every requested code has at least an empty list in the
        # map so downstream iteration doesn't miss "no data" cases.
        Enum.reduce(codes, grouped, &Map.put_new(&2, &1, []))
      end

      url
      |> Req.get(req_options())
      |> handle_response(parse, "IEM ASOS")
    end)
  end

  @spec fetch_raob(String.t(), DateTime.t()) :: {:ok, [map()]} | {:error, term()}
  def fetch_raob(station_id, dt) do
    url = raob_url(station_id, dt)

    Microwaveprop.Instrument.span([:iem, :fetch_raob], %{station: station_id}, fn ->
      IemRateLimiter.acquire()

      url
      |> Req.get(req_options())
      |> handle_response(&parse_raob_json/1, "IEM RAOB")
    end)
  end

  @spec fetch_iemre(float(), float(), Date.t()) :: {:ok, [map()]} | {:error, term()}
  def fetch_iemre(lat, lon, date) do
    Microwaveprop.Instrument.span([:iem, :fetch_iemre], %{}, fn ->
      url = iemre_url(lat, lon, date)
      IemRateLimiter.acquire()

      url
      |> Req.get(req_options())
      |> handle_response(&parse_iemre_json/1, "IEM IEMRE")
    end)
  end

  # Central response handler: fires adaptive-limiter feedback signals
  # (widen on 429, narrow on 200) before returning the parsed body or
  # an error tuple. `kind` is the upstream label used in error strings.
  defp handle_response({:ok, %{status: 200, body: body}}, parser, _kind) do
    _ = IemRateLimiter.signal_success()
    {:ok, parser.(body)}
  end

  defp handle_response({:ok, %{status: 429}}, _parser, kind) do
    _ = IemRateLimiter.signal_429()
    {:error, "#{kind} HTTP 429"}
  end

  defp handle_response({:ok, %{status: status}}, _parser, kind) do
    {:error, "#{kind} HTTP #{status}"}
  end

  defp handle_response({:error, reason}, _parser, _kind), do: {:error, reason}

  defp req_options do
    defaults = [retry: &retry?/2, max_retries: 5, retry_delay: &retry_delay/1]
    overrides = Application.get_env(:microwaveprop, :iem_req_options, [])
    Keyword.merge(defaults, overrides)
  end

  defp retry?(_request, response) do
    case response do
      %Req.Response{status: status} when status in [429, 500, 502, 503, 504] -> true
      %{__exception__: true} -> true
      _ -> false
    end
  end

  defp retry_delay(n) do
    base = Integer.pow(2, n) * 1_000
    jitter = :rand.uniform(1_000)
    base + jitter
  end

  @doc """
  Fetch current ASOS observations for all stations in the given state networks.
  Returns `{:ok, [%{station_code, lat, lon, temp_f, dewpoint_f, ...}]}`.
  """
  @spec fetch_current_asos([String.t()]) :: {:ok, [map()]}
  def fetch_current_asos(state_networks) do
    results =
      state_networks
      |> Task.async_stream(
        fn network ->
          url = "#{@iem_base}/api/1/currents.json?network=#{network}"
          IemRateLimiter.acquire()

          case Req.get(url, req_options()) do
            {:ok, %{status: 200, body: %{"data" => data}}} -> {:ok, data}
            {:ok, %{status: status}} -> {:error, "IEM currents HTTP #{status}"}
            {:error, reason} -> {:error, reason}
          end
        end,
        max_concurrency: 10,
        timeout: 30_000
      )
      |> Enum.zip(state_networks)
      |> Enum.flat_map(fn
        {{:ok, {:ok, data}}, _network} ->
          data

        {{:ok, {:error, reason}}, network} ->
          Logger.warning("IemClient.fetch_current_asos #{network}: #{inspect(reason)}")
          []

        {{:exit, reason}, network} ->
          Logger.error("IemClient.fetch_current_asos #{network} crashed: #{inspect(reason)}")
          []
      end)

    observations =
      Enum.flat_map(results, fn obs ->
        # Skip stations with missing core data
        if obs["tmpf"] && obs["dwpf"] && obs["lat"] && obs["lon"] do
          [
            %{
              station_code: obs["station"],
              lat: obs["lat"],
              lon: obs["lon"],
              utc_valid: parse_utc_valid(obs["utc_valid"]),
              temp_f: obs["tmpf"],
              dewpoint_f: obs["dwpf"],
              wind_speed_kts: obs["sknt"],
              sky_condition: obs["skyc1"],
              sea_level_pressure_mb: obs["mslp"],
              altimeter_setting: obs["alti"],
              precip_1h_in: obs["phour"]
            }
          ]
        else
          []
        end
      end)

    {:ok, observations}
  end

  defp parse_utc_valid(nil), do: nil

  defp parse_utc_valid(str) when is_binary(str) do
    case DateTime.from_iso8601(str) do
      {:ok, dt, _} -> DateTime.truncate(dt, :second)
      _ -> nil
    end
  end

  # --- Parsers ---

  @spec parse_iemre_json(map() | binary() | nil) :: [map()]
  def parse_iemre_json(json) when is_map(json) do
    Map.get(json, "data", [])
  end

  # IEM ERDDAP sometimes serves a 200 with an empty body for valid
  # station/date combinations during backfill windows. Treat as "no
  # data" rather than letting the worker stack-trace on a function
  # clause mismatch.
  def parse_iemre_json(body) when body in [nil, ""], do: []

  @spec parse_network_json(map()) :: [map()]
  def parse_network_json(json) when is_map(json) do
    json
    |> Map.get("stations", [])
    |> Enum.map(fn entry ->
      %{
        station_code: entry["id"],
        name: entry["name"],
        lat: entry["lat"],
        lon: entry["lon"]
      }
    end)
  end

  @spec parse_asos_csv(String.t()) :: [map()]
  def parse_asos_csv(csv_text) do
    csv_text
    |> String.split("\n")
    |> Enum.reject(fn line ->
      line == "" or String.starts_with?(line, "#") or String.starts_with?(line, "station")
    end)
    |> Enum.map(&parse_asos_row/1)
  end

  @spec parse_raob_json(map()) :: [map()]
  def parse_raob_json(json) when is_map(json) do
    json
    |> Map.get("profiles", [])
    |> Enum.map(fn entry ->
      %{
        observed_at: parse_raob_timestamp(entry["valid"]),
        profile: entry["profile"] || []
      }
    end)
  end

  # --- Private ---

  defp parse_asos_row(line) do
    parts = String.split(line, ",")

    %{
      station_code: parse_nullable_string(Enum.at(parts, 0)),
      observed_at: parse_asos_timestamp(Enum.at(parts, 1)),
      temp_f: parse_float(Enum.at(parts, 2)),
      dewpoint_f: parse_float(Enum.at(parts, 3)),
      relative_humidity: parse_float(Enum.at(parts, 4)),
      wind_speed_kts: parse_float(Enum.at(parts, 5)),
      wind_direction_deg: parse_int(Enum.at(parts, 6)),
      sea_level_pressure_mb: parse_float(Enum.at(parts, 7)),
      altimeter_setting: parse_float(Enum.at(parts, 8)),
      sky_condition: parse_nullable_string(Enum.at(parts, 9)),
      precip_1h_in: parse_float(Enum.at(parts, 10)),
      wx_codes: parse_nullable_string(Enum.at(parts, 11))
    }
  end

  defp parse_asos_timestamp(str) when is_binary(str) do
    # Format: "2026-03-28 18:53"
    case DateTime.from_iso8601(String.trim(str) <> ":00Z") do
      {:ok, dt, _} -> DateTime.truncate(dt, :second)
      _ -> nil
    end
  end

  defp parse_asos_timestamp(_), do: nil

  defp parse_raob_timestamp(str) when is_binary(str) do
    # Format: "2026-03-28 12:00:00+00:00"
    cleaned =
      str
      |> String.trim()
      |> String.replace(~r/\+00:00$/, "Z")
      |> String.replace(" ", "T")

    case DateTime.from_iso8601(cleaned) do
      {:ok, dt, _} -> DateTime.truncate(dt, :second)
      _ -> nil
    end
  end

  defp parse_raob_timestamp(_), do: nil

  defp parse_float(nil), do: nil
  defp parse_float("null"), do: nil

  defp parse_float(str) do
    case Float.parse(String.trim(str)) do
      {val, _} -> val
      :error -> nil
    end
  end

  defp parse_int(nil), do: nil
  defp parse_int("null"), do: nil

  defp parse_int(str) do
    case Integer.parse(String.trim(str)) do
      {val, _} -> val
      :error -> nil
    end
  end

  defp parse_nullable_string(nil), do: nil
  defp parse_nullable_string("null"), do: nil

  defp parse_nullable_string(str) do
    trimmed = String.trim(str)
    if trimmed == "", do: nil, else: trimmed
  end

  defp format_iem_ts(dt) do
    y = Integer.to_string(dt.year)
    mo = dt.month |> Integer.to_string() |> String.pad_leading(2, "0")
    d = dt.day |> Integer.to_string() |> String.pad_leading(2, "0")
    h = dt.hour |> Integer.to_string() |> String.pad_leading(2, "0")
    mi = dt.minute |> Integer.to_string() |> String.pad_leading(2, "0")
    "#{y}#{mo}#{d}#{h}#{mi}"
  end
end