From 1e7313c12acc417595884991947b7c4de4fedf97 Mon Sep 17 00:00:00 2001 From: Aleksei Larkov Date: Thu, 1 Jan 2026 22:25:44 +0200 Subject: [PATCH] feat: add more data integration --- scheduler/config.go | 9 + scheduler/data.go | 357 ++++++++++++++++++++++++++++++++++++ scheduler/miners.go | 8 +- scheduler/pv.go | 201 -------------------- scheduler/scheduler.go | 224 ++++++++++++---------- scheduler/scheduler_test.go | 79 ++++++-- test_data/metrics.sql | 37 +++- 7 files changed, 597 insertions(+), 318 deletions(-) create mode 100644 scheduler/data.go delete mode 100644 scheduler/pv.go diff --git a/scheduler/config.go b/scheduler/config.go index 7c3cd9c..e928077 100644 --- a/scheduler/config.go +++ b/scheduler/config.go @@ -340,6 +340,15 @@ func (c *Config) Validate() error { return fmt.Errorf("miner_power_super must be non-negative, got: %f", c.MinerPowerSuper) } + // Validate PV integration settings + if c.PVPollInterval <= 0 { + return fmt.Errorf("pv_poll_interval must be greater than 0, got: %s", c.PVPollInterval) + } + + if c.PVIntegrationPeriod <= 0 { + return fmt.Errorf("pv_integration_period must be greater than 0, got: %s", c.PVIntegrationPeriod) + } + return nil } diff --git a/scheduler/data.go b/scheduler/data.go new file mode 100644 index 0000000..f5c569f --- /dev/null +++ b/scheduler/data.go @@ -0,0 +1,357 @@ +package scheduler + +import ( + "database/sql" + "sync" + "time" + + "github.com/devskill-org/ems/meteo" + "github.com/devskill-org/ems/sigenergy" +) + +type WeatherForecastCache struct { + mu sync.RWMutex + forecast *meteo.METJSONForecast + fetchedAt time.Time + cacheDuration time.Duration +} + +func (w *WeatherForecastCache) Get() (*meteo.METJSONForecast, bool) { + w.mu.RLock() + defer w.mu.RUnlock() + + if w.forecast == nil { + return nil, false + } + + if time.Since(w.fetchedAt) > w.cacheDuration { + return nil, false + } + + return w.forecast, true +} + +func (w *WeatherForecastCache) Set(forecast *meteo.METJSONForecast) { + w.mu.Lock() + defer w.mu.Unlock() + + w.forecast = forecast + w.fetchedAt = time.Now() +} + +type DataSample struct { + pvPower float64 + gridPower float64 // positive = import, negative = export + batteryPower float64 // positive = charging, negative = discharging + evdcPower float64 + batterySoc float64 // % + ts time.Time +} + +type DataSamples struct { + mu sync.Mutex + samples []DataSample +} + +func (d *DataSamples) AddSample(pvPower, gridPower, batteryPower, evdcPower, batterySoc float64, ts time.Time) { + d.mu.Lock() + defer d.mu.Unlock() + d.samples = append(d.samples, DataSample{ + pvPower: pvPower, + gridPower: gridPower, + batteryPower: batteryPower, + evdcPower: evdcPower, + batterySoc: batterySoc, + ts: ts, + }) +} + +type IntegratedData struct { + pvTotalPower float64 + gridExportPower float64 + gridImportPower float64 + batteryChargePower float64 + batteryDischargePower float64 + evdcChargePower float64 + loadPower float64 + batterySoc float64 +} + +func (d *DataSamples) IntegrateSamples(pollInterval time.Duration) IntegratedData { + d.mu.Lock() + defer d.mu.Unlock() + + var result IntegratedData + + for _, sample := range d.samples { + energyKWh := pollInterval.Seconds() / 3600.0 // Convert to hours + + result.pvTotalPower += sample.pvPower * energyKWh + + // Grid power: positive = import, negative = export + if sample.gridPower > 0 { + result.gridImportPower += sample.gridPower * energyKWh + } else if sample.gridPower < 0 { + result.gridExportPower += -sample.gridPower * energyKWh + } + + // Battery power: positive = charging, negative = discharging + if sample.batteryPower > 0 { + result.batteryChargePower += sample.batteryPower * energyKWh + } else if sample.batteryPower < 0 { + result.batteryDischargePower += -sample.batteryPower * energyKWh + } + + // EV DC charging power + result.evdcChargePower += sample.evdcPower * energyKWh + + // Keep the last battery SOC + result.batterySoc = sample.batterySoc + } + + // Calculate load: Load = PV + Battery Discharge + Grid Import - Battery Charge - Grid Export - EV Charge + result.loadPower = result.pvTotalPower + result.batteryDischargePower + result.gridImportPower - + result.batteryChargePower - result.gridExportPower - result.evdcChargePower + + d.samples = d.samples[:0] + return result +} + +func (d *DataSamples) IsEmpty() bool { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.samples) == 0 +} + +// GetLatestPower returns the most recent PV power sample, or 0 if no samples exist +func (d *DataSamples) GetLatestPower() float64 { + d.mu.Lock() + defer d.mu.Unlock() + if len(d.samples) == 0 { + return 0 + } + return d.samples[len(d.samples)-1].pvPower +} + +func (s *MinerScheduler) runDataPoll(samples *DataSamples) { + if s.config.PlantModbusAddress == "" { + return + } + client, err := sigenergy.NewTCPClient(s.config.PlantModbusAddress, sigenergy.PlantAddress) + if err != nil { + s.logger.Printf("Data integration: failed to create modbus client: %v", err) + return + } + defer client.Close() + info, err := client.ReadPlantRunningInfo() + if err != nil { + s.logger.Printf("Data integration: failed to read PlantRunningInfo: %v", err) + return + } + samples.AddSample( + info.PhotovoltaicPower, + info.GridSensorActivePower, + info.ESSPower, + info.DCChargerOutputPower, + info.ESSSOC, + time.Now(), + ) +} + +func (s *MinerScheduler) runDataIntegration(samples *DataSamples, pollInterval time.Duration, dataDB *sql.DB, deviceID int, dryRun bool) { + if samples.IsEmpty() { + s.logger.Printf("Data integration: no samples collected in period") + return + } + + data := samples.IntegrateSamples(pollInterval) + timestamp := time.Now() + + if dataDB == nil { + return + } + + // Fetch weather data from meteo API + cloudCoverage, err := s.fetchCloudCoverage() + if err != nil { + s.logger.Printf("Data integration: failed to fetch cloud coverage: %v", err) + } + + weatherSymbol, err := s.fetchWeatherSymbol() + if err != nil { + s.logger.Printf("Data integration: failed to fetch weather symbol: %v", err) + } + + // Calculate costs using current energy prices + config := s.GetConfig() + + // Get current spot price for cost calculations + var gridImportCost, gridExportCost float64 + s.mu.RLock() + marketData := s.pricesMarketData + s.mu.RUnlock() + + if marketData != nil { + spotPrice, found := marketData.LookupAveragePriceInHourByTime(timestamp) + if found && spotPrice > 0 { + // Import cost: (spot price + operator fee + delivery fee) * energy in MWh + importPricePerMWh := spotPrice + config.ImportPriceOperatorFee + config.ImportPriceDeliveryFee + gridImportCost = (importPricePerMWh / 1000.0) * data.gridImportPower // Convert to EUR + + // Export revenue (negative cost): (spot price - operator fee) * energy in MWh + exportPricePerMWh := spotPrice - config.ExportPriceOperatorFee + gridExportCost = (exportPricePerMWh / 1000.0) * data.gridExportPower // Convert to EUR + } + } + + if dryRun { + // DRY-RUN MODE: Log the action without saving to database + s.logger.Printf("Data integration [DRY-RUN]: would save metrics for device_id=%d at %s", deviceID, timestamp.Format(time.RFC3339)) + s.logger.Printf(" PV: %.3f kWh, Grid Import: %.3f kWh (€%.3f), Grid Export: %.3f kWh (€%.3f)", + data.pvTotalPower, data.gridImportPower, gridImportCost, data.gridExportPower, gridExportCost) + s.logger.Printf(" Battery Charge: %.3f kWh, Battery Discharge: %.3f kWh, SOC: %.1f%%", + data.batteryChargePower, data.batteryDischargePower, data.batterySoc) + s.logger.Printf(" EV Charge: %.3f kWh, Load: %.3f kWh", data.evdcChargePower, data.loadPower) + if cloudCoverage != nil { + s.logger.Printf(" Cloud Coverage: %.1f%%", *cloudCoverage) + } + if weatherSymbol != nil { + s.logger.Printf(" Weather: %s", *weatherSymbol) + } + } else { + // Insert comprehensive energy flow data + _, err = dataDB.Exec( + `INSERT INTO metrics ( + timestamp, device_id, metric_name, + pv_total_power, cloud_coverage, weather_symbol, + grid_export_power, grid_import_power, + battery_charge_power, battery_discharge_power, battery_soc, + evdc_charge_power, load_power, + grid_export_cost, grid_import_cost + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)`, + timestamp, deviceID, "energy_flow", + data.pvTotalPower, cloudCoverage, weatherSymbol, + data.gridExportPower, data.gridImportPower, + data.batteryChargePower, data.batteryDischargePower, data.batterySoc, + data.evdcChargePower, data.loadPower, + gridExportCost, gridImportCost, + ) + if err != nil { + s.logger.Printf("Data integration: failed to insert metrics: %v", err) + return + } + + s.logger.Printf("Data integration: saved metrics for device_id=%d at %s", deviceID, timestamp.Format(time.RFC3339)) + s.logger.Printf(" PV: %.3f kWh, Grid Import: %.3f kWh (€%.3f), Grid Export: %.3f kWh (€%.3f)", + data.pvTotalPower, data.gridImportPower, gridImportCost, data.gridExportPower, gridExportCost) + s.logger.Printf(" Battery Charge: %.3f kWh, Battery Discharge: %.3f kWh, SOC: %.1f%%", + data.batteryChargePower, data.batteryDischargePower, data.batterySoc) + s.logger.Printf(" EV Charge: %.3f kWh, Load: %.3f kWh", data.evdcChargePower, data.loadPower) + } +} + +func (s *MinerScheduler) fetchCloudCoverage() (*float64, error) { + // Check cache first + if cachedForecast, ok := s.weatherCache.Get(); ok { + current := cachedForecast.GetCurrentWeather() + if current == nil { + return nil, nil + } + return current.GetCloudCoverage(), nil + } + + // Cache miss, fetch from API + s.logger.Printf("Data integration: fetching weather forecast from API") + config := s.GetConfig() + client := meteo.NewClient(config.UserAgent) + + location := meteo.Location{ + Latitude: config.Latitude, + Longitude: config.Longitude, + } + + params := meteo.QueryParams{Location: location} + forecast, err := client.GetCompact(params) + if err != nil { + return nil, err + } + + // Store in cache + s.weatherCache.Set(forecast) + + current := forecast.GetCurrentWeather() + if current == nil { + return nil, nil + } + + return current.GetCloudCoverage(), nil +} + +func (s *MinerScheduler) fetchWeatherSymbol() (*string, error) { + // Check cache first + if cachedForecast, ok := s.weatherCache.Get(); ok { + current := cachedForecast.GetCurrentWeather() + if current == nil { + return nil, nil + } + symbol := current.GetSymbolCode() + if symbol == nil { + return nil, nil + } + symbolStr := string(*symbol) + return &symbolStr, nil + } + + // Cache miss, fetch from API + config := s.GetConfig() + client := meteo.NewClient(config.UserAgent) + + location := meteo.Location{ + Latitude: config.Latitude, + Longitude: config.Longitude, + } + + params := meteo.QueryParams{Location: location} + forecast, err := client.GetCompact(params) + if err != nil { + return nil, err + } + + // Store in cache + s.weatherCache.Set(forecast) + + current := forecast.GetCurrentWeather() + if current == nil { + return nil, nil + } + + symbol := current.GetSymbolCode() + if symbol == nil { + return nil, nil + } + symbolStr := string(*symbol) + return &symbolStr, nil +} + +// GetPlantRunningInfo returns the current plant running information +// If PlantModbusAddress is not configured, returns nil +func (s *MinerScheduler) GetPlantRunningInfo() *sigenergy.PlantRunningInfo { + if s.config.PlantModbusAddress == "" { + return nil + } + + client, err := sigenergy.NewTCPClient(s.config.PlantModbusAddress, sigenergy.PlantAddress) + if err != nil { + s.logger.Printf("Failed to create modbus client for plant info: %v", err) + return nil + } + defer client.Close() + + info, err := client.ReadPlantRunningInfo() + if err != nil { + s.logger.Printf("Failed to read plant running info: %v", err) + return nil + } + + return info +} diff --git a/scheduler/miners.go b/scheduler/miners.go index 33c166e..6e405b5 100644 --- a/scheduler/miners.go +++ b/scheduler/miners.go @@ -13,7 +13,13 @@ import ( func (s *MinerScheduler) discoverMiners(ctx context.Context) error { s.logger.Printf("Discovering miners on network: %s", s.config.Network) - newlyDiscoveredMiners := miners.Discover(ctx, s.config.Network) + // Use injected discovery function for testing, otherwise use default + var newlyDiscoveredMiners []*miners.AvalonQHost + if s.minerDiscoveryFunc != nil { + newlyDiscoveredMiners = s.minerDiscoveryFunc(ctx, s.config.Network) + } else { + newlyDiscoveredMiners = miners.Discover(ctx, s.config.Network) + } s.mu.Lock() defer s.mu.Unlock() diff --git a/scheduler/pv.go b/scheduler/pv.go deleted file mode 100644 index 645fe54..0000000 --- a/scheduler/pv.go +++ /dev/null @@ -1,201 +0,0 @@ -package scheduler - -import ( - "database/sql" - "sync" - "time" - - "github.com/devskill-org/ems/meteo" - "github.com/devskill-org/ems/sigenergy" -) - -type WeatherForecastCache struct { - mu sync.RWMutex - forecast *meteo.METJSONForecast - fetchedAt time.Time - cacheDuration time.Duration -} - -func (w *WeatherForecastCache) Get() (*meteo.METJSONForecast, bool) { - w.mu.RLock() - defer w.mu.RUnlock() - - if w.forecast == nil { - return nil, false - } - - if time.Since(w.fetchedAt) > w.cacheDuration { - return nil, false - } - - return w.forecast, true -} - -func (w *WeatherForecastCache) Set(forecast *meteo.METJSONForecast) { - w.mu.Lock() - defer w.mu.Unlock() - - w.forecast = forecast - w.fetchedAt = time.Now() -} - -type PVSample struct { - value float64 - ts time.Time -} - -type PVSamples struct { - mu sync.Mutex - samples []PVSample -} - -func (p *PVSamples) AddSample(value float64, ts time.Time) { - p.mu.Lock() - defer p.mu.Unlock() - p.samples = append(p.samples, PVSample{value: value, ts: ts}) -} - -func (p *PVSamples) IntegrateSamples(pollInterval time.Duration) float64 { - p.mu.Lock() - defer p.mu.Unlock() - var total float64 - for _, sample := range p.samples { - total += sample.value * pollInterval.Seconds() / 3600.0 // kW * sec / 3600 = kWh - } - p.samples = p.samples[:0] - return total -} - -func (p *PVSamples) IsEmpty() bool { - p.mu.Lock() - defer p.mu.Unlock() - return len(p.samples) == 0 -} - -// GetLatestPower returns the most recent PV power sample, or 0 if no samples exist -func (p *PVSamples) GetLatestPower() float64 { - p.mu.Lock() - defer p.mu.Unlock() - if len(p.samples) == 0 { - return 0 - } - return p.samples[len(p.samples)-1].value -} - -func (s *MinerScheduler) runPVPoll(samples *PVSamples) { - if s.config.PlantModbusAddress == "" { - return - } - client, err := sigenergy.NewTCPClient(s.config.PlantModbusAddress, sigenergy.PlantAddress) - if err != nil { - s.logger.Printf("PV integration: failed to create modbus client: %v", err) - return - } - defer client.Close() - info, err := client.ReadPlantRunningInfo() - if err != nil { - s.logger.Printf("PV integration: failed to read PlantRunningInfo: %v", err) - return - } - samples.AddSample(info.PhotovoltaicPower, time.Now()) -} - -func (s *MinerScheduler) runPVIntegration(samples *PVSamples, pollInterval time.Duration, pvDB *sql.DB, deviceID int, dryRun bool) { - if samples.IsEmpty() { - s.logger.Printf("PV integration: no samples collected in period") - return - } - total := samples.IntegrateSamples(pollInterval) - timestamp := time.Now() - if pvDB != nil { - // Fetch cloud coverage from meteo API - cloudCoverage, err := s.fetchCloudCoverage() - if err != nil { - s.logger.Printf("PV integration: failed to fetch cloud coverage: %v", err) - } - - if dryRun { - // DRY-RUN MODE: Log the action without saving to database - if cloudCoverage != nil { - s.logger.Printf("PV integration [DRY-RUN]: would save %.3f kWh with cloud coverage %.1f%% for device_id=%d at %s", total, *cloudCoverage, deviceID, timestamp.Format(time.RFC3339)) - } else { - s.logger.Printf("PV integration [DRY-RUN]: would save %.3f kWh without cloud coverage for device_id=%d at %s", total, deviceID, timestamp.Format(time.RFC3339)) - } - } else { - // Insert with cloud coverage - _, err = pvDB.Exec( - `INSERT INTO metrics (timestamp, device_id, metric_name, pv_total_power, cloud_coverage) VALUES ($1, $2, $3, $4, $5)`, - timestamp, deviceID, "pv_total_power", total, cloudCoverage, - ) - if err != nil { - s.logger.Printf("PV integration: failed to insert metric: %v", err) - return - } - if cloudCoverage != nil { - s.logger.Printf("PV integration: saved %.3f kWh with cloud coverage %.1f%% for device_id=%d at %s", total, *cloudCoverage, deviceID, timestamp.Format(time.RFC3339)) - } else { - s.logger.Printf("PV integration: saved %.3f kWh without cloud coverage for device_id=%d at %s", total, deviceID, timestamp.Format(time.RFC3339)) - } - } - } -} - -func (s *MinerScheduler) fetchCloudCoverage() (*float64, error) { - // Check cache first - if cachedForecast, ok := s.weatherCache.Get(); ok { - s.logger.Printf("PV integration: using cached weather forecast") - current := cachedForecast.GetCurrentWeather() - if current == nil { - return nil, nil - } - return current.GetCloudCoverage(), nil - } - - // Cache miss, fetch from API - s.logger.Printf("PV integration: fetching weather forecast from API") - client := meteo.NewClient(s.config.UserAgent) - - location := meteo.Location{ - Latitude: s.config.Latitude, - Longitude: s.config.Longitude, - } - - params := meteo.QueryParams{Location: location} - forecast, err := client.GetCompact(params) - if err != nil { - return nil, err - } - - // Store in cache - s.weatherCache.Set(forecast) - - current := forecast.GetCurrentWeather() - if current == nil { - return nil, nil - } - - return current.GetCloudCoverage(), nil -} - -// GetPlantRunningInfo returns the current plant running information -// If PlantModbusAddress is not configured, returns nil -func (s *MinerScheduler) GetPlantRunningInfo() *sigenergy.PlantRunningInfo { - if s.config.PlantModbusAddress == "" { - return nil - } - - client, err := sigenergy.NewTCPClient(s.config.PlantModbusAddress, sigenergy.PlantAddress) - if err != nil { - s.logger.Printf("Failed to create modbus client for plant info: %v", err) - return nil - } - defer client.Close() - - info, err := client.ReadPlantRunningInfo() - if err != nil { - s.logger.Printf("Failed to read plant running info: %v", err) - return nil - } - - return info -} diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index 8c82803..a424701 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -14,6 +14,57 @@ import ( _ "github.com/lib/pq" ) +// PeriodicTask represents a task that runs periodically with an optional initial delay +type PeriodicTask struct { + name string + initialDelay time.Duration + interval time.Duration + runFunc func() +} + +// run executes the periodic task in a loop, respecting the initial delay and context cancellation +func (pt *PeriodicTask) run(ctx context.Context, stopChan <-chan struct{}, logger *log.Logger) { + // Wait for initial delay + if pt.initialDelay > 0 { + logger.Printf("[%s] Waiting for initial delay: %v", pt.name, pt.initialDelay) + select { + case <-time.After(pt.initialDelay): + // Initial delay passed, run the task + logger.Printf("[%s] Initial delay passed, running first iteration", pt.name) + pt.runFunc() + case <-ctx.Done(): + logger.Printf("[%s] Stopped during initial delay due to context cancellation", pt.name) + return + case <-stopChan: + logger.Printf("[%s] Stopped during initial delay due to stop signal", pt.name) + return + } + } else { + // No initial delay, run immediately + logger.Printf("[%s] Running immediately (no initial delay)", pt.name) + pt.runFunc() + } + + // Create ticker for periodic execution + ticker := time.NewTicker(pt.interval) + defer ticker.Stop() + + logger.Printf("[%s] Started with interval: %v", pt.name, pt.interval) + + for { + select { + case <-ticker.C: + pt.runFunc() + case <-ctx.Done(): + logger.Printf("[%s] Stopped due to context cancellation", pt.name) + return + case <-stopChan: + logger.Printf("[%s] Stopped due to stop signal", pt.name) + return + } + } +} + type MinerScheduler struct { // Configuration config *Config @@ -37,6 +88,9 @@ type MinerScheduler struct { // Logging logger *log.Logger + + // Test hooks for dependency injection + minerDiscoveryFunc func(ctx context.Context, network string) []*miners.AvalonQHost } // NewMinerScheduler creates a new scheduler instance @@ -70,7 +124,6 @@ func (s *MinerScheduler) SetConfig(config *Config) { s.mu.Lock() defer s.mu.Unlock() s.config = config - s.logger.Printf("Configuration updated") } // GetConfig returns the current configuration @@ -130,119 +183,88 @@ func (s *MinerScheduler) Start(ctx context.Context, serverOnly bool) error { } } - // Run the first checks immediately - go func() { - s.runMinerDiscovery(ctx) - s.runPriceCheck(ctx) - s.runMPCOptimize(ctx) - }() - config := s.GetConfig() - // PV integration state - pvSamples := &PVSamples{} - var pvDB *sql.DB - var pvDBErr error + // Data integration state + dataSamples := &DataSamples{} + var dataDB *sql.DB + var dataDBErr error if s.config.PostgresConnString != "" { - pvDB, pvDBErr = sql.Open("postgres", s.config.PostgresConnString) - if pvDBErr != nil { - s.logger.Printf("PV integration: failed to connect to DB: %v", pvDBErr) - pvDB = nil + dataDB, dataDBErr = sql.Open("postgres", s.config.PostgresConnString) + if dataDBErr != nil { + s.logger.Printf("Data integration: failed to connect to DB: %v", dataDBErr) + dataDB = nil } } - // Get initial delays + // Calculate initial delays now := time.Now() minersControlInitialDelay := s.getInitialDelay(now, config.CheckPriceInterval) + time.Second pvDataInitialDelay := s.getInitialDelay(now, config.PVIntegrationPeriod) - s.logger.Printf("Schedule next miners check for %v", now.Add(minersControlInitialDelay)) - s.logger.Printf("Schedule next PV data collection for %v", now.Add(pvDataInitialDelay)) - - // Only create PV tickers if intervals are configured - var pvTicker *time.Ticker - var pvResetTicker *time.Ticker - if config.PVPollInterval > 0 { - pvTicker = time.NewTicker(config.PVPollInterval) - defer pvTicker.Stop() - } - if config.PVIntegrationPeriod > 0 { - pvResetTicker = time.NewTicker(config.PVIntegrationPeriod) - defer pvResetTicker.Stop() + stateCheckInitialDelay := s.getInitialDelay(now, config.MinersStateCheckInterval) + + // Create periodic tasks + tasks := []PeriodicTask{ + { + name: "MinerDiscovery", + initialDelay: 0, // Run immediately + interval: config.MinerDiscoveryInterval, + runFunc: func() { + s.runMinerDiscovery(ctx) + }, + }, + { + name: "PriceCheckAndMPC", + initialDelay: minersControlInitialDelay, + interval: config.CheckPriceInterval, + runFunc: func() { + s.runPriceCheck(ctx) + s.runMPCOptimize(ctx) + }, + }, + { + name: "StateCheck", + initialDelay: stateCheckInitialDelay, + interval: config.MinersStateCheckInterval, + runFunc: func() { + s.runStateCheck(ctx) + }, + }, + { + name: "DataPoll", + initialDelay: pvDataInitialDelay, + interval: config.PVPollInterval, + runFunc: func() { + s.runDataPoll(dataSamples) + }, + }, + { + name: "DataIntegration", + initialDelay: pvDataInitialDelay, + interval: config.PVIntegrationPeriod, + runFunc: func() { + s.runDataIntegration(dataSamples, config.PVPollInterval, dataDB, config.DeviceID, config.DryRun) + }, + }, } - priceCheckTicker := time.NewTicker(config.CheckPriceInterval) - defer priceCheckTicker.Stop() - - stateCheckTicker := time.NewTicker(config.MinersStateCheckInterval) - defer stateCheckTicker.Stop() - - minerDiscoveryTicker := time.NewTicker(config.MinerDiscoveryInterval) - defer minerDiscoveryTicker.Stop() - - minersControlInitialDelayTick := time.After(minersControlInitialDelay) - minersControlInitialDelayPassed := false + // Start each periodic task in its own goroutine + var wg sync.WaitGroup + for _, task := range tasks { + wg.Add(1) + task := task // capture loop variable + go func() { + defer wg.Done() + task.run(ctx, s.stopChan, s.logger) + }() + } - pvDataInitialDelayTick := time.After(pvDataInitialDelay) - pvDataInitialDelayPassed := false + // Wait for all tasks to complete + wg.Wait() - for { - select { - case <-ctx.Done(): - s.logger.Printf("Scheduler stopping due to context cancellation") - s.stop() - return ctx.Err() - case <-s.stopChan: - s.logger.Printf("Scheduler stopping due to stop signal") - return nil - case <-minersControlInitialDelayTick: - go s.runPriceCheck(ctx) - go s.runMPCOptimize(ctx) - priceCheckTicker.Reset(config.CheckPriceInterval) - minersControlInitialDelayPassed = true - case <-pvDataInitialDelayTick: - if config.PVPollInterval > 0 { - go s.runPVPoll(pvSamples) - if pvTicker != nil { - pvTicker.Reset(config.PVPollInterval) - } - if pvResetTicker != nil { - pvResetTicker.Reset(config.PVIntegrationPeriod) - } - pvDataInitialDelayPassed = true - } - case <-priceCheckTicker.C: - if minersControlInitialDelayPassed { - go s.runPriceCheck(ctx) - go s.runMPCOptimize(ctx) - } - case <-stateCheckTicker.C: - go s.runStateCheck(ctx) - case <-minerDiscoveryTicker.C: - go s.runMinerDiscovery(ctx) - case <-func() <-chan time.Time { - if pvTicker != nil { - return pvTicker.C - } - // Return a channel that never sends - ch := make(chan time.Time) - return ch - }(): - if pvDataInitialDelayPassed && pvTicker != nil { - go s.runPVPoll(pvSamples) - } - case <-func() <-chan time.Time { - if pvResetTicker != nil { - return pvResetTicker.C - } - // Return a channel that never sends - ch := make(chan time.Time) - return ch - }(): - if pvDataInitialDelayPassed && pvResetTicker != nil { - go s.runPVIntegration(pvSamples, config.PVPollInterval, pvDB, config.DeviceID, config.DryRun) - } - } - } + s.logger.Printf("All periodic tasks stopped") + s.stop() + return nil } // Stop gracefully stops the scheduler diff --git a/scheduler/scheduler_test.go b/scheduler/scheduler_test.go index c723056..5381698 100644 --- a/scheduler/scheduler_test.go +++ b/scheduler/scheduler_test.go @@ -76,6 +76,11 @@ func mockEnergyPricesServer() *httptest.Server { })) } +// mockMinerDiscovery returns an empty list of miners to avoid real network requests +func mockMinerDiscovery(ctx context.Context, network string) []*miners.AvalonQHost { + return []*miners.AvalonQHost{} +} + // testConfig creates a basic config for testing with all required fields func testConfig() *Config { return testConfigWithServer(nil) @@ -130,6 +135,8 @@ func TestNewMinerScheduler(t *testing.T) { t.Fatal("NewMinerScheduler returned nil") } + scheduler.minerDiscoveryFunc = mockMinerDiscovery + status := scheduler.GetStatus() if status.IsRunning { @@ -149,20 +156,19 @@ func TestDryRunConfiguration(t *testing.T) { dryRun bool }{ { - name: "dry-run enabled", + name: "dry run enabled", dryRun: true, }, { - name: "dry-run disabled", + name: "dry run disabled", dryRun: false, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - config := DefaultConfig() + config := testConfig() config.DryRun = tt.dryRun - config.SecurityToken = "test-token" scheduler := NewMinerScheduler(config, nil) @@ -170,6 +176,8 @@ func TestDryRunConfiguration(t *testing.T) { t.Fatal("NewMinerScheduler returned nil") } + scheduler.minerDiscoveryFunc = mockMinerDiscovery + actualConfig := scheduler.GetConfig() if actualConfig.DryRun != tt.dryRun { t.Errorf("Expected DryRun to be %v, got %v", tt.dryRun, actualConfig.DryRun) @@ -183,6 +191,7 @@ func TestSchedulerRunningState(t *testing.T) { defer mockServer.Close() scheduler := NewMinerScheduler(testConfigWithServer(mockServer), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Initially not running if scheduler.IsRunning() { @@ -211,8 +220,8 @@ func TestSchedulerRunningState(t *testing.T) { // Wait for completion select { case err := <-done: - if err != context.Canceled { - t.Errorf("Expected context.Canceled error, got %v", err) + if err != nil { + t.Errorf("Unexpected error, got %v", err) } case <-time.After(2 * time.Second): t.Error("Scheduler did not stop within timeout") @@ -228,6 +237,7 @@ func TestSchedulerDoubleStart(t *testing.T) { defer mockServer.Close() scheduler := NewMinerScheduler(testConfigWithServer(mockServer), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -256,6 +266,7 @@ func TestSchedulerStop(t *testing.T) { defer mockServer.Close() scheduler := NewMinerScheduler(testConfigWithServer(mockServer), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery ctx := context.Background() // Start scheduler @@ -289,6 +300,7 @@ func TestSchedulerStop(t *testing.T) { func TestGetDiscoveredMiners(t *testing.T) { scheduler := NewMinerScheduler(testConfig(), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Initially empty minersList := scheduler.GetDiscoveredMiners() @@ -330,6 +342,8 @@ func TestDiscoverMinersPreservesExisting(t *testing.T) { config := testConfig() config.Network = "127.0.0.1/32" scheduler := NewMinerScheduler(config, nil) + // Mock discovery to return no miners (simulating network scan that finds nothing) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Mock some initial miners initialMiners := []*miners.AvalonQHost{ @@ -375,6 +389,7 @@ func TestDiscoverMinersPreservesExisting(t *testing.T) { func TestGetStatus(t *testing.T) { scheduler := NewMinerScheduler(testConfig(), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery status := scheduler.GetStatus() @@ -393,6 +408,7 @@ func TestGetStatus(t *testing.T) { func TestSchedulerStatus_WithData(t *testing.T) { scheduler := NewMinerScheduler(testConfig(), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Add some mock data mockMiners := []*miners.AvalonQHost{ @@ -427,6 +443,7 @@ func TestSchedulerStatus_WithData(t *testing.T) { func TestSchedulerConcurrency(t *testing.T) { scheduler := NewMinerScheduler(testConfig(), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Test concurrent access to methods done := make(chan bool, 10) @@ -467,6 +484,7 @@ func TestSchedulerConcurrency(t *testing.T) { // Benchmark tests func BenchmarkSchedulerGetStatus(b *testing.B) { scheduler := NewMinerScheduler(testConfig(), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Add some mock data mockMiners := make([]*miners.AvalonQHost, 100) @@ -625,6 +643,7 @@ func TestGetInitialDelay(t *testing.T) { func BenchmarkSchedulerGetDiscoveredMiners(b *testing.B) { scheduler := NewMinerScheduler(testConfig(), nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Add mock miners mockMiners := make([]*miners.AvalonQHost, 1000) @@ -646,13 +665,13 @@ func BenchmarkSchedulerGetDiscoveredMiners(b *testing.B) { } func TestMinersStateCheckInterval(t *testing.T) { - config := DefaultConfig() - config.MinersStateCheckInterval = 30 * time.Second - + config := testConfig() + config.MinersStateCheckInterval = 2 * time.Second scheduler := NewMinerScheduler(config, nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery - if scheduler.config.MinersStateCheckInterval != 30*time.Second { - t.Errorf("Expected MinersStateCheckInterval 30s, got %v", scheduler.config.MinersStateCheckInterval) + if scheduler.config.MinersStateCheckInterval != 2*time.Second { + t.Errorf("Expected MinersStateCheckInterval 2s, got %v", scheduler.config.MinersStateCheckInterval) } // Test that default is 1 minute @@ -663,11 +682,12 @@ func TestMinersStateCheckInterval(t *testing.T) { } func TestRunStateCheckDryRun(t *testing.T) { - config := DefaultConfig() + config := testConfig() config.DryRun = true config.MinersStateCheckInterval = 10 * time.Second scheduler := NewMinerScheduler(config, nil) + scheduler.minerDiscoveryFunc = mockMinerDiscovery // Test that runStateCheck doesn't panic with no miners scheduler.runStateCheck(context.Background()) @@ -677,3 +697,38 @@ func TestRunStateCheckDryRun(t *testing.T) { t.Error("Expected DryRun to be true") } } + +func TestMockMinerDiscovery(t *testing.T) { + config := testConfig() + scheduler := NewMinerScheduler(config, nil) + + // Create a custom mock that tracks if it was called + mockCalled := false + customMock := func(ctx context.Context, network string) []*miners.AvalonQHost { + mockCalled = true + // Return some test miners + return []*miners.AvalonQHost{ + {Address: "192.168.1.100", Port: 4028}, + {Address: "192.168.1.101", Port: 4028}, + } + } + + scheduler.minerDiscoveryFunc = customMock + + // Run discovery + err := scheduler.discoverMiners(context.Background()) + if err != nil { + t.Errorf("discoverMiners failed: %v", err) + } + + // Verify mock was called + if !mockCalled { + t.Error("Expected mock discovery function to be called") + } + + // Verify miners were discovered + miners := scheduler.GetDiscoveredMiners() + if len(miners) != 2 { + t.Errorf("Expected 2 miners, got %d", len(miners)) + } +} diff --git a/test_data/metrics.sql b/test_data/metrics.sql index 6b89561..f530a76 100644 --- a/test_data/metrics.sql +++ b/test_data/metrics.sql @@ -4,6 +4,16 @@ CREATE TABLE metrics ( metric_name VARCHAR(100), pv_total_power NUMERIC, cloud_coverage NUMERIC, + grid_export_power NUMERIC, + grid_import_power NUMERIC, + battery_charge_power NUMERIC, + battery_discharge_power NUMERIC, + evdc_charge_power NUMERIC, + load_power NUMERIC, + grid_export_cost NUMERIC, + grid_import_cost NUMERIC, + battery_soc NUMERIC, + weather_symbol VARCHAR(100), PRIMARY KEY (timestamp, device_id, metric_name) ); @@ -12,9 +22,30 @@ CREATE INDEX idx_metrics_time ON metrics USING BRIN (timestamp); -- Supported metric_name values: -- - 'pv_total_power': Total PV power in kWh for the integration period +-- - 'energy_flow': Combined energy flow metrics for the integration period --- cloud_coverage column stores the cloud area fraction percentage (0-100) from weather API +-- Column descriptions: +-- pv_total_power: Total PV power in kWh +-- cloud_coverage: Cloud area fraction percentage (0-100) from weather API +-- grid_export_power: Total power exported to grid in kWh +-- grid_import_power: Total power imported from grid in kWh +-- battery_charge_power: Total power charged to battery in kWh +-- battery_discharge_power: Total power discharged from battery in kWh +-- evdc_charge_power: Total power charged to EV DC charger in kWh +-- load_power: Total power consumed by load in kWh +-- grid_export_cost: Total cost of power exported to grid in EUR +-- grid_import_cost: Total cost of power imported from grid in EUR +-- battery_soc: Battery state of charge in % (snapshot at end of period) +-- weather_symbol: Weather condition symbol code (e.g., 'clearsky_day', 'rain', etc.) -- Migration for existing tables: --- ALTER TABLE metrics ADD COLUMN cloud_coverage NUMERIC; --- ALTER TABLE metrics RENAME COLUMN value TO pv_total_power; +-- ALTER TABLE metrics ADD COLUMN grid_export_power NUMERIC; +-- ALTER TABLE metrics ADD COLUMN grid_import_power NUMERIC; +-- ALTER TABLE metrics ADD COLUMN battery_charge_power NUMERIC; +-- ALTER TABLE metrics ADD COLUMN battery_discharge_power NUMERIC; +-- ALTER TABLE metrics ADD COLUMN evdc_charge_power NUMERIC; +-- ALTER TABLE metrics ADD COLUMN load_power NUMERIC; +-- ALTER TABLE metrics ADD COLUMN grid_export_cost NUMERIC; +-- ALTER TABLE metrics ADD COLUMN grid_import_cost NUMERIC; +-- ALTER TABLE metrics ADD COLUMN battery_soc NUMERIC; +-- ALTER TABLE metrics ADD COLUMN weather_symbol VARCHAR(100);