feat(commands): add ST-Storage commands

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

@@ -26,7 +26,9 @@
     Link,
     Simulation,
     transform_name_to_id,
-    Storage,
+)
+from antarest.study.storage.rawstudy.model.filesystem.config.st_storage import (
+    STStorageConfig,
 )
 from antarest.study.storage.rawstudy.model.filesystem.root.settings.generaldata import (
     DUPLICATE_KEYS,
@@ -49,35 +51,38 @@ def build(
     study_path: Path, study_id: str, output_path: Optional[Path] = None
 ) -> "FileStudyTreeConfig":
     """
-    Extract data from filesystem to build config study.
-    Args:
-        study_path: study_path with files inside.
-        study_id: uuid of the study
-        output_path: output_path if not in study_path/output
+    Extracts data from the filesystem to build a study config.
 
-    Returns: study config fill with data
+    Args:
+        study_path: Path to the study directory or ZIP file containing the study.
+        study_id: UUID of the study.
+        output_path: Optional path for the output directory.
+            If not provided, it will be set to `{study_path}/output`.
 
+    Returns:
+        An instance of `FileStudyTreeConfig` filled with the study data.
     """
-    (sns, asi, enr_modelling) = _parse_parameters(study_path)
+    is_zip_file = study_path.suffix.lower() == ".zip"
 
-    study_path_without_zip_extension = study_path.parent / (
-        study_path.stem if study_path.suffix == ".zip" else study_path.name
-    )
+    # Study directory to use if the study is compressed
+    study_dir = study_path.with_suffix("") if is_zip_file else study_path
+    (sns, asi, enr_modelling) = _parse_parameters(study_path)
 
+    outputs_dir: Path = output_path or study_path / "output"
     return FileStudyTreeConfig(
         study_path=study_path,
-        output_path=output_path or study_path / "output",
-        path=study_path_without_zip_extension,
+        output_path=outputs_dir,
+        path=study_dir,
         study_id=study_id,
         version=_parse_version(study_path),
         areas=_parse_areas(study_path),
         sets=_parse_sets(study_path),
-        outputs=_parse_outputs(output_path or study_path / "output"),
+        outputs=_parse_outputs(outputs_dir),
         bindings=_parse_bindings(study_path),
         store_new_set=sns,
         archive_input_series=asi,
         enr_modelling=enr_modelling,
-        zip_path=study_path if study_path.suffix == ".zip" else None,
+        zip_path=study_path if is_zip_file else None,
     )
 
 
@@ -359,7 +364,7 @@ def parse_area(root: Path, area: str) -> "Area":
         renewables=_parse_renewables(root, area_id),
         filters_synthesis=_parse_filters_synthesis(root, area_id),
         filters_year=_parse_filters_year(root, area_id),
-        st_storage=_parse_st_storage(root, area_id),
+        st_storages=_parse_st_storage(root, area_id),
     )
 
 
@@ -379,21 +384,18 @@ def _parse_thermal(root: Path, area: str) -> List[Cluster]:
     ]
 
 
-def _parse_st_storage(root: Path, area: str) -> List[Storage]:
+def _parse_st_storage(root: Path, area: str) -> List[STStorageConfig]:
     """
     Parse the short-term storage INI file, return an empty list if missing.
     """
-    list_ini: Dict[str, Any] = _extract_data_from_file(
+    config_dict: Dict[str, Any] = _extract_data_from_file(
         root=root,
         inside_root_path=Path(f"input/st-storage/clusters/{area}/list.ini"),
         file_type=FileType.SIMPLE_INI,
     )
     return [
-        Storage(
-            id=transform_name_to_id(key),
-            name=values.get("name", key),
-        )
-        for key, values in list_ini.items()
+        STStorageConfig(**dict(values, id=storage_id))
+        for storage_id, values in config_dict.items()
     ]
 
 

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

@@ -3,10 +3,14 @@
 from pathlib import Path
 from typing import Dict, List, Optional, Set
 
+from pydantic import Extra
+
 from antarest.core.model import JSON
 from antarest.core.utils.utils import DTO
 from pydantic.main import BaseModel
 
+from .st_storage import STStorageConfig
+
 
 class ENR_MODELLING(Enum):
     AGGREGATED = "aggregated"
@@ -23,15 +27,6 @@ class Cluster(BaseModel):
     enabled: bool = True
 
 
-class Storage(BaseModel):
-    """
-    Short-term storage model used in Area creation
-    """
-
-    id: str
-    name: str
-
-
 class Link(BaseModel):
     """
     Object linked to /input/links/<link>/properties.ini information
@@ -59,14 +54,17 @@ class Area(BaseModel):
     Object linked to /input/<area>/optimization.ini information
     """
 
+    class Config:
+        extra = Extra.forbid
+
     name: str
     links: Dict[str, Link]
     thermals: List[Cluster]
     renewables: List[Cluster]
     filters_synthesis: List[str]
     filters_year: List[str]
     # since v8.6
-    storages: List[Storage] = []
+    st_storages: List[STStorageConfig] = []
 
 
 class DistrictSet(BaseModel):
@@ -143,14 +141,14 @@ def __init__(
         self.study_id = study_id
         self.version = version
         self.output_path = output_path
-        self.areas = areas or dict()
-        self.sets = sets or dict()
-        self.outputs = outputs or dict()
-        self.bindings = bindings or list()
+        self.areas = areas or {}
+        self.sets = sets or {}
+        self.outputs = outputs or {}
+        self.bindings = bindings or []
         self.store_new_set = store_new_set
-        self.archive_input_series = archive_input_series or list()
+        self.archive_input_series = archive_input_series or []
         self.enr_modelling = enr_modelling
-        self.cache = cache or dict()
+        self.cache = cache or {}
         self.zip_path = zip_path
 
     def next_file(
@@ -218,8 +216,7 @@ def get_thermal_names(
 
     def get_st_storage_names(self, area: str) -> List[str]:
         return self.cache.get(
-            f"%st-storage%{area}",
-            [storage.id for storage in self.areas[area].storages],
+            f"%st-storage%{area}", [s.id for s in self.areas[area].st_storages]
         )
 
     def get_renewable_names(

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

@@ -0,0 +1,115 @@
+from typing import Dict, Any
+
+from pydantic import BaseModel, Extra, Field, root_validator
+
+from antarest.study.business.enum_ignore_case import EnumIgnoreCase
+
+
+class STStorageGroup(EnumIgnoreCase):
+    """
+    This class defines the specific energy storage systems.
+
+    Enum values:
+
+        - PSP_OPEN: Represents an open pumped storage plant.
+        - PSP_CLOSED: Represents a closed pumped storage plant.
+        - PONDAGE: Represents a pondage storage system (reservoir storage system).
+        - BATTERY: Represents a battery storage system.
+        - OTHER: Represents other energy storage systems.
+    """
+
+    PSP_OPEN = "PSP_open"
+    PSP_CLOSED = "PSP_closed"
+    PONDAGE = "Pondage"
+    BATTERY = "Battery"
+    OTHER = "Other"
+
+
+# noinspection SpellCheckingInspection
+class STStorageConfig(BaseModel):
+    """
+    Manage the configuration files in the context of Short-Term Storage.
+    It provides a convenient way to read and write configuration data from/to an INI file format.
+    """
+
+    class Config:
+        extra = Extra.forbid
+        allow_population_by_field_name = True
+
+    # The `id` field is a calculated field that is excluded when converting
+    # the model to a dictionary or JSON format (`model_dump`).
+    id: str = Field(
+        description="Short-term storage ID",
+        regex=r"[a-zA-Z0-9_(),& -]+",
+        exclude=True,
+    )
+    name: str = Field(
+        description="Short-term storage name",
+        regex=r"[a-zA-Z0-9_(),& -]+",
+    )
+    group: STStorageGroup = Field(
+        ...,
+        description="Energy storage system group (mandatory)",
+    )
+    injection_nominal_capacity: float = Field(
+        0,
+        description="Injection nominal capacity (MW)",
+        ge=0,
+        alias="injectionnominalcapacity",
+    )
+    withdrawal_nominal_capacity: float = Field(
+        0,
+        description="Withdrawal nominal capacity (MW)",
+        ge=0,
+        alias="withdrawalnominalcapacity",
+    )
+    reservoir_capacity: float = Field(
+        0,
+        description="Reservoir capacity (MWh)",
+        ge=0,
+        alias="reservoircapacity",
+    )
+    efficiency: float = Field(
+        1,
+        description="Efficiency of the storage system",
+        ge=0,
+        le=1,
+    )
+    initial_level: float = Field(
+        0,
+        description="Initial level of the storage system",
+        ge=0,
+        alias="initiallevel",
+    )
+    initial_level_optim: bool = Field(
+        False,
+        description="Flag indicating if the initial level is optimized",
+        alias="initialleveloptim",
+    )
+
+    @root_validator(pre=True)
+    def calculate_storage_id(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Calculate the short-term storage ID based on the storage name, if not provided.
+
+        Args:
+            values: values used to construct the object.
+
+        Returns:
+            The updated values.
+        """
+        # Avoid circular imports
+        from antarest.study.storage.rawstudy.model.filesystem.config.model import (
+            transform_name_to_id,
+        )
+
+        if values.get("id") or not values.get("name"):
+            return values
+        storage_name = values["name"]
+        if storage_id := transform_name_to_id(storage_name):
+            values["id"] = storage_id
+        else:
+            raise ValueError(
+                f"Invalid short term storage name '{storage_name}'."
+            )
+        return values

antarest/study/storage/rawstudy/model/filesystem/root/input/st_storage/series/area/st_storage/st_storage.py

@@ -13,12 +13,12 @@
 class InputSTStorageAreaStorage(FolderNode):
     def build(self) -> TREE:
         children: TREE = {
-            "PMAX-injection": InputSeriesMatrix(
+            "pmax_injection": InputSeriesMatrix(
                 self.context,
                 self.config.next_file("PMAX-injection.txt"),
                 default_empty=series.pmax_injection,
             ),
-            "PMAX-withdrawal": InputSeriesMatrix(
+            "pmax_withdrawal": InputSeriesMatrix(
                 self.context,
                 self.config.next_file("PMAX-withdrawal.txt"),
                 default_empty=series.pmax_withdrawal,
@@ -28,12 +28,12 @@ def build(self) -> TREE:
                 self.config.next_file("inflows.txt"),
                 default_empty=series.inflows,
             ),
-            "lower-rule-curve": InputSeriesMatrix(
+            "lower_rule_curve": InputSeriesMatrix(
                 self.context,
                 self.config.next_file("lower-rule-curve.txt"),
                 default_empty=series.lower_rule_curve,
             ),
-            "upper-rule-curve": InputSeriesMatrix(
+            "upper_rule_curve": InputSeriesMatrix(
                 self.context,
                 self.config.next_file("upper-rule-curve.txt"),
                 default_empty=series.upper_rule_curve,

antarest/study/storage/variantstudy/business/command_reverter.py

@@ -33,6 +33,9 @@
 from antarest.study.storage.variantstudy.model.command.create_renewables_cluster import (
     CreateRenewablesCluster,
 )
+from antarest.study.storage.variantstudy.model.command.create_st_storage import (
+    CreateSTStorage,
+)
 from antarest.study.storage.variantstudy.model.command.icommand import ICommand
 from antarest.study.storage.variantstudy.model.command.remove_area import (
     RemoveArea,
@@ -52,6 +55,9 @@
 from antarest.study.storage.variantstudy.model.command.remove_renewables_cluster import (
     RemoveRenewablesCluster,
 )
+from antarest.study.storage.variantstudy.model.command.remove_st_storage import (
+    RemoveSTStorage,
+)
 from antarest.study.storage.variantstudy.model.command.replace_matrix import (
     ReplaceMatrix,
 )
@@ -267,6 +273,31 @@ def _revert_remove_renewables_cluster(
             "The revert function for RemoveRenewablesCluster is not available"
         )
 
+    @staticmethod
+    def _revert_create_st_storage(
+        base_command: CreateSTStorage,
+        history: List["ICommand"],
+        base: FileStudy,
+    ) -> List[ICommand]:
+        storage_id = base_command.parameters.id
+        return [
+            RemoveSTStorage(
+                area_id=base_command.area_id,
+                storage_id=storage_id,
+                command_context=base_command.command_context,
+            )
+        ]
+
+    @staticmethod
+    def _revert_remove_st_storage(
+        base_command: RemoveSTStorage,
+        history: List["ICommand"],
+        base: FileStudy,
+    ) -> List[ICommand]:
+        raise NotImplementedError(
+            "The revert function for RemoveSTStorage is not available"
+        )
+
     @staticmethod
     def _revert_replace_matrix(
         base_command: ReplaceMatrix, history: List["ICommand"], base: FileStudy

antarest/study/storage/variantstudy/business/matrix_constants/__init__.py

@@ -1 +1 @@
-from . import hydro, prepro, thermals, link
+from . import hydro, prepro, thermals, link, st_storage

antarest/study/storage/variantstudy/business/matrix_constants/st_storage/__init__.py

@@ -0,0 +1 @@
+from . import series