sungrow/main.go

/*
   Copyright 2023 Josh Deprez

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

	   http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

// The sungrow binary periodically reads inverter data from a sungrow inverter
// and exports the data as prometheus metrics.
package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"log"
	"math/rand"
	"net/http"
	"strings"
	"sync"
	"time"

	"gitea.drjosh.dev/josh/sungrow/modbus"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

const maxScrapeAge = 3 * time.Second

var (
	httpAddr       = flag.String("http-addr", ":9455", "Address to listen on")
	inverterAddrs  = flag.String("inverter-addrs", "rakmodule_00DBC1:502,192.168.86.6:502", "Comma-separated list of inverter addresses (modbus-tcp with 'encryption')")
	scrapeInterval = flag.Duration("scrape-interval", 15*time.Second, "Period of modbus scraping loop")

	promHandler = promhttp.Handler()
	scrapeMu    sync.RWMutex
	lastScrape  time.Time
	lastValues  = make(map[uint16]float64)

	scrapeCounter = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "sungrow",
		Subsystem: "scraper",
		Name:      "scrapes_total",
		Help:      "Number of successful scrapes of the inverter input registers",
	})
	scrapeStart = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: "sungrow",
		Subsystem: "scraper",
		Name:      "scrape_start",
		Help:      "Start time of the most recent scrape attempt",
	})
	scrapeEnd = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: "sungrow",
		Subsystem: "scraper",
		Name:      "scrape_end",
		Help:      "End time of the most recent successful scrape",
	})
	scrapeDuration = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: "sungrow",
		Subsystem: "scraper",
		Name:      "scrape_duration",
		Help:      "units:s",
	})
)

func init() {
	promauto.NewGaugeFunc(
		prometheus.GaugeOpts{
			Namespace: "sungrow",
			Subsystem: "scraper",
			Name:      "scrape_interval",
			Help:      "units:s",
		},
		func() float64 { return scrapeInterval.Seconds() },
	)
	promauto.NewGaugeFunc(
		prometheus.GaugeOpts{
			Namespace: "sungrow",
			Subsystem: "tariff",
			Name:      "daily_charge",
			Help:      "units:$",
		},
		func() float64 { return dailySupplyCharge },
	)
	promauto.NewGaugeFunc(
		prometheus.GaugeOpts{
			Namespace: "sungrow",
			Subsystem: "tariff",
			Name:      "import_tariff",
			Help:      "units:$",
		},
		func() float64 { return tariff93.pricePerKWh(time.Now()) },
	)
	promauto.NewGaugeFunc(
		prometheus.GaugeOpts{
			Namespace: "sungrow",
			Subsystem: "tariff",
			Name:      "export_tariff",
			Help:      "units:$",
		},
		func() float64 { return solarFeedInTariff.pricePerKWh(time.Now()) },
	)
}

func statusHandler(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintf(w, "current time: %v\n", time.Now())
}

func dialInverter() (*sungrowConn, error) {
	for _, addr := range strings.Split(*inverterAddrs, ",") {
		conn, err := dialSungrow(addr)
		if err != nil {
			log.Printf("Couldn't dial inverter: %v", err)
			continue
		}
		return conn, nil
	}
	return nil, fmt.Errorf("all addresses unreachable")
}

// Called under scrapeMu.
func readRegs(vals map[uint16]float64, client modbus.Client, start, qty uint16) error {
	data, err := client.ReadInputRegisters(start, qty)
	if err != nil {
		return fmt.Errorf("read input registers %d-%d: %v", start+1, start+qty, err)
	}
	if len(data) != int(2*qty) {
		return fmt.Errorf("read input registers %d-%d: len(data) = %d != %d = 2*qty", start+1, start+qty, len(data), 2*qty)
	}
	for addr, reg := range sungrowInputRegs {
		if addr <= start || addr > start+qty {
			continue
		}
		val, err := reg.read(data[(addr-start-1)*2:])
		if err != nil {
			if !errors.Is(err, errSkippableRead) {
				return fmt.Errorf("parse input register data at %d: %v", addr, err)
			}
			log.Printf("Couldn't parse input register data at %d, skipping: %v", addr, err)
			val = lastValues[addr]
		}
		//fmt.Printf("%s: %v %s\n", reg.name, val, reg.unit)
		vals[addr] = val
	}
	return nil
}

// Called under scrapeMu.
func scrape() error {
	sgc, err := dialInverter()
	if err != nil {
		return err
	}
	defer sgc.Close()

	handler := modbus.TCPHandlerFromConnection(sgc)
	handler.SlaveId = 0x01
	if err := handler.Connect(); err != nil {
		return err
	}
	defer handler.Close()

	client := modbus.NewClient(handler)

	vals := make(map[uint16]float64)
	start := time.Now()
	scrapeStart.SetToCurrentTime()

	ranges := []struct{ start, qty uint16 }{
		{5000, 24}, {5030, 7}, {5048, 1},
		{5082, 18}, {5112, 1}, {5143, 6},
	}
	rand.Shuffle(len(ranges), func(i, j int) {
		ranges[i], ranges[j] = ranges[j], ranges[i]
	})
	for _, r := range ranges {
		if err := readRegs(vals, client, r.start, r.qty); err != nil {
			return err
		}
	}

	lastValues = vals
	scrapeEnd.SetToCurrentTime()
	lastScrape = time.Now()
	scrapeDuration.Set(time.Since(start).Seconds())
	scrapeCounter.Inc()

	return nil
}

func retries(ctx context.Context, tries int, base time.Duration, mul float64) <-chan int {
	ch := make(chan int)
	go func() {
		defer close(ch)
		i := 0
		for {
			select {
			case ch <- i:
				i++
				if i == tries {
					return
				}

				t := time.NewTimer(time.Duration(rand.Int63n(int64(base))))
				select {
				case <-t.C:
					// next iteration
				case <-ctx.Done():
					t.Stop()
					return
				}

				base = time.Duration(mul * float64(base))
			case <-ctx.Done():
				return
			}
		}
	}()
	return ch
}

func metricsHandler(w http.ResponseWriter, r *http.Request) {
	// In normal mode, always serve metrics
	defer promHandler.ServeHTTP(w, r)

	scrapeMu.Lock()
	defer scrapeMu.Unlock()
	if time.Since(lastScrape) <= maxScrapeAge {
		return
	}

	ctx, canc := context.WithCancel(context.Background())
	defer canc()
	for range retries(ctx, 4, 2*time.Second, 2) {
		if err := scrape(); err != nil {
			log.Printf("Scrape error: %v", err)
			continue
		}
		return
	}
	log.Fatal("Multiple scrape attempts failed, aborting entirely")
}

func main() {
	flag.Parse()

	// These are GaugeFuncs to more closely align Prometheus scrape time with
	// the modbus scrape time.
	for addr, reg := range sungrowInputRegs {
		addr, reg := addr, reg
		promauto.NewGaugeFunc(
			prometheus.GaugeOpts{
				Namespace: "sungrow",
				Subsystem: "inverter",
				Name:      reg.name,
				Help:      fmt.Sprintf("addr: %d, unit: %s", addr, reg.unit),
			},
			func() float64 {
				scrapeMu.RLock()
				defer scrapeMu.RUnlock()
				return lastValues[addr]
			},
		)
	}

	// Startup paranoia check: Is the inverter reachable?
	sgc, err := dialInverter()
	if err != nil {
		log.Fatal("Couldn't dial any addresses, aborting")
	}
	sgc.Close()

	// HTTP setup
	http.HandleFunc("/metrics", metricsHandler)
	http.HandleFunc("/", statusHandler)

	log.Fatalf("http.ListenAndServe: %v", http.ListenAndServe(*httpAddr, nil))
}