skruecken/powerpods
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden BLE host link and expose latency tools in the dashboard.

Browse Source 
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 <cursoragent@cursor.com>
This commit is contained in:
skruecken
2026-10-23 23:27:39 +02:00
co-authored by Cursor
parent aa67edfe94
commit af20635047
14 changed files with 893 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goTool/api_stream.go
+99 -100
View File
@@ -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)
}
}