Merge pull request #22 from rddl-network/eckelj/single_values_mqtt_topic

added device status mqtt support and disabled the old aggregated powe…

Author: eckelj

README.md

@@ -87,6 +87,7 @@ topic = "energy-consumption-reports"
     {"value": 50.000, "timestamp": "2025-07-15 00:15:00"},
     {"value": 50.100, "timestamp": "2025-07-15 00:30:00"},
     ... (total 96 entries) ...
+    {"value": 60.100, "timestamp": "2025-07-16 00:00:00"}, // <-- the last entry is always 00:00:00 of the next day 
   ]
 }
 ```

internal/model/devicestatus.go

@@ -0,0 +1,14 @@
+package model
+
+type DeviceStatus struct {
+	IsOn                bool     `json:"isOn"`
+	CurrentVoltage      float64  `json:"currentVoltage"`
+	CurrentAmps         float64  `json:"currentAmps"`
+	CurrentActivePower  float64  `json:"currentActivePower"`
+	TotalEnergyConsumed *float64 `json:"totalEnergyConsumed"`
+}
+
+type DeviceStatusExt struct {
+	ID           string       `json:"id"`
+	DeviceStatus DeviceStatus `json:"deviceStatus"`
+}

internal/server/helper.go

@@ -57,6 +57,31 @@ func (s *Server) write2InfluxDB(data model.EnergyData) error {
 	return nil
 }
 
+func (s *Server) writeDeviceStatus2InfluxDB(data model.DeviceStatusExt) error {
+	writeAPI := s.influxDBClient
+	if writeAPI == nil {
+		log.Printf("No InfluxDB write API set")
+		return nil
+	}
+
+	err := writeAPI.WritePoint(
+		context.Background(),
+		"device_status",
+		map[string]string{
+			"ID": data.ID,
+		},
+		map[string]interface{}{
+			"kW/h": *data.DeviceStatus.TotalEnergyConsumed,
+		},
+		time.Now(),
+	)
+	if err != nil {
+		log.Printf("Failed to write to InfluxDB: %v", err)
+		return err
+	}
+	return nil
+}
+
 // sendJSONResponse sends a JSON response with the given status code
 func sendJSONResponse(w http.ResponseWriter, resp Response, statusCode int) {
 	w.Header().Set("Content-Type", "application/json")

internal/server/mqtt.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"strings"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/rddl-network/energy-service/internal/config"
@@ -25,23 +26,71 @@ func (s *Server) initMQTT() {
 	tlsConfig := &tls.Config{}
 	opts.SetTLSConfig(tlsConfig)
 
+	// Add connection lost handler
+	opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
+		log.Printf("MQTT connection lost: %v", err)
+	})
+	opts.AutoReconnect = true
+
 	client := mqtt.NewClient(opts)
 	if token := client.Connect(); token.Wait() && token.Error() != nil {
 		log.Printf("MQTT connect error: %v", token.Error())
 		return
 	}
 	s.mqttClient = client
-	topic := mqttCfg.Topic
-	if topic == "" {
-		topic = "energy-consumption-reports"
-	}
-	if token := client.Subscribe(topic, 0, s.handleMQTTMessage); token.Wait() && token.Error() != nil {
+	//subscribe(client, mqttCfg.Topic, s.handleMQTTMessage)
+	subscribe(client, "dirigera/+", s.handleSimpleDataMQTTMessage)
+}
+
+func subscribe(client mqtt.Client, topic string, callback mqtt.MessageHandler) {
+	if token := client.Subscribe(topic, 0, callback); token.Wait() && token.Error() != nil {
 		log.Printf("MQTT subscribe error: %v", token.Error())
 	} else {
 		log.Printf("Subscribed to MQTT topic: %s", topic)
 	}
 }
 
+func extractIDFromTopic(topic string) string {
+	parts := strings.Split(topic, "/")
+	if len(parts) == 2 && parts[0] == "dirigera" {
+		return parts[1]
+	}
+	return ""
+}
+
+func (s *Server) handleSimpleDataMQTTMessage(client mqtt.Client, msg mqtt.Message) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("MQTT handler panic: %v", r)
+		}
+	}()
+
+	var deviceStatusExt model.DeviceStatusExt
+
+	deviceStatusExt.ID = extractIDFromTopic(msg.Topic())
+	if deviceStatusExt.ID == "" {
+		log.Printf("MQTT: Invalid topic format: %s", msg.Topic())
+		return
+	}
+
+	if err := json.Unmarshal(msg.Payload(), &deviceStatusExt.DeviceStatus); err != nil {
+		log.Printf("MQTT: Failed to decode JSON: %v", err)
+		return
+	}
+
+	if deviceStatusExt.DeviceStatus.TotalEnergyConsumed == nil {
+		log.Printf("MQTT: JSON does not contain consumed energy values.")
+		return
+	}
+
+	err := s.writeDeviceStatus2InfluxDB(deviceStatusExt)
+	if err != nil {
+		log.Printf("MQTT: Failed to write to database: %v", err)
+		return
+	}
+	log.Printf("MQTT: Energy data received and written to database successfully")
+}
+
 // handleMQTTMessage processes incoming MQTT messages as energy data
 func (s *Server) handleMQTTMessage(client mqtt.Client, msg mqtt.Message) {
 	var energyData model.EnergyData