Merge pull request #202 from Project-OMOTES/164-add-port-type-to-assets

164 add port type to assets

pyproject.toml

@@ -86,7 +86,7 @@ include-package-data = true
 #packages = ["src/omotes_simulator_core"]
 
 [tool.pytest.ini_options]
-# addopts = "--cov=omotes_simulator_core --cov-report html --cov-report term-missing --cov-fail-under 80"
+addopts = "--cov=omotes_simulator_core --cov-report html --cov-report term-missing --cov-fail-under 80"
 testpaths = ["unit_test"]
 
 [tool.coverage.run]

src/omotes_simulator_core/adapter/transforms/mappers.py

@@ -171,7 +171,7 @@ def _convert_assets(self, network: Network) -> List[AssetAbstract]:
         """
         py_assets_list = []
         for esdl_asset in self.esdl_object.get_all_assets_of_type("asset"):
-            # Esdl Junctions need to be skipped for now, are added later.
+            # Esdl Junctions need to be skipped in this method, they are added in another method.
             if isinstance(esdl_asset.esdl_asset, esdl_junction):
                 continue
             py_assets_list.append(EsdlAssetMapper.to_entity(esdl_asset))
@@ -190,7 +190,6 @@ def _get_junction(self) -> dict[str, list[tuple[str, str]]]:
         """
         py_joint_dict = {}
         for esdl_joint in self.esdl_object.get_all_assets_of_type("joint"):
-            # Esdl Junctions need to be skipped for now, are added later.
             temp_list = [
                 self.esdl_object.get_connected_assets(
                     asset_id=esdl_joint.esdl_asset.id, port_id=port

src/omotes_simulator_core/adapter/transforms/transform_utils.py

@@ -15,6 +15,65 @@
 
 """File containing utility functions for the transforms."""
 from typing import Dict, List
+from enum import Enum
+from dataclasses import dataclass
+
+
+class PortType(Enum):
+    """Enum to define the type of port."""
+
+    IN = 1
+    OUT = 2
+
+
+@dataclass
+class Port:
+    """Dataclass to hold the port information."""
+
+    port_id: str
+    port_name: str
+    port_type: PortType
+
+
+def sort_ports(connected_ports: list[Port]) -> list[str]:
+    """Sort the ports of the asset based on the port type.
+
+    The sort order is Inport, Outport, Starting with the first inport, then the first outport
+    and so on.
+
+    :param connected_ports: List of tuples with the port id, name and the port type.
+    :return: List of port ids sorted by port type.
+    """
+    in_ports = [port for port in connected_ports if port.port_type == PortType.IN]
+    out_ports = [port for port in connected_ports if port.port_type == PortType.OUT]
+    if len(in_ports) != len(out_ports):
+        raise ValueError("The number of in ports and out ports are not equal")
+    sorted_port_list = []
+    for in_port, out_port in zip(in_ports, out_ports):
+        sorted_port_list.append(in_port)
+        sorted_port_list.append(out_port)
+    if len(sorted_port_list) == 4:
+        sorted_port_list = order_prim_sec_ports(sorted_port_list)
+    return [port.port_id for port in sorted_port_list]
+
+
+def order_prim_sec_ports(connected_ports: list[Port]) -> list[Port]:
+    """Order the primary and secondary ports. in correct order.
+
+    The correct order is first primary port, then secondary port.
+    This can only be done by checking the port names.
+    In primary port prim is in the name for secondary port sec is in the name.
+
+    :param connected_ports: List of connected ports to be sorted.
+    :return: List of connected ports sorted by primary and secondary ports.
+    """
+    primary_ports = [port for port in connected_ports if "Prim" in port.port_name]
+    if len(primary_ports) != 2:
+        raise ValueError("The number of ports with prim in the name is not equal to 2")
+    secondary_ports = [port for port in connected_ports if "Sec" in port.port_name]
+    if len(secondary_ports) != 2:
+        raise ValueError("The number of ports with sec in the name is not equal to 2")
+    return primary_ports + secondary_ports
 
 
 def reverse_dict(original_dict: dict) -> Dict[str, List[type]]:

src/omotes_simulator_core/entities/assets/esdl_asset_object.py

@@ -20,6 +20,7 @@
 import pandas as pd
 from esdl import esdl
 from omotes_simulator_core.entities.utility.influxdb_reader import get_data_from_profile
+from omotes_simulator_core.adapter.transforms.transform_utils import PortType, sort_ports, Port
 
 logger = logging.getLogger(__name__)
 
@@ -78,8 +79,18 @@ def get_return_temperature(self, port_type: str) -> float:
         raise ValueError(f"No port found with type: {port_type} for asset: {self.esdl_asset.name}")
 
     def get_port_ids(self) -> list[str]:
-        """Get the port ids of the asset."""
-        return [port.id for port in self.esdl_asset.port]
+        """Returns a sorted list of the port ids of the asset."""
+        list_of_ports = sort_ports(
+            [
+                Port(
+                    port.id,
+                    port.name,
+                    PortType.IN if isinstance(port, esdl.InPort) else PortType.OUT,
+                )
+                for port in self.esdl_asset.port
+            ]
+        )
+        return list_of_ports
 
     def get_port_type(self, port_type: str) -> Type[esdl.Port]:
         """Get the port type of the port."""

unit_test/entities/test_esdl_object.py

@@ -265,7 +265,7 @@ def test_multiple_connection(self):
         esdl_object = EsdlObject(pyesdl_from_file(esdl_file_path))
         pipe = esdl_object.get_all_assets_of_type("pipe")[0]
         assets = esdl_object.get_all_assets_of_type("producer")
-        test_list = [(asset.get_id(), asset.get_port_ids()[0]) for asset in assets]
+        test_list = [(asset.get_id(), asset.get_port_ids()[1]) for asset in assets]
 
         # Act
         connected_assets = esdl_object.get_connected_assets(pipe.get_id(), pipe.get_port_ids()[0])

unit_test/transforms/test_transform_utils.py

@@ -0,0 +1,200 @@
+#  Copyright (c) 2023. Deltares & TNO
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+
+import unittest
+from omotes_simulator_core.adapter.transforms.transform_utils import (
+    sort_ports,
+    order_prim_sec_ports,
+    reverse_dict,
+    PortType,
+    Port,
+)
+
+
+class TransformUtilsTest(unittest.TestCase):
+    """Class to test the transform utils functions."""
+
+    def setUp(self) -> None:
+        """Set up test case."""
+        self.connected_ports = [
+            Port(port_id="port1", port_name="Prim port1", port_type=PortType.IN),
+            Port(port_id="port2", port_name="Prim port2", port_type=PortType.OUT),
+        ]
+
+    def test_sort_ports_basic(self):
+        """Test the port sorting function. THe order is nto changed."""
+        # arrange
+
+        # act
+        result = sort_ports(self.connected_ports)
+
+        # assert
+        self.assertEqual(result, [port.port_id for port in self.connected_ports])
+
+    def test_sort_ports_reverse(self):
+        """Test the port sorting function. The order is reversed."""
+        # arrange
+        connected_ports = self.connected_ports[::-1]
+
+        # act
+        result = sort_ports(connected_ports)
+
+        # assert
+        self.assertEqual(result, [port.port_id for port in self.connected_ports])
+
+    def test_sort_ports_4_ports(self):
+        """Test the port sorting function with 4 ports, the order remains the same."""
+        # arrange
+        self.connected_ports.append(
+            Port(port_id="port3", port_name="Sec port3", port_type=PortType.IN)
+        )
+        self.connected_ports.append(
+            Port(port_id="port4", port_name="Sec port4", port_type=PortType.OUT)
+        )
+        # act
+        result = sort_ports(self.connected_ports)
+
+        # assert
+        self.assertEqual(result, [port.port_id for port in self.connected_ports])
+
+    def test_sort_ports_4_ports_reverse(self):
+        """Test the port sorting function with 4 ports, the order is reversed."""
+        # arrange
+        self.connected_ports.append(
+            Port(port_id="port3", port_name="Sec port3", port_type=PortType.IN)
+        )
+        self.connected_ports.append(
+            Port(port_id="port4", port_name="Sec port4", port_type=PortType.OUT)
+        )
+        connected_ports = self.connected_ports[::-1]
+
+        # act
+        result = sort_ports(connected_ports)
+
+        # assert
+        self.assertEqual(result, [port.port_id for port in self.connected_ports])
+
+    def test_sort_port_error_number_of_ports(self):
+        """Test the port sorting function.
+
+        This test has different number of in and out port resulting in error.
+        """
+        # arrange
+        self.connected_ports.append(
+            Port(port_id="port3", port_name="Sec port3", port_type=PortType.IN)
+        )
+
+        # act
+        with self.assertRaises(ValueError) as cm:
+            sort_ports(self.connected_ports)
+
+        # assert
+        self.assertEqual(str(cm.exception), "The number of in ports and out ports are not equal")
+
+    def test_order_prim_sec_ports(self):
+        """Test the order primary secondary ports function."""
+        # arrange
+        self.connected_ports.append(
+            Port(port_id="port3", port_name="Sec port3", port_type=PortType.IN)
+        )
+        self.connected_ports.append(
+            Port(port_id="port4", port_name="Sec port4", port_type=PortType.OUT)
+        )
+
+        # act
+        result = order_prim_sec_ports(self.connected_ports)
+
+        # assert
+        self.assertEqual(result, self.connected_ports)
+
+    def test_order_prim_sec_ports_error_prim(self):
+        """Test the order primary secondary ports function.
+
+        This test has different number of primary ports resulting in error.
+        """
+        # arrange
+        self.connected_ports.append(
+            Port(port_id="port3", port_name="Prim port3", port_type=PortType.IN)
+        )
+        self.connected_ports.append(
+            Port(port_id="port4", port_name="Sec port4", port_type=PortType.OUT)
+        )
+
+        # act
+        with self.assertRaises(ValueError) as cm:
+            order_prim_sec_ports(self.connected_ports)
+
+        # assert
+        self.assertEqual(
+            str(cm.exception), "The number of ports with prim in the name is not equal to 2"
+        )
+
+    def test_order_prim_sec_ports_error_sec(self):
+        """Test the order primary secondary ports function.
+
+        This test has different number of secondary ports resulting in error.
+        """
+        # arrange
+        self.connected_ports.append(
+            Port(port_id="port3", port_name="Sec port3", port_type=PortType.IN)
+        )
+        self.connected_ports.append(
+            Port(port_id="port4", port_name="Sec port4", port_type=PortType.OUT)
+        )
+        self.connected_ports.append(
+            Port(port_id="port5", port_name="Sec port5", port_type=PortType.OUT)
+        )
+
+        # act
+        with self.assertRaises(ValueError) as cm:
+            order_prim_sec_ports(self.connected_ports)
+
+        # assert
+        self.assertEqual(
+            str(cm.exception), "The number of ports with sec in the name is not equal to 2"
+        )
+
+    def test_order_prim_sec_ports_reversed(self):
+        """Test the order primary secondary ports function.
+
+        This test has the order reversed.
+        """
+        # arrange
+        self.connected_ports.insert(
+            0, Port(port_id="port4", port_name="Sec port4", port_type=PortType.OUT)
+        )
+        self.connected_ports.insert(
+            0, Port(port_id="port3", port_name="Sec port3", port_type=PortType.IN)
+        )
+        # act
+        result = order_prim_sec_ports(self.connected_ports)
+
+        # assert
+        self.assertEqual(result[0], self.connected_ports[2])
+        self.assertEqual(result[1], self.connected_ports[3])
+        self.assertEqual(result[2], self.connected_ports[0])
+        self.assertEqual(result[3], self.connected_ports[1])
+
+    def test_reverse_dict(self):
+        """Test the reverse dict function."""
+        # arrange
+        original_dict = {"key1": "value1", "key2": "value2", "key3": "value1"}
+
+        # act
+        result = reverse_dict(original_dict)
+
+        # assert
+        self.assertEqual(result, {"value1": ["key1", "key3"], "value2": ["key2"]})