From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 23:08:38 +0000 (-0300) Subject: Poll M31 assemblies with verified reconnect setup X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/2392f423feb689c99600e5cb2c37ee93c62e603d Poll M31 assemblies with verified reconnect setup The M31 driver validates settings, verifies every output mode and pulse width before online, and establishes an input baseline without events. Later reads publish state and timed input transitions through the shared publisher. Offline recovery uses bounded backoff without duplicate publication mechanics. Publication and cleanup errors return to the runner, and cancellation closes the owned connection. Issue: 8fdc39a --- diff --git a/README.rst b/README.rst index 4114653..69b14d2 100644 --- a/README.rst +++ b/README.rst @@ -5,7 +5,7 @@ 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 diff --git a/doc/m31.rst b/doc/m31.rst index a1b6eea..5b374ef 100644 --- a/doc/m31.rst +++ b/doc/m31.rst @@ -42,3 +42,33 @@ package is required. 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. diff --git a/driver/m31/client.go b/driver/m31/client.go index 140e13e..8c65d03 100644 --- a/driver/m31/client.go +++ b/driver/m31/client.go @@ -5,8 +5,8 @@ import "context" // 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 diff --git a/driver/m31/connection.go b/driver/m31/connection.go new file mode 100644 index 0000000..2230f91 --- /dev/null +++ b/driver/m31/connection.go @@ -0,0 +1,215 @@ +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 +} diff --git a/driver/m31/connection_test.go b/driver/m31/connection_test.go new file mode 100644 index 0000000..f2e67a6 --- /dev/null +++ b/driver/m31/connection_test.go @@ -0,0 +1,132 @@ +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) +} diff --git a/driver/m31/device_error.go b/driver/m31/device_error.go new file mode 100644 index 0000000..1b60654 --- /dev/null +++ b/driver/m31/device_error.go @@ -0,0 +1,15 @@ +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 +} diff --git a/driver/m31/driver.go b/driver/m31/driver.go new file mode 100644 index 0000000..9201a59 --- /dev/null +++ b/driver/m31/driver.go @@ -0,0 +1,164 @@ +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 + } +} diff --git a/driver/m31/driver_test.go b/driver/m31/driver_test.go new file mode 100644 index 0000000..1ba6190 --- /dev/null +++ b/driver/m31/driver_test.go @@ -0,0 +1,234 @@ +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) + }) +} diff --git a/driver/m31/fake_helper_test.go b/driver/m31/fake_helper_test.go new file mode 100644 index 0000000..c53dbac --- /dev/null +++ b/driver/m31/fake_helper_test.go @@ -0,0 +1,274 @@ +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 +} diff --git a/driver/m31/module.go b/driver/m31/module.go index 173aebd..55b6668 100644 --- a/driver/m31/module.go +++ b/driver/m31/module.go @@ -22,8 +22,8 @@ const ( // 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 diff --git a/driver/m31/settings.go b/driver/m31/settings.go new file mode 100644 index 0000000..b2692ab --- /dev/null +++ b/driver/m31/settings.go @@ -0,0 +1,16 @@ +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 +} diff --git a/driver/m31/snapshot.go b/driver/m31/snapshot.go new file mode 100644 index 0000000..8c29d12 --- /dev/null +++ b/driver/m31/snapshot.go @@ -0,0 +1,10 @@ +package m31 + +import "time" + +type snapshot struct { + inputs []bool + outputs []bool + analog []uint16 + observedAt time.Time +}