add abstract class for links

antarest/study/business/link_management.py

@@ -128,7 +128,9 @@ def update_link(self, study: RawStudy, link_creation_info: LinkInfoDTOType) -> L
         command = UpdateLink(
             area1=link_creation_info.area1,
             area2=link_creation_info.area2,
-            parameters=existing_link.model_dump(exclude={"area1", "area2"}, exclude_none=True, by_alias=True),
+            parameters=existing_link.model_dump(
+                mode="json", exclude={"area1", "area2"}, exclude_none=True, by_alias=True
+            ),
             command_context=self.storage_service.variant_study_service.command_factory.command_context,
         )
 

antarest/study/storage/variantstudy/business/command_reverter.py

@@ -94,6 +94,10 @@ def _revert_create_link(base_command: CreateLink, history: t.List["ICommand"], b
             )
         ]
 
+    @staticmethod
+    def _revert_update_link(base_command: CreateLink, history: t.List["ICommand"], base: FileStudy) -> t.List[ICommand]:
+        raise NotImplementedError("The revert function for UpdateLink is not available")
+
     @staticmethod
     def _revert_remove_link(base_command: RemoveLink, history: t.List["ICommand"], base: FileStudy) -> t.List[ICommand]:
         raise NotImplementedError("The revert function for RemoveLink is not available")

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

@@ -83,25 +83,6 @@ class LinkProperties(LinkInfoProperties820):
 
 
 class AbstractLinkCommand(ICommand, metaclass=ABCMeta):
-    # area1: str
-    # area2: str
-    # parameters: Optional[Dict[str, Any]] = None
-    # series: Optional[Union[List[List[MatrixData]], str]] = None
-    # direct: Optional[Union[List[List[MatrixData]], str]] = None
-    # indirect: Optional[Union[List[List[MatrixData]], str]] = None
-    pass
-
-class CreateLink(ICommand):
-    """
-    Command used to create a link between two areas.
-    """
-
-    # Overloaded metadata
-    # ===================
-
-    command_name: CommandName = CommandName.CREATE_LINK
-    version: int = 1
-
     # Command parameters
     # ==================
 
@@ -120,11 +101,94 @@ def validate_series(
         return validate_matrix(v, new_values) if v is not None else v
 
     @model_validator(mode="after")
-    def validate_areas(self) -> "CreateLink":
+    def validate_areas(self) -> "AbstractLinkCommand":
         if self.area1 == self.area2:
             raise ValueError("Cannot create link on same node")
         return self
 
+    def to_dto(self) -> CommandDTO:
+        args = {
+            "area1": self.area1,
+            "area2": self.area2,
+            "parameters": self.parameters,
+        }
+        if self.series:
+            args["series"] = strip_matrix_protocol(self.series)
+        if self.direct:
+            args["direct"] = strip_matrix_protocol(self.direct)
+        if self.indirect:
+            args["indirect"] = strip_matrix_protocol(self.indirect)
+        return CommandDTO(
+            action=CommandName.CREATE_LINK.value,
+            args=args,
+        )
+
+    def match(self, other: ICommand, equal: bool = False) -> bool:
+        if not isinstance(other, CreateLink):
+            return False
+        simple_match = self.area1 == other.area1 and self.area2 == other.area2
+        if not equal:
+            return simple_match
+        return (
+            simple_match
+            and self.parameters == other.parameters
+            and self.series == other.series
+            and self.direct == other.direct
+            and self.indirect == other.indirect
+        )
+
+    def _create_diff(self, other: "ICommand") -> List["ICommand"]:
+        other = cast(CreateLink, other)
+        from antarest.study.storage.variantstudy.model.command.replace_matrix import ReplaceMatrix
+        from antarest.study.storage.variantstudy.model.command.update_config import UpdateConfig
+
+        commands: List[ICommand] = []
+        area_from, area_to = sorted([self.area1, self.area2])
+        if self.parameters != other.parameters:
+            properties = LinkProperties.model_validate(other.parameters or {})
+            link_property = properties.model_dump(mode="json", by_alias=True, exclude_none=True)
+            commands.append(
+                UpdateConfig(
+                    target=f"input/links/{area_from}/properties/{area_to}",
+                    data=link_property,
+                    command_context=self.command_context,
+                )
+            )
+        if self.series != other.series:
+            commands.append(
+                ReplaceMatrix(
+                    target=f"@links_series/{area_from}/{area_to}",
+                    matrix=strip_matrix_protocol(other.series),
+                    command_context=self.command_context,
+                )
+            )
+        return commands
+
+    def get_inner_matrices(self) -> List[str]:
+        list_matrices = []
+        if self.series:
+            assert_this(isinstance(self.series, str))
+            list_matrices.append(strip_matrix_protocol(self.series))
+        if self.direct:
+            assert_this(isinstance(self.direct, str))
+            list_matrices.append(strip_matrix_protocol(self.direct))
+        if self.indirect:
+            assert_this(isinstance(self.indirect, str))
+            list_matrices.append(strip_matrix_protocol(self.indirect))
+        return list_matrices
+
+
+class CreateLink(AbstractLinkCommand):
+    """
+    Command used to create a link between two areas.
+    """
+
+    # Overloaded metadata
+    # ===================
+
+    command_name: CommandName = CommandName.CREATE_LINK
+    version: int = 1
+
     def _create_link_in_config(self, area_from: str, area_to: str, study_data: FileStudyTreeConfig) -> None:
         self.parameters = self.parameters or {}
         study_data.areas[area_from].links[area_to] = Link(
@@ -276,55 +340,10 @@ def match_signature(self) -> str:
         )
 
     def match(self, other: ICommand, equal: bool = False) -> bool:
-        if not isinstance(other, CreateLink):
-            return False
-        simple_match = self.area1 == other.area1 and self.area2 == other.area2
-        if not equal:
-            return simple_match
-        return (
-            simple_match
-            and self.parameters == other.parameters
-            and self.series == other.series
-            and self.direct == other.direct
-            and self.indirect == other.indirect
-        )
+        return super().match(other, equal)
 
     def _create_diff(self, other: "ICommand") -> List["ICommand"]:
-        other = cast(CreateLink, other)
-        from antarest.study.storage.variantstudy.model.command.replace_matrix import ReplaceMatrix
-        from antarest.study.storage.variantstudy.model.command.update_config import UpdateConfig
-
-        commands: List[ICommand] = []
-        area_from, area_to = sorted([self.area1, self.area2])
-        if self.parameters != other.parameters:
-            properties = LinkProperties.model_validate(other.parameters or {})
-            link_property = properties.model_dump(mode="json", by_alias=True, exclude_none=True)
-            commands.append(
-                UpdateConfig(
-                    target=f"input/links/{area_from}/properties/{area_to}",
-                    data=link_property,
-                    command_context=self.command_context,
-                )
-            )
-        if self.series != other.series:
-            commands.append(
-                ReplaceMatrix(
-                    target=f"@links_series/{area_from}/{area_to}",
-                    matrix=strip_matrix_protocol(other.series),
-                    command_context=self.command_context,
-                )
-            )
-        return commands
+        return super()._create_diff(other)
 
     def get_inner_matrices(self) -> List[str]:
-        list_matrices = []
-        if self.series:
-            assert_this(isinstance(self.series, str))
-            list_matrices.append(strip_matrix_protocol(self.series))
-        if self.direct:
-            assert_this(isinstance(self.direct, str))
-            list_matrices.append(strip_matrix_protocol(self.direct))
-        if self.indirect:
-            assert_this(isinstance(self.indirect, str))
-            list_matrices.append(strip_matrix_protocol(self.indirect))
-        return list_matrices
+        return super().get_inner_matrices()

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

@@ -11,14 +11,17 @@
 # This file is part of the Antares project.
 import typing as t
 
+from antarest.study.model import STUDY_VERSION_8_2
 from antarest.study.storage.rawstudy.model.filesystem.config.model import FileStudyTreeConfig
 from antarest.study.storage.rawstudy.model.filesystem.factory import FileStudy
+from antarest.study.storage.variantstudy.business.utils import strip_matrix_protocol
 from antarest.study.storage.variantstudy.model.command.common import CommandName, CommandOutput
+from antarest.study.storage.variantstudy.model.command.create_link import AbstractLinkCommand
 from antarest.study.storage.variantstudy.model.command.icommand import MATCH_SIGNATURE_SEPARATOR, ICommand, OutputTuple
 from antarest.study.storage.variantstudy.model.model import CommandDTO
 
 
-class UpdateLink(ICommand):
+class UpdateLink(AbstractLinkCommand):
     """
     Command used to create a link between two areas.
     """
@@ -29,12 +32,6 @@ class UpdateLink(ICommand):
     command_name: CommandName = CommandName.UPDATE_LINK
     version: int = 1
 
-    # Command parameters
-    # ==================
-    area1: str
-    area2: str
-    parameters: t.Optional[t.Dict[str, t.Any]] = None
-
     def _apply_config(self, study_data: FileStudyTreeConfig) -> OutputTuple:
         area_from, area_to = sorted([self.area1, self.area2])
 
@@ -47,9 +44,51 @@ def _apply_config(self, study_data: FileStudyTreeConfig) -> OutputTuple:
         )
 
     def _apply(self, study_data: FileStudy) -> CommandOutput:
+        version = study_data.config.version
         area_from, area_to = sorted([self.area1, self.area2])
         study_data.tree.save(self.parameters, ["input", "links", area_from, "properties", area_to])
         output, _ = self._apply_config(study_data.config)
+
+        self.series = self.series or (self.command_context.generator_matrix_constants.get_link(version=version))
+        self.direct = self.direct or (self.command_context.generator_matrix_constants.get_link_direct())
+        self.indirect = self.indirect or (self.command_context.generator_matrix_constants.get_link_indirect())
+
+        assert type(self.series) is str
+        if version < STUDY_VERSION_8_2:
+            study_data.tree.save(self.series, ["input", "links", area_from, area_to])
+        else:
+            study_data.tree.save(
+                self.series,
+                ["input", "links", area_from, f"{area_to}_parameters"],
+            )
+
+            study_data.tree.save({}, ["input", "links", area_from, "capacities"])
+            if self.direct:
+                assert isinstance(self.direct, str)
+                study_data.tree.save(
+                    self.direct,
+                    [
+                        "input",
+                        "links",
+                        area_from,
+                        "capacities",
+                        f"{area_to}_direct",
+                    ],
+                )
+
+            if self.indirect:
+                assert isinstance(self.indirect, str)
+                study_data.tree.save(
+                    self.indirect,
+                    [
+                        "input",
+                        "links",
+                        area_from,
+                        "capacities",
+                        f"{area_to}_indirect",
+                    ],
+                )
+
         return output
 
     def to_dto(self) -> CommandDTO:
@@ -58,6 +97,12 @@ def to_dto(self) -> CommandDTO:
             "area2": self.area2,
             "parameters": self.parameters,
         }
+        if self.series:
+            args["series"] = strip_matrix_protocol(self.series)
+        if self.direct:
+            args["direct"] = strip_matrix_protocol(self.direct)
+        if self.indirect:
+            args["indirect"] = strip_matrix_protocol(self.indirect)
         return CommandDTO(
             action=CommandName.UPDATE_LINK.value,
             args=args,
@@ -69,7 +114,7 @@ def match_signature(self) -> str:
         )
 
     def _create_diff(self, other: "ICommand") -> t.List["ICommand"]:
-        pass
+        return super()._create_diff(other)
 
     def get_inner_matrices(self) -> t.List[str]:
-        pass
+        return super().get_inner_matrices()

tests/integration/study_data_blueprint/test_link.py

@@ -19,7 +19,6 @@
 
 @pytest.mark.unit_test
 class TestLink:
-
     @pytest.mark.parametrize("study_type", ["raw", "variant"])
     def test_link_update(self, client: TestClient, user_access_token: str, study_type: str) -> None:
         client.headers = {"Authorization": f"Bearer {user_access_token}"}  # type: ignore

tests/variantstudy/test_command_factory.py

@@ -94,6 +94,17 @@
             }
         ],
     ),
+    CommandDTO(
+        action=CommandName.UPDATE_LINK.value,
+        args=[
+            {
+                "area1": "area1",
+                "area2": "area2",
+                "parameters": {},
+                "series": "series",
+            }
+        ],
+    ),
     CommandDTO(
         action=CommandName.REMOVE_LINK.value,
         args={
@@ -411,7 +422,7 @@ def _get_command_classes(self) -> Set[str]:
                 f".{name}",
                 package="antarest.study.storage.variantstudy.model.command",
             )
-        abstract_commands = {"AbstractBindingConstraintCommand"}
+        abstract_commands = {"AbstractBindingConstraintCommand", "AbstractLinkCommand"}
         return {cmd.__name__ for cmd in ICommand.__subclasses__() if cmd.__name__ not in abstract_commands}
 
     def test_all_commands_are_tested(self, command_factory: CommandFactory):