feat(clusters): add new endpoint for clusters duplication

antarest/core/exceptions.py

@@ -312,3 +312,13 @@ def __init__(self, area_id: str) -> None:
             HTTPStatus.NOT_FOUND,
             f"Cluster configuration for area: '{area_id}' not found",
         )
+
+
+class ClusterAlreadyExists(HTTPException):
+    """Exception raised when attempting to create a cluster with an already existing ID."""
+
+    def __init__(self, cluster_type: str, cluster_id: str) -> None:
+        super().__init__(
+            HTTPStatus.CONFLICT,
+            f"{cluster_type} cluster with ID '{cluster_id}' already exists and could not be created.",
+        )

antarest/study/business/areas/renewable_management.py

@@ -3,10 +3,11 @@
 
 from pydantic import validator
 
-from antarest.core.exceptions import ClusterConfigNotFound, ClusterNotFound
+from antarest.core.exceptions import ClusterAlreadyExists, ClusterConfigNotFound, ClusterNotFound
 from antarest.study.business.enum_ignore_case import EnumIgnoreCase
 from antarest.study.business.utils import AllOptionalMetaclass, camel_case_model, execute_or_add_commands
 from antarest.study.model import Study
+from antarest.study.storage.rawstudy.model.filesystem.config.model import transform_name_to_id
 from antarest.study.storage.rawstudy.model.filesystem.config.renewable import (
     RenewableConfig,
     RenewableConfigType,
@@ -17,6 +18,7 @@
 from antarest.study.storage.storage_service import StudyStorageService
 from antarest.study.storage.variantstudy.model.command.create_renewables_cluster import CreateRenewablesCluster
 from antarest.study.storage.variantstudy.model.command.remove_renewables_cluster import RemoveRenewablesCluster
+from antarest.study.storage.variantstudy.model.command.replace_matrix import ReplaceMatrix
 from antarest.study.storage.variantstudy.model.command.update_config import UpdateConfig
 
 __all__ = (
@@ -47,7 +49,7 @@ class Config:
         def schema_extra(schema: t.MutableMapping[str, t.Any]) -> None:
             schema["example"] = RenewableClusterInput(
                 group="Gas",
-                name="2 avail and must 1",
+                name="Gas Cluster XY",
                 enabled=False,
                 unitCount=100,
                 nominalCapacity=1000.0,
@@ -85,9 +87,9 @@ class Config:
         @staticmethod
         def schema_extra(schema: t.MutableMapping[str, t.Any]) -> None:
             schema["example"] = RenewableClusterOutput(
-                id="2 avail and must 1",
+                id="Gas cluster YZ",
                 group="Gas",
-                name="2 avail and must 1",
+                name="Gas Cluster YZ",
                 enabled=False,
                 unitCount=100,
                 nominalCapacity=1000.0,
@@ -157,23 +159,25 @@ def create_cluster(
             The newly created cluster.
         """
         file_study = self._get_file_study(study)
-        study_version = study.version
-        cluster = cluster_data.to_config(study_version)
-
-        command = CreateRenewablesCluster(
-            area_id=area_id,
-            cluster_name=cluster.id,
-            parameters=cluster.dict(by_alias=True, exclude={"id"}),
-            command_context=self.storage_service.variant_study_service.command_factory.command_context,
-        )
+        cluster = cluster_data.to_config(study.version)
+        command = self._make_create_cluster_cmd(area_id, cluster)
         execute_or_add_commands(
             study,
             file_study,
             [command],
             self.storage_service,
         )
+        output = self.get_cluster(study, area_id, cluster.id)
+        return output
 
-        return self.get_cluster(study, area_id, cluster.id)
+    def _make_create_cluster_cmd(self, area_id: str, cluster: RenewableConfigType) -> CreateRenewablesCluster:
+        command = CreateRenewablesCluster(
+            area_id=area_id,
+            cluster_name=cluster.id,
+            parameters=cluster.dict(by_alias=True, exclude={"id"}),
+            command_context=self.storage_service.variant_study_service.command_factory.command_context,
+        )
+        return command
 
     def get_cluster(self, study: Study, area_id: str, cluster_id: str) -> RenewableClusterOutput:
         """
@@ -273,3 +277,53 @@ def delete_clusters(self, study: Study, area_id: str, cluster_ids: t.Sequence[st
         ]
 
         execute_or_add_commands(study, file_study, commands, self.storage_service)
+
+    def duplicate_cluster(
+        self,
+        study: Study,
+        area_id: str,
+        source_id: str,
+        new_cluster_name: str,
+    ) -> RenewableClusterOutput:
+        """
+        Creates a duplicate cluster within the study area with a new name.
+
+        Args:
+            study: The study in which the cluster will be duplicated.
+            area_id: The identifier of the area where the cluster will be duplicated.
+            source_id: The identifier of the cluster to be duplicated.
+            new_cluster_name: The new name for the duplicated cluster.
+
+        Returns:
+            The duplicated cluster configuration.
+
+        Raises:
+            ClusterAlreadyExists: If a cluster with the new name already exists in the area.
+        """
+        new_id = transform_name_to_id(new_cluster_name, lower=False)
+        lower_new_id = new_id.lower()
+        if any(lower_new_id == cluster.id.lower() for cluster in self.get_clusters(study, area_id)):
+            raise ClusterAlreadyExists("Renewable", new_id)
+
+        # Cluster duplication
+        current_cluster = self.get_cluster(study, area_id, source_id)
+        current_cluster.name = new_cluster_name
+        creation_form = RenewableClusterCreation(**current_cluster.dict(by_alias=False, exclude={"id"}))
+        new_config = creation_form.to_config(study.version)
+        create_cluster_cmd = self._make_create_cluster_cmd(area_id, new_config)
+
+        # Matrix edition
+        lower_source_id = source_id.lower()
+        source_path = f"input/renewables/series/{area_id}/{lower_source_id}/series"
+        new_path = f"input/renewables/series/{area_id}/{lower_new_id}/series"
+
+        # Prepare and execute commands
+        storage_service = self.storage_service.get_storage(study)
+        command_context = self.storage_service.variant_study_service.command_factory.command_context
+        current_matrix = storage_service.get(study, source_path)["data"]
+        replace_matrix_cmd = ReplaceMatrix(target=new_path, matrix=current_matrix, command_context=command_context)
+        commands = [create_cluster_cmd, replace_matrix_cmd]
+
+        execute_or_add_commands(study, self._get_file_study(study), commands, self.storage_service)
+
+        return RenewableClusterOutput(**new_config.dict(by_alias=False))

antarest/study/business/areas/st_storage_management.py

@@ -8,12 +8,14 @@
 from typing_extensions import Literal
 
 from antarest.core.exceptions import (
+    ClusterAlreadyExists,
     STStorageConfigNotFoundError,
     STStorageFieldsNotFoundError,
     STStorageMatrixNotFoundError,
 )
 from antarest.study.business.utils import AllOptionalMetaclass, camel_case_model, execute_or_add_commands
 from antarest.study.model import Study
+from antarest.study.storage.rawstudy.model.filesystem.config.model import transform_name_to_id
 from antarest.study.storage.rawstudy.model.filesystem.config.st_storage import (
     STStorageConfig,
     STStorageGroup,
@@ -24,6 +26,7 @@
 from antarest.study.storage.storage_service import StudyStorageService
 from antarest.study.storage.variantstudy.model.command.create_st_storage import CreateSTStorage
 from antarest.study.storage.variantstudy.model.command.remove_st_storage import RemoveSTStorage
+from antarest.study.storage.variantstudy.model.command.replace_matrix import ReplaceMatrix
 from antarest.study.storage.variantstudy.model.command.update_config import UpdateConfig
 
 __all__ = (
@@ -72,8 +75,8 @@ def validate_name(cls, name: t.Optional[str]) -> str:
             raise ValueError("'name' must not be empty")
         return name
 
-    @property
-    def to_config(self) -> STStorageConfig:
+    # noinspection PyUnusedLocal
+    def to_config(self, study_version: t.Union[str, int]) -> STStorageConfig:
         values = self.dict(by_alias=False, exclude_none=True)
         return STStorageConfig(**values)
 
@@ -203,7 +206,7 @@ def validate_rule_curve(
             upper_array = np.array(upper_rule_curve.data, dtype=np.float64)
             # noinspection PyUnresolvedReferences
             if (lower_array > upper_array).any():
-                raise ValueError("Each 'lower_rule_curve' value must be lower" " or equal to each 'upper_rule_curve'")
+                raise ValueError("Each 'lower_rule_curve' value must be lower or equal to each 'upper_rule_curve'")
         return values
 
 
@@ -257,21 +260,25 @@ def create_storage(
         Returns:
             The ID of the newly created short-term storage.
         """
-        storage = form.to_config
-        command = CreateSTStorage(
-            area_id=area_id,
-            parameters=storage,
-            command_context=self.storage_service.variant_study_service.command_factory.command_context,
-        )
         file_study = self._get_file_study(study)
+        storage = form.to_config(study.version)
+        command = self._make_create_cluster_cmd(area_id, storage)
         execute_or_add_commands(
             study,
             file_study,
             [command],
             self.storage_service,
         )
+        output = self.get_storage(study, area_id, storage_id=storage.id)
+        return output
 
-        return self.get_storage(study, area_id, storage_id=storage.id)
+    def _make_create_cluster_cmd(self, area_id: str, cluster: STStorageConfig) -> CreateSTStorage:
+        command = CreateSTStorage(
+            area_id=area_id,
+            parameters=cluster,
+            command_context=self.storage_service.variant_study_service.command_factory.command_context,
+        )
+        return command
 
     def get_storages(
         self,
@@ -418,6 +425,59 @@ def delete_storages(
             file_study = self._get_file_study(study)
             execute_or_add_commands(study, file_study, [command], self.storage_service)
 
+    def duplicate_cluster(self, study: Study, area_id: str, source_id: str, new_cluster_name: str) -> STStorageOutput:
+        """
+        Creates a duplicate cluster within the study area with a new name.
+
+        Args:
+            study: The study in which the cluster will be duplicated.
+            area_id: The identifier of the area where the cluster will be duplicated.
+            source_id: The identifier of the cluster to be duplicated.
+            new_cluster_name: The new name for the duplicated cluster.
+
+        Returns:
+            The duplicated cluster configuration.
+
+        Raises:
+            ClusterAlreadyExists: If a cluster with the new name already exists in the area.
+        """
+        new_id = transform_name_to_id(new_cluster_name)
+        lower_new_id = new_id.lower()
+        if any(lower_new_id == storage.id.lower() for storage in self.get_storages(study, area_id)):
+            raise ClusterAlreadyExists("Short-term storage", new_id)
+
+        # Cluster duplication
+        current_cluster = self.get_storage(study, area_id, source_id)
+        current_cluster.name = new_cluster_name
+        creation_form = STStorageCreation(**current_cluster.dict(by_alias=False, exclude={"id"}))
+        new_config = creation_form.to_config(study.version)
+        create_cluster_cmd = self._make_create_cluster_cmd(area_id, new_config)
+
+        # Matrix edition
+        lower_source_id = source_id.lower()
+        ts_names = ["pmax_injection", "pmax_withdrawal", "lower_rule_curve", "upper_rule_curve", "inflows"]
+        source_paths = [
+            STORAGE_SERIES_PATH.format(area_id=area_id, storage_id=lower_source_id, ts_name=ts_name)
+            for ts_name in ts_names
+        ]
+        new_paths = [
+            STORAGE_SERIES_PATH.format(area_id=area_id, storage_id=lower_new_id, ts_name=ts_name)
+            for ts_name in ts_names
+        ]
+
+        # Prepare and execute commands
+        commands: t.List[t.Union[CreateSTStorage, ReplaceMatrix]] = [create_cluster_cmd]
+        storage_service = self.storage_service.get_storage(study)
+        command_context = self.storage_service.variant_study_service.command_factory.command_context
+        for source_path, new_path in zip(source_paths, new_paths):
+            current_matrix = storage_service.get(study, source_path)["data"]
+            command = ReplaceMatrix(target=new_path, matrix=current_matrix, command_context=command_context)
+            commands.append(command)
+
+        execute_or_add_commands(study, self._get_file_study(study), commands, self.storage_service)
+
+        return STStorageOutput(**new_config.dict(by_alias=False))
+
     def get_matrix(
         self,
         study: Study,
@@ -484,12 +544,11 @@ def _save_matrix_obj(
         ts_name: STStorageTimeSeries,
         matrix_obj: t.Dict[str, t.Any],
     ) -> None:
-        file_study = self._get_file_study(study)
         path = STORAGE_SERIES_PATH.format(area_id=area_id, storage_id=storage_id, ts_name=ts_name)
-        try:
-            file_study.tree.save(matrix_obj, path.split("/"))
-        except KeyError:
-            raise STStorageMatrixNotFoundError(study.id, area_id, storage_id, ts_name) from None
+        matrix = matrix_obj["data"]
+        command_context = self.storage_service.variant_study_service.command_factory.command_context
+        command = ReplaceMatrix(target=path, matrix=matrix, command_context=command_context)
+        execute_or_add_commands(study, self._get_file_study(study), [command], self.storage_service)
 
     def validate_matrices(
         self,