Accomulate in db Counters, Refactor db.go, add configCounters

Author: skillcoder

config.go

@@ -31,9 +31,14 @@ type configT struct {
 		Pass string `yaml:"pass"`
 		Name string `yaml:"name"`
 	}
+	Counters struct {
+		WaterC []string `yaml:"water-c"`
+		WaterH []string `yaml:"water-h"`
+	}
 }
 
 var config = configT{}
+var configCounters = make(map[string]uint8)
 
 // TODO: switch to github.com/spf13/viper
 func configLoad() {
@@ -93,6 +98,15 @@ func configLoad() {
 	check(err)
 
 	configPrintSummary()
+
+	//make optimize varibles
+	for _, v := range config.Counters.WaterC {
+		configCounters[v] = 1 // water cold
+	}
+
+	for _, v := range config.Counters.WaterH {
+		configCounters[v] = 2 // water hot
+	}
 }
 
 func logConfigItem(key string, rawValue interface{}) {

config.yml.sample

@@ -35,3 +35,7 @@ clickhouse:
     pass: '*secret*'
     # HOMER_CLICKHOUSE_NAME - database name
     name: 'homer'
+
+counters:
+    water-c: ['water-c']
+    water-h: ['water-h']

db.go

@@ -4,8 +4,9 @@ package main
 
 import (
 	"time"
-	//  "reflect"
+	//"reflect"
 	"sort"
+	"strconv"
 	"sync"
 
 	"github.com/montanaflynn/stats"
@@ -14,7 +15,7 @@ import (
 type dbItemT struct {
 	ColName string
 	ColType string
-	ColVal  float64
+	ColVal  string
 	Time    int64
 }
 
@@ -31,12 +32,14 @@ func (q *dbQueueT) getChan() <-chan dbItemT {
 }
 
 type dbT struct {
-	m map[int64][]dbItemT
+	m      map[int64][]dbItemT
+	clicks map[uint8][]int64
 }
 
 func newDatabase() *dbT {
 	return &dbT{
-		m: make(map[int64][]dbItemT),
+		m:      make(map[int64][]dbItemT),
+		clicks: make(map[uint8][]int64),
 	}
 }
 
@@ -45,7 +48,7 @@ func (c *dbT) Load(key int64) ([]dbItemT, bool) {
 	return val, ok
 }
 
-func (c *dbT) Add(key int64, item dbItemT) {
+func (c *dbT) AddMetric(key int64, item dbItemT) {
 	column, ok := c.Load(key)
 	if !ok {
 		column = make([]dbItemT, 0, clickhouseMetricCount+1)
@@ -55,6 +58,48 @@ func (c *dbT) Add(key int64, item dbItemT) {
 	c.m[key] = column
 }
 
+func (c *dbT) AddClick(key uint8, time int64) {
+	click, ok := c.clicks[key]
+	if !ok {
+		click = make([]int64, 0, 10)
+	}
+
+	click = append(click, time)
+	c.clicks[key] = click
+}
+
+func (c *dbT) popMetricsFrom(from int64) (rows map[string][]float64) {
+	keys := make([]int64, 0, len(c.m))
+	for k := range c.m {
+		keys = append(keys, k)
+	}
+
+	sort.Slice(keys, func(i, j int) bool { return keys[i] < keys[j] })
+	rows = make(map[string][]float64)
+	for _, key := range keys {
+
+		if key >= from {
+			//log.Debugf("key [%d] %v", key, item);
+			for _, item := range c.m[key] {
+				rowKey := item.ColName + ":" + item.ColType
+				value, err := strconv.ParseFloat(item.ColVal, 64)
+				if err != nil {
+					log.Errorf("[%d] %s %s %s convert to float: %s", item.Time, item.ColName, item.ColType, item.ColVal, err)
+					continue
+				}
+
+				rows[rowKey] = append(rows[rowKey], value)
+			}
+			delete(c.m, key)
+		} else {
+			// FIXME need send it too (its late data)
+			log.Warnf("key [%d] %v", key, c.m[key])
+		}
+	}
+
+	return rows
+}
+
 var dbQueue dbQueueT
 var database *dbT
 var dbShutdownChan = make(chan bool)
@@ -68,45 +113,53 @@ func init() {
 	}
 }
 
-func dbAddMetric(fieldName string, fieldType string, valueInterface float64, time int64) {
+func dbInsert(fieldName string, fieldType string, valueInterface string, time int64) {
 	dbQueue.Add(dbItemT{fieldName, fieldType, valueInterface, time})
 }
 
-func dbAddEvent(fieldName string, fieldType string, valueInterface uint64, time int64) {
-	//dbQueue.Add(dbItemT{fieldName, fieldType, valueInterface, time})
+func getClickTime(item dbItemT) (timestamp int64) {
+	timestamp = item.Time
+	value, err := strconv.ParseUint(item.ColVal, 10, 64)
+	if err != nil {
+		// handle error
+		log.Errorf("Cant convert to time %s %s %s convert to int: %s", item.ColName, item.ColType, item.ColVal, err)
+		return timestamp
+	}
+
+	if value > 1000000000 {
+		timestamp = int64(value)
+	}
+
+	return timestamp
 }
 
 func dbStore(item dbItemT) {
 	//value := reflect.ValueOf(item.ColVal)
 	//valueType := value.Type()
 	log.Debugf("DB [%d] %s:%s = %v", item.Time, item.ColName, item.ColType, item.ColVal)
-	database.Add(item.Time, item)
+	switch item.ColType {
+	case "count":
+		if counterID, ok := configCounters[item.ColName]; ok {
+			database.AddClick(counterID, getClickTime(item))
+		} else {
+			log.Warn("Unknown counter:", item.ColName)
+		}
+	case "move":
+		log.Warn("MOVE")
+	case "led":
+		log.Warn("LED")
+	case "temp", "humd", "pres":
+		database.AddMetric(item.Time, item)
+	default:
+		log.Warnf("Unknown item.ColType: %s", item.ColType)
+	}
 }
 
-func dbDoTransfer() {
+func dbSaveMetrics() {
 	now := time.Now()
 	var timestamp = now.Unix()
 	from := timestamp - 5
-	keys := make([]int64, 0, len(database.m))
-	for k := range database.m {
-		keys = append(keys, k)
-	}
-
-	sort.Slice(keys, func(i, j int) bool { return keys[i] < keys[j] })
-	rows := make(map[string][]float64)
-	for _, key := range keys {
-		if key >= from {
-			//log.Debugf("key [%d] %v", key, item);
-			for _, item := range database.m[key] {
-				rowKey := item.ColName + ":" + item.ColType
-				rows[rowKey] = append(rows[rowKey], item.ColVal)
-			}
-			delete(database.m, key)
-		} else {
-			// FIXME need send it too (its late data)
-			log.Warnf("key [%d] %v", key, database.m[key])
-		}
-	}
+	rows := database.popMetricsFrom(from)
 
 	row := make(map[string]float64)    // Result
 	nowrow := make(map[string]float64) //next old row
@@ -133,23 +186,37 @@ func dbDoTransfer() {
 	go clickhouseMetricInsert(timestamp, row)
 }
 
+func dbProcessEvents() {
+	// TODO
+	// get current time in HH:mm
+	// create consumption speed stats
+	// save to db
+}
+
 // resend accomulated data to clickhouse in one row per sec
 func dbLoop(ms uint32) {
 	dbWg.Add(1)
 	defer dbWg.Done()
 	ticker := time.NewTicker(time.Millisecond * time.Duration(ms))
+	tickerCounter := time.NewTicker(60 * time.Second)
 	defer ticker.Stop()
+	defer tickerCounter.Stop()
 	itemChan := dbQueue.getChan()
 	for {
 		select {
 		case item := <-itemChan:
 			dbStore(item)
 		case <-ticker.C:
 			log.Debug("Make clickhouse row")
-			dbDoTransfer()
+			dbSaveMetrics()
+		case <-ticker.C:
+			log.Debug("Make counter row")
+			dbProcessEvents()
 		case <-dbShutdownChan:
 			log.Debug("DB shuting down")
-			dbDoTransfer()
+			// TODO make it in gorutines/parallel
+			dbSaveMetrics()
+			dbProcessEvents()
 			return
 		}
 	}

esp.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"strconv"
 	"strings"
 	"time"
 )
@@ -82,32 +81,6 @@ func espHandleStat(payload []byte, espName string) {
 		espName, packet.Time, packet.Ap, packet.Rssi, packet.Free, packet.Vcc, packet.Uptime)
 }
 
-func espHandleEvent(espRoom, espTheme, payloadStr string, timestamp int64) {
-	value, err := strconv.ParseUint(payloadStr, 10, 64)
-	if err != nil {
-		// handle error
-		log.Errorf("[%d] %s %s %s convert to int: %s", timestamp, espRoom, espTheme, payloadStr, err)
-		return
-	}
-
-	if value > 1000000000 {
-		timestamp = int64(value)
-	}
-
-	dbAddEvent(espRoom, espTheme, value, timestamp)
-}
-
-func espHandleMetric(espRoom, espTheme, payloadStr string, timestamp int64) {
-	value, err := strconv.ParseFloat(payloadStr, 64)
-	if err != nil {
-		// handle error
-		log.Errorf("[%d] %s %s %s convert to float: %s", timestamp, espRoom, espTheme, payloadStr, err)
-		return
-	}
-
-	dbAddMetric(espRoom, espTheme, value, timestamp)
-}
-
 func parseTopic(topic string) (espName, espTheme, espTag string, err error) {
 	s := strings.Split(topic, "/")
 	if s[1] == "esp" {
@@ -153,11 +126,8 @@ func espMessageHandler(topic string, payload []byte) {
 
 		switch espTheme {
 		// int
-		case "count", "move", "led":
-			espHandleEvent(espRoom, espTheme, payloadStr, timestamp)
-		// float
-		case "temp", "humd", "pres":
-			espHandleMetric(espRoom, espTheme, payloadStr, timestamp)
+		case "count", "move", "led", "temp", "humd", "pres":
+			dbInsert(espRoom, espTheme, payloadStr, timestamp)
 
 		case "debug":
 			log.Debugf("Debug")