fuelprices/functions.go

package main

import (
	"archive/zip"
	"bytes"
	"errors"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"strconv"
	"time"

	"github.com/antchfx/xmlquery"
	_ "github.com/influxdata/influxdb1-client"
	client "github.com/influxdata/influxdb1-client/v2"
	"gopkg.in/ini.v1"
)

// GetConfig fetch config from ini file
func (fpc *FuelPricesConfig) GetConfig() error {
	flag.Usage = Usage

	flag.StringVar(&fpc.ConfigPath, "configfile", "common.ini", "config file to use with fuelprices section")
	flag.Parse()

	config, err := ini.Load(fpc.ConfigPath)
	if err != nil {
		return err
	}

	fuelpricesSection := config.Section("fuelprices")
	fpc.RemoteURL = fuelpricesSection.Key("remote_url").MustString("https://donnees.roulez-eco.fr/opendata/instantane")
	fpc.RemoteFilename = fuelpricesSection.Key("remote_filename").MustString("PrixCarburants_instantane.xml")
	fpc.XPathBase = fuelpricesSection.Key("xpath_base").MustString(".//pdv[@id='%s']/prix[@nom='%s']")
	fpc.Table = fuelpricesSection.Key("table").MustString("fuel_price")
	fpc.Pos = fuelpricesSection.Key("pos").Strings(",")
	if len(fpc.Pos) < 1 {
		err := errors.New("No pos defined")
		return err
	}
	fpc.Types = fuelpricesSection.Key("types").Strings(",")
	if len(fpc.Types) < 1 {
		err := errors.New("No fuel types defined")
		return err
	}

	influxdbSection := config.Section("influxdb")
	fpc.InfluxURL = influxdbSection.Key("url").MustString("http://localhost:8086")
	fpc.InfluxUser = influxdbSection.Key("username").MustString("username")
	fpc.InfluxPass = influxdbSection.Key("password").MustString("password")
	fpc.InfluxDB = influxdbSection.Key("database").MustString("me")

	if err != nil {
		return err
	}

	return nil
}

// DownloadFile fetch file from webserver
func DownloadFile(fpc *FuelPricesConfig, zipfile *ZipFile) error {
	pollTo := 30 * time.Millisecond

	client := &http.Client{Timeout: pollTo * time.Second, Transport: &http.Transport{
		IdleConnTimeout:    pollTo,
		DisableCompression: false,
	}}

	resp, err := client.Get(fpc.RemoteURL)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	zipfile.Content, err = ioutil.ReadAll(resp.Body)
	if err != nil {
		return err
	}

	time.Sleep(pollTo)

	return nil
}

// ExtractZip get the XML file to be processed
func ExtractZip(fpc *FuelPricesConfig, zipfile *ZipFile, xmlfile *XMLFile) error {
	unzipped, err := zip.NewReader(bytes.NewReader(zipfile.Content), int64(len(zipfile.Content)))
	if err != nil {
		return err
	}

	for _, file := range unzipped.File {
		if file.Name == fpc.RemoteFilename {
			rc, err := file.Open()
			if err != nil {
				return err
			}
			xmlfile.Content, err = ioutil.ReadAll(rc)
			rc.Close()
		} else {
			log.Fatal("File not found")
		}
	}
	return err
}

// GetPrices parses the XML file and get values of prices
func GetPrices(fpc *FuelPricesConfig, prices *[]Price, xmlfile *XMLFile) error {
	var xml *xmlquery.Node
	var valueattr = "valeur"

	file := bytes.NewReader(xmlfile.Content)
	xml, err := xmlquery.Parse(file)
	if err != nil {
		return err
	}

	for _, station := range fpc.Pos {
		for _, fuel := range fpc.Types {

			query := fmt.Sprintf(fpc.XPathBase, station, fuel)
			list := xmlquery.FindOne(xml, query)

			if list != nil {
				for _, i := range list.Attr {
					if i.Name.Local == valueattr {
						if s, err := strconv.ParseFloat(i.Value, 64); err == nil {
							*prices = append(*prices, Price{ID: station, Fuel: fuel, Amount: s})
						}
					}
				}
			} else {
				log.Println(fmt.Sprintf("Fuel type not found for point of sale, skipping. Query : %s", query))
			}
		}
	}
	return err
}

// SendToInflux sends time series data to influxdb
func SendToInflux(fpc *FuelPricesConfig, prices *[]Price) error {
	httpClient, err := client.NewHTTPClient(client.HTTPConfig{
		Addr:     fpc.InfluxURL,
		Username: fpc.InfluxUser,
		Password: fpc.InfluxPass,
	})
	if err != nil {
		return err
	}

	defer httpClient.Close()

	bp, err := client.NewBatchPoints(client.BatchPointsConfig{
		Database: fpc.InfluxDB,
	})
	if err != nil {
		return err
	}

	for _, p := range *prices {

		tags := map[string]string{"pdv": p.ID, "fuel": p.Fuel}
		fields := map[string]interface{}{"value": p.Amount}

		point, _ := client.NewPoint(
			fpc.Table,
			tags,
			fields,
			time.Now(),
		)

		log.Println(point)

		bp.AddPoint(point)
		err = httpClient.Write(bp)
		if err != nil {
			return err
		}
	}

	return nil
}

// Usage displays possible arguments
func Usage() {
	flag.PrintDefaults()
	os.Exit(1)
}