aboutsummaryrefslogtreecommitdiffstats
path: root/main.go
diff options
context:
space:
mode:
Diffstat (limited to 'main.go')
-rw-r--r--main.go769
1 files changed, 769 insertions, 0 deletions
diff --git a/main.go b/main.go
new file mode 100644
index 0000000..f9afd40
--- /dev/null
+++ b/main.go
@@ -0,0 +1,769 @@
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "encoding/xml"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/url"
+ "os"
+ "os/signal"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "syscall"
+ "time"
+
+ mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+/* ============================================================
+ Configuration (now configurable)
+ ============================================================ */
+
+type Config struct {
+ // FMI API
+ FMIEndpoint string `env:"FMI_ENDPOINT" default:"https://opendata.fmi.fi/wfs"`
+
+ // Observation
+ StationID int `env:"STATION_ID" default:"100968"`
+ ObsPollEvery time.Duration `env:"OBS_POLL_EVERY" default:"5m"`
+ ObsSafetyLag time.Duration `env:"OBS_SAFETY_LAG" default:"2m"`
+ ObsTimestep int `env:"OBS_TIMESTEP" default:"60"`
+ WatermarkFile string `env:"WATERMARK_FILE" default:"obs_watermark.txt"`
+
+ // Forecast
+ FcPollEvery time.Duration `env:"FC_POLL_EVERY" default:"1h"`
+ HelLat float64 `env:"HEL_LAT" default:"60.1699"`
+ HelLon float64 `env:"HEL_LON" default:"24.9384"`
+
+ // MQTT
+ MQTTBroker string `env:"MQTT_BROKER" default:"tcp://localhost:1883"`
+ MQTTClientID string `env:"MQTT_CLIENT_ID" default:"fmi-poller"`
+ MQTTUsername string `env:"MQTT_USERNAME"`
+ MQTTPassword string `env:"MQTT_PASSWORD"`
+
+ // Application
+ LogLevel string `env:"LOG_LEVEL" default:"info"`
+ HTTPTimeout time.Duration `env:"HTTP_TIMEOUT" default:"30s"`
+ MaxRetries int `env:"MAX_RETRIES" default:"3"`
+ RetryDelay time.Duration `env:"RETRY_DELAY" default:"5s"`
+}
+
+/* ============================================================
+ Application State
+ ============================================================ */
+
+type App struct {
+ config *Config
+ logger *slog.Logger
+ mqttClient mqtt.Client
+ httpClient *http.Client
+
+ // State
+ lastObsWatermark atomic.Value // stores time.Time
+ lastForecastRun atomic.Value // stores time.Time
+ shutdownCtx context.Context
+ shutdownCancel context.CancelFunc
+
+ // Wait group for graceful shutdown
+ wg sync.WaitGroup
+}
+
+/* ============================================================
+ Domain models
+ ============================================================ */
+
+type Observation struct {
+ Station int `json:"station"`
+ Parameter string `json:"parameter"`
+ Time time.Time `json:"time"`
+ Value float64 `json:"value"`
+}
+
+type ForecastValue struct {
+ Location struct {
+ Lat float64 `json:"lat"`
+ Lon float64 `json:"lon"`
+ } `json:"location"`
+
+ Model string `json:"model"`
+ RunTime time.Time `json:"run_time"`
+ ForecastTime time.Time `json:"forecast_time"`
+ Parameter string `json:"parameter"`
+ Value float64 `json:"value"`
+}
+
+/* ============================================================
+ Initialization
+ ============================================================ */
+
+func NewApp() (*App, error) {
+ config := loadConfig()
+
+ // Setup logging
+ logLevel := parseLogLevel(config.LogLevel)
+ logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
+ Level: logLevel,
+ }))
+
+ // Create HTTP client with timeout
+ httpClient := &http.Client{
+ Timeout: config.HTTPTimeout,
+ Transport: &http.Transport{
+ MaxIdleConns: 10,
+ IdleConnTimeout: 30 * time.Second,
+ DisableCompression: false,
+ },
+ }
+
+ // Create MQTT client
+ mqttOpts := mqtt.NewClientOptions().
+ AddBroker(config.MQTTBroker).
+ SetClientID(fmt.Sprintf("%s-%d", config.MQTTClientID, os.Getpid())).
+ SetCleanSession(true).
+ SetAutoReconnect(true).
+ SetMaxReconnectInterval(10 * time.Second).
+ SetConnectionLostHandler(func(c mqtt.Client, err error) {
+ logger.Error("MQTT connection lost", "error", err)
+ }).
+ SetOnConnectHandler(func(c mqtt.Client) {
+ logger.Info("MQTT connected successfully")
+ })
+
+ if config.MQTTUsername != "" {
+ mqttOpts.SetUsername(config.MQTTUsername)
+ if config.MQTTPassword != "" {
+ mqttOpts.SetPassword(config.MQTTPassword)
+ }
+ }
+
+ mqttClient := mqtt.NewClient(mqttOpts)
+
+ ctx, cancel := context.WithCancel(context.Background())
+
+ app := &App{
+ config: config,
+ logger: logger,
+ httpClient: httpClient,
+ mqttClient: mqttClient,
+ shutdownCtx: ctx,
+ shutdownCancel: cancel,
+ }
+
+ // Initialize atomic values
+ app.lastObsWatermark.Store(loadWatermark(config.WatermarkFile))
+ app.lastForecastRun.Store(time.Time{})
+
+ return app, nil
+}
+
+func (app *App) ConnectMQTT() error {
+ token := app.mqttClient.Connect()
+ if !token.WaitTimeout(app.config.HTTPTimeout) {
+ return errors.New("MQTT connection timeout")
+ }
+ if err := token.Error(); err != nil {
+ return fmt.Errorf("MQTT connection failed: %w", err)
+ }
+ app.logger.Info("Connected to MQTT broker")
+ return nil
+}
+
+/* ============================================================
+ Configuration loading
+ ============================================================ */
+
+func loadConfig() *Config {
+ cfg := &Config{
+ FMIEndpoint: getEnv("FMI_ENDPOINT", "https://opendata.fmi.fi/wfs"),
+ StationID: parseInt(getEnv("STATION_ID", "100968")),
+ ObsPollEvery: parseDuration(getEnv("OBS_POLL_EVERY", "5m")),
+ ObsSafetyLag: parseDuration(getEnv("OBS_SAFETY_LAG", "2m")),
+ ObsTimestep: parseInt(getEnv("OBS_TIMESTEP", "60")),
+ WatermarkFile: getEnv("WATERMARK_FILE", "obs_watermark.txt"),
+ FcPollEvery: parseDuration(getEnv("FC_POLL_EVERY", "1h")),
+ HelLat: parseFloat(getEnv("HEL_LAT", "60.1699")),
+ HelLon: parseFloat(getEnv("HEL_LON", "24.9384")),
+ MQTTBroker: getEnv("MQTT_BROKER", "tcp://localhost:1883"),
+ MQTTClientID: getEnv("MQTT_CLIENT_ID", "fmi-poller"),
+ MQTTUsername: getEnv("MQTT_USERNAME", ""),
+ MQTTPassword: getEnv("MQTT_PASSWORD", ""),
+ LogLevel: getEnv("LOG_LEVEL", "info"),
+ HTTPTimeout: parseDuration(getEnv("HTTP_TIMEOUT", "30s")),
+ MaxRetries: parseInt(getEnv("MAX_RETRIES", "3")),
+ RetryDelay: parseDuration(getEnv("RETRY_DELAY", "5s")),
+ }
+ return cfg
+}
+
+func getEnv(key, defaultValue string) string {
+ if value, exists := os.LookupEnv(key); exists {
+ return value
+ }
+ return defaultValue
+}
+
+func parseDuration(s string) time.Duration {
+ d, err := time.ParseDuration(s)
+ if err != nil {
+ panic(fmt.Sprintf("invalid duration %s: %v", s, err))
+ }
+ return d
+}
+
+func parseInt(s string) int {
+ i, err := strconv.Atoi(s)
+ if err != nil {
+ panic(fmt.Sprintf("invalid int %s: %v", s, err))
+ }
+ return i
+}
+
+func parseFloat(s string) float64 {
+ f, err := strconv.ParseFloat(s, 64)
+ if err != nil {
+ panic(fmt.Sprintf("invalid float %s: %v", s, err))
+ }
+ return f
+}
+
+func parseLogLevel(level string) slog.Level {
+ switch strings.ToLower(level) {
+ case "debug":
+ return slog.LevelDebug
+ case "warn", "warning":
+ return slog.LevelWarn
+ case "error":
+ return slog.LevelError
+ default:
+ return slog.LevelInfo
+ }
+}
+
+/* ============================================================
+ Watermark persistence
+ ============================================================ */
+
+func loadWatermark(filename string) time.Time {
+ b, err := os.ReadFile(filename)
+ if err != nil {
+ // If file doesn't exist, start from 24 hours ago
+ return time.Now().UTC().Add(-24 * time.Hour)
+ }
+ sec, err := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64)
+ if err != nil {
+ return time.Now().UTC().Add(-24 * time.Hour)
+ }
+ return time.Unix(sec, 0).UTC()
+}
+
+func storeWatermark(filename string, t time.Time) error {
+ data := []byte(fmt.Sprintf("%d", t.Unix()))
+ return os.WriteFile(filename, data, 0644)
+}
+
+/* ============================================================
+ Retry helper
+ ============================================================ */
+
+func withRetry(app *App, name string, maxRetries int, fn func() error) error {
+ var lastErr error
+ for i := 0; i < maxRetries; i++ {
+ if err := fn(); err != nil {
+ lastErr = err
+ app.logger.Warn("Operation failed, retrying",
+ "operation", name,
+ "attempt", i+1,
+ "max_attempts", maxRetries,
+ "error", err)
+
+ if i < maxRetries-1 {
+ select {
+ case <-time.After(app.config.RetryDelay):
+ continue
+ case <-app.shutdownCtx.Done():
+ return app.shutdownCtx.Err()
+ }
+ }
+ } else {
+ return nil
+ }
+ }
+ return fmt.Errorf("%s failed after %d attempts: %w", name, maxRetries, lastErr)
+}
+
+/* ============================================================
+ HTTP helpers with retry
+ ============================================================ */
+
+func (app *App) fetchWithRetry(url string) (io.ReadCloser, error) {
+ var body io.ReadCloser
+ err := withRetry(app, "HTTP fetch", app.config.MaxRetries, func() error {
+ req, err := http.NewRequestWithContext(app.shutdownCtx, http.MethodGet, url, nil)
+ if err != nil {
+ return err
+ }
+ req.Header.Set("Accept", "application/xml")
+ req.Header.Set("User-Agent", "FMI-Weather-Poller/1.0")
+
+ resp, err := app.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+
+ if resp.StatusCode != http.StatusOK {
+ resp.Body.Close()
+ return fmt.Errorf("HTTP %d: %s", resp.StatusCode, resp.Status)
+ }
+
+ body = resp.Body
+ return nil
+ })
+
+ if err != nil {
+ return nil, err
+ }
+ return body, nil
+}
+
+/* ============================================================
+ Observation polling
+ ============================================================ */
+
+func (app *App) fetchObservations(start, end time.Time) (io.ReadCloser, error) {
+ q := url.Values{}
+ q.Set("service", "WFS")
+ q.Set("version", "2.0.0")
+ q.Set("request", "GetFeature")
+ q.Set("storedquery_id", "fmi::observations::weather::timevaluepair")
+ q.Set("fmisid", fmt.Sprintf("%d", app.config.StationID))
+ q.Set("starttime", start.Format(time.RFC3339))
+ q.Set("endtime", end.Format(time.RFC3339))
+ q.Set("timestep", fmt.Sprintf("%d", app.config.ObsTimestep))
+
+ url := app.config.FMIEndpoint + "?" + q.Encode()
+ app.logger.Debug("Fetching observations", "url", url)
+ return app.fetchWithRetry(url)
+}
+
+func (app *App) parseObservations(r io.Reader, minTime time.Time) ([]Observation, time.Time, error) {
+ dec := xml.NewDecoder(r)
+
+ var (
+ param string
+ ts time.Time
+ maxObserved = minTime
+ out []Observation
+ parseErr error
+ )
+
+ for {
+ tok, err := dec.Token()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ return nil, maxObserved, err
+ }
+
+ switch el := tok.(type) {
+ case xml.StartElement:
+ switch el.Name.Local {
+ case "MeasurementTimeseries":
+ for _, a := range el.Attr {
+ if a.Name.Local == "id" {
+ param = normalizeParameter(a.Value)
+ }
+ }
+ case "time":
+ var s string
+ if err := dec.DecodeElement(&s, &el); err != nil {
+ parseErr = fmt.Errorf("failed to decode time: %w", err)
+ continue
+ }
+ ts, err = time.Parse(time.RFC3339, strings.TrimSpace(s))
+ if err != nil {
+ parseErr = fmt.Errorf("failed to parse time %s: %w", s, err)
+ continue
+ }
+ case "value":
+ var v float64
+ if err := dec.DecodeElement(&v, &el); err != nil {
+ parseErr = fmt.Errorf("failed to decode value: %w", err)
+ continue
+ }
+ if ts.After(minTime) {
+ out = append(out, Observation{
+ Station: -1, // Will be set by caller
+ Parameter: param,
+ Time: ts,
+ Value: v,
+ })
+ if ts.After(maxObserved) {
+ maxObserved = ts
+ }
+ }
+ }
+ }
+ }
+
+ if parseErr != nil {
+ app.logger.Warn("Partial parse errors", "error", parseErr)
+ }
+
+ return out, maxObserved, nil
+}
+
+/* ============================================================
+ Forecast polling
+ ============================================================ */
+
+func (app *App) fetchForecast(start, end time.Time) (io.ReadCloser, error) {
+ q := url.Values{}
+ q.Set("service", "WFS")
+ q.Set("version", "2.0.0")
+ q.Set("request", "GetFeature")
+ q.Set("storedquery_id", "fmi::forecast::harmonie::surface::point::simple")
+ q.Set("latlon", fmt.Sprintf("%.4f,%.4f", app.config.HelLat, app.config.HelLon))
+ q.Set("starttime", start.Format(time.RFC3339))
+ q.Set("endtime", end.Format(time.RFC3339))
+
+ url := app.config.FMIEndpoint + "?" + q.Encode()
+ app.logger.Debug("Fetching forecast", "url", url)
+ return app.fetchWithRetry(url)
+}
+
+func (app *App) parseForecast(r io.Reader) ([]ForecastValue, time.Time, error) {
+ dec := xml.NewDecoder(r)
+
+ var (
+ param string
+ runTime time.Time
+ fcTime time.Time
+ out []ForecastValue
+ parseErr error
+ )
+
+ for {
+ tok, err := dec.Token()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ return nil, runTime, err
+ }
+
+ switch el := tok.(type) {
+ case xml.StartElement:
+ switch el.Name.Local {
+ case "resultTime":
+ var s string
+ if err := dec.DecodeElement(&s, &el); err != nil {
+ parseErr = fmt.Errorf("failed to decode resultTime: %w", err)
+ continue
+ }
+ runTime, err = time.Parse(time.RFC3339, strings.TrimSpace(s))
+ if err != nil {
+ parseErr = fmt.Errorf("failed to parse resultTime %s: %w", s, err)
+ continue
+ }
+ case "MeasurementTimeseries":
+ for _, a := range el.Attr {
+ if a.Name.Local == "id" {
+ param = normalizeParameter(a.Value)
+ }
+ }
+ case "time":
+ var s string
+ if err := dec.DecodeElement(&s, &el); err != nil {
+ parseErr = fmt.Errorf("failed to decode forecast time: %w", err)
+ continue
+ }
+ fcTime, err = time.Parse(time.RFC3339, strings.TrimSpace(s))
+ if err != nil {
+ parseErr = fmt.Errorf("failed to parse forecast time %s: %w", s, err)
+ continue
+ }
+ case "value":
+ var v float64
+ if err := dec.DecodeElement(&v, &el); err != nil {
+ parseErr = fmt.Errorf("failed to decode forecast value: %w", err)
+ continue
+ }
+ out = append(out, ForecastValue{
+ Model: "harmonie",
+ RunTime: runTime,
+ ForecastTime: fcTime,
+ Parameter: param,
+ Value: v,
+ })
+ }
+ }
+ }
+
+ if parseErr != nil {
+ app.logger.Warn("Partial parse errors in forecast", "error", parseErr)
+ }
+
+ return out, runTime, nil
+}
+
+/* ============================================================
+ Polling workers
+ ============================================================ */
+
+func (app *App) runObservationPoller() {
+ defer app.wg.Done()
+
+ app.logger.Info("Starting observation poller",
+ "poll_interval", app.config.ObsPollEvery,
+ "station", app.config.StationID)
+
+ ticker := time.NewTicker(app.config.ObsPollEvery)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-app.shutdownCtx.Done():
+ app.logger.Info("Observation poller stopping")
+ return
+ case <-ticker.C:
+ app.pollObservations()
+ }
+ }
+}
+
+func (app *App) pollObservations() {
+ startTime := time.Now()
+ lastObs := app.lastObsWatermark.Load().(time.Time)
+ obsEnd := time.Now().UTC().Add(-app.config.ObsSafetyLag)
+
+ if !obsEnd.After(lastObs) {
+ return // Nothing new to fetch
+ }
+
+ app.logger.Info("Polling observations",
+ "start", lastObs,
+ "end", obsEnd)
+
+ body, err := app.fetchObservations(lastObs, obsEnd)
+ if err != nil {
+ app.logger.Error("Failed to fetch observations", "error", err)
+ return
+ }
+ defer body.Close()
+
+ obs, maxT, err := app.parseObservations(body, lastObs)
+ if err != nil {
+ app.logger.Error("Failed to parse observations", "error", err)
+ return
+ }
+
+ // Set station ID for all observations
+ for i := range obs {
+ obs[i].Station = app.config.StationID
+ }
+
+ if len(obs) > 0 {
+ if err := app.publishObs(obs); err != nil {
+ app.logger.Error("Failed to publish observations", "error", err)
+ return
+ }
+
+ app.lastObsWatermark.Store(maxT)
+ if err := storeWatermark(app.config.WatermarkFile, maxT); err != nil {
+ app.logger.Error("Failed to store watermark", "error", err)
+ }
+
+ app.logger.Info("Published observations",
+ "count", len(obs),
+ "new_watermark", maxT,
+ "duration", time.Since(startTime))
+ } else {
+ app.logger.Debug("No new observations")
+ }
+}
+
+func (app *App) runForecastPoller() {
+ defer app.wg.Done()
+
+ app.logger.Info("Starting forecast poller",
+ "poll_interval", app.config.FcPollEvery,
+ "location", fmt.Sprintf("%.4f,%.4f", app.config.HelLat, app.config.HelLon))
+
+ ticker := time.NewTicker(app.config.FcPollEvery)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-app.shutdownCtx.Done():
+ app.logger.Info("Forecast poller stopping")
+ return
+ case <-ticker.C:
+ app.pollForecast()
+ }
+ }
+}
+
+func (app *App) pollForecast() {
+ startTime := time.Now()
+ now := time.Now().UTC()
+ start := now
+ end := now.Add(48 * time.Hour)
+
+ app.logger.Info("Polling forecast",
+ "start", start,
+ "end", end)
+
+ body, err := app.fetchForecast(start, end)
+ if err != nil {
+ app.logger.Error("Failed to fetch forecast", "error", err)
+ return
+ }
+ defer body.Close()
+
+ fc, runTime, err := app.parseForecast(body)
+ if err != nil {
+ app.logger.Error("Failed to parse forecast", "error", err)
+ return
+ }
+
+ // Set location for all forecasts
+ for i := range fc {
+ fc[i].Location.Lat = app.config.HelLat
+ fc[i].Location.Lon = app.config.HelLon
+ }
+
+ if len(fc) > 0 {
+ // Check if this is a new forecast run
+ lastRun := app.lastForecastRun.Load().(time.Time)
+ if runTime.After(lastRun) {
+ if err := app.publishForecast(fc); err != nil {
+ app.logger.Error("Failed to publish forecast", "error", err)
+ return
+ }
+
+ app.lastForecastRun.Store(runTime)
+ app.logger.Info("Published forecast",
+ "count", len(fc),
+ "run_time", runTime,
+ "duration", time.Since(startTime))
+ } else {
+ app.logger.Debug("Forecast run already published", "run_time", runTime)
+ }
+ }
+}
+
+/* ============================================================
+ MQTT publishing with error handling
+ ============================================================ */
+
+func (app *App) publishObs(obs []Observation) error {
+ for _, o := range obs {
+ topic := fmt.Sprintf("weather/obs/fmi/%d/%s", o.Station, o.Parameter)
+ b, err := json.Marshal(o)
+ if err != nil {
+ return fmt.Errorf("failed to marshal observation: %w", err)
+ }
+
+ token := app.mqttClient.Publish(topic, 1, false, b)
+ if !token.WaitTimeout(app.config.HTTPTimeout) {
+ return errors.New("MQTT publish timeout")
+ }
+ if err := token.Error(); err != nil {
+ return fmt.Errorf("failed to publish observation: %w", err)
+ }
+ }
+ return nil
+}
+
+func (app *App) publishForecast(fc []ForecastValue) error {
+ for _, v := range fc {
+ topic := fmt.Sprintf(
+ "weather/forecast/fmi/harmonie/helsinki/run=%s/%s",
+ v.RunTime.Format(time.RFC3339),
+ v.Parameter,
+ )
+ b, err := json.Marshal(v)
+ if err != nil {
+ return fmt.Errorf("failed to marshal forecast: %w", err)
+ }
+
+ token := app.mqttClient.Publish(topic, 1, true, b)
+ if !token.WaitTimeout(app.config.HTTPTimeout) {
+ return errors.New("MQTT publish timeout")
+ }
+ if err := token.Error(); err != nil {
+ return fmt.Errorf("failed to publish forecast: %w", err)
+ }
+ }
+ return nil
+}
+
+/* ============================================================
+ Utilities
+ ============================================================ */
+
+func normalizeParameter(id string) string {
+ return strings.TrimPrefix(strings.ToLower(id), "ts_")
+}
+
+/* ============================================================
+ Main
+ ============================================================ */
+
+func main() {
+ // Create and initialize app
+ app, err := NewApp()
+ if err != nil {
+ panic(fmt.Sprintf("Failed to initialize app: %v", err))
+ }
+ defer app.shutdownCancel()
+
+ // Setup signal handling
+ signalChan := make(chan os.Signal, 1)
+ signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
+
+ // Connect to MQTT
+ if err := app.ConnectMQTT(); err != nil {
+ app.logger.Error("Failed to connect to MQTT", "error", err)
+ os.Exit(1)
+ }
+ defer app.mqttClient.Disconnect(250)
+
+ // Start pollers
+ app.wg.Add(2)
+ go app.runObservationPoller()
+ go app.runForecastPoller()
+
+ app.logger.Info("Weather poller started")
+
+ // Wait for shutdown signal
+ select {
+ case sig := <-signalChan:
+ app.logger.Info("Received shutdown signal", "signal", sig)
+ app.shutdownCancel()
+ case <-app.shutdownCtx.Done():
+ app.logger.Info("Shutdown initiated")
+ }
+
+ // Wait for graceful shutdown
+ shutdownDone := make(chan struct{})
+ go func() {
+ app.wg.Wait()
+ close(shutdownDone)
+ }()
+
+ select {
+ case <-shutdownDone:
+ app.logger.Info("Shutdown completed")
+ case <-time.After(30 * time.Second):
+ app.logger.Warn("Shutdown timed out")
+ }
+}
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-28 22:04:28 +0000