first commit

internal/mqttingest/latest.go

@@ -0,0 +1,171 @@
+package mqttingest
+
+import (
+	"sync"
+	"time"
+)
+
+type rainMode int
+
+const (
+	rainModeUnknown rainMode = iota
+	rainModeCumulative
+	rainModeIncremental
+)
+
+type Latest struct {
+	mu sync.RWMutex
+
+	lastTS time.Time
+	last   *WS90Payload
+
+	// Rain tracking
+	mode       rainMode
+	lastRainMM *float64
+
+	// rolling sums built from "rain increment" values (mm)
+	rainIncs  []rainIncPoint // last 1h
+	dailyIncs []rainIncPoint // since midnight (or since start; we’ll trim daily by midnight)
+}
+
+type rainIncPoint struct {
+	ts time.Time
+	mm float64 // incremental rainfall at this timestamp (mm)
+}
+
+func (l *Latest) Update(ts time.Time, p *WS90Payload) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	l.lastTS = ts
+	l.last = p
+
+	inc := l.computeRainIncrement(ts, p.RainMM)
+
+	// Track last hour increments
+	l.rainIncs = append(l.rainIncs, rainIncPoint{ts: ts, mm: inc})
+	cutoff := ts.Add(-1 * time.Hour)
+	l.rainIncs = trimBefore(l.rainIncs, cutoff)
+
+	// Track daily increments: trim before local midnight
+	l.dailyIncs = append(l.dailyIncs, rainIncPoint{ts: ts, mm: inc})
+	midnight := localMidnight(ts)
+	l.dailyIncs = trimBefore(l.dailyIncs, midnight)
+}
+
+func trimBefore(a []rainIncPoint, cutoff time.Time) []rainIncPoint {
+	i := 0
+	for ; i < len(a); i++ {
+		if !a[i].ts.Before(cutoff) {
+			break
+		}
+	}
+	if i > 0 {
+		return a[i:]
+	}
+	return a
+}
+
+// localMidnight returns midnight in the local timezone of the *process*.
+// If you want a specific timezone (e.g. Australia/Sydney) we can wire that in later.
+func localMidnight(t time.Time) time.Time {
+	lt := t.Local()
+	return time.Date(lt.Year(), lt.Month(), lt.Day(), 0, 0, 0, 0, lt.Location())
+}
+
+// computeRainIncrement returns the “incremental rain” in mm for this sample,
+// regardless of whether the incoming rain_mm is cumulative or incremental.
+func (l *Latest) computeRainIncrement(ts time.Time, rainMM float64) float64 {
+	// First sample: we can’t infer anything yet
+	if l.lastRainMM == nil {
+		l.lastRainMM = &rainMM
+		return 0
+	}
+
+	prev := *l.lastRainMM
+	l.lastRainMM = &rainMM
+
+	// Heuristic:
+	// - If value often stays 0 and occasionally jumps by small amounts, it might be cumulative OR incremental.
+	// - If it monotonically increases over time (with occasional resets), that’s cumulative.
+	// - If it is usually small per message (e.g. 0, 0.2, 0, 0, 0.2) and not trending upward, that’s incremental.
+	//
+	// We’ll decide based on “trendiness” and deltas:
+	delta := rainMM - prev
+
+	// Handle reset (counter rollover / daily reset / device reboot)
+	if delta < -0.001 {
+		// If cumulative, after reset the “increment” is 0 for that sample.
+		// If incremental, a reset doesn’t really make sense but we still treat as 0.
+		if l.mode == rainModeUnknown {
+			l.mode = rainModeCumulative
+		}
+		return 0
+	}
+
+	// If we already decided
+	switch l.mode {
+	case rainModeCumulative:
+		if delta > 0 {
+			return delta
+		}
+		return 0
+	case rainModeIncremental:
+		// in incremental mode we treat the sample as “this message’s rain”
+		if rainMM > 0 {
+			return rainMM
+		}
+		return 0
+	}
+
+	// Decide mode (unknown):
+	// If delta is consistently positive when rainMM > 0, cumulative is likely.
+	// If delta is ~0 while rainMM occasionally > 0, incremental is likely.
+	//
+	// Single-sample heuristic:
+	// - if rainMM > 0 and delta > 0 => lean cumulative
+	// - if rainMM > 0 and delta ~ 0 => lean incremental
+	if rainMM > 0 {
+		if delta > 0.0009 {
+			l.mode = rainModeCumulative
+			return delta
+		}
+		// delta near zero but rainMM nonzero suggests incremental
+		l.mode = rainModeIncremental
+		return rainMM
+	}
+
+	return 0
+}
+
+type Snapshot struct {
+	TS time.Time
+	P  WS90Payload
+
+	RainLastHourMM float64
+	DailyRainMM    float64
+}
+
+func (l *Latest) Snapshot() (Snapshot, bool) {
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+
+	if l.last == nil || l.lastTS.IsZero() {
+		return Snapshot{}, false
+	}
+
+	var hourSum, daySum float64
+	for _, rp := range l.rainIncs {
+		hourSum += rp.mm
+	}
+	for _, rp := range l.dailyIncs {
+		daySum += rp.mm
+	}
+
+	return Snapshot{
+		TS:             l.lastTS,
+		P:              *l.last,
+		RainLastHourMM: hourSum,
+		DailyRainMM:    daySum,
+	}, true
+}

internal/mqttingest/mqtt.go

@@ -0,0 +1,53 @@
+package mqttingest
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+type MQTTConfig struct {
+	Broker   string
+	ClientID string
+	Username string
+	Password string
+	Topic    string
+	QoS      byte
+}
+
+type Handler func(ctx context.Context, topic string, payload []byte) error
+
+func RunSubscriber(ctx context.Context, cfg MQTTConfig, h Handler) error {
+	opts := mqtt.NewClientOptions().
+		AddBroker(cfg.Broker).
+		SetClientID(cfg.ClientID).
+		SetAutoReconnect(true).
+		SetConnectRetry(true).
+		SetConnectRetryInterval(5 * time.Second)
+
+	if cfg.Username != "" {
+		opts.SetUsername(cfg.Username)
+		opts.SetPassword(cfg.Password)
+	}
+
+	client := mqtt.NewClient(opts)
+	if tok := client.Connect(); tok.Wait() && tok.Error() != nil {
+		return fmt.Errorf("mqtt connect: %w", tok.Error())
+	}
+
+	// Subscribe
+	if tok := client.Subscribe(cfg.Topic, cfg.QoS, func(_ mqtt.Client, msg mqtt.Message) {
+		// Keep callback short; do work with context
+		_ = h(ctx, msg.Topic(), msg.Payload())
+	}); tok.Wait() && tok.Error() != nil {
+		client.Disconnect(250)
+		return fmt.Errorf("mqtt subscribe: %w", tok.Error())
+	}
+
+	// Block until ctx cancelled
+	<-ctx.Done()
+	client.Disconnect(250)
+	return nil
+}

internal/mqttingest/ws90.go

@@ -0,0 +1,39 @@
+package mqttingest
+
+import "encoding/json"
+
+// WS90Payload matches your JSON keys.
+type WS90Payload struct {
+	Model        string  `json:"model"`
+	ID           int64   `json:"id"`
+	BatteryOK    int     `json:"battery_ok"`
+	BatteryMV    int     `json:"battery_mV"`
+	TemperatureC float64 `json:"temperature_C"`
+	Humidity     float64 `json:"humidity"`
+	WindDirDeg   float64 `json:"wind_dir_deg"`
+	WindAvgMS    float64 `json:"wind_avg_m_s"`
+	WindMaxMS    float64 `json:"wind_max_m_s"`
+	UVI          float64 `json:"uvi"`
+	LightLux     float64 `json:"light_lux"`
+	Flags        int     `json:"flags"`
+	RainMM       float64 `json:"rain_mm"`
+	RainStart    int64   `json:"rain_start"`
+	SupercapV    float64 `json:"supercap_V"`
+	Firmware     int     `json:"firmware"`
+	Data         string  `json:"data"`
+	MIC          string  `json:"mic"`
+	Protocol     string  `json:"protocol"`
+	RSSI         int     `json:"rssi"`
+	Duration     int64   `json:"duration"`
+}
+
+func ParseWS90(b []byte) (*WS90Payload, map[string]any, error) {
+	var p WS90Payload
+	if err := json.Unmarshal(b, &p); err != nil {
+		return nil, nil, err
+	}
+	// Keep the full payload as JSONB too.
+	var raw map[string]any
+	_ = json.Unmarshal(b, &raw)
+	return &p, raw, nil
+}