From af206350473f3e7a48188e4a1d1d4336b46996b7 Mon Sep 17 00:00:00 2001 From: simon Date: Fri, 23 Oct 2026 23:27:39 +0200 Subject: [PATCH] Harden BLE host link and expose latency tools in the dashboard. Add BLE RTT/stream tests and live tick median in the UI, fix concurrent WebSocket writes and late BATTERY replies, and skip 2M PHY updates that assert under ESP-NOW coexistence. Co-authored-by: Cursor --- goTool/api_ble_measure.go | 129 ++++++++++++++++++++++ goTool/api_serve.go | 1 + goTool/api_stream.go | 199 +++++++++++++++++---------------- goTool/ble_link.go | 59 ++++++++-- goTool/ble_measure.go | 224 ++++++++++++++++++++++++++++++++++++++ goTool/cache_status.go | 10 +- goTool/cmd_ble_rtt.go | 61 +---------- goTool/cmd_ble_stream.go | 112 ++----------------- goTool/dashboard.go | 147 +++++++++++++++++++++++-- goTool/docs/API_REST.md | 14 +++ goTool/host_managed.go | 27 ++++- goTool/webui/index.html | 185 ++++++++++++++++++++++++++++++- main/ble_host.c | 12 +- main/esp_now_master.c | 6 +- 14 files changed, 893 insertions(+), 293 deletions(-) create mode 100644 goTool/api_ble_measure.go create mode 100644 goTool/ble_measure.go diff --git a/goTool/api_ble_measure.go b/goTool/api_ble_measure.go new file mode 100644 index 0000000..841ef91 --- /dev/null +++ b/goTool/api_ble_measure.go @@ -0,0 +1,129 @@ +package main + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "time" +) + +type bleRttAPIRequest struct { + Count *int `json:"count"` + Warmup *int `json:"warmup"` + SettleS *float64 `json:"settle_s"` +} + +type bleStreamAPIRequest struct { + ClientID *uint32 `json:"client_id"` + PeriodMs *uint32 `json:"period_ms"` + Count *int `json:"count"` + TolMs *float64 `json:"tol_ms"` + TimeoutS *float64 `json:"timeout_s"` +} + +func mountBleTestAPI(mux *http.ServeMux, link *managedHost) { + mux.HandleFunc("/api/ble-rtt", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + serveBleRtt(w, r, link) + }) + mux.HandleFunc("/api/ble-stream", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + serveBleStreamTest(w, r, link) + }) +} + +func requireBLEHost(w http.ResponseWriter, link *managedHost) bool { + if link.Transport() != hostTransportBLE || !link.IsConnected() { + writeJSON(w, http.StatusBadRequest, bleRttResult{ + Error: "BLE host required (connect via BLE, DIP8=1)", + }) + return false + } + return true +} + +func serveBleRtt(w http.ResponseWriter, r *http.Request, link *managedHost) { + if !requireBLEHost(w, link) { + return + } + var body bleRttAPIRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil && !errors.Is(err, io.EOF) { + writeJSON(w, http.StatusBadRequest, bleRttResult{Error: "invalid JSON"}) + return + } + count, warmup := 50, 5 + settle := 0.0 + if body.Count != nil { + count = *body.Count + } + if body.Warmup != nil { + warmup = *body.Warmup + } + if body.SettleS != nil { + settle = *body.SettleS + } + + if !link.beginPausePoll() { + writeJSON(w, http.StatusConflict, bleRttResult{Error: "host busy"}) + return + } + defer link.endPausePoll() + + out := measureBleRtt(link, count, warmup, time.Duration(settle*float64(time.Second))) + status := http.StatusOK + if !out.Success { + status = http.StatusBadRequest + } + writeJSON(w, status, out) +} + +func serveBleStreamTest(w http.ResponseWriter, r *http.Request, link *managedHost) { + if !requireBLEHost(w, link) { + return + } + var body bleStreamAPIRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil && !errors.Is(err, io.EOF) { + writeJSON(w, http.StatusBadRequest, bleStreamResult{Error: "invalid JSON"}) + return + } + count := 50 + periodMs := uint32(16) + tolMs := 2.0 + timeoutS := 15.0 + var clientID uint32 + if body.Count != nil { + count = *body.Count + } + if body.PeriodMs != nil { + periodMs = *body.PeriodMs + } + if body.TolMs != nil { + tolMs = *body.TolMs + } + if body.TimeoutS != nil { + timeoutS = *body.TimeoutS + } + if body.ClientID != nil { + clientID = *body.ClientID + } + + if !link.beginPausePoll() { + writeJSON(w, http.StatusConflict, bleStreamResult{Error: "host busy"}) + return + } + defer link.endPausePoll() + + out := measureBleStream(link, clientID, periodMs, count, tolMs, timeoutS) + status := http.StatusOK + if !out.Success { + status = http.StatusBadRequest + } + writeJSON(w, status, out) +} diff --git a/goTool/api_serve.go b/goTool/api_serve.go index e3b1211..419a384 100644 --- a/goTool/api_serve.go +++ b/goTool/api_serve.go @@ -94,6 +94,7 @@ type otaAPIResponse struct { func mountServeAPI(mux *http.ServeMux, link *managedSerial, hub *wsHub, streamCtl *accelStreamCtl, tapCtl *tapNotifyCtl) { mountHostAPI(mux, link) + mountBleTestAPI(mux, link) mountLiveStreamAPI(mux, hub) mountAccelStreamAPI(mux, link, hub, streamCtl) mountTapAPI(mux, link, hub, tapCtl) diff --git a/goTool/api_stream.go b/goTool/api_stream.go index 26fbfff..d9d6d6b 100644 --- a/goTool/api_stream.go +++ b/goTool/api_stream.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "log" "net/http" "sync" @@ -180,12 +181,31 @@ type cachedTapEvent struct { type wsSubscriber struct { conn *websocket.Conn + writeMu sync.Mutex // gorilla: one writer per connection receiveInput bool interval time.Duration preFetch time.Duration lastInputSent time.Time } +func (sub *wsSubscriber) writeBytes(data []byte) error { + if sub == nil { + return fmt.Errorf("nil subscriber") + } + sub.writeMu.Lock() + defer sub.writeMu.Unlock() + _ = sub.conn.SetWriteDeadline(time.Now().Add(2 * time.Second)) + return sub.conn.WriteMessage(websocket.TextMessage, data) +} + +func (sub *wsSubscriber) writeJSON(v any) { + data, err := json.Marshal(v) + if err != nil { + return + } + _ = sub.writeBytes(data) +} + type pendingInputCache struct { cache *pb.CacheStatusResponse readAt time.Time @@ -269,9 +289,7 @@ func (h *accelStreamHub) register(conn *websocket.Conn, link *managedSerial) *ws "set_led_ring", "get_battery", }, } - if data, err := json.Marshal(hello); err == nil { - _ = conn.WriteMessage(websocket.TextMessage, data) - } + sub.writeJSON(hello) return sub } @@ -467,7 +485,11 @@ func (h *accelStreamHub) deliverInput(msg InputStreamMessage) { now := time.Now() h.mu.Lock() - defer h.mu.Unlock() + type pending struct { + conn *websocket.Conn + sub *wsSubscriber + } + var out []pending for conn, sub := range h.clients { if !sub.receiveInput { continue @@ -476,13 +498,26 @@ func (h *accelStreamHub) deliverInput(msg InputStreamMessage) { continue } sub.lastInputSent = now - if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { - delete(h.clients, conn) - _ = conn.Close() + out = append(out, pending{conn: conn, sub: sub}) + } + h.mu.Unlock() + + for _, p := range out { + if err := p.sub.writeBytes(data); err != nil { + h.mu.Lock() + delete(h.clients, p.conn) + h.mu.Unlock() + _ = p.conn.Close() } } } +func (h *accelStreamHub) clientsFromCache(cache *pb.CacheStatusResponse) []InputClientSample { + h.mu.Lock() + defer h.mu.Unlock() + return h.inputClientsFromCacheLocked(cache, time.Now()) +} + func runInputStreamer(link *managedSerial, hub *accelStreamHub, dash *wsHub, ctl *accelStreamCtl, tapCtl *tapNotifyCtl, stop <-chan struct{}) { ticker := time.NewTicker(minAPIStreamInterval) defer ticker.Stop() @@ -496,7 +531,7 @@ func runInputStreamer(link *managedSerial, hub *accelStreamHub, dash *wsHub, ctl case <-hub.configChanged: pending = nil case now := <-ticker.C: - if !hub.anyWantsInput() || !inputPollingActive(dash, ctl, tapCtl) { + if !hub.anyWantsInput() || !inputPollingActive(dash, ctl, tapCtl) || link.PollPaused() { pending = nil continue } @@ -508,14 +543,11 @@ func runInputStreamer(link *managedSerial, hub *accelStreamHub, dash *wsHub, ctl cache, cerr := cacheStatusFromBLEStreamPayload(payload) if cerr == nil { hub.ingestTapFromCache(cache) - hub.mu.RLock() - clients := hub.inputClientsFromCacheLocked(cache, time.Now()) - hub.mu.RUnlock() hub.deliverInput(InputStreamMessage{ Type: "input", T: time.Now().UnixNano(), Success: true, - Clients: clients, + Clients: hub.clientsFromCache(cache), }) pending = nil continue @@ -557,14 +589,11 @@ func runInputStreamer(link *managedSerial, hub *accelStreamHub, dash *wsHub, ctl } if pending.cache != nil && now.Sub(pending.readAt) >= waitPreFetch { - hub.mu.RLock() - clients := hub.inputClientsFromCacheLocked(pending.cache, now) - hub.mu.RUnlock() hub.deliverInput(InputStreamMessage{ Type: "input", T: ts, Success: true, - Clients: clients, + Clients: hub.clientsFromCache(pending.cache), }) pending = nil } @@ -582,46 +611,29 @@ func inputPollingActive(dash *wsHub, ctl *accelStreamCtl, tapCtl *tapNotifyCtl) return dash != nil && dash.anyAccelStreamEnabled() } -func writeStreamStatus(conn *websocket.Conn, msg StreamStatusMessage) { - data, err := json.Marshal(msg) - if err != nil { - return - } - _ = conn.WriteMessage(websocket.TextMessage, data) +func writeStreamStatus(sub *wsSubscriber, msg StreamStatusMessage) { + sub.writeJSON(msg) } -func writeBatteryStatus(conn *websocket.Conn, out batteryAPIResponse) { +func writeBatteryStatus(sub *wsSubscriber, out batteryAPIResponse) { out.Type = "battery_status" - data, err := json.Marshal(out) - if err != nil { - return - } - _ = conn.WriteMessage(websocket.TextMessage, data) + sub.writeJSON(out) } -func writeLedRingStatus(conn *websocket.Conn, out ledRingAPIResponse) { +func writeLedRingStatus(sub *wsSubscriber, out ledRingAPIResponse) { out.Type = "led_ring_status" - data, err := json.Marshal(out) - if err != nil { - return - } - _ = conn.WriteMessage(websocket.TextMessage, data) + sub.writeJSON(out) } -func writeInputStreamStatus(conn *websocket.Conn, out accelStreamAPIResponse) { - msg := InputStreamStatusMessage{ +func writeInputStreamStatus(sub *wsSubscriber, out accelStreamAPIResponse) { + sub.writeJSON(InputStreamStatusMessage{ Type: "input_stream_status", ClientID: out.ClientID, Enabled: out.Enabled, Success: out.Success, SlavesUpdated: out.SlavesUpdated, Error: out.Error, - } - data, err := json.Marshal(msg) - if err != nil { - return - } - _ = conn.WriteMessage(websocket.TextMessage, data) + }) } func clientInfoToAPI(c *pb.ClientInfo) APIClientInfo { @@ -640,16 +652,12 @@ func clientInfoToAPI(c *pb.ClientInfo) APIClientInfo { } } -func writeClientList(conn *websocket.Conn, msg ClientListMessage) { - data, err := json.Marshal(msg) - if err != nil { - return - } - _ = conn.WriteMessage(websocket.TextMessage, data) +func writeClientList(sub *wsSubscriber, msg ClientListMessage) { + sub.writeJSON(msg) } -func writeTapNotifyStatus(conn *websocket.Conn, out tapNotifyAPIResponse) { - msg := TapNotifyStatusMessage{ +func writeTapNotifyStatus(sub *wsSubscriber, out tapNotifyAPIResponse) { + sub.writeJSON(TapNotifyStatusMessage{ Type: "tap_notify_status", ClientID: out.ClientID, Single: out.Single, @@ -658,12 +666,7 @@ func writeTapNotifyStatus(conn *websocket.Conn, out tapNotifyAPIResponse) { Success: out.Success, SlavesUpdated: out.SlavesUpdated, Error: out.Error, - } - data, err := json.Marshal(msg) - if err != nil { - return - } - _ = conn.WriteMessage(websocket.TextMessage, data) + }) } func applyTapNotifyClientWS(link *managedSerial, dash *wsHub, tapCtl *tapNotifyCtl, clientID uint32, single, doubleTap, triple bool) tapNotifyAPIResponse { @@ -696,16 +699,16 @@ func applyTapNotifyClientWS(link *managedSerial, dash *wsHub, tapCtl *tapNotifyC return out } -func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, link *managedSerial, dash *wsHub, ctl *accelStreamCtl, tapCtl *tapNotifyCtl, hub *accelStreamHub) { +func handleAccelWSCommand(sub *wsSubscriber, data []byte, link *managedSerial, dash *wsHub, ctl *accelStreamCtl, tapCtl *tapNotifyCtl, hub *accelStreamHub) { var cmd accelWSCommand if err := json.Unmarshal(data, &cmd); err != nil { - writeStreamStatus(conn, StreamStatusMessage{Type: "stream_status", Error: "invalid JSON"}) + writeStreamStatus(sub, StreamStatusMessage{Type: "stream_status", Error: "invalid JSON"}) return } switch cmd.Type { case "get_host": - writeHostStatus(conn, hostStatusFromLink(link, true, "")) + writeHostStatus(sub, hostStatusFromLink(link, true, "")) case "list_serial_ports": ports, err := listSerialPorts() @@ -714,7 +717,7 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, msg.Error = err.Error() msg.Ports = []string{} } - writeWSJSON(conn, msg) + writeWSJSON(sub, msg) case "ble_scan": timeout := time.Duration(cmd.TimeoutS * float64(time.Second)) @@ -730,7 +733,7 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, msg.Error = err.Error() msg.Devices = []bleScanDevice{} } - writeWSJSON(conn, msg) + writeWSJSON(sub, msg) case "host_connect": var err error @@ -740,26 +743,26 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, case hostTransportBLE: err = link.ConnectBLE(cmd.Address, cmd.Name) default: - writeHostStatus(conn, HostStatusMessage{ + writeHostStatus(sub, HostStatusMessage{ Type: "host_status", Error: "transport must be uart or ble", }) return } if err != nil { - writeHostStatus(conn, hostStatusFromLink(link, false, err.Error())) + writeHostStatus(sub, hostStatusFromLink(link, false, err.Error())) return } - writeHostStatus(conn, hostStatusFromLink(link, true, "")) + writeHostStatus(sub, hostStatusFromLink(link, true, "")) case "host_disconnect": _ = link.Disconnect() - writeHostStatus(conn, hostStatusFromLink(link, true, "")) + writeHostStatus(sub, hostStatusFromLink(link, true, "")) case "list_clients": clients, err := link.listClientsPoll() if err != nil { - writeClientList(conn, ClientListMessage{ + writeClientList(sub, ClientListMessage{ Type: "client_list", Error: err.Error(), }) @@ -769,7 +772,7 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, for _, c := range clients { out = append(out, clientInfoToAPI(c)) } - writeClientList(conn, ClientListMessage{ + writeClientList(sub, ClientListMessage{ Type: "client_list", Success: true, Clients: out, @@ -777,34 +780,34 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, case "set_stream": if cmd.Enable == nil { - writeStreamStatus(conn, StreamStatusMessage{ + writeStreamStatus(sub, StreamStatusMessage{ Type: "stream_status", Error: "enable required", }) return } - writeStreamStatus(conn, hub.setStream(sub, *cmd.Enable, cmd.IntervalMs, cmd.PreFetch)) + writeStreamStatus(sub, hub.setStream(sub, *cmd.Enable, cmd.IntervalMs, cmd.PreFetch)) case "get_stream": - writeStreamStatus(conn, hub.getStream(sub)) + writeStreamStatus(sub, hub.getStream(sub)) case "set_input_stream": if cmd.ClientID == 0 { - writeInputStreamStatus(conn, accelStreamAPIResponse{Error: "client_id required"}) + writeInputStreamStatus(sub, accelStreamAPIResponse{Error: "client_id required"}) return } if cmd.Enable == nil { - writeInputStreamStatus(conn, accelStreamAPIResponse{ + writeInputStreamStatus(sub, accelStreamAPIResponse{ ClientID: cmd.ClientID, Error: "enable required", }) return } - writeInputStreamStatus(conn, applyAccelStreamClient(link, dash, ctl, cmd.ClientID, *cmd.Enable)) + writeInputStreamStatus(sub, applyAccelStreamClient(link, dash, ctl, cmd.ClientID, *cmd.Enable)) case "get_input_stream": if cmd.ClientID == 0 { - writeInputStreamStatus(conn, accelStreamAPIResponse{Error: "client_id required"}) + writeInputStreamStatus(sub, accelStreamAPIResponse{Error: "client_id required"}) return } resp, err := link.AccelStreamPoll(&pb.AccelStreamRequest{ @@ -812,7 +815,7 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, ClientId: cmd.ClientID, }) if err != nil { - writeInputStreamStatus(conn, accelStreamAPIResponse{ + writeInputStreamStatus(sub, accelStreamAPIResponse{ ClientID: cmd.ClientID, Error: err.Error(), }) @@ -821,7 +824,7 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, if ctl != nil { ctl.Set(cmd.ClientID, resp.GetEnabled()) } - writeInputStreamStatus(conn, accelStreamAPIResponse{ + writeInputStreamStatus(sub, accelStreamAPIResponse{ Enabled: resp.GetEnabled(), ClientID: resp.GetClientId(), Success: resp.GetSuccess(), @@ -830,15 +833,15 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, case "set_tap_notify": if cmd.AllClients { if cmd.Single == nil || cmd.DoubleTap == nil || cmd.Triple == nil { - writeTapNotifyStatus(conn, tapNotifyAPIResponse{Error: "single, double_tap, triple required"}) + writeTapNotifyStatus(sub, tapNotifyAPIResponse{Error: "single, double_tap, triple required"}) return } updated, err := applyTapNotifyAll(link, dash, tapCtl, *cmd.Single, *cmd.DoubleTap, *cmd.Triple) if err != nil { - writeTapNotifyStatus(conn, tapNotifyAPIResponse{Error: err.Error()}) + writeTapNotifyStatus(sub, tapNotifyAPIResponse{Error: err.Error()}) return } - writeTapNotifyStatus(conn, tapNotifyAPIResponse{ + writeTapNotifyStatus(sub, tapNotifyAPIResponse{ Success: updated > 0, SlavesUpdated: updated, Single: *cmd.Single, @@ -848,21 +851,21 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, return } if cmd.ClientID == 0 { - writeTapNotifyStatus(conn, tapNotifyAPIResponse{Error: "client_id required"}) + writeTapNotifyStatus(sub, tapNotifyAPIResponse{Error: "client_id required"}) return } if cmd.Single == nil || cmd.DoubleTap == nil || cmd.Triple == nil { - writeTapNotifyStatus(conn, tapNotifyAPIResponse{ + writeTapNotifyStatus(sub, tapNotifyAPIResponse{ ClientID: cmd.ClientID, Error: "single, double_tap, triple required", }) return } - writeTapNotifyStatus(conn, applyTapNotifyClientWS(link, dash, tapCtl, cmd.ClientID, *cmd.Single, *cmd.DoubleTap, *cmd.Triple)) + writeTapNotifyStatus(sub, applyTapNotifyClientWS(link, dash, tapCtl, cmd.ClientID, *cmd.Single, *cmd.DoubleTap, *cmd.Triple)) case "get_tap_notify": if cmd.ClientID == 0 { - writeTapNotifyStatus(conn, tapNotifyAPIResponse{Error: "client_id required"}) + writeTapNotifyStatus(sub, tapNotifyAPIResponse{Error: "client_id required"}) return } resp, err := link.TapNotifyPoll(&pb.TapNotifyRequest{ @@ -870,7 +873,7 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, ClientId: cmd.ClientID, }) if err != nil { - writeTapNotifyStatus(conn, tapNotifyAPIResponse{ + writeTapNotifyStatus(sub, tapNotifyAPIResponse{ ClientID: cmd.ClientID, Error: err.Error(), }) @@ -879,7 +882,7 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, if tapCtl != nil { tapCtl.Set(cmd.ClientID, resp.GetSingle(), resp.GetDoubleTap(), resp.GetTriple()) } - writeTapNotifyStatus(conn, tapNotifyAPIResponse{ + writeTapNotifyStatus(sub, tapNotifyAPIResponse{ ClientID: cmd.ClientID, Success: resp.GetSuccess(), Single: resp.GetSingle(), @@ -890,28 +893,28 @@ func handleAccelWSCommand(conn *websocket.Conn, sub *wsSubscriber, data []byte, case "set_led_ring": var body ledRingAPIRequest if err := json.Unmarshal(data, &body); err != nil { - writeLedRingStatus(conn, ledRingAPIResponse{Error: "invalid JSON"}) + writeLedRingStatus(sub, ledRingAPIResponse{Error: "invalid JSON"}) return } if body.Mode == "" { - writeLedRingStatus(conn, ledRingAPIResponse{Error: "mode required"}) + writeLedRingStatus(sub, ledRingAPIResponse{Error: "mode required"}) return } - writeLedRingStatus(conn, applyLedRing(link, body)) + writeLedRingStatus(sub, applyLedRing(link, body)) case "get_battery": var body batteryAPIRequest if err := json.Unmarshal(data, &body); err != nil { - writeBatteryStatus(conn, batteryAPIResponse{Error: "invalid JSON"}) + writeBatteryStatus(sub, batteryAPIResponse{Error: "invalid JSON"}) return } if !body.AllClients && body.ClientID == 0 { body.AllClients = true } - writeBatteryStatus(conn, applyBatteryStatus(link, body)) + writeBatteryStatus(sub, applyBatteryStatus(link, body)) default: - writeStreamStatus(conn, StreamStatusMessage{ + writeStreamStatus(sub, StreamStatusMessage{ Type: "stream_status", Error: "unknown type (get_host, list_serial_ports, ble_scan, host_connect, host_disconnect, list_clients, set_stream, get_stream, set_input_stream, get_input_stream, set_tap_notify, get_tap_notify, set_led_ring, get_battery)", }) @@ -931,16 +934,12 @@ func hostStatusFromLink(link *managedSerial, success bool, errMsg string) HostSt } } -func writeHostStatus(conn *websocket.Conn, msg HostStatusMessage) { - writeWSJSON(conn, msg) +func writeHostStatus(sub *wsSubscriber, msg HostStatusMessage) { + sub.writeJSON(msg) } -func writeWSJSON(conn *websocket.Conn, msg any) { - data, err := json.Marshal(msg) - if err != nil { - return - } - _ = conn.WriteMessage(websocket.TextMessage, data) +func writeWSJSON(sub *wsSubscriber, msg any) { + sub.writeJSON(msg) } func serveExternalWS(conn *websocket.Conn, link *managedSerial, dash *wsHub, ctl *accelStreamCtl, tapCtl *tapNotifyCtl, hub *accelStreamHub) { @@ -953,7 +952,7 @@ func serveExternalWS(conn *websocket.Conn, link *managedSerial, dash *wsHub, ctl if err != nil { return } - handleAccelWSCommand(conn, sub, data, link, dash, ctl, tapCtl, hub) + handleAccelWSCommand(sub, data, link, dash, ctl, tapCtl, hub) } } diff --git a/goTool/ble_link.go b/goTool/ble_link.go index 36f0146..87a1229 100644 --- a/goTool/ble_link.go +++ b/goTool/ble_link.go @@ -175,7 +175,17 @@ func (b *bleLink) onNotify(buf []byte) { select { case b.respCh <- payload: default: - log.Printf("BLE: response channel full, dropping frame type=0x%02x", payload[0]) + // Prefer dropping the oldest queued frame so a late/stale reply + // does not block the matching response for the in-flight command. + select { + case <-b.respCh: + default: + } + select { + case b.respCh <- payload: + default: + log.Printf("BLE: response channel full, dropping frame type=0x%02x", payload[0]) + } } } } @@ -189,9 +199,21 @@ func (b *bleLink) exchange(cmdID byte, cmdName string) ([]byte, error) { } func (b *bleLink) exchangePayload(payload []byte, cmdName string) ([]byte, error) { + return b.exchangePayloadTimeout(payload, cmdName, readTimeout) +} + +func (b *bleLink) exchangePayloadTimeout(payload []byte, cmdName string, timeout time.Duration) ([]byte, error) { b.mu.Lock() defer b.mu.Unlock() + if len(payload) == 0 { + return nil, fmt.Errorf("empty payload") + } + expect := payload[0] + if timeout <= 0 { + timeout = readTimeout + } + frame, err := uartframe.EncodeFrame(payload) if err != nil { return nil, fmt.Errorf("encode frame: %w", err) @@ -212,16 +234,33 @@ sent: return nil, fmt.Errorf("BLE write: %w (check DIP8=BLE)", err) } - timer := time.NewTimer(readTimeout) - defer timer.Stop() - select { - case resp := <-b.respCh: - if !b.quiet { - log.Printf("BLE response (%d bytes)", len(resp)) + deadline := time.Now().Add(timeout) + for { + remain := time.Until(deadline) + if remain <= 0 { + return nil, fmt.Errorf("BLE read timeout for %s (check DIP8=BLE / -ble)", cmdName) + } + timer := time.NewTimer(remain) + select { + case resp := <-b.respCh: + timer.Stop() + if len(resp) == 0 { + continue + } + if resp[0] != expect { + // Stale reply (e.g. late BATTERY_STATUS after a short timeout). + if !b.quiet { + log.Printf("BLE: discard type=0x%02x while waiting for 0x%02x (%s)", resp[0], expect, cmdName) + } + continue + } + if !b.quiet { + log.Printf("BLE response (%d bytes)", len(resp)) + } + return resp, nil + case <-timer.C: + return nil, fmt.Errorf("BLE read timeout for %s (check DIP8=BLE / -ble)", cmdName) } - return resp, nil - case <-timer.C: - return nil, fmt.Errorf("BLE read timeout for %s (check DIP8=BLE / -ble)", cmdName) } } diff --git a/goTool/ble_measure.go b/goTool/ble_measure.go new file mode 100644 index 0000000..4f6627e --- /dev/null +++ b/goTool/ble_measure.go @@ -0,0 +1,224 @@ +package main + +import ( + "fmt" + "math" + "sort" + "time" + + pb "powerpod/gotool/pb" + + "google.golang.org/protobuf/proto" +) + +type latencyStats struct { + N int `json:"n"` + MinMs float64 `json:"min_ms"` + MedMs float64 `json:"median_ms"` + MeanMs float64 `json:"mean_ms"` + MaxMs float64 `json:"max_ms"` +} + +type bleRttResult struct { + Success bool `json:"success"` + Count int `json:"count"` + Warmup int `json:"warmup"` + Stats latencyStats `json:"stats"` + Error string `json:"error,omitempty"` +} + +type bleStreamResult struct { + Success bool `json:"success"` + Received int `json:"received"` + Requested int `json:"requested"` + PeriodMs uint32 `json:"period_ms"` + TolMs float64 `json:"tol_ms"` + OutsideTol int `json:"outside_tol"` + WindowOK bool `json:"window_ok"` + DTEsp latencyStats `json:"dt_esp_ms"` + DRPc latencyStats `json:"dr_pc_ms"` + ExtraVsFirst latencyStats `json:"extra_vs_first_ms"` + Error string `json:"error,omitempty"` +} + +func measureBleRtt(h hostConn, count, warmup int, settle time.Duration) bleRttResult { + out := bleRttResult{Count: count, Warmup: warmup} + if count < 1 { + out.Error = "count must be >= 1" + return out + } + if warmup < 0 { + out.Error = "warmup must be >= 0" + return out + } + if settle > 0 { + time.Sleep(settle) + } + + samples := make([]float64, 0, count) + total := warmup + count + for i := 0; i < total; i++ { + t0 := time.Now() + if _, err := hostGetVersion(h); err != nil { + out.Error = fmt.Sprintf("VERSION round-trip %d: %v", i, err) + return out + } + ms := float64(time.Since(t0).Microseconds()) / 1000.0 + if i >= warmup { + samples = append(samples, ms) + } + } + out.Stats = makeLatencyStats(samples) + out.Success = true + return out +} + +func measureBleStream(h hostConn, clientID, periodMs uint32, count int, tolMs, timeoutSec float64) bleStreamResult { + out := bleStreamResult{ + Requested: count, + PeriodMs: periodMs, + TolMs: tolMs, + } + if periodMs < 8 || periodMs > 16 { + out.Error = "period_ms must be 8..16" + return out + } + if count < 2 { + out.Error = "count must be >= 2" + return out + } + sc, ok := h.(hostStreamConn) + if !ok { + out.Error = "BLE stream ticks require BLE host" + return out + } + + req := &pb.AccelStreamRequest{ + Write: true, + Enable: true, + PeriodMs: periodMs, + } + if clientID == 0 { + req.AllClients = true + } else { + req.ClientId = clientID + } + resp, err := hostAccelStream(h, req) + if err != nil { + out.Error = err.Error() + return out + } + if !resp.GetSuccess() { + out.Error = "accel stream enable failed" + return out + } + + type sample struct { + espUs uint64 + pcT float64 + } + samples := make([]sample, 0, count) + deadline := time.Now().Add(time.Duration(timeoutSec * float64(time.Second))) + if timeoutSec <= 0 { + deadline = time.Now().Add(15 * time.Second) + } + + for len(samples) < count { + remain := time.Until(deadline) + if remain <= 0 { + break + } + payload, err := sc.NextStreamTick(remain) + if err != nil { + break + } + pcT := time.Now().UnixMicro() + var msg pb.UartMessage + if err := proto.Unmarshal(payload[1:], &msg); err != nil { + continue + } + tick := msg.GetBleStreamTick() + if tick == nil { + continue + } + samples = append(samples, sample{espUs: tick.GetEspTimestampUs(), pcT: float64(pcT)}) + } + + off := &pb.AccelStreamRequest{Write: true, Enable: false, AllClients: req.AllClients, ClientId: req.ClientId} + _, _ = hostAccelStream(h, off) + + out.Received = len(samples) + if len(samples) < 2 { + out.Error = fmt.Sprintf("need >=2 ticks, got %d (stream running / DIP8=BLE?)", len(samples)) + return out + } + + var dT, dR, extra []float64 + esp0 := samples[0].espUs + pc0 := samples[0].pcT + for i := 1; i < len(samples); i++ { + dT = append(dT, float64(samples[i].espUs-samples[i-1].espUs)/1000.0) + dR = append(dR, (samples[i].pcT-samples[i-1].pcT)/1000.0) + relEsp := float64(samples[i].espUs-esp0) / 1000.0 + relPc := (samples[i].pcT - pc0) / 1000.0 + extra = append(extra, relPc-relEsp) + } + + period := float64(periodMs) + outCount := 0 + for _, dt := range dT { + if math.Abs(dt-period) > tolMs { + outCount++ + } + } + out.OutsideTol = outCount + out.WindowOK = outCount*2 <= len(dT) + out.DTEsp = makeLatencyStats(dT) + out.DRPc = makeLatencyStats(dR) + out.ExtraVsFirst = makeLatencyStats(extra) + out.Success = true + return out +} + +func makeLatencyStats(v []float64) latencyStats { + if len(v) == 0 { + return latencyStats{} + } + s := append([]float64(nil), v...) + sort.Float64s(s) + return latencyStats{ + N: len(s), + MinMs: s[0], + MedMs: percentile(s, 0.5), + MeanMs: mean(s), + MaxMs: s[len(s)-1], + } +} + +func mean(v []float64) float64 { + if len(v) == 0 { + return math.NaN() + } + var s float64 + for _, x := range v { + s += x + } + return s / float64(len(v)) +} + +func percentile(sorted []float64, p float64) float64 { + if len(sorted) == 0 { + return math.NaN() + } + if len(sorted) == 1 { + return sorted[0] + } + idx := p * float64(len(sorted)-1) + lo := int(idx) + hi := lo + 1 + if hi >= len(sorted) { + return sorted[lo] + } + frac := idx - float64(lo) + return sorted[lo]*(1-frac) + sorted[hi]*frac +} diff --git a/goTool/cache_status.go b/goTool/cache_status.go index c0361a8..9406c71 100644 --- a/goTool/cache_status.go +++ b/goTool/cache_status.go @@ -57,6 +57,14 @@ func tapEventsFromCacheStatus(r *pb.CacheStatusResponse) []*pb.TapEvent { // cacheStatusFromBLEStreamPayload parses a framed BLE_STREAM_TICK payload // (type byte + protobuf UartMessage) into a CacheStatusResponse-shaped view. func cacheStatusFromBLEStreamPayload(payload []byte) (*pb.CacheStatusResponse, error) { + tick, err := parseBleStreamTickPayload(payload) + if err != nil { + return nil, err + } + return &pb.CacheStatusResponse{Clients: tick.GetClients()}, nil +} + +func parseBleStreamTickPayload(payload []byte) (*pb.BleStreamTick, error) { if len(payload) < 2 { return nil, fmt.Errorf("BLE stream tick too short") } @@ -71,5 +79,5 @@ func cacheStatusFromBLEStreamPayload(payload []byte) (*pb.CacheStatusResponse, e if tick == nil { return nil, fmt.Errorf("missing ble_stream_tick") } - return &pb.CacheStatusResponse{Clients: tick.GetClients()}, nil + return tick, nil } diff --git a/goTool/cmd_ble_rtt.go b/goTool/cmd_ble_rtt.go index 6f03f91..28a0e25 100644 --- a/goTool/cmd_ble_rtt.go +++ b/goTool/cmd_ble_rtt.go @@ -3,8 +3,6 @@ package main import ( "flag" "fmt" - "math" - "sort" "time" ) @@ -16,62 +14,13 @@ func runBleRtt(h hostConn, args []string) error { if err := fs.Parse(args); err != nil { return err } - if *count < 1 { - return fmt.Errorf("--count must be >= 1") - } - if *warmup < 0 { - return fmt.Errorf("--warmup must be >= 0") - } - if *settle > 0 { - fmt.Printf("settling %.1fs…\n", *settle) - time.Sleep(time.Duration(*settle * float64(time.Second))) + out := measureBleRtt(h, *count, *warmup, time.Duration(*settle*float64(time.Second))) + if !out.Success { + return fmt.Errorf("%s", out.Error) } - - total := *warmup + *count - samples := make([]float64, 0, *count) - for i := 0; i < total; i++ { - t0 := time.Now() - if _, err := hostGetVersion(h); err != nil { - return fmt.Errorf("VERSION round-trip %d: %w", i, err) - } - ms := float64(time.Since(t0).Microseconds()) / 1000.0 - if i >= *warmup { - samples = append(samples, ms) - } - } - - sort.Float64s(samples) - fmt.Printf("ble-rtt (VERSION): n=%d warmup=%d\n", len(samples), *warmup) + fmt.Printf("ble-rtt (VERSION): n=%d warmup=%d\n", out.Stats.N, out.Warmup) fmt.Printf(" min=%.3f median=%.3f mean=%.3f max=%.3f ms\n", - samples[0], percentile(samples, 0.5), mean(samples), samples[len(samples)-1]) + out.Stats.MinMs, out.Stats.MedMs, out.Stats.MeanMs, out.Stats.MaxMs) return nil } - -func mean(v []float64) float64 { - if len(v) == 0 { - return math.NaN() - } - var s float64 - for _, x := range v { - s += x - } - return s / float64(len(v)) -} - -func percentile(sorted []float64, p float64) float64 { - if len(sorted) == 0 { - return math.NaN() - } - if len(sorted) == 1 { - return sorted[0] - } - idx := p * float64(len(sorted)-1) - lo := int(idx) - hi := lo + 1 - if hi >= len(sorted) { - return sorted[lo] - } - frac := idx - float64(lo) - return sorted[lo]*(1-frac) + sorted[hi]*frac -} diff --git a/goTool/cmd_ble_stream.go b/goTool/cmd_ble_stream.go index f386bb7..9768301 100644 --- a/goTool/cmd_ble_stream.go +++ b/goTool/cmd_ble_stream.go @@ -3,12 +3,6 @@ package main import ( "flag" "fmt" - "math" - "sort" - "time" - - pb "powerpod/gotool/pb" - "google.golang.org/protobuf/proto" ) func runBleStream(h hostConn, args []string) error { @@ -21,108 +15,26 @@ func runBleStream(h hostConn, args []string) error { if err := fs.Parse(args); err != nil { return err } - if *periodMs < 8 || *periodMs > 16 { - return fmt.Errorf("--period must be 8..16") - } - if *count < 2 { - return fmt.Errorf("--count must be >= 2") - } - sc, ok := h.(hostStreamConn) - if !ok { - return fmt.Errorf("ble-stream requires -ble (UART has no push ticks)") - } - - req := &pb.AccelStreamRequest{ - Write: true, - Enable: true, - PeriodMs: uint32(*periodMs), - } - if *clientID == 0 { - req.AllClients = true - } else { - req.ClientId = uint32(*clientID) - } - resp, err := hostAccelStream(h, req) - if err != nil { - return err - } - if !resp.GetSuccess() { - return fmt.Errorf("accel stream enable failed") - } fmt.Printf("accel stream on; period_ms=%d; collecting %d ticks…\n", *periodMs, *count) - - type sample struct { - espUs uint64 - pcT float64 + out := measureBleStream(h, uint32(*clientID), uint32(*periodMs), *count, *tolMs, *duration) + if !out.Success { + return fmt.Errorf("%s", out.Error) } - samples := make([]sample, 0, *count) - deadline := time.Now().Add(time.Duration(*duration * float64(time.Second))) - - for len(samples) < *count { - remain := time.Until(deadline) - if remain <= 0 { - break - } - payload, err := sc.NextStreamTick(remain) - if err != nil { - break - } - pcT := time.Now().UnixMicro() - var msg pb.UartMessage - if err := proto.Unmarshal(payload[1:], &msg); err != nil { - continue - } - tick := msg.GetBleStreamTick() - if tick == nil { - continue - } - samples = append(samples, sample{espUs: tick.GetEspTimestampUs(), pcT: float64(pcT)}) - } - - // Best-effort disable - off := &pb.AccelStreamRequest{Write: true, Enable: false, AllClients: req.AllClients, ClientId: req.ClientId} - _, _ = hostAccelStream(h, off) - - if len(samples) < 2 { - return fmt.Errorf("need >=2 ticks, got %d (is stream running / DIP8=BLE?)", len(samples)) - } - - var dT, dR, extra []float64 - esp0 := samples[0].espUs - pc0 := samples[0].pcT - for i := 1; i < len(samples); i++ { - dT = append(dT, float64(samples[i].espUs-samples[i-1].espUs)/1000.0) - dR = append(dR, (samples[i].pcT-samples[i-1].pcT)/1000.0) - relEsp := float64(samples[i].espUs-esp0) / 1000.0 - relPc := (samples[i].pcT - pc0) / 1000.0 - extra = append(extra, relPc-relEsp) - } - - period := float64(*periodMs) - outCount := 0 - for _, dt := range dT { - if math.Abs(dt-period) > *tolMs { - outCount++ - } - } - windowOK := outCount*2 <= len(dT) // majority inside tolerance - - fmt.Printf("received=%d/%d period_ms=%d tol=±%.1f ms\n", len(samples), *count, *periodMs, *tolMs) - printStat("dT ESP inter-depart", dT) - printStat("dR PC inter-arrive", dR) - printStat("extra vs first", extra) - fmt.Printf("outside_tol=%d/%d window_ok=%s\n", outCount, len(dT), map[bool]string{true: "yes", false: "no"}[windowOK]) + fmt.Printf("received=%d/%d period_ms=%d tol=±%.1f ms\n", out.Received, out.Requested, out.PeriodMs, out.TolMs) + printStat("dT ESP inter-depart", out.DTEsp) + printStat("dR PC inter-arrive", out.DRPc) + printStat("extra vs first", out.ExtraVsFirst) + fmt.Printf("outside_tol=%d/%d window_ok=%s\n", out.OutsideTol, out.DTEsp.N, + map[bool]string{true: "yes", false: "no"}[out.WindowOK]) return nil } -func printStat(name string, v []float64) { - if len(v) == 0 { +func printStat(name string, s latencyStats) { + if s.N == 0 { fmt.Printf("%s: (none)\n", name) return } - s := append([]float64(nil), v...) - sort.Float64s(s) fmt.Printf("%s: n=%d min=%.3f median=%.3f mean=%.3f max=%.3f ms\n", - name, len(s), s[0], percentile(s, 0.5), mean(s), s[len(s)-1]) + name, s.N, s.MinMs, s.MedMs, s.MeanMs, s.MaxMs) } diff --git a/goTool/dashboard.go b/goTool/dashboard.go index 4e86506..36c2d4d 100644 --- a/goTool/dashboard.go +++ b/goTool/dashboard.go @@ -77,15 +77,80 @@ type DashboardState struct { SerialError string `json:"serial_error,omitempty"` /** Host: fast CACHE_STATUS poll (~16 ms) for accel + tap. */ LiveStream bool `json:"live_stream"` - Master MasterView `json:"master"` - Clients []ClientView `json:"clients"` + /** BLE live path: inter-tick stats from BleStreamTick.esp_timestamp_us. */ + BleTick *BleTickView `json:"ble_tick,omitempty"` + Master MasterView `json:"master"` + Clients []ClientView `json:"clients"` +} + +// BleTickView is median/mean inter-departure of BLE_STREAM_TICK (ESP timer). +type BleTickView struct { + PeriodMs uint32 `json:"period_ms"` + MedianMs float64 `json:"median_ms"` + MeanMs float64 `json:"mean_ms"` + MinMs float64 `json:"min_ms"` + MaxMs float64 `json:"max_ms"` + Samples int `json:"samples"` + LastEspUs uint64 `json:"last_esp_us,omitempty"` +} + +const bleTickDeltaCap = 64 + +type bleTickTracker struct { + lastEspUs uint64 + haveLast bool + periodMs uint32 + deltas [bleTickDeltaCap]float64 + next int + count int +} + +func (t *bleTickTracker) reset() { + *t = bleTickTracker{} +} + +func (t *bleTickTracker) add(espUs uint64, periodMs uint32) BleTickView { + if periodMs != 0 { + t.periodMs = periodMs + } + if t.haveLast && espUs > t.lastEspUs { + dt := float64(espUs-t.lastEspUs) / 1000.0 + t.deltas[t.next] = dt + t.next = (t.next + 1) % bleTickDeltaCap + if t.count < bleTickDeltaCap { + t.count++ + } + } + t.lastEspUs = espUs + t.haveLast = true + + samples := make([]float64, 0, t.count) + if t.count < bleTickDeltaCap { + samples = append(samples, t.deltas[:t.count]...) + } else { + samples = append(samples, t.deltas[t.next:]...) + samples = append(samples, t.deltas[:t.next]...) + } + st := makeLatencyStats(samples) + return BleTickView{ + PeriodMs: t.periodMs, + MedianMs: st.MedMs, + MeanMs: st.MeanMs, + MinMs: st.MinMs, + MaxMs: st.MaxMs, + Samples: st.N, + LastEspUs: espUs, + } } type wsHub struct { mu sync.RWMutex + writeMu sync.Mutex // gorilla websocket: one writer at a time clients map[*websocket.Conn]struct{} state DashboardState liveStream bool + bleTicks bleTickTracker + lastPush time.Time // throttle BLE tick WS pushes } func newWSHub() *wsHub { @@ -96,6 +161,10 @@ func (h *wsHub) setState(st DashboardState) { h.mu.Lock() prev := h.state st.LiveStream = prev.LiveStream + if st.BleTick == nil && prev.BleTick != nil && st.LiveStream { + cp := *prev.BleTick + st.BleTick = &cp + } st.Clients = preserveClientAccel(st.Clients, prev.Clients, st.LiveStream) st.Clients = preserveClientBattery(st.Clients, prev.Clients) st.Clients = preserveClientTap(st.Clients, prev.Clients) @@ -140,15 +209,25 @@ func (h *wsHub) unregister(c *websocket.Conn) { h.mu.Unlock() } -// writeJSON sends to one client; removes it on error or panic (closed connection). +// dropClient removes the client and closes the socket so the browser reconnects. +func (h *wsHub) dropClient(c *websocket.Conn) { + h.unregister(c) + _ = c.Close() +} + +// writeJSON sends to one client; removes+closes it on error or panic. func (h *wsHub) writeJSON(c *websocket.Conn, data []byte) { defer func() { if recover() != nil { - h.unregister(c) + h.dropClient(c) } }() - if err := c.WriteMessage(websocket.TextMessage, data); err != nil { - h.unregister(c) + h.writeMu.Lock() + _ = c.SetWriteDeadline(time.Now().Add(2 * time.Second)) + err := c.WriteMessage(websocket.TextMessage, data) + h.writeMu.Unlock() + if err != nil { + h.dropClient(c) } } @@ -417,6 +496,8 @@ func (h *wsHub) patchLiveStream(enabled bool) { st := h.state st.LiveStream = enabled if !enabled { + h.bleTicks.reset() + st.BleTick = nil for i := range st.Clients { st.Clients[i].AccelValid = false st.Clients[i].AccelX = 0 @@ -501,6 +582,49 @@ func (h *wsHub) mergeAccel(samples []*pb.AccelSample) { } } +// mergeBleStreamTick applies accel/tap from a BLE tick and updates inter-tick stats. +// State is updated every tick; WebSocket push is throttled (~50 Hz) so concurrent +// writers (poller/battery) cannot flood or race the gorilla conn. +func (h *wsHub) mergeBleStreamTick(tick *pb.BleStreamTick) { + if tick == nil || !h.liveStreamEnabled() { + return + } + cache := &pb.CacheStatusResponse{Clients: tick.GetClients()} + const minPushGap = 20 * time.Millisecond + + h.mu.Lock() + view := h.bleTicks.add(tick.GetEspTimestampUs(), tick.GetPeriodMs()) + st := h.state + st.Clients = applyAccelSamples(st.Clients, accelSamplesFromCacheStatus(cache)) + st.Clients = applyTapEvents(st.Clients, tapEventsFromCacheStatus(cache)) + st.BleTick = &view + st.UpdatedAt = time.Now().Format(time.RFC3339) + h.state = st + push := time.Since(h.lastPush) >= minPushGap + if push { + h.lastPush = time.Now() + } + var conns []*websocket.Conn + if push { + conns = make([]*websocket.Conn, 0, len(h.clients)) + for c := range h.clients { + conns = append(conns, c) + } + } + h.mu.Unlock() + + if !push { + return + } + data, err := json.Marshal(st) + if err != nil { + return + } + for _, c := range conns { + h.writeJSON(c, data) + } +} + func (h *wsHub) mergeTap(events []*pb.TapEvent) { if len(events) == 0 || !h.liveStreamEnabled() { return @@ -673,7 +797,7 @@ func runBatteryPoller(link *managedSerial, hub *wsHub, interval time.Duration, s case <-stop: return case <-ticker.C: - if !link.IsConnected() { + if !link.IsConnected() || link.PollPaused() { continue } bat, err := link.BatteryStatusPoll(&pb.BatteryStatusRequest{AllClients: true}) @@ -694,7 +818,7 @@ func runCacheStatusDashboardPoller(link *managedSerial, hub *wsHub, interval tim case <-stop: return case <-ticker.C: - if !hub.liveStreamEnabled() { + if !hub.liveStreamEnabled() || link.PollPaused() { continue } // BLE live path prefers BLE_STREAM_TICK pushes. @@ -720,7 +844,7 @@ func runBLEStreamDashboardPoller(link *managedSerial, hub *wsHub, streamCtl *acc return default: } - if link.Transport() != hostTransportBLE || !hub.liveStreamEnabled() { + if link.Transport() != hostTransportBLE || !hub.liveStreamEnabled() || link.PollPaused() { select { case <-stop: return @@ -732,12 +856,11 @@ func runBLEStreamDashboardPoller(link *managedSerial, hub *wsHub, streamCtl *acc if err != nil { continue } - cache, err := cacheStatusFromBLEStreamPayload(payload) + tick, err := parseBleStreamTickPayload(payload) if err != nil { continue } - hub.mergeAccel(accelSamplesFromCacheStatus(cache)) - hub.mergeTap(tapEventsFromCacheStatus(cache)) + hub.mergeBleStreamTick(tick) } } diff --git a/goTool/docs/API_REST.md b/goTool/docs/API_REST.md index f3548a7..5b2991a 100644 --- a/goTool/docs/API_REST.md +++ b/goTool/docs/API_REST.md @@ -140,6 +140,20 @@ Dashboard WebSocket state includes `host_connected`, `host_transport`, `host_end **OTA** (`POST /api/ota`) returns **400** unless `transport==uart` and connected. +### BLE latency tests (BLE host only) + +```http +POST /api/ble-rtt +{"count":50,"warmup":5,"settle_s":0} +→ {"success":true,"count":50,"warmup":5,"stats":{"n":50,"min_ms":…,"median_ms":…,"mean_ms":…,"max_ms":…}} + +POST /api/ble-stream +{"count":50,"period_ms":16,"tol_ms":2,"timeout_s":15,"client_id":0} +→ {"success":true,"received":50,"requested":50,"window_ok":true,"dt_esp_ms":{…},"dr_pc_ms":{…},…} +``` + +Pauses dashboard/API polling while running. Requires active BLE host (`DIP8=1`). + ### Live stream (host `CACHE_STATUS` poll ~16 ms; BLE uses `BLE_STREAM_TICK`) ```http diff --git a/goTool/host_managed.go b/goTool/host_managed.go index b13c8b4..f545ddb 100644 --- a/goTool/host_managed.go +++ b/goTool/host_managed.go @@ -33,6 +33,7 @@ type managedHost struct { sp *serialPort ble *bleLink otaActive bool + pausePoll bool // true while BLE latency tests run (pollers yield) uartPort string // remembered UART port for reconnect } @@ -291,10 +292,32 @@ func (m *managedHost) exchangePayloadPoll(payload []byte, cmdName string) ([]byt return m.exchangePayloadMode(true, payload, cmdName, readTimeout) } +func (m *managedHost) beginPausePoll() bool { + m.mu.Lock() + defer m.mu.Unlock() + if m.otaActive || m.pausePoll { + return false + } + m.pausePoll = true + return true +} + +func (m *managedHost) endPausePoll() { + m.mu.Lock() + m.pausePoll = false + m.mu.Unlock() +} + +func (m *managedHost) PollPaused() bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.pausePoll || m.otaActive +} + func (m *managedHost) exchangePayloadMode(poll bool, payload []byte, cmdName string, timeout time.Duration) ([]byte, error) { m.mu.Lock() defer m.mu.Unlock() - if m.otaActive { + if m.otaActive || (poll && m.pausePoll) { return nil, errHostBusy } switch m.transport { @@ -302,7 +325,7 @@ func (m *managedHost) exchangePayloadMode(poll bool, payload []byte, cmdName str if m.ble == nil { return nil, fmt.Errorf("BLE not connected") } - resp, err := m.ble.exchangePayload(payload, cmdName) + resp, err := m.ble.exchangePayloadTimeout(payload, cmdName, timeout) if err != nil { m.invalidateLocked(err) } diff --git a/goTool/webui/index.html b/goTool/webui/index.html index 0752ff4..c143dea 100644 --- a/goTool/webui/index.html +++ b/goTool/webui/index.html @@ -269,6 +269,89 @@ +
+
BLE-Tests
+
+

+ Host↔Master-Latenz und Stream-Tick-Fenster (wie CLI ble-rtt / ble-stream). + Dashboard-Polling pausiert während des Laufs. +

+
+
+
RTT (VERSION)
+
+
+ + +
+
+ + +
+ +
+ +
+
+
Stream-Ticks
+
+
+ + +
+
+ + +
+
+ + +
+ +
+ +
+
+
+
+
@@ -461,9 +544,9 @@

- Live-Stream startet die schnelle CACHE_STATUS-Abfrage (~16 ms). - Pro Slave Accel aktiviert den ESP-NOW-Accel-Stream; Tap-Notify (S/D/T) steuert - die Tap-Arten auf dem Slave. + Live-Stream: UART ≈ CACHE_STATUS-Poll (~16 ms); BLE ≈ + BLE_STREAM_TICK-Push. Pro Slave Accel aktiviert den ESP-NOW-Stream; + Tap-Notify (S/D/T) steuert die Tap-Arten auf dem Slave.

+ Tap alle Slaves: @@ -815,6 +901,9 @@ bleScanning: false, bleScanErr: '', hostBusy: false, + bleTestBusy: false, + bleRtt: { count: 50, warmup: 5, result: null }, + bleStream: { count: 50, period_ms: 16, tol_ms: 2, result: null }, masterDz: 100, masterDzSeeded: false, masterLogLevel: 0, @@ -867,6 +956,82 @@ isUARTHost() { return this.hostUp() && (this.state?.host_transport || 'uart') === 'uart'; }, + isBLEHost() { + return this.hostUp() && this.state?.host_transport === 'ble'; + }, + bleTickLabel() { + const t = this.state?.ble_tick; + if (!t || !t.samples) return 'BLE tick: …'; + const med = Number(t.median_ms).toFixed(2); + const period = t.period_ms != null ? t.period_ms : '—'; + return `BLE tick median ${med} ms (n=${t.samples}, period ${period} ms)`; + }, + fmtMs(v) { + if (v == null || Number.isNaN(v)) return '—'; + return Number(v).toFixed(3) + ' ms'; + }, + fmtStats(s) { + if (!s?.n) return '—'; + return `min ${this.fmtMs(s.min_ms)} · med ${this.fmtMs(s.median_ms)} · mean ${this.fmtMs(s.mean_ms)} · max ${this.fmtMs(s.max_ms)}`; + }, + async runBleRtt() { + this.bleTestBusy = 'rtt'; + this.bleRtt.result = null; + try { + const r = await fetch('/api/ble-rtt', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + count: this.bleRtt.count, + warmup: this.bleRtt.warmup, + }), + }); + const data = await r.json(); + this.bleRtt.result = data; + if (!data.success) { + this.flash(data.error || 'BLE RTT fehlgeschlagen', false); + } else { + this.flash( + `BLE RTT: mean ${Number(data.stats?.mean_ms).toFixed(3)} ms (n=${data.stats?.n})`, + true + ); + } + } catch (e) { + this.flash(String(e), false); + } finally { + this.bleTestBusy = false; + } + }, + async runBleStream() { + this.bleTestBusy = 'stream'; + this.bleStream.result = null; + try { + const r = await fetch('/api/ble-stream', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + count: this.bleStream.count, + period_ms: this.bleStream.period_ms, + tol_ms: this.bleStream.tol_ms, + timeout_s: 20, + }), + }); + const data = await r.json(); + this.bleStream.result = data; + if (!data.success) { + this.flash(data.error || 'BLE Stream-Test fehlgeschlagen', false); + } else { + this.flash( + `BLE Stream: ${data.received}/${data.requested} ticks, window_ok=${data.window_ok ? 'yes' : 'no'}`, + true + ); + } + } catch (e) { + this.flash(String(e), false); + } finally { + this.bleTestBusy = false; + } + }, hostBadge() { const t = this.state?.host_transport || 'none'; if (!this.hostUp()) return 'off'; @@ -981,9 +1146,14 @@ if (this._tapTimer) clearInterval(this._tapTimer); this._tapTimer = setInterval(() => { this._tapClock++; }, 250); const connect = () => { + if (this.ws && (this.ws.readyState === WebSocket.OPEN || + this.ws.readyState === WebSocket.CONNECTING)) { + try { this.ws.close(); } catch (_) {} + } this.ws = new WebSocket(url); this.ws.onopen = () => { this.wsConnected = true; + this._lastWsAt = Date.now(); this.refreshBattery(); this.refreshSerialPorts(); }; @@ -993,6 +1163,7 @@ }; this.ws.onmessage = (e) => { try { + this._lastWsAt = Date.now(); const msg = JSON.parse(e.data); if (msg.type === 'ota_progress') { this.applyOTAProgress(msg); @@ -1024,6 +1195,14 @@ } catch (_) {} }; }; + if (this._wsWatch) clearInterval(this._wsWatch); + // If server dropped us from the hub without closing, force reconnect. + this._wsWatch = setInterval(() => { + if (!this.wsConnected || !this.state?.live_stream) return; + if (Date.now() - (this._lastWsAt || 0) > 4000) { + try { this.ws?.close(); } catch (_) {} + } + }, 2000); connect(); }, preserveBatteryInState(prev, next) { diff --git a/main/ble_host.c b/main/ble_host.c index 910b74f..246d0c1 100644 --- a/main/ble_host.c +++ b/main/ble_host.c @@ -140,15 +140,13 @@ static void ble_request_fast_conn_params(uint16_t handle) { static void ble_optimize_link(uint16_t handle) { int rc; + /* + * Do NOT request 2M PHY here. With ESP-NOW/WiFi coexistence the controller + * can assert in llc_phy_upd (e.g. "BLE assert llc_phy_upd.c …") and the + * host link dies — no further responses. Stay on default 1M. + */ ble_request_fast_conn_params(handle); - rc = ble_gap_set_prefered_le_phy(handle, BLE_GAP_LE_PHY_2M_MASK, - BLE_GAP_LE_PHY_2M_MASK, - BLE_GAP_LE_PHY_CODED_ANY); - if (rc != 0) { - ESP_LOGW(TAG, "2M PHY request failed: %d", rc); - } - rc = ble_gap_set_data_len(handle, 251, 2120); if (rc != 0) { ESP_LOGW(TAG, "data len request failed: %d", rc); diff --git a/main/esp_now_master.c b/main/esp_now_master.c index 04fd3f0..51392ff 100644 --- a/main/esp_now_master.c +++ b/main/esp_now_master.c @@ -435,11 +435,13 @@ static void handle_accel_sample(const uint8_t mac[CLIENT_MAC_LEN], esp_err_t err = client_registry_update_accel( mac, sample->slave_id, (int16_t)sample->x, (int16_t)sample->y, (int16_t)sample->z); - if (err == ESP_ERR_NOT_FOUND) { + if (err == ESP_ERR_NOT_FOUND || err == ESP_ERR_INVALID_STATE) { + // Unknown MAC, or stream already off (in-flight samples after disable). return; } if (err != ESP_OK) { - ESP_LOGW(TAG, "accel sample id mismatch from %02x:…:%02x", mac[0], mac[5]); + ESP_LOGW(TAG, "accel sample id mismatch from %02x:…:%02x (slave_id=%lu)", + mac[0], mac[5], (unsigned long)sample->slave_id); } }