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

Add CACHE_STATUS UART poll and dashboard live stream.

Browse Source 
Combine cached accel and tap in one low-overhead master command for ~16 ms
host polling. The dashboard uses a single live-stream toggle plus per-slave
accel-stream controls; fix live_stream state so polling is not cleared every
slow client refresh.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
skruecken
2026-05-29 20:57:54 +02:00
co-authored by Cursor
parent a8d4d42920
commit f512936d97
21 changed files with 764 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goTool/dashboard.go
+84 -122
View File
@@ -44,8 +44,6 @@ type ClientView struct {
TapNotifySingle bool `json:"tap_notify_single"`
TapNotifyDouble bool `json:"tap_notify_double"`
TapNotifyTriple bool `json:"tap_notify_triple"`
/** Host-side: poll master tap cache for this slave (~16 ms). */
TapReceive bool `json:"tap_receive"`
LastTap string `json:"last_tap,omitempty"`
LastTapAt int64 `json:"last_tap_at,omitempty"`
Lipo1 lipoReadingJSON `json:"lipo1"`
@@ -59,14 +57,17 @@ type DashboardState struct {
UARTConnected bool `json:"uart_connected"`
SerialOK bool `json:"serial_ok"`
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"`
}
type wsHub struct {
mu sync.RWMutex
clients map[*websocket.Conn]struct{}
state DashboardState
mu sync.RWMutex
clients map[*websocket.Conn]struct{}
state DashboardState
liveStream bool
}
func newWSHub() *wsHub {
@@ -76,9 +77,9 @@ func newWSHub() *wsHub {
func (h *wsHub) setState(st DashboardState) {
h.mu.Lock()
prev := h.state
st.Clients = preserveClientAccel(st.Clients, prev.Clients)
st.LiveStream = prev.LiveStream
st.Clients = preserveClientAccel(st.Clients, prev.Clients, st.LiveStream)
st.Clients = preserveClientBattery(st.Clients, prev.Clients)
st.Clients = preserveClientTapReceive(st.Clients, prev.Clients)
st.Clients = preserveClientTap(st.Clients, prev.Clients)
if !st.Master.Lipo1.Valid && !st.Master.Lipo2.Valid {
if prev.Master.Lipo1.Valid || prev.Master.Lipo2.Valid {
@@ -87,6 +88,7 @@ func (h *wsHub) setState(st DashboardState) {
st.Master.BatteryAgeMs = prev.Master.BatteryAgeMs
}
}
h.liveStream = st.LiveStream
h.state = st
conns := make([]*websocket.Conn, 0, len(h.clients))
for c := range h.clients {
@@ -150,7 +152,7 @@ func applyAccelSamples(clients []ClientView, samples []*pb.AccelSample) []Client
return out
}
func preserveClientAccel(newClients, oldClients []ClientView) []ClientView {
func preserveClientAccel(newClients, oldClients []ClientView, liveStream bool) []ClientView {
if len(oldClients) == 0 {
return newClients
}
@@ -161,7 +163,15 @@ func preserveClientAccel(newClients, oldClients []ClientView) []ClientView {
out := make([]ClientView, len(newClients))
for i, c := range newClients {
out[i] = c
if !c.AccelStream {
if !liveStream && !c.AccelStream {
continue
}
if liveStream && !c.AccelStream {
out[i].AccelValid = false
out[i].AccelX = 0
out[i].AccelY = 0
out[i].AccelZ = 0
out[i].AccelAgeMs = 0
continue
}
prev, ok := oldByID[c.ID]
@@ -224,45 +234,6 @@ func anyClientTapNotify(clients []ClientView) bool {
return false
}
func anyClientTapReceive(clients []ClientView) bool {
for _, c := range clients {
if c.TapReceive {
return true
}
}
return false
}
func preserveClientTapReceive(newClients, oldClients []ClientView) []ClientView {
if len(oldClients) == 0 {
return newClients
}
oldByID := make(map[uint32]ClientView, len(oldClients))
for _, c := range oldClients {
oldByID[c.ID] = c
}
out := make([]ClientView, len(newClients))
for i, c := range newClients {
out[i] = c
prev, ok := oldByID[c.ID]
if !ok {
continue
}
out[i].TapReceive = prev.TapReceive
if !prev.TapReceive {
continue
}
if c.LastTap == "" && prev.LastTap != "" {
cutoff := time.Now().Add(-clientTapDisplayMinMs * time.Millisecond).UnixMilli()
if prev.LastTapAt >= cutoff {
out[i].LastTap = prev.LastTap
out[i].LastTapAt = prev.LastTapAt
}
}
}
return out
}
func tapKindLabelPB(k pb.TapKind) string {
switch k {
case pb.TapKind_TAP_SINGLE:
@@ -293,7 +264,7 @@ func applyTapEvents(clients []ClientView, events []*pb.TapEvent) []ClientView {
out := make([]ClientView, len(clients))
for i, c := range clients {
out[i] = c
if !c.TapReceive {
if !clientTapNotifyAny(c) {
continue
}
e, ok := byID[c.ID]
@@ -308,6 +279,10 @@ func applyTapEvents(clients []ClientView, events []*pb.TapEvent) []ClientView {
const clientTapDisplayMinMs = 2000
func clientTapNotifyAny(c ClientView) bool {
return c.TapNotifySingle || c.TapNotifyDouble || c.TapNotifyTriple
}
func preserveClientTap(newClients, oldClients []ClientView) []ClientView {
if len(oldClients) == 0 {
return newClients
@@ -379,10 +354,51 @@ func (h *wsHub) anyTapNotifyEnabled() bool {
return anyClientTapNotify(h.state.Clients)
}
func (h *wsHub) anyTapReceiveEnabled() bool {
func (h *wsHub) liveStreamEnabled() bool {
h.mu.RLock()
defer h.mu.RUnlock()
return anyClientTapReceive(h.state.Clients)
return h.liveStream
}
func (h *wsHub) snapshotClients() []ClientView {
h.mu.RLock()
defer h.mu.RUnlock()
out := make([]ClientView, len(h.state.Clients))
copy(out, h.state.Clients)
return out
}
// patchLiveStream toggles host CACHE_STATUS polling (~16 ms).
func (h *wsHub) patchLiveStream(enabled bool) {
h.mu.Lock()
h.liveStream = enabled
st := h.state
st.LiveStream = enabled
if !enabled {
for i := range st.Clients {
st.Clients[i].AccelValid = false
st.Clients[i].AccelX = 0
st.Clients[i].AccelY = 0
st.Clients[i].AccelZ = 0
st.Clients[i].AccelAgeMs = 0
st.Clients[i].LastTap = ""
st.Clients[i].LastTapAt = 0
}
}
h.state = st
conns := make([]*websocket.Conn, 0, len(h.clients))
for c := range h.clients {
conns = append(conns, c)
}
h.mu.Unlock()
data, err := json.Marshal(st)
if err != nil {
return
}
for _, c := range conns {
_ = c.WriteMessage(websocket.TextMessage, data)
}
}
// patchClientTapNotify updates tap notify flags immediately (e.g. after REST) and pushes WS.
@@ -420,6 +436,9 @@ func (h *wsHub) patchClientTapNotify(clientID uint32, single, doubleTap, triple
// mergeAccel updates cached accel on clients and pushes state to dashboard WebSockets.
func (h *wsHub) mergeAccel(samples []*pb.AccelSample) {
if !h.liveStreamEnabled() {
return
}
h.mu.Lock()
st := h.state
st.Clients = applyAccelSamples(st.Clients, samples)
@@ -440,38 +459,8 @@ func (h *wsHub) mergeAccel(samples []*pb.AccelSample) {
}
}
func (h *wsHub) patchClientTapReceive(clientID uint32, enabled bool) {
h.mu.Lock()
for i := range h.state.Clients {
if h.state.Clients[i].ID != clientID {
continue
}
h.state.Clients[i].TapReceive = enabled
if !enabled {
h.state.Clients[i].LastTap = ""
h.state.Clients[i].LastTapAt = 0
}
break
}
st := h.state
st.UpdatedAt = time.Now().Format(time.RFC3339)
conns := make([]*websocket.Conn, 0, len(h.clients))
for c := range h.clients {
conns = append(conns, c)
}
h.mu.Unlock()
data, err := json.Marshal(st)
if err != nil {
return
}
for _, c := range conns {
_ = c.WriteMessage(websocket.TextMessage, data)
}
}
func (h *wsHub) mergeTap(events []*pb.TapEvent) {
if len(events) == 0 {
if len(events) == 0 || !h.liveStreamEnabled() {
return
}
h.mu.Lock()
@@ -565,25 +554,16 @@ func pollDashboard(link *managedSerial, portName string, last *DashboardState, s
st.Clients = append(st.Clients, cv)
}
applyBatteryToState(link, &st)
if anyClientAccelStream(st.Clients) {
for i := range st.Clients {
if !st.Clients[i].AccelStream {
continue
}
if dz, err := readDeadzonePoll(link, st.Clients[i].ID); err == nil {
st.Clients[i].Deadzone = dz
}
}
if snap, err := link.readAccelSnapshotPoll(0); err == nil {
st.Clients = applyAccelSamples(st.Clients, snap.GetSamples())
}
} else {
if last == nil || !last.LiveStream {
for i, c := range clients {
if dz, err := readDeadzonePoll(link, c.GetId()); err == nil {
st.Clients[i].Deadzone = dz
}
}
}
if last != nil {
st.LiveStream = last.LiveStream
}
if streamCtl != nil {
streamCtl.SyncFromClients(st.Clients)
}
@@ -647,7 +627,7 @@ func runBatteryPoller(link *managedSerial, hub *wsHub, interval time.Duration, s
}
}
func runAccelDashboardPoller(link *managedSerial, hub *wsHub, interval time.Duration, stop <-chan struct{}) {
func runCacheStatusDashboardPoller(link *managedSerial, hub *wsHub, interval time.Duration, stop <-chan struct{}) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
@@ -656,35 +636,15 @@ func runAccelDashboardPoller(link *managedSerial, hub *wsHub, interval time.Dura
case <-stop:
return
case <-ticker.C:
if hub.clientCount() == 0 || !hub.anyAccelStreamEnabled() {
if !hub.liveStreamEnabled() {
continue
}
snap, err := link.readAccelSnapshotPoll(0)
cache, err := link.readCacheStatusPoll()
if err != nil {
continue
}
hub.mergeAccel(snap.GetSamples())
}
}
}
func runTapDashboardPoller(link *managedSerial, hub *wsHub, interval time.Duration, stop <-chan struct{}) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-stop:
return
case <-ticker.C:
if hub.clientCount() == 0 || !hub.anyTapReceiveEnabled() {
continue
}
snap, err := link.readTapSnapshotPoll(0)
if err != nil {
continue
}
hub.mergeTap(snap.GetEvents())
hub.mergeAccel(cache.GetAccel())
hub.mergeTap(cache.GetTaps())
}
}
}
@@ -752,14 +712,16 @@ func runPoller(link *managedSerial, portName string, hub *wsHub, streamCtl *acce
var lastGood DashboardState
publish := func() {
st := pollDashboard(link, portName, &lastGood, streamCtl, tapCtl)
hub.setState(st)
if st.UARTConnected && st.SerialOK {
lastGood = st
hub.mu.RLock()
lastGood = hub.state
hub.mu.RUnlock()
}
if st.UARTConnected && !uartUp {
log.Printf("UART %s connected", portName)
}
uartUp = st.UARTConnected
hub.setState(st)
}
publish()