diff --git a/lib/microwaveprop/pskr.ex b/lib/microwaveprop/pskr.ex index 8e3bb0bb..8852df23 100644 --- a/lib/microwaveprop/pskr.ex +++ b/lib/microwaveprop/pskr.ex @@ -73,9 +73,14 @@ defmodule Microwaveprop.Pskr do 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. + + Callsigns are taken from the JSON payload (`sc`/`rc` keys). + When those keys are absent or empty (common for FT8/Q65 multi- + decoder spots), falls back to callsigns extracted from the MQTT + topic which always carries them in the routing path. """ - @spec parse_spot(binary()) :: {:ok, spot()} | {:error, term()} - def parse_spot(payload) when is_binary(payload) do + @spec parse_spot(binary(), keyword()) :: {:ok, spot()} | {:error, term()} + def parse_spot(payload, opts \\ []) 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"), @@ -91,8 +96,12 @@ defmodule Microwaveprop.Pskr do transmit_time: transmit_time, sender_grid: sender_grid, receiver_grid: receiver_grid, - sender_callsign: normalize_callsign(Map.get(json, "sc")), - receiver_callsign: normalize_callsign(Map.get(json, "rc")), + sender_callsign: + normalize_callsign(Map.get(json, "sc")) || + normalize_callsign(Keyword.get(opts, :sender_callsign)), + receiver_callsign: + normalize_callsign(Map.get(json, "rc")) || + normalize_callsign(Keyword.get(opts, :receiver_callsign)), sender_country: Map.get(json, "sa"), receiver_country: Map.get(json, "ra") }} diff --git a/lib/microwaveprop/pskr/aggregator.ex b/lib/microwaveprop/pskr/aggregator.ex index 2f550e63..b0ee00c4 100644 --- a/lib/microwaveprop/pskr/aggregator.ex +++ b/lib/microwaveprop/pskr/aggregator.ex @@ -41,10 +41,15 @@ defmodule Microwaveprop.Pskr.Aggregator do 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}) + @doc """ + Push one raw MQTT payload (binary JSON) into the accumulator. + + `opts` is forwarded to `Pskr.parse_spot/2` for topic-derived + fallback values (e.g. callsigns when the JSON payload omits them). + """ + @spec ingest(GenServer.server(), binary(), keyword()) :: :ok + def ingest(server \\ __MODULE__, payload, opts \\ []) when is_binary(payload) do + GenServer.cast(server, {:ingest, payload, opts}) end @doc "Force a synchronous flush. Returns the number of rows written." @@ -65,8 +70,13 @@ defmodule Microwaveprop.Pskr.Aggregator do end @impl true + # Accept both old (2-tuple) and new (3-tuple with opts) ingest messages. def handle_cast({:ingest, payload}, state) do - case Pskr.parse_spot(payload) do + handle_cast({:ingest, payload, []}, state) + end + + def handle_cast({:ingest, payload, opts}, state) do + case Pskr.parse_spot(payload, opts) do {:ok, spot} -> key = Pskr.path_key(spot) rows = Map.update(state.rows, key, Pskr.merge_spot(nil, spot), &Pskr.merge_spot(&1, spot)) diff --git a/lib/microwaveprop/pskr/client.ex b/lib/microwaveprop/pskr/client.ex index c3549b6b..07e5ee0a 100644 --- a/lib/microwaveprop/pskr/client.ex +++ b/lib/microwaveprop/pskr/client.ex @@ -248,12 +248,30 @@ defmodule Microwaveprop.Pskr.Client do # etc. before the spots ever reach us. defp topic_for(band), do: "pskr/filter/v2/#{band}/+/+/+/+/+/#{@us_dxcc}/#{@us_dxcc}" + # Extract sender/receiver callsigns from a received topic. + # Topic format: pskr/filter/v2//////// + # Some FT8/Q65 multi-decoder spots omit `sc`/`rc` from the JSON + # payload; the topic always carries them in the routing path. + defp topic_callsigns(topic) do + case String.split(topic, "/") do + ["pskr", "filter", "v2", _band, _mode, sc, rc | _rest] -> + opts = [] + opts = if sc != "+" and sc != "", do: Keyword.put(opts, :sender_callsign, sc), else: opts + opts = if rc != "+" and rc != "", do: Keyword.put(opts, :receiver_callsign, rc), else: opts + opts + + _ -> + [] + end + end + # ---------- Inbound dispatch ---------- defp process_buffer(state) do case Mqtt.parse(state.buffer) do - {:ok, {:publish, _topic, payload}, rest} -> - Aggregator.ingest(state.aggregator, payload) + {:ok, {:publish, topic, payload}, rest} -> + opts = topic_callsigns(topic) + Aggregator.ingest(state.aggregator, payload, opts) process_buffer(%{state | buffer: rest, received: state.received + 1}) {:ok, {:suback, _id, results}, rest} -> diff --git a/test/microwaveprop/pskr_test.exs b/test/microwaveprop/pskr_test.exs index 767a7c0a..df06803b 100644 --- a/test/microwaveprop/pskr_test.exs +++ b/test/microwaveprop/pskr_test.exs @@ -231,5 +231,37 @@ defmodule Microwaveprop.PskrTest do assert acc.sender_callsigns == ["K5ABC"] assert acc.receiver_callsigns == ["K7XYZ"] end + + test "falls back to topic-derived callsign when JSON payload omits sc/rc" do + # FT8/Q65 multi-decoder spots sometimes lack sc/rc in the JSON + # payload, but the MQTT topic always carries them. + payload = @sample |> Map.drop(["sc", "rc"]) |> Jason.encode!() + opts = [sender_callsign: "K5TOPIC", receiver_callsign: "K7TOPIC"] + + {:ok, spot} = Pskr.parse_spot(payload, opts) + + assert spot.sender_callsign == "K5TOPIC" + assert spot.receiver_callsign == "K7TOPIC" + end + + test "prefers JSON callsign over topic fallback when both present" do + payload = Jason.encode!(@sample) + opts = [sender_callsign: "K5TOPIC", receiver_callsign: "K7TOPIC"] + + {:ok, spot} = Pskr.parse_spot(payload, opts) + + assert spot.sender_callsign == "K5ABC" + assert spot.receiver_callsign == "K7XYZ" + end + + test "topic fallback with nil/empty JSON callsign prefers topic" do + payload = @sample |> Map.put("sc", "") |> Jason.encode!() + opts = [sender_callsign: "K5TOPIC"] + + {:ok, spot} = Pskr.parse_spot(payload, opts) + + assert spot.sender_callsign == "K5TOPIC" + assert spot.receiver_callsign == "K7XYZ" + end end end