fix(links): fix a bug that occurred when updating links via table-mode (

#2256)

antarest/study/business/link_management.py

@@ -11,34 +11,21 @@
 # This file is part of the Antares project.
 
 import typing as t
-from typing import Any, Dict
+from typing import Any
 
 from antares.study.version import StudyVersion
 
-from antarest.core.exceptions import ConfigFileNotFound, LinkNotFound, LinkValidationError
+from antarest.core.exceptions import LinkNotFound
 from antarest.core.model import JSON
-from antarest.study.business.all_optional_meta import all_optional_model, camel_case_model
 from antarest.study.business.model.link_model import LinkBaseDTO, LinkDTO, LinkInternal
 from antarest.study.business.utils import execute_or_add_commands
 from antarest.study.model import RawStudy, Study
-from antarest.study.storage.rawstudy.model.filesystem.config.links import LinkProperties
 from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy
 from antarest.study.storage.storage_service import StudyStorageService
 from antarest.study.storage.variantstudy.model.command.create_link import CreateLink
 from antarest.study.storage.variantstudy.model.command.remove_link import RemoveLink
-from antarest.study.storage.variantstudy.model.command.update_config import UpdateConfig
 from antarest.study.storage.variantstudy.model.command.update_link import UpdateLink
 
-_ALL_LINKS_PATH = "input/links"
-
-
-@all_optional_model
-@camel_case_model
-class LinkOutput(LinkProperties):
-    """
-    DTO object use to get the link information.
-    """
-
 
 class LinkManager:
     def __init__(self, storage_service: StudyStorageService) -> None:
@@ -61,6 +48,17 @@ def get_all_links(self, study: Study) -> t.List[LinkDTO]:
 
         return result
 
+    def get_link(self, study: RawStudy, link: LinkInternal) -> LinkInternal:
+        file_study = self.storage_service.get_storage(study).get_raw(study)
+
+        link_properties = self._get_link_if_exists(file_study, link)
+
+        link_properties.update({"area1": link.area1, "area2": link.area2})
+
+        updated_link = LinkInternal.model_validate(link_properties)
+
+        return updated_link
+
     def create_link(self, study: Study, link_creation_dto: LinkDTO) -> LinkDTO:
         link = link_creation_dto.to_internal(StudyVersion.parse(study.version))
 
@@ -80,16 +78,16 @@ def create_link(self, study: Study, link_creation_dto: LinkDTO) -> LinkDTO:
         return link_creation_dto
 
     def update_link(self, study: RawStudy, area_from: str, area_to: str, link_update_dto: LinkBaseDTO) -> LinkDTO:
-        link_dto = LinkDTO(area1=area_from, area2=area_to, **link_update_dto.model_dump())
+        link_dto = LinkDTO(area1=area_from, area2=area_to, **link_update_dto.model_dump(exclude_unset=True))
 
         link = link_dto.to_internal(StudyVersion.parse(study.version))
         file_study = self.storage_service.get_storage(study).get_raw(study)
 
-        self.get_link_if_exists(file_study, link)
+        self._get_link_if_exists(file_study, link)
 
         command = UpdateLink(
-            area1=area_from,
-            area2=area_to,
+            area1=link.area1,
+            area2=link.area2,
             parameters=link.model_dump(
                 include=link_update_dto.model_fields_set, exclude={"area1", "area2"}, exclude_none=True
             ),
@@ -99,25 +97,21 @@ def update_link(self, study: RawStudy, area_from: str, area_to: str, link_update
 
         execute_or_add_commands(study, file_study, [command], self.storage_service)
 
-        updated_link = self.get_internal_link(study, link)
+        updated_link = self.get_link(study, link)
 
         return updated_link.to_dto()
 
-    def get_link_if_exists(self, file_study: FileStudy, link: LinkInternal) -> dict[str, Any]:
-        try:
-            return file_study.tree.get(["input", "links", link.area1, "properties", link.area2])
-        except KeyError:
-            raise LinkNotFound(f"The link {link.area1} -> {link.area2} is not present in the study")
-
-    def get_internal_link(self, study: RawStudy, link: LinkInternal) -> LinkInternal:
-        file_study = self.storage_service.get_storage(study).get_raw(study)
-
-        link_properties = self.get_link_if_exists(file_study, link)
-
-        link_properties.update({"area1": link.area1, "area2": link.area2})
-        updated_link = LinkInternal.model_validate(link_properties)
+    def update_links(
+        self,
+        study: RawStudy,
+        update_links_by_ids: t.Mapping[t.Tuple[str, str], LinkBaseDTO],
+    ) -> t.Mapping[t.Tuple[str, str], LinkBaseDTO]:
+        new_links_by_ids = {}
+        for (area1, area2), update_link_dto in update_links_by_ids.items():
+            updated_link = self.update_link(study, area1, area2, update_link_dto)
+            new_links_by_ids[(area1, area2)] = updated_link
 
-        return updated_link
+        return new_links_by_ids
 
     def delete_link(self, study: RawStudy, area1_id: str, area2_id: str) -> None:
         file_study = self.storage_service.get_storage(study).get_raw(study)
@@ -129,69 +123,12 @@ def delete_link(self, study: RawStudy, area1_id: str, area2_id: str) -> None:
         )
         execute_or_add_commands(study, file_study, [command], self.storage_service)
 
-    def get_all_links_props(self, study: RawStudy) -> t.Mapping[t.Tuple[str, str], LinkOutput]:
-        """
-        Retrieves all links properties from the study.
-
-        Args:
-            study: The raw study object.
-        Returns:
-            A mapping of link IDS `(area1_id, area2_id)` to link properties.
-        Raises:
-            ConfigFileNotFound: if a configuration file is not found.
-        """
-        file_study = self.storage_service.get_storage(study).get_raw(study)
-
-        # Get the link information from the `input/links/{area1}/properties.ini` file.
-        path = _ALL_LINKS_PATH
+    def _get_link_if_exists(self, file_study: FileStudy, link: LinkInternal) -> dict[str, Any]:
         try:
-            links_cfg = file_study.tree.get(path.split("/"), depth=5)
+            return file_study.tree.get(["input", "links", link.area1, "properties", link.area2])
         except KeyError:
-            raise ConfigFileNotFound(path) from None
-
-        # areas_cfg contains a dictionary where the keys are the area IDs,
-        # and the values are objects that can be converted to `LinkFolder`.
-        links_by_ids = {}
-        for area1_id, entries in links_cfg.items():
-            property_map = entries.get("properties") or {}
-            for area2_id, properties_cfg in property_map.items():
-                area1_id, area2_id = sorted([area1_id, area2_id])
-                properties = LinkProperties(**properties_cfg)
-                links_by_ids[(area1_id, area2_id)] = LinkOutput(**properties.model_dump(mode="json", by_alias=False))
-
-        return links_by_ids
-
-    def update_links_props(
-        self,
-        study: RawStudy,
-        update_links_by_ids: t.Mapping[t.Tuple[str, str], LinkOutput],
-    ) -> t.Mapping[t.Tuple[str, str], LinkOutput]:
-        old_links_by_ids = self.get_all_links_props(study)
-        new_links_by_ids = {}
-        file_study = self.storage_service.get_storage(study).get_raw(study)
-        commands = []
-        for (area1, area2), update_link_dto in update_links_by_ids.items():
-            # Update the link properties.
-            old_link_dto = old_links_by_ids[(area1, area2)]
-            new_link_dto = old_link_dto.copy(
-                update=update_link_dto.model_dump(mode="json", by_alias=False, exclude_none=True)
-            )
-            new_links_by_ids[(area1, area2)] = new_link_dto
-
-            # Convert the DTO to a configuration object and update the configuration file.
-            properties = LinkProperties(**new_link_dto.model_dump(by_alias=False))
-            path = f"{_ALL_LINKS_PATH}/{area1}/properties/{area2}"
-            cmd = UpdateConfig(
-                target=path,
-                data=properties.to_config(),
-                command_context=self.storage_service.variant_study_service.command_factory.command_context,
-                study_version=file_study.config.version,
-            )
-            commands.append(cmd)
-
-        execute_or_add_commands(study, file_study, commands, self.storage_service)
-        return new_links_by_ids
+            raise LinkNotFound(f"The link {link.area1} -> {link.area2} is not present in the study")
 
     @staticmethod
     def get_table_schema() -> JSON:
-        return LinkOutput.schema()
+        return LinkBaseDTO.model_json_schema()

antarest/study/business/model/link_model.py

@@ -12,19 +12,122 @@
 import typing as t
 
 from antares.study.version import StudyVersion
-from pydantic import ConfigDict, Field, model_validator
+from pydantic import BeforeValidator, ConfigDict, Field, PlainSerializer, model_validator
 
 from antarest.core.exceptions import LinkValidationError
 from antarest.core.serialization import AntaresBaseModel
 from antarest.core.utils.string import to_camel_case, to_kebab_case
+from antarest.study.business.enum_ignore_case import EnumIgnoreCase
 from antarest.study.model import STUDY_VERSION_8_2
-from antarest.study.storage.rawstudy.model.filesystem.config.links import (
-    AssetType,
-    FilterOption,
-    LinkStyle,
-    TransmissionCapacity,
-    comma_separated_enum_list,
-)
+
+
+class AssetType(EnumIgnoreCase):
+    """
+    Enum representing the type of asset for a link between two areas.
+
+    Attributes:
+        AC: Represents an Alternating Current link. This is the most common type of electricity transmission.
+        DC: Represents a Direct Current link. This is typically used for long-distance transmission.
+        GAZ: Represents a gas link. This is used when the link is related to gas transmission.
+        VIRT: Represents a virtual link. This is used when the link doesn't physically exist
+            but is used for modeling purposes.
+        OTHER: Represents any other type of link that doesn't fall into the above categories.
+    """
+
+    AC = "ac"
+    DC = "dc"
+    GAZ = "gaz"
+    VIRT = "virt"
+    OTHER = "other"
+
+
+class TransmissionCapacity(EnumIgnoreCase):
+    """
+    Enum representing the transmission capacity of a link.
+
+    Attributes:
+        INFINITE: Represents a link with infinite transmission capacity.
+            This means there are no limits on the amount of electricity that can be transmitted.
+        IGNORE: Represents a link where the transmission capacity is ignored.
+            This means the capacity is not considered during simulations.
+        ENABLED: Represents a link with a specific transmission capacity.
+            This means the capacity is considered in the model and has a certain limit.
+    """
+
+    INFINITE = "infinite"
+    IGNORE = "ignore"
+    ENABLED = "enabled"
+
+
+class LinkStyle(EnumIgnoreCase):
+    """
+    Enum representing the style of a link in a network visualization.
+
+    Attributes:
+        DOT: Represents a dotted line style.
+        PLAIN: Represents a solid line style.
+        DASH: Represents a dashed line style.
+        DOT_DASH: Represents a line style with alternating dots and dashes.
+    """
+
+    DOT = "dot"
+    PLAIN = "plain"
+    DASH = "dash"
+    DOT_DASH = "dotdash"
+    OTHER = "other"
+
+
+class FilterOption(EnumIgnoreCase):
+    """
+    Enum representing the time filter options for data visualization or analysis in Antares Web.
+
+    Attributes:
+        HOURLY: Represents filtering data by the hour.
+        DAILY: Represents filtering data by the day.
+        WEEKLY: Represents filtering data by the week.
+        MONTHLY: Represents filtering data by the month.
+        ANNUAL: Represents filtering data by the year.
+    """
+
+    HOURLY = "hourly"
+    DAILY = "daily"
+    WEEKLY = "weekly"
+    MONTHLY = "monthly"
+    ANNUAL = "annual"
+
+
+def validate_filters(
+    filter_value: t.Union[t.List[FilterOption], str], enum_cls: t.Type[FilterOption]
+) -> t.List[FilterOption]:
+    if isinstance(filter_value, str):
+        if not filter_value.strip():
+            return []
+
+        filter_accepted_values = [e for e in enum_cls]
+
+        options = filter_value.replace(" ", "").split(",")
+
+        invalid_options = [opt for opt in options if opt not in filter_accepted_values]
+        if invalid_options:
+            raise LinkValidationError(
+                f"Invalid value(s) in filters: {', '.join(invalid_options)}. "
+                f"Allowed values are: {', '.join(filter_accepted_values)}."
+            )
+        options_enum: t.List[FilterOption] = list(dict.fromkeys(enum_cls(opt) for opt in options))
+        return options_enum
+
+    return filter_value
+
+
+def join_with_comma(values: t.List[FilterOption]) -> str:
+    return ", ".join(value.name.lower() for value in values)
+
+
+comma_separated_enum_list = t.Annotated[
+    t.List[FilterOption],
+    BeforeValidator(lambda x: validate_filters(x, FilterOption)),
+    PlainSerializer(lambda x: join_with_comma(x)),
+]
 
 DEFAULT_COLOR = 112
 FILTER_VALUES: t.List[FilterOption] = [

antarest/study/business/table_mode_management.py

@@ -25,7 +25,8 @@
 from antarest.study.business.areas.thermal_management import ThermalClusterInput, ThermalManager
 from antarest.study.business.binding_constraint_management import BindingConstraintManager, ConstraintInput
 from antarest.study.business.enum_ignore_case import EnumIgnoreCase
-from antarest.study.business.link_management import LinkManager, LinkOutput
+from antarest.study.business.link_management import LinkManager
+from antarest.study.business.model.link_model import LinkBaseDTO
 from antarest.study.model import STUDY_VERSION_8_2, RawStudy
 
 _TableIndex = str  # row name
@@ -98,15 +99,15 @@ def _get_table_data_unsafe(self, study: RawStudy, table_type: TableModeType) ->
             areas_map = self._area_manager.get_all_area_props(study)
             data = {area_id: area.model_dump(mode="json", by_alias=True) for area_id, area in areas_map.items()}
         elif table_type == TableModeType.LINK:
-            links_map = self._link_manager.get_all_links_props(study)
+            links_map = self._link_manager.get_all_links(study)
             excludes = (
                 set()
                 if StudyVersion.parse(study.version) >= STUDY_VERSION_8_2
                 else {"filter_synthesis", "filter_year_by_year"}
             )
             data = {
-                f"{area1_id} / {area2_id}": link.model_dump(mode="json", by_alias=True, exclude=excludes)
-                for (area1_id, area2_id), link in links_map.items()
+                f"{link.area1} / {link.area2}": link.model_dump(mode="json", by_alias=True, exclude=excludes)
+                for link in links_map
             }
         elif table_type == TableModeType.THERMAL:
             thermals_by_areas = self._thermal_manager.get_all_thermals_props(study)
@@ -199,8 +200,8 @@ def update_table_data(
             data = {area_id: area.model_dump(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()}
-            updated_map = self._link_manager.update_links_props(study, links_map)  # type: ignore
+            links_map = {tuple(key.split(" / ")): LinkBaseDTO(**values) for key, values in data.items()}
+            updated_map = self._link_manager.update_links(study, links_map)  # type: ignore
             excludes = (
                 set()
                 if StudyVersion.parse(study.version) >= STUDY_VERSION_8_2