bugfix(sb): handle missing objects in Scenario Builder configuration (#…

…2038)

antarest/study/business/scenario_builder_management.py

@@ -1,87 +1,108 @@
-from typing import Any, Dict, List
+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.storage_service import StudyStorageService
 from antarest.study.storage.variantstudy.model.command.update_scenario_builder import UpdateScenarioBuilder
 
-KEY_SEP = ","
-HL_COEF = 100
+# Symbols used in scenario builder data
+_AREA_RELATED_SYMBOLS = "l", "h", "w", "s", "bc", "hgp"
+_LINK_RELATED_SYMBOLS = ("ntc",)
+_HYDRO_LEVEL_RELATED_SYMBOLS = "hl", "hfl"
+_CLUSTER_RELATED_SYMBOLS = "t", "r"
+
+_HYDRO_LEVEL_PERCENT = 100
+
+_Section: te.TypeAlias = t.MutableMapping[str, t.Union[int, float]]
+_Sections: te.TypeAlias = t.MutableMapping[str, _Section]
+
+Ruleset: te.TypeAlias = t.MutableMapping[str, t.Any]
+Rulesets: te.TypeAlias = t.MutableMapping[str, Ruleset]
 
 
 class ScenarioBuilderManager:
     def __init__(self, storage_service: StudyStorageService) -> None:
         self.storage_service = storage_service
 
-    def get_config(self, study: Study) -> Dict[str, Any]:
-        temp = self.storage_service.get_storage(study).get(study, "/settings/scenariobuilder")
-
-        def format_key(key: str) -> List[str]:
-            parts = key.split(KEY_SEP)
-            # "ntc,area1,area2,0" to ["ntc", "area1 / area2", "0"]
-            if parts[0] == "ntc":
-                return [parts[0], f"{parts[1]} / {parts[2]}", parts[3]]
-            # "t,area1,0,thermal1" to ["t", "area1", "thermal1", "0"]
-            elif parts[0] == "t":
-                return [parts[0], parts[1], parts[3], parts[2]]
-            # "[symbol],area1,0" to [[symbol], "area1", "0"]
-            return parts
-
-        def format_obj(obj: Dict[str, Any]) -> Dict[str, Any]:
-            result = {}
-            for k, v in obj.items():
-                if isinstance(v, dict):
-                    result[k] = format_obj(v)
-                else:
-                    keys = format_key(k)
-                    nested_dict = result
-                    for key in keys[:-1]:
-                        nested_dict = nested_dict.setdefault(key, {})
-                    nested_dict[keys[-1]] = v * HL_COEF if keys[0] == "hl" else v
-            return result
-
-        return format_obj(temp)
-
-    def update_config(self, study: Study, data: Dict[str, Any]) -> None:
+    def get_config(self, study: Study) -> Rulesets:
+        sections = t.cast(_Sections, self.storage_service.get_storage(study).get(study, "/settings/scenariobuilder"))
+
+        rulesets: Rulesets = {}
+        for ruleset_name, data in sections.items():
+            ruleset = rulesets.setdefault(ruleset_name, {})
+            for key, value in data.items():
+                symbol, *parts = key.split(",")
+                scenario = ruleset.setdefault(symbol, {})
+                if symbol in _AREA_RELATED_SYMBOLS:
+                    scenario_area = scenario.setdefault(parts[0], {})
+                    scenario_area[parts[1]] = int(value)
+                elif symbol in _HYDRO_LEVEL_RELATED_SYMBOLS:
+                    scenario_area = scenario.setdefault(parts[0], {})
+                    scenario_area[parts[1]] = float(value) * _HYDRO_LEVEL_PERCENT
+                elif symbol in _LINK_RELATED_SYMBOLS:
+                    scenario_link = scenario.setdefault(f"{parts[0]} / {parts[1]}", {})
+                    scenario_link[parts[2]] = int(value)
+                elif symbol in _CLUSTER_RELATED_SYMBOLS:
+                    scenario_area = scenario.setdefault(parts[0], {})
+                    scenario_area_cluster = scenario_area.setdefault(parts[2], {})
+                    scenario_area_cluster[parts[1]] = int(value)
+                else:  # pragma: no cover
+                    raise NotImplementedError(f"Unknown symbol {symbol}")
+
+        return rulesets
+
+    def update_config(self, study: Study, rulesets: Rulesets) -> None:
         file_study = self.storage_service.get_storage(study).get_raw(study)
 
-        def to_valid_key(key: str) -> str:
-            parts = key.split(KEY_SEP)
-            # "ntc,area1 / area2,0" to "ntc,area1,area2,0"
-            if parts[0] == "ntc":
-                area1, area2 = parts[1].split(" / ")
-                return KEY_SEP.join([parts[0], area1, area2, parts[2]])
-            # "t,area1,thermal1,0" to "t,area1,0,thermal1"
-            elif parts[0] == "t":
-                return KEY_SEP.join([parts[0], parts[1], parts[3], parts[2]])
-            # "[symbol],area1,0"
-            return key
-
-        def flatten_obj(obj: Dict[str, Any], parent_key: str = "") -> Dict[str, Dict[str, int]]:
-            items = []  # type: ignore
-            for k, v in obj.items():
-                new_key = parent_key + KEY_SEP + k if parent_key else k
-                if isinstance(v, dict):
-                    items.extend(flatten_obj(v, new_key).items())
-                else:
-                    symbol = new_key.split(KEY_SEP)[0]
-                    items.append(
-                        (
-                            to_valid_key(new_key),
-                            v / HL_COEF if symbol == "hl" else v,
-                        )
-                    )
-            return dict(items)
+        sections: _Sections = {}
+        for ruleset_name, ruleset in rulesets.items():
+            section = sections[ruleset_name] = {}
+            for symbol, data in ruleset.items():
+                if symbol in _AREA_RELATED_SYMBOLS:
+                    _populate_common(section, symbol, data)
+                elif symbol in _HYDRO_LEVEL_RELATED_SYMBOLS:
+                    _populate_hydro_levels(section, symbol, data)
+                elif symbol in _LINK_RELATED_SYMBOLS:
+                    _populate_links(section, symbol, data)
+                elif symbol in _CLUSTER_RELATED_SYMBOLS:
+                    _populate_clusters(section, symbol, data)
+                else:  # pragma: no cover
+                    raise NotImplementedError(f"Unknown symbol {symbol}")
 
+        context = self.storage_service.variant_study_service.command_factory.command_context
         execute_or_add_commands(
             study,
             file_study,
-            [
-                UpdateScenarioBuilder(
-                    # The value is a string when it is a ruleset cloning/deleting
-                    data={k: flatten_obj(v) if isinstance(v, dict) else v for k, v in data.items()},
-                    command_context=self.storage_service.variant_study_service.command_factory.command_context,
-                )
-            ],
+            [UpdateScenarioBuilder(data=sections, command_context=context)],
             self.storage_service,
         )
+
+
+def _populate_common(section: _Section, symbol: str, data: t.Mapping[str, t.Mapping[str, t.Any]]) -> None:
+    for area, scenario_area in data.items():
+        for year, value in scenario_area.items():
+            section[f"{symbol},{area},{year}"] = value
+
+
+def _populate_hydro_levels(section: _Section, symbol: str, data: t.Mapping[str, t.Mapping[str, t.Any]]) -> None:
+    for area, scenario_area in data.items():
+        for year, value in scenario_area.items():
+            if isinstance(value, (int, float)) and value != float("nan"):
+                value /= _HYDRO_LEVEL_PERCENT
+            section[f"{symbol},{area},{year}"] = value
+
+
+def _populate_links(section: _Section, symbol: str, data: t.Mapping[str, t.Mapping[str, t.Any]]) -> None:
+    for link, scenario_link in data.items():
+        for year, value in scenario_link.items():
+            area1, area2 = link.split(" / ")
+            section[f"{symbol},{area1},{area2},{year}"] = value
+
+
+def _populate_clusters(section: _Section, symbol: str, data: t.Mapping[str, t.Mapping[str, t.Any]]) -> None:
+    for area, scenario_area in data.items():
+        for cluster, scenario_area_cluster in scenario_area.items():
+            for year, value in scenario_area_cluster.items():
+                section[f"{symbol},{area},{year},{cluster}"] = value

antarest/study/business/timeseries_config_management.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional
+import typing as t
 
 from pydantic import StrictBool, StrictInt, root_validator, validator
 
@@ -28,30 +28,30 @@ class SeasonCorrelation(EnumIgnoreCase):
 
 
 class TSFormFieldsForType(FormFieldsBaseModel):
-    stochastic_ts_status: Optional[StrictBool]
-    number: Optional[StrictInt]
-    refresh: Optional[StrictBool]
-    refresh_interval: Optional[StrictInt]
-    season_correlation: Optional[SeasonCorrelation]
-    store_in_input: Optional[StrictBool]
-    store_in_output: Optional[StrictBool]
-    intra_modal: Optional[StrictBool]
-    inter_modal: Optional[StrictBool]
+    stochastic_ts_status: t.Optional[StrictBool]
+    number: t.Optional[StrictInt]
+    refresh: t.Optional[StrictBool]
+    refresh_interval: t.Optional[StrictInt]
+    season_correlation: t.Optional[SeasonCorrelation]
+    store_in_input: t.Optional[StrictBool]
+    store_in_output: t.Optional[StrictBool]
+    intra_modal: t.Optional[StrictBool]
+    inter_modal: t.Optional[StrictBool]
 
 
 class TSFormFields(FormFieldsBaseModel):
-    load: Optional[TSFormFieldsForType] = None
-    hydro: Optional[TSFormFieldsForType] = None
-    thermal: Optional[TSFormFieldsForType] = None
-    wind: Optional[TSFormFieldsForType] = None
-    solar: Optional[TSFormFieldsForType] = None
-    renewables: Optional[TSFormFieldsForType] = None
-    ntc: Optional[TSFormFieldsForType] = None
+    load: t.Optional[TSFormFieldsForType] = None
+    hydro: t.Optional[TSFormFieldsForType] = None
+    thermal: t.Optional[TSFormFieldsForType] = None
+    wind: t.Optional[TSFormFieldsForType] = None
+    solar: t.Optional[TSFormFieldsForType] = None
+    renewables: t.Optional[TSFormFieldsForType] = None
+    ntc: t.Optional[TSFormFieldsForType] = None
 
     @root_validator(pre=True)
     def check_type_validity(
-        cls, values: Dict[str, Optional[TSFormFieldsForType]]
-    ) -> Dict[str, Optional[TSFormFieldsForType]]:
+        cls, values: t.Dict[str, t.Optional[TSFormFieldsForType]]
+    ) -> t.Dict[str, t.Optional[TSFormFieldsForType]]:
         def has_type(ts_type: TSType) -> bool:
             return values.get(ts_type.value, None) is not None
 
@@ -117,7 +117,7 @@ def __set_field_values_for_type(
         ts_type: TSType,
         field_values: TSFormFieldsForType,
     ) -> None:
-        commands: List[UpdateConfig] = []
+        commands: t.List[UpdateConfig] = []
         values = field_values.dict()
 
         for field, path in PATH_BY_TS_STR_FIELD.items():
@@ -155,7 +155,7 @@ def __set_field_values_for_type(
         if len(commands) > 0:
             execute_or_add_commands(study, file_study, commands, self.storage_service)
 
-    def __set_ts_types_str(self, file_study: FileStudy, path: str, values: Dict[TSType, bool]) -> UpdateConfig:
+    def __set_ts_types_str(self, file_study: FileStudy, path: str, values: t.Dict[TSType, bool]) -> UpdateConfig:
         """
         Set string value with the format: "[ts_type_1], [ts_type_2]"
         """
@@ -188,20 +188,22 @@ def __get_form_fields_for_type(
         file_study: FileStudy,
         ts_type: TSType,
         general_data: JSON,
-    ) -> Optional[TSFormFieldsForType]:
+    ) -> t.Optional[TSFormFieldsForType]:
         general = general_data.get("general", {})
         input_ = general_data.get("input", {})
         output = general_data.get("output", {})
 
-        is_aggregated = file_study.config.enr_modelling == EnrModelling.AGGREGATED.value
+        config = file_study.config
+        study_version = config.version
+        has_renewables = config.version >= 810 and EnrModelling(config.enr_modelling) == EnrModelling.CLUSTERS
 
-        if ts_type == TSType.RENEWABLES and is_aggregated:
+        if ts_type == TSType.RENEWABLES and not has_renewables:
             return None
 
-        if ts_type in [TSType.WIND, TSType.SOLAR] and not is_aggregated:
+        if ts_type in [TSType.WIND, TSType.SOLAR] and has_renewables:
             return None
 
-        if ts_type == TSType.NTC and file_study.config.version < 820:
+        if ts_type == TSType.NTC and study_version < 820:
             return None
 
         is_special_type = ts_type == TSType.RENEWABLES or ts_type == TSType.NTC

antarest/study/storage/rawstudy/model/filesystem/config/files.py

@@ -17,6 +17,7 @@
 )
 from antarest.study.storage.rawstudy.model.filesystem.config.field_validators import extract_filtering
 from antarest.study.storage.rawstudy.model.filesystem.config.model import (
+    DEFAULT_GROUP,
     Area,
     BindingConstraintDTO,
     DistrictSet,
@@ -224,8 +225,16 @@ def _parse_bindings(root: Path) -> t.List[BindingConstraintDTO]:
                 cluster_set.add(key)
                 area_set.add(key.split(".", 1)[0])
 
+        group = bind.get("group", DEFAULT_GROUP)
+
         output_list.append(
-            BindingConstraintDTO(id=bind["id"], areas=area_set, clusters=cluster_set, time_step=time_step)
+            BindingConstraintDTO(
+                id=bind["id"],
+                areas=area_set,
+                clusters=cluster_set,
+                time_step=time_step,
+                group=group,
+            )
         )
 
     return output_list

antarest/study/storage/rawstudy/model/filesystem/config/model.py

@@ -13,11 +13,26 @@
 from .st_storage import STStorageConfigType
 from .thermal import ThermalConfigType
 
+DEFAULT_GROUP = "default"
+"""Default group for binding constraints (since v8.7)."""
+
 
 class EnrModelling(EnumIgnoreCase):
+    """
+    Renewable energy modelling type (since v8.1).
+
+    Attributes:
+        AGGREGATED: Simulations are done using aggregated data from wind and solar.
+        CLUSTERS: Simulations are done using renewable clusters
+    """
+
     AGGREGATED = "aggregated"
     CLUSTERS = "clusters"
 
+    def __str__(self) -> str:
+        """Return the string representation of the enum value."""
+        return self.value
+
 
 class Link(BaseModel, extra="ignore"):
     """
@@ -105,6 +120,8 @@ class BindingConstraintDTO(BaseModel):
     areas: t.Set[str]
     clusters: t.Set[str]
     time_step: BindingConstraintFrequency
+    # since v8.7
+    group: str = DEFAULT_GROUP
 
 
 class FileStudyTreeConfig(DTO):
@@ -125,7 +142,7 @@ def __init__(
         bindings: t.Optional[t.List[BindingConstraintDTO]] = None,
         store_new_set: bool = False,
         archive_input_series: t.Optional[t.List[str]] = None,
-        enr_modelling: str = EnrModelling.AGGREGATED.value,
+        enr_modelling: str = str(EnrModelling.AGGREGATED),
         cache: t.Optional[t.Dict[str, t.List[str]]] = None,
         zip_path: t.Optional[Path] = None,
     ):
@@ -185,7 +202,7 @@ def at_file(self, filepath: Path) -> "FileStudyTreeConfig":
         )
 
     def area_names(self) -> t.List[str]:
-        return self.cache.get("%areas", list(self.areas.keys()))
+        return self.cache.get("%areas", list(self.areas))
 
     def set_names(self, only_output: bool = True) -> t.List[str]:
         return self.cache.get(
@@ -211,7 +228,16 @@ def get_st_storage_ids(self, area: str) -> t.List[str]:
         return self.cache.get(f"%st-storage%{area}", [s.id for s in self.areas[area].st_storages])
 
     def get_links(self, area: str) -> t.List[str]:
-        return self.cache.get(f"%links%{area}", list(self.areas[area].links.keys()))
+        return self.cache.get(f"%links%{area}", list(self.areas[area].links))
+
+    def get_binding_constraint_groups(self) -> t.List[str]:
+        """
+        Returns the list of binding constraint groups, without duplicates and
+        sorted alphabetically (case-insensitive).
+        Note that groups are stored in lower case in the binding constraints file.
+        """
+        lower_groups = {bc.group.lower(): bc.group for bc in self.bindings}
+        return self.cache.get("%binding-constraints", [grp for _, grp in sorted(lower_groups.items())])
 
     def get_filters_synthesis(self, area: str, link: t.Optional[str] = None) -> t.List[str]:
         if link:
@@ -260,7 +286,7 @@ class FileStudyTreeConfigDTO(BaseModel):
     bindings: t.List[BindingConstraintDTO] = list()
     store_new_set: bool = False
     archive_input_series: t.List[str] = list()
-    enr_modelling: str = EnrModelling.AGGREGATED.value
+    enr_modelling: str = str(EnrModelling.AGGREGATED)
     zip_path: t.Optional[Path] = None
 
     @staticmethod