feat(bandopen): read PSK Reporter, and arrange the sources the watch needs
The detection shipped reading whatever the operator's cluster nodes happened to carry. 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, and Nexus flagged it on the same PC while OpsLog stayed silent. internal/pskr subscribes to pskr/filter/v2/<band>/# on PSK Reporter's MQTT broker. Every ordinary station running WSJT-X reports what it decodes, so the difference is two orders of magnitude rather than a threshold. The feed's shape suits us exactly: BOTH grids are in each message, so distance and bearing are arithmetic — no lookup, and no DXCC-centre approximation, which is what made the cluster path's bearings coarse. The geometry is injected from app.go so it stays the same arithmetic the cluster path uses; two answers to one question is how a bearing quietly becomes wrong. Volume was the design constraint, not the protocol. Six metres open is thousands of messages a minute and this runs on some very old PCs, so nothing is kept or persisted in the watcher: each message is parsed, measured and handed on or dropped, and the detector's existing window does the deciding. And the part that was the real bug: enabling the watch now ARRANGES ITS OWN SOURCES. It adds the two RBN nodes when missing and brings the feed up. A feature that silently depends on sources nobody can know are needed does not look unconfigured, it looks broken. Matched on host and port, not name, so an operator who renamed theirs does not get a duplicate — which the detector would read as twice as many stations, and announce an opening that is not there. Turning it off leaves the nodes alone: they may have been wanted for their own sake, and removing a node someone is using is worse than leaving one they are not.
This commit is contained in:
@@ -0,0 +1,230 @@
|
||||
// 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"
|
||||
|
||||
// Bands offered. HF below 12 m is deliberately absent: an "opening" on 20 m is
|
||||
// the normal state of the band and announcing it says nothing. These are the
|
||||
// bands where an opening is an event worth interrupting an operator for.
|
||||
var Bands = []string{"12m", "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
|
||||
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
|
||||
// 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)
|
||||
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.Logf == nil {
|
||||
cfg.Logf = func(string, ...any) {}
|
||||
}
|
||||
return &Watcher{cfg: cfg}
|
||||
}
|
||||
|
||||
// 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)
|
||||
for _, b := range w.cfg.Bands {
|
||||
// Every mode, every pair of stations, on this band. That firehose IS
|
||||
// the point: the detector's job is to find the shape in it.
|
||||
topic := "pskr/filter/v2/" + b + "/#"
|
||||
if tok := c.Subscribe(topic, 0, w.handle); tok.Wait() && tok.Error() != nil {
|
||||
w.cfg.Logf("pskr: subscribe %s failed: %v", topic, tok.Error())
|
||||
continue
|
||||
}
|
||||
w.cfg.Logf("pskr: watching %s", topic)
|
||||
}
|
||||
}
|
||||
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))
|
||||
if call == "" || len(grid) < 4 {
|
||||
return
|
||||
}
|
||||
// The TRANSMITTER is the station on the air; the receiver is whoever
|
||||
// happened to be listening. An opening is described by where the signals are
|
||||
// coming from, so it is the transmitter's grid that is measured.
|
||||
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],
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user