git.giacomolli.uk - onehouse/blob - driver/m31/command_handler.go

   1 package m31
   2
   3 import (
   4         "cmp"
   5         "context"
   6         "errors"
   7         "fmt"
   8         "slices"
   9         "time"
  10
  11         "onehouse/driver"
  12         "onehouse/protocol"
  13 )
  14
  15 type commandHandler struct {
  16         connection *connection
  17 }
  18
  19 func (h *commandHandler) handle(ctx context.Context, commands []driver.Command) error {
  20         pending, err := h.prepare(ctx, commands)
  21         if err != nil {
  22                 return fmt.Errorf("prepare output commands: %w", err)
  23         }
  24
  25         if len(pending) == 0 {
  26                 return nil
  27         }
  28
  29         if len(commands) == 1 && (pending[0].analog || pending[0].point.pulse != 0) {
  30                 err = h.special(ctx, pending[0])
  31                 if err != nil {
  32                         return fmt.Errorf("apply special output: %w", err)
  33                 }
  34
  35                 return nil
  36         }
  37
  38         slices.SortFunc(pending, func(first outputRequest, second outputRequest) int {
  39                 return cmp.Compare(first.point.address, second.point.address)
  40         })
  41
  42         active := make([]outputRequest, 0, len(pending))
  43
  44         for start := 0; start < len(pending); {
  45                 if pending[start].command.Expired(time.Now()) {
  46                         err = h.result(ctx, pending[start].command, protocol.ResultPayload{
  47                                 Status: "rejected",
  48                                 Reason: "Command expired",
  49                         })
  50                         if err != nil {
  51                                 return fmt.Errorf("reject expired write: %w", err)
  52                         }
  53
  54                         start++
  55
  56                         continue
  57                 }
  58
  59                 end := start + 1
  60
  61                 for end < len(pending) && pending[end].point.address == pending[end-1].point.address+1 && !pending[end].command.Expired(time.Now()) {
  62                         end++
  63                 }
  64
  65                 err = h.writeRange(ctx, pending[start:end])
  66                 if err != nil {
  67                         return fmt.Errorf("fail coil writes: %w", h.fail(ctx, pending, err))
  68                 }
  69
  70                 active = append(active, pending[start:end]...)
  71                 start = end
  72         }
  73
  74         if len(active) == 0 {
  75                 return nil
  76         }
  77
  78         next := h.connection.state
  79         next.outputs, err = h.connection.owner.client.ReadCoils(ctx, uint16(len(h.connection.owner.topology.outputs)))
  80         if err != nil {
  81                 return fmt.Errorf("fail coil readback: %w", h.fail(ctx, active, fmt.Errorf("read command coils: %w", err)))
  82         }
  83
  84         if len(next.outputs) != len(h.connection.owner.topology.outputs) {
  85                 return fmt.Errorf("fail incomplete readback: %w", h.fail(ctx, active, fmt.Errorf("incomplete coil readback")))
  86         }
  87
  88         next.observedAt = time.Now().UTC()
  89
  90         err = h.complete(ctx, active, next)
  91         if err != nil {
  92                 return fmt.Errorf("complete coil commands: %w", err)
  93         }
  94
  95         return nil
  96 }
  97
  98 func (h *commandHandler) prepare(ctx context.Context, commands []driver.Command) ([]outputRequest, error) {
  99         var pending []outputRequest
 100
 101         for _, command := range commands {
 102                 record, exists := h.connection.owner.ledger.Lookup(command.Metadata.MessageID)
 103                 if exists {
 104                         err := h.send(ctx, command.Metadata.MessageID, record)
 105                         if err != nil {
 106                                 return nil, fmt.Errorf("replay result: %w", err)
 107                         }
 108
 109                         continue
 110                 }
 111
 112                 request, reason := newOutputRequest(command, h.connection.owner.topology, h.connection.owner.deviceID, len(commands) > 1)
 113                 if reason != "" {
 114                         err := h.result(ctx, command, protocol.ResultPayload{
 115                                 Status: "rejected",
 116                                 Reason: reason,
 117                         })
 118                         if err != nil {
 119                                 return nil, fmt.Errorf("reject command: %w", err)
 120                         }
 121
 122                         continue
 123                 }
 124
 125                 pending = append(pending, request)
 126         }
 127
 128         return pending, nil
 129 }
 130
 131 func (h *commandHandler) send(ctx context.Context, commandID string, record driver.ResultRecord) error {
 132         err := h.connection.owner.publisher.Result(ctx, h.connection.owner.deviceID, commandID, record.Metadata, record.Payload)
 133         if err != nil {
 134                 return fmt.Errorf("publish result: %w", err)
 135         }
 136
 137         return nil
 138 }
 139
 140 func (h *commandHandler) result(ctx context.Context, command driver.Command, payload protocol.ResultPayload) error {
 141         record, err := h.connection.owner.ledger.Remember(command.Metadata.MessageID, payload)
 142         if err != nil {
 143                 return fmt.Errorf("remember command result: %w", err)
 144         }
 145
 146         err = h.send(ctx, command.Metadata.MessageID, record)
 147         if err != nil {
 148                 return fmt.Errorf("send command result: %w", err)
 149         }
 150
 151         return nil
 152 }
 153
 154 func (h *commandHandler) writeRange(ctx context.Context, requests []outputRequest) error {
 155         expires := *requests[0].command.Metadata.ExpiresAt
 156         values := make([]bool, len(requests))
 157
 158         for index, request := range requests {
 159                 values[index] = request.value == 1
 160                 if request.command.Metadata.ExpiresAt.Before(expires) {
 161                         expires = *request.command.Metadata.ExpiresAt
 162                 }
 163         }
 164
 165         ctx, cancel := context.WithDeadline(ctx, expires)
 166         defer cancel()
 167
 168         var err error
 169
 170         if len(requests) == 1 {
 171                 err = h.connection.owner.client.WriteCoil(ctx, requests[0].point.address, values[0])
 172         } else {
 173                 err = h.connection.owner.client.WriteCoils(ctx, requests[0].point.address, values)
 174         }
 175
 176         if err != nil {
 177                 return fmt.Errorf("write output coils: %w", err)
 178         }
 179
 180         return nil
 181 }
 182
 183 func (h *commandHandler) fail(ctx context.Context, requests []outputRequest, cause error) error {
 184         records := make([]driver.ResultRecord, len(requests))
 185
 186         for index, request := range requests {
 187                 record, err := h.connection.owner.ledger.Remember(request.command.Metadata.MessageID, protocol.ResultPayload{
 188                         Status: "failed",
 189                         Reason: "Modbus operation failed",
 190                 })
 191                 if err != nil {
 192                         return fmt.Errorf("remember failed commands: %w", errors.Join(cause, err))
 193                 }
 194
 195                 records[index] = record
 196         }
 197
 198         for index, request := range requests {
 199                 err := h.send(ctx, request.command.Metadata.MessageID, records[index])
 200                 if err != nil {
 201                         return fmt.Errorf("send failed commands: %w", errors.Join(cause, err))
 202                 }
 203         }
 204
 205         return &deviceError{err: cause}
 206 }
 207
 208 func (h *commandHandler) complete(ctx context.Context, requests []outputRequest, next snapshot) error {
 209         records := make([]driver.ResultRecord, len(requests))
 210         causes := make(map[string]string, len(requests))
 211
 212         for index, request := range requests {
 213                 actual := uint16(0)
 214                 if request.analog {
 215                         actual = next.analog[request.point.address]
 216                 } else if next.outputs[request.point.address] {
 217                         actual = 1
 218                 }
 219
 220                 payload := protocol.ResultPayload{Status: "completed"}
 221                 if actual != request.value {
 222                         payload.Status, payload.Reason = "failed", "Output readback differs from command"
 223                 }
 224
 225                 record, err := h.connection.owner.ledger.Remember(request.command.Metadata.MessageID, payload)
 226                 if err != nil {
 227                         return fmt.Errorf("remember output results: %w", err)
 228                 }
 229
 230                 records[index] = record
 231                 causes[request.point.id] = request.command.Metadata.MessageID
 232         }
 233
 234         err := h.connection.publish(ctx, next, false, causes)
 235         if err != nil {
 236                 return fmt.Errorf("publish output feedback: %w", err)
 237         }
 238
 239         for index, request := range requests {
 240                 err = h.send(ctx, request.command.Metadata.MessageID, records[index])
 241                 if err != nil {
 242                         return fmt.Errorf("publish completed results: %w", err)
 243                 }
 244         }
 245
 246         return nil
 247 }
 248
 249 func (h *commandHandler) special(ctx context.Context, request outputRequest) error {
 250         next, err := h.applySpecial(ctx, request)
 251         if err != nil {
 252                 return fmt.Errorf("fail special command: %w", h.fail(ctx, []outputRequest{request}, err))
 253         }
 254
 255         if !request.analog {
 256                 err = h.result(ctx, request.command, protocol.ResultPayload{Status: "completed"})
 257                 if err != nil {
 258                         return fmt.Errorf("complete pulse: %w", err)
 259                 }
 260
 261                 return nil
 262         }
 263
 264         err = h.complete(ctx, []outputRequest{request}, next)
 265         if err != nil {
 266                 return fmt.Errorf("complete voltage command: %w", err)
 267         }
 268
 269         return nil
 270 }
 271
 272 func (h *commandHandler) applySpecial(ctx context.Context, request outputRequest) (snapshot, error) {
 273         ctx, cancel := context.WithDeadline(ctx, *request.command.Metadata.ExpiresAt)
 274         defer cancel()
 275
 276         if !request.analog {
 277                 err := h.connection.owner.client.WriteCoil(ctx, request.point.address, true)
 278                 if err != nil {
 279                         return snapshot{}, fmt.Errorf("write pulse: %w", err)
 280                 }
 281
 282                 return h.connection.state, nil
 283         }
 284
 285         err := h.connection.owner.client.WriteRegister(ctx, request.point.address, request.value)
 286         if err != nil {
 287                 return snapshot{}, fmt.Errorf("write voltage: %w", err)
 288         }
 289
 290         next := h.connection.state
 291         next.analog, err = h.connection.owner.client.ReadRegisters(ctx, 0, uint16(len(h.connection.owner.topology.analog)))
 292         if err != nil {
 293                 return snapshot{}, fmt.Errorf("read voltage feedback: %w", err)
 294         }
 295
 296         if len(next.analog) != len(h.connection.owner.topology.analog) {
 297                 return snapshot{}, fmt.Errorf("incomplete voltage feedback")
 298         }
 299
 300         err = validateVoltage(next.analog)
 301         if err != nil {
 302                 return snapshot{}, fmt.Errorf("validate voltage feedback: %w", err)
 303         }
 304
 305         next.observedAt = time.Now().UTC()
 306
 307         return next, nil
 308 }
 309
 310 func (h *commandHandler) unavailable(ctx context.Context, commands []driver.Command) error {
 311         pending, err := h.prepare(ctx, commands)
 312         if err != nil {
 313                 return fmt.Errorf("prepare unavailable commands: %w", err)
 314         }
 315
 316         for _, request := range pending {
 317                 err = h.result(ctx, request.command, protocol.ResultPayload{
 318                         Status: "failed",
 319                         Reason: "Modbus device is unavailable",
 320                 })
 321                 if err != nil {
 322                         return fmt.Errorf("publish unavailable result: %w", err)
 323                 }
 324         }
 325
 326         return nil
 327 }