add co2level support

storage/influxdb-storage.go

@@ -38,7 +38,8 @@ func (storage *influxStorage) Save(data WeatherData) error {
 	fields := map[string]interface{}{
 		"temperature": data.Temperature,
 		"humidity":    data.Humidity,
-		"pressure":    data.Pressure}
+		"pressure":    data.Pressure,
+		"co2level":    data.CO2Level}
 
 	datapoint := influxdb2.NewPoint(storage.measurement,
 		tags,
@@ -88,6 +89,9 @@ func (storage *influxStorage) executeFluxQuery(query string) ([]*WeatherData, er
 		if result.Record().Field() == "humidity" {
 			data.Humidity = result.Record().Value().(float64)
 		}
+		if result.Record().Field() == "co2level" {
+			data.CO2Level = result.Record().Value().(float64)
+		}
 
 		if !contained {
 			data.Location = location

storage/weather-data.go

@@ -17,6 +17,7 @@ type WeatherData struct {
 	Humidity    float64   `json:"humidity"`
 	Pressure    float64   `json:"airPressure"`
 	Temperature float64   `json:"temperature"`
+	CO2Level    float64   `json:"co2level"`
 	Location    string    `json:"location"`
 	TimeStamp   time.Time `json:"timestamp"`
 }
@@ -28,6 +29,7 @@ func NewRandomWeatherData(location string) WeatherData {
 	data.Humidity = rand.Float64() * 100
 	data.Pressure = rand.Float64()*80 + 960
 	data.Temperature = rand.Float64()*40 - 5
+	data.CO2Level = rand.Float64()*50 + 375
 	data.Location = location
 	data.TimeStamp = time.Now()
 	return data

weathersource/mqtt-source.go

@@ -70,6 +70,10 @@ func (source *mqttWeatherSource) mqttMessageHandler() mqtt.MessageHandler {
 			source.lastData.Temperature, _ = strconv.ParseFloat(string(msg.Payload()), 64)
 			source.lastData.TimeStamp = time.Now()
 		}
+		if strings.HasSuffix(msg.Topic(), "co2level") {
+			source.lastData.CO2Level, _ = strconv.ParseFloat(string(msg.Payload()), 64)
+			source.lastData.TimeStamp = time.Now()
+		}
 	}
 }