Hydro heuristic

src/andromede/hydro_heuristic/data.py

@@ -10,118 +10,154 @@
 #
 # This file is part of the Antares project.
 
+from dataclasses import dataclass
 from pathlib import Path
 from typing import List, Optional
 
 import numpy as np
 
 
+@dataclass
+class ReservoirParameters:
+    capacity: float
+    initial_level: float
+    folder_name: str
+    scenario: int
+
+
+@dataclass
+class HydroHeuristicParameters:
+    inter_breakdown: int = 1
+    total_target: Optional[float] = None
+
+
+@dataclass
+class DataAggregatorParameters:
+    hours_aggregated_time_steps: List[int]
+    timesteps: List[int]
+
+
 def get_number_of_days_in_month(month: int) -> int:
     number_day_month = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31][month]
     return number_day_month
 
 
+class RawHydroData:
+    def __init__(self, folder_name: str, scenario: int) -> None:
+        self.folder_name = folder_name
+        self.scenario = scenario
+
+        self.name_file = {
+            "demand": "load",
+            "inflow": "mod",
+            "lower_rule_curve": "reservoir",
+            "upper_rule_curve": "reservoir",
+            "max_generating": "maxpower",
+        }
+
+        self.column = {
+            "demand": self.scenario,
+            "inflow": self.scenario,
+            "lower_rule_curve": 0,
+            "upper_rule_curve": 2,
+            "max_generating": 0,
+        }
+
+    def read_data(self, name: str) -> list[float]:
+        hours_input = 1 if name == "demand" else 24
+
+        data = np.loadtxt(
+            Path(__file__).parent
+            / (
+                "../../../tests/functional/data/"
+                + self.folder_name
+                + "/"
+                + self.name_file[name]
+                + ".txt"
+            )
+        )
+        if len(data.shape) >= 2:
+            data = data[:, self.column[name]]
+        data = np.repeat(data, hours_input)
+        if self.name_file[name] == "mod":
+            data = data / hours_input
+
+        return list(data)
+
+
 class HydroHeuristicData:
     def __init__(
         self,
-        scenario: int,
-        horizon: str,
-        folder_name: str,
-        timesteps: List[int],
-        capacity: float,
-        initial_level: float,
+        data_aggregator_parameters: DataAggregatorParameters,
+        reservoir_data: ReservoirParameters,
     ):
-        self.folder_name = folder_name
-        if horizon == "monthly":
-            hours_aggregated_time_steps = [
-                24 * get_number_of_days_in_month(m) for m in range(12)
-            ]
-        elif horizon == "daily":
-            hours_aggregated_time_steps = [24 for d in range(365)]
-        self.hours_aggregated_time_steps = hours_aggregated_time_steps
-        self.timesteps = timesteps
-        self.capacity = capacity
-        self.initial_level = initial_level
+        self.reservoir_data = reservoir_data
+
+        data_aggregator = DataAggregator(
+            data_aggregator_parameters.hours_aggregated_time_steps,
+            data_aggregator_parameters.timesteps,
+        )
+
+        raw_data_reader = RawHydroData(
+            reservoir_data.folder_name, reservoir_data.scenario
+        )
 
-        self.demand = self.get_input_data(
-            name_file="load",
-            column=scenario,
-            hours_input=1,
+        self.demand = data_aggregator.aggregate_data(
             operator="sum",
+            data=raw_data_reader.read_data("demand"),
         )
-        self.inflow = self.get_input_data(
-            name_file="mod",
-            column=scenario,
-            hours_input=24,
+        self.inflow = data_aggregator.aggregate_data(
             operator="sum",
+            data=raw_data_reader.read_data("inflow"),
         )
-        self.lower_rule_curve = self.get_input_data(
-            name_file="reservoir",
-            column=0,
-            hours_input=24,
+        self.lower_rule_curve = data_aggregator.aggregate_data(
             operator="lag_first_element",
+            data=raw_data_reader.read_data("lower_rule_curve"),
         )
-        self.upper_rule_curve = self.get_input_data(
-            name_file="reservoir",
-            column=2,
-            hours_input=24,
+        self.upper_rule_curve = data_aggregator.aggregate_data(
             operator="lag_first_element",
+            data=raw_data_reader.read_data("upper_rule_curve"),
         )
-        max_generating = self.get_input_data(
-            name_file="maxpower",
-            column=0,
-            hours_input=24,
+        self.max_generating = data_aggregator.aggregate_data(
             operator="sum",
+            data=raw_data_reader.read_data("max_generating"),
         )
-        self.max_generating = [x * 24 for x in max_generating]
 
-    def get_input_data(
-        self,
-        name_file: str,
-        column: int,
-        hours_input: int,
-        operator: str,
-    ) -> List[float]:
-        data = np.loadtxt(
-            Path(__file__).parent
-            / (
-                "../../../tests/functional/data/"
-                + self.folder_name
-                + "/"
-                + name_file
-                + ".txt"
-            )
+    def compute_target(self, heuristic_parameters: HydroHeuristicParameters) -> None:
+        if heuristic_parameters.total_target is None:
+            total_target = sum(self.inflow)
+        else:
+            total_target = heuristic_parameters.total_target
+        target = (
+            total_target
+            * np.power(self.demand, heuristic_parameters.inter_breakdown)
+            / sum(np.power(self.demand, heuristic_parameters.inter_breakdown))
         )
-        data = data[:, column]
+
+        self.target = list(target)
+
+
+class DataAggregator:
+    def __init__(
+        self,
+        hours_aggregated_time_steps: List[int],
+        timesteps: List[int],
+    ) -> None:
+        self.hours_aggregated_time_steps = hours_aggregated_time_steps
+        self.timesteps = timesteps
+
+    def aggregate_data(self, operator: str, data: list[float]) -> List[float]:
         aggregated_data: List[float] = []
         hour = 0
         for time_step, hours_time_step in enumerate(self.hours_aggregated_time_steps):
-            assert hours_time_step % hours_input == 0
             if time_step in self.timesteps:
                 if operator == "sum":
-                    aggregated_data.append(
-                        np.sum(data[hour : hour + hours_time_step // hours_input])
-                    )
+                    aggregated_data.append(np.sum(data[hour : hour + hours_time_step]))
                 elif operator == "lag_first_element":
-                    aggregated_data.append(
-                        data[(hour + hours_time_step // hours_input) % len(data)]
-                    )
-            hour += hours_time_step // hours_input
+                    aggregated_data.append(data[(hour + hours_time_step) % len(data)])
+            hour += hours_time_step
         return aggregated_data
 
-    def compute_target(
-        self, total_target: Optional[float], inter_breakdown: int = 1
-    ) -> None:
-        if total_target is None:
-            total_target = sum(self.inflow)
-        target = (
-            total_target
-            * np.power(self.demand, inter_breakdown)
-            / sum(np.power(self.demand, inter_breakdown))
-        )
-
-        self.target = list(target)
-
 
 def update_generation_target(
     all_daily_generation: list[float], daily_generation: list[float]

src/andromede/hydro_heuristic/heuristic_model.py

@@ -35,30 +35,53 @@ def __init__(
         self.hydro_model = hydro_model
         self.horizon = horizon
 
-    def get_model(self) -> Model:
+        self.objective_function_cost = {
+            "gamma_d": 1,
+            "gamma_delta": 1 if horizon == "monthly" else 2,
+            "gamma_y": 100000 if horizon == "monthly" else 68,
+            "gamma_w": 0 if horizon == "monthly" else 34,
+            "gamma_v+": 100 if horizon == "monthly" else 0,
+            "gamma_v-": 100 if horizon == "monthly" else 68,
+            "gamma_o": 0 if horizon == "monthly" else 23 * 68 + 1,
+            "gamma_s": 0 if horizon == "monthly" else -1 / 32,
+        }
+
+    def get_model(
+        self,
+    ) -> Model:
+        assert "level" in self.hydro_model.variables.keys()
+        assert "generating" in self.hydro_model.variables.keys()
+        assert "overflow" in self.hydro_model.variables.keys()
+        assert "lower_rule_curve" in self.hydro_model.parameters.keys()
+        assert "upper_rule_curve" in self.hydro_model.parameters.keys()
         HYDRO_HEURISTIC = model(
             id="H",
-            parameters=[p for p in self.hydro_model.parameters.values()]
+            parameters=list(self.hydro_model.parameters.values())
             + self.get_heuristic_parameters(),
-            variables=[v for v in self.hydro_model.variables.values()]
+            variables=list(self.hydro_model.variables.values())
             + self.get_heuristic_variables(),
-            constraints=[c for c in self.hydro_model.constraints.values()]
+            constraints=list(self.hydro_model.constraints.values())
             + self.get_heuristic_constraints(),
             objective_operational_contribution=self.get_heuristic_objective(),
         )
         return HYDRO_HEURISTIC
 
     def get_heuristic_objective(self) -> ExpressionNode:
         return (
-            param("gamma_d") * var("distance_between_target_and_generating")
-            + param("gamma_v+") * var("violation_upper_rule_curve")
-            + param("gamma_v-") * var("violation_lower_rule_curve")
-            + param("gamma_o") * var("overflow")
-            + param("gamma_s") * var("level")
+            self.objective_function_cost["gamma_d"]
+            * var("distance_between_target_and_generating")
+            + self.objective_function_cost["gamma_v+"]
+            * var("violation_upper_rule_curve")
+            + self.objective_function_cost["gamma_v-"]
+            * var("violation_lower_rule_curve")
+            + self.objective_function_cost["gamma_o"] * var("overflow")
+            + self.objective_function_cost["gamma_s"] * var("level")
         ).sum().expec() + (
-            param("gamma_delta") * var("max_distance_between_target_and_generating")
-            + param("gamma_y") * var("max_violation_lower_rule_curve")
-            + param("gamma_w") * var("gap_to_target")
+            self.objective_function_cost["gamma_delta"]
+            * var("max_distance_between_target_and_generating")
+            + self.objective_function_cost["gamma_y"]
+            * var("max_violation_lower_rule_curve")
+            + self.objective_function_cost["gamma_w"] * var("gap_to_target")
         ).expec()
 
     def get_heuristic_variables(self) -> List[Variable]:
@@ -99,14 +122,6 @@ def get_heuristic_parameters(self) -> List[Parameter]:
         return [
             float_parameter("generating_target", TIME_AND_SCENARIO_FREE),
             float_parameter("overall_target", CONSTANT_PER_SCENARIO),
-            float_parameter("gamma_d", CONSTANT),
-            float_parameter("gamma_delta", CONSTANT),
-            float_parameter("gamma_y", CONSTANT),
-            float_parameter("gamma_w", CONSTANT),
-            float_parameter("gamma_v+", CONSTANT),
-            float_parameter("gamma_v-", CONSTANT),
-            float_parameter("gamma_o", CONSTANT),
-            float_parameter("gamma_s", CONSTANT),
         ]
 
     def get_heuristic_constraints(self) -> List[Constraint]: