feat(bc): add constraint duplication endpoint (#2295)

antarest/study/business/areas/st_storage_management.py

@@ -548,7 +548,7 @@ def duplicate_cluster(self, study: Study, area_id: str, source_id: str, new_clus
             The duplicated cluster configuration.
 
         Raises:
-            ClusterAlreadyExists: If a cluster with the new name already exists in the area.
+            DuplicateSTStorage: 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()

antarest/study/business/areas/thermal_management.py

@@ -429,7 +429,7 @@ def duplicate_cluster(
             The duplicated cluster configuration.
 
         Raises:
-            ClusterAlreadyExists: If a cluster with the new name already exists in the area.
+            DuplicateThermalCluster: 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()

antarest/study/business/binding_constraint_management.py

@@ -795,6 +795,75 @@ def create_binding_constraint(
         new_constraint["id"] = bc_id
         return self.constraint_model_adapter(new_constraint, version)
 
+    def duplicate_binding_constraint(self, study: Study, source_id: str, new_constraint_name: str) -> ConstraintOutput:
+        """
+        Creates a duplicate constraint with a new name.
+
+        Args:
+            study: The study in which the cluster will be duplicated.
+            source_id: The identifier of the constraint to be duplicated.
+            new_constraint_name: The new name for the duplicated constraint.
+
+        Returns:
+            The duplicated constraint configuration.
+
+        Raises:
+            DuplicateConstraintName: If a constraint with the new name already exists in the study.
+        """
+
+        # Checks if the new constraint already exists
+        new_constraint_id = transform_name_to_id(new_constraint_name)
+        existing_constraints = self.get_binding_constraints(study)
+        if new_constraint_id in {bc.id for bc in existing_constraints}:
+            raise DuplicateConstraintName(
+                f"A binding constraint with the same name already exists: {new_constraint_name}."
+            )
+
+        # Retrieval of the source constraint properties
+        source_constraint = next(iter(bc for bc in existing_constraints if bc.id == source_id), None)
+        if not source_constraint:
+            raise BindingConstraintNotFound(f"Binding constraint '{source_id}' not found")
+
+        new_constraint = {
+            "name": new_constraint_name,
+            **source_constraint.model_dump(mode="json", exclude={"terms", "name", "id"}),
+        }
+        args = {
+            **new_constraint,
+            "command_context": self.storage_service.variant_study_service.command_factory.command_context,
+            "study_version": StudyVersion.parse(study.version),
+        }
+        if source_constraint.terms:
+            args["coeffs"] = self.terms_to_coeffs(source_constraint.terms)
+
+        # Retrieval of the source constraint matrices
+        file_study = self.storage_service.get_storage(study).get_raw(study)
+        if file_study.config.version < STUDY_VERSION_8_7:
+            matrix = file_study.tree.get(["input", "bindingconstraints", source_id])
+            args["values"] = matrix["data"]
+        else:
+            correspondence_map = {
+                "lt": TermMatrices.LESS.value,
+                "gt": TermMatrices.GREATER.value,
+                "eq": TermMatrices.EQUAL.value,
+            }
+            source_matrices = OPERATOR_MATRIX_FILE_MAP[source_constraint.operator]
+            for matrix_name in source_matrices:
+                matrix = file_study.tree.get(["input", "bindingconstraints", matrix_name.format(bc_id=source_id)])[
+                    "data"
+                ]
+                command_attribute = correspondence_map[matrix_name.removeprefix("{bc_id}_")]
+                args[command_attribute] = matrix
+
+        # Creates and applies constraint
+        command = CreateBindingConstraint(**args)
+        execute_or_add_commands(study, file_study, [command], self.storage_service)
+
+        # Returns the new constraint
+        source_constraint.name = new_constraint_name
+        source_constraint.id = new_constraint_id
+        return source_constraint
+
     def update_binding_constraint(
         self,
         study: Study,

antarest/study/web/study_data_blueprint.py

@@ -1340,6 +1340,27 @@ def create_binding_constraint(
         study = study_service.check_study_access(uuid, StudyPermissionType.READ, params)
         return study_service.binding_constraint_manager.create_binding_constraint(study, data)
 
+    @bp.post(
+        "/studies/{uuid}/bindingconstraints/{binding_constraint_id}",
+        tags=[APITag.study_data],
+        summary="Duplicates a given binding constraint",
+    )
+    def duplicate_binding_constraint(
+        uuid: str,
+        binding_constraint_id: str,
+        new_constraint_name: str,
+        current_user: JWTUser = Depends(auth.get_current_user),
+    ) -> ConstraintOutput:
+        logger.info(
+            f"Duplicates constraint {binding_constraint_id} for study {uuid}",
+            extra={"user": current_user.id},
+        )
+        params = RequestParameters(user=current_user)
+        study = study_service.check_study_access(uuid, StudyPermissionType.WRITE, params)
+        return study_service.binding_constraint_manager.duplicate_binding_constraint(
+            study, binding_constraint_id, new_constraint_name
+        )
+
     @bp.delete(
         "/studies/{uuid}/bindingconstraints/{binding_constraint_id}",
         tags=[APITag.study_data],

tests/integration/study_data_blueprint/test_binding_constraints.py

@@ -462,10 +462,61 @@ def test_lifecycle__nominal(self, client: TestClient, user_access_token: str, st
         assert len(dataframe["index"]) == 366
         assert len(dataframe["columns"]) == 3  # less, equal, greater
 
+        # =============================
+        # CONSTRAINT DUPLICATION
+        # =============================
+
+        # Change source constraint matrix to ensure it will be copied correctly
+        new_matrix = np.ones((366, 3)).tolist()
+        res = client.post(
+            f"/v1/studies/{study_id}/raw", params={"path": f"input/bindingconstraints/{bc_id}"}, json=new_matrix
+        )
+        res.raise_for_status()
+
+        # Get the source constraint properties to ensure there are copied correctly
+        res = client.get(f"/v1/studies/{study_id}/bindingconstraints/{bc_id}")
+        res.raise_for_status()
+        current_constraint = res.json()
+        current_constraint.pop("name")
+        current_constraint.pop("id")
+
+        # Duplicates the constraint
+        duplicated_name = "BC_4"
+        res = client.post(
+            f"/v1/studies/{study_id}/bindingconstraints/{bc_id}", params={"new_constraint_name": duplicated_name}
+        )
+        res.raise_for_status()
+        duplicated_constraint = res.json()
+
+        # Asserts the duplicated constraint has the right name and the right properties
+        assert duplicated_constraint.pop("name") == duplicated_name
+        new_id = duplicated_constraint.pop("id")
+        assert current_constraint == duplicated_constraint
+
+        # Asserts the matrix is duplicated correctly
+        res = client.get(f"/v1/studies/{study_id}/raw", params={"path": f"input/bindingconstraints/{new_id}"})
+        res.raise_for_status()
+        assert res.json()["data"] == new_matrix
+
         # =============================
         # ERRORS
         # =============================
 
+        # Asserts duplication fails if given an non-exisiting constraint
+        fake_name = "fake_name"
+        res = client.post(
+            f"/v1/studies/{study_id}/bindingconstraints/{fake_name}", params={"new_constraint_name": "aa"}
+        )
+        assert res.status_code == 404
+        assert res.json()["exception"] == "BindingConstraintNotFound"
+        assert res.json()["description"] == f"Binding constraint '{fake_name}' not found"
+
+        # Asserts duplication fails if given an already existing name
+        res = client.post(f"/v1/studies/{study_id}/bindingconstraints/{bc_id}", params={"new_constraint_name": bc_id})
+        assert res.status_code == 409
+        assert res.json()["exception"] == "DuplicateConstraintName"
+        assert res.json()["description"] == f"A binding constraint with the same name already exists: {bc_id}."
+
         # Assert empty name
         res = client.post(
             f"/v1/studies/{study_id}/bindingconstraints",
@@ -921,6 +972,45 @@ def test_for_version_870(self, client: TestClient, user_access_token: str, study
         binding_constraints_list = preparer.get_binding_constraints(study_id)
         assert len(binding_constraints_list) == 2
 
+        # =============================
+        # CONSTRAINT DUPLICATION
+        # =============================
+
+        # Change source constraint matrix to ensure it will be copied correctly
+        new_matrix = np.ones((366, 1)).tolist()
+        res = client.post(
+            f"/v1/studies/{study_id}/raw",
+            params={"path": f"input/bindingconstraints/{bc_id_w_matrix}_lt"},
+            json=new_matrix,
+        )
+        res.raise_for_status()
+
+        # Get the source constraint properties to ensure there are copied correctly
+        res = client.get(f"/v1/studies/{study_id}/bindingconstraints/{bc_id_w_matrix}")
+        res.raise_for_status()
+        current_constraint = res.json()
+        current_constraint.pop("name")
+        current_constraint.pop("id")
+
+        # Duplicates the constraint
+        duplicated_name = "BC_4"
+        res = client.post(
+            f"/v1/studies/{study_id}/bindingconstraints/{bc_id_w_matrix}",
+            params={"new_constraint_name": duplicated_name},
+        )
+        res.raise_for_status()
+        duplicated_constraint = res.json()
+
+        # Asserts the duplicated constraint has the right name and the right properties
+        assert duplicated_constraint.pop("name") == duplicated_name
+        new_id = duplicated_constraint.pop("id")
+        assert current_constraint == duplicated_constraint
+
+        # Asserts the matrix is duplicated correctly
+        res = client.get(f"/v1/studies/{study_id}/raw", params={"path": f"input/bindingconstraints/{new_id}_lt"})
+        res.raise_for_status()
+        assert res.json()["data"] == new_matrix
+
         # =============================
         #  ERRORS
         # =============================