feat: add 1min default task retry interval (#12)

Author: gmlexx

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()

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)

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) {

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
 }

scheduler/mpc.go

@@ -13,22 +13,22 @@ 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()
 
 	// 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,12 +588,13 @@ 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
 	s.lastExecutedDecision = currentDecision
 	s.mu.Unlock()
 
 	s.logger.Printf("Successfully executed MPC decision")
+	return nil
 }