implement scenario builder

src/andromede/main/main.py

@@ -151,19 +151,17 @@ def main() -> None:
 
     network = build_network(components)
 
-    for scenario in range(1, args.scenario + 1):
-        timeblock = TimeBlock(1, list(range(args.duration)))
-        try:
-            problem = build_problem(network, database, timeblock, scenario)
-        except IndexError as e:
-            raise IndexError(
-                str(e)
-                + ". Did you correctly use the '--duration' and '--scenario' parameters ?"
-            )
-
-        status = problem.solver.Solve()
-        print("scenario ", scenario)
-        print("status : ", status)
+    timeblock = TimeBlock(1, list(range(args.duration)))
+    try:
+        problem = build_problem(network, database, timeblock, args.scenario)
+    except IndexError as e:
+        raise IndexError(
+            str(e)
+            + ". Did you correctly use the '--duration' and '--scenario' parameters ?"
+        )
+
+    status = problem.solver.Solve()
+    print("status : ", status)
 
     print("avarage final cost : ", problem.solver.Objective().Value())
 

src/andromede/study/__init__.py

@@ -15,6 +15,7 @@
     DataBase,
     ScenarioIndex,
     ScenarioSeriesData,
+    Scenarization,
     TimeIndex,
     TimeScenarioIndex,
     TimeScenarioSeriesData,

src/andromede/study/data.py

@@ -118,6 +118,19 @@ def load_ts_from_txt(
         raise Exception(f"An error has arrived when processing '{ts_path}'")
 
 
+@dataclass(frozen=True)
+class Scenarization:
+    _scenarization: Dict[int, int]
+
+    def get_scenario_for_year(self, year: int) -> int:
+        return self._scenarization[year]
+
+    def add_year(self, year: int, scenario: int) -> None:
+        if year in self._scenarization:
+            raise ValueError(f"the year {year} is already defined")
+        self._scenarization[year] = scenario
+
+
 @dataclass(frozen=True)
 class TimeScenarioSeriesData(AbstractDataStructure):
     """
@@ -127,8 +140,11 @@ class TimeScenarioSeriesData(AbstractDataStructure):
     """
 
     time_scenario_series: pd.DataFrame
+    scenarization: Optional[Scenarization] = None
 
     def get_value(self, timestep: int, scenario: int) -> float:
+        if self.scenarization:
+            scenario = self.scenarization.get_scenario_for_year(scenario)
         value = str(self.time_scenario_series.iloc[timestep, scenario])
         return float(value)
 

src/andromede/study/parsing.py

@@ -36,15 +36,23 @@ class InputPortConnections(BaseModel):
 class InputComponentParameter(BaseModel):
     name: str
     type: str
+    scenario_group: Optional[str] = None
     value: Optional[float] = None
     timeseries: Optional[str] = None
 
+    class Config:
+        alias_generator = _to_kebab
+
 
 class InputComponent(BaseModel):
     id: str
     model: str
+    scenario_group: Optional[str] = None
     parameters: Optional[List[InputComponentParameter]] = None
 
+    class Config:
+        alias_generator = _to_kebab
+
 
 class InputComponents(BaseModel):
     nodes: List[InputComponent] = Field(default_factory=list)

src/andromede/study/resolve_components.py

@@ -25,6 +25,7 @@
     Node,
     PortRef,
     PortsConnection,
+    Scenarization,
 )
 from andromede.study.data import (
     AbstractDataStructure,
@@ -160,11 +161,48 @@ def _evaluate_param_type(
     param_value: Optional[float],
     timeseries_name: Optional[str],
     timeseries_dir: Optional[Path],
+    scenarization: Optional[Scenarization] = None,
 ) -> AbstractDataStructure:
     if param_type == "constant" and param_value is not None:
         return ConstantData(float(param_value))
 
     elif param_type == "timeseries":
-        return TimeScenarioSeriesData(load_ts_from_txt(timeseries_name, timeseries_dir))
+        return TimeScenarioSeriesData(
+            load_ts_from_txt(timeseries_name, timeseries_dir), scenarization
+        )
 
     raise ValueError(f"Data should be either constant or timeseries ")
+
+
+def _resolve_scenarization(
+    scenario_builder_data: pd.core.frame.DataFrame,
+) -> Dict[str, Scenarization]:
+    output: Dict[str, Scenarization] = {}
+    for i, row in scenario_builder_data.iterrows():
+        if row["name"] in output:
+            output[row["name"]].add_year(row["year"], row["scenario"])
+        else:
+            output[row["name"]] = Scenarization({row["year"]: row["scenario"]})
+    return output
+
+
+def build_scenarized_data_base(
+    input_comp: InputComponents,
+    scenario_builder_data: pd.core.frame.DataFrame,
+    timeseries_dir: Optional[Path],
+) -> DataBase:
+    database = DataBase()
+    scenarizations = _resolve_scenarization(scenario_builder_data)
+    for comp in input_comp.components:
+        scenarization = None
+        if comp.scenario_group:
+            scenarization = scenarizations[comp.scenario_group]
+        for param in comp.parameters or []:
+            if param.scenario_group:
+                scenarization = scenarizations[param.scenario_group]
+            param_value = _evaluate_param_type(
+                param.type, param.value, param.timeseries, timeseries_dir, scenarization
+            )
+            database.add_data(comp.id, param.name, param_value)
+
+    return database

tests/functional/input/components/components.yml

@@ -54,6 +54,7 @@ study:
       model: demand
       parameters:
         - name: demand
+          scenario-group: g1
           type: timeseries
           timeseries: demand-ts
     - id: S

tests/functional/input/models/lib_3.yml

@@ -0,0 +1,26 @@
+# Copyright (c) 2024, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+library:
+  id: basic 3
+  dependence: basic
+  description: Basic library
+
+  models:
+
+    - id: node
+      description: A basic balancing node model
+      ports:
+        - name: injection_port
+          type: flow
+      binding-constraints:
+        - name: balance
+          expression:  sum_connections(injection_port.flow) = 0