Include influxdb

main.go

@@ -2,19 +2,25 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"flag"
 	"fmt"
 	"io"
 	"log"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"time"
 	"unicode"
 
+	"github.com/influxdata/influxdb-client-go/v2"
+	"github.com/influxdata/influxdb-client-go/v2/api/write"
 	"github.com/tarm/serial"
 )
 
+var myZone = ""
+
 func readSerial(s *serial.Port, c chan []byte) {
 	var unread bytes.Buffer
 	buf := make([]byte, 128)
@@ -44,8 +50,109 @@ func readSerial(s *serial.Port, c chan []byte) {
 	}
 }
 
-func treatFrames(frames chan []byte) {
+func defaultTags() map[string]string {
+	ret := map[string]string{}
+
+	if myZone != "" {
+		ret["zone"] = myZone
+	}
+
+	if host, err := os.Hostname(); err == nil {
+		ret["host"] = host
+	}
+
+	return ret
+}
+
+func fillUnit(tags *map[string]string, key string) {
+	if v, ok := MeasurementUnits[key]; ok {
+		(*tags)["unit"] = v
+	}
+}
+
+func getValue(key string, data []byte) interface{} {
+	if _, ok := MeasurementUnits[key]; ok {
+		ret, _ := strconv.Atoi(string(data))
+		return ret
+	} else {
+		return string(data)
+	}
+}
+
+func genMeasurements(m map[string][]byte, horodate time.Time) (points []*write.Point) {
+	measurmentdone := []string{}
+
+	// Treat MeasurementGroups
+	for kgrp, grp := range MeasurementGroups {
+		fields := map[string]interface{}{}
+
+		for _, k := range grp {
+			measurmentdone = append(measurmentdone, k)
+
+			if v, ok := m[k]; ok {
+				fields[k] = getValue(k, v)
+			}
+		}
+
+		if len(fields) == 0 {
+			continue
+		}
+
+		tags := defaultTags()
+		fillUnit(&tags, grp[0])
+
+		points = append(points, influxdb2.NewPoint(kgrp,
+			tags,
+			fields,
+			horodate,
+		))
+	}
+
+	// Treat int
+munits:
+	for k := range MeasurementUnits {
+		if _, ok := m[k]; !ok {
+			continue
+		}
+
+		for _, mg := range measurmentdone {
+			if mg == k {
+				continue munits
+			}
+		}
+
+		tags := defaultTags()
+		fillUnit(&tags, k)
+
+		points = append(points, influxdb2.NewPoint(k,
+			tags,
+			map[string]interface{}{"value": getValue(k, m[k])},
+			horodate,
+		))
+	}
+
+	// Treat string
+	for _, k := range MeasurementStrings {
+		if _, ok := m[k]; !ok {
+			continue
+		}
+
+		tags := defaultTags()
+		fillUnit(&tags, k)
+
+		points = append(points, influxdb2.NewPoint(k,
+			tags,
+			map[string]interface{}{"value": getValue(k, m[k])},
+			horodate,
+		))
+	}
+
+	return
+}
+
+func treatFrames(frames chan []byte, client influxdb2.Client) {
 	first := true
+	writeAPI := client.WriteAPIBlocking("", "teleinfo/autogen")
 	for {
 		frame := <-frames
 
@@ -55,6 +162,12 @@ func treatFrames(frames chan []byte) {
 			continue
 		}
 
+		points := []*write.Point{}
+
+		var defaultHorodate time.Time
+
+		m := map[string][]byte{}
+
 		for _, line := range bytes.Split(frame, []byte("\r\n")) {
 			key, horodate, data, err := treatLine(line)
 
@@ -63,7 +176,35 @@ func treatFrames(frames chan []byte) {
 				continue
 			}
 
-			fmt.Println(key, horodate, data)
+			// Skip ADCO, this is the Linky address, confidential and unrelevant
+			if key == "ADCO" {
+				continue
+			}
+
+			if key == "DATE" {
+				defaultHorodate = *horodate
+				continue
+			}
+
+			if horodate == nil {
+				m[key] = data
+			} else {
+				tags := defaultTags()
+				fillUnit(&tags, key)
+
+				points = append(points, influxdb2.NewPoint(key,
+					tags,
+					map[string]interface{}{"value": getValue(key, data)},
+					*horodate,
+				))
+			}
+		}
+
+		points = append(points, genMeasurements(m, defaultHorodate)...)
+
+		err := writeAPI.WritePoint(context.Background(), points...)
+		if err != nil {
+			log.Println("Unable to write points:", err)
 		}
 	}
 }
@@ -87,9 +228,9 @@ func getHorodate(fields *[][]byte) (*time.Time, error) {
 
 		// Handle "saison"
 		if (*fields)[1][0] == 'E' || (*fields)[1][0] == 'e' {
-			horodate = horodate.Add(2 * time.Hour)
+			horodate = horodate.Add(-2 * time.Hour)
 		} else {
-			horodate = horodate.Add(1 * time.Hour)
+			horodate = horodate.Add(-1 * time.Hour)
 		}
 
 		// Mark field as treated
@@ -101,7 +242,7 @@ func getHorodate(fields *[][]byte) (*time.Time, error) {
 	return nil, nil
 }
 
-func treatLine(line []byte) (key string, horodate *time.Time, data [][]byte, err error) {
+func treatLine(line []byte) (key string, horodate *time.Time, data []byte, err error) {
 	line = bytes.TrimSpace(line)
 
 	if len(line) <= 1 {
@@ -122,17 +263,20 @@ func treatLine(line []byte) (key string, horodate *time.Time, data [][]byte, err
 		return
 	}
 
-	data = fields[1 : len(fields)-1]
+	data = bytes.Join(fields[1:len(fields)-1], []byte{' '})
 
 	return
 }
 
 func main() {
-	var interval = flag.Duration("interval", 5*time.Second, "Minimum time between releve")
+	var influxScheme = flag.String("influx-scheme", "http", "Scheme to use to contact InfluxDB")
+	var influxHost = flag.String("influx-host", "localhost", "Host where lives InfluxDB")
+	var influxPort = flag.Uint("influx-port", 8086, "Port where InfluxDB is accessible")
+	var influxUser = flag.String("influx-username", "influx", "Username to use to perform authentication")
+	var influxPass = flag.String("influx-password", "", "Password to use to perform authentication")
+	flag.StringVar(&myZone, "tag-zone", myZone, "Tag zone to add to points")
 	flag.Parse()
 
-	log.Println(interval)
-
 	if len(flag.Args()) < 1 {
 		log.Println("missing required argument: serial device (eg. /dev/ttyUSB0)")
 		return
@@ -151,11 +295,15 @@ func main() {
 		log.Fatal(err)
 	}
 
+	client := influxdb2.NewClient(fmt.Sprintf("%s://%s:%d", *influxScheme, *influxHost, *influxPort), fmt.Sprintf("%s:%s", *influxUser, *influxPass))
+
 	frames := make(chan []byte)
 	go readSerial(s, frames)
-	go treatFrames(frames)
+	go treatFrames(frames, client)
 
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt, syscall.SIGTERM)
 	<-interrupt
+
+	client.Close()
 }