v2.16.8

antarest/__init__.py

@@ -7,9 +7,9 @@
 
 # Standard project metadata
 
-__version__ = "2.16.7"
+__version__ = "2.16.8"
 __author__ = "RTE, Antares Web Team"
-__date__ = "2024-03-05"
+__date__ = "2024-04-19"
 # noinspection SpellCheckingInspection
 __credits__ = "(c) Réseau de Transport de l’Électricité (RTE)"
 

antarest/core/exceptions.py

@@ -34,13 +34,35 @@ class STStorageConfigNotFoundError(HTTPException):
     """Configuration for short-term storage is not found"""
 
     def __init__(self, study_id: str, area_id: str) -> None:
-        detail = f"The short-term storage configuration of area '{area_id}' not found:"
+        detail = f"The short-term storage configuration of area '{area_id}' not found"
         super().__init__(HTTPStatus.NOT_FOUND, detail)
 
     def __str__(self) -> str:
         return self.detail
 
 
+class STStorageNotFoundError(HTTPException):
+    """Short-term storage is not found"""
+
+    def __init__(self, study_id: str, area_id: str, st_storage_id: str) -> None:
+        detail = f"Short-term storage '{st_storage_id}' not found in area '{area_id}'"
+        super().__init__(HTTPStatus.NOT_FOUND, detail)
+
+    def __str__(self) -> str:
+        return self.detail
+
+
+class DuplicateSTStorageId(HTTPException):
+    """Exception raised when trying to create a short-term storage with an already existing id."""
+
+    def __init__(self, study_id: str, area_id: str, st_storage_id: str) -> None:
+        detail = f"Short term storage '{st_storage_id}' already exists in area '{area_id}'"
+        super().__init__(HTTPStatus.CONFLICT, detail)
+
+    def __str__(self) -> str:
+        return self.detail
+
+
 class UnknownModuleError(Exception):
     def __init__(self, message: str) -> None:
         super(UnknownModuleError, self).__init__(message)
@@ -304,3 +326,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))