koolnova2mqtt/watcher/watcher.go

// Package watcher represents a cache of a range of modbus registers in a device
// Can fire events if a watched register changes
package watcher

import (
	"errors"
	"sync"
)

type Modbus interface {
	ReadRegister(slaveID byte, address uint16, quantity uint16) (results []uint16, err error)
	WriteRegister(slaveID byte, address uint16, value uint16) (results []uint16, err error)
	Close() error
}

// Config contains the configuration parameters for a new Watcher instance
type Config struct {
	Address  uint16 // Start address
	Quantity uint16 // Number of registers to watch
	SlaveID  byte   // SlaveID to watch
	Modbus   Modbus // Modbus interface
}

// Watcher represents a cache of modbus registers in a device
type Watcher struct {
	Config
	state     []uint16                        // current view of the modbus register states
	callbacks map[uint16]func(address uint16) // set of callbacks
	lock      *sync.RWMutex
}

var ErrAddressOutOfRange = errors.New("Register address out of range")
var ErrUninitialized = errors.New("State uninitialized. Call Poll() first.")
var ErrCannotIncreaseRange = errors.New("Cannot increase range")

// New returns a new Watcher instance
func New(config *Config) *Watcher {
	return &Watcher{
		Config:    *config,
		callbacks: make(map[uint16]func(address uint16)),
		lock:      &sync.RWMutex{},
	}
}

// RegisterCallback registers a new callback that will be fired when the specific register address changes values
func (w *Watcher) RegisterCallback(address uint16, callback func(address uint16)) {
	w.callbacks[address] = callback
}

// Poll refreshes the cache by reading the watched register range from the slave device
func (w *Watcher) Poll() error {
	w.lock.Lock()
	newState, err := w.Modbus.ReadRegister(w.SlaveID, w.Address, w.Quantity)
	if err != nil {
		w.lock.Unlock()
		return err
	}

	oldState := w.state
	w.state = newState
	var callbackAddresses []uint16

	first := len(oldState) != len(newState)
	for n := 0; n < len(newState); n++ {
		address := uint16(n) + w.Address
		callback := w.callbacks[address]
		if callback == nil {
			continue
		}
		if first || oldState[n] != newState[n] {
			callbackAddresses = append(callbackAddresses, address)
		}
	}
	w.lock.Unlock()
	for _, address := range callbackAddresses {
		callback := w.callbacks[address]
		callback(address)
	}
	return nil
}

// ReadRegister reads one register from the cache
func (w *Watcher) ReadRegister(address uint16) (value uint16) {
	w.lock.Lock()
	defer w.lock.Unlock()
	if address < w.Address || address > w.Address+uint16(w.Quantity) {
		panic(ErrAddressOutOfRange)
	}
	if w.state == nil {
		panic(ErrUninitialized)
	}
	return w.state[int(address-w.Address)]

}

// WriteRegister writes the value to the slave device and updates the cache if successful
func (w *Watcher) WriteRegister(address uint16, value uint16) error {
	w.lock.Lock()
	results, err := w.Modbus.WriteRegister(w.SlaveID, address, value)
	if err != nil {
		w.lock.Unlock()
		return err
	}
	w.state[int(address-w.Address)] = results[0]
	callback := w.callbacks[address]
	w.lock.Unlock()
	if callback != nil {
		callback(address)
	}
	return nil
}

// TriggerCallbacks calls all callbacks
func (w *Watcher) TriggerCallbacks() {
	for address, callback := range w.callbacks {
		callback(address)
	}
}

// Resize reduces the watched range
func (w *Watcher) Resize(newQuantity int) {
	w.lock.Lock()
	defer w.lock.Unlock()
	if newQuantity <= int(w.Quantity) {
		w.state = w.state[:newQuantity]
		for address := range w.callbacks {
			if address > w.Address+uint16(newQuantity)-1 {
				delete(w.callbacks, address)
			}
		}
	} else {
		panic(ErrCannotIncreaseRange)
	}
	w.Quantity = uint16(newQuantity)
}