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