DriverVault/API Server/internal/modbus/client.go

// Package modbus is a minimal Modbus TCP client, written against the
// MODBUS Application Protocol Specification V1.1b3 and the MODBUS Messaging on
// TCP/IP Implementation Guide V1.0b rather than pulled in as a dependency: the
// whole API server is stdlib-only, so this sits beside internal/mqtt's
// hand-rolled MQTT and internal/ocpp's hand-rolled RFC 6455 WebSocket for the
// same reason.
//
// It is deliberately scoped to what the Anker SOLIX V1 EV charger needs
// (internal/plugins/builtin/ankersolix): read a contiguous block of holding
// registers, and write one register to issue a command. Concretely that means:
//
//   - Function codes 0x03 (Read Holding Registers), 0x04 (Read Input Registers)
//     and 0x06 (Write Single Register) only. The charger splits its map across
//     the first two: everything it measures or reports (20000-20100) answers on
//     FC04 and rejects FC03 with an illegal-address exception, while the
//     controls it accepts (21000-21005) are holding registers.
//   - One request in flight at a time. Modbus TCP allows pipelining by
//     transaction identifier; nothing here needs it, and a strictly synchronous
//     exchange means a desynchronised peer cannot silently mismatch replies.
//   - No reconnect and no pooling. The charger accepts at most two simultaneous
//     clients (Anker's Modbus spec, §"Activate Modbus TCP"), one of which an
//     operator may well be using for a debugging tool, so a Client is meant to
//     be opened for one exchange and closed again rather than held open.
//
// A Client is safe for concurrent use; requests are serialized.
package modbus

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"time"
)

// Defaults applied by Options.normalize when a field is left zero.
const (
	defaultUnitID         = 1
	defaultTimeout        = 5 * time.Second
	defaultConnectTimeout = 5 * time.Second
)

// ErrClosed is returned once the connection is gone.
var ErrClosed = errors.New("modbus: connection closed")

// Protocol constants. The MBAP header is seven bytes — transaction id (2),
// protocol id (2, always zero for Modbus), length (2), unit id (1) — and is
// followed by the PDU, whose first byte is the function code.
const (
	mbapLen     = 7
	protocolID  = 0
	fcReadHold  = 0x03
	fcReadInput = 0x04
	fcWriteReg  = 0x06
	excMask     = 0x80 // set on the echoed function code when the reply is an exception
	maxReadRegs = 125  // §6.3/§6.4: the most registers one read request may ask for
)

// maxPDU bounds what this client will read back from a peer. The spec's own
// ceiling is 253 bytes of PDU, so anything larger is a broken or hostile peer
// and the stream can no longer be framed.
const maxPDU = 253

// Options configures a connection.
type Options struct {
	// Address is the charger's host:port. Required. Anker's Modbus TCP server
	// listens on port 502.
	Address string
	// UnitID addresses a device behind a gateway. The charger answers directly,
	// so this is 1 unless a deployment puts a bridge in the way.
	UnitID byte
	// Timeout bounds one request/response exchange.
	Timeout time.Duration
	// ConnectTimeout bounds the TCP dial.
	ConnectTimeout time.Duration
}

func (o *Options) normalize() {
	if o.UnitID == 0 {
		o.UnitID = defaultUnitID
	}
	if o.Timeout <= 0 {
		o.Timeout = defaultTimeout
	}
	if o.ConnectTimeout <= 0 {
		o.ConnectTimeout = defaultConnectTimeout
	}
}

// Error is a Modbus exception response: the server understood the frame and
// refused it. It is distinct from a transport failure, because it says something
// about the request (a bad address, an out-of-range value) rather than the link.
type Error struct {
	Function byte
	Code     byte
}

func (e *Error) Error() string {
	return fmt.Sprintf("modbus: server rejected function 0x%02x: %s", e.Function, exceptionText(e.Code))
}

// exceptionText names the exception codes in §7 that this client can provoke.
func exceptionText(code byte) string {
	switch code {
	case 0x01:
		return "illegal function"
	case 0x02:
		return "illegal data address"
	case 0x03:
		return "illegal data value"
	case 0x04:
		return "server device failure"
	case 0x05:
		return "acknowledge (request accepted, still processing)"
	case 0x06:
		return "server device busy"
	case 0x0B:
		return "gateway target device failed to respond"
	default:
		return fmt.Sprintf("exception code 0x%02x", code)
	}
}

// Client is a connected Modbus TCP session.
type Client struct {
	conn    net.Conn
	unitID  byte
	timeout time.Duration

	mu     sync.Mutex // serializes exchanges; one request is in flight at a time
	nextTx uint16
	closed bool
}

// Connect dials the server. The returned Client owns the connection; call Close
// when done.
func Connect(ctx context.Context, opt Options) (*Client, error) {
	opt.normalize()
	if opt.Address == "" {
		return nil, errors.New("modbus: server address is required")
	}
	d := net.Dialer{Timeout: opt.ConnectTimeout}
	conn, err := d.DialContext(ctx, "tcp", opt.Address)
	if err != nil {
		return nil, fmt.Errorf("modbus: cannot reach %s: %w", opt.Address, err)
	}
	return &Client{conn: conn, unitID: opt.UnitID, timeout: opt.Timeout}, nil
}

// Close releases the connection. It is safe to call more than once, which
// matters because callers defer it and may also close on an error path.
func (c *Client) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return nil
	}
	c.closed = true
	return c.conn.Close()
}

// ReadHolding reads count consecutive holding registers starting at addr
// (function code 0x03). Register values are returned in the order read.
func (c *Client) ReadHolding(ctx context.Context, addr, count uint16) ([]uint16, error) {
	return c.readRegisters(ctx, fcReadHold, addr, count)
}

// ReadInput reads count consecutive input registers starting at addr (function
// code 0x04). It is the read the Anker charger's measurement block answers:
// the same request shape as ReadHolding, a different table on the device.
func (c *Client) ReadInput(ctx context.Context, addr, count uint16) ([]uint16, error) {
	return c.readRegisters(ctx, fcReadInput, addr, count)
}

// readRegisters is the body both reads share; FC03 and FC04 differ only in
// which table the server looks the addresses up in, not in their framing.
func (c *Client) readRegisters(ctx context.Context, fc byte, addr, count uint16) ([]uint16, error) {
	if count == 0 || count > maxReadRegs {
		return nil, fmt.Errorf("modbus: cannot read %d registers in one request (1-%d)", count, maxReadRegs)
	}
	req := make([]byte, 5)
	req[0] = fc
	binary.BigEndian.PutUint16(req[1:], addr)
	binary.BigEndian.PutUint16(req[3:], count)

	pdu, err := c.exchange(ctx, req)
	if err != nil {
		return nil, err
	}
	// Reply PDU: function code, byte count, then two bytes per register.
	if len(pdu) < 2 {
		return nil, errors.New("modbus: truncated read reply")
	}
	n := int(pdu[1])
	if n != int(count)*2 || len(pdu) < 2+n {
		return nil, fmt.Errorf("modbus: read reply carries %d bytes, want %d", len(pdu)-2, int(count)*2)
	}
	out := make([]uint16, count)
	for i := range out {
		out[i] = binary.BigEndian.Uint16(pdu[2+i*2:])
	}
	return out, nil
}

// WriteSingle writes one holding register (function code 0x06). The server
// echoes the address and value it applied, which is verified here: a charger
// that clamps a value reports the clamp rather than silently diverging.
func (c *Client) WriteSingle(ctx context.Context, addr, value uint16) error {
	req := make([]byte, 5)
	req[0] = fcWriteReg
	binary.BigEndian.PutUint16(req[1:], addr)
	binary.BigEndian.PutUint16(req[3:], value)

	pdu, err := c.exchange(ctx, req)
	if err != nil {
		return err
	}
	if len(pdu) < 5 {
		return errors.New("modbus: truncated write reply")
	}
	if got := binary.BigEndian.Uint16(pdu[1:]); got != addr {
		return fmt.Errorf("modbus: write reply echoes register %d, want %d", got, addr)
	}
	if got := binary.BigEndian.Uint16(pdu[3:]); got != value {
		return fmt.Errorf("modbus: register %d accepted %d, not the %d requested", addr, got, value)
	}
	return nil
}

// exchange sends one PDU and returns the reply PDU, having checked the MBAP
// header, the transaction identifier and the exception bit.
func (c *Client) exchange(ctx context.Context, pdu []byte) ([]byte, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return nil, ErrClosed
	}

	// A deadline drawn from both the context and the configured timeout, so a
	// cancelled caller and a mute peer are both bounded.
	deadline := time.Now().Add(c.timeout)
	if d, ok := ctx.Deadline(); ok && d.Before(deadline) {
		deadline = d
	}
	if err := c.conn.SetDeadline(deadline); err != nil {
		return nil, err
	}

	c.nextTx++
	tx := c.nextTx

	frame := make([]byte, mbapLen+len(pdu))
	binary.BigEndian.PutUint16(frame[0:], tx)
	binary.BigEndian.PutUint16(frame[2:], protocolID)
	// Length counts the unit id plus the PDU, per the messaging guide §3.1.3.
	binary.BigEndian.PutUint16(frame[4:], uint16(len(pdu)+1))
	frame[6] = c.unitID
	copy(frame[mbapLen:], pdu)

	if _, err := c.conn.Write(frame); err != nil {
		return nil, fmt.Errorf("modbus: sending request: %w", err)
	}

	var header [mbapLen]byte
	if _, err := io.ReadFull(c.conn, header[:]); err != nil {
		return nil, fmt.Errorf("modbus: reading reply header: %w", err)
	}
	if got := binary.BigEndian.Uint16(header[2:]); got != protocolID {
		return nil, fmt.Errorf("modbus: reply has protocol id %d, want 0", got)
	}
	length := int(binary.BigEndian.Uint16(header[4:]))
	if length < 2 || length-1 > maxPDU {
		return nil, fmt.Errorf("modbus: reply announces %d bytes, which is not a valid PDU length", length)
	}
	body := make([]byte, length-1) // less the unit id, already in the header
	if _, err := io.ReadFull(c.conn, body); err != nil {
		return nil, fmt.Errorf("modbus: reading reply: %w", err)
	}
	if got := binary.BigEndian.Uint16(header[0:]); got != tx {
		// Synchronous exchange, so this is a desynchronised stream rather than a
		// late reply to something else; the connection is no longer trustworthy.
		return nil, fmt.Errorf("modbus: reply is for transaction %d, want %d", got, tx)
	}

	fc := body[0]
	if fc&excMask != 0 {
		if len(body) < 2 {
			return nil, errors.New("modbus: exception reply carries no code")
		}
		return nil, &Error{Function: fc &^ excMask, Code: body[1]}
	}
	if fc != pdu[0] {
		return nil, fmt.Errorf("modbus: reply is for function 0x%02x, want 0x%02x", fc, pdu[0])
	}
	return body, nil
}