From 65e97bec6b90a0d5e3ed08c880a606fa6c6a4758 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 1 May 2026 12:47:35 -0500 Subject: [PATCH] feat(aprs): add Aprs context for read-only queries against aprs.me --- lib/microwaveprop/aprs.ex | 116 +++++++++++++++++ test/microwaveprop/aprs_test.exs | 214 +++++++++++++++++++++++++++++++ test/test_helper.exs | 5 +- 3 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 lib/microwaveprop/aprs.ex create mode 100644 test/microwaveprop/aprs_test.exs diff --git a/lib/microwaveprop/aprs.ex b/lib/microwaveprop/aprs.ex new file mode 100644 index 00000000..0a572c53 --- /dev/null +++ b/lib/microwaveprop/aprs.ex @@ -0,0 +1,116 @@ +defmodule Microwaveprop.Aprs do + @moduledoc """ + Read-only access to aprs.me's `packets` table for 144 MHz calibration. + + This module never writes, never schemas, never mirrors. Each call + issues a SELECT against `Microwaveprop.AprsRepo` (which connects to + aprs.me's database) and returns raw maps for downstream parsing by + `Microwaveprop.Aprs.PathParser` and consumption by the + `Calibrate.Aprs144` mix task. + + aprs.me retains 24 h of packet history in production + (`PACKET_RETENTION_DAYS=1`), so any "historical" query is bounded by + that window. + + Tests run against a local `aprsme_test` database whose schema mirrors + aprs.me's partitioned `packets` table; rows are inserted via raw SQL + in a setup block under the `Ecto.Adapters.SQL.Sandbox`. + """ + + alias Ecto.Adapters.SQL + alias Microwaveprop.AprsRepo + + @type packet_row :: %{ + id: binary(), + sender: String.t(), + base_callsign: String.t(), + lat: float(), + lon: float(), + path: String.t(), + received_at: DateTime.t() + } + + @type position :: {lat :: float(), lon :: float(), last_heard_at :: DateTime.t()} + + @recent_packets_sql """ + SELECT id, sender, base_callsign, lat, lon, path, received_at + FROM packets + WHERE has_position = true + AND lat IS NOT NULL + AND lon IS NOT NULL + AND path IS NOT NULL + AND path <> '' + AND is_item = false + AND is_object = false + AND sender IS NOT NULL + AND received_at >= $1 + ORDER BY received_at ASC + LIMIT $2 + """ + + @station_positions_sql """ + SELECT DISTINCT ON (base_callsign) base_callsign, lat, lon, received_at + FROM packets + WHERE base_callsign = ANY($1) + AND has_position = true + AND lat IS NOT NULL + AND lon IS NOT NULL + ORDER BY base_callsign, received_at DESC + """ + + @doc """ + Returns recent position-bearing packets with non-empty paths, oldest first. + + Filters out item/object packets and rows missing position or path. + + ## Options + + * `:since` — `%DateTime{}`. Default `now - 1h`. + * `:limit` — non-neg integer. Default `50_000`. + """ + @spec recent_packets_with_paths(keyword()) :: [packet_row()] + def recent_packets_with_paths(opts \\ []) do + since = Keyword.get_lazy(opts, :since, fn -> DateTime.add(DateTime.utc_now(), -3600, :second) end) + limit = Keyword.get(opts, :limit, 50_000) + + %Postgrex.Result{rows: rows} = + SQL.query!(AprsRepo, @recent_packets_sql, [DateTime.to_naive(since), limit]) + + Enum.map(rows, &decode_packet_row/1) + end + + @doc """ + Returns the most recent known position for each callsign in `callsigns`. + + Callsigns with no positioned packets simply don't appear in the result. + """ + @spec station_positions([String.t()]) :: %{String.t() => position()} + def station_positions([]), do: %{} + + def station_positions(callsigns) when is_list(callsigns) do + %Postgrex.Result{rows: rows} = SQL.query!(AprsRepo, @station_positions_sql, [callsigns]) + + Map.new(rows, fn [base_callsign, lat, lon, received_at] -> + {base_callsign, {to_float(lat), to_float(lon), to_utc_datetime(received_at)}} + end) + end + + defp decode_packet_row([id, sender, base_callsign, lat, lon, path, received_at]) do + %{ + id: id, + sender: sender, + base_callsign: base_callsign, + lat: to_float(lat), + lon: to_float(lon), + path: path || "", + received_at: to_utc_datetime(received_at) + } + end + + defp to_float(%Decimal{} = d), do: Decimal.to_float(d) + defp to_float(n) when is_float(n), do: n + defp to_float(n) when is_integer(n), do: n * 1.0 + + defp to_utc_datetime(%NaiveDateTime{} = naive), do: DateTime.from_naive!(naive, "Etc/UTC") + defp to_utc_datetime(%DateTime{} = dt), do: dt +end diff --git a/test/microwaveprop/aprs_test.exs b/test/microwaveprop/aprs_test.exs new file mode 100644 index 00000000..583ab7c5 --- /dev/null +++ b/test/microwaveprop/aprs_test.exs @@ -0,0 +1,214 @@ +defmodule Microwaveprop.AprsTest do + use ExUnit.Case, async: false + + alias Ecto.Adapters.SQL + alias Ecto.Adapters.SQL.Sandbox + alias Microwaveprop.Aprs + alias Microwaveprop.AprsRepo + + setup do + pid = Sandbox.start_owner!(AprsRepo, shared: true) + on_exit(fn -> Sandbox.stop_owner(pid) end) + :ok + end + + defp insert_packet(opts) do + fields = + Keyword.merge( + [ + sender: "TEST-A", + base_callsign: "TEST-A", + lat: 33.0, + lon: -97.0, + path: "WIDE1*", + received_at: NaiveDateTime.utc_now(), + has_position: true, + is_item: false, + is_object: false + ], + opts + ) + + SQL.query!( + AprsRepo, + """ + INSERT INTO packets + (id, sender, base_callsign, lat, lon, path, received_at, + has_position, is_item, is_object, inserted_at, updated_at) + VALUES + (gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, now(), now()) + """, + [ + fields[:sender], + fields[:base_callsign], + fields[:lat], + fields[:lon], + fields[:path], + fields[:received_at], + fields[:has_position], + fields[:is_item], + fields[:is_object] + ] + ) + + :ok + end + + describe "recent_packets_with_paths/1" do + test "filters out is_item, is_object, has_position=false, empty path, and null lat/lon" do + now = NaiveDateTime.utc_now() + since = DateTime.add(DateTime.utc_now(), -3600, :second) + + # 1 valid row + :ok = insert_packet(sender: "TEST-VALID", path: "K5GVL-10*,WIDE1-1", received_at: now) + + # 7 rejected rows + :ok = insert_packet(sender: "TEST-ITEM", is_item: true, received_at: now) + :ok = insert_packet(sender: "TEST-OBJECT", is_object: true, received_at: now) + :ok = insert_packet(sender: "TEST-NOPOS", has_position: false, received_at: now) + :ok = insert_packet(sender: "TEST-EMPTYPATH", path: "", received_at: now) + :ok = insert_packet(sender: "TEST-NULLPATH", path: nil, received_at: now) + :ok = insert_packet(sender: "TEST-NULLLAT", lat: nil, received_at: now) + :ok = insert_packet(sender: "TEST-NULLLON", lon: nil, received_at: now) + + rows = Aprs.recent_packets_with_paths(since: since) + + assert length(rows) == 1 + assert hd(rows).sender == "TEST-VALID" + end + + test "honors :since" do + now = DateTime.utc_now() + recent = NaiveDateTime.add(NaiveDateTime.utc_now(), -1800, :second) + old = NaiveDateTime.add(NaiveDateTime.utc_now(), -5400, :second) + + :ok = insert_packet(sender: "TEST-RECENT", received_at: recent) + :ok = insert_packet(sender: "TEST-OLD", received_at: old) + + since = DateTime.add(now, -3600, :second) + rows = Aprs.recent_packets_with_paths(since: since) + + senders = Enum.map(rows, & &1.sender) + assert "TEST-RECENT" in senders + refute "TEST-OLD" in senders + end + + test "honors :limit" do + since = DateTime.add(DateTime.utc_now(), -3600, :second) + + Enum.each(1..5, fn i -> + # Stagger by seconds so ordering is deterministic. + ts = NaiveDateTime.add(NaiveDateTime.utc_now(), -i, :second) + :ok = insert_packet(sender: "TEST-L#{i}", received_at: ts) + end) + + rows = Aprs.recent_packets_with_paths(since: since, limit: 2) + assert length(rows) == 2 + end + + test "returns rows in received_at ascending order" do + since = DateTime.add(DateTime.utc_now(), -3600, :second) + base = NaiveDateTime.utc_now() + + :ok = insert_packet(sender: "TEST-MID", received_at: NaiveDateTime.add(base, -120, :second)) + :ok = insert_packet(sender: "TEST-OLDEST", received_at: NaiveDateTime.add(base, -300, :second)) + :ok = insert_packet(sender: "TEST-NEWEST", received_at: NaiveDateTime.add(base, -10, :second)) + + rows = + since + |> then(&Aprs.recent_packets_with_paths(since: &1)) + |> Enum.filter(&String.starts_with?(&1.sender, "TEST-")) + + senders = Enum.map(rows, & &1.sender) + assert senders == ["TEST-OLDEST", "TEST-MID", "TEST-NEWEST"] + end + + test "decodes Decimal lat/lon as floats" do + since = DateTime.add(DateTime.utc_now(), -3600, :second) + :ok = insert_packet(sender: "TEST-FLOAT", lat: 33.123, lon: -97.456) + + rows = Aprs.recent_packets_with_paths(since: since) + row = Enum.find(rows, &(&1.sender == "TEST-FLOAT")) + + assert is_float(row.lat) + assert is_float(row.lon) + assert_in_delta row.lat, 33.123, 0.0001 + assert_in_delta row.lon, -97.456, 0.0001 + end + + test "decodes received_at as a UTC DateTime" do + since = DateTime.add(DateTime.utc_now(), -3600, :second) + :ok = insert_packet(sender: "TEST-TZ") + + rows = Aprs.recent_packets_with_paths(since: since) + row = Enum.find(rows, &(&1.sender == "TEST-TZ")) + + assert %DateTime{} = row.received_at + assert row.received_at.time_zone == "Etc/UTC" + end + end + + describe "station_positions/1" do + test "returns the most recent fix per callsign and omits unknown callsigns" do + base = NaiveDateTime.utc_now() + + :ok = + insert_packet( + sender: "TEST-A", + base_callsign: "TEST-A", + lat: 30.0, + lon: -90.0, + received_at: NaiveDateTime.add(base, -3600, :second) + ) + + :ok = + insert_packet( + sender: "TEST-A", + base_callsign: "TEST-A", + lat: 31.0, + lon: -91.0, + received_at: NaiveDateTime.add(base, -1800, :second) + ) + + :ok = + insert_packet( + sender: "TEST-A", + base_callsign: "TEST-A", + lat: 32.5, + lon: -92.5, + received_at: NaiveDateTime.add(base, -60, :second) + ) + + :ok = + insert_packet( + sender: "TEST-B", + base_callsign: "TEST-B", + lat: 40.0, + lon: -100.0, + received_at: NaiveDateTime.add(base, -300, :second) + ) + + result = Aprs.station_positions(["TEST-A", "TEST-B", "TEST-C"]) + + assert result |> Map.keys() |> Enum.sort() == ["TEST-A", "TEST-B"] + + {lat_a, lon_a, heard_a} = result["TEST-A"] + assert_in_delta lat_a, 32.5, 0.0001 + assert_in_delta lon_a, -92.5, 0.0001 + assert %DateTime{} = heard_a + assert heard_a.time_zone == "Etc/UTC" + + {lat_b, lon_b, _heard_b} = result["TEST-B"] + assert_in_delta lat_b, 40.0, 0.0001 + assert_in_delta lon_b, -100.0, 0.0001 + + refute Map.has_key?(result, "TEST-C") + end + + test "empty list short-circuits to %{} without querying" do + # Stop the sandbox owner so a real query would crash with NoConnectionError. + Sandbox.checkin(AprsRepo) + assert Aprs.station_positions([]) == %{} + end + end +end diff --git a/test/test_helper.exs b/test/test_helper.exs index 718d7007..d453a0fa 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -1,2 +1,5 @@ +alias Ecto.Adapters.SQL.Sandbox + ExUnit.start(exclude: [:slow], capture_log: true) -Ecto.Adapters.SQL.Sandbox.mode(Microwaveprop.Repo, :manual) +Sandbox.mode(Microwaveprop.Repo, :manual) +Sandbox.mode(Microwaveprop.AprsRepo, :manual)