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 }