aa55 udp: add delay option to space inverter sends (#30832)

plugin/aa55/aa55udp.go

@@ -26,7 +26,8 @@ type AA55UDP struct {
 	offset   int    // byte offset into the response payload (0 for register reads)
 	decode   string // int32be | uint32be | uint32nan | int16be | uint16be | float32be
 	scale    float64
-	cacheKey []byte // precomputed cache key (remoteAddr/pdu); nil disables caching
+	cacheKey []byte        // precomputed cache key (remoteAddr/pdu); nil disables caching
+	delay    time.Duration // minimum gap between sends to the inverter (0 disables)
 }
 
 // Block describes the enclosing register block to fetch in block-read mode.
@@ -80,7 +81,7 @@ func buildReadConfig(id int, register, count uint16, block *Block) (readConfig,
 // New constructs an AA55UDP from a high-level configuration. It validates
 // decode, resolves the read mode (register vs block), and wraps the conn.
 // The caller is responsible for dialling conn.
-func New(log *util.Logger, conn *net.UDPConn, id int, register, count uint16, block *Block, decode string, scale float64) (*AA55UDP, error) {
+func New(log *util.Logger, conn *net.UDPConn, id int, register, count uint16, block *Block, decode string, scale float64, delay time.Duration) (*AA55UDP, error) {
 	if err := validateDecode(decode); err != nil {
 		return nil, err
 	}
@@ -93,6 +94,7 @@ func New(log *util.Logger, conn *net.UDPConn, id int, register, count uint16, bl
 		conn:   conn,
 		decode: decode,
 		scale:  scale,
+		delay:  delay,
 		pdu:    cfg.pdu,
 		offset: cfg.offset,
 	}
@@ -157,8 +159,16 @@ func (p *AA55UDP) exchange() ([]byte, error) {
 	return stripHeader(raw)
 }
 
-// sendRecv sends packet over p.conn and returns the raw response bytes.
+// sendRecv sends packet over p.conn and returns the raw response bytes. When a
+// delay is set it serializes and spaces exchanges to the same inverter.
 func (p *AA55UDP) sendRecv(packet []byte) ([]byte, error) {
+	if p.delay > 0 {
+		g := pace.gate(p.conn.RemoteAddr().String())
+		g.mu.Lock()
+		defer g.mu.Unlock()
+		g.wait(p.delay)
+	}
+
 	p.log.TRACE.Printf("send to %s: %x", p.conn.RemoteAddr(), packet)
 
 	if _, err := p.conn.Write(packet); err != nil {

plugin/aa55/pacer.go

@@ -0,0 +1,43 @@
+package aa55
+
+import (
+	"sync"
+	"time"
+)
+
+// pace spaces sends per inverter. Each source dials its own UDP connection, so
+// the gap is enforced per remote address at package level, not per connection.
+var pace = &pacer{gates: make(map[string]*inverterGate)}
+
+type pacer struct {
+	mu    sync.Mutex
+	gates map[string]*inverterGate
+}
+
+// inverterGate serializes exchanges to one inverter and tracks its last send.
+type inverterGate struct {
+	mu   sync.Mutex
+	last time.Time
+}
+
+// gate returns the shared gate for the inverter at addr, creating it on first use.
+func (p *pacer) gate(addr string) *inverterGate {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	g := p.gates[addr]
+	if g == nil {
+		g = &inverterGate{}
+		p.gates[addr] = g
+	}
+	return g
+}
+
+// wait blocks until delay has elapsed since the previous send to this inverter,
+// then records the new send time. The caller must hold g.mu.
+func (g *inverterGate) wait(delay time.Duration) {
+	if d := time.Until(g.last.Add(delay)); d > 0 {
+		time.Sleep(d)
+	}
+	g.last = time.Now()
+}

plugin/aa55/pacer_test.go

@@ -0,0 +1,27 @@
+package aa55
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPacerGatePerInverter(t *testing.T) {
+	p := &pacer{gates: make(map[string]*inverterGate)}
+
+	g := p.gate("1.2.3.4:8899")
+	assert.Same(t, g, p.gate("1.2.3.4:8899"), "same inverter shares one gate")
+	assert.NotSame(t, g, p.gate("5.6.7.8:8899"), "different inverters get separate gates")
+}
+
+func TestGateSpacing(t *testing.T) {
+	g := &inverterGate{}
+	const delay = 30 * time.Millisecond
+
+	g.wait(delay) // first send: no prior, returns immediately
+	start := time.Now()
+	g.wait(delay) // second send: must wait ~delay since the first
+
+	assert.GreaterOrEqual(t, time.Since(start), delay/2, "second send is spaced from the first")
+}

plugin/aa55udp.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net"
 	"net/netip"
+	"time"
 
 	"github.com/evcc-io/evcc/plugin/aa55"
 	"github.com/evcc-io/evcc/util"
@@ -39,6 +40,7 @@ func init() {
 //	  count:    125          # block length (registers)
 //	decode:   int32be        # int32be | uint32be | uint32nan | int16be | uint16be | float32be
 //	scale:    1.0            # optional multiplier (default 1.0)
+//	delay:    100ms          # optional min gap between sends to one inverter (0 disables)
 func NewAA55UDPFromConfig(ctx context.Context, other map[string]any) (Plugin, error) {
 	cc := struct {
 		Host     string
@@ -48,6 +50,7 @@ func NewAA55UDPFromConfig(ctx context.Context, other map[string]any) (Plugin, er
 		Block    *aa55.Block
 		Decode   string
 		Scale    float64
+		Delay    time.Duration
 	}{
 		Id:    int(aa55.InverterAddr),
 		Count: 2,
@@ -69,7 +72,7 @@ func NewAA55UDPFromConfig(ctx context.Context, other map[string]any) (Plugin, er
 		return nil, err
 	}
 
-	res, err := aa55.New(util.NewLogger("aa55udp"), conn, cc.Id, cc.Register, cc.Count, cc.Block, cc.Decode, cc.Scale)
+	res, err := aa55.New(util.NewLogger("aa55udp"), conn, cc.Id, cc.Register, cc.Count, cc.Block, cc.Decode, cc.Scale, cc.Delay)
 	if err != nil {
 		return nil, fmt.Errorf("aa55udp: %w", err)
 	}