resolve_components.py completed

src/andromede/model/parsing.py

@@ -21,11 +21,6 @@ def parse_yaml_library(input: typing.TextIO) -> "InputLibrary":
     return InputLibrary.model_validate(tree["library"])
 
 
-def parse_yaml_components(input_components: typing.TextIO) -> "InputComponent":
-    tree = safe_load(input_components)
-    return InputComponent.model_validate(tree["study"])
-
-
 # Design note: actual parsing and validation is delegated to pydantic models
 def _to_kebab(snake: str) -> str:
     return snake.replace("_", "-")
@@ -96,23 +91,6 @@ class Config:
         alias_generator = _to_kebab
 
 
-class InputPortConnections(BaseModel):
-    id: str
-    component1: str
-    port_1: str
-    component2: str
-    port_2: str
-
-
-class InputComponent(BaseModel):
-    nodes: List[InputModel] = Field(default_factory=list)
-    components: List[InputModel] = Field(default_factory=list)
-    connections: List[InputPortConnections] = Field(default_factory=list)
-
-    class Config:
-        alias_generator = _to_kebab
-
-
 class InputLibrary(BaseModel):
     id: str
     port_types: List[InputPortType] = Field(default_factory=list)

src/andromede/model/resolve_library.py

@@ -9,8 +9,8 @@
 # SPDX-License-Identifier: MPL-2.0
 #
 # This file is part of the Antares project.
-from pydoc import text
-from typing import Dict, List, Optional
+
+from typing import Dict, Optional
 
 from andromede.expression import ExpressionNode
 from andromede.expression.indexing_structure import IndexingStructure
@@ -30,23 +30,19 @@
     Variable,
     model,
 )
-from andromede.model.components import Components, components
 from andromede.model.library import Library, library
 from andromede.model.model import PortFieldDefinition, port_field_def
 from andromede.model.parsing import (
-    InputComponent,
     InputConstraint,
     InputField,
     InputLibrary,
     InputModel,
     InputModelPort,
     InputParameter,
-    InputPortConnections,
     InputPortFieldDefinition,
     InputPortType,
     InputVariable,
 )
-from andromede.study import Component, PortRef, PortsConnection
 
 
 def resolve_library(input_lib: InputLibrary) -> Library:
@@ -59,18 +55,8 @@ def resolve_library(input_lib: InputLibrary) -> Library:
     port_types = [_convert_port_type(p) for p in input_lib.port_types]
     port_types_dict = dict((p.id, p) for p in port_types)
     models = [_resolve_model(m, port_types_dict) for m in input_lib.models]
-    return library(port_types, models)
-
-
-def resolve_components_and_cnx(input_comp: InputComponent) -> Components:
-    components_list = [_resolve_component(m, m.id) for m in input_comp.components]
 
-    connections = []
-    for c in input_comp.connections:
-        resolved_cnx = _resolve_connections(c, components_list)
-        connections.extend(resolved_cnx)
-
-    return components(components_list, connections)
+    return library(port_types, models)
 
 
 def _convert_field(field: InputField) -> PortField:
@@ -165,30 +151,3 @@ def _to_constraint(
         lower_bound=_to_expression_if_present(constraint.lower_bound, identifiers),
         upper_bound=_to_expression_if_present(constraint.upper_bound, identifiers),
     )
-
-
-def _resolve_component(model_for_component: InputModel, component_id: str) -> Component:
-    return Component(
-        model=_resolve_model_identifier(model_for_component), id=component_id
-    )
-
-
-def _resolve_connections(
-    connection: InputPortConnections,
-    components_list: List[Component],
-) -> List[PortRef]:
-    component1 = connection.component1
-    component2 = connection.component2
-    port1 = connection.port_1
-    port2 = connection.port_2
-
-    port_ref_1 = PortRef(_get_component_by_id(components_list, component1), port1)
-    port_ref_2 = PortRef(_get_component_by_id(components_list, component2), port2)
-    return [port_ref_1, port_ref_2]
-
-
-def _get_component_by_id(
-    components_list: List[Component], component_id: str
-) -> Optional[Component]:
-    components_dict = {component.id: component for component in components_list}
-    return components_dict.get(component_id)

src/andromede/model/components.py → src/andromede/study/components.py

@@ -19,14 +19,14 @@
 @dataclass(frozen=True)
 class Components:
     components: Dict[str, Component]
-    ports_to_connect: List[PortRef]
+    connections: List[PortRef]
 
 
 def components(
     components_list: Iterable[Component],
-    ports_to_connect: Iterable[PortRef],
+    connections: Iterable[PortRef],
 ) -> Components:
     return Components(
         components=dict((m.id, m) for m in components_list),
-        ports_to_connect=list(p for p in ports_to_connect),
+        connections=list(p for p in connections),
     )

src/andromede/study/parsing.py

@@ -0,0 +1,45 @@
+# Copyright (c) 2024, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+import typing
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+from yaml import safe_load
+
+from andromede.model.parsing import InputModel
+
+
+def parse_yaml_components(input_components: typing.TextIO) -> "InputComponent":
+    tree = safe_load(input_components)
+    return InputComponent.model_validate(tree["study"])
+
+
+# Design note: actual parsing and validation is delegated to pydantic models
+def _to_kebab(snake: str) -> str:
+    return snake.replace("_", "-")
+
+
+class InputPortConnections(BaseModel):
+    id: str
+    component1: str
+    port_1: str
+    component2: str
+    port_2: str
+
+
+class InputComponent(BaseModel):
+    nodes: List[InputModel] = Field(default_factory=list)
+    components: List[InputModel] = Field(default_factory=list)
+    connections: List[InputPortConnections] = Field(default_factory=list)
+
+    class Config:
+        alias_generator = _to_kebab

src/andromede/study/resolve_components.py

@@ -0,0 +1,64 @@
+# Copyright (c) 2024, RTE (https://www.rte-france.com)
+#
+# See AUTHORS.txt
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This file is part of the Antares project.
+from typing import List, Optional
+
+from andromede.model.parsing import InputModel
+from andromede.model.resolve_library import _resolve_model_identifier
+from andromede.study import Component, PortRef
+from andromede.study.components import Components, components
+from andromede.study.parsing import InputComponent, InputPortConnections
+
+
+def resolve_components_and_cnx(input_comp: InputComponent) -> Components:
+    """
+    Resolves:
+    - components to be used for study
+    - connections between components"""
+    components_list = [_resolve_component(m, m.id) for m in input_comp.components]
+    components_list.extend(_resolve_component(n, n.id) for n in input_comp.nodes)
+    connections = []
+
+    for cnx in input_comp.connections:
+        resolved_cnx = _resolve_connections(cnx, components_list)
+        connections.extend(resolved_cnx)
+
+    return components(components_list, connections)
+
+
+def _resolve_component(model_for_component: InputModel, component_id: str) -> Component:
+    return Component(
+        model=_resolve_model_identifier(model_for_component), id=component_id
+    )
+
+
+def _resolve_connections(
+    connection: InputPortConnections,
+    components_list: List[Component],
+) -> List[PortRef]:
+    cnx_component1 = connection.component1
+    cnx_component2 = connection.component2
+    port1 = connection.port_1
+    port2 = connection.port_2
+
+    component_1 = _get_component_by_id(components_list, cnx_component1)
+    component_2 = _get_component_by_id(components_list, cnx_component2)
+    assert component_1 is not None and component_2 is not None
+    port_ref_1 = PortRef(component_1, port1)
+    port_ref_2 = PortRef(component_2, port2)
+    return [port_ref_1, port_ref_2]
+
+
+def _get_component_by_id(
+    components_list: List[Component], component_id: str
+) -> Optional[Component]:
+    components_dict = {component.id: component for component in components_list}
+    return components_dict.get(component_id)

tests/unittests/data/components.yml

@@ -38,5 +38,11 @@ study:
       component2: D
       port_2: injection_port
 
+    - id: cnx2
+      component1: N
+      port_1: injection_port
+      component2: G
+      port_2: injection_port
+
 
 

tests/unittests/study/test_components_parsing.py

@@ -0,0 +1,19 @@
+from pathlib import Path
+
+from andromede.study.parsing import parse_yaml_components
+from andromede.study.resolve_components import resolve_components_and_cnx
+
+
+def test_parsing_components_ok(data_dir: Path):
+    compo_file = data_dir / "components.yml"
+
+    with compo_file.open() as c:
+        input_compo = parse_yaml_components(c)
+    assert len(input_compo.components) == 2
+    assert len(input_compo.nodes) == 1
+    assert len(input_compo.connections) == 2
+
+    result = resolve_components_and_cnx(input_compo)
+
+    assert len(result.components) == 3
+    assert len(result.connections) == 4