From 661fec975d421387e9c765472f90315f2c48e497 Mon Sep 17 00:00:00 2001 From: maze Date: Tue, 17 Mar 2026 09:51:16 +0100 Subject: [PATCH] Refactoring stats gathering --- protocol/meshcore/node_repeater.go | 195 +++++++++++++++++++---------- 1 file changed, 130 insertions(+), 65 deletions(-) diff --git a/protocol/meshcore/node_repeater.go b/protocol/meshcore/node_repeater.go index 2d29319..0d81a2d 100644 --- a/protocol/meshcore/node_repeater.go +++ b/protocol/meshcore/node_repeater.go @@ -25,6 +25,7 @@ type repeaterDriver struct { lastFrameAt time.Time info repeaterInfo stats chan map[string]any + rawStats map[string]any err error } @@ -101,9 +102,10 @@ type repeaterInfo struct { func newRepeaterDriver(conn io.ReadWriteCloser) *repeaterDriver { return &repeaterDriver{ - conn: conn, - waiting: make(chan *repeaterDriverWaiting, 16), - stats: make(chan map[string]any, 2), + conn: conn, + waiting: make(chan *repeaterDriverWaiting, 16), + stats: make(chan map[string]any, 2), + rawStats: make(map[string]any), } } @@ -122,7 +124,6 @@ func (drv *repeaterDriver) Setup() (err error) { if err = drv.queryDeviceInfo(); err != nil { return err } - go drv.pollStats() return drv.err } @@ -191,6 +192,14 @@ func (drv *repeaterDriver) Stats() <-chan map[string]any { } func (drv *repeaterDriver) queryDeviceInfo() (err error) { + // Start with sending blank lines: + for i := 0; i < 3; i++ { + if _, err = fmt.Fprintf(drv.conn, "\r\n"); err != nil { + return + } + time.Sleep(time.Millisecond * 100) + } + if drv.info.FirmwareVersion, err = drv.writeCommand("ver"); err != nil { return } @@ -253,7 +262,10 @@ func (drv *repeaterDriver) queryDeviceInfo() (err error) { return } - return err + // Now that we're online, start the stats collector; + go drv.pollStats() + + return nil } func (drv *repeaterDriver) writeCommand(command string, args ...string) (response string, err error) { @@ -400,11 +412,18 @@ func (drv *repeaterDriver) poll() { Logger.Tracef("meshcore: handle %q", line) switch { case strings.HasPrefix(line, "-> "): + line = strings.TrimPrefix(line[3:], "> ") select { case waiting := <-drv.waiting: - waiting.Respond(strings.TrimPrefix(line[3:], "> ")) + Logger.Tracef("meshcore: send response to waiting: %q", line) + waiting.Respond(line) default: - Logger.Warnf("meshcore: unhandled response %q", line[3:]) + if strings.HasPrefix(line, "{") { + Logger.Tracef("meshcore: handle stats update %q", line) + drv.processStats(line) + } else { + Logger.Warnf("meshcore: unhandled response %q", line[3:]) + } } case strings.Contains(line, " RAW: "): @@ -417,6 +436,9 @@ func (drv *repeaterDriver) poll() { case strings.Contains(line, ": TX,"): // ignore (for now) + case strings.HasPrefix(line, "stats-"): + // ignore echo of the stats commands we send + default: Logger.Tracef("meshcore: repeater sent gibberish %q", line) } @@ -428,80 +450,123 @@ func (drv *repeaterDriver) pollStats() { defer ticker.Stop() for { - stats := make(map[string]any) + /* + stats := make(map[string]any) - neighbors, err := drv.getNeighbors() - if err != nil { - Logger.Warnf("meshcore: failed to get neighbors: %v", err) - } else { - stats["neighbors"] = neighbors - } - - response, err := drv.writeCommand("stats-core") - if err != nil { - Logger.Warnf("meshcore: failed to get stats: %v", err) - return - } - - neighborStats := make(map[string]any) - for _, line := range strings.Split(response, "\n") { - parts := strings.SplitN(line, "=", 2) - if len(parts) != 2 { - continue - } - key := parts[0] - value := parts[1] - - if i, err := strconv.Atoi(value); err == nil { - neighborStats[key] = i - } else if f, err := strconv.ParseFloat(value, 64); err == nil { - neighborStats[key] = f + neighbors, err := drv.getNeighbors() + if err != nil { + Logger.Warnf("meshcore: failed to get neighbors: %v", err) } else { - neighborStats[key] = value + stats["neighbors"] = neighbors } - } - var ( - coreStats = make(map[string]any) - radioStats = make(map[string]any) - packetStats = make(map[string]any) - ) - if response, err := drv.writeCommand("stats-core"); err == nil { - if err = json.Unmarshal([]byte(response), &coreStats); err != nil { - Logger.Warnf("meshcore: failed to decode core stats: %v", err) + response, err := drv.writeCommand("stats-core") + if err != nil { + Logger.Warnf("meshcore: failed to get stats: %v", err) + return } - } else { - Logger.Warnf("meshcore: failed to get core stats: %v", err) - } - if response, err := drv.writeCommand("stats-radio"); err == nil { - if err = json.Unmarshal([]byte(response), &radioStats); err != nil { - Logger.Warnf("meshcore: failed to decode radio stats: %v", err) + neighborStats := make(map[string]any) + for _, line := range strings.Split(response, "\n") { + parts := strings.SplitN(line, "=", 2) + if len(parts) != 2 { + continue + } + key := parts[0] + value := parts[1] + + if i, err := strconv.Atoi(value); err == nil { + neighborStats[key] = i + } else if f, err := strconv.ParseFloat(value, 64); err == nil { + neighborStats[key] = f + } else { + neighborStats[key] = value + } } - } else { - Logger.Warnf("meshcore: failed to get radio stats: %v", err) - } - if response, err := drv.writeCommand("stats-packets"); err == nil { - if err = json.Unmarshal([]byte(response), &packetStats); err != nil { - Logger.Warnf("meshcore: failed to decode packet stats: %v", err) + var ( + coreStats = make(map[string]any) + radioStats = make(map[string]any) + packetStats = make(map[string]any) + ) + if response, err := drv.writeCommand("stats-core"); err == nil { + if err = json.Unmarshal([]byte(response), &coreStats); err != nil { + Logger.Warnf("meshcore: failed to decode core stats: %v", err) + } + } else { + Logger.Warnf("meshcore: failed to get core stats: %v", err) } - } else { - Logger.Warnf("meshcore: failed to get packet stats: %v", err) + + if response, err := drv.writeCommand("stats-radio"); err == nil { + if err = json.Unmarshal([]byte(response), &radioStats); err != nil { + Logger.Warnf("meshcore: failed to decode radio stats: %v", err) + } + } else { + Logger.Warnf("meshcore: failed to get radio stats: %v", err) + } + + if response, err := drv.writeCommand("stats-packets"); err == nil { + if err = json.Unmarshal([]byte(response), &packetStats); err != nil { + Logger.Warnf("meshcore: failed to decode packet stats: %v", err) + } + } else { + Logger.Warnf("meshcore: failed to get packet stats: %v", err) + } + + stats["neighbors"] = neighborStats + stats["core"] = coreStats + stats["radio"] = radioStats + stats["packets"] = packetStats + + select { + case drv.stats <- stats: + default: + Logger.Warn("meshcore: stats channel full, dropping stats") + } + */ + + // Request the stats, we let the main poll loop handle the response, as we don't want to have multiple concurrent requests. + if _, err := drv.writeCommand("stats-core"); err != nil { + Logger.Warnf("meshcore: failed to request stats: %v", err) } + time.Sleep(time.Millisecond * 100) // small delay to avoid overwhelming the device + if _, err := drv.writeCommand("stats-radio"); err != nil { + Logger.Warnf("meshcore: failed to request stats: %v", err) + } + time.Sleep(time.Millisecond * 100) // small delay to avoid overwhelming the device + if _, err := drv.writeCommand("stats-packets"); err != nil { + Logger.Warnf("meshcore: failed to request stats: %v", err) + } + time.Sleep(time.Millisecond * 100) // small delay to avoid overwhelming the device - stats["neighbors"] = neighborStats - stats["core"] = coreStats - stats["radio"] = radioStats - stats["packets"] = packetStats + <-ticker.C + } +} +func (drv *repeaterDriver) processStats(line string) { + var stats map[string]any + if err := json.Unmarshal([]byte(line), &stats); err != nil { + Logger.Warnf("meshcore: failed to decode stats: %v", err) + return + } + + // TRAC[2026-03-17T09:42:10+01:00] meshcore: handle "-> {\"battery_mv\":4454,\"uptime_secs\":3751,\"errors\":0,\"queue_len\":0}" + // TRAC[2026-03-17T09:42:10+01:00] meshcore: handle "-> {\"noise_floor\":-115,\"last_rssi\":-78,\"last_snr\":11.50,\"tx_air_secs\":0,\"rx_air_secs\":147}" + // TRAC[2026-03-17T09:42:10+01:00] meshcore: handle "-> {\"recv\":404,\"sent\":1,\"flood_tx\":0,\"direct_tx\":1,\"flood_rx\":403,\"direct_rx\":1,\"recv_errors\":0}" + if _, ok := stats["battery_mv"]; ok { + drv.rawStats["core"] = stats + } else if _, ok := stats["noise_floor"]; ok { + drv.rawStats["radio"] = stats + } else if _, ok := stats["recv"]; ok { + drv.rawStats["packets"] = stats select { - case drv.stats <- stats: + case drv.stats <- drv.rawStats: + Logger.Trace("meshcore: published stats update") default: Logger.Warn("meshcore: stats channel full, dropping stats") } - - <-ticker.C + } else { + Logger.Warnf("meshcore: unknown stats update: %v", stats) } }