334 lines
12 KiB
Go
334 lines
12 KiB
Go
// Package pskr subscribes to PSK Reporter's MQTT feed and turns it into the
|
||
// spots the band-opening detector already eats.
|
||
//
|
||
// Why this exists at all: the detector was fed from the DX cluster and the RBN,
|
||
// and on VHF that is a few hundred skimmers, nearly all of them on HF. A 6 m
|
||
// opening carrying 869 stations reached OpsLog as a handful of spots, or none.
|
||
// PSK Reporter is every ordinary station running WSJT-X and reporting what it
|
||
// decodes — the difference is two orders of magnitude, not a threshold.
|
||
//
|
||
// The feed's shape happens to suit us exactly:
|
||
//
|
||
// topic pskr/filter/v2/<band>/<mode>/<tx call>/<rx call>/...
|
||
// payload {"f":50313000,"md":"FT8","rp":-12,"sc":"F4BPO","sl":"JN36",
|
||
// "rc":"OH5CX","rl":"KP30","b":"6m"}
|
||
//
|
||
// BOTH grids are in the message, so distance and bearing are arithmetic. No
|
||
// lookup, no DXCC-centre approximation, no extra network call — which is what
|
||
// made the cluster path's bearings coarse.
|
||
//
|
||
// Volume is the real design constraint. Six metres open is thousands of
|
||
// messages a minute, and OpsLog runs on some very old PCs. So: no history is
|
||
// kept here, nothing is persisted, each message is parsed and handed on or
|
||
// dropped, and the callback does the deciding.
|
||
package pskr
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||
)
|
||
|
||
// DefaultBroker is PSK Reporter's public MQTT endpoint, TLS.
|
||
const DefaultBroker = "tls://mqtt.pskreporter.info:1884"
|
||
|
||
// DefaultNearKm is the radius that counts as "around here" for a receiver.
|
||
const DefaultNearKm = 300
|
||
|
||
// Bands offered. Anything below 10 m is deliberately absent, 12 m included: an
|
||
// "opening" on 20 m is the normal state of the band and announcing it says
|
||
// nothing, and 12 m is close enough to that at this point in the cycle to be the
|
||
// same problem. These are the bands where an opening is an event.
|
||
//
|
||
// Fewer subscriptions is also less traffic, which matters here: the operator's
|
||
// PC pays for every message the broker sends, whether or not anything comes of it.
|
||
var Bands = []string{"10m", "6m", "4m", "2m"}
|
||
|
||
// Spot is one decode, already reduced to what a detector needs.
|
||
type Spot struct {
|
||
Call string // the transmitting station
|
||
Band string
|
||
Mode string
|
||
Grid string // transmitter's grid, 4 characters
|
||
// FreqHz is where the decode happened. The band alone is enough for an
|
||
// opening, but a station worth chasing has to be tuned to.
|
||
FreqHz int64
|
||
DistKm int // from the operator
|
||
Bearing int // degrees from the operator, short path
|
||
At time.Time
|
||
}
|
||
|
||
// Config is what the watcher needs to run.
|
||
type Config struct {
|
||
Broker string
|
||
Bands []string
|
||
// OpLat/OpLon are the operator's position: every spot is measured from it,
|
||
// so with no position there is nothing to measure and the watcher stays down.
|
||
OpLat, OpLon float64
|
||
// NearKm is how close a RECEIVER must be to count as "around here". A report
|
||
// collected further away proves a path that is not the operator's.
|
||
//
|
||
// 300 km by default: far enough to borrow the ears of a whole region, which
|
||
// is the point — an opening reaches an area, not a postcode, and waiting for
|
||
// a decode at one's own station is just working the band. Close enough that
|
||
// the ionosphere doing something there is the ionosphere doing it here.
|
||
NearKm int
|
||
// Geo turns two grids into distance and bearing. Injected rather than
|
||
// implemented here so it stays the SAME arithmetic the cluster path uses —
|
||
// two answers for one question is how a bearing quietly becomes wrong.
|
||
Geo func(grid string) (distKm int, bearing int, ok bool)
|
||
// OnSpot receives every accepted decode. Called from the MQTT goroutine, so
|
||
// it must not block: the broker's buffer is what pays for it if it does.
|
||
OnSpot func(Spot)
|
||
// OnGrid receives the transmitter of EVERY message, before any geographic
|
||
// filtering, for the callsign-to-locator store. Same goroutine as OnSpot and
|
||
// the same rule: do not block.
|
||
OnGrid func(call, grid string)
|
||
// RxGrids filters at the BROKER: only reports collected by a receiver in one
|
||
// of these squares are sent at all. Empty keeps the old behaviour, which was
|
||
// to receive the world and discard it here — measured at 83 messages a second
|
||
// for the four opening bands, of which about one in a hundred survived.
|
||
RxGrids []string
|
||
Logf func(string, ...any)
|
||
}
|
||
|
||
// Watcher owns the MQTT connection and its subscriptions.
|
||
type Watcher struct {
|
||
mu sync.Mutex
|
||
cfg Config
|
||
client mqtt.Client
|
||
running bool
|
||
|
||
// received counts accepted spots since start, for the status panel: a
|
||
// connection that is up but silent looks identical to one that is working
|
||
// until you can see a number moving.
|
||
received uint64
|
||
lastAt time.Time
|
||
lastErr string
|
||
}
|
||
|
||
func New(cfg Config) *Watcher {
|
||
if cfg.Broker == "" {
|
||
cfg.Broker = DefaultBroker
|
||
}
|
||
if len(cfg.Bands) == 0 {
|
||
cfg.Bands = Bands
|
||
}
|
||
if cfg.NearKm <= 0 {
|
||
cfg.NearKm = DefaultNearKm
|
||
}
|
||
if cfg.Logf == nil {
|
||
cfg.Logf = func(string, ...any) {}
|
||
}
|
||
return &Watcher{cfg: cfg}
|
||
}
|
||
|
||
// topics builds the subscription list.
|
||
//
|
||
// The v2 topic is
|
||
//
|
||
// pskr/filter/v2/<band>/<mode>/<tx call>/<rx call>/<tx grid>/<rx grid>/<tx dxcc>/<rx dxcc>
|
||
//
|
||
// so the receiver's square is a level the broker can filter on, and a band of
|
||
// "+" means every band. Filtering by RECEIVER square rather than by receiver
|
||
// DXCC is deliberate: measured on 20 m, one country ranged from 1.2 messages a
|
||
// second (OH) to 72.5 (K), because a DXCC can be a continent. By square the
|
||
// same measurement is 0.2 to 1.2 — the load follows distance, which is what the
|
||
// feed is actually about, and it is the same for every operator.
|
||
func (w *Watcher) topics() []string {
|
||
bands := w.cfg.Bands
|
||
// One subscription per band per square multiplies, and past a point the
|
||
// cheaper trade is to take every band from those squares and drop the
|
||
// unwanted ones here: four bands over six hundred squares is 2400
|
||
// subscriptions, where "+" is 600 for maybe three times the messages —
|
||
// which are then filtered locally, as they already are for everything else.
|
||
if len(bands) > 1 && len(bands)*len(w.cfg.RxGrids) > 1000 {
|
||
bands = []string{"+"}
|
||
}
|
||
out := []string{}
|
||
for _, b := range bands {
|
||
if len(w.cfg.RxGrids) == 0 {
|
||
out = append(out, "pskr/filter/v2/"+b+"/#")
|
||
continue
|
||
}
|
||
for _, g := range w.cfg.RxGrids {
|
||
out = append(out, "pskr/filter/v2/"+b+"/+/+/+/+/"+strings.ToUpper(g)+"/+/+")
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// Start connects and subscribes. Safe to call when already running.
|
||
func (w *Watcher) Start() error {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
if w.running {
|
||
return nil
|
||
}
|
||
if w.cfg.Geo == nil || w.cfg.OnSpot == nil {
|
||
return fmt.Errorf("pskr: Geo and OnSpot are required")
|
||
}
|
||
|
||
opts := mqtt.NewClientOptions().
|
||
AddBroker(w.cfg.Broker).
|
||
// A stable client id would collide with another OpsLog on the same
|
||
// account; the broker is anonymous, so uniqueness is ours to provide.
|
||
SetClientID(fmt.Sprintf("opslog-%d", time.Now().UnixNano())).
|
||
SetCleanSession(true).
|
||
SetAutoReconnect(true).
|
||
SetConnectRetry(true).
|
||
SetConnectRetryInterval(30 * time.Second).
|
||
SetConnectTimeout(15 * time.Second).
|
||
// No message is worth keeping if we cannot handle it now: an opening is
|
||
// a thing happening at this moment, and a queue of stale decodes would
|
||
// announce one that finished twenty minutes ago.
|
||
SetOrderMatters(false)
|
||
|
||
opts.OnConnect = func(c mqtt.Client) {
|
||
w.cfg.Logf("pskr: connected to %s", w.cfg.Broker)
|
||
topics := w.topics()
|
||
// In batches, not one at a time. Each Subscribe waits for its own
|
||
// acknowledgement, which is fine for the nine squares this started with
|
||
// and is minutes of waiting for the six hundred a 2000 km radius asks
|
||
// for — during which the feed is only partly subscribed and the panel
|
||
// looks broken.
|
||
const batch = 100
|
||
subbed := 0
|
||
for i := 0; i < len(topics); i += batch {
|
||
end := i + batch
|
||
if end > len(topics) {
|
||
end = len(topics)
|
||
}
|
||
filters := make(map[string]byte, end-i)
|
||
for _, t := range topics[i:end] {
|
||
filters[t] = 0
|
||
}
|
||
if tok := c.SubscribeMultiple(filters, w.handle); tok.Wait() && tok.Error() != nil {
|
||
w.cfg.Logf("pskr: subscribing to %d topics failed: %v", len(filters), tok.Error())
|
||
continue
|
||
}
|
||
subbed += len(filters)
|
||
}
|
||
w.cfg.Logf("pskr: watching %d topics (%d bands × %d receiver squares)", subbed, len(w.cfg.Bands), max(len(w.cfg.RxGrids), 1))
|
||
}
|
||
opts.OnConnectionLost = func(_ mqtt.Client, err error) {
|
||
w.mu.Lock()
|
||
w.lastErr = err.Error()
|
||
w.mu.Unlock()
|
||
w.cfg.Logf("pskr: connection lost: %v (will retry)", err)
|
||
}
|
||
|
||
c := mqtt.NewClient(opts)
|
||
// Deliberately NOT waiting on the connect token: the broker may be slow or
|
||
// unreachable and startup must not hang on a feature that is decoration.
|
||
// ConnectRetry brings it up in the background when it can.
|
||
c.Connect()
|
||
w.client = c
|
||
w.running = true
|
||
return nil
|
||
}
|
||
|
||
// Stop disconnects. Safe to call when already stopped.
|
||
func (w *Watcher) Stop() {
|
||
w.mu.Lock()
|
||
c, running := w.client, w.running
|
||
w.client, w.running = nil, false
|
||
w.mu.Unlock()
|
||
if running && c != nil {
|
||
c.Disconnect(250)
|
||
}
|
||
}
|
||
|
||
// wire is the payload as PSK Reporter sends it — short keys, no nesting.
|
||
type wire struct {
|
||
Freq int64 `json:"f"`
|
||
Mode string `json:"md"`
|
||
SNR int `json:"rp"`
|
||
TxCall string `json:"sc"`
|
||
TxGrid string `json:"sl"`
|
||
RxCall string `json:"rc"`
|
||
RxGrid string `json:"rl"`
|
||
Band string `json:"b"`
|
||
}
|
||
|
||
func (w *Watcher) handle(_ mqtt.Client, m mqtt.Message) {
|
||
var p wire
|
||
if err := json.Unmarshal(m.Payload(), &p); err != nil {
|
||
return // malformed payloads are not worth a log line at this rate
|
||
}
|
||
call := strings.ToUpper(strings.TrimSpace(p.TxCall))
|
||
grid := strings.ToUpper(strings.TrimSpace(p.TxGrid))
|
||
rxGrid := strings.ToUpper(strings.TrimSpace(p.RxGrid))
|
||
if call == "" || len(grid) < 4 || len(rxGrid) < 4 {
|
||
return
|
||
}
|
||
|
||
// The locator store takes every transmitter, before any of the geography
|
||
// below. What it wants is "which square is this callsign in", and that is
|
||
// true whoever happened to hear the report.
|
||
if w.cfg.OnGrid != nil {
|
||
w.cfg.OnGrid(call, grid[:4])
|
||
}
|
||
|
||
// THE RECEIVER HAS TO BE NEAR THE OPERATOR. This is the whole difference
|
||
// between a useful feed and a world map.
|
||
//
|
||
// A PSK Reporter message says "X was heard BY Y". Without this check, a
|
||
// station 1400 km from here heard by somebody in Japan counted as evidence
|
||
// of an opening — it proves the path from X to JAPAN, and says nothing at all
|
||
// about whether anything reaches this station. That is how a "2 m opening"
|
||
// came to be announced from a KX9X in the United States.
|
||
//
|
||
// Only reports collected by a receiver in the operator's own region show that
|
||
// signals are actually arriving HERE, which is the only question worth asking.
|
||
if rxDist, _, ok := w.cfg.Geo(rxGrid[:4]); !ok || rxDist > w.cfg.NearKm {
|
||
return
|
||
}
|
||
|
||
// The TRANSMITTER is the station on the air, so it is the transmitter's grid
|
||
// that gives the direction and length of the path that just proved itself.
|
||
dist, brg, ok := w.cfg.Geo(grid[:4])
|
||
if !ok {
|
||
return
|
||
}
|
||
s := Spot{
|
||
Call: call, Band: strings.ToLower(strings.TrimSpace(p.Band)),
|
||
Mode: strings.ToUpper(strings.TrimSpace(p.Mode)), Grid: grid[:4],
|
||
FreqHz: p.Freq,
|
||
DistKm: dist, Bearing: brg,
|
||
// Stamped on receipt: the broker's own timestamps vary between payload
|
||
// versions, and the window this feeds is measured in minutes.
|
||
At: time.Now(),
|
||
}
|
||
w.mu.Lock()
|
||
w.received++
|
||
w.lastAt = s.At
|
||
w.mu.Unlock()
|
||
w.cfg.OnSpot(s)
|
||
}
|
||
|
||
// Status is the snapshot the settings panel shows.
|
||
type Status struct {
|
||
Running bool `json:"running"`
|
||
Received uint64 `json:"received"`
|
||
LastAt time.Time `json:"last_at"`
|
||
LastErr string `json:"last_err,omitempty"`
|
||
Broker string `json:"broker"`
|
||
Bands []string `json:"bands"`
|
||
}
|
||
|
||
func (w *Watcher) Status() Status {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
st := Status{
|
||
Running: w.running, Received: w.received, LastAt: w.lastAt,
|
||
LastErr: w.lastErr, Broker: w.cfg.Broker,
|
||
}
|
||
st.Bands = append(st.Bands, w.cfg.Bands...)
|
||
return st
|
||
}
|