// Package pskrtgt answers one question about one station: can they hear me? // // It is the other way round from internal/pskr. That watcher asks what is // happening AROUND HERE — reports collected near the operator, whoever sent // them — and it is the right shape for finding a band opening or a new entity. // This one starts from a callsign the operator wants to work and gathers the // evidence about that path, in both directions: // // - did the DX decode MY call, and how long ago // - who NEAR ME did the DX decode (the path is open at my end) // - who near the DX decoded ME (the path is open at his end, even when he // uploads nothing himself) // - how many stations he is decoding right now (the pileup I am up against) // - where in his receive passband those decodes land, so a caller can pick a // slot he is not already covered on // // Nothing here is persisted and nothing is inferred from a QSO: it is a sliding // window of PSK Reporter reports, and when the window empties the answer goes // back to "not known", which is the honest answer. // // The window is FIVE minutes. An FT8 cycle is fifteen seconds, so that is // twenty chances for a path to show itself — short enough that "he decoded you" // still means now, long enough that one missed cycle does not erase it. package pskrtgt import ( "encoding/json" "encoding/xml" "fmt" "net/http" "net/url" "sort" "strconv" "strings" "sync" "time" mqtt "github.com/eclipse/paho.mqtt.golang" ) // DefaultBroker is PSK Reporter's public MQTT endpoint, TLS. Same one the // band-opening watcher uses — two connections to it, because the two want // opposite slices of the feed and neither can be filtered out of the other's. const DefaultBroker = "tls://mqtt.pskreporter.info:1884" const ( // window is how far back a report still counts. // // TEN minutes. Five was chosen as "recent enough to still mean now", and on // the air it meant half the evidence: PSK Reporter's uploaders batch their // reports, many of them every five minutes, so a five-minute window catches // roughly one upload cycle per station. Side by side with DXHunter on the // same DX, the same second: 18 decodes here against 27 there, and a station // missing from "from your area" that was simply six minutes old. // // It is a window on ONE station's activity, not on the band: ten minutes of // a DX working a pileup is still what he is doing now. window = 10 * time.Minute // pileupWindow is the tighter one for "how many stations is he working // through". A station he decoded four minutes ago has very likely moved on, // and counting it inflates the only number an operator uses to decide // whether it is worth calling at all. pileupWindow = 2 * time.Minute // The passband histogram: 60 Hz bins from 200 Hz to 4000 Hz. Above 4 kHz // there is essentially no FT8, and drawing the empty space made the strip // look broken rather than empty. binHz = 60 lowHz = 200 highHz = 4000 ) // Scope decides how much of the feed is subscribed to. type Scope string const ( // ScopeTarget subscribes to three filters: what the DX transmits, what he // receives, and who hears the operator. A handful of messages a second, and // the REST backfill fills the window the moment the target changes. ScopeTarget Scope = "target" // ScopeBand subscribes to the whole band's FTx traffic. Switching target is // then instant with no backfill, at the cost of every message on the band — // hundreds a second when 20 m is busy. ScopeBand Scope = "band" ) // spot is one PSK Reporter reception report, as the v2 payload carries it. type spot 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"` // at is stamped on arrival. The payload's own timestamps differ between // versions of the feed, and everything here is measured in minutes. at time.Time } // Entry is one station in one of the lists the panel shows. type Entry struct { Call string `json:"call"` Grid string `json:"grid"` SNR int `json:"snr"` OffsetHz int `json:"offset_hz"` // audio offset from the operator's dial AgeSec int `json:"age_sec"` } // Bin is one 60 Hz slice of the DX's receive passband. type Bin struct { OffsetHz int `json:"offset_hz"` Count int `json:"count"` AvgSNR float64 `json:"avg_snr"` } // Analysis is the whole snapshot the panel draws, recomputed on demand. type Analysis struct { Target string `json:"target"` Mode string `json:"mode,omitempty"` // Enabled is the operator's switch; Online is whether the broker is // actually connected. A panel that says nothing has to be able to say WHY. Enabled bool `json:"enabled"` Online bool `json:"online"` // Spots is everything in the window, the sign that the feed is alive even // when every counter below is legitimately zero. Spots int `json:"spots"` // HeMe is the answer to the question. The rest is what to do when it is no. HeMe bool `json:"he_me"` HeMeSeconds int `json:"he_me_seconds"` HeMeSNR int `json:"he_me_snr"` HeMeOffset int `json:"he_me_offset_hz"` // TargetUploads distinguishes "he is not hearing anybody" from "his software // tells PSK Reporter nothing" — without it, a silent panel reads as a dead // band when it may be a full one. TargetUploads bool `json:"target_uploads"` TargetGrid string `json:"target_grid,omitempty"` // Near the DX: stations in his square that decoded the operator. This is // what still works when he uploads nothing himself. NearHimCount int `json:"near_him_count"` NearHimTop []Entry `json:"near_him_top"` // Near the operator: stations in his own field that the DX decoded. FromMyAreaCount int `json:"from_my_area_count"` FromMyAreaTop []Entry `json:"from_my_area_top"` PathOpen bool `json:"path_open"` // Who heard the DX, worldwide and locally. HeardByCount int `json:"heard_by_count"` HeardNearMe int `json:"heard_near_me"` HeardNearMeTop []Entry `json:"heard_near_me_top"` // The pileup: everyone he decoded (window), and the recent slice of it. DecodedByCount int `json:"decoded_by_count"` DecodedByTop []Entry `json:"decoded_by_top"` DecodedByCalls []string `json:"decoded_by_calls"` PileupCount int `json:"pileup_count"` // His receive passband, and a slot in it that nobody is using. DialHz int64 `json:"dial_hz"` CeilingHz int `json:"ceiling_hz"` DecodesInWindow int `json:"decodes_in_window"` Bins []Bin `json:"bins"` SuggestedOffset int `json:"suggested_offset"` } // Config is what the watcher needs from the application. type Config struct { Broker string Scope Scope // MyCall and MyGrid are the operator's. Both matter: the callsign is what // "he decoded you" is looked up by, and the grid decides what counts as // "near me" — its first two characters, a Maidenhead FIELD, which is a few // hundred kilometres rather than a whole continent. MyCall string MyGrid string // Continent resolves a callsign to EU/NA/AS/… It is only a FALLBACK, for an // operator whose grid is not set: without a grid there is nothing to compare // squares with, and a continent is better than nothing. Injected so this // package does not pull in the country file. Continent func(call string) string Logf func(string, ...any) } // Watcher owns the MQTT connection and the window. type Watcher struct { mu sync.Mutex cfg Config client mqtt.Client target string // the callsign being analysed, upper case mode string // FT8 / FT4 — the target's mode, for the band-scope topic band string // band tag currently subscribed to under ScopeBand dialHz int64 // the operator's dial, for audio offsets // subs is what we are subscribed to right now, so a target change can take // the old filters down without guessing at their shape. subs []string spots []spot // backfilled remembers the target the REST history was fetched for, so the // panel's polling cannot re-fetch it every second. PSK Reporter's query API // answers that with a rate limit, and rightly. backfilled string } func New(cfg Config) *Watcher { if cfg.Broker == "" { cfg.Broker = DefaultBroker } if cfg.Scope == "" { cfg.Scope = ScopeTarget } if cfg.Logf == nil { cfg.Logf = func(string, ...any) {} } return &Watcher{cfg: cfg} } // Watch points the analysis at a callsign. Connects on the first call, so an // operator who never opens the panel never opens a socket. // // Called repeatedly with the same target — the panel re-asserts it as the // operator works — so everything expensive here is guarded on an actual change. func (w *Watcher) Watch(target, mode string, dialHz int64) error { target = strings.ToUpper(strings.TrimSpace(target)) mode = strings.ToUpper(strings.TrimSpace(mode)) if mode == "" { mode = "FT8" } if target == "" { w.Stop() return nil } w.mu.Lock() changed := target != w.target || mode != w.mode w.target, w.mode = target, mode if dialHz > 0 { w.dialHz = dialHz } band := bandTag(w.dialHz) bandChanged := band != "" && band != w.band // Set BEFORE any connect: the subscription is built from it, and a first // connect that found it empty would subscribe to every band at once under // the band-wide scope — the one case where that is expensive. if band != "" { w.band = band } client := w.client w.mu.Unlock() if client == nil || !client.IsConnected() { c, err := w.connect() if err != nil { return err } w.mu.Lock() w.client, client = c, c w.mu.Unlock() // connect() subscribes on its own OnConnect handler; anything below // would only repeat it. changed = false bandChanged = false } if !changed && !bandChanged { return nil } w.mu.Lock() // The window belongs to the target it was collected for. Keeping it across a // change would answer the new question with the old station's evidence. if changed { w.spots = w.spots[:0] } w.mu.Unlock() if err := w.resubscribe(client); err != nil { return err } if changed && w.cfg.Scope == ScopeTarget { // Under the band-wide subscription the window is already full of the new // target's reports; under the narrow one it is empty, and the REST query // is what makes the panel useful in the first fifteen seconds instead of // after five minutes. go w.backfill(target, mode) } return nil } // resubscribe replaces every filter with the ones the current target and scope // want. Takes the old ones down first: a target change that only ADDED filters // would leave the previous station's reports arriving for ever. func (w *Watcher) resubscribe(c mqtt.Client) error { w.mu.Lock() old := w.subs topics := w.topicsLocked() w.subs = topics target, scope := w.target, w.cfg.Scope w.mu.Unlock() if len(old) > 0 { if tok := c.Unsubscribe(old...); tok.Wait() && tok.Error() != nil { w.cfg.Logf("pskr target: unsubscribe failed: %v", tok.Error()) } } if len(topics) == 0 { return nil } filters := make(map[string]byte, len(topics)) for _, t := range topics { filters[t] = 0 } if tok := c.SubscribeMultiple(filters, w.handle); tok.Wait() && tok.Error() != nil { return fmt.Errorf("pskr target: subscribe: %w", tok.Error()) } w.cfg.Logf("pskr target: watching %s (%s scope, %d filters)", target, scope, len(topics)) return nil } // topicsLocked builds the subscription list. The v2 topic is // // pskr/filter/v2//////// // // so both directions of one callsign are addressable at the broker, which is // the whole reason the narrow scope costs almost nothing. func (w *Watcher) topicsLocked() []string { if w.target == "" { return nil } if w.cfg.Scope == ScopeBand { band := w.band if band == "" { band = "+" } return []string{"pskr/filter/v2/" + band + "/" + w.mode + "/#"} } out := []string{ // What he is transmitting: who is hearing him. "pskr/filter/v2/+/" + w.mode + "/" + w.target + "/#", // What he is receiving: the pileup, and whether the operator is in it. "pskr/filter/v2/+/" + w.mode + "/+/" + w.target + "/#", } // Who hears the OPERATOR. Only some of those receivers are near the DX, and // those are the ones that answer "can I be heard over there" on a DX who // uploads nothing himself. Left out when the callsign is not configured // rather than subscribing to a filter with an empty level in it. if my := strings.ToUpper(strings.TrimSpace(w.cfg.MyCall)); my != "" { out = append(out, "pskr/filter/v2/+/"+w.mode+"/"+my+"/#") } return out } func (w *Watcher) connect() (mqtt.Client, error) { opts := mqtt.NewClientOptions(). AddBroker(w.cfg.Broker). SetClientID(fmt.Sprintf("opslog-tgt-%d", time.Now().UnixNano())). SetCleanSession(true). SetAutoReconnect(true). SetConnectRetry(true). SetConnectRetryInterval(30 * time.Second). SetConnectTimeout(15 * time.Second). SetOrderMatters(false) // Re-subscribe on every connect, reconnects included: the session is clean, // so the broker remembers nothing and a dropped link would otherwise come // back up subscribed to nothing at all — a panel that goes quiet for ever // while still saying "online". opts.OnConnect = func(c mqtt.Client) { w.mu.Lock() w.subs = nil target, mode := w.target, w.mode w.mu.Unlock() if err := w.resubscribe(c); err != nil { w.cfg.Logf("pskr target: %v", err) } if target != "" && w.cfg.Scope == ScopeTarget { go w.backfill(target, mode) } } opts.OnConnectionLost = func(_ mqtt.Client, err error) { w.cfg.Logf("pskr target: connection lost: %v (will retry)", err) } c := mqtt.NewClient(opts) tok := c.Connect() if !tok.WaitTimeout(15*time.Second) || tok.Error() != nil { err := tok.Error() if err == nil { err = fmt.Errorf("timeout") } return nil, fmt.Errorf("pskr target: connect %s: %w", w.cfg.Broker, err) } return c, nil } func (w *Watcher) handle(_ mqtt.Client, m mqtt.Message) { var s spot if err := json.Unmarshal(m.Payload(), &s); err != nil { return } if s.TxCall == "" || s.RxCall == "" { return } s.TxCall = strings.ToUpper(s.TxCall) s.RxCall = strings.ToUpper(s.RxCall) s.TxGrid = strings.ToUpper(s.TxGrid) s.RxGrid = strings.ToUpper(s.RxGrid) s.at = time.Now() w.mu.Lock() w.spots = append(w.spots, s) w.mu.Unlock() } // Stop drops the target and the connection. The window goes with it: it is // evidence about a station nobody is asking about any more. func (w *Watcher) Stop() { w.mu.Lock() c := w.client w.client, w.target, w.band, w.subs, w.backfilled = nil, "", "", nil, "" w.spots = nil w.mu.Unlock() if c != nil { c.Disconnect(250) } } // SetDial updates the frequency audio offsets are measured against. func (w *Watcher) SetDial(hz int64) { if hz <= 0 { return } w.mu.Lock() w.dialHz = hz w.mu.Unlock() } // SetOperator refreshes the operator's own callsign and grid. Called when the // station profile changes: every "near me" answer is measured from these, and a // stale pair would quietly measure them from somebody else's station. func (w *Watcher) SetOperator(call, grid string) { w.mu.Lock() w.cfg.MyCall = strings.ToUpper(strings.TrimSpace(call)) w.cfg.MyGrid = strings.ToUpper(strings.TrimSpace(grid)) w.mu.Unlock() } // Snapshot recomputes the analysis from the window. func (w *Watcher) Snapshot() Analysis { w.mu.Lock() defer w.mu.Unlock() now := time.Now() cutoff := now.Add(-window) kept := w.spots[:0] for _, s := range w.spots { if s.at.After(cutoff) { kept = append(kept, s) } } w.spots = kept a := Analysis{ Target: w.target, Mode: w.mode, Online: w.client != nil && w.client.IsConnected(), DialHz: w.dialHz, Spots: len(w.spots), } if w.target == "" { return a } myCall := strings.ToUpper(strings.TrimSpace(w.cfg.MyCall)) myField := "" if g := strings.ToUpper(strings.TrimSpace(w.cfg.MyGrid)); len(g) >= 2 { myField = g[:2] } myCont := "" if myField == "" && myCall != "" && w.cfg.Continent != nil { myCont = strings.ToUpper(w.cfg.Continent(myCall)) } // His square, taken from any report where he was transmitting. It is what // "near him" is measured against, so without it that whole answer is // unavailable rather than approximated. for i := range w.spots { if w.spots[i].TxCall == w.target && len(w.spots[i].TxGrid) >= 4 { a.TargetGrid = w.spots[i].TxGrid[:4] } } entry := func(call, grid string, s *spot) Entry { off := 0 if w.dialHz > 0 { off = int(s.Freq - w.dialHz) } return Entry{Call: call, Grid: grid, SNR: s.SNR, OffsetHz: off, AgeSec: int(now.Sub(s.at).Seconds())} } // One entry per station, overwritten as newer reports arrive, so a station // calling every cycle counts once and shows its latest report. heardBy := map[string]Entry{} heardNearMe := map[string]Entry{} fromMyArea := map[string]Entry{} decodedBy := map[string]Entry{} nearHim := map[string]Entry{} pileup := map[string]struct{}{} pileupCutoff := now.Add(-pileupWindow) type acc struct { n int sum float64 } bins := map[int]*acc{} var lastHeMe *spot near := func(theirGrid, call string) bool { if myField != "" { return strings.HasPrefix(strings.ToUpper(theirGrid), myField) } if myCont != "" && w.cfg.Continent != nil { return strings.ToUpper(w.cfg.Continent(call)) == myCont } return false } for i := range w.spots { s := &w.spots[i] // He transmitted: somebody heard him. if s.TxCall == w.target { e := entry(s.RxCall, s.RxGrid, s) heardBy[s.RxCall] = e if near(s.RxGrid, s.RxCall) { heardNearMe[s.RxCall] = e } } // The operator transmitted and a station in the DX's own square heard // it. That is a path to his region, proved without his help. if a.TargetGrid != "" && myCall != "" && s.TxCall == myCall && strings.HasPrefix(s.RxGrid, a.TargetGrid) { nearHim[s.RxCall] = entry(s.RxCall, s.RxGrid, s) } // He received: this is the pileup, the passband, and the answer. if s.RxCall == w.target { a.DecodesInWindow++ if s.TxCall == myCall { if lastHeMe == nil || s.at.After(lastHeMe.at) { lastHeMe = s } continue // the operator is not part of his own pileup } decodedBy[s.TxCall] = entry(s.TxCall, s.TxGrid, s) if s.at.After(pileupCutoff) { pileup[s.TxCall] = struct{}{} } if near(s.TxGrid, s.TxCall) { fromMyArea[s.TxCall] = entry(s.TxCall, s.TxGrid, s) } if w.dialHz > 0 { off := int(s.Freq - w.dialHz) if off >= lowHz && off <= highHz { edge := (off / binHz) * binHz b := bins[edge] if b == nil { b = &acc{} bins[edge] = b } b.n++ b.sum += float64(s.SNR) if off > a.CeilingHz { a.CeilingHz = off } } } } } if lastHeMe != nil { a.HeMe = true a.HeMeSeconds = int(now.Sub(lastHeMe.at).Seconds()) a.HeMeSNR = lastHeMe.SNR if w.dialHz > 0 { a.HeMeOffset = int(lastHeMe.Freq - w.dialHz) } } a.TargetUploads = a.DecodesInWindow > 0 a.HeardByCount = len(heardBy) a.HeardNearMe = len(heardNearMe) a.HeardNearMeTop = top(heardNearMe, 5) a.FromMyAreaCount = len(fromMyArea) a.FromMyAreaTop = top(fromMyArea, 5) a.PathOpen = a.FromMyAreaCount > 0 a.NearHimCount = len(nearHim) a.NearHimTop = top(nearHim, 5) a.DecodedByCount = len(decodedBy) a.DecodedByTop = top(decodedBy, 10) a.DecodedByCalls = make([]string, 0, len(decodedBy)) for c := range decodedBy { a.DecodedByCalls = append(a.DecodedByCalls, c) } sort.Strings(a.DecodedByCalls) a.PileupCount = len(pileup) a.Bins = make([]Bin, 0, len(bins)) for edge, b := range bins { avg := 0.0 if b.n > 0 { avg = b.sum / float64(b.n) } a.Bins = append(a.Bins, Bin{OffsetHz: edge, Count: b.n, AvgSNR: avg}) } sort.Slice(a.Bins, func(i, j int) bool { return a.Bins[i].OffsetHz < a.Bins[j].OffsetHz }) a.SuggestedOffset = suggestOffset(a.Bins, a.CeilingHz) return a } // top returns the freshest entries from a per-callsign map, newest first. func top(m map[string]Entry, limit int) []Entry { out := make([]Entry, 0, len(m)) for _, e := range m { out = append(out, e) } sort.Slice(out, func(i, j int) bool { return out[i].AgeSec < out[j].AgeSec }) if len(out) > limit { out = out[:limit] } return out } // suggestOffset picks an audio slot to call on: the middle of the widest run of // empty bins below the ceiling. // // Below the CEILING, not below 4000 Hz. The ceiling is the highest offset he // has actually decoded, and it is the only evidence available about how wide // his receiver is set — plenty of stations run 2500 Hz. Suggesting 3400 Hz to // somebody whose passband stops at 2700 is advice to transmit into a filter. func suggestOffset(bins []Bin, ceiling int) int { if ceiling < 1000 { return 0 } used := map[int]bool{} for _, b := range bins { if b.Count > 0 { used[b.OffsetHz] = true // The neighbours too: FT8 is 50 Hz wide and the bins are 60, so a // signal on a bin edge covers the next one as surely as its own. used[b.OffsetHz-binHz] = true used[b.OffsetHz+binHz] = true } } bestStart, bestLen := -1, 0 start, run := -1, 0 // From 1000 Hz up: below that is where every default transmit offset sits, // so it is the most crowded part of the passband and the least useful // advice. for edge := 1020; edge+binHz <= ceiling; edge += binHz { if used[edge] { start, run = -1, 0 continue } if start < 0 { start = edge } run++ if run > bestLen { bestStart, bestLen = start, run } } if bestStart < 0 || bestLen < 2 { return 0 } return bestStart + bestLen*binHz/2 } // bandTag names the band a dial frequency is on, in PSK Reporter's own // vocabulary ("20m"). Only used by the band-wide scope, to subscribe to one // band instead of all of them. func bandTag(hz int64) string { khz := hz / 1000 switch { case khz >= 1800 && khz <= 2000: return "160m" case khz >= 3500 && khz <= 4000: return "80m" case khz >= 5250 && khz <= 5450: return "60m" case khz >= 7000 && khz <= 7300: return "40m" case khz >= 10100 && khz <= 10150: return "30m" case khz >= 14000 && khz <= 14350: return "20m" case khz >= 18068 && khz <= 18168: return "17m" case khz >= 21000 && khz <= 21450: return "15m" case khz >= 24890 && khz <= 24990: return "12m" case khz >= 28000 && khz <= 29700: return "10m" case khz >= 50000 && khz <= 54000: return "6m" case khz >= 70000 && khz <= 70500: return "4m" case khz >= 144000 && khz <= 148000: return "2m" case khz >= 430000 && khz <= 440000: return "70cm" } return "" } // ── REST backfill ───────────────────────────────────────────────────────── // // The narrow subscription starts empty, and five minutes of waiting is not an // answer to "should I call this station now". PSK Reporter's query API hands // back the last quarter hour in one request, so the window is populated before // the first cycle finishes. // // Fetched ONCE per target. The panel polls every second, and a query per poll // is what gets an application rate-limited off the service for everyone. type pskrReport struct { Sender string `xml:"senderCallsign,attr"` SenderGrid string `xml:"senderLocator,attr"` Receiver string `xml:"receiverCallsign,attr"` ReceiverGrid string `xml:"receiverLocator,attr"` Frequency string `xml:"frequency,attr"` SNR string `xml:"sNR,attr"` Mode string `xml:"mode,attr"` FlowStartSecs string `xml:"flowStartSeconds,attr"` } type pskrReports struct { XMLName xml.Name `xml:"receptionReports"` Reports []pskrReport `xml:"receptionReport"` } // backfill fetches the last quarter hour for a target, in BOTH directions. // // Two queries, because the panel asks two questions and the service answers // them separately: what the target RECEIVED (his pileup, the passband, whether // he decoded us) and what he TRANSMITTED (who is hearing him, and how much of // that is near us). The live feed fills both eventually; a target picked ten // seconds ago has neither, and with the narrow subscription there is nothing in // the window at all until his own uploader next reports. // // Fetched ONCE per target. The panel polls every second, and a query per poll // is what gets an application rate-limited off the service for everyone. func (w *Watcher) backfill(target, mode string) { w.mu.Lock() if w.backfilled == target { w.mu.Unlock() return } w.backfilled = target w.mu.Unlock() got := 0 for _, dir := range []struct{ param, what string }{ {"receiverCallsign", "decoded by him"}, {"senderCallsign", "who is hearing him"}, } { q := url.Values{} q.Set(dir.param, target) q.Set("mode", mode) q.Set("flowStartSeconds", strconv.Itoa(-900)) q.Set("nolocator", "0") // The pskquery5 endpoint rather than retrieve.pskreporter.info: this is // the one DXHunter has been using against the live service, and a // backfill that silently returns nothing is worse than none at all. req, err := http.NewRequest("GET", "https://pskreporter.info/cgi-bin/pskquery5.pl?"+q.Encode(), nil) if err != nil { continue } req.Header.Set("User-Agent", "OpsLog (PSK Reporter target analysis)") resp, err := (&http.Client{Timeout: 15 * time.Second}).Do(req) if err != nil { w.cfg.Logf("pskr target: history for %s (%s) unavailable: %v", target, dir.what, err) continue } if resp.StatusCode != http.StatusOK { // 503 is the service saying "too often". Worth a line, because the // panel then fills at the live feed's pace and looks slow for no // visible reason. w.cfg.Logf("pskr target: history for %s (%s) refused (HTTP %d)", target, dir.what, resp.StatusCode) resp.Body.Close() continue } var rr pskrReports err = xml.NewDecoder(resp.Body).Decode(&rr) resp.Body.Close() if err != nil { continue } got += w.absorb(target, rr.Reports) } if got > 0 { w.cfg.Logf("pskr target: %d recent reports for %s from the history queries", got, target) } } // absorb adds fetched reports to the window, skipping what the live feed has // already delivered. Without the check the same report arrives twice — once by // MQTT, once by query — and every count that is not per-callsign doubles: the // decode total, and the bars of the passband. func (w *Watcher) absorb(target string, reports []pskrReport) int { now := time.Now() w.mu.Lock() defer w.mu.Unlock() // Still the same target? The operator may have moved on while this was in // flight, and dropping a stale answer into the window would attribute one // station's pileup to another. if w.target != target { return 0 } type key struct { tx, rx string hz int64 } seen := make(map[key]bool, len(w.spots)) for i := range w.spots { seen[key{w.spots[i].TxCall, w.spots[i].RxCall, w.spots[i].Freq}] = true } added := 0 for _, r := range reports { hz, _ := strconv.ParseInt(r.Frequency, 10, 64) snr, _ := strconv.Atoi(r.SNR) k := key{strings.ToUpper(r.Sender), strings.ToUpper(r.Receiver), hz} if hz == 0 || seen[k] { continue } at := now if secs, err := strconv.ParseInt(r.FlowStartSecs, 10, 64); err == nil { switch { case secs > 1_000_000_000: at = time.Unix(secs, 0) // an absolute time case secs < 0: at = now.Add(time.Duration(secs) * time.Second) // an age in seconds } } // Stamped with its REAL age, so it ages out of the window on its own and // a quarter-hour-old decode is never read as "he heard you just now". if at.Before(now.Add(-window)) { continue } seen[k] = true w.spots = append(w.spots, spot{ Freq: hz, Mode: strings.ToUpper(r.Mode), SNR: snr, TxCall: k.tx, TxGrid: strings.ToUpper(r.SenderGrid), RxCall: k.rx, RxGrid: strings.ToUpper(r.ReceiverGrid), at: at, }) added++ } return added }