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. ``doc/m31.rst`` describes the M31 integration.
+its wire contract. ``doc/m31.rst`` and ``doc/alamo.rst`` describe the
+device integrations.
*************
Development
``idle``, or ``heating``), and ``locked``. Commands contain the complete
mode and target settings only. House-specific target bounds and steps
belong to core validation. The driver accepts finite targets.
+
+****************
+ Gateway driver
+****************
+
+``New`` takes ``Settings``, the shared publisher and router, and a raw
+``Client``. Settings list all stable thermostat IDs and their unique MAC
+values for one physical gateway. The default poll interval is fifteen
+seconds. An explicit interval is from ten seconds through five minutes.
+The caller runs the publisher, supplies command intake, and owns the
+HTTP transport. The command keeps its explicit empty driver list.
+
+``Run`` serializes polling and commands across the gateway. One status
+request supplies every configured thermostat. It publishes no startup
+availability before evidence. Three failed or missing observations mark
+a thermostat offline. A successful observation resets that count and
+restores online state. A command error does not count as a missed poll.
+Recoverable HTTP failures appear as missed polls or failed command
+results, not logs. Publication failures return to the framework runner.
+Shutdown does not publish an outage. ``Refresh`` coalesces a
+rate-limited status poll and asks the shared publisher to resend cached
+snapshots.
+
+Each command reads a fresh action baseline, turns off before a target
+change, sets and verifies the target, and only then turns on if
+requested. Every action response is feedback. A failed intermediate
+readback stops later actions. A command can leave the thermostat off if
+a later action fails or expires. The driver publishes the last
+successful observation with its observation time and command causation.
+A completed result requires matching power and target readback. A
+complete no-change command still reads the gateway.
+
+One queue consumer per thermostat preserves command order. Work from
+separate thermostats has no cross-device order guarantee. The shared
+ledger stores an outcome before feedback or result publication.
+Duplicate commands replay the first result without another gateway
+exchange, also after a failed publication. The ledger is memory-only
+with the framework limit of 256 results per device. Restart or eviction
+ends that duplicate protection. A gateway request can have an uncertain
+physical outcome when its response fails. The driver does not retry that
+command action.
// Client performs one raw gateway exchange. The driver serializes exchanges and
// applies the shared request interval. Implementations honor ctx, return full
// validated records, and do not log. The caller owns the client and its
-// cleanup.
+// cleanup. Errors omit addresses, queries, and response data.
type Client interface {
Status(ctx context.Context) ([]State, error)
Action(ctx context.Context, mac string, action Action) (State, error)
--- /dev/null
+package alamo
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+type commandHandler struct {
+ gateway *gateway
+ thermostat *thermostatState
+}
+
+func (h *commandHandler) handle(ctx context.Context, commands []driver.Command) error {
+ for _, command := range commands {
+ err := h.one(ctx, command)
+ if err != nil {
+ return fmt.Errorf("process thermostat command: %w", err)
+ }
+ }
+
+ return nil
+}
+
+func (h *commandHandler) one(ctx context.Context, command driver.Command) error {
+ s := h.thermostat
+ if command.DeviceID != s.settings.DeviceID {
+ return fmt.Errorf("command device is not owned by this thermostat")
+ }
+
+ record, exists := s.ledger.Lookup(command.Metadata.MessageID)
+ if exists {
+ err := s.publisher.Result(ctx, s.settings.DeviceID, command.Metadata.MessageID, record.Metadata, record.Payload)
+ if err != nil {
+ return fmt.Errorf("replay thermostat result: %w", err)
+ }
+
+ return nil
+ }
+
+ reason := ""
+ settings, err := decodeControl(command.Payload.Value)
+
+ switch {
+ case command.Expired(time.Now()):
+ reason = "Command expired"
+ case command.Capability != "control":
+ reason = "Capability is not the thermostat control"
+ case err != nil:
+ reason = "Thermostat settings are invalid"
+ }
+
+ if reason != "" {
+ err = s.result(ctx, command, protocol.ResultPayload{
+ Status: "rejected",
+ Reason: reason,
+ })
+ if err != nil {
+ return fmt.Errorf("reject thermostat command: %w", err)
+ }
+
+ return nil
+ }
+
+ state, observedAt, applyErr := h.apply(ctx, command, settings)
+ payload := protocol.ResultPayload{Status: "completed"}
+ if applyErr != nil {
+ payload.Status = "failed"
+ payload.Reason = "Thermostat request or readback failed"
+ if command.Expired(time.Now()) {
+ payload.Reason = "Command expired"
+ }
+ }
+
+ record, err = s.ledger.Remember(command.Metadata.MessageID, payload)
+ if err != nil {
+ return fmt.Errorf("remember applied thermostat result: %w", err)
+ }
+
+ if ctx.Err() != nil {
+ if applyErr != nil {
+ return fmt.Errorf("stop thermostat command: %w", errors.Join(ctx.Err(), applyErr))
+ }
+
+ return ctx.Err()
+ }
+
+ if !observedAt.IsZero() {
+ err = s.record(ctx, state, observedAt, command.Metadata.MessageID)
+ if err != nil {
+ return fmt.Errorf("record command feedback: %w", err)
+ }
+ }
+
+ err = s.publisher.Result(ctx, s.settings.DeviceID, command.Metadata.MessageID, record.Metadata, record.Payload)
+ if err != nil {
+ return fmt.Errorf("publish applied thermostat result: %w", err)
+ }
+
+ return nil
+}
+
+func (h *commandHandler) apply(ctx context.Context, command driver.Command, settings controlSettings) (State, time.Time, error) {
+ ctx, cancel := context.WithDeadline(ctx, *command.Metadata.ExpiresAt)
+ defer cancel()
+
+ mac := h.thermostat.settings.MAC
+ state, err := h.gateway.action(ctx, mac, Action{})
+ if err != nil {
+ return State{}, time.Time{}, fmt.Errorf("read command baseline: %w", err)
+ }
+
+ observedAt := time.Now().UTC()
+
+ var next State
+
+ wantPower := settings.Mode == "heat"
+ if state.Power && (!wantPower || state.TargetTemperature != settings.TargetTemperature) {
+ power := false
+ next, err = h.gateway.action(ctx, mac, Action{Power: &power})
+ if err != nil {
+ return state, observedAt, fmt.Errorf("turn thermostat off: %w", err)
+ }
+
+ state = next
+ observedAt = time.Now().UTC()
+ if state.Power {
+ return state, observedAt, fmt.Errorf("thermostat remains on before target change")
+ }
+ }
+
+ if state.TargetTemperature != settings.TargetTemperature {
+ next, err = h.gateway.action(ctx, mac, Action{TargetTemperature: &settings.TargetTemperature})
+ if err != nil {
+ return state, observedAt, fmt.Errorf("set thermostat target: %w", err)
+ }
+
+ state = next
+ observedAt = time.Now().UTC()
+ if state.TargetTemperature != settings.TargetTemperature {
+ return state, observedAt, fmt.Errorf("thermostat target readback differs")
+ }
+ }
+
+ if wantPower && !state.Power {
+ power := true
+ next, err = h.gateway.action(ctx, mac, Action{Power: &power})
+ if err != nil {
+ return state, observedAt, fmt.Errorf("turn thermostat on: %w", err)
+ }
+
+ state = next
+ observedAt = time.Now().UTC()
+ }
+
+ if state.Power != wantPower || state.TargetTemperature != settings.TargetTemperature {
+ return state, observedAt, fmt.Errorf("thermostat settings readback differs")
+ }
+
+ return state, observedAt, nil
+}
--- /dev/null
+package alamo
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+func TestCommandHandlerOrdersPowerAndTargetWithActionReadback(t *testing.T) {
+ testCases := []struct {
+ name string
+ initialPower bool
+ value string
+ wantActions []string
+ }{
+ {
+ name: "target precedes power on",
+ value: `{"mode":"heat","target_temperature":22}`,
+ wantActions: []string{"read", "target", "on"},
+ },
+ {
+ name: "power off precedes target change",
+ initialPower: true,
+ value: `{"mode":"off","target_temperature":22}`,
+ wantActions: []string{"read", "off", "target"},
+ },
+ {
+ name: "active target change turns off then on",
+ initialPower: true,
+ value: `{"mode":"heat","target_temperature":22}`,
+ wantActions: []string{"read", "off", "target", "on"},
+ },
+ {
+ name: "equal settings still get fresh readback",
+ value: `{"mode":"off","target_temperature":20}`,
+ wantActions: []string{"read"},
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ state := testState()
+ state.Power = tc.initialPower
+ client := &fakeGateway{states: []State{state}}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+ handler := &commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ command := testCommand()
+ command.Payload.Value = []byte(tc.value)
+ require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
+ require.Equal(t, tc.wantActions, actionNames(client.actions))
+ require.Len(t, transport.messages(t, protocol.Result), 1)
+
+ result := transport.messages(t, protocol.Result)[0]
+ require.Equal(t, protocol.ResultPayload{Status: "completed"}, result.Payload)
+
+ reported := transport.messages(t, protocol.Reported)
+ require.Len(t, reported, 1)
+ require.Equal(t, command.Metadata.MessageID, reported[0].Metadata.CausationID)
+
+ for index := 1; index < len(client.starts); index++ {
+ require.GreaterOrEqual(t, client.starts[index].Sub(client.starts[index-1]), 6*time.Second)
+ }
+ })
+ })
+ }
+}
+
+func TestCommandHandlerStopsAtUnmatchedIntermediateReadback(t *testing.T) {
+ testCases := []struct {
+ name string
+ power bool
+ ignorePower bool
+ ignoreTarget bool
+ wantActions []string
+ }{
+ {
+ name: "failed off confirmation prevents target write",
+ power: true,
+ ignorePower: true,
+ wantActions: []string{"read", "off"},
+ },
+ {
+ name: "failed target confirmation prevents power on",
+ ignoreTarget: true,
+ wantActions: []string{"read", "target"},
+ },
+ {
+ name: "failed power on confirmation does not complete",
+ ignorePower: true,
+ wantActions: []string{"read", "target", "on"},
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ state := testState()
+ state.Power = tc.power
+ client := &fakeGateway{
+ states: []State{state},
+ ignorePower: tc.ignorePower,
+ ignoreTarget: tc.ignoreTarget,
+ }
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ require.NoError(t, handler.handle(t.Context(), []driver.Command{testCommand()}))
+ require.Equal(t, tc.wantActions, actionNames(client.actions))
+
+ result, ok := transport.messages(t, protocol.Result)[0].Payload.(protocol.ResultPayload)
+ require.True(t, ok)
+ require.Equal(t, "failed", result.Status)
+ require.NotEmpty(t, result.Reason)
+ })
+ })
+ }
+}
+
+func TestCommandHandlerReplaysFirstResultAfterFeedbackOrResultSendFailure(t *testing.T) {
+ for _, kind := range []protocol.Kind{protocol.Reported, protocol.Result, protocol.Availability} {
+ t.Run(string(kind), func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ failure := errors.New("transport unavailable")
+ transport := &fakeTransport{
+ failKind: kind,
+ failure: failure,
+ }
+ client := &fakeGateway{states: []State{testState()}}
+ integration, _ := testDriver(t, client, transport)
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ command := testCommand()
+ require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
+
+ calls := len(client.actions)
+ record, exists := integration.thermostats[0].ledger.Lookup(command.Metadata.MessageID)
+ require.True(t, exists)
+ require.Equal(t, "completed", record.Payload.Status)
+
+ transport.failKind = ""
+
+ require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
+ require.Len(t, client.actions, calls)
+
+ messages := transport.messages(t, protocol.Result)
+ last := messages[len(messages)-1]
+ require.Equal(t, record.Metadata, last.Metadata)
+ require.Equal(t, record.Payload, last.Payload)
+ })
+ })
+ }
+}
+
+func TestCommandHandlerRejectsInvalidCommandsWithoutGatewayWork(t *testing.T) {
+ testCases := []struct {
+ name string
+ change func(*driver.Command)
+ }{
+ {
+ name: "expired work is rejected",
+ change: func(command *driver.Command) {
+ expires := time.Now().Add(-time.Second).UTC()
+ command.Metadata.ExpiresAt = &expires
+ },
+ },
+ {
+ name: "missing expiry is rejected",
+ change: func(command *driver.Command) {
+ command.Metadata.ExpiresAt = nil
+ },
+ },
+ {
+ name: "another capability is rejected",
+ change: func(command *driver.Command) {
+ command.Capability = "other"
+ },
+ },
+ {
+ name: "incomplete settings are rejected",
+ change: func(command *driver.Command) {
+ command.Payload.Value = []byte(`{}`)
+ },
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ client := &fakeGateway{}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ command := testCommand()
+ tc.change(&command)
+ require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
+ require.Empty(t, client.starts)
+
+ messages := transport.messages(t, protocol.Result)
+ require.Len(t, messages, 2)
+ require.Equal(t, messages[0], messages[1])
+
+ result, ok := messages[0].Payload.(protocol.ResultPayload)
+ require.True(t, ok)
+ require.Equal(t, "rejected", result.Status)
+ })
+ }
+}
+
+func TestCommandHandlerCommandExpiryStopsLaterActionsAndReplay(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := &fakeGateway{states: []State{testState()}}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ command := testCommand()
+ expires := time.Now().Add(7 * time.Second).UTC()
+ command.Metadata.ExpiresAt = &expires
+ require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
+ require.Equal(t, []string{"read", "target"}, actionNames(client.actions))
+ require.False(t, client.states[0].Power)
+
+ reported := transport.messages(t, protocol.Reported)
+ require.Len(t, reported, 1)
+ require.Equal(t, client.starts[1].UTC(), *reported[0].Metadata.SentAt)
+ require.NoError(t, handler.handle(t.Context(), []driver.Command{command}))
+ require.Len(t, client.actions, 2)
+
+ messages := transport.messages(t, protocol.Result)
+ require.Len(t, messages, 2)
+ require.Equal(t, messages[0], messages[1])
+ require.Equal(t, protocol.ResultPayload{
+ Status: "failed",
+ Reason: "Command expired",
+ }, messages[0].Payload)
+ })
+}
+
+func TestCommandHandlerRetainsFailuresWithoutRepeatingPhysicalActions(t *testing.T) {
+ for _, position := range []int{1, 2, 3, 4} {
+ t.Run(actionFailureName(position), func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ state := testState()
+ state.Power = true
+ client := &fakeGateway{
+ states: []State{state},
+ failActionAt: position,
+ }
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ command := testCommand()
+ require.NoError(t, handler.handle(t.Context(), []driver.Command{command, command}))
+ require.Len(t, client.actions, position)
+
+ messages := transport.messages(t, protocol.Result)
+ require.Len(t, messages, 2)
+ require.Equal(t, messages[0], messages[1])
+
+ payload, ok := messages[0].Payload.(protocol.ResultPayload)
+ require.True(t, ok)
+ require.Equal(t, "failed", payload.Status)
+ })
+ })
+ }
+}
+
+func TestCommandHandlerValidatesOwnershipAndReturnsPublicationFailures(t *testing.T) {
+ client := &fakeGateway{}
+ failure := errors.New("result unavailable")
+ transport := &fakeTransport{
+ failKind: protocol.Result,
+ failure: failure,
+ }
+ integration, _ := testDriver(t, client, transport)
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ command := testCommand()
+ command.DeviceID = testSecondID
+ require.ErrorContains(t, handler.handle(t.Context(), []driver.Command{command}), "not owned")
+
+ command = testCommand()
+ command.Capability = "invalid"
+ require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
+ require.ErrorIs(t, handler.handle(t.Context(), []driver.Command{command}), failure)
+ require.Empty(t, client.starts)
+}
+
+func TestCommandHandlerCancellationStopsBeforeAnotherActionAndKeepsOutcome(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := &fakeGateway{states: []State{testState()}}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ done := make(chan error, 1)
+ command := testCommand()
+
+ go func() {
+ done <- handler.handle(ctx, []driver.Command{command})
+ }()
+
+ synctest.Wait()
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ require.Len(t, client.actions, 1)
+
+ _, exists := integration.thermostats[0].ledger.Lookup(command.Metadata.MessageID)
+ require.True(t, exists)
+ require.Empty(t, transport.messages(t, protocol.Result))
+ })
+}
+
+func TestCommandHandlerKeepsIndependentFailureDuringCancellation(t *testing.T) {
+ failure := errors.New("independent gateway failure")
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ client := &cancelingGateway{
+ cancel: cancel,
+ failure: failure,
+ }
+ integration, _ := testDriver(t, client, &fakeTransport{})
+ handler := commandHandler{
+ gateway: &integration.gateway,
+ thermostat: integration.thermostats[0],
+ }
+ err := handler.handle(ctx, []driver.Command{testCommand()})
+ require.ErrorIs(t, err, context.Canceled)
+ require.ErrorIs(t, err, failure)
+}
+
+func actionNames(actions []Action) []string {
+ var names []string
+
+ for _, action := range actions {
+ name := "read"
+ if action.TargetTemperature != nil {
+ name = "target"
+ }
+
+ if action.Power != nil {
+ name = "off"
+ if *action.Power {
+ name = "on"
+ }
+ }
+
+ names = append(names, name)
+ }
+
+ return names
+}
+
+func actionFailureName(position int) string {
+ return []string{"", "failed initial read", "failed power off", "failed target write", "failed power on"}[position]
+}
--- /dev/null
+package alamo
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "sync"
+ "time"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+// Driver owns one gateway's shared polling, request budget, and command work.
+// Run has one owner. HTTP failures count toward missed polls or command failure
+// results. Publication failures return to the runner. No method logs. The
+// caller runs the shared publisher alongside the driver and owns the Client.
+// Refresh is safe during Run. Ledger protection is memory-only and ends at
+// eviction or process restart. Construct one driver per physical gateway.
+type Driver struct {
+ gateway gateway
+ pollInterval time.Duration
+ thermostats []*thermostatState
+ refresh chan struct{}
+ pending chan queuedCommands
+ publisher *driver.Publisher
+}
+
+// New validates settings and command queues without I/O. The caller supplies
+// valid publisher, router, and client dependencies, and starts router intake
+// separately. The constructor retains no caller-owned settings slice.
+func New(settings Settings, publisher *driver.Publisher, router *driver.Router, client Client) (*Driver, error) {
+ if len(settings.Thermostats) == 0 {
+ return nil, fmt.Errorf("gateway requires at least one thermostat")
+ }
+
+ interval := settings.PollInterval
+ if interval == 0 {
+ interval = 15 * time.Second
+ }
+
+ if interval < 10*time.Second || interval > 5*time.Minute {
+ return nil, fmt.Errorf("poll interval must be from ten seconds through five minutes")
+ }
+
+ integration := &Driver{
+ gateway: gateway{client: client},
+ pollInterval: interval,
+ refresh: make(chan struct{}, 1),
+ pending: make(chan queuedCommands),
+ publisher: publisher,
+ }
+ ids := make(map[string]bool)
+ macs := make(map[string]bool)
+
+ for _, thermostat := range settings.Thermostats {
+ if !protocol.ValidID(thermostat.DeviceID) || ids[thermostat.DeviceID] {
+ return nil, fmt.Errorf("thermostat device IDs must be unique canonical ULIDs")
+ }
+
+ mac, err := normalizeMAC(thermostat.MAC)
+ if err != nil {
+ return nil, fmt.Errorf("validate thermostat MAC: %w", err)
+ }
+
+ if macs[mac] {
+ return nil, fmt.Errorf("thermostat MAC values must be unique")
+ }
+
+ queue, exists := router.Queue(thermostat.DeviceID)
+ if !exists {
+ return nil, fmt.Errorf("thermostat has no command queue")
+ }
+
+ ledger, err := driver.NewLedger(thermostat.DeviceID)
+ if err != nil {
+ return nil, fmt.Errorf("create thermostat ledger: %w", err)
+ }
+
+ ids[thermostat.DeviceID] = true
+ macs[mac] = true
+ thermostat.MAC = mac
+ integration.thermostats = append(integration.thermostats, &thermostatState{
+ settings: thermostat,
+ queue: queue,
+ ledger: ledger,
+ publisher: publisher,
+ })
+ }
+
+ return integration, nil
+}
+
+// Run publishes descriptors, then shares each poll across the thermostat list.
+// Three missed observations mark a thermostat offline. A full observation
+// restores it. Shutdown does not publish offline. A command and its actions run
+// in order through the gateway budget and the shared result ledger.
+func (d *Driver) Run(ctx context.Context) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ var workers sync.WaitGroup
+
+ defer func() {
+ cancel()
+ workers.Wait()
+ }()
+
+ for _, thermostat := range d.thermostats {
+ err := thermostat.descriptor(ctx)
+ if err != nil {
+ return fmt.Errorf("publish thermostat descriptor: %w", err)
+ }
+
+ workers.Go(func() {
+ d.forward(ctx, thermostat)
+ })
+ }
+
+ err := d.poll(ctx)
+ if err != nil {
+ return fmt.Errorf("poll initial thermostat states: %w", err)
+ }
+
+ ticker := time.NewTicker(d.pollInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-d.refresh:
+ err = d.poll(ctx)
+ if err != nil {
+ return fmt.Errorf("refresh thermostat states: %w", err)
+ }
+ case <-ticker.C:
+ err = d.poll(ctx)
+ if err != nil {
+ return fmt.Errorf("poll thermostat states: %w", err)
+ }
+ case pending := <-d.pending:
+ handler := commandHandler{
+ gateway: &d.gateway,
+ thermostat: pending.thermostat,
+ }
+
+ err = handler.handle(ctx, pending.commands)
+ if err != nil {
+ return fmt.Errorf("process thermostat commands: %w", err)
+ }
+ }
+ }
+}
+
+func (d *Driver) forward(ctx context.Context, thermostat *thermostatState) {
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case commands := <-thermostat.queue:
+ select {
+ case <-ctx.Done():
+ return
+ case d.pending <- queuedCommands{
+ thermostat: thermostat,
+ commands: commands,
+ }:
+ }
+ }
+ }
+}
+
+func (d *Driver) poll(ctx context.Context) error {
+ states, err := d.gateway.status(ctx)
+ if ctx.Err() != nil {
+ if err != nil {
+ return fmt.Errorf("stop gateway poll: %w", errors.Join(ctx.Err(), err))
+ }
+
+ return ctx.Err()
+ }
+
+ observedAt := time.Now().UTC()
+ observed := make(map[string]State, len(states))
+ if err == nil {
+ for _, state := range states {
+ observed[state.MAC] = state
+ }
+ }
+
+ for _, thermostat := range d.thermostats {
+ state, exists := observed[thermostat.settings.MAC]
+ if exists {
+ err = thermostat.record(ctx, state, observedAt, "")
+ } else {
+ err = thermostat.miss(ctx)
+ }
+
+ if err != nil {
+ return fmt.Errorf("record thermostat poll: %w", err)
+ }
+ }
+
+ return nil
+}
+
+// Refresh coalesces one rate-limited status poll and asks the shared publisher
+// to resend its snapshots. It does not repeat physical command actions.
+func (d *Driver) Refresh() {
+ d.publisher.Refresh()
+
+ select {
+ case d.refresh <- struct{}{}:
+ default:
+ }
+}
--- /dev/null
+package alamo
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+func TestDriverPollSharesOneSnapshotAcrossThermostats(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ first := testState()
+ second := first
+ second.MAC = "fc610e7f98ab"
+ client := &fakeGateway{states: []State{first, second}}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport, Thermostat{
+ DeviceID: testDeviceID,
+ MAC: testMAC,
+ }, Thermostat{
+ DeviceID: testSecondID,
+ MAC: second.MAC,
+ })
+
+ err := integration.poll(t.Context())
+ require.NoError(t, err)
+ require.Len(t, client.starts, 1)
+ require.Len(t, transport.messages(t, protocol.Availability), 2)
+ require.Len(t, transport.messages(t, protocol.Reported), 2)
+ })
+}
+
+func TestDriverMarksOnlyMissingThermostatOfflineAfterThreePollsAndRecovers(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ second := testState()
+ second.MAC = "fc610e7f98ab"
+ client := &fakeGateway{states: []State{testState(), second}}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport, Thermostat{
+ DeviceID: testDeviceID,
+ MAC: testMAC,
+ }, Thermostat{
+ DeviceID: testSecondID,
+ MAC: second.MAC,
+ })
+
+ require.NoError(t, integration.poll(t.Context()))
+
+ client.states = []State{second}
+
+ for range 2 {
+ require.NoError(t, integration.poll(t.Context()))
+ require.Len(t, transport.messages(t, protocol.Availability), 2)
+ }
+
+ require.NoError(t, integration.poll(t.Context()))
+
+ messages := transport.messages(t, protocol.Availability)
+ require.Len(t, messages, 3)
+ require.Equal(t, testDeviceID, messages[2].Address.ID)
+ require.Equal(t, protocol.AvailabilityPayload{Status: "offline"}, messages[2].Payload)
+ require.NoError(t, integration.poll(t.Context()))
+ require.Len(t, transport.messages(t, protocol.Availability), 3)
+
+ client.states = []State{testState(), second}
+
+ require.NoError(t, integration.poll(t.Context()))
+
+ messages = transport.messages(t, protocol.Availability)
+ require.Len(t, messages, 4)
+ require.Equal(t, protocol.AvailabilityPayload{Status: "online"}, messages[3].Payload)
+ })
+}
+
+func TestDriverFailedPollsHaveNoPrematureStartupAvailability(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := &fakeGateway{err: errors.New("gateway unavailable")}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+
+ for range 2 {
+ require.NoError(t, integration.poll(t.Context()))
+ require.Empty(t, transport.messages(t, protocol.Availability))
+ }
+
+ require.NoError(t, integration.poll(t.Context()))
+
+ messages := transport.messages(t, protocol.Availability)
+ require.Len(t, messages, 1)
+ require.Equal(t, protocol.AvailabilityPayload{Status: "offline"}, messages[0].Payload)
+
+ client.err = nil
+ client.states = []State{testState()}
+
+ require.NoError(t, integration.poll(t.Context()))
+
+ client.err = errors.New("gateway unavailable")
+
+ require.NoError(t, integration.poll(t.Context()))
+ require.Len(t, transport.messages(t, protocol.Availability), 2)
+ })
+}
+
+func TestDriverRunForwardsCommandsRefreshesAndStopsWithoutOutage(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := &fakeGateway{states: []State{testState()}}
+ transport := &fakeTransport{}
+ integration, router := testDriver(t, client, transport)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ integration.Refresh()
+ integration.Refresh()
+
+ done := make(chan error, 1)
+
+ go func() {
+ done <- integration.Run(ctx)
+ }()
+
+ synctest.Wait()
+ require.Len(t, transport.messages(t, protocol.Descriptor), 1)
+ require.Equal(t, 1, client.startCount())
+ time.Sleep(6 * time.Second)
+ synctest.Wait()
+ require.Equal(t, 2, client.startCount())
+
+ command := testCommand()
+ publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+ require.NoError(t, err)
+ require.NoError(t, router.Handle(ctx, publication.Topic, publication.Data, false))
+ require.NoError(t, router.Handle(ctx, publication.Topic, publication.Data, false))
+
+ time.Sleep(40 * time.Second)
+ synctest.Wait()
+ require.Len(t, transport.messages(t, protocol.Result), 2)
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ require.Len(t, client.actions, 3)
+
+ messages := transport.messages(t, protocol.Availability)
+
+ for _, message := range messages {
+ require.Equal(t, protocol.AvailabilityPayload{Status: "online"}, message.Payload)
+ }
+ })
+}
+
+func TestDriverConstructorValidatesAndCopiesSettings(t *testing.T) {
+ router, err := driver.NewRouter([]string{testDeviceID, testSecondID})
+ require.NoError(t, err)
+
+ publisher := driver.NewPublisher(&fakeTransport{})
+ valid := Thermostat{
+ DeviceID: testDeviceID,
+ MAC: testMAC,
+ }
+ testCases := []struct {
+ name string
+ settings Settings
+ }{
+ {
+ name: "empty gateway is rejected",
+ },
+ {
+ name: "invalid ID is rejected",
+ settings: Settings{Thermostats: []Thermostat{{
+ DeviceID: "bad",
+ MAC: testMAC,
+ }}},
+ },
+ {
+ name: "invalid MAC is rejected",
+ settings: Settings{Thermostats: []Thermostat{{
+ DeviceID: testDeviceID,
+ MAC: "bad",
+ }}},
+ },
+ {
+ name: "duplicate ID is rejected",
+ settings: Settings{Thermostats: []Thermostat{valid, valid}},
+ },
+ {
+ name: "MAC uniqueness ignores letter case",
+ settings: Settings{Thermostats: []Thermostat{valid, {
+ DeviceID: testSecondID,
+ MAC: strings.ToUpper(testMAC),
+ }}},
+ },
+ {
+ name: "poll interval cannot exceed five minutes",
+ settings: Settings{
+ PollInterval: time.Hour,
+ Thermostats: []Thermostat{valid},
+ },
+ },
+ {
+ name: "poll interval cannot be below ten seconds",
+ settings: Settings{
+ PollInterval: time.Second,
+ Thermostats: []Thermostat{valid},
+ },
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ _, err = New(tc.settings, publisher, router, &fakeGateway{})
+ require.Error(t, err)
+ })
+ }
+
+ emptyRouter, err := driver.NewRouter(nil)
+ require.NoError(t, err)
+
+ _, err = New(Settings{Thermostats: []Thermostat{valid}}, publisher, emptyRouter, &fakeGateway{})
+ require.ErrorContains(t, err, "queue")
+
+ settings := Settings{
+ PollInterval: 10 * time.Second,
+ Thermostats: []Thermostat{valid},
+ }
+ integration, err := New(settings, publisher, router, &fakeGateway{})
+ require.NoError(t, err)
+
+ settings.Thermostats[0].MAC = "bad"
+ require.Equal(t, testMAC, integration.thermostats[0].settings.MAC)
+ require.Equal(t, 10*time.Second, integration.pollInterval)
+}
+
+func TestDriverReturnsPublicationFailuresAndCallerCancellation(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ failure := errors.New("publication unavailable")
+ transport := &fakeTransport{
+ failKind: protocol.Descriptor,
+ failure: failure,
+ }
+ integration, _ := testDriver(t, &fakeGateway{states: []State{testState()}}, transport)
+ require.ErrorIs(t, integration.Run(t.Context()), failure)
+
+ transport.failKind = protocol.Reported
+
+ require.ErrorIs(t, integration.Run(t.Context()), failure)
+
+ transport.failKind = protocol.Availability
+ integration.thermostats[0].missed = 2
+ integration.gateway.client = &fakeGateway{}
+ require.ErrorIs(t, integration.poll(t.Context()), failure)
+
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+
+ require.ErrorIs(t, integration.poll(ctx), context.Canceled)
+ })
+}
+
+func TestDriverReturnsRefreshAndPeriodicPublicationFailures(t *testing.T) {
+ for _, refresh := range []bool{false, true} {
+ t.Run(publicationFailureName(refresh), func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := &fakeGateway{states: []State{testState()}}
+ transport := &fakeTransport{}
+ integration, _ := testDriver(t, client, transport)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ done := make(chan error, 1)
+
+ go func() {
+ done <- integration.Run(ctx)
+ }()
+
+ synctest.Wait()
+
+ failure := errors.New("publication unavailable")
+
+ transport.mu.Lock()
+
+ transport.failKind = protocol.Reported
+ transport.failure = failure
+ transport.mu.Unlock()
+ client.mu.Lock()
+
+ client.states[0].TargetTemperature = 25
+ client.mu.Unlock()
+
+ if refresh {
+ integration.Refresh()
+ }
+
+ require.ErrorIs(t, <-done, failure)
+ })
+ })
+ }
+}
+
+func TestDriverReturnsCommandPublicationFailuresAndJoinsForwarders(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := &fakeGateway{states: []State{testState()}}
+ transport := &fakeTransport{
+ failKind: protocol.Result,
+ failure: errors.New("result unavailable"),
+ }
+ integration, router := testDriver(t, client, transport)
+ command := testCommand()
+ publication, err := protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+ require.NoError(t, err)
+ require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, false))
+ require.ErrorIs(t, integration.Run(t.Context()), transport.failure)
+ })
+}
+
+func TestDriverPollKeepsIndependentFailureDuringCancellation(t *testing.T) {
+ failure := errors.New("independent gateway failure")
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ client := &cancelingGateway{
+ cancel: cancel,
+ failure: failure,
+ }
+ integration, _ := testDriver(t, client, &fakeTransport{})
+ err := integration.poll(ctx)
+ require.ErrorIs(t, err, context.Canceled)
+ require.ErrorIs(t, err, failure)
+}
+
+func publicationFailureName(refresh bool) string {
+ if refresh {
+ return "refresh failure returns to runner"
+ }
+
+ return "periodic failure returns to runner"
+}
import (
"context"
+ "errors"
+ "fmt"
+ "sync"
+ "testing"
"time"
+
+ "github.com/stretchr/testify/require"
+
+ "onehouse/driver"
+ "onehouse/protocol"
)
-const testMAC = "fc610e7f98aa"
+const (
+ validRecord = `{"mac":"FC610E7F98AA","power":0,"stemp":20,"temp":19.5,"sensor":23,"ativo":0,"bloqueio":1}`
+
+ testMAC = "fc610e7f98aa"
+ testDeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+ testSecondID = "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+ testCommandID = "01ARZ3NDEKTSV4RRFFQ69G5FAX"
+)
type fakeGateway struct {
- states []State
- starts []time.Time
- actions []Action
- err error
- delay time.Duration
+ mu sync.Mutex
+ failActionAt int
+ ignorePower bool
+ ignoreTarget bool
+ states []State
+ starts []time.Time
+ actions []Action
+ err error
+ delay time.Duration
}
func (g *fakeGateway) Status(ctx context.Context) ([]State, error) {
+ g.mu.Lock()
+ defer g.mu.Unlock()
+
err := g.request(ctx)
if err != nil {
return nil, err
return append([]State(nil), g.states...), nil
}
+func (g *fakeGateway) request(ctx context.Context) error {
+ g.starts = append(g.starts, time.Now())
+ if g.delay > 0 {
+ timer := time.NewTimer(g.delay)
+ defer timer.Stop()
+
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-timer.C:
+ }
+ }
+
+ return g.err
+}
+
func (g *fakeGateway) Action(ctx context.Context, mac string, action Action) (State, error) {
+ g.mu.Lock()
+ defer g.mu.Unlock()
+
+ g.actions = append(g.actions, action)
+
err := g.request(ctx)
if err != nil {
return State{}, err
}
- g.actions = append(g.actions, action)
+ if len(g.actions) == g.failActionAt {
+ return State{}, errors.New("action unavailable")
+ }
for index := range g.states {
state := &g.states[index]
continue
}
- if action.Power != nil {
+ if action.Power != nil && !g.ignorePower {
state.Power = *action.Power
}
- if action.TargetTemperature != nil {
+ if action.TargetTemperature != nil && !g.ignoreTarget {
state.TargetTemperature = *action.TargetTemperature
}
return State{}, nil
}
-func (g *fakeGateway) request(ctx context.Context) error {
- g.starts = append(g.starts, time.Now())
- if g.delay > 0 {
- timer := time.NewTimer(g.delay)
- defer timer.Stop()
+func (g *fakeGateway) startCount() int {
+ g.mu.Lock()
+ defer g.mu.Unlock()
- select {
- case <-ctx.Done():
- return ctx.Err()
- case <-timer.C:
+ return len(g.starts)
+}
+
+type fakeTransport struct {
+ mu sync.Mutex
+ publications []protocol.Publication
+ failKind protocol.Kind
+ failure error
+}
+
+func (p *fakeTransport) Publish(_ context.Context, publication protocol.Publication) error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ p.publications = append(p.publications, publication)
+ 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 == p.failKind {
+ return p.failure
+ }
+
+ return nil
+}
+
+func (p *fakeTransport) messages(t *testing.T, kind protocol.Kind) []protocol.Message[any] {
+ t.Helper()
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ var messages []protocol.Message[any]
+
+ for _, publication := range p.publications {
+ message, err := protocol.Parse(publication.Topic, publication.Data, publication.Retain)
+ require.NoError(t, err)
+
+ if message.Address.Kind == kind {
+ messages = append(messages, message)
}
}
- return g.err
+ return messages
+}
+
+type cancelingGateway struct {
+ cancel context.CancelFunc
+ failure error
+}
+
+func (g *cancelingGateway) Status(_ context.Context) ([]State, error) {
+ g.cancel()
+
+ return nil, errors.Join(context.Canceled, g.failure)
+}
+
+func (g *cancelingGateway) Action(_ context.Context, _ string, _ Action) (State, error) {
+ g.cancel()
+
+ return State{}, errors.Join(context.Canceled, g.failure)
+}
+
+func testDriver(t *testing.T, client Client, transport *fakeTransport, thermostats ...Thermostat) (*Driver, *driver.Router) {
+ t.Helper()
+
+ if len(thermostats) == 0 {
+ thermostats = []Thermostat{{
+ DeviceID: testDeviceID,
+ MAC: testMAC,
+ }}
+ }
+
+ var ids []string
+
+ for _, thermostat := range thermostats {
+ ids = append(ids, thermostat.DeviceID)
+ }
+
+ router, err := driver.NewRouter(ids)
+ require.NoError(t, err)
+
+ integration, err := New(Settings{Thermostats: thermostats}, driver.NewPublisher(transport), router, client)
+ require.NoError(t, err)
+
+ return integration, router
+}
+
+func testCommand() driver.Command {
+ expires := time.Now().Add(time.Minute).UTC()
+
+ return driver.Command{
+ DeviceID: testDeviceID,
+ Capability: "control",
+ Metadata: protocol.Metadata{
+ MessageID: testCommandID,
+ ExpiresAt: &expires,
+ },
+ Payload: protocol.ValuePayload{Value: []byte(`{"mode":"heat","target_temperature":22}`)},
+ }
}
func testState() State {
return states, nil
}
-func (g *gateway) action(ctx context.Context, mac string, action Action) (State, error) {
- err := g.wait(ctx)
- if err != nil {
- return State{}, fmt.Errorf("wait for action slot: %w", err)
- }
-
- ctx, cancel := context.WithTimeout(ctx, gatewayExchangeTimeout)
- defer cancel()
-
- state, err := g.client.Action(ctx, mac, action)
- if err != nil {
- return State{}, fmt.Errorf("exchange gateway action: %w", err)
- }
-
- return state, nil
-}
-
func (g *gateway) wait(ctx context.Context) error {
err := ctx.Err()
if err != nil {
return err
}
+ deadline, exists := ctx.Deadline()
+ if exists && !time.Now().Before(deadline) {
+ return context.DeadlineExceeded
+ }
+
g.nextStart = time.Now().Add(gatewayRequestInterval)
return nil
}
+
+func (g *gateway) action(ctx context.Context, mac string, action Action) (State, error) {
+ err := g.wait(ctx)
+ if err != nil {
+ return State{}, fmt.Errorf("wait for action slot: %w", err)
+ }
+
+ ctx, cancel := context.WithTimeout(ctx, gatewayExchangeTimeout)
+ defer cancel()
+
+ state, err := g.client.Action(ctx, mac, action)
+ if err != nil {
+ return State{}, fmt.Errorf("exchange gateway action: %w", err)
+ }
+
+ return state, nil
+}
--- /dev/null
+package alamo
+
+import (
+ "encoding/json"
+ "math"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestGatewayRecordRejectsInvalidMACAndNonfiniteTemperatures(t *testing.T) {
+ var record gatewayRecord
+
+ require.NoError(t, json.Unmarshal([]byte(validRecord), &record))
+
+ invalid := "zzzzzzzzzzzz"
+ record.MAC = &invalid
+ _, err := record.state()
+ require.Error(t, err)
+
+ valid := testMAC
+ record.MAC = &valid
+ nonfinite := math.NaN()
+ record.Floor = &nonfinite
+ _, err = record.state()
+ require.ErrorContains(t, err, "finite")
+}
require.ErrorIs(t, err, client.err)
})
}
+
+func TestGatewayDoesNotStartAtTheExactCommandDeadline(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := &fakeGateway{states: []State{testState()}}
+ gateway := &gateway{client: client}
+ _, err := gateway.status(t.Context())
+ require.NoError(t, err)
+
+ ctx, cancel := context.WithTimeout(t.Context(), 6*time.Second)
+ defer cancel()
+
+ _, err = gateway.action(ctx, testMAC, Action{})
+ require.ErrorIs(t, err, context.DeadlineExceeded)
+ require.Len(t, client.starts, 1)
+ })
+}
return c.exchange(ctx, "/status", nil)
}
-// Action changes one field or reads a thermostat, and returns its complete
-// readback. It rejects a response for another MAC or more than one thermostat.
-func (c *HTTPClient) Action(ctx context.Context, mac string, action Action) (State, error) {
- mac, err := normalizeMAC(mac)
- if err != nil {
- return State{}, fmt.Errorf("validate action MAC: %w", err)
- }
-
- if action.Power != nil && action.TargetTemperature != nil {
- return State{}, fmt.Errorf("action must change at most one field")
- }
-
- query := url.Values{"mac": {mac}}
- if action.Power != nil {
- query.Set("power", "0")
-
- if *action.Power {
- query.Set("power", "1")
- }
- }
-
- if action.TargetTemperature != nil {
- if !finite(*action.TargetTemperature) {
- return State{}, fmt.Errorf("target temperature must be finite")
- }
-
- query.Set("stemp", strconv.FormatFloat(*action.TargetTemperature, 'f', -1, 64))
- }
-
- states, err := c.exchange(ctx, "/acao", query)
- if err != nil {
- return State{}, fmt.Errorf("exchange action: %w", err)
- }
-
- if len(states) != 1 || states[0].MAC != mac {
- return State{}, fmt.Errorf("action response must contain the requested thermostat only")
- }
-
- return states[0], nil
-}
-
func (c *HTTPClient) exchange(ctx context.Context, path string, query url.Values) (states []State, err error) {
ctx, cancel := context.WithTimeout(ctx, httpRequestTimeout)
defer cancel()
response, err := c.client.Do(request)
if err != nil {
- if ctx.Err() != nil {
- return nil, ctx.Err()
- }
-
- return nil, fmt.Errorf("send gateway request")
+ return nil, fmt.Errorf("send gateway request: %w", safeRequestError(err))
}
defer func() {
body, err := io.ReadAll(io.LimitReader(response.Body, gatewayResponseLimit+1))
if err != nil {
- if ctx.Err() != nil {
- return nil, ctx.Err()
- }
-
- return nil, fmt.Errorf("read gateway response")
+ return nil, fmt.Errorf("read gateway response: %w", safeRequestError(err))
}
if len(body) > gatewayResponseLimit {
seen := make(map[string]bool, len(records))
for _, record := range records {
- state, recordErr := record.state()
- if recordErr != nil {
- return nil, fmt.Errorf("validate gateway record: %w", recordErr)
+ var state State
+
+ state, err = record.state()
+ if err != nil {
+ return nil, fmt.Errorf("validate gateway record: %w", err)
}
if seen[state.MAC] {
return states, nil
}
+
+// Action changes one field or reads a thermostat, and returns its complete
+// readback. It rejects a response for another MAC or more than one thermostat.
+func (c *HTTPClient) Action(ctx context.Context, mac string, action Action) (State, error) {
+ mac, err := normalizeMAC(mac)
+ if err != nil {
+ return State{}, fmt.Errorf("validate action MAC: %w", err)
+ }
+
+ if action.Power != nil && action.TargetTemperature != nil {
+ return State{}, fmt.Errorf("action must change at most one field")
+ }
+
+ query := url.Values{"mac": {mac}}
+ if action.Power != nil {
+ query.Set("power", "0")
+
+ if *action.Power {
+ query.Set("power", "1")
+ }
+ }
+
+ if action.TargetTemperature != nil {
+ if !finite(*action.TargetTemperature) {
+ return State{}, fmt.Errorf("target temperature must be finite")
+ }
+
+ query.Set("stemp", strconv.FormatFloat(*action.TargetTemperature, 'f', -1, 64))
+ }
+
+ states, err := c.exchange(ctx, "/acao", query)
+ if err != nil {
+ return State{}, fmt.Errorf("exchange action: %w", err)
+ }
+
+ if len(states) != 1 || states[0].MAC != mac {
+ return State{}, fmt.Errorf("action response must contain the requested thermostat only")
+ }
+
+ return states[0], nil
+}
+
+func safeRequestError(err error) error {
+ var joined interface{ Unwrap() []error }
+
+ if errors.As(err, &joined) {
+ var causes []error
+
+ for _, cause := range joined.Unwrap() {
+ if cause != nil {
+ causes = append(causes, safeRequestError(cause))
+ }
+ }
+
+ return errors.Join(causes...)
+ }
+
+ if errors.Is(err, context.Canceled) {
+ return context.Canceled
+ }
+
+ if errors.Is(err, context.DeadlineExceeded) {
+ return context.DeadlineExceeded
+ }
+
+ return fmt.Errorf("gateway transport failed")
+}
"github.com/stretchr/testify/require"
)
-const validRecord = `{"mac":"FC610E7F98AA","power":0,"stemp":20,"temp":19.5,"sensor":23,"ativo":0,"bloqueio":1}`
-
func TestHTTPClientReadsAllRecordsAndOrderedActionFields(t *testing.T) {
var requests []*http.Request
require.Equal(t, 1, calls)
}
+func TestHTTPClientKeepsIndependentFailureAlongsideCancellation(t *testing.T) {
+ client, err := NewHTTPClient("http://example.com", &http.Client{Transport: roundTripFunc(func(_ *http.Request) (*http.Response, error) {
+ return nil, errors.Join(context.Canceled, errors.New("secret"))
+ })})
+ require.NoError(t, err)
+
+ _, err = client.Status(t.Context())
+ require.ErrorIs(t, err, context.Canceled)
+ require.ErrorContains(t, err, "gateway transport failed")
+ require.NotContains(t, err.Error(), "secret")
+}
+
type roundTripFunc func(*http.Request) (*http.Response, error)
func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
--- /dev/null
+package alamo
+
+import "onehouse/driver"
+
+type queuedCommands struct {
+ thermostat *thermostatState
+ commands []driver.Command
+}
--- /dev/null
+package alamo
+
+import "time"
+
+// Settings defines one gateway's complete thermostat list. PollInterval
+// defaults to fifteen seconds when zero, and otherwise is from ten seconds
+// through five minutes. Use exactly one driver and client owner per physical
+// gateway. The constructor copies the list and performs no I/O.
+type Settings struct {
+ PollInterval time.Duration
+ Thermostats []Thermostat
+}
--- /dev/null
+package alamo
+
+// Thermostat maps one stable canonical device ID to one gateway MAC. MAC
+// contains twelve hexadecimal digits and is normalized to lowercase. IDs and
+// MAC values must be unique in the gateway list.
+type Thermostat struct {
+ DeviceID string
+ MAC string
+}
--- /dev/null
+package alamo
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "time"
+
+ "onehouse/driver"
+ "onehouse/protocol"
+)
+
+type thermostatState struct {
+ settings Thermostat
+ queue <-chan []driver.Command
+ ledger *driver.Ledger
+ publisher *driver.Publisher
+ missed int
+}
+
+func (s *thermostatState) descriptor(ctx context.Context) error {
+ err := s.publisher.Descriptor(ctx, s.settings.DeviceID, protocol.Metadata{}, protocol.DescriptorPayload{
+ Manufacturer: "alamo",
+ Model: "slim-wifi",
+ Firmware: "unknown",
+ Capabilities: []protocol.Capability{{
+ ID: "control",
+ Type: "floor-heating",
+ ReportsState: true,
+ AcceptsCommands: true,
+ }},
+ })
+ if err != nil {
+ return fmt.Errorf("publish descriptor: %w", err)
+ }
+
+ return nil
+}
+
+func (s *thermostatState) record(ctx context.Context, state State, observedAt time.Time, causationID string) error {
+ s.missed = 0
+ metadata := protocol.Metadata{
+ SentAt: &observedAt,
+ CausationID: causationID,
+ }
+
+ err := s.publisher.Availability(ctx, s.settings.DeviceID, protocol.Metadata{SentAt: &observedAt}, protocol.AvailabilityPayload{Status: "online"})
+ if err != nil {
+ return fmt.Errorf("publish thermostat contact: %w", err)
+ }
+
+ value, err := json.Marshal(newReportedControl(state))
+ if err != nil {
+ return fmt.Errorf("encode thermostat feedback: %w", err)
+ }
+
+ err = s.publisher.Reported(ctx, s.settings.DeviceID, "control", metadata, protocol.ValuePayload{Value: value})
+ if err != nil {
+ return fmt.Errorf("publish thermostat feedback: %w", err)
+ }
+
+ return nil
+}
+
+func (s *thermostatState) miss(ctx context.Context) error {
+ s.missed = min(s.missed+1, 3)
+ if s.missed < 3 {
+ return nil
+ }
+
+ err := s.publisher.Availability(ctx, s.settings.DeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"})
+ if err != nil {
+ return fmt.Errorf("publish offline: %w", err)
+ }
+
+ return nil
+}
+
+func (s *thermostatState) result(ctx context.Context, command driver.Command, payload protocol.ResultPayload) error {
+ record, err := s.ledger.Remember(command.Metadata.MessageID, payload)
+ if err != nil {
+ return fmt.Errorf("remember thermostat result: %w", err)
+ }
+
+ err = s.publisher.Result(ctx, s.settings.DeviceID, command.Metadata.MessageID, record.Metadata, record.Payload)
+ if err != nil {
+ return fmt.Errorf("publish thermostat result: %w", err)
+ }
+
+ return nil
+}
projects / onehouse / commitdiff