MQTT Discovery, filtering, custom component

.gitignore

@@ -1,6 +1,5 @@
-/sofar
-sofar_g3_lsw3_logger_reader
 config.yaml
-sofar
+sofar-x86
 sofar-arm
-.idea
+.vscode
+.idea

Makefile

@@ -1,5 +1,7 @@
+all: build-arm build-x86
+
 build-arm:
-	env GOOS=linux GOARCH=arm GOARM=5 go build -o sofar-arm
+	env GOOS=linux GOARCH=arm GOARM=5 go build -o custom_components/sofar_g3_lsw3_logger_reader/sofar-arm
 
-build:
-	go build -o sofar
+build-x86:
+	go build -o custom_components/sofar_g3_lsw3_logger_reader/sofar-x86

adapters/comms/tcpip/tcpip.go

@@ -1,7 +1,6 @@
 package tcpip
 
 import (
-	"bufio"
 	"fmt"
 	"net"
 	"time"
@@ -51,13 +50,11 @@ func (s *tcpIpPort) Read(buf []byte) (int, error) {
 		return 0, fmt.Errorf("connection is not open")
 	}
 
-	reader := bufio.NewReader(s.conn)
-
 	if err := s.conn.SetReadDeadline(time.Now().Add(timeout)); err != nil {
 		return 0, err
 	}
 
-	return reader.Read(buf)
+	return s.conn.Read(buf)
 }
 
 func (s *tcpIpPort) Write(payload []byte) (int, error) {

adapters/devices/sofar/device.go

@@ -1,21 +1,71 @@
 package sofar
 
-import "github.com/kubaceg/sofar_g3_lsw3_logger_reader/ports"
+import (
+	"log"
+	"regexp"
+
+	"github.com/kubaceg/sofar_g3_lsw3_logger_reader/ports"
+)
 
 type Logger struct {
-	serialNumber uint
-	connPort     ports.CommunicationPort
+	serialNumber  uint
+	connPort      ports.CommunicationPort
+	attrWhiteList map[string]struct{}
+	attrBlackList []*regexp.Regexp
+}
+
+// for a set in go we use a map of keys -> empty struct
+func toSet(slice []string) map[string]struct{} {
+	set := make(map[string]struct{}, len(slice))
+	v := struct{}{}
+	for _, s := range slice {
+		set[s] = v
+	}
+	return set
+}
+
+func toREs(patterns []string) []*regexp.Regexp {
+	res := make([]*regexp.Regexp, 0, len(patterns))
+	for idx, p := range patterns {
+		re, err := regexp.Compile(p)
+		if err == nil {
+			res = append(res, re)
+		} else {
+			log.Printf("config attrBlackList item %d '%s' not a valid regexp; %v", idx, p, err)
+		}
+	}
+	return res
 }
 
-func NewSofarLogger(serialNumber uint, connPort ports.CommunicationPort) *Logger {
+func NewSofarLogger(serialNumber uint, connPort ports.CommunicationPort, attrWhiteList []string, attrBlackList []string) *Logger {
 	return &Logger{
-		serialNumber: serialNumber,
-		connPort:     connPort,
+		serialNumber:  serialNumber,
+		connPort:      connPort,
+		attrWhiteList: toSet(attrWhiteList),
+		attrBlackList: toREs(attrBlackList),
+	}
+}
+
+func (s *Logger) nameFilter(k string) bool {
+	if len(s.attrWhiteList) > 0 {
+		_, ok := s.attrWhiteList[k]
+		return ok
+	} else {
+		for _, re := range s.attrBlackList {
+			if re.MatchString(k) {
+				return false
+			}
+		}
 	}
+	return true
+}
+
+func (s *Logger) GetDiscoveryFields() []ports.DiscoveryField {
+	return getDiscoveryFields(s.nameFilter)
 }
 
 func (s *Logger) Query() (map[string]interface{}, error) {
-	return readData(s.connPort, s.serialNumber)
+	return readData(s.connPort, s.serialNumber, s.nameFilter)
 }
 
 func (s *Logger) Name() string {

adapters/devices/sofar/lsw.go

@@ -73,17 +73,18 @@ func (l LSWRequest) checksum(buf []byte) uint8 {
 	return checksum
 }
 
-func readData(connPort ports.CommunicationPort, serialNumber uint) (map[string]interface{}, error) {
+func readData(connPort ports.CommunicationPort, serialNumber uint, nameFilter func(string) bool) (map[string]interface{}, error) {
 	result := make(map[string]interface{})
 
 	for _, rr := range allRegisterRanges {
 		reply, err := readRegisterRange(rr, connPort, serialNumber)
 		if err != nil {
 			return nil, err
 		}
-
 		for k, v := range reply {
-			result[k] = v
+			if nameFilter(k) {
+				result[k] = v
+			}
 		}
 	}
 	return result, nil
@@ -111,27 +112,25 @@ func readRegisterRange(rr registerRange, connPort ports.CommunicationPort, seria
 		return nil, err
 	}
 
-	// read the result
-	buf := make([]byte, 2048)
-	n, err := connPort.Read(buf)
-	if err != nil {
-		return nil, err
-	}
-
-	// truncate the buffer
-	buf = buf[:n]
-	if len(buf) < 48 {
-		// short reply
-		return nil, fmt.Errorf("short reply: %d bytes", n)
+	// read enough bytes
+	buf := []byte{}
+	for {
+		b := make([]byte, 2048)
+		n, err := connPort.Read(b)
+		if n > 0 {
+			buf = append(buf, b[:n]...)
+		}
+		if err != nil {
+			return nil, err
+		}
+		if len(buf) >= 28 && len(buf) >= 28+int(buf[27]) {
+			break
+		}
 	}
-
-	replyBytesCount := buf[27]
-
-	modbusReply := buf[28 : 28+replyBytesCount]
+	modbusReply := buf[28 : 28+buf[27]]
 
 	// shove the data into the reply
 	reply := make(map[string]interface{})
-
 	for _, f := range rr.replyFields {
 		fieldOffset := (f.register - rr.start) * 2
 

adapters/devices/sofar/sofar_protocol.go

@@ -1,5 +1,9 @@
 package sofar
 
+import (
+	"github.com/kubaceg/sofar_g3_lsw3_logger_reader/ports"
+)
+
 type field struct {
 	register  int
 	name      string
@@ -37,6 +41,18 @@ func GetAllRegisterNames() []string {
 	return result
 }
 
+func getDiscoveryFields(nameFilter func(string) bool) []ports.DiscoveryField {
+	result := make([]ports.DiscoveryField, 0)
+	for _, rr := range allRegisterRanges {
+		for _, f := range rr.replyFields {
+			if f.name != "" && f.valueType != "" && nameFilter(f.name) {
+				result = append(result, ports.DiscoveryField{Name: f.name, Factor: f.factor, Unit: f.unit})
+			}
+		}
+	}
+	return result
+}
+
 var rrSystemInfo = registerRange{
 	start: 0x400,
 	end:   0x43a,
@@ -75,18 +91,18 @@ var rrSystemInfo = registerRange{
 		{0x0423, "Temp_Rsvd1", "I16", "1", "℃"},
 		{0x0424, "Temp_Rsvd2", "I16", "1", "℃"},
 		{0x0425, "Temp_Rsvd3", "I16", "1", "℃"},
-		{0x0426, "GenerationTime_Today", "U16", "1", "Minute"},
-		{0x0427, "GenerationTime_Total", "U32", "1", "Minute"},
+		{0x0426, "GenerationTime_Today", "U16", "1", "min"}, // HA uses d, h, min, s not Minute
+		{0x0427, "GenerationTime_Total", "U32", "1", "min"},
 		{0x0428, "", "", "", ""},
-		{0x0429, "ServiceTime_Total", "U32", "1", "Minute"},
+		{0x0429, "ServiceTime_Total", "U32", "1", "min"},
 		{0x042A, "", "", "", ""},
 		{0x042B, "InsulationResistance", "U16", "1", "kΩ"},
 		{0x042C, "SysTime_Year", "U16", "", ""},
 		{0x042D, "SysTime_Month", "U16", "", ""},
-		{0x042E, "SysTime_Date", "U16", "", ""},
-		{0x042F, "SysTime_Hour", "U16", "", ""},
-		{0x0430, "SysTime_Minute", "U16", "", ""},
-		{0x0431, "SysTime_Second", "U16", "", ""},
+		{0x042E, "SysTime_Date", "U16", "1", "d"},
+		{0x042F, "SysTime_Hour", "U16", "1", "h"},
+		{0x0430, "SysTime_Minute", "U16", "1", "min"},
+		{0x0431, "SysTime_Second", "U16", "1", "s"},
 		{0x0432, "Fault19", "U16", "", ""},
 		{0x0433, "Fault20", "U16", "", ""},
 		{0x0434, "Fault21", "U16", "", ""},
@@ -104,11 +120,11 @@ var rrEnergyTodayTotals = registerRange{
 	replyFields: []field{
 		{0x684, "PV_Generation_Today", "U32", "0.01", "kWh"},
 		{0x686, "PV_Generation_Total", "U32", "0.1", "kWh"},
-		{0x688, "Load_Consumption_Today", "U32", "0.1", "kWh"},
+		{0x688, "Load_Consumption_Today", "U32", "0.01", "kWh"},
 		{0x68A, "Load_Consumption_Total", "U32", "0.1", "kWh"},
-		{0x68C, "Energy_Purchase_Today", "U32", "0.1", "kWh"},
+		{0x68C, "Energy_Purchase_Today", "U32", "0.01", "kWh"},
 		{0x68E, "Energy_Purchase_Total", "U32", "0.1", "kWh"},
-		{0x690, "Energy_Selling_Today", "U32", "0.1", "kWh"},
+		{0x690, "Energy_Selling_Today", "U32", "0.01", "kWh"},
 		{0x692, "Energy_Selling_Total", "U32", "0.1", "kWh"},
 		{0x694, "Bat_Charge_Today", "U32", "0.01", "kWh"},
 		{0x696, "Bat_Charge_Total", "U32", "0.1", "kWh"},

adapters/export/mosquitto/mqtt.go

@@ -4,21 +4,24 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"strings"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/kubaceg/sofar_g3_lsw3_logger_reader/ports"
 )
 
 type MqttConfig struct {
-	Url      string `yaml:"url"`
-	User     string `yaml:"user"`
-	Password string `yaml:"password"`
-	Prefix   string `yaml:"prefix"`
+	Url       string `yaml:"url"`
+	User      string `yaml:"user"`
+	Password  string `yaml:"password"`
+	Discovery string `yaml:"discovery"`
+	State     string `yaml:"state"`
 }
 
 type Connection struct {
 	client mqtt.Client
-	prefix string
+	state  string
 }
 
 var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
@@ -46,7 +49,7 @@ func New(config *MqttConfig) (*Connection, error) {
 
 	conn := &Connection{}
 	conn.client = mqtt.NewClient(opts)
-	conn.prefix = config.Prefix
+	conn.state = config.State
 	if token := conn.client.Connect(); token.Wait() && token.Error() != nil {
 		return nil, token.Error()
 	}
@@ -55,27 +58,77 @@ func New(config *MqttConfig) (*Connection, error) {
 
 }
 
-func (conn *Connection) InsertRecord(measurement map[string]interface{}) error {
-	measurementCopy := make(map[string]interface{}, len(measurement))
-	for k, v := range measurement {
-		measurementCopy[k] = v
+func (conn *Connection) publish(topic string, msg string, retain bool) {
+	token := conn.client.Publish(topic, 0, retain, msg)
+	res := token.WaitTimeout(1 * time.Second)
+	if !res || token.Error() != nil {
+		log.Printf("error inserting to MQTT: %s", token.Error())
 	}
-	go func(measurement map[string]interface{}) {
-		// timestamp it
-		measurement["LastTimestamp"] = time.Now().UnixNano() / int64(time.Millisecond)
-		m, _ := json.Marshal(measurement)
-		measurement["All"] = string(m)
-
-		for k, v := range measurement {
-			token := conn.client.Publish(fmt.Sprintf("%s/%s", conn.prefix, k), 0, true, fmt.Sprintf("%v", v))
-			res := token.WaitTimeout(1 * time.Second)
-			if !res || token.Error() != nil {
-				log.Printf("error inserting to MQTT: %s", token.Error())
-			}
-		}
-
-	}(measurementCopy)
+}
+
+// return "power" for kW etc., "energy" for kWh etc.
+func unit2DeviceClass(unit string) string {
+	if strings.HasSuffix(unit, "Wh") {
+		return "energy"
+	} else if strings.HasSuffix(unit, "W") {
+		return "power"
+	} else if strings.HasSuffix(unit, "Hz") {
+		return "frequency"
+	} else if strings.HasSuffix(unit, "VA") {
+		return "apparent_power"
+	} else if strings.HasSuffix(unit, "VAR") {
+		return "reactive_power"
+	} else if strings.HasSuffix(unit, "V") {
+		return "voltage"
+	} else if strings.HasSuffix(unit, "A") {
+		return "current"
+	} else if strings.HasSuffix(unit, "Ω") {
+		return "voltage" // resistance not valid in https://developers.home-assistant.io/docs/core/entity/sensor/ so use "voltage"
+	} else if strings.HasSuffix(unit, "℃") {
+		return "temperature"
+	} else if strings.HasSuffix(unit, "min") {
+		return "duration"
+	} else {
+		return ""
+	}
+}
+
+func unit2StateClass(unit string) string {
+	if strings.HasSuffix(unit, "Wh") {
+		return "total"
+	} else {
+		return "measurement"
+	}
+}
+
+// MQTT Discovery: https://www.home-assistant.io/integrations/mqtt/#mqtt-discovery
+func (conn *Connection) InsertDiscoveryRecord(discovery string, state string, expireAfter int, fields []ports.DiscoveryField) error {
+	uniq := "01ad" // TODO: get from config?
+	for _, f := range fields {
+		topic := fmt.Sprintf("%s/%s/config", discovery, f.Name)
+		json, _ := json.Marshal(map[string]interface{}{
+			"name":                  f.Name,
+			"unique_id":             fmt.Sprintf("%s_%s", f.Name, uniq),
+			"device_class":          unit2DeviceClass(f.Unit),
+			"state_class":           unit2StateClass(f.Unit),
+			"state_topic":           state,
+			"unit_of_measurement":   f.Unit,
+			"value_template":        fmt.Sprintf("{{ value_json.%s|int * %s }}", f.Name, f.Factor),
+			"availability_topic":    state,
+			"availability_template": "{{ value_json.availability }}",
+			"device": map[string]interface{}{
+				"identifiers": [...]string{fmt.Sprintf("Inverter_%s", uniq)},
+				"name":        "Inverter",
+			},
+		})
+		conn.publish(topic, string(json), true) // MQTT Discovery messages should be retained, but in dev it can become a pain
+	}
+	return nil
+}
 
+func (conn *Connection) InsertRecord(m map[string]interface{}) error {
+	json, _ := json.Marshal(m)
+	conn.publish(conn.state, string(json), false) // state messages should not be retained
 	return nil
 }