fix: extract callsigns from MQTT topic when JSON payload omits sc/rc
Some FT8/Q65 multi-decoder spots omit sender/receiver callsigns from the JSON payload, but the MQTT topic routing path always carries them in positions 5 (sc) and 6 (rc): pskr/filter/v2/<band>/<mode>/<sc>/<rc>/<sl>/<rl>/<sa>/<ra> Client now extracts callsigns from the topic and threads them through Aggregator.ingest → Pskr.parse_spot, where they act as fallback when the JSON sc/rc keys are nil or empty. JSON payload values still take precedence when present — the topic is only used as a safety net. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
af50631a2a
commit
ba17a41683
4 changed files with 80 additions and 11 deletions
|
|
@ -73,9 +73,14 @@ defmodule Microwaveprop.Pskr do
|
||||||
reported (4/6/8/10 chars) and uppercased for consistent storage.
|
reported (4/6/8/10 chars) and uppercased for consistent storage.
|
||||||
Returns `{:error, reason}` for any payload missing the fields we
|
Returns `{:error, reason}` for any payload missing the fields we
|
||||||
need — the caller should count and discard.
|
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()}
|
@spec parse_spot(binary(), keyword()) :: {:ok, spot()} | {:error, term()}
|
||||||
def parse_spot(payload) when is_binary(payload) do
|
def parse_spot(payload, opts \\ []) when is_binary(payload) do
|
||||||
with {:ok, json} <- Jason.decode(payload),
|
with {:ok, json} <- Jason.decode(payload),
|
||||||
{:ok, sender_grid} <- fetch_grid(json, "sl"),
|
{:ok, sender_grid} <- fetch_grid(json, "sl"),
|
||||||
{:ok, receiver_grid} <- fetch_grid(json, "rl"),
|
{:ok, receiver_grid} <- fetch_grid(json, "rl"),
|
||||||
|
|
@ -91,8 +96,12 @@ defmodule Microwaveprop.Pskr do
|
||||||
transmit_time: transmit_time,
|
transmit_time: transmit_time,
|
||||||
sender_grid: sender_grid,
|
sender_grid: sender_grid,
|
||||||
receiver_grid: receiver_grid,
|
receiver_grid: receiver_grid,
|
||||||
sender_callsign: normalize_callsign(Map.get(json, "sc")),
|
sender_callsign:
|
||||||
receiver_callsign: normalize_callsign(Map.get(json, "rc")),
|
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"),
|
sender_country: Map.get(json, "sa"),
|
||||||
receiver_country: Map.get(json, "ra")
|
receiver_country: Map.get(json, "ra")
|
||||||
}}
|
}}
|
||||||
|
|
|
||||||
|
|
@ -41,10 +41,15 @@ defmodule Microwaveprop.Pskr.Aggregator do
|
||||||
GenServer.start_link(__MODULE__, opts, name: name)
|
GenServer.start_link(__MODULE__, opts, name: name)
|
||||||
end
|
end
|
||||||
|
|
||||||
@doc "Push one raw MQTT payload (binary JSON) into the accumulator."
|
@doc """
|
||||||
@spec ingest(GenServer.server(), binary()) :: :ok
|
Push one raw MQTT payload (binary JSON) into the accumulator.
|
||||||
def ingest(server \\ __MODULE__, payload) when is_binary(payload) do
|
|
||||||
GenServer.cast(server, {:ingest, payload})
|
`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
|
end
|
||||||
|
|
||||||
@doc "Force a synchronous flush. Returns the number of rows written."
|
@doc "Force a synchronous flush. Returns the number of rows written."
|
||||||
|
|
@ -65,8 +70,13 @@ defmodule Microwaveprop.Pskr.Aggregator do
|
||||||
end
|
end
|
||||||
|
|
||||||
@impl true
|
@impl true
|
||||||
|
# Accept both old (2-tuple) and new (3-tuple with opts) ingest messages.
|
||||||
def handle_cast({:ingest, payload}, state) do
|
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} ->
|
{:ok, spot} ->
|
||||||
key = Pskr.path_key(spot)
|
key = Pskr.path_key(spot)
|
||||||
rows = Map.update(state.rows, key, Pskr.merge_spot(nil, spot), &Pskr.merge_spot(&1, spot))
|
rows = Map.update(state.rows, key, Pskr.merge_spot(nil, spot), &Pskr.merge_spot(&1, spot))
|
||||||
|
|
|
||||||
|
|
@ -248,12 +248,30 @@ defmodule Microwaveprop.Pskr.Client do
|
||||||
# etc. before the spots ever reach us.
|
# etc. before the spots ever reach us.
|
||||||
defp topic_for(band), do: "pskr/filter/v2/#{band}/+/+/+/+/+/#{@us_dxcc}/#{@us_dxcc}"
|
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/<band>/<mode>/<sc>/<rc>/<sl>/<rl>/<sa>/<ra>
|
||||||
|
# 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 ----------
|
# ---------- Inbound dispatch ----------
|
||||||
|
|
||||||
defp process_buffer(state) do
|
defp process_buffer(state) do
|
||||||
case Mqtt.parse(state.buffer) do
|
case Mqtt.parse(state.buffer) do
|
||||||
{:ok, {:publish, _topic, payload}, rest} ->
|
{:ok, {:publish, topic, payload}, rest} ->
|
||||||
Aggregator.ingest(state.aggregator, payload)
|
opts = topic_callsigns(topic)
|
||||||
|
Aggregator.ingest(state.aggregator, payload, opts)
|
||||||
process_buffer(%{state | buffer: rest, received: state.received + 1})
|
process_buffer(%{state | buffer: rest, received: state.received + 1})
|
||||||
|
|
||||||
{:ok, {:suback, _id, results}, rest} ->
|
{:ok, {:suback, _id, results}, rest} ->
|
||||||
|
|
|
||||||
|
|
@ -231,5 +231,37 @@ defmodule Microwaveprop.PskrTest do
|
||||||
assert acc.sender_callsigns == ["K5ABC"]
|
assert acc.sender_callsigns == ["K5ABC"]
|
||||||
assert acc.receiver_callsigns == ["K7XYZ"]
|
assert acc.receiver_callsigns == ["K7XYZ"]
|
||||||
end
|
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
|
||||||
end
|
end
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue