diff --git a/main.go b/main.go index 7ed03cc..edde650 100644 --- a/main.go +++ b/main.go @@ -108,11 +108,16 @@ func runMPCOptimize(config *scheduler.Config) { ctx := context.Background() - minerScheduler.RunMinerDiscovery(ctx) + if err := minerScheduler.RunMinerDiscovery(ctx); err != nil { + logger.Printf("Error during miner discovery: %v", err) + } // Run MPC optimization logger.Printf("Running MPC optimization...") - minerScheduler.RunMPCOptimize(ctx) + if err := minerScheduler.RunMPCOptimize(ctx); err != nil { + logger.Printf("Error during MPC optimization: %v", err) + return + } // Get and log all decisions decisions := minerScheduler.GetMPCDecisions() diff --git a/mpc/mpc_test.go b/mpc/mpc_test.go index 3a29a64..7d92cbb 100644 --- a/mpc/mpc_test.go +++ b/mpc/mpc_test.go @@ -452,7 +452,7 @@ func TestOptimize(t *testing.T) { BatteryMinSOC: 0.0, // 0% minimum SOC BatteryMaxSOC: 1.0, // 100% maximum SOC BatteryEfficiency: 0.9, // 90% round-trip efficiency - BatteryDegradationCost: 0.0, // €0.00/kWh degradation cost + BatteryDegradationCost: 0.0, // 0.00/kWh degradation cost MaxGridImport: 30.0, // 30 kW max grid import MaxGridExport: 30.0, // 30 kW max grid export } @@ -474,7 +474,7 @@ func TestOptimize(t *testing.T) { // Verify all decisions have valid data for i, dec := range decisions { - t.Logf("Hour %2d: SOC=%.3f, Charge=%.3f, Discharge=%.3f, Import=%.3f, Export=%.3f, Profit=€%.4f, Price=€%.5f/kWh", + t.Logf("Hour %2d: SOC=%.3f, Charge=%.3f, Discharge=%.3f, Import=%.3f, Export=%.3f, Profit=%.4f, Price=%.5f/kWh", i, dec.BatterySOC, dec.BatteryCharge, dec.BatteryDischarge, dec.GridImport, dec.GridExport, dec.Profit, dec.ImportPrice) @@ -527,7 +527,7 @@ func TestOptimize(t *testing.T) { for _, dec := range decisions { totalProfit += dec.Profit } - t.Logf("Total profit over 36 hours: €%.4f", totalProfit) + t.Logf("Total profit over 36 hours: %.4f", totalProfit) // Detailed arbitrage analysis t.Log("\n=== Arbitrage Analysis ===") @@ -543,17 +543,17 @@ func TestOptimize(t *testing.T) { exportPriceHour22 := forecast[22].ExportPrice t.Logf("\nHour 4 (LOWEST PRICE):") - t.Logf(" Import Price: €%.5f/kWh", importPriceHour4) - t.Logf(" Export Price: €%.5f/kWh", exportPriceHour4) + t.Logf(" Import Price: %.5f/kWh", importPriceHour4) + t.Logf(" Export Price: %.5f/kWh", exportPriceHour4) t.Logf(" Charged: %.3f kW (added %.3f kWh to battery)", hour4.BatteryCharge, hour4.BatteryCharge*0.9) - t.Logf(" Cost: €%.4f (import + degradation)", hour4.BatteryCharge*importPriceHour4+hour4.BatteryCharge*config.BatteryDegradationCost) + t.Logf(" Cost: %.4f (import + degradation)", hour4.BatteryCharge*importPriceHour4+hour4.BatteryCharge*config.BatteryDegradationCost) t.Logf("\nHour 22 (HIGHEST PRICE):") - t.Logf(" Import Price: €%.5f/kWh", importPriceHour22) - t.Logf(" Export Price: €%.5f/kWh", exportPriceHour22) + t.Logf(" Import Price: %.5f/kWh", importPriceHour22) + t.Logf(" Export Price: %.5f/kWh", exportPriceHour22) t.Logf(" Discharged: %.3f kW (%.3f kWh from battery)", hour22.BatteryDischarge, hour22.BatteryDischarge) t.Logf(" Exported: %.3f kW", hour22.GridExport) - t.Logf(" Revenue: €%.4f (export - degradation)", hour22.GridExport*exportPriceHour22-hour22.BatteryDischarge*config.BatteryDegradationCost) + t.Logf(" Revenue: %.4f (export - degradation)", hour22.GridExport*exportPriceHour22-hour22.BatteryDischarge*config.BatteryDegradationCost) // Calculate theoretical max arbitrage if charged more at hour 4 maxAdditionalCharge := config.BatteryMaxCharge - hour4.BatteryCharge @@ -564,9 +564,9 @@ func TestOptimize(t *testing.T) { t.Logf("\nTheoretical Additional Arbitrage Opportunity:") t.Logf(" Could charge additional: %.3f kW at hour 4", actualMaxCharge) t.Logf(" Energy stored after efficiency: %.3f kWh", actualMaxCharge*config.BatteryEfficiency) - t.Logf(" Cost to charge: €%.4f", actualMaxCharge*importPriceHour4+actualMaxCharge*config.BatteryDegradationCost) - t.Logf(" Revenue if exported at hour 22: €%.4f", actualMaxCharge*config.BatteryEfficiency*exportPriceHour22-actualMaxCharge*config.BatteryDegradationCost) - t.Logf(" Net profit from additional arbitrage: €%.4f", + t.Logf(" Cost to charge: %.4f", actualMaxCharge*importPriceHour4+actualMaxCharge*config.BatteryDegradationCost) + t.Logf(" Revenue if exported at hour 22: %.4f", actualMaxCharge*config.BatteryEfficiency*exportPriceHour22-actualMaxCharge*config.BatteryDegradationCost) + t.Logf(" Net profit from additional arbitrage: %.4f", actualMaxCharge*config.BatteryEfficiency*exportPriceHour22- actualMaxCharge*importPriceHour4- 2*actualMaxCharge*config.BatteryDegradationCost) @@ -579,7 +579,7 @@ func TestOptimize(t *testing.T) { for i, dec := range decisions { if dec.BatteryCharge > 0.1 { chargedTotal += dec.BatteryCharge - t.Logf(" Hour %2d: CHARGED %.3f kW at price €%.5f/kWh (SOC: %.3f)", + t.Logf(" Hour %2d: CHARGED %.3f kW at price %.5f/kWh (SOC: %.3f)", i, dec.BatteryCharge, dec.ImportPrice, dec.BatterySOC) } if dec.BatteryDischarge > 0.1 { @@ -588,12 +588,12 @@ func TestOptimize(t *testing.T) { if dec.GridExport > 0.001 { exported = fmt.Sprintf(" → EXPORTED %.3f kW", dec.GridExport) } - t.Logf(" Hour %2d: DISCHARGED %.3f kW at price €%.5f/kWh (SOC: %.3f)%s", + t.Logf(" Hour %2d: DISCHARGED %.3f kW at price %.5f/kWh (SOC: %.3f)%s", i, dec.BatteryDischarge, dec.ExportPrice, dec.BatterySOC, exported) } } t.Logf("\nTotal charged: %.3f kW, Total discharged: %.3f kW", chargedTotal, dischargedTotal) - t.Logf("Degradation cost: €%.4f (%.3f kW throughput × €%.2f/kWh)", + t.Logf("Degradation cost: %.4f (%.3f kW throughput × %.2f/kWh)", (chargedTotal+dischargedTotal)*config.BatteryDegradationCost, chargedTotal+dischargedTotal, config.BatteryDegradationCost) @@ -617,8 +617,8 @@ func TestOptimize(t *testing.T) { } } - t.Logf("Lowest price: €%.5f/kWh at hour %d", minPrice, minPriceHour) - t.Logf("Highest price: €%.5f/kWh at hour %d", maxPrice, maxPriceHour) + t.Logf("Lowest price: %.5f/kWh at hour %d", minPrice, minPriceHour) + t.Logf("Highest price: %.5f/kWh at hour %d", maxPrice, maxPriceHour) // Check that battery tends to charge during low prices // (Not strict since it depends on SOC constraints and overall optimization) diff --git a/scheduler/data.go b/scheduler/data.go index ded1aae..a107078 100644 --- a/scheduler/data.go +++ b/scheduler/data.go @@ -142,20 +142,20 @@ func (d *DataSamples) GetLatestPower() float64 { return d.samples[len(d.samples)-1].pvPower } -func (s *MinerScheduler) runDataPoll(samples *DataSamples) { +func (s *MinerScheduler) runDataPoll(samples *DataSamples) error { if s.config.PlantModbusAddress == "" { - return + return nil } 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 + return err } defer client.Close() info, err := client.ReadPlantRunningInfo() if err != nil { s.logger.Printf("Data integration: failed to read PlantRunningInfo: %v", err) - return + return err } samples.AddSample( info.PhotovoltaicPower, @@ -165,19 +165,20 @@ func (s *MinerScheduler) runDataPoll(samples *DataSamples) { info.ESSSOC, time.Now(), ) + return nil } -func (s *MinerScheduler) runDataIntegration(samples *DataSamples, pollInterval time.Duration, dataDB *sql.DB, deviceID int, dryRun bool) { +func (s *MinerScheduler) runDataIntegration(samples *DataSamples, pollInterval time.Duration, dataDB *sql.DB, deviceID int, dryRun bool) error { if samples.IsEmpty() { s.logger.Printf("Data integration: no samples collected in period") - return + return nil } data := samples.IntegrateSamples(pollInterval) timestamp := time.Now() if dataDB == nil { - return + return nil } // Fetch weather data from meteo API @@ -205,18 +206,18 @@ func (s *MinerScheduler) runDataIntegration(samples *DataSamples, pollInterval t 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 + gridImportCost = (importPricePerMWh / 1000.0) * data.gridImportPower // Export revenue (negative cost): (spot price - operator fee) * energy in MWh exportPricePerMWh := spotPrice - config.ExportPriceOperatorFee - gridExportCost = (exportPricePerMWh / 1000.0) * data.gridExportPower // Convert to EUR + gridExportCost = (exportPricePerMWh / 1000.0) * data.gridExportPower } } 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)", + 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) @@ -247,16 +248,17 @@ func (s *MinerScheduler) runDataIntegration(samples *DataSamples, pollInterval t ) if err != nil { s.logger.Printf("Data integration: failed to insert metrics: %v", err) - return + return nil } 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)", + 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) } + return nil } func (s *MinerScheduler) fetchCloudCoverage() (*float64, error) { diff --git a/scheduler/miners.go b/scheduler/miners.go index 100ee27..86ebf2d 100644 --- a/scheduler/miners.go +++ b/scheduler/miners.go @@ -42,15 +42,16 @@ func (s *MinerScheduler) discoverMiners(ctx context.Context) error { } // RunMinerDiscovery runs the miner discovery process as a scheduled task -func (s *MinerScheduler) RunMinerDiscovery(ctx context.Context) { +func (s *MinerScheduler) RunMinerDiscovery(ctx context.Context) error { s.logger.Printf("Starting miner discovery task at %s", time.Now().Format(time.RFC3339)) if err := s.discoverMiners(ctx); err != nil { s.logger.Printf("Error discovering miners: %v", err) - return + return err } s.logger.Printf("Miner discovery task completed successfully") + return nil } // getMinerPowerConsumption returns the power consumption in kW for a given miner state and work mode @@ -301,10 +302,10 @@ func (s *MinerScheduler) controlMiner(m *miners.AvalonQHost, totalPower float64, } // runStateCheck executes the state monitoring task for miners -func (s *MinerScheduler) runStateCheck(ctx context.Context) { +func (s *MinerScheduler) runStateCheck(ctx context.Context) error { minersList := s.refreshMinersState(ctx) if len(minersList) == 0 { - return + return nil } isDryRun := s.config.DryRun @@ -409,5 +410,7 @@ func (s *MinerScheduler) runStateCheck(ctx context.Context) { for _, err := range errors { s.logger.Printf(" - %v", err) } + return fmt.Errorf("errors during state check") } + return nil } diff --git a/scheduler/mpc.go b/scheduler/mpc.go index 231cb66..c8ce738 100644 --- a/scheduler/mpc.go +++ b/scheduler/mpc.go @@ -13,7 +13,7 @@ import ( ) // RunMPCOptimize executes the MPC optimization task -func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { +func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) error { s.logger.Printf("Starting MPC optimization task at %s", time.Now().Format(time.RFC3339)) config := s.GetConfig() @@ -21,14 +21,14 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { // Check if Plant Modbus Address is configured if config.PlantModbusAddress == "" { s.logger.Printf("MPC optimization skipped: PlantModbusAddress not configured") - return + return nil } // Step 1: Read plant running info from inverter plantInfo, err := s.readPlantRunningInfo(config) if err != nil { s.logger.Printf("Error reading plant running info from inverter: %v", err) - return + return err } // Extract initial SOC from plant info @@ -39,12 +39,12 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { forecast, err := s.buildMPCForecast(ctx, config, plantInfo) if err != nil { s.logger.Printf("Error building MPC forecast: %v", err) - return + return err } if len(forecast) == 0 { s.logger.Printf("No forecast data available for MPC optimization") - return + return nil } s.logger.Printf("Built forecast with %d time slots", len(forecast)) @@ -69,7 +69,7 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { decisions := controller.Optimize(forecast) if len(decisions) == 0 { s.logger.Printf("MPC optimization produced no decisions") - return + return nil } // Step 5: Save optimization results to memory @@ -110,10 +110,11 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { if err != nil { s.logger.Printf("Error executing MPC decision: %v (will retry every minute)", err) - return + return err } s.logger.Printf("MPC optimization task completed successfully") + return nil } // readPlantRunningInfo reads the plant running information from the inverter @@ -536,21 +537,15 @@ func (s *MinerScheduler) executeMPCDecision(decision *mpc.ControlDecision, dryRu // runMPCExecution re-executes the current MPC decision only if previous execution failed // This ensures the decision is applied even if previous execution failed -func (s *MinerScheduler) runMPCExecution() { +func (s *MinerScheduler) runMPCExecution() error { s.mu.RLock() config := s.GetConfig() - // Check if Plant Modbus Address is configured - if config.PlantModbusAddress == "" { + // Check if Plant Modbus Address is configured and there are decisions + if config.PlantModbusAddress == "" || len(s.mpcDecisions) == 0 { s.mu.RUnlock() - return - } - - // Get current MPC decision based on current timestamp - if len(s.mpcDecisions) == 0 { - s.mu.RUnlock() - return + return nil } now := time.Now().Unix() @@ -570,7 +565,7 @@ func (s *MinerScheduler) runMPCExecution() { if currentDecision == nil { // No matching decision found for current timestamp s.mu.RUnlock() - return + return nil } lastExecuted := s.lastExecutedDecision @@ -579,7 +574,7 @@ func (s *MinerScheduler) runMPCExecution() { // Check if this decision has already been executed if lastExecuted != nil && currentDecision.Timestamp == lastExecuted.Timestamp { // Decision already executed, no need to retry - return + return nil } s.logger.Printf("Executing MPC decision for timestamp %d (hour %d)", currentDecision.Timestamp, currentDecision.Hour) @@ -593,7 +588,7 @@ func (s *MinerScheduler) runMPCExecution() { s.lastExecutedDecision = nil s.mu.Unlock() s.logger.Printf("Error executing MPC decision: %v (will retry again in 1 minute)", err) - return + return err } // Execution succeeded, store the executed decision @@ -601,4 +596,5 @@ func (s *MinerScheduler) runMPCExecution() { s.mu.Unlock() s.logger.Printf("Successfully executed MPC decision") + return nil } diff --git a/scheduler/pricing.go b/scheduler/pricing.go index 8f3f9f3..219de68 100644 --- a/scheduler/pricing.go +++ b/scheduler/pricing.go @@ -67,14 +67,14 @@ func (s *MinerScheduler) GetMarketData(ctx context.Context) (*entsoe.Publication } // runPriceCheck executes the main scheduler task -func (s *MinerScheduler) runPriceCheck(ctx context.Context) { +func (s *MinerScheduler) runPriceCheck(ctx context.Context) error { s.logger.Printf("Starting price check task at %s", time.Now().Format(time.RFC3339)) // Step 1: Get current electricity price currentPrice, err := s.getCurrentAvgPrice(ctx) if err != nil { s.logger.Printf("Error getting current price: %v", err) - return + return err } s.logger.Printf("Current hourly average electricity price: %.2f EUR/MWh", currentPrice) @@ -83,10 +83,11 @@ func (s *MinerScheduler) runPriceCheck(ctx context.Context) { // Step 2: Manage miners based on price if err := s.manageMiners(ctx, currentPrice); err != nil { s.logger.Printf("Error managing miners: %v", err) - return + return err } s.logger.Printf("Price check task completed successfully") + return nil } // getCurrentAvgPrice gets the current hourly average electricity price, downloading new data if needed diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index 8856672..5a83b73 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -16,10 +16,12 @@ import ( // 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() + name string + initialDelay time.Duration + interval time.Duration + runFunc func() error + retryInterval *time.Duration + err error } // run executes the periodic task in a loop, respecting the initial delay and context cancellation @@ -31,7 +33,7 @@ func (pt *PeriodicTask) run(ctx context.Context, stopChan <-chan struct{}, logge case <-time.After(pt.initialDelay): // Initial delay passed, run the task logger.Printf("[%s] Initial delay passed, running first iteration", pt.name) - pt.runFunc() + pt.err = pt.runFunc() case <-ctx.Done(): logger.Printf("[%s] Stopped during initial delay due to context cancellation", pt.name) return @@ -42,19 +44,31 @@ func (pt *PeriodicTask) run(ctx context.Context, stopChan <-chan struct{}, logge } else { // No initial delay, run immediately logger.Printf("[%s] Running immediately (no initial delay)", pt.name) - pt.runFunc() + pt.err = pt.runFunc() } // Create ticker for periodic execution ticker := time.NewTicker(pt.interval) defer ticker.Stop() + // Default value, but the retry is disabled if the pt.retryInterval is nil + retryInterval := time.Hour + if pt.retryInterval != nil { + retryInterval = *pt.retryInterval + } + retryTicker := time.NewTicker(retryInterval) + defer retryTicker.Stop() + logger.Printf("[%s] Started with interval: %v", pt.name, pt.interval) for { select { case <-ticker.C: - pt.runFunc() + pt.err = pt.runFunc() + case <-retryTicker.C: + if pt.retryInterval != nil && pt.err != nil { + pt.err = pt.runFunc() + } case <-ctx.Done(): logger.Printf("[%s] Stopped due to context cancellation", pt.name) return @@ -223,55 +237,67 @@ func (s *MinerScheduler) Start(ctx context.Context, serverOnly bool) error { stateCheckInitialDelay := s.getInitialDelay(now, config.MinersStateCheckInterval) mpcExecutionInitialDelay := s.getInitialDelay(now, config.MPCExecutionInterval) + 2*time.Second + taskRetryInterval := time.Minute + // Create periodic tasks tasks := []PeriodicTask{ { name: "MinerDiscovery", initialDelay: 0, // Run immediately interval: config.MinerDiscoveryInterval, - runFunc: func() { - s.RunMinerDiscovery(ctx) + runFunc: func() error { + return s.RunMinerDiscovery(ctx) }, }, { - name: "PriceCheckAndMPC", - initialDelay: minersControlInitialDelay, - interval: config.CheckPriceInterval, - runFunc: func() { - s.runPriceCheck(ctx) - s.RunMPCOptimize(ctx) + name: "PriceCheck", + initialDelay: minersControlInitialDelay, + interval: config.CheckPriceInterval, + retryInterval: &taskRetryInterval, + runFunc: func() error { + return s.runPriceCheck(ctx) + }, + }, + { + name: "MPC", + initialDelay: minersControlInitialDelay, + interval: config.CheckPriceInterval, + retryInterval: &taskRetryInterval, + runFunc: func() error { + return s.RunMPCOptimize(ctx) }, }, { name: "StateCheck", initialDelay: stateCheckInitialDelay, interval: config.MinersStateCheckInterval, - runFunc: func() { - s.runStateCheck(ctx) + runFunc: func() error { + return s.runStateCheck(ctx) }, }, { name: "DataPoll", initialDelay: pvDataInitialDelay, interval: config.PVPollInterval, - runFunc: func() { - s.runDataPoll(dataSamples) + runFunc: func() error { + return s.runDataPoll(dataSamples) }, }, { - name: "DataIntegration", - initialDelay: pvDataInitialDelay, - interval: config.PVIntegrationPeriod, - runFunc: func() { - s.runDataIntegration(dataSamples, config.PVPollInterval, dataDB, config.DeviceID, config.DryRun) + name: "DataIntegration", + initialDelay: pvDataInitialDelay, + interval: config.PVIntegrationPeriod, + retryInterval: &taskRetryInterval, + runFunc: func() error { + return s.runDataIntegration(dataSamples, config.PVPollInterval, dataDB, config.DeviceID, config.DryRun) }, }, { name: "MPCExecution", initialDelay: mpcExecutionInitialDelay, interval: config.MPCExecutionInterval, - runFunc: func() { - s.runMPCExecution() + runFunc: func() error { + return s.runMPCExecution() }, }, }