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

Enable host OTA over BLE and harden ESP-NOW slave distribution.

Browse Source 
Allow the same staged firmware upload path on BLE as UART, fix distribute stack/event-group crashes, and recover sticky slave OTA sessions instead of failing with err=4.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
skruecken
2026-11-01 20:42:14 +01:00
co-authored by Cursor
parent 567dab6d0a
commit ae8ef40820
16 changed files with 400 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goTool/ota_upload.go
+124 -102
View File
@@ -6,14 +6,14 @@ import (
"google.golang.org/protobuf/proto"
uartframe "powerpod/gotool/uart"
"powerpod/gotool/pb"
uartframe "powerpod/gotool/uart"
)
const (
otaHostChunkSize = 200
otaFlashBlockSize = 4096
otaPrepareTimeout = 120 * time.Second
otaHostChunkSize = 200
otaFlashBlockSize = 4096
otaPrepareTimeout = 120 * time.Second
otaDefaultTimeout = 15 * time.Second
otaStatusPollTimeout = 3 * time.Second
otaDistReadTimeout = 400 * time.Millisecond
@@ -21,8 +21,12 @@ const (
otaDistQueryTimeout = 2 * time.Second
otaDistEmitMinInterval = 150 * time.Millisecond
// Pace host chunks so the master UART RX ring is not overrun (~20 frames/block).
otaHostChunkPace = 3 * time.Millisecond
otaBlockMaxRetries = 3
otaHostChunkPaceUART = 3 * time.Millisecond
// Slightly slower for BLE WriteWithoutResponse flow control.
otaHostChunkPaceBLE = 6 * time.Millisecond
otaBlockMaxRetries = 3
// 200 B OTA_PAYLOAD → ~216 B framed; needs ATT MTU >= 219.
otaMinATTMTU = 219
)
const (
@@ -37,6 +41,14 @@ const (
otaDistTimeout = 45 * time.Minute
)
// otaTransport is the host IO used by the OTA protocol (UART or BLE).
type otaTransport interface {
writeMsg(msg *pb.UartMessage) error
readFrame(timeout time.Duration) ([]byte, error)
flushInput()
chunkPace() time.Duration
}
// OtaSlaveDetail is per-slave ESP-NOW OTA state from OTA_SLAVE_PROGRESS.
type OtaSlaveDetail struct {
BytesWritten uint32 `json:"bytes_written"`
@@ -47,20 +59,20 @@ type OtaSlaveDetail struct {
// OTAProgress is pushed to the dashboard during web uploads.
type OTAProgress struct {
Type string `json:"type"` // always "ota_progress"
Phase string `json:"phase"`
Step string `json:"step,omitempty"` // master, slaves
Percent int `json:"percent"`
MasterPercent int `json:"master_percent,omitempty"`
MasterDone bool `json:"master_done,omitempty"`
Message string `json:"message"`
MasterMessage string `json:"master_message,omitempty"`
Bytes uint32 `json:"bytes_written,omitempty"`
Slot uint32 `json:"target_slot,omitempty"`
Slaves uint32 `json:"slaves,omitempty"`
ImageSize uint32 `json:"image_size,omitempty"`
SlaveProgress map[uint32]uint32 `json:"slave_progress,omitempty"` // client_id -> bytes
SlaveDetails map[uint32]OtaSlaveDetail `json:"slave_details,omitempty"`
Type string `json:"type"` // always "ota_progress"
Phase string `json:"phase"`
Step string `json:"step,omitempty"` // master, slaves
Percent int `json:"percent"`
MasterPercent int `json:"master_percent,omitempty"`
MasterDone bool `json:"master_done,omitempty"`
Message string `json:"message"`
MasterMessage string `json:"master_message,omitempty"`
Bytes uint32 `json:"bytes_written,omitempty"`
Slot uint32 `json:"target_slot,omitempty"`
Slaves uint32 `json:"slaves,omitempty"`
ImageSize uint32 `json:"image_size,omitempty"`
SlaveProgress map[uint32]uint32 `json:"slave_progress,omitempty"` // client_id -> bytes
SlaveDetails map[uint32]OtaSlaveDetail `json:"slave_details,omitempty"`
}
type otaProgressFn func(OTAProgress)
@@ -80,27 +92,52 @@ func runOTAUpload(m *managedSerial, firmware []byte, onProgress otaProgressFn) e
Percent: 0, Message: msg, MasterMessage: msg,
})
}
push("preparing", "UART wird vorbereitet…")
push("preparing", "Host wird für OTA vorbereitet…")
// Block until the UART is free, then hold m.mu for the entire upload so
// dashboard/API polling cannot interleave on the serial port.
// Hold m.mu for the entire upload so dashboard/API polling cannot interleave.
m.mu.Lock()
if m.otaActive {
m.mu.Unlock()
return errOTAInProgress
}
m.otaActive = true
if m.sp == nil {
if err := m.openLocked(); err != nil {
var t otaTransport
switch m.transport {
case hostTransportBLE:
if m.ble == nil {
m.otaActive = false
m.mu.Unlock()
err := fmt.Errorf("BLE not connected")
push("error", err.Error())
return err
}
if err := m.ble.ensureOTAMTU(); err != nil {
m.otaActive = false
m.mu.Unlock()
push("error", err.Error())
return err
}
t = m.ble
case hostTransportUART, hostTransportNone:
if m.sp == nil {
if err := m.openLocked(); err != nil {
m.otaActive = false
m.mu.Unlock()
push("error", err.Error())
return err
}
}
t = m.sp
default:
m.otaActive = false
m.mu.Unlock()
err := fmt.Errorf("not connected")
push("error", err.Error())
return err
}
sp := m.sp
err := runOTAOnPortUnlocked(sp, firmware, onProgress)
err := runOTAUnlocked(t, firmware, onProgress)
if err != nil {
m.invalidateLocked(err)
}
@@ -109,7 +146,7 @@ func runOTAUpload(m *managedSerial, firmware []byte, onProgress otaProgressFn) e
return err
}
func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgressFn) error {
func runOTAUnlocked(t otaTransport, firmware []byte, onProgress otaProgressFn) error {
if len(firmware) == 0 {
return fmt.Errorf("empty firmware")
}
@@ -151,16 +188,11 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
onProgress(p)
}
if err := sp.port.SetReadTimeout(readTimeout); err != nil {
notify("error", "", 0, err.Error())
return err
}
notify("preparing", otaStepMaster, 0, fmt.Sprintf("Master: OTA start (%d bytes)…", imageSize))
flushSerialInput(sp)
t.flushInput()
if err := writeUartMessage(sp, &pb.UartMessage{
if err := t.writeMsg(&pb.UartMessage{
Type: pb.MessageType_OTA_START,
Payload: &pb.UartMessage_OtaStart{
OtaStart: &pb.OtaStartPayload{TotalSize: uint32(imageSize)},
@@ -170,13 +202,7 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
return err
}
if err := sp.port.SetReadTimeout(otaPrepareTimeout); err != nil {
notify("error", "", 0, err.Error())
return err
}
defer func() { _ = sp.port.SetReadTimeout(readTimeout) }()
ready, err := waitOtaStatus(sp, otaStReady, otaPrepareTimeout, func(msg string) {
ready, err := waitOtaStatus(t, otaStReady, otaPrepareTimeout, func(msg string) {
notify("preparing", otaStepMaster, 2, msg)
})
if err != nil {
@@ -185,11 +211,7 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
}
notify("ready", otaStepMaster, 5, fmt.Sprintf("Master: Slot %d bereit", ready.GetTargetSlot()))
if err := sp.port.SetReadTimeout(otaDefaultTimeout); err != nil {
notify("error", "", 0, err.Error())
return err
}
pace := t.chunkPace()
var seq uint32
for offset := 0; offset < imageSize; {
blockStart := offset
@@ -208,7 +230,7 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
}
chunk := firmware[offset : offset+n]
if err := writeUartMessage(sp, &pb.UartMessage{
if err := t.writeMsg(&pb.UartMessage{
Type: pb.MessageType_OTA_PAYLOAD,
Payload: &pb.UartMessage_OtaPayload{
OtaPayload: &pb.OtaPayload{Seq: seq, Data: chunk},
@@ -216,7 +238,7 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
}); err != nil {
return false, err
}
time.Sleep(otaHostChunkPace)
time.Sleep(pace)
seq++
offset += n
bytesInBlock += n
@@ -252,7 +274,7 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
return err
}
}
st, ackErr = waitOtaStatus(sp, otaStBlockAck, otaDefaultTimeout, nil)
st, ackErr = waitOtaStatus(t, otaStBlockAck, otaDefaultTimeout, nil)
if ackErr == nil {
break
}
@@ -272,10 +294,10 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
}
masterPct = 100
masterMsg = "Master: UART-Upload abgeschlossen"
masterMsg = "Master: Upload abgeschlossen"
notify("uploading", otaStepMaster, 100, masterMsg)
if err := writeUartMessage(sp, &pb.UartMessage{
if err := t.writeMsg(&pb.UartMessage{
Type: pb.MessageType_OTA_END,
Payload: &pb.UartMessage_OtaEnd{
OtaEnd: &pb.OtaEndPayload{},
@@ -314,7 +336,8 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
}
}
if slaveCount == 0 {
return "Keine verfügbaren Slaves — Verteilung übersprungen", 0, 0
// Do not claim "skipped" mid-flight — details often arrive after the first query.
return "ESP-NOW: warte auf Slave-Fortschritt…", 0, 0
}
return fmt.Sprintf("ESP-NOW: %d / %d bytes (%d Slaves)",
aggBytes, imageSize, slaveCount), aggBytes, slaveCount
@@ -334,7 +357,7 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
return
}
lastQuery = time.Now()
prog, err := queryOtaSlaveProgressLocked(sp, 0, onDistStatus, otaDistQueryTimeout)
prog, err := queryOtaSlaveProgress(t, 0, onDistStatus, otaDistQueryTimeout)
if err != nil {
if len(slaveDetails) > 0 {
emitSlaveThrottled(true)
@@ -356,13 +379,13 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
lastQuery = time.Time{} // first query immediately when distribution starts
querySlaveProgress()
st, err := waitOtaComplete(sp, otaDistTimeout, pushSlaveDist, onWaitTick, otaDistReadTimeout)
st, err := waitOtaComplete(t, otaDistTimeout, pushSlaveDist, onWaitTick, otaDistReadTimeout)
if err != nil {
notify("error", "", 0, err.Error())
return err
}
if prog, err := queryOtaSlaveProgressLocked(sp, 0, nil, otaDistQueryTimeout); err == nil {
if prog, err := queryOtaSlaveProgress(t, 0, nil, otaDistQueryTimeout); err == nil {
mergeSlaveProgressResponse(prog, slaveBytes, slaveDetails)
}
notify("done", "", 100,
@@ -378,13 +401,11 @@ func runOTAOnPortUnlocked(sp *serialPort, firmware []byte, onProgress otaProgres
}
// QueryOtaSlaveProgress queries the master for per-slave ESP-NOW OTA progress.
func QueryOtaSlaveProgress(sp *serialPort, clientID uint32) (*pb.OtaSlaveProgressResponse, error) {
sp.mu.Lock()
defer sp.mu.Unlock()
return queryOtaSlaveProgressLocked(sp, clientID, nil, otaDefaultTimeout)
func QueryOtaSlaveProgress(t otaTransport, clientID uint32) (*pb.OtaSlaveProgressResponse, error) {
return queryOtaSlaveProgress(t, clientID, nil, otaDefaultTimeout)
}
func queryOtaSlaveProgressLocked(sp *serialPort, clientID uint32,
func queryOtaSlaveProgress(t otaTransport, clientID uint32,
onStatus func(*pb.OtaStatusPayload), queryTimeout time.Duration) (*pb.OtaSlaveProgressResponse, error) {
req := &pb.UartMessage{
Type: pb.MessageType_OTA_SLAVE_PROGRESS,
@@ -394,14 +415,14 @@ func queryOtaSlaveProgressLocked(sp *serialPort, clientID uint32,
},
},
}
if err := writeUartMessage(sp, req); err != nil {
if err := t.writeMsg(req); err != nil {
return nil, err
}
if queryTimeout <= 0 {
queryTimeout = otaDefaultTimeout
}
deadline := time.Now().Add(queryTimeout)
msg, err := readUartMessageUntil(sp, deadline, pb.MessageType_OTA_SLAVE_PROGRESS, onStatus, otaDistReadTimeout)
msg, err := readUartMessageUntil(t, deadline, pb.MessageType_OTA_SLAVE_PROGRESS, onStatus, otaDistReadTimeout)
if err != nil {
return nil, err
}
@@ -434,7 +455,7 @@ func applyDistributingOtaStatus(st *pb.OtaStatusPayload, imageSize int,
details[id] = d
}
func readUartMessageUntil(sp *serialPort, deadline time.Time, want pb.MessageType,
func readUartMessageUntil(t otaTransport, deadline time.Time, want pb.MessageType,
onStatus func(*pb.OtaStatusPayload), readChunk time.Duration) (*pb.UartMessage, error) {
if readChunk <= 0 {
readChunk = otaStatusPollTimeout
@@ -447,10 +468,7 @@ func readUartMessageUntil(sp *serialPort, deadline time.Time, want pb.MessageTyp
if wait > readChunk {
wait = readChunk
}
if err := sp.port.SetReadTimeout(wait); err != nil {
return nil, err
}
payload, err := uartframe.ReadFrame(sp.port, nil, wait)
payload, err := t.readFrame(wait)
if err != nil {
return nil, err
}
@@ -505,7 +523,7 @@ func copySlaveMap(m map[uint32]uint32) map[uint32]uint32 {
return out
}
func waitOtaComplete(sp *serialPort, timeout time.Duration,
func waitOtaComplete(t otaTransport, timeout time.Duration,
onDistributing func(*pb.OtaStatusPayload), onInterval func(),
readTimeout time.Duration) (*pb.OtaStatusPayload, error) {
if readTimeout <= 0 {
@@ -520,10 +538,7 @@ func waitOtaComplete(sp *serialPort, timeout time.Duration,
if readWait > readTimeout {
readWait = readTimeout
}
if err := sp.port.SetReadTimeout(readWait); err != nil {
return nil, err
}
st, err := readOtaStatus(sp)
st, err := readOtaStatus(t, readWait)
if err != nil {
if onInterval != nil {
onInterval()
@@ -548,16 +563,7 @@ func waitOtaComplete(sp *serialPort, timeout time.Duration,
}
}
func writeUartMessage(sp *serialPort, msg *pb.UartMessage) error {
frame, err := encodeUartMessage(msg)
if err != nil {
return err
}
_, err = sp.port.Write(frame)
return err
}
func waitOtaStatus(sp *serialPort, want uint32, timeout time.Duration, onPreparing func(string)) (*pb.OtaStatusPayload, error) {
func waitOtaStatus(t otaTransport, want uint32, timeout time.Duration, onPreparing func(string)) (*pb.OtaStatusPayload, error) {
deadline := time.Now().Add(timeout)
for {
if time.Now().After(deadline) {
@@ -567,10 +573,7 @@ func waitOtaStatus(sp *serialPort, want uint32, timeout time.Duration, onPrepari
if readWait > otaStatusPollTimeout {
readWait = otaStatusPollTimeout
}
if err := sp.port.SetReadTimeout(readWait); err != nil {
return nil, err
}
payload, err := uartframe.ReadFrame(sp.port, nil, readWait)
payload, err := t.readFrame(readWait)
if err != nil {
continue
}
@@ -595,8 +598,8 @@ func waitOtaStatus(sp *serialPort, want uint32, timeout time.Duration, onPrepari
}
}
func readOtaStatus(sp *serialPort) (*pb.OtaStatusPayload, error) {
payload, err := uartframe.ReadFrame(sp.port, nil, readTimeout)
func readOtaStatus(t otaTransport, timeout time.Duration) (*pb.OtaStatusPayload, error) {
payload, err := t.readFrame(timeout)
if err != nil {
return nil, fmt.Errorf("read response: %w", err)
}
@@ -623,8 +626,37 @@ func encodeUartMessage(msg *pb.UartMessage) ([]byte, error) {
return uartframe.EncodeFrame(payload)
}
// flushSerialInput drops stale RX bytes (not full frames — avoids ReadFrame blocking).
func flushSerialInput(sp *serialPort) {
func decodeUartPayload(payload []byte) (*pb.UartMessage, error) {
if len(payload) == 0 {
return nil, fmt.Errorf("empty response")
}
var msg pb.UartMessage
if err := proto.Unmarshal(payload[1:], &msg); err != nil {
return nil, err
}
msg.Type = pb.MessageType(payload[0])
return &msg, nil
}
// --- UART otaTransport ---
func (sp *serialPort) writeMsg(msg *pb.UartMessage) error {
frame, err := encodeUartMessage(msg)
if err != nil {
return err
}
_, err = sp.port.Write(frame)
return err
}
func (sp *serialPort) readFrame(timeout time.Duration) ([]byte, error) {
if err := sp.port.SetReadTimeout(timeout); err != nil {
return nil, err
}
return uartframe.ReadFrame(sp.port, nil, timeout)
}
func (sp *serialPort) flushInput() {
if sp == nil {
return
}
@@ -639,14 +671,4 @@ func flushSerialInput(sp *serialPort) {
}
}
func decodeUartPayload(payload []byte) (*pb.UartMessage, error) {
if len(payload) == 0 {
return nil, fmt.Errorf("empty response")
}
var msg pb.UartMessage
if err := proto.Unmarshal(payload[1:], &msg); err != nil {
return nil, err
}
msg.Type = pb.MessageType(payload[0])
return &msg, nil
}
func (sp *serialPort) chunkPace() time.Duration { return otaHostChunkPaceUART }