fermentord/pkg/temperature/ds18b20.go

package temperature

// Kernel documentation: https://www.kernel.org/doc/html/latest/w1/slaves/w1_therm.html

import (
	"context"
	"errors"
	"io/ioutil"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"git.joco.dk/sng/fermentord/internal/metrics"
	"github.com/getsentry/sentry-go"
)

const (
	sensorConversionTime = 750 * time.Millisecond
)

var (
	// ErrReadSensor indicates that the sensor could not be read
	ErrReadSensor = errors.New("failed to read sensor")
)

var (
	// C receives the temperature readings
	C chan TemperatureReading

	// ConfigUpdates receives the list of sensors to read
	ConfigUpdates chan []string
	sensors       []string
	accErrTrigger int
	accErrSensor  map[string]int
	filters       map[string]*EWMAFilter
)

func Initialize() {
	sensors = make([]string, 0)
	filters = make(map[string]*EWMAFilter)
	C = make(chan TemperatureReading, 30)
	ConfigUpdates = make(chan []string, 1)
	accErrSensor = make(map[string]int)
}

func PollSensors(ctx context.Context, wg *sync.WaitGroup, readingInterval time.Duration, filterWeight float64) {
	hub := sentry.CurrentHub().Clone()
	defer hub.Flush(10 * time.Second)
	defer wg.Done()

	if readingInterval < sensorConversionTime {
		log.Fatalf("Reading interval must be at least %v ms.", sensorConversionTime)
	}

	ticker := time.NewTicker(readingInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			start := time.Now()

			if accErrTrigger > 60 {
				log.Fatal("Thermal bulk read failed 60 times in a row -- terminating")
			}

			for _, sensor := range sensors {
				if accErrSensor[sensor] > 60 {
					log.Fatalf("Thermal sensor read of %v failed 60 times in a row -- terminating", sensor)
				}
			}

			// Trigger a bulk read to start conversion on all sensors.
			if err := triggerBulkRead(); err != nil {
				accErrTrigger++
				hub.CaptureException(err)
				log.Print(err)
				break
			}

			accErrTrigger = 0

			// Ensure that we wait for sensors to convert data.
			deltaSleep := sensorConversionTime - time.Since(start)
			if deltaSleep > 0 {
				// Wait for sensors to convert data.
				time.Sleep(deltaSleep)
			}

			// Read all sensors.
			readSensors(hub)

		case c := <-ConfigUpdates:
			sensors = c

			for k := range filters {
				delete(filters, k)
			}
			for _, x := range sensors {
				filters[x] = NewEWMAFilter(filterWeight)
			}

		case <-ctx.Done():
			close(C)
			return
		}
	}
}

func readSensors(hub *sentry.Hub) {
	for _, sensor := range sensors {
		t, err := read(sensor)

		if err != nil {
			accErrSensor[sensor]++
			hub.CaptureException(err)
			log.Printf("Error reading temperature sensor %v: %v", sensor, err)
			metrics.TemperatureSensorReadingStatus.WithLabelValues(sensor, "failed").Inc()
			continue
		}

		accErrSensor[sensor] = 0
		tw := filters[sensor].Update(float64(t) / 1000)

		r := TemperatureReading{
			Time:    time.Now(),
			Sensor:  sensor,
			Degrees: tw,
		}

		metrics.TemperatureSensorReadingDegreesCelcius.
			WithLabelValues(sensor).
			Set(tw)

		C <- r

		metrics.TemperatureSensorReadingStatus.WithLabelValues(sensor, "ok").Inc()
	}
}

func triggerBulkRead() error {
	f, err := os.OpenFile("/sys/bus/w1/devices/w1_bus_master1/therm_bulk_read", os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	defer f.Close()

	_, err = f.WriteString("trigger\n")
	return err
}

// read returns the temperature of the specified sensor in millidegrees celcius.
func read(sensor string) (int64, error) {
	path := filepath.Join("/sys/bus/w1/devices", sensor, "w1_slave")
	data, err := ioutil.ReadFile(path)
	if err != nil {
		return 0.0, err
	}

	raw := string(data)

	if !strings.Contains(raw, " YES") {
		return 0.0, ErrReadSensor
	}

	i := strings.LastIndex(raw, "t=")
	if i == -1 {
		return 0.0, ErrReadSensor
	}

	c, err := strconv.ParseInt(raw[i+2:len(raw)-1], 10, 64)
	if err != nil {
		return 0.0, err
	}

	return c, nil
}