feat(ui-api): add scenario builder v8.7 full support (#2054)

antarest/study/business/scenario_builder_management.py

@@ -1,9 +1,12 @@
+import enum
 import typing as t
 
 import typing_extensions as te
 
 from antarest.study.business.utils import execute_or_add_commands
 from antarest.study.model import Study
+from antarest.study.storage.rawstudy.model.filesystem.config.ruleset_matrices import RulesetMatrices, TableForm
+from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy
 from antarest.study.storage.storage_service import StudyStorageService
 from antarest.study.storage.variantstudy.model.command.update_scenario_builder import UpdateScenarioBuilder
 
@@ -22,6 +25,128 @@
 Rulesets: te.TypeAlias = t.MutableMapping[str, Ruleset]
 
 
+class ScenarioType(str, enum.Enum):
+    """
+    Scenario type
+
+    - LOAD: Load scenario
+    - THERMAL: Thermal cluster scenario
+    - HYDRO: Hydraulic scenario
+    - WIND: Wind scenario
+    - SOLAR: Solar scenario
+    - NTC: NTC scenario (link)
+    - RENEWABLE: Renewable scenario
+    - BINDING_CONSTRAINTS: Binding constraints scenario
+    - HYDRO_INITIAL_LEVEL: hydraulic Initial level scenario
+    - HYDRO_FINAL_LEVEL: hydraulic Final level scenario
+    - HYDRO_GENERATION_POWER: hydraulic Generation power scenario
+    """
+
+    LOAD = "load"
+    THERMAL = "thermal"
+    HYDRO = "hydro"
+    WIND = "wind"
+    SOLAR = "solar"
+    LINK = "ntc"
+    RENEWABLE = "renewable"
+    BINDING_CONSTRAINTS = "bindingConstraints"
+    HYDRO_INITIAL_LEVEL = "hydroInitialLevels"
+    HYDRO_FINAL_LEVEL = "hydroFinalLevels"
+    HYDRO_GENERATION_POWER = "hydroGenerationPower"
+
+    def __str__(self) -> str:
+        """Return the string representation of the enum value."""
+        return self.value
+
+
+SYMBOLS_BY_SCENARIO_TYPES = {
+    ScenarioType.LOAD: "l",
+    ScenarioType.HYDRO: "h",
+    ScenarioType.WIND: "w",
+    ScenarioType.SOLAR: "s",
+    ScenarioType.THERMAL: "t",
+    ScenarioType.RENEWABLE: "r",
+    ScenarioType.LINK: "ntc",
+    ScenarioType.BINDING_CONSTRAINTS: "bc",
+    ScenarioType.HYDRO_INITIAL_LEVEL: "hl",
+    ScenarioType.HYDRO_FINAL_LEVEL: "hfl",
+    ScenarioType.HYDRO_GENERATION_POWER: "hgp",
+}
+
+
+def _get_ruleset_config(
+    file_study: FileStudy,
+    ruleset_name: str,
+    symbol: str = "",
+) -> t.Dict[str, t.Union[int, float]]:
+    try:
+        suffix = f"/{symbol}" if symbol else ""
+        url = f"settings/scenariobuilder/{ruleset_name}{suffix}".split("/")
+        ruleset_cfg = t.cast(t.Dict[str, t.Union[int, float]], file_study.tree.get(url))
+    except KeyError:
+        ruleset_cfg = {}
+    return ruleset_cfg
+
+
+def _get_nb_years(file_study: FileStudy) -> int:
+    try:
+        # noinspection SpellCheckingInspection
+        url = "settings/generaldata/general/nbyears".split("/")
+        nb_years = t.cast(int, file_study.tree.get(url))
+    except KeyError:
+        nb_years = 1
+    return nb_years
+
+
+def _get_active_ruleset_name(file_study: FileStudy, default_ruleset: str = "Default Ruleset") -> str:
+    """
+    Get the active ruleset name stored in the configuration at the following path:
+    ``settings/generaldata.ini``, in the section "general", key "active-rules-scenario".
+
+    This ruleset name must match a section name in the scenario builder configuration
+    at the following path: ``settings/scenariobuilder``.
+
+    Args:
+        file_study: Object representing the study file
+        default_ruleset: Name of the default ruleset
+
+    Returns:
+        The active ruleset name if found in the configuration, or the default ruleset name if missing.
+    """
+    try:
+        url = "settings/generaldata/general/active-rules-scenario".split("/")
+        active_ruleset = t.cast(str, file_study.tree.get(url))
+    except KeyError:
+        active_ruleset = default_ruleset
+    else:
+        # In some old studies, the active ruleset is stored in lowercase.
+        if not active_ruleset or active_ruleset.lower() == "default ruleset":
+            active_ruleset = default_ruleset
+    return active_ruleset
+
+
+def _build_ruleset(file_study: FileStudy, symbol: str = "") -> RulesetMatrices:
+    ruleset_name = _get_active_ruleset_name(file_study)
+    nb_years = _get_nb_years(file_study)
+    ruleset_config = _get_ruleset_config(file_study, ruleset_name, symbol)
+
+    # Create and populate the RulesetMatrices
+    areas = file_study.config.areas
+    groups = file_study.config.get_binding_constraint_groups() if file_study.config.version >= 870 else []
+    scenario_types = {s: str(st) for st, s in SYMBOLS_BY_SCENARIO_TYPES.items()}
+    ruleset = RulesetMatrices(
+        nb_years=nb_years,
+        areas=areas,
+        links=((a1, a2) for a1 in areas for a2 in file_study.config.get_links(a1)),
+        thermals={a: file_study.config.get_thermal_ids(a) for a in areas},
+        renewables={a: file_study.config.get_renewable_ids(a) for a in areas},
+        groups=groups,
+        scenario_types=scenario_types,
+    )
+    ruleset.update_rules(ruleset_config)
+    return ruleset
+
+
 class ScenarioBuilderManager:
     def __init__(self, storage_service: StudyStorageService) -> None:
         self.storage_service = storage_service
@@ -79,6 +204,33 @@ def update_config(self, study: Study, rulesets: Rulesets) -> None:
             self.storage_service,
         )
 
+    def get_scenario_by_type(self, study: Study, scenario_type: ScenarioType) -> TableForm:
+        symbol = SYMBOLS_BY_SCENARIO_TYPES[scenario_type]
+        file_study = self.storage_service.get_storage(study).get_raw(study)
+        ruleset = _build_ruleset(file_study, symbol)
+        ruleset.sort_scenarios()
+
+        # Extract the table form for the given scenario type
+        table_form = ruleset.get_table_form(str(scenario_type), nan_value="")
+        return table_form
+
+    def update_scenario_by_type(self, study: Study, table_form: TableForm, scenario_type: ScenarioType) -> TableForm:
+        file_study = self.storage_service.get_storage(study).get_raw(study)
+        ruleset = _build_ruleset(file_study)
+        ruleset.update_table_form(table_form, str(scenario_type), nan_value="")
+        ruleset.sort_scenarios()
+
+        # Create the UpdateScenarioBuilder command
+        ruleset_name = _get_active_ruleset_name(file_study)
+        data = {ruleset_name: ruleset.get_rules(allow_nan=True)}
+        command_context = self.storage_service.variant_study_service.command_factory.command_context
+        update_scenario = UpdateScenarioBuilder(data=data, command_context=command_context)
+        execute_or_add_commands(study, file_study, [update_scenario], self.storage_service)
+
+        # Extract the updated table form for the given scenario type
+        table_form = ruleset.get_table_form(str(scenario_type), nan_value="")
+        return table_form
+
 
 def _populate_common(section: _Section, symbol: str, data: t.Mapping[str, t.Mapping[str, t.Any]]) -> None:
     for area, scenario_area in data.items():