feat(aprs): add Aprs context for read-only queries against aprs.me

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

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

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)