EMS/internal/collector/collector.go

package collector

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"strconv"
	"time"

	"github.com/tb/ems/internal/config"
)

// SystemState represents the current state of the energy system,
// collected from Prometheus metrics.
type SystemState struct {
	Timestamp        time.Time
	GridPowerW       float64 // positive = import, negative = export
	BatterySOC       float64 // 0-100
	PVProductionW    float64 // current PV production in watts
	BatteryPowerW    float64 // battery charge/discharge power
	CompressorPowerW float64 // heat pump compressor power
	AmbientTempC     float64 // outdoor temperature
	PhaseL1PowerW    float64 // per-phase grid power L1 (positive=import, negative=export)
	PhaseL2PowerW    float64 // per-phase grid power L2
	PhaseL3PowerW    float64 // per-phase grid power L3
	WWTopTempC       float64 // DHW cylinder top temperature (°C)
	PVYieldTodayKWh  float64 // cumulative PV production today (kWh)
}

// IsExporting returns true if the system is exporting to grid.
func (s SystemState) IsExporting() bool {
	return s.GridPowerW < 0
}

// ExportW returns the export power as a positive number, or 0 if importing.
func (s SystemState) ExportW() float64 {
	if s.GridPowerW < 0 {
		return -s.GridPowerW
	}
	return 0
}

// ImportW returns the import power as a positive number, or 0 if exporting.
func (s SystemState) ImportW() float64 {
	if s.GridPowerW > 0 {
		return s.GridPowerW
	}
	return 0
}

// Collector reads the current system state from a Prometheus instance.
type Collector struct {
	client  *http.Client
	baseURL string
	metrics map[string]string
	logger  *slog.Logger
}

// NewCollector creates a new Prometheus collector.
func NewCollector(cfg *config.Config, logger *slog.Logger) *Collector {
	return &Collector{
		client: &http.Client{
			Timeout: 10 * time.Second,
		},
		baseURL: cfg.Prometheus.URL,
		metrics: cfg.Prometheus.Metrics,
		logger:  logger,
	}
}

// Collect queries Prometheus for the current system state.
func (c *Collector) Collect(ctx context.Context) (SystemState, error) {
	state := SystemState{
		Timestamp: time.Now(),
	}

	type metricTarget struct {
		name  string
		dest  *float64
		scale float64 // unit conversion multiplier (1.0 = no conversion)
	}

	targets := []metricTarget{
		{"grid_power_exchange", &state.GridPowerW, 1},
		{"battery_soc", &state.BatterySOC, 1},
		{"pv_production", &state.PVProductionW, 1000}, // kW → W
		{"battery_power", &state.BatteryPowerW, 1},
		{"compressor_power", &state.CompressorPowerW, 1},
		{"ambient_temp", &state.AmbientTempC, 1},
		{"phase_l1_power", &state.PhaseL1PowerW, 1},
		{"phase_l2_power", &state.PhaseL2PowerW, 1},
		{"phase_l3_power", &state.PhaseL3PowerW, 1},
		{"ww_top_temp", &state.WWTopTempC, 1},
		{"pv_yield_today", &state.PVYieldTodayKWh, 0.001}, // Wh → kWh
	}

	for _, t := range targets {
		metricName, ok := c.metrics[t.name]
		if !ok {
			c.logger.Warn("metric not configured", "key", t.name)
			continue
		}

		val, err := c.queryInstant(ctx, metricName)
		if err != nil {
			c.logger.Error("failed to query metric",
				"key", t.name,
				"metric", metricName,
				"error", err,
			)
			continue
		}
		*t.dest = val * t.scale
	}

	c.logger.Info("collected system state",
		"grid_w", state.GridPowerW,
		"soc", state.BatterySOC,
		"pv_w", state.PVProductionW,
		"pv_today_kwh", fmt.Sprintf("%.1f", state.PVYieldTodayKWh),
		"compressor_w", state.CompressorPowerW,
		"ambient_c", state.AmbientTempC,
		"l1_w", state.PhaseL1PowerW,
		"l2_w", state.PhaseL2PowerW,
		"l3_w", state.PhaseL3PowerW,
		"ww_top_c", state.WWTopTempC,
	)

	return state, nil
}

// promResponse represents the Prometheus API response for instant queries.
type promResponse struct {
	Status string   `json:"status"`
	Data   promData `json:"data"`
}

type promData struct {
	ResultType string       `json:"resultType"`
	Result     []promResult `json:"result"`
}

type promResult struct {
	Metric map[string]string `json:"metric"`
	Value  [2]interface{}    `json:"value"` // [timestamp, "value"]
}

// queryInstant performs a Prometheus instant query and returns the scalar value.
func (c *Collector) queryInstant(ctx context.Context, query string) (float64, error) {
	u, err := url.Parse(c.baseURL + "/api/v1/query")
	if err != nil {
		return 0, fmt.Errorf("parsing URL: %w", err)
	}

	q := u.Query()
	q.Set("query", query)
	u.RawQuery = q.Encode()

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
	if err != nil {
		return 0, fmt.Errorf("creating request: %w", err)
	}

	resp, err := c.client.Do(req)
	if err != nil {
		return 0, fmt.Errorf("querying prometheus: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return 0, fmt.Errorf("prometheus returned status %d", resp.StatusCode)
	}

	var promResp promResponse
	if err := json.NewDecoder(resp.Body).Decode(&promResp); err != nil {
		return 0, fmt.Errorf("decoding response: %w", err)
	}

	if promResp.Status != "success" {
		return 0, fmt.Errorf("prometheus query failed: %s", promResp.Status)
	}

	if len(promResp.Data.Result) == 0 {
		return 0, fmt.Errorf("no data for query %q", query)
	}

	// Value is [timestamp, "string_value"]
	valStr, ok := promResp.Data.Result[0].Value[1].(string)
	if !ok {
		return 0, fmt.Errorf("unexpected value type for %q", query)
	}

	val, err := strconv.ParseFloat(valStr, 64)
	if err != nil {
		return 0, fmt.Errorf("parsing value %q: %w", valStr, err)
	}

	return val, nil
}