OneHouse is one program that controls a whole house. Linux is supported.
The command runs an empty driver list and stops on SIGINT or SIGTERM.
``doc/driver.rst`` describes the framework and ``doc/mqtt.rst`` defines
-its wire contract.
+its wire contract. ``doc/m31.rst`` describes the M31 integration.
*************
Development
In-memory stream tests check request bytes, response validation,
exceptions, truncation, cancellation, and timeouts. They do not verify a
physical host or TCP network behavior.
+
+*******************
+ Polling lifecycle
+*******************
+
+``New`` validates ``Settings`` and takes a shared publisher and a
+client. The caller runs both the publisher and the integration through
+``driver.Run``. The command keeps its explicit driver list empty and
+does not open hardware connections.
+
+Every connection writes and reads back all digital output modes. Pulse
+outputs also write and verify their widths before online. The first
+complete read establishes a baseline without events. A later input
+change sends retained state followed by a pressed or released event. All
+events from one input read share its completion time, before coil and
+analog reads. A steady output or analog change sends reported state.
+Pulse coils do not send transient feedback.
+
+A device error publishes offline and retries after one second, doubling
+to thirty seconds until a complete baseline succeeds. Device failures do
+not log inside the integration. Publication and cleanup failures return
+to the runner. Cancellation closes the client without a false offline
+publication. Refresh requests cached snapshots through the shared
+publisher without repeating events or physical work.
+
+Polling can miss an edge that starts and ends between reads. It does not
+expose input counters. A configured poll interval is a target, not a
+guaranteed edge latency. Tests use fake clients and simulated time for
+baseline, transitions, pulse verification, reconnect, refresh, shutdown,
+and error propagation. Physical timing remains unverified.
// Client owns one serialized Modbus connection. Connect replaces a closed
// connection and selects a unit from one through 247. Reads must return the
// exact requested count. Writes validate their acknowledgements. All operations
-// honor caller cancellation. The driver closes each connection before retry.
-// A caller must not use the client concurrently.
+// honor caller cancellation. The driver closes each connection before retry. A
+// caller must not use the client concurrently.
type Client interface {
Connect(ctx context.Context, address string, unit byte) error
Close() error
--- /dev/null
+package m31
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "strconv"
+ "time"
+
+ "onehouse/protocol"
+)
+
+const (
+ pulseWidthBase = 0x30d4
+ outputModeBase = 0x32c8
+ maximumVoltage = 10000
+)
+
+type connection struct {
+ owner *Driver
+ state snapshot
+}
+
+func (c *connection) start(ctx context.Context) error {
+ err := c.configure(ctx)
+ if err != nil {
+ return fmt.Errorf("configure outputs: %w", err)
+ }
+
+ next, err := c.read(ctx)
+ if err != nil {
+ return fmt.Errorf("read baseline: %w", err)
+ }
+
+ err = c.owner.publisher.Availability(ctx, c.owner.deviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "online"})
+ if err != nil {
+ return fmt.Errorf("publish online: %w", err)
+ }
+
+ err = c.publish(ctx, next, true, nil)
+ if err != nil {
+ return fmt.Errorf("publish baseline: %w", err)
+ }
+
+ return nil
+}
+
+func (c *connection) configure(ctx context.Context) error {
+ for _, output := range c.owner.topology.outputs {
+ mode := uint16(0)
+ if output.pulse != 0 {
+ err := c.configureRegister(ctx, pulseWidthBase+output.address, output.pulse)
+ if err != nil {
+ return fmt.Errorf("set pulse width: %w", err)
+ }
+
+ mode = 1
+ }
+
+ err := c.configureRegister(ctx, outputModeBase+output.address, mode)
+ if err != nil {
+ return fmt.Errorf("set output mode: %w", err)
+ }
+ }
+
+ return nil
+}
+
+func (c *connection) configureRegister(ctx context.Context, address uint16, value uint16) error {
+ err := c.owner.client.WriteRegister(ctx, address, value)
+ if err != nil {
+ return &deviceError{err: fmt.Errorf("write configuration register: %w", err)}
+ }
+
+ values, err := c.owner.client.ReadRegisters(ctx, address, 1)
+ if err != nil {
+ return &deviceError{err: fmt.Errorf("read configuration register: %w", err)}
+ }
+
+ if len(values) != 1 || values[0] != value {
+ return &deviceError{err: fmt.Errorf("configuration register readback differs")}
+ }
+
+ return nil
+}
+
+func (c *connection) read(ctx context.Context) (snapshot, error) {
+ inputs, err := c.owner.client.ReadInputs(ctx, uint16(len(c.owner.topology.inputs)))
+ if err != nil {
+ return snapshot{}, &deviceError{err: fmt.Errorf("read inputs: %w", err)}
+ }
+
+ observedAt := time.Now().UTC()
+ outputs, err := c.owner.client.ReadCoils(ctx, uint16(len(c.owner.topology.outputs)))
+ if err != nil {
+ return snapshot{}, &deviceError{err: fmt.Errorf("read coils: %w", err)}
+ }
+
+ var analog []uint16
+
+ if len(c.owner.topology.analog) != 0 {
+ analog, err = c.owner.client.ReadRegisters(ctx, 0, uint16(len(c.owner.topology.analog)))
+ if err != nil {
+ return snapshot{}, &deviceError{err: fmt.Errorf("read analog outputs: %w", err)}
+ }
+ }
+
+ if len(inputs) != len(c.owner.topology.inputs) || len(outputs) != len(c.owner.topology.outputs) || len(analog) != len(c.owner.topology.analog) {
+ return snapshot{}, &deviceError{err: fmt.Errorf("incomplete assembly read")}
+ }
+
+ err = validateVoltage(analog)
+ if err != nil {
+ return snapshot{}, &deviceError{err: fmt.Errorf("validate analog outputs: %w", err)}
+ }
+
+ return snapshot{
+ inputs: inputs,
+ outputs: outputs,
+ analog: analog,
+ observedAt: observedAt,
+ }, nil
+}
+
+func (c *connection) publish(ctx context.Context, next snapshot, baseline bool, causes map[string]string) error {
+ for index, point := range c.owner.topology.inputs {
+ if !baseline && next.inputs[index] == c.state.inputs[index] {
+ continue
+ }
+
+ value := strconv.FormatBool(next.inputs[index])
+ err := c.report(ctx, point.id, value, next.observedAt, "")
+ if err != nil {
+ return fmt.Errorf("publish input state: %w", err)
+ }
+
+ if !baseline {
+ action := `"released"`
+ if next.inputs[index] {
+ action = `"pressed"`
+ }
+
+ err = c.owner.publisher.Event(ctx, c.owner.deviceID, point.id, protocol.Metadata{SentAt: &next.observedAt}, protocol.EventPayload{
+ "action": json.RawMessage(action),
+ "value": json.RawMessage(value),
+ })
+ if err != nil {
+ return fmt.Errorf("publish input transition: %w", err)
+ }
+ }
+ }
+
+ for index, point := range c.owner.topology.outputs {
+ if point.pulse != 0 || !baseline && next.outputs[index] == c.state.outputs[index] {
+ continue
+ }
+
+ err := c.report(ctx, point.id, strconv.FormatBool(next.outputs[index]), next.observedAt, causes[point.id])
+ if err != nil {
+ return fmt.Errorf("publish output state: %w", err)
+ }
+ }
+
+ for index, point := range c.owner.topology.analog {
+ if !baseline && next.analog[index] == c.state.analog[index] {
+ continue
+ }
+
+ value := strconv.FormatFloat(float64(next.analog[index])/1000, 'f', -1, 64)
+ err := c.report(ctx, point.id, value, next.observedAt, causes[point.id])
+ if err != nil {
+ return fmt.Errorf("publish analog state: %w", err)
+ }
+ }
+
+ c.state = next
+
+ return nil
+}
+
+func (c *connection) report(ctx context.Context, capability string, value string, observedAt time.Time, cause string) error {
+ err := c.owner.publisher.Reported(ctx, c.owner.deviceID, capability, protocol.Metadata{
+ SentAt: &observedAt,
+ CausationID: cause,
+ }, protocol.ValuePayload{Value: json.RawMessage(value)})
+ if err != nil {
+ return fmt.Errorf("publish point: %w", err)
+ }
+
+ return nil
+}
+
+func (c *connection) poll(ctx context.Context) error {
+ next, err := c.read(ctx)
+ if err != nil {
+ return fmt.Errorf("read assembly state: %w", err)
+ }
+
+ err = c.publish(ctx, next, false, nil)
+ if err != nil {
+ return fmt.Errorf("publish assembly state: %w", err)
+ }
+
+ return nil
+}
+
+func validateVoltage(values []uint16) error {
+ for _, value := range values {
+ if value > maximumVoltage {
+ return fmt.Errorf("analog output exceeds ten volts")
+ }
+ }
+
+ return nil
+}
--- /dev/null
+package m31
+
+import (
+ "encoding/json"
+ "errors"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/protocol"
+)
+
+func TestConnectionBaselineAndInputEvents(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ c, client, transport := testConnection(t)
+ require.Empty(t, transport.ofKind(protocol.Event))
+
+ client.inputs[0], client.inputs[15] = true, true
+ client.inputDelay = time.Millisecond
+ observedAt := time.Now().Add(time.Millisecond).UTC()
+
+ require.NoError(t, c.poll(t.Context()))
+
+ events := transport.ofKind(protocol.Event)
+ require.Len(t, events, 2)
+
+ for _, event := range events {
+ require.Equal(t, &observedAt, event.Metadata.SentAt)
+ require.JSONEq(t, `{"action":"pressed","value":true}`, eventJSON(t, event.Payload))
+ }
+
+ require.NoError(t, c.poll(t.Context()))
+ require.Len(t, transport.ofKind(protocol.Event), 2)
+
+ client.inputs[0] = false
+
+ require.NoError(t, c.poll(t.Context()))
+
+ events = transport.ofKind(protocol.Event)
+ require.Len(t, events, 3)
+ require.JSONEq(t, `{"action":"released","value":false}`, eventJSON(t, events[2].Payload))
+
+ client.coils[2] = true
+ transport.messages = nil
+
+ require.NoError(t, c.poll(t.Context()))
+ require.Empty(t, transport.messages)
+ })
+}
+
+func TestConnectionRejectsConfigurationAndReadFailuresBeforeOnline(t *testing.T) {
+ testCases := []struct {
+ name string
+ change func(*fakeClient)
+ }{
+ {name: "rejects pulse register write failure", change: func(c *fakeClient) {
+ c.registerErr = errors.New("configuration failed")
+ }},
+ {name: "rejects configuration mismatch", change: func(c *fakeClient) {
+ c.wrongConfig = true
+ }},
+ {name: "rejects input read failure", change: func(c *fakeClient) {
+ c.inputErr = errors.New("input read failed")
+ }},
+ {name: "rejects coil read failure", change: func(c *fakeClient) {
+ c.readErr = errors.New("coil read failed")
+ }},
+ {name: "rejects incomplete read", change: func(c *fakeClient) {
+ c.shortRead = true
+ }},
+ {name: "rejects unsafe voltage", change: func(c *fakeClient) {
+ c.analog[0] = 10001
+ }},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ d, client, transport := testAssembly(t)
+ tc.change(client)
+
+ c := &connection{owner: d}
+ err := c.start(t.Context())
+
+ var failure *deviceError
+
+ require.ErrorAs(t, err, &failure)
+ require.Empty(t, transport.messages)
+ })
+ }
+}
+
+func TestConnectionReturnsPublicationFailures(t *testing.T) {
+ testCases := []struct {
+ name string
+ kind protocol.Kind
+ change func(*fakeClient)
+ }{
+ {name: "returns input report failure", kind: protocol.Reported, change: func(c *fakeClient) {
+ c.inputs[0] = true
+ }},
+ {name: "returns transition failure", kind: protocol.Event, change: func(c *fakeClient) {
+ c.inputs[0] = true
+ }},
+ {name: "returns output report failure", kind: protocol.Reported, change: func(c *fakeClient) {
+ c.coils[0] = true
+ }},
+ {name: "returns voltage report failure", kind: protocol.Reported, change: func(c *fakeClient) {
+ c.analog[0] = 5000
+ }},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ c, client, transport := testConnection(t)
+ failure := errors.New("publish failed")
+ transport.failKind, transport.err = tc.kind, failure
+ tc.change(client)
+ require.ErrorIs(t, c.poll(t.Context()), failure)
+ })
+ }
+}
+
+func eventJSON(t *testing.T, value any) string {
+ t.Helper()
+
+ data, err := json.Marshal(value)
+ require.NoError(t, err)
+
+ return string(data)
+}
--- /dev/null
+package m31
+
+// deviceError selects reconnect recovery for a physical transport or feedback
+// failure. Publication and cleanup failures return to the framework runner.
+type deviceError struct {
+ err error
+}
+
+func (f *deviceError) Error() string {
+ return f.err.Error()
+}
+
+func (f *deviceError) Unwrap() error {
+ return f.err
+}
--- /dev/null
+package m31
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "strconv"
+ "time"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+// Driver polls one assembly. Run has one owner and closes the supplied Client
+// on each attempt. Recoverable device failures publish offline and retry with
+// backoff. Publication or cleanup failures return to the framework runner. The
+// shared publisher must run alongside this driver. Refresh is safe during Run.
+// No operation starts a physical connection before Run.
+type Driver struct {
+ deviceID string
+ address string
+ unit byte
+ pollInterval time.Duration
+ topology topology
+ publisher *driver.Publisher
+ client Client
+}
+
+// New validates settings. The caller supplies valid publisher and client
+// dependencies. The driver owns the client until Run returns.
+func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver, error) {
+ if !protocol.ValidID(settings.DeviceID) {
+ return nil, fmt.Errorf("device ID must be a canonical ULID")
+ }
+
+ host, port, err := net.SplitHostPort(settings.Address)
+ if err != nil || host == "" {
+ return nil, fmt.Errorf("address must contain a host and TCP port")
+ }
+
+ number, err := strconv.Atoi(port)
+ if err != nil || number < 1 || number > 65535 {
+ return nil, fmt.Errorf("TCP port is outside its range")
+ }
+
+ if settings.Unit == 0 || settings.Unit > 247 {
+ return nil, fmt.Errorf("unit must be from one through 247")
+ }
+
+ if settings.PollInterval < 10*time.Millisecond || settings.PollInterval > time.Minute {
+ return nil, fmt.Errorf("poll interval must be from ten milliseconds through one minute")
+ }
+
+ points, err := newTopology(settings.Modules)
+ if err != nil {
+ return nil, fmt.Errorf("validate topology: %w", err)
+ }
+
+ return &Driver{
+ deviceID: settings.DeviceID,
+ address: settings.Address,
+ unit: settings.Unit,
+ pollInterval: settings.PollInterval,
+ topology: points,
+ publisher: publisher,
+ client: client,
+ }, nil
+}
+
+// Run configures and verifies output modes before online or polling. A
+// successful first read establishes a baseline without input events. Device
+// failures retry after one second, doubling to thirty seconds until a complete
+// baseline succeeds.
+func (d *Driver) Run(ctx context.Context) error {
+ err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
+ if err != nil {
+ return fmt.Errorf("publish descriptor: %w", err)
+ }
+
+ delay := time.Second
+
+ for {
+ healthy, retry, err := d.attempt(ctx)
+ if !retry || ctx.Err() != nil {
+ return fmt.Errorf("run assembly: %w", err)
+ }
+
+ err = d.publisher.Availability(ctx, d.deviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"})
+ if err != nil {
+ return fmt.Errorf("publish offline: %w", err)
+ }
+
+ if healthy {
+ delay = time.Second
+ }
+
+ err = d.wait(ctx, delay)
+ if err != nil {
+ return fmt.Errorf("wait for reconnect: %w", err)
+ }
+
+ delay = min(delay*2, 30*time.Second)
+ }
+}
+
+// Refresh requests retained snapshots through the shared publisher. It does not
+// repeat input events, physical commands, or pulse operations.
+func (d *Driver) Refresh() {
+ d.publisher.Refresh()
+}
+
+func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err error) {
+ defer func() {
+ var failure *deviceError
+
+ retry = errors.As(err, &failure)
+
+ closeErr := d.client.Close()
+ if closeErr != nil {
+ retry = false
+ err = fmt.Errorf("close assembly: %w", errors.Join(err, closeErr))
+ }
+ }()
+
+ err = d.client.Connect(ctx, d.address, d.unit)
+ if err != nil {
+ return false, false, &deviceError{err: fmt.Errorf("connect assembly: %w", err)}
+ }
+
+ connection := &connection{owner: d}
+ err = connection.start(ctx)
+ if err != nil {
+ return false, false, fmt.Errorf("initialize assembly: %w", err)
+ }
+
+ healthy = true
+ ticker := time.NewTicker(d.pollInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return healthy, false, ctx.Err()
+ case <-ticker.C:
+ err = connection.poll(ctx)
+ if err != nil {
+ return healthy, false, fmt.Errorf("poll assembly: %w", err)
+ }
+ }
+ }
+}
+
+func (d *Driver) wait(ctx context.Context, delay time.Duration) error {
+ timer := time.NewTimer(delay)
+ defer timer.Stop()
+
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-timer.C:
+ return nil
+ }
+}
--- /dev/null
+package m31
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/protocol"
+)
+
+func TestNewRejectsInvalidAssemblySettings(t *testing.T) {
+ testCases := []struct {
+ name string
+ change func(*Settings)
+ }{
+ {name: "rejects noncanonical ID", change: func(s *Settings) {
+ s.DeviceID = "bad"
+ }},
+ {name: "requires host and port", change: func(s *Settings) {
+ s.Address = "example.com"
+ }},
+ {name: "requires a host", change: func(s *Settings) {
+ s.Address = ":502"
+ }},
+ {name: "requires numeric port", change: func(s *Settings) {
+ s.Address = "example.com:bad"
+ }},
+ {name: "rejects port zero", change: func(s *Settings) {
+ s.Address = "example.com:0"
+ }},
+ {name: "rejects excess port", change: func(s *Settings) {
+ s.Address = "example.com:65536"
+ }},
+ {name: "rejects broadcast unit", change: func(s *Settings) {
+ s.Unit = 0
+ }},
+ {name: "rejects reserved unit", change: func(s *Settings) {
+ s.Unit = 248
+ }},
+ {name: "rejects fast polling", change: func(s *Settings) {
+ s.PollInterval = time.Millisecond
+ }},
+ {name: "rejects slow polling", change: func(s *Settings) {
+ s.PollInterval = time.Minute + time.Nanosecond
+ }},
+ {name: "requires valid modules", change: func(s *Settings) {
+ s.Modules = nil
+ }},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ d, client, _ := testAssembly(t)
+ settings := Settings{
+ DeviceID: testDeviceID,
+ Address: "example.com:502",
+ Unit: 1,
+ PollInterval: 50 * time.Millisecond,
+ Modules: []Module{{Model: HostModel}},
+ }
+ tc.change(&settings)
+
+ _, err := New(settings, d.publisher, client)
+ require.Error(t, err)
+ })
+ }
+}
+
+func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ d, client, transport := testAssembly(t)
+ client.connectErrors = []error{errors.New("unavailable")}
+ client.inputs[1] = true
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ done := make(chan error, 1)
+
+ go func() {
+ done <- d.Run(ctx)
+ }()
+
+ synctest.Wait()
+ client.withLock(func(client *fakeClient) {
+ require.Len(t, client.attempts, 1)
+ require.Empty(t, client.registerWrites)
+ })
+ require.Empty(t, transport.ofKind(protocol.Event))
+
+ time.Sleep(time.Second)
+ synctest.Wait()
+ client.withLock(func(client *fakeClient) {
+ require.Len(t, client.attempts, 2)
+ require.Len(t, client.registerWrites, 33)
+ require.EqualValues(t, 200, client.registers[pulseWidthBase+2])
+ require.EqualValues(t, 1, client.registers[outputModeBase+2])
+ })
+ require.Empty(t, transport.ofKind(protocol.Event))
+
+ client.withLock(func(client *fakeClient) {
+ client.inputs[1] = false
+ })
+
+ time.Sleep(50 * time.Millisecond)
+ synctest.Wait()
+ require.Len(t, transport.ofKind(protocol.Event), 1)
+
+ client.withLock(func(client *fakeClient) {
+ client.inputErr = errors.New("connection lost")
+ })
+
+ time.Sleep(50 * time.Millisecond)
+ synctest.Wait()
+
+ availability, valid := transport.ofKind(protocol.Availability)[2].Payload.(protocol.AvailabilityPayload)
+ require.True(t, valid)
+ require.Equal(t, "offline", availability.Status)
+
+ client.withLock(func(client *fakeClient) {
+ client.inputErr = nil
+ client.inputs[1] = true
+ })
+
+ time.Sleep(time.Second)
+ synctest.Wait()
+ client.withLock(func(client *fakeClient) {
+ require.Len(t, client.attempts, 3)
+ require.Len(t, client.registerWrites, 66)
+ })
+ require.Len(t, transport.ofKind(protocol.Event), 1)
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ client.withLock(func(client *fakeClient) {
+ require.Equal(t, 3, client.closes)
+ })
+ })
+}
+
+func TestDriverBoundsReconnectBackoff(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ d, client, _ := testAssembly(t)
+ for range 8 {
+ client.withLock(func(client *fakeClient) {
+ client.connectErrors = append(client.connectErrors, errors.New("offline"))
+ })
+ }
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ done := make(chan error, 1)
+
+ go func() {
+ done <- d.Run(ctx)
+ }()
+
+ synctest.Wait()
+
+ for _, delay := range []time.Duration{1, 2, 4, 8, 16, 30, 30} {
+ time.Sleep(delay * time.Second)
+ synctest.Wait()
+ }
+
+ client.withLock(func(client *fakeClient) {
+ require.Len(t, client.attempts, 8)
+ require.Equal(t, 30*time.Second, client.attempts[7].Sub(client.attempts[6]))
+ })
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ })
+}
+
+func TestDriverReturnsPublicationAndCleanupFailures(t *testing.T) {
+ testCases := []struct {
+ name string
+ kind protocol.Kind
+ closeFailure bool
+ }{
+ {name: "returns descriptor failure", kind: protocol.Descriptor},
+ {name: "returns online failure", kind: protocol.Availability},
+ {name: "returns baseline failure", kind: protocol.Reported},
+ {name: "returns cleanup failure", closeFailure: true},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ d, client, transport := testAssembly(t)
+ failure := errors.New("independent failure")
+ if tc.closeFailure {
+ client.closeErr = failure
+ client.connectErrors = []error{errors.New("device unavailable")}
+ } else {
+ transport.failKind, transport.err = tc.kind, failure
+ }
+
+ require.ErrorIs(t, d.Run(t.Context()), failure)
+ require.LessOrEqual(t, len(client.attempts), 1)
+ })
+ }
+}
+
+func TestDriverRefreshDoesNotRepeatCommandsOrEvents(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ d, client, transport := testAssembly(t)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ done := make(chan error, 2)
+
+ go func() {
+ done <- d.Run(ctx)
+ }()
+ go func() {
+ done <- d.publisher.Run(ctx)
+ }()
+
+ synctest.Wait()
+
+ count := len(transport.all())
+
+ d.Refresh()
+ synctest.Wait()
+ require.Greater(t, len(transport.all()), count)
+ require.Empty(t, client.writes)
+ require.Empty(t, transport.ofKind(protocol.Event))
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ require.ErrorIs(t, <-done, context.Canceled)
+ })
+}
--- /dev/null
+package m31
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "slices"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+const testDeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+type fakeClient struct {
+ mu sync.Mutex
+ inputs []bool
+ coils []bool
+ analog []uint16
+ registers map[uint16]uint16
+ writes []coilWrite
+ registerWrites []uint16
+ attempts []time.Time
+ closes int
+ coilReads int
+ connectErrors []error
+ inputErr error
+ readErr error
+ writeErr error
+ registerErr error
+ closeErr error
+ wrongConfig bool
+ shortRead bool
+ mismatch int
+ writeDelay time.Duration
+ inputDelay time.Duration
+}
+
+func (c *fakeClient) Connect(ctx context.Context, _ string, _ byte) error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ c.attempts = append(c.attempts, time.Now())
+ if len(c.connectErrors) != 0 {
+ err := c.connectErrors[0]
+ c.connectErrors = c.connectErrors[1:]
+
+ return err
+ }
+
+ return ctx.Err()
+}
+
+func (c *fakeClient) Close() error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ c.closes++
+
+ return c.closeErr
+}
+
+func (c *fakeClient) ReadInputs(ctx context.Context, _ uint16) ([]bool, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ time.Sleep(c.inputDelay)
+
+ if c.inputErr != nil || ctx.Err() != nil {
+ return nil, errors.Join(c.inputErr, ctx.Err())
+ }
+
+ return slices.Clone(c.inputs), nil
+}
+
+func (c *fakeClient) ReadCoils(ctx context.Context, _ uint16) ([]bool, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ c.coilReads++
+ if c.readErr != nil || ctx.Err() != nil {
+ return nil, errors.Join(c.readErr, ctx.Err())
+ }
+
+ if c.shortRead {
+ return nil, nil
+ }
+
+ return slices.Clone(c.coils), nil
+}
+
+func (c *fakeClient) ReadRegisters(ctx context.Context, address uint16, count uint16) ([]uint16, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ if c.registerErr != nil || ctx.Err() != nil {
+ return nil, errors.Join(c.registerErr, ctx.Err())
+ }
+
+ if address >= pulseWidthBase {
+ value := c.registers[address]
+ if c.wrongConfig {
+ value++
+ }
+
+ return []uint16{value}, nil
+ }
+
+ return slices.Clone(c.analog[:min(int(count), len(c.analog))]), nil
+}
+
+func (c *fakeClient) WriteCoil(ctx context.Context, address uint16, value bool) error {
+ return c.write(ctx, address, []bool{value}, false)
+}
+
+func (c *fakeClient) WriteCoils(ctx context.Context, address uint16, values []bool) error {
+ return c.write(ctx, address, values, true)
+}
+
+func (c *fakeClient) write(ctx context.Context, address uint16, values []bool, multiple bool) error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ time.Sleep(c.writeDelay)
+
+ if c.writeErr != nil || ctx.Err() != nil {
+ return errors.Join(c.writeErr, ctx.Err())
+ }
+
+ c.writes = append(c.writes, coilWrite{
+ address: address,
+ values: slices.Clone(values),
+ multiple: multiple,
+ })
+
+ for offset, value := range values {
+ if int(address)+offset != c.mismatch {
+ c.coils[int(address)+offset] = value
+ }
+ }
+
+ return nil
+}
+
+func (c *fakeClient) WriteRegister(ctx context.Context, address uint16, value uint16) error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ if c.registerErr != nil || ctx.Err() != nil {
+ return errors.Join(c.registerErr, ctx.Err())
+ }
+
+ c.registerWrites = append(c.registerWrites, address)
+ if address >= pulseWidthBase {
+ c.registers[address] = value
+ } else if int(address) != c.mismatch {
+ c.analog[address] = value
+ }
+
+ return nil
+}
+
+func (c *fakeClient) withLock(fn func(*fakeClient)) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ fn(c)
+}
+
+type coilWrite struct {
+ address uint16
+ values []bool
+ multiple bool
+}
+
+type fakeTransport struct {
+ mu sync.Mutex
+ messages []protocol.Message[any]
+ failKind protocol.Kind
+ err error
+}
+
+func (f *fakeTransport) Publish(ctx context.Context, publication protocol.Publication) error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
+ message, err := protocol.Parse(publication.Topic, publication.Data, publication.Retain)
+ if err != nil {
+ return fmt.Errorf("parse fake publication: %w", err)
+ }
+
+ if message.Address.Kind == f.failKind && f.err != nil {
+ return f.err
+ }
+
+ if ctx.Err() != nil {
+ return ctx.Err()
+ }
+
+ f.messages = append(f.messages, message)
+
+ return nil
+}
+
+func (f *fakeTransport) ofKind(kind protocol.Kind) []protocol.Message[any] {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
+ var messages []protocol.Message[any]
+
+ for _, message := range f.messages {
+ if message.Address.Kind == kind {
+ messages = append(messages, message)
+ }
+ }
+
+ return messages
+}
+
+func (f *fakeTransport) all() []protocol.Message[any] {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
+ return slices.Clone(f.messages)
+}
+
+func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport) {
+ t.Helper()
+
+ transport := &fakeTransport{}
+ publisher := driver.NewPublisher(transport)
+ client := &fakeClient{
+ inputs: make([]bool, 16),
+ coils: make([]bool, 32),
+ analog: make([]uint16, 8),
+ registers: make(map[uint16]uint16),
+ mismatch: -1,
+ }
+ settings := Settings{
+ DeviceID: testDeviceID,
+ Address: "example.com:502",
+ Unit: 1,
+ PollInterval: 50 * time.Millisecond,
+ Modules: []Module{
+ {Model: HostModel, PulseWidths: map[int]time.Duration{2: 200 * time.Millisecond}},
+ {Model: OutputModel},
+ {Model: AnalogModel},
+ {Model: MixedModel},
+ },
+ }
+ d, err := New(settings, publisher, client)
+ require.NoError(t, err)
+
+ return d, client, transport
+}
+
+func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
+ t.Helper()
+
+ d, client, transport := testAssembly(t)
+ connection := &connection{owner: d}
+ require.NoError(t, connection.start(t.Context()))
+
+ transport.messages = nil
+ client.writes = nil
+ client.coilReads = 0
+
+ return connection, client, transport
+}
// Module occupies its slice index in Settings.Modules. PulseWidths maps digital
// output channels to hardware pulse widths. Widths must be whole milliseconds
-// from 50 through 65535 milliseconds. Other digital outputs use level mode.
-// The constructor copies the topology and does not retain these maps.
+// from 50 through 65535 milliseconds. Other digital outputs use level mode. The
+// constructor copies the topology and does not retain these maps.
type Module struct {
Model string
PulseWidths map[int]time.Duration
--- /dev/null
+package m31
+
+import "time"
+
+// Settings defines one assembly. DeviceID is a canonical ULID. Address contains
+// a host and TCP port. Unit is from one through 247. PollInterval is from ten
+// milliseconds through one minute. Modules has one host followed by at most
+// fifteen expansions in physical order. The constructor retains no slice or map
+// from the caller.
+type Settings struct {
+ DeviceID string
+ Address string
+ Unit byte
+ PollInterval time.Duration
+ Modules []Module
+}
--- /dev/null
+package m31
+
+import "time"
+
+type snapshot struct {
+ inputs []bool
+ outputs []bool
+ analog []uint16
+ observedAt time.Time
+}
projects / onehouse / commitdiff