From 9610461167642d912fc7e37bc33329f20d9112d1 Mon Sep 17 00:00:00 2001 From: Aleksei Larkov Date: Wed, 7 Jan 2026 14:24:57 +0200 Subject: [PATCH 1/3] fix: mpc retry logic --- entsoe/api_client.go | 4 +-- scheduler/mpc.go | 67 ++++++++++++++++++++++++++++++------------ scheduler/pricing.go | 6 ++-- scheduler/scheduler.go | 4 +-- 4 files changed, 55 insertions(+), 26 deletions(-) diff --git a/entsoe/api_client.go b/entsoe/api_client.go index c226634..d7d641e 100644 --- a/entsoe/api_client.go +++ b/entsoe/api_client.go @@ -57,8 +57,8 @@ func DownloadPublicationMarketData(ctx context.Context, securityToken string, ur return nil, err } - // If current time is >= 13:00, also download data for the next day - if now.Hour() >= 13 { + // If current time is >= 14:00, also download data for the next day + if now.Hour() >= 14 { tomorrow := now.AddDate(0, 0, 1) urlNextDay := buildPublicationMarketDataURL(securityToken, urlFormat, tomorrow) diff --git a/scheduler/mpc.go b/scheduler/mpc.go index 0490912..a5bff1a 100644 --- a/scheduler/mpc.go +++ b/scheduler/mpc.go @@ -75,7 +75,7 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { // Step 5: Save optimization results s.mu.Lock() s.mpcDecisions = decisions - s.lastMPCExecutionError = nil // Clear any previous execution error + s.lastExecutedDecision = nil // Clear last executed decision for new optimization s.mu.Unlock() // Log summary @@ -91,7 +91,13 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { // Record execution status s.mu.Lock() - s.lastMPCExecutionError = err + if err != nil { + // Execution failed, set lastExecutedDecision to nil + s.lastExecutedDecision = nil + } else { + // Execution succeeded, store the executed decision + s.lastExecutedDecision = &decisions[0] + } s.mu.Unlock() if err != nil { @@ -524,15 +530,7 @@ func (s *MinerScheduler) executeMPCDecision(decision *mpc.ControlDecision, dryRu // This ensures the decision is applied even if previous execution failed func (s *MinerScheduler) runMPCExecution() { - // Check if last execution failed s.mu.RLock() - lastError := s.lastMPCExecutionError - if lastError == nil { - // Last execution succeeded, no need to retry - s.mu.RUnlock() - return - } - config := s.GetConfig() // Check if Plant Modbus Address is configured @@ -541,27 +539,58 @@ func (s *MinerScheduler) runMPCExecution() { return } - // Get current MPC decision + // Get current MPC decision based on current timestamp if len(s.mpcDecisions) == 0 { s.mu.RUnlock() return } - currentDecision := s.mpcDecisions[0] + + now := time.Now().Unix() + var currentDecision *mpc.ControlDecision + + // Find the decision that matches the current hour + for i := range s.mpcDecisions { + decision := &s.mpcDecisions[i] + // Check if current time falls within this decision's hour + // Each decision covers a 1-hour window starting from its timestamp + if now >= decision.Timestamp && now < decision.Timestamp+3600 { + currentDecision = decision + break + } + } + + if currentDecision == nil { + // No matching decision found for current timestamp + s.mu.RUnlock() + return + } + + lastExecuted := s.lastExecutedDecision s.mu.RUnlock() - s.logger.Printf("Retrying MPC decision execution (previous error: %v)", lastError) + // Check if this decision has already been executed + if lastExecuted != nil && currentDecision.Timestamp == lastExecuted.Timestamp { + // Decision already executed, no need to retry + return + } + + s.logger.Printf("Executing MPC decision for timestamp %d (hour %d)", currentDecision.Timestamp, currentDecision.Hour) // Execute the current decision - err := s.executeMPCDecision(¤tDecision, config.DryRun) + err := s.executeMPCDecision(currentDecision, config.DryRun) s.mu.Lock() - s.lastMPCExecutionError = err - s.mu.Unlock() - if err != nil { - s.logger.Printf("Error retrying MPC decision: %v (will retry again in 1 minute)", err) + // Execution failed, set lastExecutedDecision to nil + s.lastExecutedDecision = nil + s.mu.Unlock() + s.logger.Printf("Error executing MPC decision: %v (will retry again in 1 minute)", err) return } - s.logger.Printf("Successfully retried MPC decision execution") + // Execution succeeded, store the executed decision + s.lastExecutedDecision = currentDecision + s.mu.Unlock() + + s.logger.Printf("Successfully executed MPC decision") } diff --git a/scheduler/pricing.go b/scheduler/pricing.go index 45781d4..29875d1 100644 --- a/scheduler/pricing.go +++ b/scheduler/pricing.go @@ -50,10 +50,10 @@ func (s *MinerScheduler) GetMarketData(ctx context.Context) (*entsoe.Publication } // Calculate next expiry time at 13:00 - nextExpiry := time.Date(now.Year(), now.Month(), now.Day(), 13, 0, 0, 0, location) + nextExpiry := time.Date(now.Year(), now.Month(), now.Day(), 14, 0, 0, 0, location) - // If it's already past 13:00 today, set expiry to 13:00 tomorrow - if now.Hour() >= 13 { + // If it's already past 14:00 today, set expiry to 14:00 tomorrow + if now.Hour() >= 14 { nextExpiry = nextExpiry.Add(24 * time.Hour) } diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index acd71ec..e5e3810 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -81,8 +81,8 @@ type MinerScheduler struct { weatherCache WeatherForecastCache // MPC optimization results - mpcDecisions []mpc.ControlDecision - lastMPCExecutionError error // Tracks if last MPC execution failed + mpcDecisions []mpc.ControlDecision + lastExecutedDecision *mpc.ControlDecision // Tracks the last successfully executed decision // Web server webServer *WebServer From bdf79c482888b0eec0cddddbc6fdd5f2ffcafa49 Mon Sep 17 00:00:00 2001 From: Aleksei Larkov Date: Wed, 7 Jan 2026 15:14:55 +0200 Subject: [PATCH 2/3] fix: prices cache expiration --- scheduler/pricing.go | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/scheduler/pricing.go b/scheduler/pricing.go index 29875d1..81ea536 100644 --- a/scheduler/pricing.go +++ b/scheduler/pricing.go @@ -17,7 +17,13 @@ func (s *MinerScheduler) GetPricesMarketData() *entsoe.PublicationMarketData { // GetMarketData returns the latest PublicationMarketData, downloading new data if needed func (s *MinerScheduler) GetMarketData(ctx context.Context) (*entsoe.PublicationMarketData, error) { - now := time.Now() + + location, err := time.LoadLocation(s.config.Location) + if err != nil { + return nil, err + } + + now := time.Now().In(location) s.mu.RLock() marketData := s.pricesMarketData @@ -36,11 +42,6 @@ func (s *MinerScheduler) GetMarketData(ctx context.Context) (*entsoe.Publication s.logger.Printf("No cached pricing data available, downloading new PublicationMarketData...") } - location, err := time.LoadLocation(s.config.Location) - if err != nil { - return nil, err - } - s.mu.Lock() defer s.mu.Unlock() @@ -49,7 +50,7 @@ func (s *MinerScheduler) GetMarketData(ctx context.Context) (*entsoe.Publication return nil, fmt.Errorf("failed to download PublicationMarketData: %w", err) } - // Calculate next expiry time at 13:00 + // Calculate next expiry time at 14:00 nextExpiry := time.Date(now.Year(), now.Month(), now.Day(), 14, 0, 0, 0, location) // If it's already past 14:00 today, set expiry to 14:00 tomorrow From 1ef2070f7fed1a4145526322183c4e25a3f26229 Mon Sep 17 00:00:00 2001 From: Aleksei Larkov Date: Wed, 7 Jan 2026 21:40:49 +0200 Subject: [PATCH 3/3] feat: persist mpc decisions in DB --- scheduler/mpc.go | 10 +- scheduler/mpc_persistence.go | 209 ++++++++++++++++++ scheduler/mpc_persistence_test.go | 343 ++++++++++++++++++++++++++++++ scheduler/scheduler.go | 10 + sql/README.md | 174 +++++++++++++++ {test_data => sql}/metrics.sql | 0 sql/mpc_decisions.sql | 32 +++ 7 files changed, 777 insertions(+), 1 deletion(-) create mode 100644 scheduler/mpc_persistence.go create mode 100644 scheduler/mpc_persistence_test.go create mode 100644 sql/README.md rename {test_data => sql}/metrics.sql (100%) create mode 100644 sql/mpc_decisions.sql diff --git a/scheduler/mpc.go b/scheduler/mpc.go index a5bff1a..5f808fe 100644 --- a/scheduler/mpc.go +++ b/scheduler/mpc.go @@ -72,12 +72,20 @@ func (s *MinerScheduler) RunMPCOptimize(ctx context.Context) { return } - // Step 5: Save optimization results + // Step 5: Save optimization results to memory s.mu.Lock() s.mpcDecisions = decisions s.lastExecutedDecision = nil // Clear last executed decision for new optimization s.mu.Unlock() + // Step 5.1: Persist decisions to database (only when not in dry run mode) + if !config.DryRun { + if err := s.saveMPCDecisions(ctx, decisions); err != nil { + s.logger.Printf("Warning: Failed to save MPC decisions to database: %v", err) + // Continue execution even if persistence fails + } + } + // Log summary s.logger.Printf("MPC optimization completed with %d decisions", len(decisions)) totalProfit := 0.0 diff --git a/scheduler/mpc_persistence.go b/scheduler/mpc_persistence.go new file mode 100644 index 0000000..b49e38f --- /dev/null +++ b/scheduler/mpc_persistence.go @@ -0,0 +1,209 @@ +package scheduler + +import ( + "context" + "database/sql" + "fmt" + "time" + + "github.com/devskill-org/ems/mpc" +) + +// saveMPCDecisions persists MPC decisions to the database +func (s *MinerScheduler) saveMPCDecisions(ctx context.Context, decisions []mpc.ControlDecision) error { + if s.db == nil { + return fmt.Errorf("database connection not available") + } + + if len(decisions) == 0 { + return nil + } + + // Use first decision timestamp as minimum + // Decisions are ordered by timestamp because: + // 1. MPC forecast is built from a map and explicitly sorted by hour + // 2. MPC controller reconstructs path in the same order as forecast + // 3. Timestamp increases monotonically with hour + minTimestamp := decisions[0].Timestamp + + // Begin transaction + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback() + + // Delete existing decisions with timestamp >= minTimestamp + _, err = tx.ExecContext(ctx, `DELETE FROM mpc_decisions WHERE timestamp >= $1`, minTimestamp) + if err != nil { + return fmt.Errorf("failed to delete existing decisions: %w", err) + } + + // Prepare upsert statement + stmt, err := tx.PrepareContext(ctx, ` + INSERT INTO mpc_decisions ( + timestamp, + hour, + battery_charge, + battery_discharge, + grid_import, + grid_export, + battery_soc, + profit, + import_price, + export_price, + solar_forecast, + load_forecast, + cloud_coverage, + weather_symbol + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) + ON CONFLICT (timestamp) DO UPDATE SET + hour = EXCLUDED.hour, + battery_charge = EXCLUDED.battery_charge, + battery_discharge = EXCLUDED.battery_discharge, + grid_import = EXCLUDED.grid_import, + grid_export = EXCLUDED.grid_export, + battery_soc = EXCLUDED.battery_soc, + profit = EXCLUDED.profit, + import_price = EXCLUDED.import_price, + export_price = EXCLUDED.export_price, + solar_forecast = EXCLUDED.solar_forecast, + load_forecast = EXCLUDED.load_forecast, + cloud_coverage = EXCLUDED.cloud_coverage, + weather_symbol = EXCLUDED.weather_symbol + `) + if err != nil { + return fmt.Errorf("failed to prepare statement: %w", err) + } + defer stmt.Close() + + // Insert all decisions + for _, decision := range decisions { + _, err := stmt.ExecContext(ctx, + decision.Timestamp, + decision.Hour, + decision.BatteryCharge, + decision.BatteryDischarge, + decision.GridImport, + decision.GridExport, + decision.BatterySOC, + decision.Profit, + decision.ImportPrice, + decision.ExportPrice, + decision.SolarForecast, + decision.LoadForecast, + decision.CloudCoverage, + decision.WeatherSymbol, + ) + if err != nil { + return fmt.Errorf("failed to insert decision for hour %d: %w", decision.Hour, err) + } + } + + // Commit transaction + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + + s.logger.Printf("Saved %d MPC decisions to database", len(decisions)) + return nil +} + +// loadLatestMPCDecisions loads MPC decisions from the database with timestamp >= now +func (s *MinerScheduler) loadLatestMPCDecisions(ctx context.Context) ([]mpc.ControlDecision, error) { + if s.db == nil { + return nil, fmt.Errorf("database connection not available") + } + + // Get current Unix timestamp + now := ctx.Value("now") + var nowTimestamp int64 + if now != nil { + if t, ok := now.(int64); ok { + nowTimestamp = t + } + } + if nowTimestamp == 0 { + nowTimestamp = s.getCurrentTimestamp() + } + + // Load decisions with timestamp >= now, ordered by timestamp + rows, err := s.db.QueryContext(ctx, ` + SELECT + timestamp, + hour, + battery_charge, + battery_discharge, + grid_import, + grid_export, + battery_soc, + profit, + import_price, + export_price, + solar_forecast, + load_forecast, + cloud_coverage, + weather_symbol + FROM mpc_decisions + WHERE timestamp >= $1 + ORDER BY timestamp ASC + `, nowTimestamp) + if err != nil { + return nil, fmt.Errorf("failed to query decisions: %w", err) + } + defer rows.Close() + + var decisions []mpc.ControlDecision + for rows.Next() { + var decision mpc.ControlDecision + var cloudCoverage sql.NullFloat64 + var weatherSymbol sql.NullString + + err := rows.Scan( + &decision.Timestamp, + &decision.Hour, + &decision.BatteryCharge, + &decision.BatteryDischarge, + &decision.GridImport, + &decision.GridExport, + &decision.BatterySOC, + &decision.Profit, + &decision.ImportPrice, + &decision.ExportPrice, + &decision.SolarForecast, + &decision.LoadForecast, + &cloudCoverage, + &weatherSymbol, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan decision: %w", err) + } + + if cloudCoverage.Valid { + decision.CloudCoverage = cloudCoverage.Float64 + } + if weatherSymbol.Valid { + decision.WeatherSymbol = weatherSymbol.String + } + + decisions = append(decisions, decision) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating decisions: %w", err) + } + + if len(decisions) == 0 { + s.logger.Printf("No future MPC decisions found in database") + return nil, nil + } + + s.logger.Printf("Loaded %d MPC decisions from database (starting from timestamp %d)", len(decisions), nowTimestamp) + + return decisions, nil +} + +// getCurrentTimestamp returns the current Unix timestamp +func (s *MinerScheduler) getCurrentTimestamp() int64 { + return time.Now().Unix() +} diff --git a/scheduler/mpc_persistence_test.go b/scheduler/mpc_persistence_test.go new file mode 100644 index 0000000..f44651f --- /dev/null +++ b/scheduler/mpc_persistence_test.go @@ -0,0 +1,343 @@ +package scheduler + +import ( + "context" + "database/sql" + "log" + "os" + "testing" + "time" + + "github.com/devskill-org/ems/mpc" + _ "github.com/lib/pq" +) + +// TestMPCPersistence_SaveAndLoad tests the save and load cycle +func TestMPCPersistence_SaveAndLoad(t *testing.T) { + // Skip if no database connection available + connString := os.Getenv("TEST_POSTGRES_CONN") + if connString == "" { + t.Skip("Skipping test: TEST_POSTGRES_CONN not set") + } + + db, err := sql.Open("postgres", connString) + if err != nil { + t.Fatalf("Failed to connect to database: %v", err) + } + defer db.Close() + + // Clean up table before test + _, err = db.Exec("DELETE FROM mpc_decisions") + if err != nil { + t.Fatalf("Failed to clean up table: %v", err) + } + + // Create scheduler with database + config := &Config{} + scheduler := &MinerScheduler{ + config: config, + db: db, + logger: log.New(os.Stdout, "TEST: ", log.LstdFlags), + } + + // Create test decisions with timestamps in the future + now := time.Now().Unix() + decisions := []mpc.ControlDecision{ + { + Hour: 0, + Timestamp: now + 3600, + BatteryCharge: 10.5, + BatteryDischarge: 0, + GridImport: 5.0, + GridExport: 0, + BatterySOC: 0.6, + Profit: 2.5, + ImportPrice: 0.1, + ExportPrice: 0.05, + SolarForecast: 15.0, + LoadForecast: 10.0, + CloudCoverage: 30.0, + WeatherSymbol: "clearsky_day", + }, + { + Hour: 1, + Timestamp: now + 7200, + BatteryCharge: 0, + BatteryDischarge: 8.0, + GridImport: 0, + GridExport: 3.0, + BatterySOC: 0.5, + Profit: 3.2, + ImportPrice: 0.12, + ExportPrice: 0.06, + SolarForecast: 20.0, + LoadForecast: 12.0, + CloudCoverage: 10.0, + WeatherSymbol: "fair_day", + }, + } + + ctx := context.Background() + + // Save decisions + err = scheduler.saveMPCDecisions(ctx, decisions) + if err != nil { + t.Fatalf("Failed to save decisions: %v", err) + } + + // Load decisions + loaded, err := scheduler.loadLatestMPCDecisions(ctx) + if err != nil { + t.Fatalf("Failed to load decisions: %v", err) + } + + // Verify loaded decisions + if len(loaded) != len(decisions) { + t.Errorf("Expected %d decisions, got %d", len(decisions), len(loaded)) + } + + for i, decision := range loaded { + if decision.Timestamp != decisions[i].Timestamp { + t.Errorf("Decision %d: expected timestamp %d, got %d", i, decisions[i].Timestamp, decision.Timestamp) + } + if decision.BatteryCharge != decisions[i].BatteryCharge { + t.Errorf("Decision %d: expected battery_charge %.2f, got %.2f", i, decisions[i].BatteryCharge, decision.BatteryCharge) + } + if decision.Profit != decisions[i].Profit { + t.Errorf("Decision %d: expected profit %.2f, got %.2f", i, decisions[i].Profit, decision.Profit) + } + } +} + +// TestMPCPersistence_DeleteOldDecisions tests that old decisions are replaced +func TestMPCPersistence_DeleteOldDecisions(t *testing.T) { + // Skip if no database connection available + connString := os.Getenv("TEST_POSTGRES_CONN") + if connString == "" { + t.Skip("Skipping test: TEST_POSTGRES_CONN not set") + } + + db, err := sql.Open("postgres", connString) + if err != nil { + t.Fatalf("Failed to connect to database: %v", err) + } + defer db.Close() + + // Clean up table before test + _, err = db.Exec("DELETE FROM mpc_decisions") + if err != nil { + t.Fatalf("Failed to clean up table: %v", err) + } + + // Create scheduler with database + config := &Config{} + scheduler := &MinerScheduler{ + config: config, + db: db, + logger: log.New(os.Stdout, "TEST: ", log.LstdFlags), + } + + now := time.Now().Unix() + ctx := context.Background() + + // First, save decisions for hours 0-2 + firstDecisions := []mpc.ControlDecision{ + {Hour: 0, Timestamp: now + 3600, Profit: 1.0}, + {Hour: 1, Timestamp: now + 7200, Profit: 2.0}, + {Hour: 2, Timestamp: now + 10800, Profit: 3.0}, + } + err = scheduler.saveMPCDecisions(ctx, firstDecisions) + if err != nil { + t.Fatalf("Failed to save first decisions: %v", err) + } + + // Then, save new decisions starting from hour 1 (should replace hours 1-2) + secondDecisions := []mpc.ControlDecision{ + {Hour: 1, Timestamp: now + 7200, Profit: 20.0}, // Updated + {Hour: 2, Timestamp: now + 10800, Profit: 30.0}, // Updated + {Hour: 3, Timestamp: now + 14400, Profit: 40.0}, // New + } + err = scheduler.saveMPCDecisions(ctx, secondDecisions) + if err != nil { + t.Fatalf("Failed to save second decisions: %v", err) + } + + // Load all decisions (including past) + var allDecisions []mpc.ControlDecision + rows, err := db.Query("SELECT timestamp, profit FROM mpc_decisions ORDER BY timestamp") + if err != nil { + t.Fatalf("Failed to query decisions: %v", err) + } + defer rows.Close() + + for rows.Next() { + var d mpc.ControlDecision + err := rows.Scan(&d.Timestamp, &d.Profit) + if err != nil { + t.Fatalf("Failed to scan decision: %v", err) + } + allDecisions = append(allDecisions, d) + } + + // Should have 4 decisions: hour 0 (unchanged), hours 1-3 (new/updated) + if len(allDecisions) != 4 { + t.Errorf("Expected 4 decisions, got %d", len(allDecisions)) + } + + // Verify hour 0 is unchanged + if allDecisions[0].Timestamp == now+3600 && allDecisions[0].Profit != 1.0 { + t.Errorf("Hour 0 should be unchanged with profit 1.0, got %.2f", allDecisions[0].Profit) + } + + // Verify hour 1 is updated + if allDecisions[1].Timestamp == now+7200 && allDecisions[1].Profit != 20.0 { + t.Errorf("Hour 1 should be updated with profit 20.0, got %.2f", allDecisions[1].Profit) + } +} + +// TestMPCPersistence_LoadOnlyFutureDecisions tests that only future decisions are loaded +func TestMPCPersistence_LoadOnlyFutureDecisions(t *testing.T) { + // Skip if no database connection available + connString := os.Getenv("TEST_POSTGRES_CONN") + if connString == "" { + t.Skip("Skipping test: TEST_POSTGRES_CONN not set") + } + + db, err := sql.Open("postgres", connString) + if err != nil { + t.Fatalf("Failed to connect to database: %v", err) + } + defer db.Close() + + // Clean up table before test + _, err = db.Exec("DELETE FROM mpc_decisions") + if err != nil { + t.Fatalf("Failed to clean up table: %v", err) + } + + // Create scheduler with database + config := &Config{} + scheduler := &MinerScheduler{ + config: config, + db: db, + logger: log.New(os.Stdout, "TEST: ", log.LstdFlags), + } + + now := time.Now().Unix() + ctx := context.Background() + + // Save decisions: some in the past, some in the future + decisions := []mpc.ControlDecision{ + {Hour: 0, Timestamp: now - 3600, Profit: 1.0}, // Past + {Hour: 1, Timestamp: now - 1800, Profit: 2.0}, // Past + {Hour: 2, Timestamp: now + 1800, Profit: 3.0}, // Future + {Hour: 3, Timestamp: now + 3600, Profit: 4.0}, // Future + {Hour: 4, Timestamp: now + 7200, Profit: 5.0}, // Future + } + + // Insert directly to test load filtering + for _, d := range decisions { + _, err := db.Exec(` + INSERT INTO mpc_decisions (timestamp, hour, battery_charge, battery_discharge, + grid_import, grid_export, battery_soc, profit, import_price, export_price, + solar_forecast, load_forecast) + VALUES ($1, $2, 0, 0, 0, 0, 0.5, $3, 0.1, 0.05, 10, 5) + `, d.Timestamp, d.Hour, d.Profit) + if err != nil { + t.Fatalf("Failed to insert decision: %v", err) + } + } + + // Load decisions (should only get future ones) + loaded, err := scheduler.loadLatestMPCDecisions(ctx) + if err != nil { + t.Fatalf("Failed to load decisions: %v", err) + } + + // Should only load 3 future decisions + if len(loaded) != 3 { + t.Errorf("Expected 3 future decisions, got %d", len(loaded)) + } + + // Verify all loaded decisions are in the future + for i, decision := range loaded { + if decision.Timestamp < now { + t.Errorf("Decision %d has past timestamp %d (now: %d)", i, decision.Timestamp, now) + } + } + + // Verify they are ordered by timestamp + for i := 1; i < len(loaded); i++ { + if loaded[i].Timestamp <= loaded[i-1].Timestamp { + t.Errorf("Decisions not properly ordered by timestamp") + } + } +} + +// TestMPCPersistence_UniqueTimestamp tests that timestamp PRIMARY KEY prevents duplicates +func TestMPCPersistence_UniqueTimestamp(t *testing.T) { + // Skip if no database connection available + connString := os.Getenv("TEST_POSTGRES_CONN") + if connString == "" { + t.Skip("Skipping test: TEST_POSTGRES_CONN not set") + } + + db, err := sql.Open("postgres", connString) + if err != nil { + t.Fatalf("Failed to connect to database: %v", err) + } + defer db.Close() + + // Clean up table before test + _, err = db.Exec("DELETE FROM mpc_decisions") + if err != nil { + t.Fatalf("Failed to clean up table: %v", err) + } + + now := time.Now().Unix() + timestamp := now + 3600 + + // Insert first decision + _, err = db.Exec(` + INSERT INTO mpc_decisions (timestamp, hour, battery_charge, battery_discharge, + grid_import, grid_export, battery_soc, profit, import_price, export_price, + solar_forecast, load_forecast) + VALUES ($1, 0, 10, 0, 5, 0, 0.6, 2.5, 0.1, 0.05, 15, 10) + `, timestamp) + if err != nil { + t.Fatalf("Failed to insert first decision: %v", err) + } + + // Try to insert duplicate timestamp (should be handled by UPSERT in saveMPCDecisions) + _, err = db.Exec(` + INSERT INTO mpc_decisions (timestamp, hour, battery_charge, battery_discharge, + grid_import, grid_export, battery_soc, profit, import_price, export_price, + solar_forecast, load_forecast) + VALUES ($1, 1, 20, 0, 10, 0, 0.7, 5.0, 0.12, 0.06, 20, 12) + ON CONFLICT (timestamp) DO UPDATE SET + hour = EXCLUDED.hour, + profit = EXCLUDED.profit + `, timestamp) + if err != nil { + t.Fatalf("UPSERT failed: %v", err) + } + + // Verify only one row exists and it's updated + var count int + var profit float64 + var hour int + err = db.QueryRow("SELECT COUNT(*), MAX(hour), MAX(profit) FROM mpc_decisions WHERE timestamp = $1", timestamp).Scan(&count, &hour, &profit) + if err != nil { + t.Fatalf("Failed to query: %v", err) + } + + if count != 1 { + t.Errorf("Expected 1 row, got %d", count) + } + if hour != 1 { + t.Errorf("Expected hour to be updated to 1, got %d", hour) + } + if profit != 5.0 { + t.Errorf("Expected profit to be updated to 5.0, got %.2f", profit) + } +} diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index e5e3810..de991e8 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -202,6 +202,16 @@ func (s *MinerScheduler) Start(ctx context.Context, serverOnly bool) error { dataDB = nil } else { s.db = dataDB + + // Load latest MPC decisions from database + if decisions, err := s.loadLatestMPCDecisions(ctx); err != nil { + s.logger.Printf("Warning: Failed to load MPC decisions from database: %v", err) + } else if len(decisions) > 0 { + s.mu.Lock() + s.mpcDecisions = decisions + s.mu.Unlock() + s.logger.Printf("Loaded %d MPC decisions from database on startup", len(decisions)) + } } } diff --git a/sql/README.md b/sql/README.md new file mode 100644 index 0000000..0e86e5c --- /dev/null +++ b/sql/README.md @@ -0,0 +1,174 @@ +# SQL Database Setup + +This directory contains SQL schema files for the EMS (Energy Management System) application. + +## Prerequisites + +- PostgreSQL database (version 9.5+) +- The `gen_random_uuid()` function is available in PostgreSQL 13+ by default +- For PostgreSQL versions < 13, you may need to enable the `pgcrypto` extension: + ```sql + CREATE EXTENSION IF NOT EXISTS pgcrypto; + ``` + +## Database Tables + +### metrics.sql + +Contains the schema for storing energy metrics and measurements from the system. + +**Table:** `metrics` + +Stores time-series data including: +- PV power generation +- Grid import/export +- Battery charge/discharge +- Load consumption +- Energy costs +- Battery state of charge +- Weather conditions + +### mpc_decisions.sql + +Contains the schema for persisting Model Predictive Control (MPC) optimization decisions. + +**Table:** `mpc_decisions` + +Stores MPC optimization decisions with timestamp as the PRIMARY KEY (only one decision per timestamp): +- Decision timestamps (PRIMARY KEY) +- Horizon hours +- Battery charge/discharge commands +- Grid import/export commands +- Expected profits +- Forecast data (prices, solar, load, weather) + +## Setup Instructions + +1. Create a PostgreSQL database for the EMS application: + ```bash + createdb ems + ``` + +2. Set the database connection string in your EMS configuration: + ```json + { + "postgres_conn_string": "postgres://username:password@localhost:5432/ems?sslmode=disable" + } + ``` + +3. Run the SQL schema files to create the tables: + ```bash + psql -d ems -f sql/metrics.sql + psql -d ems -f sql/mpc_decisions.sql + ``` + + Or from within `psql`: + ```sql + \i sql/metrics.sql + \i sql/mpc_decisions.sql + ``` + +## MPC Decisions Persistence + +The MPC controller automatically: + +1. **Saves decisions** to the database after each optimization run + - Deletes all existing decisions with timestamp >= minimum timestamp of new decisions + - Inserts/updates new decisions using UPSERT (ON CONFLICT DO UPDATE) + - Each decision includes forecast data and expected outcomes + - Timestamp is the PRIMARY KEY, ensuring only one decision per time slot + +2. **Loads decisions** on scheduler startup + - Loads only future decisions (timestamp >= current time) + - Orders decisions by timestamp + - This allows the system to continue executing decisions after a restart + +## Data Retention + +Consider implementing data retention policies: + +```sql +-- Example: Delete MPC decisions older than current time (past decisions) +DELETE FROM mpc_decisions WHERE timestamp < EXTRACT(EPOCH FROM NOW()); + +-- Example: Delete metrics older than 1 year +DELETE FROM metrics WHERE timestamp < NOW() - INTERVAL '1 year'; +``` + +You can set up a periodic cleanup job using `pg_cron` or an external scheduler. + +## Indexes + +The `mpc_decisions` table uses timestamp as PRIMARY KEY which provides: + +- **Fast lookups** by timestamp (primary key constraint creates implicit index) +- **Uniqueness guarantee** - only one decision per timestamp +- **Efficient range queries** for loading future decisions + +The `metrics` table includes BRIN indexes for time-series efficiency. + +## Monitoring + +Query examples for monitoring: + +```sql +-- Check future MPC decisions (from now onwards) +SELECT + timestamp, + hour, + battery_charge, + battery_discharge, + grid_import, + grid_export, + profit +FROM mpc_decisions +WHERE timestamp >= EXTRACT(EPOCH FROM NOW()) +ORDER BY timestamp ASC; + +-- Check all MPC decisions +SELECT + timestamp, + hour, + battery_charge, + battery_discharge, + grid_import, + grid_export, + profit +FROM mpc_decisions +ORDER BY timestamp ASC; + +-- Check recent energy flow metrics +SELECT + timestamp, + pv_total_power, + grid_import_power, + grid_export_power, + battery_soc +FROM metrics +WHERE metric_name = 'energy_flow' + AND timestamp > NOW() - INTERVAL '24 hours' +ORDER BY timestamp DESC; +``` + +## Troubleshooting + +**Issue:** UUID generation fails with error about `gen_random_uuid()` + +**Solution:** Enable the pgcrypto extension: +```sql +CREATE EXTENSION IF NOT EXISTS pgcrypto; +``` + +**Issue:** Connection refused to PostgreSQL + +**Solution:** Check that PostgreSQL is running and the connection string is correct: +```bash +psql -d ems # Test connection +``` + +**Issue:** Permission denied when creating tables + +**Solution:** Ensure your database user has CREATE privileges: +```sql +GRANT CREATE ON DATABASE ems TO your_username; +``` diff --git a/test_data/metrics.sql b/sql/metrics.sql similarity index 100% rename from test_data/metrics.sql rename to sql/metrics.sql diff --git a/sql/mpc_decisions.sql b/sql/mpc_decisions.sql new file mode 100644 index 0000000..ee10373 --- /dev/null +++ b/sql/mpc_decisions.sql @@ -0,0 +1,32 @@ +CREATE TABLE mpc_decisions ( + timestamp BIGINT PRIMARY KEY, + hour INTEGER NOT NULL, + battery_charge NUMERIC NOT NULL, + battery_discharge NUMERIC NOT NULL, + grid_import NUMERIC NOT NULL, + grid_export NUMERIC NOT NULL, + battery_soc NUMERIC NOT NULL, + profit NUMERIC NOT NULL, + import_price NUMERIC NOT NULL, + export_price NUMERIC NOT NULL, + solar_forecast NUMERIC NOT NULL, + load_forecast NUMERIC NOT NULL, + cloud_coverage NUMERIC, + weather_symbol VARCHAR(100) +); + +-- Column descriptions: +-- timestamp: Unix timestamp when this time slot begins (PRIMARY KEY) +-- hour: Hour index in the optimization horizon (0-based) +-- battery_charge: Battery charging power in kW (positive = charging) +-- battery_discharge: Battery discharging power in kW (positive = discharging) +-- grid_import: Grid import power in kW (positive = importing) +-- grid_export: Grid export power in kW (positive = exporting) +-- battery_soc: Battery state of charge as fraction (0-1) +-- profit: Expected profit for this hour in EUR +-- import_price: Electricity import price in EUR/kWh +-- export_price: Electricity export price in EUR/kWh +-- solar_forecast: Forecasted solar generation in kW (average for the hour) +-- load_forecast: Forecasted load consumption in kW (average for the hour) +-- cloud_coverage: Cloud coverage percentage (0-100) +-- weather_symbol: Weather condition symbol code