diff --git a/.changeset/modbus-proxy.md b/.changeset/modbus-proxy.md
new file mode 100644
index 000000000..3475df7d5
--- /dev/null
+++ b/.changeset/modbus-proxy.md
@@ -0,0 +1,5 @@
+---
+"ftw": minor
+---
+
+Optional Modbus TCP proxy. Drivers share one socket per host:port, and other LAN integrations can talk to that same session through FTW. Off by default; writes stay blocked unless you opt in.
diff --git a/config.example.yaml b/config.example.yaml
index 43c431467..8f1f938fa 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -42,6 +42,7 @@ drivers:
host: 192.168.1.10
port: 502
unit_id: 1
+ # proxy_listen: ":1502" # only needed when modbus_proxy is on and this is not the only Modbus device
# Pixii PowerShaper. When troubleshooting mode is enabled in Settings,
# Pixii also exposes calibration/control status and setpoint readback metrics.
@@ -143,6 +144,21 @@ homeassistant:
password: homeems
publish_interval_s: 5
+# Modbus TCP proxy. FTW already holds the inverter's single socket; other
+# integrations (Home Assistant's native Modbus / Sungrow / SolarEdge
+# integrations, Node-RED, …) share it by talking to this box instead of the
+# device. Off by default: the listener has no Modbus authentication, and
+# writes would bypass FTW's control loop.
+# modbus_proxy:
+# enabled: true
+# listen: ":1502" # used when the site has one unique Modbus TCP endpoint
+# allow_write: false # keep off unless you trust every host on the LAN
+#
+# Two inverters (different host:port) each need their own listen address:
+# capabilities.modbus.proxy_listen: ":1502" / ":1503"
+# Docker host-networking already publishes 1502; other compose files need
+# the port mapped.
+
# Calendar-based planner constraints (#498). FTW hosts its OWN in-process,
# pure-Go CalDAV server (emersion/go-webdav, MIT — no sidecar, works in a
# single container incl. a Home Assistant add-on; objects persist in state.db)
diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go
index ddc0803c5..0f19bdc3c 100644
--- a/go/cmd/ftw/main.go
+++ b/go/cmd/ftw/main.go
@@ -626,8 +626,33 @@ func main() {
reg.MQTTFactory = func(name string, c *config.MQTTConfig) (drivers.MQTTCap, error) {
return mqttcli.DialWithOptions(c.Host, c.Port, c.Username, c.Password, "ftw-"+name, c.AllowUnverifiedLocal)
}
+ modbusEngine := modbuscli.NewEngine()
reg.ModbusFactory = func(name string, c *config.ModbusConfig) (drivers.ModbusCap, error) {
- return modbuscli.DialWithOptions(c.Host, c.Port, c.UnitID, c.AllowUnverifiedLocal)
+ return modbusEngine.Open(c.Host, c.Port, c.UnitID, c.AllowUnverifiedLocal)
+ }
+ if cfg.ModbusProxy.On() {
+ binds, err := cfg.ModbusProxyBinds()
+ if err != nil {
+ slog.Error("modbus proxy not started", "err", err)
+ } else if len(binds) == 0 {
+ slog.Warn("modbus proxy enabled but no Modbus TCP drivers to expose")
+ } else {
+ mbBinds := make([]modbuscli.Bind, 0, len(binds))
+ for _, b := range binds {
+ mbBinds = append(mbBinds, modbuscli.Bind{
+ Listen: b.Listen,
+ Host: b.Host,
+ Port: b.Port,
+ AllowUnverifiedLocal: b.AllowUnverifiedLocal,
+ })
+ }
+ proxy, err := modbusEngine.Listen(mbBinds, cfg.ModbusProxy.AllowWrite)
+ if err != nil {
+ slog.Error("modbus proxy listen failed", "err", err)
+ } else {
+ defer proxy.Close()
+ }
+ }
}
reg.SerialFactory = func(name string, c *config.SerialConfig) (drivers.SerialCap, error) {
return drivers.OpenSerial(c)
diff --git a/go/internal/config/config.go b/go/internal/config/config.go
index 628e654a3..24c70a634 100644
--- a/go/internal/config/config.go
+++ b/go/internal/config/config.go
@@ -28,6 +28,7 @@ type Config struct {
Drivers []Driver `yaml:"drivers" json:"drivers"`
API API `yaml:"api" json:"api"`
HomeAssistant *HomeAssistant `yaml:"homeassistant,omitempty" json:"homeassistant,omitempty"`
+ ModbusProxy *ModbusProxy `yaml:"modbus_proxy,omitempty" json:"modbus_proxy,omitempty"`
State *StateConf `yaml:"state,omitempty" json:"state,omitempty"`
Price *Price `yaml:"price,omitempty" json:"price,omitempty"`
Weather *Weather `yaml:"weather,omitempty" json:"weather,omitempty"`
@@ -924,6 +925,11 @@ type ModbusConfig struct {
Host string `yaml:"host" json:"host"`
Port int `yaml:"port,omitempty" json:"port,omitempty"` // default 502
UnitID int `yaml:"unit_id,omitempty" json:"unit_id,omitempty"` // default 1
+ // ProxyListen is the local Modbus TCP address FTW binds for this
+ // backend when modbus_proxy is enabled. Required when the site has
+ // more than one unique host:port; a single-endpoint site uses
+ // modbus_proxy.listen.
+ ProxyListen string `yaml:"proxy_listen,omitempty" json:"proxy_listen,omitempty"`
// AllowUnverifiedLocal is copied from capabilities.allow_unverified_local
// by the core before this config reaches the transport factory. It is
// runtime-only and never comes from this nested YAML block.
@@ -1607,6 +1613,9 @@ func applyDefaults(c *Config) {
c.HomeAssistant.PublishIntervalS = 5
}
}
+ if c.ModbusProxy != nil && strings.TrimSpace(c.ModbusProxy.Listen) == "" {
+ c.ModbusProxy.Listen = DefaultModbusProxyListen
+ }
// Backfill for configs that predate notifications: — lands a
// populated-but-disabled stub so upgrading an existing install
// lights up the Notifications tab with the defaults instead of an
@@ -1701,6 +1710,9 @@ func (c *Config) Validate() error {
if err := c.CalDAV.Validate(); err != nil {
return err
}
+ if err := c.validateModbusProxy(); err != nil {
+ return err
+ }
if err := c.FleetPing.Validate(); err != nil {
return err
}
diff --git a/go/internal/config/modbus_proxy.go b/go/internal/config/modbus_proxy.go
new file mode 100644
index 000000000..207245dbb
--- /dev/null
+++ b/go/internal/config/modbus_proxy.go
@@ -0,0 +1,205 @@
+package config
+
+import (
+ "fmt"
+ "net"
+ "reflect"
+ "strconv"
+ "strings"
+)
+
+// DefaultModbusProxyListen is the bind used when the site has exactly one
+// Modbus TCP endpoint and the driver does not set proxy_listen.
+const DefaultModbusProxyListen = ":1502"
+
+// ModbusProxy exposes driver Modbus TCP sessions on the LAN so other
+// integrations can share the socket FTW already holds. Off by default:
+// Modbus TCP has no authentication, and writes bypass the control loop.
+type ModbusProxy struct {
+ Enabled bool `yaml:"enabled" json:"enabled"`
+ Listen string `yaml:"listen,omitempty" json:"listen,omitempty"`
+ AllowWrite bool `yaml:"allow_write,omitempty" json:"allow_write,omitempty"`
+}
+
+// ModbusProxyBind is one listener attached to a driver Modbus TCP endpoint.
+type ModbusProxyBind struct {
+ Listen string
+ Host string
+ Port int
+ AllowUnverifiedLocal bool
+}
+
+// On reports whether the proxy should bind. Nil-safe.
+func (p *ModbusProxy) On() bool {
+ return p != nil && p.Enabled
+}
+
+// ListenAddr is the default bind for a single-endpoint site.
+func (p *ModbusProxy) ListenAddr() string {
+ if p != nil && strings.TrimSpace(p.Listen) != "" {
+ return strings.TrimSpace(p.Listen)
+ }
+ return DefaultModbusProxyListen
+}
+
+func (c *Config) validateModbusProxy() error {
+ if c == nil || !c.ModbusProxy.On() {
+ return nil
+ }
+ if _, err := NormalizeListenAddr(c.ModbusProxy.ListenAddr()); err != nil {
+ return fmt.Errorf("modbus_proxy.listen: %w", err)
+ }
+ if _, err := c.ModbusProxyBinds(); err != nil {
+ return err
+ }
+ return nil
+}
+
+// ModbusProxyBinds is the listen/backend map the runtime engine should serve.
+func (c *Config) ModbusProxyBinds() ([]ModbusProxyBind, error) {
+ if c == nil || !c.ModbusProxy.On() {
+ return nil, nil
+ }
+ type ep struct {
+ host, listen string
+ port int
+ allowUnverified bool
+ drivers []string
+ }
+ byKey := map[string]*ep{}
+ order := []string{}
+ for _, d := range c.Drivers {
+ if d.Disabled {
+ continue
+ }
+ mb := d.EffectiveModbus()
+ if mb == nil || strings.TrimSpace(mb.Host) == "" {
+ continue
+ }
+ port := mb.Port
+ if port == 0 {
+ port = 502
+ }
+ key := net.JoinHostPort(mb.Host, strconv.Itoa(port))
+ e := byKey[key]
+ if e == nil {
+ e = &ep{host: mb.Host, port: port, allowUnverified: d.Capabilities.AllowUnverifiedLocal}
+ byKey[key] = e
+ order = append(order, key)
+ }
+ if d.Capabilities.AllowUnverifiedLocal {
+ e.allowUnverified = true
+ }
+ e.drivers = append(e.drivers, d.Name)
+ pl := strings.TrimSpace(mb.ProxyListen)
+ if pl == "" {
+ continue
+ }
+ norm, err := NormalizeListenAddr(pl)
+ if err != nil {
+ return nil, fmt.Errorf("driver %q: proxy_listen: %w", d.Name, err)
+ }
+ if e.listen != "" && e.listen != norm {
+ return nil, fmt.Errorf("modbus_proxy: endpoint %s has conflicting proxy_listen (%s vs %s)", key, e.listen, norm)
+ }
+ e.listen = norm
+ }
+
+ if len(order) == 0 {
+ return nil, nil
+ }
+ if len(order) > 1 {
+ for _, key := range order {
+ if byKey[key].listen == "" {
+ return nil, fmt.Errorf("modbus_proxy: multiple Modbus endpoints; set capabilities.modbus.proxy_listen on each (missing for %s, drivers %s)", key, strings.Join(byKey[key].drivers, ", "))
+ }
+ }
+ }
+ defListen, err := NormalizeListenAddr(c.ModbusProxy.ListenAddr())
+ if err != nil {
+ return nil, fmt.Errorf("modbus_proxy.listen: %w", err)
+ }
+
+ used := map[string]string{}
+ out := make([]ModbusProxyBind, 0, len(order))
+ for _, key := range order {
+ e := byKey[key]
+ listen := e.listen
+ if listen == "" {
+ listen = defListen
+ }
+ if other := used[listen]; other != "" {
+ return nil, fmt.Errorf("modbus_proxy: listen %s used by both %s and %s", listen, other, key)
+ }
+ used[listen] = key
+ out = append(out, ModbusProxyBind{
+ Listen: listen,
+ Host: e.host,
+ Port: e.port,
+ AllowUnverifiedLocal: e.allowUnverified,
+ })
+ }
+ return out, nil
+}
+
+// NormalizeListenAddr accepts ":1502", "1502", "0.0.0.0:1502".
+func NormalizeListenAddr(s string) (string, error) {
+ s = strings.TrimSpace(s)
+ if s == "" {
+ s = DefaultModbusProxyListen
+ }
+ if !strings.Contains(s, ":") {
+ s = ":" + s
+ }
+ host, port, err := net.SplitHostPort(s)
+ if err != nil {
+ return "", fmt.Errorf("invalid listen address %q", s)
+ }
+ p, err := strconv.Atoi(port)
+ if err != nil || p < 1 || p > 65535 {
+ return "", fmt.Errorf("invalid listen port in %q", s)
+ }
+ return net.JoinHostPort(host, port), nil
+}
+
+func modbusProxyRestartReasons(oldCfg, newCfg *Config) []string {
+ var reasons []string
+ if !reflect.DeepEqual(oldCfg.ModbusProxy, newCfg.ModbusProxy) {
+ reasons = append(reasons, "modbus_proxy — TCP listener binds at startup")
+ }
+ if oldCfg.ModbusProxy.On() || newCfg.ModbusProxy.On() {
+ if !reflect.DeepEqual(modbusProxySignature(oldCfg), modbusProxySignature(newCfg)) {
+ reasons = append(reasons, "modbus_proxy endpoints — driver Modbus host/port/listen feeds the proxy at startup")
+ }
+ }
+ return reasons
+}
+
+type proxySig struct {
+ Host, Listen string
+ Port int
+ AllowUnverified bool
+}
+
+func modbusProxySignature(c *Config) []proxySig {
+ if c == nil {
+ return nil
+ }
+ var out []proxySig
+ for _, d := range c.Drivers {
+ if d.Disabled {
+ continue
+ }
+ mb := d.EffectiveModbus()
+ if mb == nil {
+ continue
+ }
+ out = append(out, proxySig{
+ Host: mb.Host,
+ Port: mb.Port,
+ Listen: mb.ProxyListen,
+ AllowUnverified: d.Capabilities.AllowUnverifiedLocal,
+ })
+ }
+ return out
+}
diff --git a/go/internal/config/modbus_proxy_test.go b/go/internal/config/modbus_proxy_test.go
new file mode 100644
index 000000000..dc65277b0
--- /dev/null
+++ b/go/internal/config/modbus_proxy_test.go
@@ -0,0 +1,101 @@
+package config
+
+import (
+ "strings"
+ "testing"
+)
+
+func TestModbusProxyBindsSingleEndpointUsesDefaultListen(t *testing.T) {
+ c := &Config{
+ ModbusProxy: &ModbusProxy{Enabled: true},
+ Drivers: []Driver{{
+ Name: "sungrow",
+ Capabilities: Capabilities{Modbus: &ModbusConfig{Host: "192.168.1.10", Port: 502, UnitID: 1}},
+ }},
+ }
+ binds, err := c.ModbusProxyBinds()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(binds) != 1 || binds[0].Listen != DefaultModbusProxyListen || binds[0].Host != "192.168.1.10" || binds[0].Port != 502 {
+ t.Fatalf("binds = %+v", binds)
+ }
+}
+
+func TestModbusProxyBindsRequiresPerEndpointListenWhenMultiple(t *testing.T) {
+ c := &Config{
+ ModbusProxy: &ModbusProxy{Enabled: true, Listen: ":1502"},
+ Drivers: []Driver{
+ {Name: "a", Capabilities: Capabilities{Modbus: &ModbusConfig{Host: "10.0.0.1", Port: 502}}},
+ {Name: "b", Capabilities: Capabilities{Modbus: &ModbusConfig{Host: "10.0.0.2", Port: 502}}},
+ },
+ }
+ _, err := c.ModbusProxyBinds()
+ if err == nil || !strings.Contains(err.Error(), "proxy_listen") {
+ t.Fatalf("err = %v, want proxy_listen required", err)
+ }
+ c.Drivers[0].Capabilities.Modbus.ProxyListen = ":1502"
+ c.Drivers[1].Capabilities.Modbus.ProxyListen = ":1503"
+ binds, err := c.ModbusProxyBinds()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(binds) != 2 || binds[0].Listen != ":1502" || binds[1].Listen != ":1503" {
+ t.Fatalf("binds = %+v", binds)
+ }
+}
+
+func TestModbusProxyBindsSharesOneListenForSameHostPort(t *testing.T) {
+ c := &Config{
+ ModbusProxy: &ModbusProxy{Enabled: true},
+ Drivers: []Driver{
+ {Name: "meter", Capabilities: Capabilities{Modbus: &ModbusConfig{Host: "10.0.0.5", Port: 502, UnitID: 1}}},
+ {Name: "inverter", Capabilities: Capabilities{Modbus: &ModbusConfig{Host: "10.0.0.5", Port: 502, UnitID: 2}}},
+ },
+ }
+ binds, err := c.ModbusProxyBinds()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(binds) != 1 {
+ t.Fatalf("binds = %+v, want one shared backend", binds)
+ }
+}
+
+func TestModbusProxyDisabledIsNoop(t *testing.T) {
+ c := &Config{
+ Drivers: []Driver{{
+ Name: "sungrow",
+ Capabilities: Capabilities{Modbus: &ModbusConfig{Host: "192.168.1.10", Port: 502}},
+ }},
+ }
+ binds, err := c.ModbusProxyBinds()
+ if err != nil || len(binds) != 0 {
+ t.Fatalf("disabled proxy binds = %v err = %v", binds, err)
+ }
+}
+
+func TestNormalizeListenAddr(t *testing.T) {
+ got, err := NormalizeListenAddr("1502")
+ if err != nil || got != ":1502" {
+ t.Fatalf("1502 -> %q %v", got, err)
+ }
+ if _, err := NormalizeListenAddr("not-a-port"); err == nil {
+ t.Fatal("expected error")
+ }
+}
+
+func TestParseModbusProxyYAML(t *testing.T) {
+ c, err := Parse([]byte(minimalYAML+`
+modbus_proxy:
+ enabled: true
+ listen: ":1502"
+ allow_write: false
+`), "/tmp")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !c.ModbusProxy.On() || c.ModbusProxy.Listen != ":1502" || c.ModbusProxy.AllowWrite {
+ t.Fatalf("parsed %+v", c.ModbusProxy)
+ }
+}
diff --git a/go/internal/config/restart_required.go b/go/internal/config/restart_required.go
index 48e12ce67..ec290d1bb 100644
--- a/go/internal/config/restart_required.go
+++ b/go/internal/config/restart_required.go
@@ -16,7 +16,7 @@ import "reflect"
// {pv_rated_w, latitude, longitude}, fleet_ping.enabled, and
// home_assistant.* reload live.
// - Everything else (api.port, state.path, price.*, planner.*, nova.*,
-// ev_charger.*, caldav.*, weather.provider/arrays,
+// ev_charger.*, caldav.*, modbus_proxy.*, weather.provider/arrays,
// site.control_interval_s, site.watchdog_timeout_s, site.smoothing_alpha,
// site.gain) needs the binary restarted to take effect.
//
@@ -52,6 +52,7 @@ func RestartRequiredFor(oldCfg, newCfg *Config) []string {
if oldCfg.API.Port != newCfg.API.Port {
reasons = append(reasons, "api.port — HTTP server binds the port at startup")
}
+ reasons = append(reasons, modbusProxyRestartReasons(oldCfg, newCfg)...)
// homeassistant.* is hot-reloadable via (*ha.Bridge).Reload; see the
// applier in cmd/ftw/main.go.
if !pointerEqual(oldCfg.State, newCfg.State) {
diff --git a/go/internal/config/restart_required_test.go b/go/internal/config/restart_required_test.go
index 0a3f1286e..ebb715da9 100644
--- a/go/internal/config/restart_required_test.go
+++ b/go/internal/config/restart_required_test.go
@@ -86,6 +86,9 @@ func TestRestartRequiredFor_BootSections(t *testing.T) {
{"caldav credentials changed", func(c *Config) {
c.CalDAV = &CalDAV{Enabled: true, Username: "calendar-user", Password: "rotated"}
}, "caldav"},
+ {"modbus proxy enabled", func(c *Config) {
+ c.ModbusProxy = &ModbusProxy{Enabled: true, Listen: ":1502"}
+ }, "modbus_proxy"},
{"weather provider change", func(c *Config) {
c.Weather = &Weather{Provider: "open_meteo", Latitude: 59, Longitude: 18}
}, "weather"},
diff --git a/go/internal/modbus/client.go b/go/internal/modbus/client.go
index 7b06a7165..8246cc7e5 100644
--- a/go/internal/modbus/client.go
+++ b/go/internal/modbus/client.go
@@ -145,11 +145,16 @@ func (c *Capability) Close() error {
// Read — implements drivers.ModbusCap. Reconnects once on transport error.
func (c *Capability) Read(addr, count uint16, kind int32) ([]uint16, error) {
+ return c.readAs(c.unitID, addr, count, kind)
+}
+
+func (c *Capability) readAs(unitID int, addr, count uint16, kind int32) ([]uint16, error) {
c.mu.Lock()
defer c.mu.Unlock()
if err := c.ensureClient(); err != nil {
return nil, err
}
+ c.applyUnit(unitID)
var fc byte
switch kind {
case drivers.ModbusInput:
@@ -171,6 +176,7 @@ func (c *Capability) Read(addr, count uint16, kind int32) ([]uint16, error) {
if rerr := c.prepareTransportRetry(); rerr != nil {
return nil, fmt.Errorf("read after reconnect: %w (original: %v)", rerr, err)
}
+ c.applyUnit(unitID)
regs, err = c.client.ReadRegisters(addr, count, fc)
c.finishRequest(err)
return regs, markTransport(err)
@@ -192,11 +198,16 @@ func markTransport(err error) error {
// WriteSingle — implements drivers.ModbusCap. Reconnects once on transport error.
func (c *Capability) WriteSingle(addr, value uint16) error {
+ return c.writeSingleAs(c.unitID, addr, value)
+}
+
+func (c *Capability) writeSingleAs(unitID int, addr, value uint16) error {
c.mu.Lock()
defer c.mu.Unlock()
if err := c.ensureClient(); err != nil {
return err
}
+ c.applyUnit(unitID)
err := c.client.WriteRegister(addr, value)
if err == nil {
c.noteLiveResponse()
@@ -209,6 +220,7 @@ func (c *Capability) WriteSingle(addr, value uint16) error {
if rerr := c.prepareTransportRetry(); rerr != nil {
return fmt.Errorf("write after reconnect: %w (original: %v)", rerr, err)
}
+ c.applyUnit(unitID)
err = c.client.WriteRegister(addr, value)
c.finishRequest(err)
return err
@@ -216,11 +228,16 @@ func (c *Capability) WriteSingle(addr, value uint16) error {
// WriteMulti — implements drivers.ModbusCap. Reconnects once on transport error.
func (c *Capability) WriteMulti(addr uint16, values []uint16) error {
+ return c.writeMultiAs(c.unitID, addr, values)
+}
+
+func (c *Capability) writeMultiAs(unitID int, addr uint16, values []uint16) error {
c.mu.Lock()
defer c.mu.Unlock()
if err := c.ensureClient(); err != nil {
return err
}
+ c.applyUnit(unitID)
err := c.client.WriteRegisters(addr, values)
if err == nil {
c.noteLiveResponse()
@@ -233,11 +250,45 @@ func (c *Capability) WriteMulti(addr uint16, values []uint16) error {
if rerr := c.prepareTransportRetry(); rerr != nil {
return fmt.Errorf("write-multi after reconnect: %w (original: %v)", rerr, err)
}
+ c.applyUnit(unitID)
err = c.client.WriteRegisters(addr, values)
c.finishRequest(err)
return err
}
+func (c *Capability) applyUnit(unitID int) {
+ if c.client == nil || unitID <= 0 {
+ return
+ }
+ c.client.SetUnitId(uint8(unitID))
+}
+
+// executePDU runs one raw PDU on this connection. Exception PDUs are
+// returned as data so a proxy can forward them. Transport errors follow
+// the same reconnect-once policy as Read.
+func (c *Capability) executePDU(unitID uint8, pdu []byte) ([]byte, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if err := c.ensureClient(); err != nil {
+ return nil, err
+ }
+ res, err := c.client.roundTrip(unitID, pdu)
+ if err == nil {
+ c.noteLiveResponse()
+ return res, nil
+ }
+ if !isTransportError(err) {
+ c.noteLiveResponse()
+ return res, err
+ }
+ if rerr := c.prepareTransportRetry(); rerr != nil {
+ return nil, fmt.Errorf("pdu after reconnect: %w (original: %v)", rerr, err)
+ }
+ res, err = c.client.roundTrip(unitID, pdu)
+ c.finishRequest(err)
+ return res, markTransport(err)
+}
+
func (c *Capability) ensureClient() error {
if c.client != nil {
return nil
diff --git a/go/internal/modbus/engine.go b/go/internal/modbus/engine.go
new file mode 100644
index 000000000..68f7e0edf
--- /dev/null
+++ b/go/internal/modbus/engine.go
@@ -0,0 +1,182 @@
+package modbus
+
+import (
+ "errors"
+ "log/slog"
+ "net"
+ "strconv"
+ "sync"
+ "sync/atomic"
+
+ "github.com/srcfl/ftw/go/internal/drivers"
+)
+
+// Engine owns Modbus TCP sessions keyed by host:port. Drivers, the
+// debug probe, fingerprinting and the optional LAN proxy all go through
+// it, so several Lua drivers can poll the same inverter (or different
+// unit IDs behind one RS-485 gateway) without each opening a socket the
+// device will not accept.
+type Engine struct {
+ mu sync.Mutex
+ sessions map[string]*leased
+}
+
+type leased struct {
+ cap *Capability
+ refs int
+}
+
+// NewEngine returns an empty session pool. One Engine lives for the
+// process; Dial stays a private connection for tests and one-shot probes
+// that must not join the pool.
+func NewEngine() *Engine {
+ return &Engine{sessions: make(map[string]*leased)}
+}
+
+// Open returns a ModbusCap on the shared session for host:port. unitID is
+// applied per request so two handles on the same socket can address
+// different slaves. The socket stays up until every handle (drivers and
+// the proxy pin) has Closed.
+func (e *Engine) Open(host string, port, unitID int, allowUnverifiedLocal bool) (drivers.ModbusCap, error) {
+ if e == nil {
+ return nil, errors.New("modbus engine is nil")
+ }
+ if err := validateEndpoint(host, port, unitID); err != nil {
+ return nil, err
+ }
+ key := sessionKey(host, port)
+
+ e.mu.Lock()
+ if s := e.sessions[key]; s != nil {
+ s.refs++
+ refs := s.refs
+ cap := s.cap
+ e.mu.Unlock()
+ if refs == 2 {
+ slog.Info("modbus session shared", "addr", key, "refs", refs)
+ }
+ return newHandle(e, key, unitID, cap), nil
+ }
+ e.mu.Unlock()
+
+ cap, err := DialWithOptions(host, port, unitID, allowUnverifiedLocal)
+ if err != nil {
+ return nil, err
+ }
+
+ e.mu.Lock()
+ if s := e.sessions[key]; s != nil {
+ s.refs++
+ refs := s.refs
+ existing := s.cap
+ e.mu.Unlock()
+ _ = cap.Close()
+ if refs == 2 {
+ slog.Info("modbus session shared", "addr", key, "refs", refs)
+ }
+ return newHandle(e, key, unitID, existing), nil
+ }
+ e.sessions[key] = &leased{cap: cap, refs: 1}
+ e.mu.Unlock()
+ return newHandle(e, key, unitID, cap), nil
+}
+
+func (e *Engine) release(key string) error {
+ e.mu.Lock()
+ s := e.sessions[key]
+ if s == nil {
+ e.mu.Unlock()
+ return nil
+ }
+ s.refs--
+ if s.refs > 0 {
+ e.mu.Unlock()
+ return nil
+ }
+ delete(e.sessions, key)
+ cap := s.cap
+ e.mu.Unlock()
+ if cap == nil {
+ return nil
+ }
+ return cap.Close()
+}
+
+func (e *Engine) lookup(key string) *Capability {
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ s := e.sessions[key]
+ if s == nil {
+ return nil
+ }
+ return s.cap
+}
+
+func (e *Engine) sessionCount() int {
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ return len(e.sessions)
+}
+
+func (e *Engine) refCount(key string) int {
+ e.mu.Lock()
+ defer e.mu.Unlock()
+ s := e.sessions[key]
+ if s == nil {
+ return 0
+ }
+ return s.refs
+}
+
+func sessionKey(host string, port int) string {
+ return net.JoinHostPort(host, strconv.Itoa(port))
+}
+
+type handle struct {
+ engine *Engine
+ key string
+ unitID int
+ cap *Capability
+ closed atomic.Bool
+}
+
+func newHandle(e *Engine, key string, unitID int, cap *Capability) *handle {
+ return &handle{engine: e, key: key, unitID: unitID, cap: cap}
+}
+
+func (h *handle) Read(addr, count uint16, kind int32) ([]uint16, error) {
+ if err := h.alive(); err != nil {
+ return nil, err
+ }
+ return h.cap.readAs(h.unitID, addr, count, kind)
+}
+
+func (h *handle) WriteSingle(addr, value uint16) error {
+ if err := h.alive(); err != nil {
+ return err
+ }
+ return h.cap.writeSingleAs(h.unitID, addr, value)
+}
+
+func (h *handle) WriteMulti(addr uint16, values []uint16) error {
+ if err := h.alive(); err != nil {
+ return err
+ }
+ return h.cap.writeMultiAs(h.unitID, addr, values)
+}
+
+func (h *handle) Close() error {
+ if !h.closed.CompareAndSwap(false, true) {
+ return nil
+ }
+ return h.engine.release(h.key)
+}
+
+func (h *handle) alive() error {
+ if h == nil || h.closed.Load() {
+ return errors.New("modbus handle closed")
+ }
+ return nil
+}
+
+var _ drivers.ModbusCap = (*handle)(nil)
diff --git a/go/internal/modbus/engine_test.go b/go/internal/modbus/engine_test.go
new file mode 100644
index 000000000..68e62d66b
--- /dev/null
+++ b/go/internal/modbus/engine_test.go
@@ -0,0 +1,104 @@
+package modbus
+
+import (
+ "testing"
+
+ "github.com/srcfl/ftw/go/internal/drivers"
+)
+
+func TestEngineSharesOneSocketPerHostPort(t *testing.T) {
+ slave := startTestSlave(t)
+ host, port := slave.Addr()
+ engine := NewEngine()
+
+ a, err := engine.Open(host, port, 1, false)
+ if err != nil {
+ t.Fatalf("open a: %v", err)
+ }
+ defer a.Close()
+ b, err := engine.Open(host, port, 2, false)
+ if err != nil {
+ t.Fatalf("open b: %v", err)
+ }
+ defer b.Close()
+
+ if got := engine.sessionCount(); got != 1 {
+ t.Fatalf("sessions = %d, want 1", got)
+ }
+ if got := engine.refCount(sessionKey(host, port)); got != 2 {
+ t.Fatalf("refs = %d, want 2", got)
+ }
+
+ regs, err := a.Read(1, 1, drivers.ModbusHolding)
+ if err != nil {
+ t.Fatalf("read a: %v", err)
+ }
+ if len(regs) != 1 || regs[0] != 0x1111 {
+ t.Fatalf("read a = %v, want [0x1111]", regs)
+ }
+ regs, err = b.Read(1, 1, drivers.ModbusHolding)
+ if err != nil {
+ t.Fatalf("read b: %v", err)
+ }
+ if len(regs) != 1 || regs[0] != 0x1111 {
+ t.Fatalf("read b = %v, want [0x1111]", regs)
+ }
+
+ if got := slave.Accepts(); got != 1 {
+ t.Fatalf("backend accepts = %d, want 1 shared socket", got)
+ }
+ units := slave.Units()
+ if len(units) < 2 || units[0] != 1 || units[1] != 2 {
+ t.Fatalf("unit IDs = %v, want [1 2 …]", units)
+ }
+
+ if err := a.Close(); err != nil {
+ t.Fatalf("close a: %v", err)
+ }
+ if engine.sessionCount() != 1 {
+ t.Fatal("closing one handle dropped the shared session")
+ }
+ if _, err := b.Read(1, 1, drivers.ModbusHolding); err != nil {
+ t.Fatalf("read after peer close: %v", err)
+ }
+ if err := b.Close(); err != nil {
+ t.Fatalf("close b: %v", err)
+ }
+ if engine.sessionCount() != 0 {
+ t.Fatalf("sessions after last close = %d, want 0", engine.sessionCount())
+ }
+}
+
+func TestDialDoesNotJoinTheEnginePool(t *testing.T) {
+ slave := startTestSlave(t)
+ host, port := slave.Addr()
+ engine := NewEngine()
+
+ pooled, err := engine.Open(host, port, 1, false)
+ if err != nil {
+ t.Fatalf("open: %v", err)
+ }
+ defer pooled.Close()
+ private, err := Dial(host, port, 1)
+ if err != nil {
+ t.Fatalf("dial: %v", err)
+ }
+ defer private.Close()
+
+ if _, err := pooled.Read(1, 1, drivers.ModbusHolding); err != nil {
+ t.Fatalf("pooled read: %v", err)
+ }
+ if _, err := private.Read(1, 1, drivers.ModbusHolding); err != nil {
+ t.Fatalf("private read: %v", err)
+ }
+ if got := slave.Accepts(); got != 2 {
+ t.Fatalf("accepts = %d, want 2 (engine + Dial)", got)
+ }
+}
+
+func TestEngineOpenRejectsBadEndpoint(t *testing.T) {
+ engine := NewEngine()
+ if _, err := engine.Open("", 502, 1, false); err == nil {
+ t.Fatal("expected error for empty host")
+ }
+}
diff --git a/go/internal/modbus/proxy.go b/go/internal/modbus/proxy.go
new file mode 100644
index 000000000..f7eca8aa6
--- /dev/null
+++ b/go/internal/modbus/proxy.go
@@ -0,0 +1,265 @@
+package modbus
+
+import (
+ "encoding/binary"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/srcfl/ftw/go/internal/drivers"
+)
+
+const (
+ proxyIdleTimeout = 90 * time.Second
+ proxyMaxPDUSize = 253
+ proxyMaxClients = 16
+ proxyMaxADULength = 1 + proxyMaxPDUSize // unit ID + PDU
+ modbusExcIllegalFn = 0x01
+ modbusExcGWPath = 0x0A
+ modbusExcGWTarget = 0x0B
+)
+
+// Bind is one proxy listener attached to a backend already in the Engine.
+type Bind struct {
+ Listen string
+ Host string
+ Port int
+ AllowUnverifiedLocal bool
+}
+
+// Proxy accepts Modbus TCP clients and multiplexes their PDUs onto the
+// Engine session for that backend. Writes are denied unless allowWrite is
+// set: a LAN client writing registers would bypass FTW's control loop.
+type Proxy struct {
+ engine *Engine
+ allowWrite bool
+
+ mu sync.Mutex
+ listeners []net.Listener
+ pins []drivers.ModbusCap
+ wg sync.WaitGroup
+ closing atomic.Bool
+ sem chan struct{}
+}
+
+// Listen pins each backend session and serves Modbus TCP on Bind.Listen.
+func (e *Engine) Listen(binds []Bind, allowWrite bool) (*Proxy, error) {
+ if e == nil {
+ return nil, errors.New("modbus engine is nil")
+ }
+ p := &Proxy{
+ engine: e,
+ allowWrite: allowWrite,
+ sem: make(chan struct{}, proxyMaxClients),
+ }
+ for _, b := range binds {
+ if err := p.addBind(b); err != nil {
+ _ = p.Close()
+ return nil, err
+ }
+ }
+ return p, nil
+}
+
+func (p *Proxy) addBind(b Bind) error {
+ if b.Listen == "" || b.Host == "" || b.Port < 1 {
+ return fmt.Errorf("modbus proxy bind incomplete: listen=%q host=%q port=%d", b.Listen, b.Host, b.Port)
+ }
+ pin, err := p.engine.Open(b.Host, b.Port, 1, b.AllowUnverifiedLocal)
+ if err != nil {
+ return fmt.Errorf("modbus proxy pin %s:%d: %w", b.Host, b.Port, err)
+ }
+ ln, err := net.Listen("tcp", b.Listen)
+ if err != nil {
+ _ = pin.Close()
+ return fmt.Errorf("modbus proxy listen %s: %w", b.Listen, err)
+ }
+ p.mu.Lock()
+ p.pins = append(p.pins, pin)
+ p.listeners = append(p.listeners, ln)
+ p.mu.Unlock()
+
+ backend := Bind{Host: b.Host, Port: b.Port, Listen: ln.Addr().String()}
+ p.wg.Add(1)
+ go p.serve(ln, backend)
+ slog.Info("modbus proxy listening",
+ "listen", ln.Addr().String(),
+ "backend", sessionKey(b.Host, b.Port),
+ "allow_write", p.allowWrite)
+ return nil
+}
+
+func (p *Proxy) serve(ln net.Listener, backend Bind) {
+ defer p.wg.Done()
+ for {
+ conn, err := ln.Accept()
+ if err != nil {
+ if p.closing.Load() {
+ return
+ }
+ slog.Warn("modbus proxy accept", "listen", ln.Addr().String(), "err", err)
+ return
+ }
+ select {
+ case p.sem <- struct{}{}:
+ default:
+ slog.Warn("modbus proxy client limit reached", "listen", ln.Addr().String())
+ _ = conn.Close()
+ continue
+ }
+ p.wg.Add(1)
+ go func(c net.Conn) {
+ defer p.wg.Done()
+ defer func() { <-p.sem }()
+ p.handleClient(c, backend)
+ }(conn)
+ }
+}
+
+func (p *Proxy) handleClient(conn net.Conn, backend Bind) {
+ defer conn.Close()
+ for {
+ if p.closing.Load() {
+ return
+ }
+ _ = conn.SetReadDeadline(time.Now().Add(proxyIdleTimeout))
+ txID, unitID, pdu, err := readMBAP(conn)
+ if err != nil {
+ return
+ }
+ resp := p.forward(backend, unitID, pdu)
+ _ = conn.SetWriteDeadline(time.Now().Add(modbusRequestTimeout))
+ if err := writeMBAP(conn, txID, unitID, resp); err != nil {
+ return
+ }
+ }
+}
+
+func (p *Proxy) forward(backend Bind, unitID uint8, pdu []byte) []byte {
+ fc := pdu[0]
+ if isModbusWrite(fc) && !p.allowWrite {
+ return exceptionPDU(fc, modbusExcIllegalFn)
+ }
+ if !isModbusRead(fc) && !isModbusWrite(fc) {
+ return exceptionPDU(fc, modbusExcIllegalFn)
+ }
+ cap := p.engine.lookup(sessionKey(backend.Host, backend.Port))
+ if cap == nil {
+ return exceptionPDU(fc, modbusExcGWPath)
+ }
+ res, err := cap.executePDU(unitID, pdu)
+ if err != nil {
+ if errors.Is(err, drivers.ErrModbusBackoff) || isTransportError(err) {
+ return exceptionPDU(fc, modbusExcGWTarget)
+ }
+ slog.Warn("modbus proxy backend", "backend", sessionKey(backend.Host, backend.Port), "err", err)
+ return exceptionPDU(fc, modbusExcGWTarget)
+ }
+ if len(res) == 0 {
+ return exceptionPDU(fc, modbusExcGWTarget)
+ }
+ return res
+}
+
+// ListenAddrs returns the bound address of each listener. Used in tests.
+func (p *Proxy) ListenAddrs() []string {
+ if p == nil {
+ return nil
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ out := make([]string, len(p.listeners))
+ for i, ln := range p.listeners {
+ out[i] = ln.Addr().String()
+ }
+ return out
+}
+
+// Close stops listeners and drops the pin refs. Driver handles keep their
+// sessions.
+func (p *Proxy) Close() error {
+ if p == nil {
+ return nil
+ }
+ p.closing.Store(true)
+ p.mu.Lock()
+ listeners := p.listeners
+ p.listeners = nil
+ pins := p.pins
+ p.pins = nil
+ p.mu.Unlock()
+ var first error
+ for _, ln := range listeners {
+ if err := ln.Close(); err != nil && first == nil {
+ first = err
+ }
+ }
+ p.wg.Wait()
+ for _, pin := range pins {
+ if err := pin.Close(); err != nil && first == nil {
+ first = err
+ }
+ }
+ return first
+}
+
+func isModbusRead(fc byte) bool {
+ switch fc {
+ case 0x01, 0x02, 0x03, 0x04:
+ return true
+ default:
+ return false
+ }
+}
+
+func isModbusWrite(fc byte) bool {
+ switch fc {
+ case 0x05, 0x06, 0x0F, 0x10, 0x15, 0x16, 0x17:
+ return true
+ default:
+ return false
+ }
+}
+
+func exceptionPDU(fc, code byte) []byte {
+ return []byte{fc | 0x80, code}
+}
+
+func readMBAP(r io.Reader) (txID uint16, unitID uint8, pdu []byte, err error) {
+ hdr := make([]byte, 7)
+ if _, err = io.ReadFull(r, hdr); err != nil {
+ return 0, 0, nil, err
+ }
+ txID = binary.BigEndian.Uint16(hdr[0:2])
+ if proto := binary.BigEndian.Uint16(hdr[2:4]); proto != 0 {
+ return 0, 0, nil, fmt.Errorf("modbus proxy protocol id %d", proto)
+ }
+ length := int(binary.BigEndian.Uint16(hdr[4:6]))
+ if length < 2 || length > proxyMaxADULength {
+ return 0, 0, nil, fmt.Errorf("modbus proxy invalid length %d", length)
+ }
+ pdu = make([]byte, length-1)
+ if _, err = io.ReadFull(r, pdu); err != nil {
+ return 0, 0, nil, err
+ }
+ return txID, hdr[6], pdu, nil
+}
+
+func writeMBAP(w io.Writer, txID uint16, unitID uint8, pdu []byte) error {
+ if len(pdu) == 0 || len(pdu) > proxyMaxPDUSize {
+ return fmt.Errorf("modbus proxy invalid response pdu length %d", len(pdu))
+ }
+ buf := make([]byte, 7+len(pdu))
+ binary.BigEndian.PutUint16(buf[0:2], txID)
+ binary.BigEndian.PutUint16(buf[2:4], 0)
+ binary.BigEndian.PutUint16(buf[4:6], uint16(len(pdu)+1))
+ buf[6] = unitID
+ copy(buf[7:], pdu)
+ _, err := w.Write(buf)
+ return err
+}
diff --git a/go/internal/modbus/proxy_test.go b/go/internal/modbus/proxy_test.go
new file mode 100644
index 000000000..a28df1aed
--- /dev/null
+++ b/go/internal/modbus/proxy_test.go
@@ -0,0 +1,146 @@
+package modbus
+
+import (
+ "io"
+ "net"
+ "strings"
+ "testing"
+
+ "github.com/srcfl/ftw/go/internal/drivers"
+)
+
+func TestProxyMultiplexesOntoSharedSession(t *testing.T) {
+ slave := startTestSlave(t)
+ host, port := slave.Addr()
+ engine := NewEngine()
+
+ driver, err := engine.Open(host, port, 1, false)
+ if err != nil {
+ t.Fatalf("driver open: %v", err)
+ }
+ defer driver.Close()
+
+ proxy, err := engine.Listen([]Bind{{
+ Listen: "127.0.0.1:0",
+ Host: host,
+ Port: port,
+ }}, false)
+ if err != nil {
+ t.Fatalf("listen: %v", err)
+ }
+ defer proxy.Close()
+ addrs := proxy.ListenAddrs()
+ if len(addrs) != 1 {
+ t.Fatalf("listen addrs = %v", addrs)
+ }
+ pHost, pPort := mustAddr(t, addrs[0])
+
+ client, err := Dial(pHost, pPort, 1)
+ if err != nil {
+ t.Fatalf("dial proxy: %v", err)
+ }
+ defer client.Close()
+
+ regs, err := client.Read(1, 2, drivers.ModbusHolding)
+ if err != nil {
+ t.Fatalf("proxy read: %v", err)
+ }
+ if len(regs) != 2 || regs[0] != 0x1111 || regs[1] != 0x2222 {
+ t.Fatalf("proxy read = %v, want [0x1111 0x2222]", regs)
+ }
+
+ if _, err := driver.Read(1, 1, drivers.ModbusInput); err != nil {
+ t.Fatalf("driver read while proxy is up: %v", err)
+ }
+ if got := slave.Accepts(); got != 1 {
+ t.Fatalf("backend accepts = %d, want 1 (driver+proxy share)", got)
+ }
+
+ err = client.WriteSingle(10, 0xBEEF)
+ if err == nil {
+ t.Fatal("proxy write succeeded; want read-only exception")
+ }
+ if !strings.Contains(err.Error(), "code=0x01") {
+ t.Fatalf("write error = %v, want illegal-function exception", err)
+ }
+ if slave.Holding(10) != 0 {
+ t.Fatalf("backend holding[10] = %v, want unchanged 0", slave.Holding(10))
+ }
+ if slave.Writes() != 0 {
+ t.Fatalf("backend writes = %d, want 0", slave.Writes())
+ }
+}
+
+func TestProxyAllowWriteForwardsRegisterWrites(t *testing.T) {
+ slave := startTestSlave(t)
+ host, port := slave.Addr()
+ engine := NewEngine()
+ proxy, err := engine.Listen([]Bind{{
+ Listen: "127.0.0.1:0",
+ Host: host,
+ Port: port,
+ }}, true)
+ if err != nil {
+ t.Fatalf("listen: %v", err)
+ }
+ defer proxy.Close()
+ pHost, pPort := mustAddr(t, proxy.ListenAddrs()[0])
+
+ client, err := Dial(pHost, pPort, 1)
+ if err != nil {
+ t.Fatalf("dial proxy: %v", err)
+ }
+ defer client.Close()
+ if err := client.WriteSingle(10, 0xBEEF); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ if got := slave.Holding(10); got != 0xBEEF {
+ t.Fatalf("holding[10] = 0x%04x, want 0xBEEF", got)
+ }
+}
+
+func TestProxyUnknownFunctionIsIllegal(t *testing.T) {
+ slave := startTestSlave(t)
+ host, port := slave.Addr()
+ engine := NewEngine()
+ proxy, err := engine.Listen([]Bind{{
+ Listen: "127.0.0.1:0",
+ Host: host,
+ Port: port,
+ }}, false)
+ if err != nil {
+ t.Fatalf("listen: %v", err)
+ }
+ defer proxy.Close()
+
+ conn, err := net.Dial("tcp", proxy.ListenAddrs()[0])
+ if err != nil {
+ t.Fatalf("dial: %v", err)
+ }
+ defer conn.Close()
+ // FC 0x2B (Encapsulated Interface Transport) is not on the read/write list.
+ req := []byte{0, 1, 0, 0, 0, 2, 1, 0x2B}
+ if _, err := conn.Write(req); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ hdr := make([]byte, 9)
+ if _, err := io.ReadFull(conn, hdr); err != nil {
+ t.Fatalf("read: %v", err)
+ }
+ if hdr[7] != 0x2B|0x80 || hdr[8] != modbusExcIllegalFn {
+ t.Fatalf("response = %x, want exception illegal function", hdr)
+ }
+}
+
+func mustAddr(t *testing.T, addr string) (string, int) {
+ t.Helper()
+ host, portStr, err := net.SplitHostPort(addr)
+ if err != nil {
+ t.Fatalf("split %q: %v", addr, err)
+ }
+ var port int
+ if _, err := fmtSscan(portStr, &port); err != nil {
+ t.Fatalf("port %q: %v", portStr, err)
+ }
+ return host, port
+}
diff --git a/go/internal/modbus/slave_test.go b/go/internal/modbus/slave_test.go
new file mode 100644
index 000000000..adfdac53d
--- /dev/null
+++ b/go/internal/modbus/slave_test.go
@@ -0,0 +1,168 @@
+package modbus
+
+import (
+ "encoding/binary"
+ "io"
+ "net"
+ "sync"
+ "sync/atomic"
+ "testing"
+)
+
+// testSlave is a tiny Modbus TCP server with holding/input maps. It
+// records Accept count and unit IDs so sharing tests can see one socket.
+type testSlave struct {
+ ln net.Listener
+ accepts atomic.Int32
+ mu sync.Mutex
+ holding map[uint16]uint16
+ input map[uint16]uint16
+ unitIDs []uint8
+ writes int
+ closing atomic.Bool
+}
+
+func startTestSlave(t *testing.T) *testSlave {
+ t.Helper()
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatalf("listen: %v", err)
+ }
+ s := &testSlave{
+ ln: ln,
+ holding: map[uint16]uint16{1: 0x1111, 2: 0x2222, 10: 0},
+ input: map[uint16]uint16{1: 0xAABB},
+ }
+ go s.loop()
+ t.Cleanup(s.Close)
+ return s
+}
+
+func (s *testSlave) Addr() (host string, port int) {
+ host, portStr, _ := net.SplitHostPort(s.ln.Addr().String())
+ var p int
+ _, _ = fmtSscan(portStr, &p)
+ return host, p
+}
+
+func (s *testSlave) Accepts() int { return int(s.accepts.Load()) }
+
+func (s *testSlave) Units() []uint8 {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make([]uint8, len(s.unitIDs))
+ copy(out, s.unitIDs)
+ return out
+}
+
+func (s *testSlave) Holding(addr uint16) uint16 {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.holding[addr]
+}
+
+func (s *testSlave) Writes() int {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.writes
+}
+
+func (s *testSlave) Close() {
+ s.closing.Store(true)
+ _ = s.ln.Close()
+}
+
+func (s *testSlave) loop() {
+ for {
+ c, err := s.ln.Accept()
+ if err != nil {
+ return
+ }
+ s.accepts.Add(1)
+ go s.serve(c)
+ }
+}
+
+func (s *testSlave) serve(c net.Conn) {
+ defer c.Close()
+ for {
+ hdr := make([]byte, 7)
+ if _, err := io.ReadFull(c, hdr); err != nil {
+ return
+ }
+ length := int(binary.BigEndian.Uint16(hdr[4:6]))
+ if length < 2 {
+ return
+ }
+ pdu := make([]byte, length-1)
+ if _, err := io.ReadFull(c, pdu); err != nil {
+ return
+ }
+ unit := hdr[6]
+ s.mu.Lock()
+ s.unitIDs = append(s.unitIDs, unit)
+ respPDU := s.handlePDU(pdu)
+ s.mu.Unlock()
+
+ resp := make([]byte, 7+len(respPDU))
+ copy(resp[0:2], hdr[0:2])
+ binary.BigEndian.PutUint16(resp[4:6], uint16(len(respPDU)+1))
+ resp[6] = unit
+ copy(resp[7:], respPDU)
+ if _, err := c.Write(resp); err != nil {
+ return
+ }
+ }
+}
+
+func (s *testSlave) handlePDU(pdu []byte) []byte {
+ if len(pdu) == 0 {
+ return []byte{0x80, 0x01}
+ }
+ fc := pdu[0]
+ switch fc {
+ case modbusReadHoldingRegisters, modbusReadInputRegisters:
+ if len(pdu) < 5 {
+ return exceptionPDU(fc, 0x03)
+ }
+ addr := binary.BigEndian.Uint16(pdu[1:3])
+ count := binary.BigEndian.Uint16(pdu[3:5])
+ src := s.holding
+ if fc == modbusReadInputRegisters {
+ src = s.input
+ }
+ out := make([]byte, 2+int(count)*2)
+ out[0] = fc
+ out[1] = byte(count * 2)
+ for i := uint16(0); i < count; i++ {
+ binary.BigEndian.PutUint16(out[2+int(i)*2:4+int(i)*2], src[addr+i])
+ }
+ return out
+ case modbusWriteSingleRegister:
+ if len(pdu) < 5 {
+ return exceptionPDU(fc, 0x03)
+ }
+ addr := binary.BigEndian.Uint16(pdu[1:3])
+ value := binary.BigEndian.Uint16(pdu[3:5])
+ s.holding[addr] = value
+ s.writes++
+ return append([]byte(nil), pdu...)
+ case modbusWriteMultipleRegs:
+ if len(pdu) < 6 {
+ return exceptionPDU(fc, 0x03)
+ }
+ addr := binary.BigEndian.Uint16(pdu[1:3])
+ count := binary.BigEndian.Uint16(pdu[3:5])
+ s.writes++
+ for i := uint16(0); i < count; i++ {
+ off := 6 + int(i)*2
+ if off+2 > len(pdu) {
+ break
+ }
+ s.holding[addr+i] = binary.BigEndian.Uint16(pdu[off : off+2])
+ }
+ return []byte{fc, pdu[1], pdu[2], pdu[3], pdu[4]}
+ default:
+ return exceptionPDU(fc, 0x01)
+ }
+}
diff --git a/go/internal/modbus/tcp_client.go b/go/internal/modbus/tcp_client.go
index a95b4f388..1414090d5 100644
--- a/go/internal/modbus/tcp_client.go
+++ b/go/internal/modbus/tcp_client.go
@@ -91,7 +91,7 @@ func (c *tcpClient) ReadRegisters(addr, count uint16, fc byte) ([]uint16, error)
byte(addr >> 8), byte(addr),
byte(count >> 8), byte(count),
}
- res, err := c.execute(pdu)
+ res, err := c.execute(c.unitID, pdu)
if err != nil {
return nil, err
}
@@ -116,7 +116,7 @@ func (c *tcpClient) WriteRegister(addr, value uint16) error {
byte(addr >> 8), byte(addr),
byte(value >> 8), byte(value),
}
- res, err := c.execute(pdu)
+ res, err := c.execute(c.unitID, pdu)
if err != nil {
return err
}
@@ -144,7 +144,7 @@ func (c *tcpClient) WriteRegisters(addr uint16, values []uint16) error {
for i, v := range values {
binary.BigEndian.PutUint16(pdu[6+i*2:8+i*2], v)
}
- res, err := c.execute(pdu)
+ res, err := c.execute(c.unitID, pdu)
if err != nil {
return err
}
@@ -156,17 +156,41 @@ func (c *tcpClient) WriteRegisters(addr uint16, values []uint16) error {
return nil
}
-func (c *tcpClient) execute(pdu []byte) ([]byte, error) {
+func (c *tcpClient) execute(unitID uint8, pdu []byte) ([]byte, error) {
+ res, err := c.roundTrip(unitID, pdu)
+ if err != nil {
+ return nil, err
+ }
+ if len(res) == 0 {
+ return nil, errors.New("modbus empty response pdu")
+ }
+ if res[0]&0x80 != 0 {
+ code := byte(0)
+ if len(res) > 1 {
+ code = res[1]
+ }
+ return nil, modbusException{function: res[0] &^ 0x80, code: code}
+ }
+ return res, nil
+}
+
+// roundTrip sends one PDU and returns the response PDU, including Modbus
+// exception PDUs. Transport failures are the only errors. unitID is taken
+// from the caller so one TCP session can serve several slaves.
+func (c *tcpClient) roundTrip(unitID uint8, pdu []byte) ([]byte, error) {
if c.conn == nil {
return nil, io.ErrClosedPipe
}
+ if len(pdu) == 0 {
+ return nil, errors.New("modbus empty request pdu")
+ }
c.txID++
txID := c.txID
req := make([]byte, 7+len(pdu))
binary.BigEndian.PutUint16(req[0:2], txID)
binary.BigEndian.PutUint16(req[2:4], 0)
binary.BigEndian.PutUint16(req[4:6], uint16(len(pdu)+1))
- req[6] = c.unitID
+ req[6] = unitID
copy(req[7:], pdu)
deadline := time.Now().Add(c.timeout)
@@ -192,16 +216,6 @@ func (c *tcpClient) execute(pdu []byte) ([]byte, error) {
if _, err := io.ReadFull(c.conn, res); err != nil {
return nil, err
}
- if len(res) == 0 {
- return nil, errors.New("modbus empty response pdu")
- }
- if res[0]&0x80 != 0 {
- code := byte(0)
- if len(res) > 1 {
- code = res[1]
- }
- return nil, modbusException{function: res[0] &^ 0x80, code: code}
- }
return res, nil
}
diff --git a/go/test/e2e/stack_test.go b/go/test/e2e/stack_test.go
index df4edff00..19ae65914 100644
--- a/go/test/e2e/stack_test.go
+++ b/go/test/e2e/stack_test.go
@@ -277,8 +277,9 @@ func setupStack(t *testing.T) *stack {
s.reg.MQTTFactory = func(name string, c *config.MQTTConfig) (drivers.MQTTCap, error) {
return mqttcli.Dial(c.Host, c.Port, c.Username, c.Password, "ftw-e2e-"+name)
}
+ modbusEngine := modbuscli.NewEngine()
s.reg.ModbusFactory = func(name string, c *config.ModbusConfig) (drivers.ModbusCap, error) {
- return modbuscli.Dial(c.Host, c.Port, c.UnitID)
+ return modbusEngine.Open(c.Host, c.Port, c.UnitID, false)
}
for _, d := range s.cfg.Drivers {
if err := s.reg.Add(ctx, d); err != nil {
diff --git a/web/index.html b/web/index.html
index 986fb9376..f0f42aaeb 100644
--- a/web/index.html
+++ b/web/index.html
@@ -941,7 +941,7 @@ Price bars (top of the chart)
-
+
diff --git a/web/settings/tabs/ha.js b/web/settings/tabs/ha.js
index 9a466df90..38112c1a6 100644
--- a/web/settings/tabs/ha.js
+++ b/web/settings/tabs/ha.js
@@ -39,6 +39,7 @@
render: function (ctx) {
var field = ctx.field, config = ctx.config;
if (!config.homeassistant) config.homeassistant = {};
+ if (!config.modbus_proxy) config.modbus_proxy = {};
return 'checking…
' +
'' +
+ '';
},
after: function () {
diff --git a/web/settings/tabs/ha.test.mjs b/web/settings/tabs/ha.test.mjs
index 7814a6891..54024865f 100644
--- a/web/settings/tabs/ha.test.mjs
+++ b/web/settings/tabs/ha.test.mjs
@@ -5,7 +5,8 @@ import assert from "node:assert/strict";
globalThis.window = {};
await import("./ha.js");
-const { statusView } = globalThis.window.FTWSettings.tabs.ha._pure;
+const haTab = globalThis.window.FTWSettings.tabs.ha;
+const { statusView } = haTab._pure;
describe("Home Assistant status", () => {
it("shows disabled only when saved config is disabled", () => {
@@ -46,3 +47,16 @@ describe("Home Assistant status", () => {
});
});
});
+
+describe("Modbus proxy settings", () => {
+ it("renders the proxy fieldset on the Home Assistant tab", () => {
+ const html = haTab.render({
+ field: function (label, path) { return "" + label + ""; },
+ config: {},
+ });
+ assert.match(html, /Modbus TCP proxy/);
+ assert.match(html, /modbus_proxy.enabled/);
+ assert.match(html, /data-path="modbus_proxy.listen"/);
+ assert.match(html, /modbus_proxy.allow_write/);
+ });
+});