prop/lib/microwaveprop/pskr.ex

defmodule Microwaveprop.Pskr do
  @moduledoc """
  Ingestion of the PSK Reporter MQTT firehose
  (`mqtt.pskreporter.info:1883`).

  ## Why we listen

  Each MQTT message is a single reception report — *station X heard
  station Y at SNR Z on band B at time T*. Read across millions of
  reports, this is a continuous, geolocated, timestamped record of
  *did propagation actually occur on this path right now?* That's
  the empirical signal we correlate against HRRR atmospheric state
  to recalibrate the propagation scoring weights.

  ## What we keep

  Topic structure (from the broker landing page):

      pskr/filter/v2/<band>/<mode>/<sendercall>/<receivercall>/<sl>/<rl>/<sa>/<ra>

  We subscribe with country-side wildcards anchored on USA (DXCC 291),
  so the broker pre-filters to spots involving at least one CONUS
  station — anything else can't be cross-referenced with HRRR
  anyway.

  Per-spot payload (verbatim keys, also documented at
  http://mqtt.pskreporter.info/):

      sq f md rp t t_tx sc sl rc rl sa ra b

  ## What we discard

  The aggregator collapses every spot inside `(hour_utc, band,
  sender_grid, receiver_grid)` into one `pskr_spots_hourly` row.
  We keep counts + SNR envelope + the set of modes — the
  individual report identity, sequence number, and exact intra-hour
  timestamp are dropped on the floor. A 50-reporter QSO becomes a
  single bumped count, not 50 rows; that's the design, not a
  limitation.

  Locator precision is preserved as reported: a spotter who shares
  `EM12kl` keeps subsquare resolution, while a `EM12kl37` reporter
  keeps extended-square resolution. The bucket key uses the full
  grid, so two spotters in different subsquares of the same field
  produce two rows — that's the right behavior, since they're
  literally different paths.
  """

  alias Microwaveprop.Geo
  alias Microwaveprop.Radio.Maidenhead

  @typedoc "Parsed PSK Reporter spot, post-`parse_spot/1`."
  @type spot :: %{
          band: String.t(),
          mode: String.t(),
          frequency_hz: pos_integer(),
          snr_db: integer(),
          transmit_time: DateTime.t(),
          sender_grid: String.t(),
          receiver_grid: String.t(),
          sender_country: integer() | nil,
          receiver_country: integer() | nil
        }

  @typedoc "Bucket key used by the aggregator's accumulator map."
  @type path_key :: {DateTime.t(), String.t(), String.t(), String.t()}

  @doc """
  Decodes a single MQTT payload (binary JSON) into a normalized
  spot map. Locators are kept at the precision the spotter
  reported (4/6/8/10 chars) and uppercased for consistent storage.
  Returns `{:error, reason}` for any payload missing the fields we
  need — the caller should count and discard.
  """
  @spec parse_spot(binary()) :: {:ok, spot()} | {:error, term()}
  def parse_spot(payload) when is_binary(payload) do
    with {:ok, json} <- Jason.decode(payload),
         {:ok, sender_grid} <- fetch_grid(json, "sl"),
         {:ok, receiver_grid} <- fetch_grid(json, "rl"),
         {:ok, band} <- fetch_string(json, "b"),
         {:ok, time_unix} <- fetch_time(json),
         {:ok, transmit_time} <- DateTime.from_unix(time_unix) do
      {:ok,
       %{
         band: band,
         mode: Map.get(json, "md", ""),
         frequency_hz: Map.get(json, "f"),
         snr_db: Map.get(json, "rp"),
         transmit_time: transmit_time,
         sender_grid: sender_grid,
         receiver_grid: receiver_grid,
         sender_country: Map.get(json, "sa"),
         receiver_country: Map.get(json, "ra")
       }}
    end
  end

  @doc "Bucket a spot to (top-of-hour, band, sender_grid, receiver_grid) at full reported precision."
  @spec path_key(spot()) :: path_key()
  def path_key(%{transmit_time: t, band: band, sender_grid: snd, receiver_grid: rcv}) do
    {%{t | minute: 0, second: 0, microsecond: {0, 0}}, band, snd, rcv}
  end

  @doc """
  Fold a spot into an accumulator row. Pass `nil` as the first
  argument to seed a fresh row. The result is an attribute map
  ready to feed into `Repo.insert_all/3` / `Ecto.Changeset.cast/3`.

  Pre-computes sender/receiver lat-lon, midpoint, and great-circle
  distance once at first sight — they don't change for subsequent
  spots on the same path, so the merge clause is cheap.
  """
  @spec merge_spot(map() | nil, spot()) :: map()
  def merge_spot(nil, spot) do
    {hour, band, snd, rcv} = path_key(spot)
    {snd_lat, snd_lon} = grid_center(snd)
    {rcv_lat, rcv_lon} = grid_center(rcv)
    distance = Geo.haversine_km(snd_lat, snd_lon, rcv_lat, rcv_lon)

    %{
      hour_utc: hour,
      band: band,
      sender_grid: snd,
      receiver_grid: rcv,
      sender_country: spot.sender_country,
      receiver_country: spot.receiver_country,
      sender_lat: snd_lat,
      sender_lon: snd_lon,
      receiver_lat: rcv_lat,
      receiver_lon: rcv_lon,
      midpoint_lat: (snd_lat + rcv_lat) / 2,
      midpoint_lon: (snd_lon + rcv_lon) / 2,
      distance_km: distance,
      spot_count: 1,
      max_snr_db: spot.snr_db,
      min_snr_db: spot.snr_db,
      modes: maybe_mode_list(spot.mode),
      first_spot_at: spot.transmit_time,
      last_spot_at: spot.transmit_time
    }
  end

  def merge_spot(%{} = acc, spot) do
    %{
      acc
      | spot_count: acc.spot_count + 1,
        max_snr_db: max_snr(acc.max_snr_db, spot.snr_db),
        min_snr_db: min_snr(acc.min_snr_db, spot.snr_db),
        modes: add_mode(acc.modes, spot.mode),
        first_spot_at: earliest(acc.first_spot_at, spot.transmit_time),
        last_spot_at: latest(acc.last_spot_at, spot.transmit_time)
    }
  end

  # Keep locators at the precision the spotter reported. Maidenhead
  # validity already enforces minimum 4 chars + even length + the
  # right alphabet at each position, so anything that passes is a
  # legal 4/6/8/10-char locator we can store as-is. Uppercase for
  # consistent indexing — the unique key is case-sensitive.
  defp fetch_grid(json, key) do
    case Map.get(json, key) do
      grid when is_binary(grid) ->
        upper = String.upcase(grid)
        if Maidenhead.valid?(upper), do: {:ok, upper}, else: {:error, {:bad_grid, key, grid}}

      other ->
        {:error, {:missing_grid, key, other}}
    end
  end

  defp fetch_string(json, key) do
    case Map.get(json, key) do
      v when is_binary(v) and v != "" -> {:ok, v}
      other -> {:error, {:missing, key, other}}
    end
  end

  # `t_tx` is the transmission time and is the right anchor for
  # hour-bucketing — that's when propagation actually carried the
  # signal. `t` (PSKReporter receive time) drifts a few seconds and
  # would split spots that crossed an hour boundary onto two rows.
  defp fetch_time(json) do
    case {Map.get(json, "t_tx"), Map.get(json, "t")} do
      {n, _} when is_integer(n) -> {:ok, n}
      {_, n} when is_integer(n) -> {:ok, n}
      _ -> {:error, :no_timestamp}
    end
  end

  defp grid_center(grid), do: Geo.maidenhead_center(grid) || {0.0, 0.0}

  defp maybe_mode_list(""), do: []
  defp maybe_mode_list(nil), do: []
  defp maybe_mode_list(mode), do: [mode]

  defp add_mode(modes, ""), do: modes
  defp add_mode(modes, nil), do: modes
  defp add_mode(modes, mode), do: if(mode in modes, do: modes, else: [mode | modes])

  defp max_snr(nil, b), do: b
  defp max_snr(a, nil), do: a
  defp max_snr(a, b), do: max(a, b)

  defp min_snr(nil, b), do: b
  defp min_snr(a, nil), do: a
  defp min_snr(a, b), do: min(a, b)

  defp earliest(a, b), do: if(DateTime.before?(a, b), do: a, else: b)
  defp latest(a, b), do: if(DateTime.after?(a, b), do: a, else: b)
end