diff --git a/pyproject.toml b/pyproject.toml index 538fef3e..01c684dd 100644 --- a/pyproject.toml +++ b/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] diff --git a/src/omotes_simulator_core/adapter/transforms/mappers.py b/src/omotes_simulator_core/adapter/transforms/mappers.py index e77a05d7..177b9b23 100644 --- a/src/omotes_simulator_core/adapter/transforms/mappers.py +++ b/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 diff --git a/src/omotes_simulator_core/adapter/transforms/transform_utils.py b/src/omotes_simulator_core/adapter/transforms/transform_utils.py index 5b4c9a55..5279a2cf 100644 --- a/src/omotes_simulator_core/adapter/transforms/transform_utils.py +++ b/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]]: diff --git a/src/omotes_simulator_core/entities/assets/esdl_asset_object.py b/src/omotes_simulator_core/entities/assets/esdl_asset_object.py index a6aeab99..4a3d9644 100644 --- a/src/omotes_simulator_core/entities/assets/esdl_asset_object.py +++ b/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.""" diff --git a/unit_test/entities/test_esdl_object.py b/unit_test/entities/test_esdl_object.py index c63959e2..2c6d513b 100644 --- a/unit_test/entities/test_esdl_object.py +++ b/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]) diff --git a/unit_test/transforms/test_transform_utils.py b/unit_test/transforms/test_transform_utils.py new file mode 100644 index 00000000..baeeb70b --- /dev/null +++ b/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 . + + +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"]})