feat(ftmap): a layer for who hears me
The map drew one direction of every path on it: what this receiver decoded. The reverse — which stations are reporting our own transmissions — is the half an operator cannot see from their own radio, and on FT8 it is the half that decides whether calling is worth the cycle. It is the narrowest slice of the PSK Reporter feed there is. The v2 topic is pskr/filter/v2/<band>/<mode>/<tx call>/<rx call>/… so putting the operator's callsign in the TRANSMIT level makes the broker send nothing else. For scale, from internal/pskr's own measured numbers: four bands unfiltered is 83 messages a second, filtered on the receiver's square 0.2 to 1.2 a second — one callsign in the transmit level is a handful per FT8 cycle however open the band is. Both grids are in the payload, so the arc is arithmetic and there is no lookup. internal/pskrme, with its own connection, for the same reason internal/pskrtgt has its own: the three want slices of the feed that cannot be filtered out of one another. It also means this keeps working with the band-opening watch off — hanging it off that feed's lifecycle would have made it fail silently for anyone not chasing openings. Nothing is persisted. One entry per STATION inside a fifteen-minute window, carrying its freshest report: PSK Reporter's uploaders batch, many every five minutes, so a tighter window would show a fraction of who actually heard the last few calls. Stop clears the window, or switching the layer back on would redraw who heard us before it was on. On the map the layer is dashed and single-coloured. Solid is what we decoded, dashed is somebody decoding us; colour alone could not carry that distinction next to fourteen band colours. The receivers are rings rather than filled dots for the same reason. Per profile, since the callsign IS the subscription — a switch resubscribes rather than going on reporting who hears the previous station.
This commit is contained in:
@@ -0,0 +1,271 @@
|
||||
// Package pskrme answers the other half of the decodes map: who is hearing ME.
|
||||
//
|
||||
// The FT map draws what this station decodes, which is one direction of every
|
||||
// path on it. The reverse — which stations are reporting our own transmissions
|
||||
// — is the half an operator cannot see from their own receiver at all, and on
|
||||
// FT8 it is the half that decides whether calling is worth the cycle.
|
||||
//
|
||||
// It is the narrowest possible slice of the PSK Reporter feed. The v2 topic is
|
||||
//
|
||||
// pskr/filter/v2/<band>/<mode>/<tx call>/<rx call>/<tx grid>/<rx grid>/…
|
||||
//
|
||||
// so putting the operator's callsign in the TX level makes the broker send
|
||||
// nothing else. That is the whole reason this is cheap: internal/pskr measured
|
||||
// 83 messages a second for four bands unfiltered, and 0.2 to 1.2 a second once
|
||||
// filtered on the receiver's square — one callsign in the transmit level is a
|
||||
// handful of messages per FT8 cycle, however open the band is.
|
||||
//
|
||||
// Its own connection, like internal/pskrtgt has its own: the three want
|
||||
// different slices of the feed, and none of them can be filtered out of
|
||||
// another's. It also means this works with the band-opening watch switched off,
|
||||
// which matters — tying it to that feed's lifecycle would have made it fail
|
||||
// silently for anyone not chasing openings.
|
||||
//
|
||||
// Nothing is persisted. A report older than the window is dropped on the next
|
||||
// read, and an empty window means nobody has reported us recently, which is the
|
||||
// honest answer rather than a stale map.
|
||||
package pskrme
|
||||
|
||||
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"
|
||||
|
||||
// Window is how long a report keeps counting.
|
||||
//
|
||||
// Fifteen minutes. PSK Reporter's uploaders batch, many of them every five, so
|
||||
// a tighter window shows a fraction of the stations that actually heard the
|
||||
// last few calls — internal/pskrtgt widened its own to ten for exactly that
|
||||
// reason. This one is looser still because it feeds a MAP: a receiver that
|
||||
// heard us twelve minutes ago is a path worth seeing on it, where the same
|
||||
// report as a live "can he hear me" verdict would be stale.
|
||||
const Window = 15 * time.Minute
|
||||
|
||||
// Report is one station's reception of us, reduced to what a map needs.
|
||||
type Report struct {
|
||||
Call string `json:"call"` // who reported us
|
||||
Grid string `json:"grid"` // their square, from the message itself
|
||||
Band string `json:"band"`
|
||||
Mode string `json:"mode"`
|
||||
SNR int `json:"snr"` // how they heard us, their report
|
||||
FreqHz int64 `json:"freq_hz"` // where we were when they did
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Status is what the panel needs to tell a working feed from a silent one.
|
||||
type Status struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
Online bool `json:"online"`
|
||||
Reports uint64 `json:"reports"` // accepted since start
|
||||
Watching string `json:"watching"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
|
||||
// Config is what the watcher needs to run.
|
||||
type Config struct {
|
||||
Broker string
|
||||
// MyCall is the callsign to watch for in the TRANSMIT level. Without one
|
||||
// there is no subscription to make: a wildcard there would be the whole
|
||||
// feed, which is the one thing this package exists not to do.
|
||||
MyCall string
|
||||
Logf func(string, ...any)
|
||||
}
|
||||
|
||||
// Watcher owns the connection, its one subscription, and the sliding window.
|
||||
type Watcher struct {
|
||||
mu sync.Mutex
|
||||
cfg Config
|
||||
client mqtt.Client
|
||||
running bool
|
||||
topic string
|
||||
|
||||
reports []Report
|
||||
received uint64
|
||||
lastErr string
|
||||
}
|
||||
|
||||
func New(cfg Config) *Watcher {
|
||||
if cfg.Broker == "" {
|
||||
cfg.Broker = DefaultBroker
|
||||
}
|
||||
if cfg.Logf == nil {
|
||||
cfg.Logf = func(string, ...any) {}
|
||||
}
|
||||
cfg.MyCall = strings.ToUpper(strings.TrimSpace(cfg.MyCall))
|
||||
return &Watcher{cfg: cfg}
|
||||
}
|
||||
|
||||
// message is the payload, the same shape internal/pskr documents.
|
||||
type message 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"`
|
||||
}
|
||||
|
||||
// Start brings the subscription up. Safe to call on a running watcher.
|
||||
func (w *Watcher) Start() error {
|
||||
w.mu.Lock()
|
||||
if w.running {
|
||||
w.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
call := w.cfg.MyCall
|
||||
w.mu.Unlock()
|
||||
if call == "" {
|
||||
return fmt.Errorf("pskrme: no station callsign")
|
||||
}
|
||||
|
||||
opts := mqtt.NewClientOptions().
|
||||
AddBroker(w.cfg.Broker).
|
||||
SetClientID(fmt.Sprintf("opslog-hearme-%d", time.Now().UnixNano())).
|
||||
SetCleanSession(true).
|
||||
SetAutoReconnect(true).
|
||||
SetConnectRetry(true).
|
||||
SetConnectRetryInterval(30 * time.Second).
|
||||
SetConnectTimeout(15 * time.Second).
|
||||
SetOrderMatters(false)
|
||||
// Subscribed on every connect, reconnects included: the session is clean, so
|
||||
// the broker remembers nothing and a silent reconnect would leave a feed
|
||||
// that looks up and delivers nothing for the rest of the evening.
|
||||
opts.OnConnect = func(c mqtt.Client) {
|
||||
// Both wildcards deliberate: every band and every mode. The filter that
|
||||
// matters is the callsign, and an operator wants to know who hears them
|
||||
// wherever they happen to be.
|
||||
topic := "pskr/filter/v2/+/+/" + call + "/#"
|
||||
if tok := c.Subscribe(topic, 0, w.handle); tok.Wait() && tok.Error() != nil {
|
||||
w.setErr(tok.Error().Error())
|
||||
w.cfg.Logf("pskrme: subscribing to %s failed: %v", topic, tok.Error())
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
w.topic = topic
|
||||
w.lastErr = ""
|
||||
w.mu.Unlock()
|
||||
w.cfg.Logf("pskrme: watching who reports %s", call)
|
||||
}
|
||||
opts.OnConnectionLost = func(_ mqtt.Client, err error) {
|
||||
w.setErr(err.Error())
|
||||
w.cfg.Logf("pskrme: connection lost: %v", err)
|
||||
}
|
||||
|
||||
client := mqtt.NewClient(opts)
|
||||
if tok := client.Connect(); tok.Wait() && tok.Error() != nil {
|
||||
return fmt.Errorf("pskrme: connect: %w", tok.Error())
|
||||
}
|
||||
w.mu.Lock()
|
||||
w.client, w.running = client, true
|
||||
w.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop drops the connection and everything collected. A feed turned off must
|
||||
// not leave a map showing who heard us before it was.
|
||||
func (w *Watcher) Stop() {
|
||||
w.mu.Lock()
|
||||
client, running := w.client, w.running
|
||||
w.client, w.running = nil, false
|
||||
w.reports = nil
|
||||
w.topic = ""
|
||||
w.mu.Unlock()
|
||||
if running && client != nil {
|
||||
client.Disconnect(250)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Watcher) setErr(msg string) {
|
||||
w.mu.Lock()
|
||||
w.lastErr = msg
|
||||
w.mu.Unlock()
|
||||
}
|
||||
|
||||
// handle runs on the MQTT goroutine, so it does the least possible.
|
||||
func (w *Watcher) handle(_ mqtt.Client, m mqtt.Message) {
|
||||
var msg message
|
||||
if err := json.Unmarshal(m.Payload(), &msg); err != nil {
|
||||
return
|
||||
}
|
||||
// The topic filter already guarantees the transmitter, but a receiver with
|
||||
// no callsign or no square cannot be drawn and is not a report of anything.
|
||||
if strings.TrimSpace(msg.RxCall) == "" || len(strings.TrimSpace(msg.RxGrid)) < 4 {
|
||||
return
|
||||
}
|
||||
r := Report{
|
||||
Call: strings.ToUpper(strings.TrimSpace(msg.RxCall)),
|
||||
Grid: strings.ToUpper(strings.TrimSpace(msg.RxGrid)),
|
||||
Band: strings.ToLower(strings.TrimSpace(msg.Band)),
|
||||
Mode: strings.ToUpper(strings.TrimSpace(msg.Mode)),
|
||||
SNR: msg.SNR,
|
||||
FreqHz: msg.Freq,
|
||||
At: time.Now(),
|
||||
}
|
||||
w.mu.Lock()
|
||||
w.reports = append(w.reports, r)
|
||||
w.received++
|
||||
// A cap as well as the window, so a pathological feed cannot grow this
|
||||
// without bound between two reads.
|
||||
if len(w.reports) > 4000 {
|
||||
w.reports = w.reports[len(w.reports)-2000:]
|
||||
}
|
||||
w.mu.Unlock()
|
||||
}
|
||||
|
||||
// Reports is who has heard us inside the window, freshest report per callsign.
|
||||
//
|
||||
// One entry per STATION, not per message: the same receiver uploading every
|
||||
// five minutes is one pair of ears on the map, and its latest report is the one
|
||||
// that says whether the path is still there.
|
||||
func (w *Watcher) Reports() []Report {
|
||||
cutoff := time.Now().Add(-Window)
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
// Pruned on read rather than on a timer: the only thing that cares about the
|
||||
// window is whoever is looking.
|
||||
kept := w.reports[:0]
|
||||
for _, r := range w.reports {
|
||||
if r.At.After(cutoff) {
|
||||
kept = append(kept, r)
|
||||
}
|
||||
}
|
||||
w.reports = kept
|
||||
|
||||
byCall := map[string]int{}
|
||||
out := make([]Report, 0, len(kept))
|
||||
for _, r := range kept {
|
||||
if i, seen := byCall[r.Call]; seen {
|
||||
if r.At.After(out[i].At) {
|
||||
out[i] = r
|
||||
}
|
||||
continue
|
||||
}
|
||||
byCall[r.Call] = len(out)
|
||||
out = append(out, r)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Status reports the connection, for the panel.
|
||||
func (w *Watcher) Status() Status {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
online := w.running && w.client != nil && w.client.IsConnected()
|
||||
return Status{
|
||||
Enabled: w.running,
|
||||
Online: online,
|
||||
Reports: w.received,
|
||||
Watching: w.cfg.MyCall,
|
||||
Error: w.lastErr,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package pskrme
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// One station uploading every five minutes must be ONE pair of ears on the map,
|
||||
// showing its freshest report — not four arcs to the same square, and not the
|
||||
// oldest of them deciding whether the path still looks open.
|
||||
func TestReportsKeepsTheFreshestPerStation(t *testing.T) {
|
||||
w := New(Config{MyCall: "F4BPO"})
|
||||
now := time.Now()
|
||||
w.reports = []Report{
|
||||
{Call: "OH5CX", Grid: "KP30", SNR: -18, At: now.Add(-9 * time.Minute)},
|
||||
{Call: "W1AW", Grid: "FN31", SNR: -5, At: now.Add(-2 * time.Minute)},
|
||||
{Call: "OH5CX", Grid: "KP30", SNR: -11, At: now.Add(-1 * time.Minute)},
|
||||
}
|
||||
got := w.Reports()
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("got %d stations, want 2: %+v", len(got), got)
|
||||
}
|
||||
for _, r := range got {
|
||||
if r.Call == "OH5CX" && r.SNR != -11 {
|
||||
t.Errorf("OH5CX kept the %d dB report, want the freshest (-11)", r.SNR)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A report older than the window is gone, and gone from the slice too: the map
|
||||
// must not show a path that stopped existing a quarter of an hour ago, and the
|
||||
// window is what keeps this from growing all evening.
|
||||
func TestReportsDropsWhatIsPastTheWindow(t *testing.T) {
|
||||
w := New(Config{MyCall: "F4BPO"})
|
||||
now := time.Now()
|
||||
w.reports = []Report{
|
||||
{Call: "OLD", Grid: "JN36", At: now.Add(-Window - time.Minute)},
|
||||
{Call: "NEW", Grid: "JN36", At: now.Add(-time.Minute)},
|
||||
}
|
||||
got := w.Reports()
|
||||
if len(got) != 1 || got[0].Call != "NEW" {
|
||||
t.Fatalf("got %+v, want only NEW", got)
|
||||
}
|
||||
if len(w.reports) != 1 {
|
||||
t.Errorf("the stale report is still held: %d kept", len(w.reports))
|
||||
}
|
||||
}
|
||||
|
||||
// Turning the feed off clears what it collected. Left in place, switching it
|
||||
// back on would redraw a map of who heard us before it was on.
|
||||
func TestStopForgetsTheReports(t *testing.T) {
|
||||
w := New(Config{MyCall: "F4BPO"})
|
||||
w.reports = []Report{{Call: "OH5CX", Grid: "KP30", At: time.Now()}}
|
||||
w.Stop()
|
||||
if got := w.Reports(); len(got) != 0 {
|
||||
t.Errorf("got %+v after Stop, want nothing", got)
|
||||
}
|
||||
}
|
||||
|
||||
// No callsign, no subscription: the transmit level would be a wildcard, which
|
||||
// is the entire feed — the one thing this package exists not to ask for.
|
||||
func TestStartRefusesWithoutACallsign(t *testing.T) {
|
||||
if err := New(Config{}).Start(); err == nil {
|
||||
t.Fatal("started with no callsign")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user