Merge branch 'release/2.17.0' into dev

antarest/__init__.py

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

antarest/core/exceptions.py

@@ -376,11 +376,6 @@ def __init__(self, message: str) -> None:
         super().__init__(HTTPStatus.NOT_FOUND, message)
 
 
-class ConstraintAlreadyExistError(HTTPException):
-    def __init__(self, message: str) -> None:
-        super().__init__(HTTPStatus.NOT_FOUND, message)
-
-
 class DuplicateConstraintName(HTTPException):
     def __init__(self, message: str) -> None:
         super().__init__(HTTPStatus.CONFLICT, message)
@@ -406,14 +401,61 @@ def __init__(self, message: str) -> None:
         super().__init__(HTTPStatus.UNPROCESSABLE_ENTITY, message)
 
 
-class MissingDataError(HTTPException):
-    def __init__(self, message: str) -> None:
+class ConstraintTermNotFound(HTTPException):
+    """
+    Exception raised when a constraint term is not found.
+    """
+
+    def __init__(self, binding_constraint_id: str, *ids: str) -> None:
+        count = len(ids)
+        id_enum = ", ".join(f"'{term}'" for term in ids)
+        message = {
+            0: f"Constraint terms not found in BC '{binding_constraint_id}'",
+            1: f"Constraint term {id_enum} not found in BC '{binding_constraint_id}'",
+            2: f"Constraint terms {id_enum} not found in BC '{binding_constraint_id}'",
+        }[min(count, 2)]
         super().__init__(HTTPStatus.NOT_FOUND, message)
 
+    def __str__(self) -> str:
+        """Return a string representation of the exception."""
+        return self.detail
 
-class ConstraintIdNotFoundError(HTTPException):
-    def __init__(self, message: str) -> None:
-        super().__init__(HTTPStatus.NOT_FOUND, message)
+
+class DuplicateConstraintTerm(HTTPException):
+    """
+    Exception raised when an attempt is made to create a constraint term which already exists.
+    """
+
+    def __init__(self, binding_constraint_id: str, *ids: str) -> None:
+        count = len(ids)
+        id_enum = ", ".join(f"'{term}'" for term in ids)
+        message = {
+            0: f"Constraint terms already exist in BC '{binding_constraint_id}'",
+            1: f"Constraint term {id_enum} already exists in BC '{binding_constraint_id}'",
+            2: f"Constraint terms {id_enum} already exist in BC '{binding_constraint_id}'",
+        }[min(count, 2)]
+        super().__init__(HTTPStatus.CONFLICT, message)
+
+    def __str__(self) -> str:
+        """Return a string representation of the exception."""
+        return self.detail
+
+
+class InvalidConstraintTerm(HTTPException):
+    """
+    Exception raised when a constraint term is not correctly specified (no term data).
+    """
+
+    def __init__(self, binding_constraint_id: str, term_json: str) -> None:
+        message = (
+            f"Invalid constraint term for binding constraint '{binding_constraint_id}': {term_json},"
+            f" term 'data' is missing or empty"
+        )
+        super().__init__(HTTPStatus.UNPROCESSABLE_ENTITY, message)
+
+    def __str__(self) -> str:
+        """Return a string representation of the exception."""
+        return self.detail
 
 
 class LayerNotFound(HTTPException):

antarest/study/business/advanced_parameters_management.py

@@ -91,7 +91,7 @@ def check_accuracy_on_correlation(cls, v: str) -> str:
             return ""
 
         allowed_values = ["wind", "load", "solar"]
-        values_list = re.split("\s*,\s*", v.strip())
+        values_list = re.split(r"\s*,\s*", v.strip())
 
         if len(values_list) != len(set(values_list)):
             raise ValueError("Duplicate value")

antarest/study/business/area_management.py

@@ -232,8 +232,11 @@ def _to_optimization(self) -> OptimizationProperties:
             nodal_optimization=nodal_optimization_section,
         )
 
-    def _to_adequacy_patch(self) -> AdequacyPathProperties:
+    def _to_adequacy_patch(self) -> t.Optional[AdequacyPathProperties]:
         obj = {name: getattr(self, name) for name in AdequacyPathProperties.AdequacyPathSection.__fields__}
+        # If all fields are `None`, the object is empty.
+        if all(value is None for value in obj.values()):
+            return None
         adequacy_path_section = AdequacyPathProperties.AdequacyPathSection(**obj)
         return AdequacyPathProperties(adequacy_patch=adequacy_path_section)
 

antarest/study/business/binding_constraint_management.py

@@ -10,14 +10,13 @@
 
 from antarest.core.exceptions import (
     BindingConstraintNotFound,
-    ConstraintAlreadyExistError,
-    ConstraintIdNotFoundError,
+    ConstraintTermNotFound,
     DuplicateConstraintName,
+    DuplicateConstraintTerm,
     InvalidConstraintName,
+    InvalidConstraintTerm,
     InvalidFieldForVersionError,
     MatrixWidthMismatchError,
-    MissingDataError,
-    NoConstraintError,
     WrongMatrixHeightError,
 )
 from antarest.core.model import JSON
@@ -805,90 +804,92 @@ def remove_binding_constraint(self, study: Study, binding_constraint_id: str) ->
         command = RemoveBindingConstraint(id=bc.id, command_context=command_context)
         execute_or_add_commands(study, file_study, [command], self.storage_service)
 
-    def update_constraint_term(
-        self,
-        study: Study,
-        binding_constraint_id: str,
-        term: ConstraintTerm,
+    def _update_constraint_with_terms(
+        self, study: Study, bc: ConstraintOutput, terms: t.Mapping[str, ConstraintTerm]
     ) -> None:
-        file_study = self.storage_service.get_storage(study).get_raw(study)
-        constraint = self.get_binding_constraint(study, binding_constraint_id)
-        constraint_terms = constraint.terms  # existing constraint terms
-        if not constraint_terms:
-            raise NoConstraintError(study.id)
-
-        term_id = term.id if isinstance(term, ConstraintTerm) else term
-        if term_id is None:
-            raise ConstraintIdNotFoundError(study.id)
-
-        term_id_index = find_constraint_term_id(constraint_terms, term_id)
-        if term_id_index < 0:
-            raise ConstraintIdNotFoundError(study.id)
-
-        if isinstance(term, ConstraintTerm):
-            updated_term_id = term.data.generate_id() if term.data else term_id
-            current_constraint = constraint_terms[term_id_index]
-
-            constraint_terms[term_id_index] = ConstraintTerm(
-                id=updated_term_id,
-                weight=term.weight or current_constraint.weight,
-                offset=term.offset,
-                data=term.data or current_constraint.data,
-            )
-        else:
-            del constraint_terms[term_id_index]
-
-        coeffs = {term.id: [term.weight, term.offset] if term.offset else [term.weight] for term in constraint_terms}
-
+        coeffs = {
+            term_id: [term.weight, term.offset] if term.offset else [term.weight] for term_id, term in terms.items()
+        }
         command = UpdateBindingConstraint(
-            id=constraint.id,
+            id=bc.id,
             coeffs=coeffs,
             command_context=self.storage_service.variant_study_service.command_factory.command_context,
         )
+        file_study = self.storage_service.get_storage(study).get_raw(study)
         execute_or_add_commands(study, file_study, [command], self.storage_service)
 
-    def create_constraint_term(
+    def update_constraint_terms(
         self,
         study: Study,
         binding_constraint_id: str,
-        constraint_term: ConstraintTerm,
+        constraint_terms: t.Sequence[ConstraintTerm],
+        update_mode: str = "replace",
     ) -> None:
-        file_study = self.storage_service.get_storage(study).get_raw(study)
-        constraint = self.get_binding_constraint(study, binding_constraint_id)
-
-        if constraint_term.data is None:
-            raise MissingDataError("Add new constraint term : data is missing")
+        """
+        Update or add the specified constraint terms.
 
-        constraint_id = constraint_term.data.generate_id()
-        constraint_terms = constraint.terms or []
-        if find_constraint_term_id(constraint_terms, constraint_id) >= 0:
-            raise ConstraintAlreadyExistError(study.id)
+        Args:
+            study: The study from which to update the binding constraint.
+            binding_constraint_id: The ID of the binding constraint to update.
+            constraint_terms: The constraint terms to update.
+            update_mode: The update mode, either "replace" or "add".
+        """
+        if update_mode == "add":
+            for term in constraint_terms:
+                if term.data is None:
+                    raise InvalidConstraintTerm(binding_constraint_id, term.json())
 
-        constraint_terms.append(
-            ConstraintTerm(
-                id=constraint_id,
-                weight=constraint_term.weight if constraint_term.weight is not None else 0.0,
-                offset=constraint_term.offset,
-                data=constraint_term.data,
-            )
-        )
+        constraint = self.get_binding_constraint(study, binding_constraint_id)
+        existing_terms = collections.OrderedDict((term.generate_id(), term) for term in constraint.terms)
+        updated_terms = collections.OrderedDict((term.generate_id(), term) for term in constraint_terms)
+
+        if update_mode == "replace":
+            missing_terms = set(updated_terms) - set(existing_terms)
+            if missing_terms:
+                raise ConstraintTermNotFound(binding_constraint_id, *missing_terms)
+        elif update_mode == "add":
+            duplicate_terms = set(updated_terms) & set(existing_terms)
+            if duplicate_terms:
+                raise DuplicateConstraintTerm(binding_constraint_id, *duplicate_terms)
+        else:  # pragma: no cover
+            raise NotImplementedError(f"Unsupported update mode: {update_mode}")
+
+        existing_terms.update(updated_terms)
+        self._update_constraint_with_terms(study, constraint, existing_terms)
+
+    def create_constraint_terms(
+        self, study: Study, binding_constraint_id: str, constraint_terms: t.Sequence[ConstraintTerm]
+    ) -> None:
+        """
+        Adds new constraint terms to an existing binding constraint.
 
-        coeffs = {term.id: [term.weight] + [term.offset] if term.offset else [term.weight] for term in constraint_terms}
-        command = UpdateBindingConstraint(
-            id=constraint.id,
-            coeffs=coeffs,
-            command_context=self.storage_service.variant_study_service.command_factory.command_context,
-        )
-        execute_or_add_commands(study, file_study, [command], self.storage_service)
+        Args:
+            study: The study from which to update the binding constraint.
+            binding_constraint_id: The ID of the binding constraint to update.
+            constraint_terms: The constraint terms to add.
+        """
+        return self.update_constraint_terms(study, binding_constraint_id, constraint_terms, update_mode="add")
 
-    # FIXME create a dedicated delete service
     def remove_constraint_term(
         self,
         study: Study,
         binding_constraint_id: str,
         term_id: str,
     ) -> None:
-        return self.update_constraint_term(study, binding_constraint_id, term_id)  # type: ignore
+        """
+        Remove a constraint term from an existing binding constraint.
+
+        Args:
+            study: The study from which to update the binding constraint.
+            binding_constraint_id: The ID of the binding constraint to update.
+            term_id: The ID of the term to remove.
+        """
+        constraint = self.get_binding_constraint(study, binding_constraint_id)
+        existing_terms = collections.OrderedDict((term.generate_id(), term) for term in constraint.terms)
+        removed_term = existing_terms.pop(term_id, None)
+        if removed_term is None:
+            raise ConstraintTermNotFound(binding_constraint_id, term_id)
+        self._update_constraint_with_terms(study, constraint, existing_terms)
 
     @staticmethod
     def get_table_schema() -> JSON:
@@ -918,14 +919,6 @@ def _replace_matrices_according_to_frequency_and_version(
     return args
 
 
-def find_constraint_term_id(constraints_term: t.Sequence[ConstraintTerm], constraint_term_id: str) -> int:
-    try:
-        index = [elm.id for elm in constraints_term].index(constraint_term_id)
-        return index
-    except ValueError:
-        return -1
-
-
 def check_attributes_coherence(data: t.Union[ConstraintCreation, ConstraintInput], study_version: int) -> None:
     if study_version < 870:
         if data.group:

antarest/study/business/table_mode_management.py

@@ -174,10 +174,10 @@ def update_table_data(
             The updated properties of the objects including the old ones.
         """
         if table_type == TableModeType.AREA:
-            # Use AreaOutput to update properties of areas
+            # Use AreaOutput to update properties of areas, which may include `None` values
             area_props_by_ids = {key: AreaOutput(**values) for key, values in data.items()}
             areas_map = self._area_manager.update_areas_props(study, area_props_by_ids)
-            data = {area_id: area.dict(by_alias=True) for area_id, area in areas_map.items()}
+            data = {area_id: area.dict(by_alias=True, exclude_none=True) for area_id, area in areas_map.items()}
             return data
         elif table_type == TableModeType.LINK:
             links_map = {tuple(key.split(" / ")): LinkOutput(**values) for key, values in data.items()}

antarest/study/storage/variantstudy/model/command/icommand.py

@@ -19,7 +19,7 @@
 logger = logging.getLogger(__name__)
 
 
-class ICommand(ABC, BaseModel, extra=Extra.forbid, arbitrary_types_allowed=True):
+class ICommand(ABC, BaseModel, extra=Extra.forbid, arbitrary_types_allowed=True, copy_on_model_validation="deep"):
     """
     Interface for all commands that can be applied to a study.