update target temperature

kn/bridge.go

@@ -4,18 +4,22 @@ import (
 	"encoding/json"
 	"fmt"
 	"koolnova2mqtt/watcher"
+	"log"
 	"strconv"
 )
 
 type Publish func(topic string, qos byte, retained bool, payload string)
+type Subscribe func(topic string, callback func(message string)) error
 
 type Config struct {
-	ModuleName   string
-	SlaveID      byte
-	Publish      Publish
-	TopicPrefix  string
-	HassPrefix   string
-	ReadRegister watcher.ReadRegister
+	ModuleName    string
+	SlaveID       byte
+	Publish       Publish
+	Subscribe     Subscribe
+	TopicPrefix   string
+	HassPrefix    string
+	ReadRegister  watcher.ReadRegister
+	WriteRegister watcher.WriteRegister
 }
 
 type Bridge struct {
@@ -51,6 +55,7 @@ func NewBridge(config *Config) *Bridge {
 		RegisterSize: 2,
 		SlaveID:      config.SlaveID,
 		Read:         config.ReadRegister,
+		Write:        config.WriteRegister,
 	})
 
 	sysw := watcher.New(&watcher.Config{
@@ -59,6 +64,7 @@ func NewBridge(config *Config) *Bridge {
 		RegisterSize: 2,
 		SlaveID:      config.SlaveID,
 		Read:         config.ReadRegister,
+		Write:        config.WriteRegister,
 	})
 
 	b := &Bridge{
@@ -70,7 +76,7 @@ func NewBridge(config *Config) *Bridge {
 	return b
 }
 
-func (b *Bridge) Start() {
+func (b *Bridge) Start() error {
 	sys := NewSys(&SysConfig{
 		Watcher: b.sysw,
 	})
@@ -78,13 +84,13 @@ func (b *Bridge) Start() {
 	err := b.zw.Poll()
 	if err != nil {
 		fmt.Println(err)
-		return
+		return err
 	}
 
 	err = b.sysw.Poll()
 	if err != nil {
 		fmt.Println(err)
-		return
+		return err
 	}
 
 	getHVACMode := func() string {
@@ -113,13 +119,22 @@ func (b *Bridge) Start() {
 	}
 
 	zones, err := getActiveZones(b.zw)
+
+	publishHvacMode := func() {
+		for _, zone := range zones {
+			if zone.IsOn() {
+				hvacModeTopic := b.getZoneTopic(zone.ZoneNumber, "hvacMode")
+				mode := getHVACMode()
+				b.Publish(hvacModeTopic, 0, true, mode)
+			}
+		}
+	}
+
 	var hvacModes []string
 	for k, _ := range KnModes.GetForwardMap() {
 		hvacModes = append(hvacModes, k.(string))
 	}
 
-	hvacModeTopic := b.getSysTopic("hvacMode")
-	hvacModeTopicSet := hvacModeTopic + "/set"
 	holdModeTopic := b.getSysTopic("holdMode")
 	holdModeSetTopic := holdModeTopic + "/set"
 
@@ -130,6 +145,8 @@ func (b *Bridge) Start() {
 		targetTempSetTopic := targetTempTopic + "/set"
 		fanModeTopic := b.getZoneTopic(zone.ZoneNumber, "fanMode")
 		fanModeSetTopic := fanModeTopic + "/set"
+		hvacModeTopic := b.getZoneTopic(zone.ZoneNumber, "hvacMode")
+		hvacModeTopicSet := hvacModeTopic + "/set"
 
 		zone.OnCurrentTempChange = func(currentTemp float32) {
 			b.Publish(currentTempTopic, 0, true, fmt.Sprintf("%g", currentTemp))
@@ -144,6 +161,18 @@ func (b *Bridge) Start() {
 
 		}
 
+		err = b.Subscribe(targetTempSetTopic, func(message string) {
+			targetTemp, err := strconv.ParseFloat(message, 32)
+			if err != nil {
+				log.Printf("Error parsing targetTemperature in topic %s: %s", targetTempSetTopic, err)
+				return
+			}
+			err = zone.SetTargetTemperature(float32(targetTemp))
+			if err != nil {
+				log.Printf("Cannot set target temperature to %g in zone %d", targetTemp, zone.ZoneNumber)
+			}
+		})
+
 		name := fmt.Sprintf("%s_zone%d", b.ModuleName, zone.ZoneNumber)
 		config := map[string]interface{}{
 			"name":                      name,
@@ -191,10 +220,10 @@ func (b *Bridge) Start() {
 	sys.OnSystemEnabledChange = func() {
 		enabled := sys.GetSystemEnabled()
 		b.Publish(b.getSysTopic("enabled"), 0, true, fmt.Sprintf("%t", enabled))
-		b.Publish(hvacModeTopic, 0, true, getHVACMode())
+		publishHvacMode()
 	}
 	sys.OnKnModeChange = func() {
-		b.Publish(hvacModeTopic, 0, true, getHVACMode())
+		publishHvacMode()
 		b.Publish(holdModeTopic, 0, true, getHoldMode())
 	}
 
@@ -204,6 +233,7 @@ func (b *Bridge) Start() {
 	b.Publish(b.getSysTopic("serialBaud"), 0, true, strconv.Itoa(sys.GetBaudRate()))
 	b.Publish(b.getSysTopic("serialParity"), 0, true, sys.GetParity())
 	b.Publish(b.getSysTopic("slaveId"), 0, true, strconv.Itoa(sys.GetSlaveID()))
+	return nil
 }
 
 func (b *Bridge) Tick() {

kn/constants.go

@@ -56,8 +56,8 @@ var KnModes = bimap.New(map[interface{}]interface{}{
 	"underfloor air heating": MODE_UNDERFLOOR_AIR_HEATING,
 })
 
-const HOLD_MODE_UNDERFLOOR_ONLY = "underfloor only"
-const HOLD_MODE_FAN_ONLY = "fan only"
+const HOLD_MODE_UNDERFLOOR_ONLY = "underfloor"
+const HOLD_MODE_FAN_ONLY = "fan"
 const HOLD_MODE_UNDERFLOOR_AND_FAN = "underfloor and fan"
 
 const HVAC_MODE_OFF = "off"

kn/zone.go

@@ -6,6 +6,7 @@ import (
 
 type Watcher interface {
 	ReadRegister(address uint16) (value []byte)
+	WriteRegister(address uint16, value uint16) error
 	RegisterCallback(address uint16, callback func(address uint16))
 }
 
@@ -62,11 +63,14 @@ func (z *Zone) RegisterCallback(num int, f func()) {
 }
 
 func (z *Zone) ReadRegister(num int) uint16 {
-
 	b := z.Watcher.ReadRegister(uint16(z.ZoneNumber*REG_PER_ZONE + num))
 	return binary.BigEndian.Uint16(b)
 }
 
+func (z *Zone) WriteRegister(num int, value uint16) error {
+	return z.Watcher.WriteRegister(uint16(z.ZoneNumber*REG_PER_ZONE+num), value)
+}
+
 func (z *Zone) IsOn() bool {
 	r1 := z.ReadRegister(REG_ENABLED)
 	return r1&0x1 != 0
@@ -87,6 +91,10 @@ func (z *Zone) GetTargetTemperature() float32 {
 	return reg2temp(r3)
 }
 
+func (z *Zone) SetTargetTemperature(targetTemp float32) error {
+	return z.WriteRegister(REG_TARGET_TEMP, temp2reg(targetTemp))
+}
+
 func (z *Zone) GetFanMode() FanMode {
 	r2 := z.ReadRegister(REG_MODE)
 	return (FanMode)(r2&0x00F0) >> 4
@@ -100,3 +108,7 @@ func (z *Zone) GetHvacMode() KnMode {
 func reg2temp(r uint16) float32 {
 	return float32(0x00FF&r) / 2.0
 }
+
+func temp2reg(t float32) uint16 {
+	return uint16(t * 2)
+}

main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"crypto/tls"
+	"errors"
 	"flag"
 	"fmt"
 	"koolnova2mqtt/kn"
@@ -12,6 +13,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -19,14 +21,26 @@ import (
 	"github.com/goburrow/modbus"
 )
 
-func buildReader(handler *modbus.RTUClientHandler, client modbus.Client) watcher.ReadRegister {
+func buildReader(handler *modbus.RTUClientHandler, client modbus.Client, lock *sync.RWMutex) watcher.ReadRegister {
 	return func(slaveID byte, address uint16, quantity uint16) (results []byte, err error) {
+		lock.Lock()
+		defer lock.Unlock()
 		handler.SlaveId = slaveID
 		results, err = client.ReadHoldingRegisters(address-1, quantity)
 		return results, err
 	}
 }
 
+func buildWriter(handler *modbus.RTUClientHandler, client modbus.Client, lock *sync.RWMutex) watcher.WriteRegister {
+	return func(slaveID byte, address uint16, value uint16) (results []byte, err error) {
+		lock.Lock()
+		defer lock.Unlock()
+		handler.SlaveId = slaveID
+		results, err = client.WriteSingleRegister(address-1, value)
+		return results, err
+	}
+}
+
 func generateNodeName(slaveID string, port string) string {
 	reg, err := regexp.Compile("[^a-zA-Z0-9]+")
 	if err != nil {
@@ -74,17 +88,41 @@ func main() {
 
 	err := handler.Connect()
 	if err != nil {
-		fmt.Println(err)
+		log.Fatalf("Error connecting modbus: %s", err)
 	}
 	defer handler.Close()
 
 	modbusClient := modbus.NewClient(handler)
 
-	registerReader := buildReader(handler, modbusClient)
+	lock := &sync.RWMutex{}
+	registerReader := buildReader(handler, modbusClient, lock)
+	registerWriter := buildWriter(handler, modbusClient, lock)
 
 	var mqttClient MQTT.Client
 	publish := func(topic string, qos byte, retained bool, payload string) {
-		mqttClient.Publish(topic, qos, retained, payload)
+		client := mqttClient
+		if client == nil {
+			log.Printf("Cannot publish message %q to topic %s. MQTT client is disconnected", payload, topic)
+			return
+		}
+		client.Publish(topic, qos, retained, payload)
+	}
+
+	subscribe := func(topic string, callback func(message string)) error {
+		client := mqttClient
+		if client == nil {
+			log.Printf("Cannot subscribe to topic %s. MQTT client is disconnected", topic)
+			return errors.New("Client is disconnected")
+		}
+		token := client.Subscribe(topic, 0, func(c MQTT.Client, m MQTT.Message) {
+			cbclient := mqttClient
+			if cbclient != client {
+				log.Printf("Cannot invoke callback to topic %s. MQTT client is disconnected", topic)
+			}
+			callback(string(m.Payload()))
+		})
+		token.Wait()
+		return token.Error()
 	}
 
 	var snameList []string
@@ -109,12 +147,14 @@ func main() {
 			log.Fatalf("Error parsing slaveID list")
 		}
 		bridge := kn.NewBridge(&kn.Config{
-			ModuleName:   slaveName,
-			SlaveID:      byte(slaveID),
-			Publish:      publish,
-			TopicPrefix:  *prefix,
-			HassPrefix:   *hassPrefix,
-			ReadRegister: registerReader,
+			ModuleName:    slaveName,
+			SlaveID:       byte(slaveID),
+			Publish:       publish,
+			Subscribe:     subscribe,
+			TopicPrefix:   *prefix,
+			HassPrefix:    *hassPrefix,
+			ReadRegister:  registerReader,
+			WriteRegister: registerWriter,
 		})
 		bridges = append(bridges, bridge)
 	}
@@ -128,26 +168,55 @@ func main() {
 	}
 	tlsConfig := &tls.Config{InsecureSkipVerify: true, ClientAuth: tls.NoClientCert}
 	connOpts.SetTLSConfig(tlsConfig)
+	onConnect := false
 	connOpts.OnConnect = func(c MQTT.Client) {
-		for _, b := range bridges {
-			b.Start()
-		}
+		onConnect = true
+	}
+	connOpts.OnConnectionLost = func(c MQTT.Client, err error) {
+		log.Printf("Connection to MQTT server lost: %s\n", err)
+		mqttClient = nil
 	}
 
-	mqttClient = MQTT.NewClient(connOpts)
+	connectMQTT := func() error {
+		mqttClient = MQTT.NewClient(connOpts)
 
-	if token := mqttClient.Connect(); token.Wait() && token.Error() != nil {
-		panic(token.Error())
-	} else {
-		fmt.Printf("Connected to %s\n", *server)
+		if token := mqttClient.Connect(); token.Wait() && token.Error() != nil {
+			mqttClient = nil
+			return token.Error()
+		} else {
+			log.Printf("Connected to %s\n", *server)
+		}
+		return nil
 	}
 
 	ticker := time.NewTicker(time.Second)
 
 	go func() {
 		for range ticker.C {
-			for _, b := range bridges {
-				b.Tick()
+			if mqttClient == nil {
+				err := connectMQTT()
+				if err != nil {
+					log.Printf("Error connecting to MQTT server: %s\n", err)
+					continue
+				}
+			}
+			client := mqttClient
+			if client != nil && client.IsConnected() {
+				if onConnect {
+					onConnect = false
+					for _, b := range bridges {
+						err := b.Start()
+						if err != nil {
+							log.Printf("Error starting bridge: %s\n", err)
+							client.Disconnect(100)
+							mqttClient = nil
+						}
+					}
+				} else {
+					for _, b := range bridges {
+						b.Tick()
+					}
+				}
 			}
 		}
 	}()

watcher/watcher.go

@@ -3,15 +3,18 @@ package watcher
 import (
 	"bytes"
 	"errors"
+	"sync"
 )
 
 type ReadRegister func(slaveID byte, address uint16, quantity uint16) (results []byte, err error)
+type WriteRegister func(slaveID byte, address uint16, value uint16) (results []byte, err error)
 
 type Config struct {
 	Address      uint16
 	Quantity     uint16
 	SlaveID      byte
 	Read         ReadRegister
+	Write        WriteRegister
 	RegisterSize int
 }
 
@@ -19,6 +22,7 @@ type Watcher struct {
 	Config
 	state     []byte
 	callbacks map[uint16]func(address uint16)
+	lock      *sync.RWMutex
 }
 
 var ErrIncorrectRegisterSize = errors.New("Incorrect register size")
@@ -29,6 +33,7 @@ func New(config *Config) *Watcher {
 	return &Watcher{
 		Config:    *config,
 		callbacks: make(map[uint16]func(address uint16)),
+		lock:      &sync.RWMutex{},
 	}
 }
 
@@ -37,24 +42,27 @@ func (w *Watcher) RegisterCallback(address uint16, callback func(address uint16)
 }
 
 func (w *Watcher) Poll() error {
+	w.lock.Lock()
 	newState, err := w.Read(w.SlaveID, w.Address, w.Quantity)
 	if err != nil {
+		w.lock.Unlock()
 		return err
 	}
 
 	if len(newState) != int(w.Quantity)*w.RegisterSize {
+		w.lock.Unlock()
 		return ErrIncorrectRegisterSize
 	}
 
 	oldState := w.state
 	w.state = newState
+	var callbackAddresses []uint16
 
 	first := len(oldState) != len(newState)
-	address := w.Address
 	for n := 0; n < len(newState); n += w.RegisterSize {
+		address := uint16(n/w.RegisterSize) + w.Address
 		callback := w.callbacks[address]
 		if callback == nil {
-			address++
 			continue
 		}
 		var oldValue []byte
@@ -65,14 +73,20 @@ func (w *Watcher) Poll() error {
 			oldValue = oldState[n : n+w.RegisterSize]
 		}
 		if bytes.Compare(oldValue, newValue) != 0 {
-			callback(address)
+			callbackAddresses = append(callbackAddresses, address)
 		}
-		address++
+	}
+	w.lock.Unlock()
+	for _, address := range callbackAddresses {
+		callback := w.callbacks[address]
+		callback(address)
 	}
 	return nil
 }
 
 func (w *Watcher) ReadRegister(address uint16) (value []byte) {
+	w.lock.Lock()
+	defer w.lock.Unlock()
 	if address < w.Address || address > w.Address+uint16(w.Quantity) {
 		panic(ErrAddressOutOfRange)
 	}
@@ -84,6 +98,23 @@ func (w *Watcher) ReadRegister(address uint16) (value []byte) {
 
 }
 
+func (w *Watcher) WriteRegister(address uint16, value uint16) error {
+	w.lock.Lock()
+	results, err := w.Write(w.SlaveID, address, value)
+	if err != nil {
+		w.lock.Unlock()
+		return err
+	}
+	registerOffset := int(address-w.Address) * w.RegisterSize
+	copy(w.state[registerOffset:registerOffset+w.RegisterSize], results)
+	callback := w.callbacks[address]
+	w.lock.Unlock()
+	if callback != nil {
+		callback(address)
+	}
+	return nil
+}
+
 func (w *Watcher) TriggerCallbacks() {
 	for address, callback := range w.callbacks {
 		callback(address)