From cfc5f7583f2eb8511869c8a4246d4767b56fd3b9 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 4 May 2026 09:24:20 -0500 Subject: [PATCH] feat(pskr): ingest PSK Reporter MQTT firehose for weather correlation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Subscribes to mqtt.pskreporter.info:1883 over plain TCP and folds every reception report into hourly aggregates per (band, sender_grid_4, receiver_grid_4). Each spot is the empirical "propagation actually occurred on this path right now" signal we'll correlate against HRRR atmospheric state to recalibrate the scoring weights. Aggregating at the path-hour bucket collapses 50-reporter QSOs to one row instead of 50. Topic filter anchors on USA (DXCC 291) on either sender or receiver side so the broker pre-filters out everything outside our HRRR domain. Bands subscribed: VHF and up (6m, 2m, 70cm, 23cm, + microwave) — HF is dominated by ionospheric propagation, which this project doesn't model. Pskr.Client cluster-elects a singleton via :global.register_name — all replicas start the GenServer but only one connects to MQTT; the rest stay in :standby and watch for the leader's nodedown to re-run election. Off in dev/test, on by default in prod (PSKR_MQTT_ENABLED=false as kill-switch). Pskr.Aggregator buffers in memory keyed by Pskr.path_key/1 and flushes every 60s via Repo.insert_all/3 with an additive ON CONFLICT (count summed, SNR envelope by GREATEST/LEAST, modes unioned via unnest+DISTINCT). Idempotent across overlapping flushes and across leader handover. Dockerfile sets BUILD_WITHOUT_QUIC=1 — emqtt's transitive quicer dep wants CMake + OpenSSL headers we'd otherwise have to add to the builder image just to never use QUIC over plain MQTT. Base image is unchanged; the new dep compiles cleanly into the existing prop-base runtime. --- Dockerfile | 8 + config/runtime.exs | 13 ++ config/test.exs | 5 + lib/microwaveprop/application.ex | 29 ++- lib/microwaveprop/pskr.ex | 198 +++++++++++++++++ lib/microwaveprop/pskr/aggregator.ex | 136 ++++++++++++ lib/microwaveprop/pskr/client.ex | 204 ++++++++++++++++++ lib/microwaveprop/pskr/spot_hourly.ex | 68 ++++++ mix.exs | 7 + mix.lock | 6 + ...0260504135243_create_pskr_spots_hourly.exs | 33 +++ test/microwaveprop/pskr/aggregator_test.exs | 79 +++++++ test/microwaveprop/pskr_test.exs | 106 +++++++++ 13 files changed, 890 insertions(+), 2 deletions(-) create mode 100644 lib/microwaveprop/pskr.ex create mode 100644 lib/microwaveprop/pskr/aggregator.ex create mode 100644 lib/microwaveprop/pskr/client.ex create mode 100644 lib/microwaveprop/pskr/spot_hourly.ex create mode 100644 priv/repo/migrations/20260504135243_create_pskr_spots_hourly.exs create mode 100644 test/microwaveprop/pskr/aggregator_test.exs create mode 100644 test/microwaveprop/pskr_test.exs diff --git a/Dockerfile b/Dockerfile index 2455c528..f01c9573 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,6 +43,14 @@ RUN mix local.hex --force \ # set build ENV ENV MIX_ENV="prod" +# Skip the QUIC native build inside emqtt's transitive `quicer` dep. +# We only need plain TCP MQTT (port 1883) for the PSK Reporter +# firehose, and quicer's CMake-driven build pulls in extra +# toolchain requirements (cmake, OpenSSL headers) that aren't worth +# adding to the slim build image just to leave the support +# unused. +ENV BUILD_WITHOUT_QUIC="1" + # install mix dependencies (changes only when mix.exs/mix.lock change) COPY mix.exs mix.lock ./ COPY vendor vendor diff --git a/config/runtime.exs b/config/runtime.exs index a443e55e..c6473cbe 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -378,6 +378,19 @@ if config_env() == :prod do url -> config :microwaveprop, :valkey_url, url end + # PSK Reporter MQTT firehose ingestion. On by default in prod — + # all replicas start the client GenServer but only the + # cluster-elected leader actually connects (`:global` registration + # in `Microwaveprop.Pskr.Client`). Set `PSKR_MQTT_ENABLED=false` + # to kill the entire feature without code change. + case System.get_env("PSKR_MQTT_ENABLED") do + disabled when disabled in ["0", "false", "FALSE"] -> + config :microwaveprop, :pskr_mqtt_enabled, false + + _ -> + config :microwaveprop, :pskr_mqtt_enabled, true + end + # Optional secondary repo for read-only access to aprs.me's database. # Used by APRS-based 144 MHz calibration tooling. When the env var is # unset OR an empty string (k8s secret resolves to "") we leave the diff --git a/config/test.exs b/config/test.exs index 854b94a4..a096844b 100644 --- a/config/test.exs +++ b/config/test.exs @@ -84,6 +84,11 @@ config :microwaveprop, :propagation_scores_dir, Path.join(System.tmp_dir!(), "microwaveprop_test_scores_#{System.unique_integer([:positive])}") +# Don't open an outbound MQTT connection from `mix test`. The +# Pskr.Aggregator still starts (so unit tests can drive it), but +# Pskr.Client is gated off here. +config :microwaveprop, :pskr_mqtt_enabled, false + # Skip the Overpass road-proximity HTTP call in tests — the rover live # test exercises the full Calculate pipeline and we don't want it # reaching out to overpass-api.de. diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 1e3f3948..ed2b96a1 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -46,7 +46,15 @@ defmodule Microwaveprop.Application do Microwaveprop.RepoListener, # Start to serve requests, typically the last entry MicrowavepropWeb.Endpoint, - Microwaveprop.Propagation.FreshnessMonitor + Microwaveprop.Propagation.FreshnessMonitor, + # PSK Reporter MQTT firehose. Aggregator runs on every pod + # (cheap idle GenServer); the Client also runs on every pod + # but only the cluster-elected leader actually opens an MQTT + # connection — the rest stay in `:standby` watching for the + # leader to drop. See `Microwaveprop.Pskr.Client`. Disabled + # in dev/test so `mix test` doesn't try to dial out. + Microwaveprop.Pskr.Aggregator, + pskr_client_child_spec() ] # See https://hexdocs.pm/elixir/Supervisor.html @@ -134,7 +142,24 @@ defmodule Microwaveprop.Application do # to Redix which rejects it as an unknown option (the valid # keys are :host, :port, :password, …, NOT :url). The bare URL # form takes the binary-clause and works correctly. - {Phoenix.PubSub, name: Microwaveprop.PubSub, adapter: Phoenix.PubSub.Redis, redis_opts: url, node_name: node_name} + opts = [ + name: Microwaveprop.PubSub, + adapter: Phoenix.PubSub.Redis, + redis_opts: url, + node_name: node_name + ] + + {Phoenix.PubSub, opts} + end + end + + # PSK Reporter MQTT client. Returns the child spec when + # `:pskr_mqtt_enabled` is true, otherwise `nil` so the supervisor + # children list filters it out. Off by default — see runtime.exs + # for the env-var gate. Only one pod should run this. + defp pskr_client_child_spec do + if Application.get_env(:microwaveprop, :pskr_mqtt_enabled, false) do + Microwaveprop.Pskr.Client end end diff --git a/lib/microwaveprop/pskr.ex b/lib/microwaveprop/pskr.ex new file mode 100644 index 00000000..f58520de --- /dev/null +++ b/lib/microwaveprop/pskr.ex @@ -0,0 +1,198 @@ +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//////// + + 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_4, receiver_grid_4)` 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. + """ + + 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. Truncates locators to 4 characters since hourly + aggregation is at subsquare resolution. 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 4-char, receiver 4-char)." + @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 + + defp fetch_grid(json, key) do + case Map.get(json, key) do + grid when is_binary(grid) and byte_size(grid) >= 4 -> + head = grid |> String.slice(0, 4) |> String.upcase() + if Maidenhead.valid?(head), do: {:ok, head}, 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 diff --git a/lib/microwaveprop/pskr/aggregator.ex b/lib/microwaveprop/pskr/aggregator.ex new file mode 100644 index 00000000..883bfe2e --- /dev/null +++ b/lib/microwaveprop/pskr/aggregator.ex @@ -0,0 +1,136 @@ +defmodule Microwaveprop.Pskr.Aggregator do + @moduledoc """ + In-memory accumulator for incoming PSK Reporter spots. + + The MQTT firehose can push 500+ messages/second when subscribed + broadly. Writing each spot as a row would (a) saturate the + Postgres write path and (b) bloat the dataset 50× because every + QSO is heard by many reporters. This GenServer keeps a small + state map keyed by `Pskr.path_key/1`, folds each spot into the + matching row with `Pskr.merge_spot/2`, and periodically batches + the whole map into a single `Repo.insert_all/3` upsert. + + Conflict handling is additive: when a row with the same + `(hour_utc, band, sender_grid, receiver_grid)` already exists in + the DB, the upsert merges counts, expands the SNR envelope, and + unions the mode set rather than overwriting. That makes flushes + idempotent in the face of overlap (e.g. a flush firing a few + seconds before the hour boundary). + + Memory profile: each pending row is small (~200 bytes), and the + unique key space is bounded by `bands × paths_per_hour`. With our + CONUS-anchored topic filter that's typically a few thousand keys + per hour — flushed every minute, the working set never exceeds a + handful of MiB. + """ + use GenServer + + import Ecto.Query + + alias Microwaveprop.Pskr + alias Microwaveprop.Pskr.SpotHourly + alias Microwaveprop.Repo + + require Logger + + @default_flush_ms 60_000 + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts \\ []) do + {name, opts} = Keyword.pop(opts, :name, __MODULE__) + GenServer.start_link(__MODULE__, opts, name: name) + end + + @doc "Push one raw MQTT payload (binary JSON) into the accumulator." + @spec ingest(GenServer.server(), binary()) :: :ok + def ingest(server \\ __MODULE__, payload) when is_binary(payload) do + GenServer.cast(server, {:ingest, payload}) + end + + @doc "Force a synchronous flush. Returns the number of rows written." + @spec flush(GenServer.server()) :: non_neg_integer() + def flush(server \\ __MODULE__) do + GenServer.call(server, :flush, 30_000) + end + + @doc "Inspect the current pending row count without flushing." + @spec pending_count(GenServer.server()) :: non_neg_integer() + def pending_count(server \\ __MODULE__), do: GenServer.call(server, :pending_count) + + @impl true + def init(opts) do + flush_ms = Keyword.get(opts, :flush_ms, @default_flush_ms) + if flush_ms > 0, do: schedule_flush(flush_ms) + {:ok, %{rows: %{}, parsed: 0, dropped: 0, flush_ms: flush_ms}} + end + + @impl true + def handle_cast({:ingest, payload}, state) do + case Pskr.parse_spot(payload) do + {:ok, spot} -> + key = Pskr.path_key(spot) + rows = Map.update(state.rows, key, Pskr.merge_spot(nil, spot), &Pskr.merge_spot(&1, spot)) + {:noreply, %{state | rows: rows, parsed: state.parsed + 1}} + + {:error, _reason} -> + {:noreply, %{state | dropped: state.dropped + 1}} + end + end + + @impl true + def handle_call(:flush, _from, state) do + written = do_flush(state.rows) + {:reply, written, %{state | rows: %{}}} + end + + def handle_call(:pending_count, _from, state) do + {:reply, map_size(state.rows), state} + end + + @impl true + def handle_info(:flush_tick, state) do + written = do_flush(state.rows) + + if written > 0 or state.dropped > 0 do + Logger.info("Pskr.Aggregator flushed #{written} rows (parsed #{state.parsed}, dropped #{state.dropped})") + end + + schedule_flush(state.flush_ms) + {:noreply, %{state | rows: %{}, parsed: 0, dropped: 0}} + end + + defp schedule_flush(ms) when ms > 0, do: Process.send_after(self(), :flush_tick, ms) + defp schedule_flush(_), do: :ok + + defp do_flush(rows) when map_size(rows) == 0, do: 0 + + defp do_flush(rows) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + entries = + rows + |> Map.values() + |> Enum.map(&Map.merge(&1, %{inserted_at: now, updated_at: now})) + + {count, _} = + Repo.insert_all(SpotHourly, entries, + on_conflict: + from(s in SpotHourly, + update: [ + set: [ + spot_count: fragment("? + EXCLUDED.spot_count", s.spot_count), + max_snr_db: fragment("GREATEST(?, EXCLUDED.max_snr_db)", s.max_snr_db), + min_snr_db: fragment("LEAST(?, EXCLUDED.min_snr_db)", s.min_snr_db), + modes: fragment("ARRAY(SELECT DISTINCT unnest(? || EXCLUDED.modes))", s.modes), + first_spot_at: fragment("LEAST(?, EXCLUDED.first_spot_at)", s.first_spot_at), + last_spot_at: fragment("GREATEST(?, EXCLUDED.last_spot_at)", s.last_spot_at), + updated_at: fragment("EXCLUDED.updated_at") + ] + ] + ), + conflict_target: [:hour_utc, :band, :sender_grid, :receiver_grid] + ) + + count + end +end diff --git a/lib/microwaveprop/pskr/client.ex b/lib/microwaveprop/pskr/client.ex new file mode 100644 index 00000000..670a4c0c --- /dev/null +++ b/lib/microwaveprop/pskr/client.ex @@ -0,0 +1,204 @@ +defmodule Microwaveprop.Pskr.Client do + @moduledoc """ + MQTT client for the PSK Reporter live spot firehose. + + Connects to `mqtt.pskreporter.info:1883` (plain TCP — the broker + doesn't expose TLS), subscribes to a curated set of topics, and + forwards every payload to `Pskr.Aggregator.ingest/2`. Each + topic anchors on USA (DXCC 291) on either the sender or receiver + side so the broker pre-filters out the bulk of the firehose + before it ever reaches us — the alternative would be subscribing + to `pskr/filter/v2/#` and dropping ~95% client-side. + + ## Cluster singleton + + All replicas start this GenServer; **only one across the cluster + actually connects to MQTT**. We elect a leader via + `:global.register_name/2` — atomic across the BEAM cluster + (libcluster gives us connectivity). The losing nodes stay in + `:standby` mode: no MQTT connection, no DB writes, just a + cheap idle process watching for the leader to disappear. + + When the leader's node goes down (deploy, OOM, network + partition), `:global` evicts its registration cluster-wide and + emits `{:nodedown, node}` to every subscriber. Standby clients + race to re-register; one wins and starts a fresh connection + inside `handle_continue/2`. The handover loses at most one flush + cycle (~60 s) of in-flight aggregates from the old leader's + Aggregator — acceptable, since the upsert-on-conflict logic + merges any overlapping path-hours additively. + + ## Bands + + Subscribes to **VHF and up** by default (6m, 2m, 70cm, 23cm). HF + spots are dominated by ionospheric propagation, which isn't what + this project models — we deliberately skip HF rather than waste + bandwidth. Microwave bands are listed too but spots there are + vanishingly rare on PSK Reporter (FT8 activity is mostly + 144 MHz / 432 MHz / 1296 MHz). + + ## Off in dev/test + + `:microwaveprop, :pskr_mqtt_enabled` defaults to `false` in + dev/test so `mix test` doesn't open an outbound connection. Prod + flips it on via runtime.exs. + """ + use GenServer + + alias Microwaveprop.Pskr.Aggregator + + require Logger + + @broker_host ~c"mqtt.pskreporter.info" + @broker_port 1883 + + # USA = 291 (continental). PSK Reporter splits Alaska (6) and + # Hawaii (110) as separate DXCCs but neither is currently in the + # HRRR domain; HRDPS will eventually add Canada (1). + @us_dxcc 291 + + @default_bands ~w(6m 2m 70cm 23cm 13cm 9cm 6cm 3cm 1.25cm) + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts \\ []) do + GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__)) + end + + @impl true + def init(opts) do + Process.flag(:trap_exit, true) + # `:net_kernel.monitor_nodes/1` lands `{:nodeup, node}` / + # `{:nodedown, node}` messages in our mailbox — that's what + # tells a standby client the leader vanished and election + # should re-run. + :ok = :net_kernel.monitor_nodes(true) + + bands = Keyword.get(opts, :bands, @default_bands) + aggregator = Keyword.get(opts, :aggregator, Aggregator) + client_id = Keyword.get(opts, :client_id, default_client_id()) + + state = %{ + role: :standby, + pid: nil, + bands: bands, + aggregator: aggregator, + client_id: client_id, + received: 0 + } + + {:ok, state, {:continue, :elect}} + end + + @impl true + def handle_continue(:elect, state) do + case :global.register_name(__MODULE__, self()) do + :yes -> + Logger.info("Pskr.Client elected leader on #{node()} — connecting to MQTT") + {:noreply, %{state | role: :active}, {:continue, :connect}} + + :no -> + leader = :global.whereis_name(__MODULE__) + Logger.info("Pskr.Client standby on #{node()} (leader: #{inspect(leader)})") + {:noreply, state} + end + end + + def handle_continue(:connect, state) do + case connect_and_subscribe(state) do + {:ok, pid} -> + Logger.info("Pskr.Client connected as #{state.client_id}, subscribed to #{length(state.bands)} bands") + {:noreply, %{state | pid: pid}} + + {:error, reason} -> + Logger.error("Pskr.Client connect failed: #{inspect(reason)} — retrying in 30s") + Process.send_after(self(), :reconnect, 30_000) + {:noreply, state} + end + end + + @impl true + def handle_info(:reconnect, state), do: {:noreply, state, {:continue, :connect}} + + # When any node joins or leaves, standby clients re-attempt + # leader election. The active leader is unaffected — its + # registration is still valid until *its* node drops out. + def handle_info({event, _node}, %{role: :standby} = state) when event in [:nodeup, :nodedown] do + {:noreply, state, {:continue, :elect}} + end + + def handle_info({event, _node}, state) when event in [:nodeup, :nodedown], do: {:noreply, state} + + # `:emqtt` delivers PUBLISH frames to the owner process as + # `{:publish, %{topic: …, payload: …, qos: …}}`. + def handle_info({:publish, %{payload: payload}}, state) do + Aggregator.ingest(state.aggregator, payload) + {:noreply, %{state | received: state.received + 1}} + end + + def handle_info({:disconnected, reason_code, _props}, state) do + Logger.warning("Pskr.Client disconnected: #{inspect(reason_code)} (received #{state.received} so far)") + Process.send_after(self(), :reconnect, 5_000) + {:noreply, %{state | pid: nil}} + end + + def handle_info({:EXIT, pid, reason}, %{pid: pid} = state) do + Logger.warning("Pskr.Client emqtt exited: #{inspect(reason)} — reconnecting in 5s") + Process.send_after(self(), :reconnect, 5_000) + {:noreply, %{state | pid: nil}} + end + + def handle_info(_other, state), do: {:noreply, state} + + defp connect_and_subscribe(state) do + opts = [ + host: @broker_host, + port: @broker_port, + clientid: state.client_id, + clean_start: true, + keepalive: 60, + owner: self() + ] + + with {:ok, pid} <- :emqtt.start_link(opts), + {:ok, _props} <- :emqtt.connect(pid), + :ok <- subscribe_all(pid, state.bands) do + {:ok, pid} + end + end + + defp subscribe_all(pid, bands) do + topics = + Enum.flat_map(bands, fn band -> + [ + {topic_for_band(band, :sender_us), 0}, + {topic_for_band(band, :receiver_us), 0} + ] + end) + + case :emqtt.subscribe(pid, %{}, topics) do + {:ok, _props, _reasons} -> :ok + other -> {:error, {:subscribe_failed, other}} + end + end + + # Topic order (from the broker's landing page): + # pskr/filter/v2//////// + # Anchor either sender or receiver country on USA so the broker + # delivers only the half of the firehose that touches our HRRR + # domain. + defp topic_for_band(band, :sender_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/#{@us_dxcc}/+" + + defp topic_for_band(band, :receiver_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/+/#{@us_dxcc}" + + defp default_client_id do + suffix = 4 |> :crypto.strong_rand_bytes() |> Base.encode16(case: :lower) + "microwaveprop-#{node_short()}-#{suffix}" + end + + defp node_short do + case System.get_env("POD_IP") do + nil -> node() |> to_string() |> String.replace(["@", "."], "-") + ip -> String.replace(ip, ".", "-") + end + end +end diff --git a/lib/microwaveprop/pskr/spot_hourly.ex b/lib/microwaveprop/pskr/spot_hourly.ex new file mode 100644 index 00000000..e38c48df --- /dev/null +++ b/lib/microwaveprop/pskr/spot_hourly.ex @@ -0,0 +1,68 @@ +defmodule Microwaveprop.Pskr.SpotHourly do + @moduledoc """ + Hourly aggregate of PSK Reporter spots between two 4-character + Maidenhead grid squares on a single band. + + A single QSO often produces 50+ reception reports (one per + monitoring station that decoded it). Storing each report verbatim + would dwarf the rest of the database — instead `Pskr.Aggregator` + collapses every spot inside `(hour_utc, band, sender_grid, + receiver_grid)` into one row, retaining `spot_count`, SNR + envelope, and the set of modes observed. Downstream weather + correlation cares about *did propagation occur in this cell during + this hour*, not about each individual reporter — so the lossy + aggregation is the desired shape, not just a space optimization. + + `midpoint_lat/lon` is the great-circle midpoint of the path, + pre-computed at insert time so HRRR / IEMRE grid lookups are a + cheap `WHERE midpoint_lat BETWEEN …` rather than a per-row + recomputation. + """ + use Ecto.Schema + + import Ecto.Changeset + + @primary_key {:id, :binary_id, autogenerate: true} + @foreign_key_type :binary_id + + schema "pskr_spots_hourly" do + field :hour_utc, :utc_datetime + field :band, :string + field :sender_grid, :string + field :receiver_grid, :string + field :sender_country, :integer + field :receiver_country, :integer + field :sender_lat, :float + field :sender_lon, :float + field :receiver_lat, :float + field :receiver_lon, :float + field :midpoint_lat, :float + field :midpoint_lon, :float + field :distance_km, :float + field :spot_count, :integer, default: 0 + field :max_snr_db, :integer + field :min_snr_db, :integer + field :modes, {:array, :string}, default: [] + field :first_spot_at, :utc_datetime + field :last_spot_at, :utc_datetime + + timestamps(type: :utc_datetime) + end + + @type t :: %__MODULE__{} + + @cast_fields ~w(hour_utc band sender_grid receiver_grid sender_country receiver_country + sender_lat sender_lon receiver_lat receiver_lon midpoint_lat midpoint_lon + distance_km spot_count max_snr_db min_snr_db modes first_spot_at + last_spot_at)a + + @required_fields ~w(hour_utc band sender_grid receiver_grid spot_count)a + + @spec changeset(t(), map()) :: Ecto.Changeset.t() + def changeset(record, attrs) do + record + |> cast(attrs, @cast_fields) + |> validate_required(@required_fields) + |> unique_constraint([:hour_utc, :band, :sender_grid, :receiver_grid]) + end +end diff --git a/mix.exs b/mix.exs index 8ede37a6..958d6e31 100644 --- a/mix.exs +++ b/mix.exs @@ -113,6 +113,13 @@ defmodule Microwaveprop.MixProject do # PG2 adapter (in-process). Wires up via application.ex's # PubSub child spec, not a config block. {:phoenix_pubsub_redis, "~> 3.0"}, + # MQTT client for the PSK Reporter live spot firehose + # (mqtt.pskreporter.info:1883). Used by `Microwaveprop.Pskr.Client` + # to ingest geolocated, timestamped reception reports — the + # ground-truth signal we correlate against HRRR atmospheric + # state. Disabled in dev/test by default; enabled in prod via + # `PSKR_MQTT_ENABLED=true`. + {:emqtt, "~> 1.13"}, {:ecto_psql_extras, "~> 0.8"}, {:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false}, {:live_table, "~> 0.4.1"}, diff --git a/mix.lock b/mix.lock index 19a88199..03edd47a 100644 --- a/mix.lock +++ b/mix.lock @@ -8,6 +8,7 @@ "circular_buffer": {:hex, :circular_buffer, "1.0.0", "25c004da0cba7bd8bc1bdabded4f9a902d095e20600fd15faf1f2ffbaea18a07", [:mix], [], "hexpm", "c829ec31c13c7bafd1f546677263dff5bfb006e929f25635878ac3cfba8749e5"}, "comeonin": {:hex, :comeonin, "5.5.1", "5113e5f3800799787de08a6e0db307133850e635d34e9fab23c70b6501669510", [:mix], [], "hexpm", "65aac8f19938145377cee73973f192c5645873dcf550a8a6b18187d17c13ccdb"}, "complex": {:hex, :complex, "0.6.0", "b0130086a7a8c33574d293b2e0e250f4685580418eac52a5658a4bd148f3ccf1", [:mix], [], "hexpm", "0a5fa95580dcaf30fcd60fe1aaf24327c0fe401e98c24d892e172e79498269f9"}, + "cowlib": {:hex, :cowlib, "2.16.0", "54592074ebbbb92ee4746c8a8846e5605052f29309d3a873468d76cdf932076f", [:make, :rebar3], [], "hexpm", "7f478d80d66b747344f0ea7708c187645cfcc08b11aa424632f78e25bf05db51"}, "credo": {:hex, :credo, "1.7.18", "5c5596bf7aedf9c8c227f13272ac499fe8eae6237bd326f2f07dfc173786f042", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "a189d164685fd945809e862fe76a7420c4398fa288d76257662aecb909d6b3e5"}, "db_connection": {:hex, :db_connection, "2.10.0", "8ff756471e41765bd5563b633f73e9a94bbc138816e8644bb17d0d91bf260a95", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "02cdd01b45efb1b550e68edbbea41be32de9b24bb07e1ea0e9cbc522ac377e54"}, "decimal": {:hex, :decimal, "2.3.0", "3ad6255aa77b4a3c4f818171b12d237500e63525c2fd056699967a3e7ea20f62", [:mix], [], "hexpm", "a4d66355cb29cb47c3cf30e71329e58361cfcb37c34235ef3bf1d7bf3773aeac"}, @@ -16,6 +17,7 @@ "ecto_psql_extras": {:hex, :ecto_psql_extras, "0.8.8", "aa02529c97f69aed5722899f5dc6360128735a92dd169f23c5d50b1f7fdede08", [:mix], [{:ecto_sql, "~> 3.7", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:postgrex, "> 0.16.0", [hex: :postgrex, repo: "hexpm", optional: false]}, {:table_rex, "~> 3.1.1 or ~> 4.0", [hex: :table_rex, repo: "hexpm", optional: false]}], "hexpm", "04c63d92b141723ad6fed2e60a4b461ca00b3594d16df47bbc48f1f4534f2c49"}, "ecto_sql": {:hex, :ecto_sql, "3.13.5", "2f8282b2ad97bf0f0d3217ea0a6fff320ead9e2f8770f810141189d182dc304e", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "aa36751f4e6a2b56ae79efb0e088042e010ff4935fc8684e74c23b1f49e25fdc"}, "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"}, + "emqtt": {:hex, :emqtt, "1.15.0", "20b75d79ea625ce31564f31af5e686e7b12252f05ae6680118587de0ce1cdc0e", [:rebar3], [{:cowlib, "~> 2.13", [hex: :cowlib, repo: "hexpm", optional: false]}, {:getopt, "1.0.3", [hex: :getopt, repo: "hexpm", optional: false]}, {:gun, "~> 2.1", [hex: :gun, repo: "hexpm", optional: false]}, {:quicer, "0.2.15", [hex: :quicer, repo: "hexpm", optional: false]}], "hexpm", "b1bc715a83eeb043407cc776deedeff2514b9ddf5b936937c0a04a0b1d05fe1e"}, "erlex": {:hex, :erlex, "0.2.8", "cd8116f20f3c0afe376d1e8d1f0ae2452337729f68be016ea544a72f767d9c12", [:mix], [], "hexpm", "9d66ff9fedf69e49dc3fd12831e12a8a37b76f8651dd21cd45fcf5561a8a7590"}, "esbuild": {:hex, :esbuild, "0.10.0", "b0aa3388a1c23e727c5a3e7427c932d89ee791746b0081bbe56103e9ef3d291f", [:mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "468489cda427b974a7cc9f03ace55368a83e1a7be12fba7e30969af78e5f8c70"}, "exla": {:hex, :exla, "0.11.0", "1428de9edcb297480a64611d3a72fcefe13c93c115bba6d38e910583c37e38c8", [:make, :mix], [{:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}, {:fine, "~> 0.1", [hex: :fine, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:nx, "~> 0.11.0", [hex: :nx, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:xla, "~> 0.10.0", [hex: :xla, repo: "hexpm", optional: false]}], "hexpm", "1067207c802bd6f28cded6a2664979ee2e25dddce95cb84be3f0a3ebfbab2c74"}, @@ -23,7 +25,9 @@ "finch": {:hex, :finch, "0.21.0", "b1c3b2d48af02d0c66d2a9ebfb5622be5c5ecd62937cf79a88a7f98d48a8290c", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.6.2 or ~> 1.7", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "87dc6e169794cb2570f75841a19da99cfde834249568f2a5b121b809588a4377"}, "fine": {:hex, :fine, "0.1.6", "4bf7151493443c454aac9f2fa2f34f5fefd0346a83fb5586a016c4a135c63247", [:mix], [], "hexpm", "5638eb4495488e885ebec167fa57973e5c35e1a50c344eb7666c90ec1c4e3b12"}, "gen_smtp": {:hex, :gen_smtp, "1.3.0", "62c3d91f0dcf6ce9db71bcb6881d7ad0d1d834c7f38c13fa8e952f4104a8442e", [:rebar3], [{:ranch, ">= 1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "0b73fbf069864ecbce02fe653b16d3f35fd889d0fdd4e14527675565c39d84e6"}, + "getopt": {:hex, :getopt, "1.0.3", "4f3320c1f6f26b2bec0f6c6446b943eb927a1e6428ea279a1c6c534906ee79f1", [:rebar3], [], "hexpm", "7e01de90ac540f21494ff72792b1e3162d399966ebbfc674b4ce52cb8f49324f"}, "glob_ex": {:hex, :glob_ex, "0.1.11", "cb50d3f1ef53f6ca04d6252c7fde09fd7a1cf63387714fe96f340a1349e62c93", [:mix], [], "hexpm", "342729363056e3145e61766b416769984c329e4378f1d558b63e341020525de4"}, + "gun": {:hex, :gun, "2.2.0", "b8f6b7d417e277d4c2b0dc3c07dfdf892447b087f1cc1caff9c0f556b884e33d", [:make, :rebar3], [{:cowlib, ">= 2.15.0 and < 3.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "76022700c64287feb4df93a1795cff6741b83fb37415c40c34c38d2a4645261a"}, "heroicons": {:git, "https://github.com/tailwindlabs/heroicons.git", "0435d4ca364a608cc75e2f8683d374e55abbae26", [tag: "v2.2.0", sparse: "optimized", depth: 1]}, "hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"}, "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"}, @@ -60,10 +64,12 @@ "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"}, "postgrex": {:hex, :postgrex, "0.22.0", "fb027b58b6eab1f6de5396a2abcdaaeb168f9ed4eccbb594e6ac393b02078cbd", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a68c4261e299597909e03e6f8ff5a13876f5caadaddd0d23af0d0a61afcc5d84"}, "prom_ex": {:hex, :prom_ex, "1.11.0", "1f6d67f2dead92224cb4f59beb3e4d319257c5728d9638b4a5e8ceb51a4f9c7e", [:mix], [{:absinthe, ">= 1.7.0", [hex: :absinthe, repo: "hexpm", optional: true]}, {:broadway, ">= 1.1.0", [hex: :broadway, repo: "hexpm", optional: true]}, {:ecto, ">= 3.11.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:finch, "~> 0.18", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:oban, ">= 2.10.0", [hex: :oban, repo: "hexpm", optional: true]}, {:octo_fetch, "~> 0.4", [hex: :octo_fetch, repo: "hexpm", optional: false]}, {:peep, "~> 3.0", [hex: :peep, repo: "hexpm", optional: false]}, {:phoenix, ">= 1.7.0", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, ">= 0.20.0", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, ">= 1.16.0", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 2.6.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:telemetry, ">= 1.0.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}, {:telemetry_metrics_prometheus_core, "~> 1.2", [hex: :telemetry_metrics_prometheus_core, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 1.1", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "76b074bc3730f0802978a7eb5c7091a65473eaaf07e99ec9e933138dcc327805"}, + "quicer": {:hex, :quicer, "0.2.15", "775fcfa09c9ce5a4a00d23c1c8b1e81cd361ebc847e43bb766efac4373cdb1b8", [:rebar3], [{:snabbkaffe, "1.0.10", [hex: :snabbkaffe, repo: "hexpm", optional: false]}], "hexpm", "52236d9384a541341706bcae438e81b209fa21d2045d3df317f8ba5beda6a8e5"}, "ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"}, "redix": {:hex, :redix, "1.5.3", "4eaae29c75e3285c0ff9957046b7c209aa7f72a023a17f0a9ea51c2a50ab5b0f", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:nimble_options, "~> 0.5.0 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7b06fb5246373af41f5826b03334dfa3f636347d4d5d98b4d455b699d425ae7e"}, "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, "rewrite": {:hex, :rewrite, "1.3.0", "67448ba7975690b35ba7e7f35717efcce317dbd5963cb0577aa7325c1923121a", [:mix], [{:glob_ex, "~> 0.1", [hex: :glob_ex, repo: "hexpm", optional: false]}, {:sourceror, "~> 1.0", [hex: :sourceror, repo: "hexpm", optional: false]}, {:text_diff, "~> 0.1", [hex: :text_diff, repo: "hexpm", optional: false]}], "hexpm", "d111ac7ff3a58a802ef4f193bbd1831e00a9c57b33276e5068e8390a212714a5"}, + "snabbkaffe": {:hex, :snabbkaffe, "1.0.10", "9be2f54f61fc6862391b666b2b5b76c3fa53598e2989a17cef1b48cf347a8a63", [:rebar3], [], "hexpm", "70a98df36ae756908d55b5770891d443d63c903833e3e87d544036e13d4fac26"}, "sourceror": {:hex, :sourceror, "1.12.0", "da354c5f35aad3cc1132f5d5b0d8437d865e2661c263260480bab51b5eedb437", [:mix], [], "hexpm", "755703683bd014ebcd5de9acc24b68fb874a660a568d1d63f8f98cd8a6ef9cd0"}, "spitfire": {:hex, :spitfire, "0.3.11", "79dfcb033762470de472c1c26ea2b4e3aca74700c685dbffd9a13466272c323d", [:mix], [], "hexpm", "eb6e2dadf63214e8bfe65ca9788cef2b03b01027365d78d3c0e3d9ebd3d5b7b4"}, "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"}, diff --git a/priv/repo/migrations/20260504135243_create_pskr_spots_hourly.exs b/priv/repo/migrations/20260504135243_create_pskr_spots_hourly.exs new file mode 100644 index 00000000..fe935de7 --- /dev/null +++ b/priv/repo/migrations/20260504135243_create_pskr_spots_hourly.exs @@ -0,0 +1,33 @@ +defmodule Microwaveprop.Repo.Migrations.CreatePskrSpotsHourly do + use Ecto.Migration + + def change do + create table(:pskr_spots_hourly, primary_key: false) do + add :id, :binary_id, primary_key: true + add :hour_utc, :utc_datetime, null: false + add :band, :string, null: false + add :sender_grid, :string, null: false + add :receiver_grid, :string, null: false + add :sender_country, :integer + add :receiver_country, :integer + add :sender_lat, :float + add :sender_lon, :float + add :receiver_lat, :float + add :receiver_lon, :float + add :midpoint_lat, :float + add :midpoint_lon, :float + add :distance_km, :float + add :spot_count, :integer, null: false, default: 0 + add :max_snr_db, :integer + add :min_snr_db, :integer + add :modes, {:array, :string}, null: false, default: [] + add :first_spot_at, :utc_datetime + add :last_spot_at, :utc_datetime + + timestamps(type: :utc_datetime) + end + + create unique_index(:pskr_spots_hourly, [:hour_utc, :band, :sender_grid, :receiver_grid]) + create index(:pskr_spots_hourly, [:hour_utc]) + end +end diff --git a/test/microwaveprop/pskr/aggregator_test.exs b/test/microwaveprop/pskr/aggregator_test.exs new file mode 100644 index 00000000..7661954b --- /dev/null +++ b/test/microwaveprop/pskr/aggregator_test.exs @@ -0,0 +1,79 @@ +defmodule Microwaveprop.Pskr.AggregatorTest do + use Microwaveprop.DataCase, async: true + + alias Ecto.Adapters.SQL.Sandbox + alias Microwaveprop.Pskr.Aggregator + alias Microwaveprop.Pskr.SpotHourly + alias Microwaveprop.Repo + + @sample %{ + "f" => 144_174_000, + "md" => "FT8", + "rp" => -5, + "t_tx" => 1_662_407_697, + "sc" => "K5ABC", + "sl" => "EM12kl", + "rc" => "K7XYZ", + "rl" => "DM43st", + "sa" => 291, + "ra" => 291, + "b" => "2m" + } + + setup do + pid = start_supervised!({Aggregator, name: :"agg_#{:erlang.unique_integer([:positive])}", flush_ms: 0}) + Sandbox.allow(Repo, self(), pid) + %{agg: pid} + end + + test "ingest collapses repeats on the same path-hour", %{agg: agg} do + payloads = [ + Jason.encode!(@sample), + Jason.encode!(%{@sample | "rp" => -3}), + Jason.encode!(%{@sample | "rp" => -12, "md" => "FT4"}) + ] + + Enum.each(payloads, &Aggregator.ingest(agg, &1)) + + assert Aggregator.flush(agg) == 1 + [row] = Repo.all(SpotHourly) + + assert row.spot_count == 3 + assert row.max_snr_db == -3 + assert row.min_snr_db == -12 + assert Enum.sort(row.modes) == ["FT4", "FT8"] + assert row.band == "2m" + assert row.sender_grid == "EM12" + assert row.receiver_grid == "DM43" + assert row.distance_km > 0 + end + + test "two paths in the same hour produce two rows", %{agg: agg} do + Aggregator.ingest(agg, Jason.encode!(@sample)) + Aggregator.ingest(agg, Jason.encode!(%{@sample | "sl" => "FN31pr"})) + + assert Aggregator.flush(agg) == 2 + assert Repo.aggregate(SpotHourly, :count) == 2 + end + + test "second flush merges into existing row via upsert", %{agg: agg} do + Aggregator.ingest(agg, Jason.encode!(@sample)) + assert Aggregator.flush(agg) == 1 + + Aggregator.ingest(agg, Jason.encode!(%{@sample | "rp" => 5})) + assert Aggregator.flush(agg) == 1 + + [row] = Repo.all(SpotHourly) + assert row.spot_count == 2 + assert row.max_snr_db == 5 + assert row.min_snr_db == -5 + end + + test "malformed payloads are dropped silently", %{agg: agg} do + Aggregator.ingest(agg, "not json") + Aggregator.ingest(agg, Jason.encode!(%{"oops" => true})) + Aggregator.ingest(agg, Jason.encode!(@sample)) + + assert Aggregator.flush(agg) == 1 + end +end diff --git a/test/microwaveprop/pskr_test.exs b/test/microwaveprop/pskr_test.exs new file mode 100644 index 00000000..864bec6c --- /dev/null +++ b/test/microwaveprop/pskr_test.exs @@ -0,0 +1,106 @@ +defmodule Microwaveprop.PskrTest do + use ExUnit.Case, async: true + + alias Microwaveprop.Pskr + + @sample %{ + "sq" => 30_142_870_791, + "f" => 144_174_000, + "md" => "FT8", + "rp" => -5, + "t" => 1_662_407_712, + "t_tx" => 1_662_407_697, + "sc" => "K5ABC", + "sl" => "EM12kl", + "rc" => "K7XYZ", + "rl" => "DM43st", + "sa" => 291, + "ra" => 291, + "b" => "2m" + } + + describe "parse_spot/1" do + test "extracts the fields needed for hourly aggregation" do + json = Jason.encode!(@sample) + + assert {:ok, spot} = Pskr.parse_spot(json) + + assert spot.band == "2m" + assert spot.mode == "FT8" + assert spot.snr_db == -5 + assert spot.sender_grid == "EM12" + assert spot.receiver_grid == "DM43" + assert spot.sender_country == 291 + assert spot.receiver_country == 291 + assert %DateTime{} = spot.transmit_time + assert DateTime.to_unix(spot.transmit_time) == 1_662_407_697 + end + + test "uses :t when :t_tx is missing" do + json = @sample |> Map.delete("t_tx") |> Jason.encode!() + + assert {:ok, spot} = Pskr.parse_spot(json) + assert DateTime.to_unix(spot.transmit_time) == 1_662_407_712 + end + + test "rejects payloads without grids" do + json = @sample |> Map.delete("sl") |> Jason.encode!() + assert {:error, _} = Pskr.parse_spot(json) + end + + test "rejects malformed JSON" do + assert {:error, _} = Pskr.parse_spot("not-json") + end + end + + describe "path_key/1" do + test "buckets to the top of the hour and 4-char grid" do + {:ok, spot} = Pskr.parse_spot(Jason.encode!(@sample)) + {hour, band, snd, rcv} = Pskr.path_key(spot) + + assert band == "2m" + assert snd == "EM12" + assert rcv == "DM43" + # 1_662_407_697 → 2022-09-05T19:54:57Z → bucket at 19:00:00Z + assert hour == ~U[2022-09-05 19:00:00Z] + end + end + + describe "merge_spot/2" do + test "initializes accumulator from a single spot" do + {:ok, spot} = Pskr.parse_spot(Jason.encode!(@sample)) + + acc = Pskr.merge_spot(nil, spot) + + assert acc.spot_count == 1 + assert acc.max_snr_db == -5 + assert acc.min_snr_db == -5 + assert acc.modes == ["FT8"] + assert acc.sender_grid == "EM12" + assert acc.receiver_grid == "DM43" + assert is_float(acc.distance_km) + assert acc.distance_km > 0 + # Midpoint should land between the two grid centers + assert acc.midpoint_lat > min(acc.sender_lat, acc.receiver_lat) + assert acc.midpoint_lat < max(acc.sender_lat, acc.receiver_lat) + end + + test "increments count and tracks SNR envelope across spots" do + {:ok, first} = Pskr.parse_spot(Jason.encode!(@sample)) + {:ok, louder} = Pskr.parse_spot(Jason.encode!(%{@sample | "rp" => -1, "md" => "FT4"})) + {:ok, quieter} = Pskr.parse_spot(Jason.encode!(%{@sample | "rp" => -18})) + + acc = + nil + |> Pskr.merge_spot(first) + |> Pskr.merge_spot(louder) + |> Pskr.merge_spot(quieter) + + assert acc.spot_count == 3 + assert acc.max_snr_db == -1 + assert acc.min_snr_db == -18 + assert Enum.sort(acc.modes) == ["FT4", "FT8"] + assert DateTime.compare(acc.first_spot_at, acc.last_spot_at) in [:lt, :eq] + end + end +end