Get rid of repetitive code, make one util function for all elements being converted as sym_load

Signed-off-by: furqan463 <[email protected]>

Author: furqan463

src/power_grid_model_io/converters/pandapower_converter.py

@@ -342,13 +342,13 @@ def _create_output_data(self):
         self._pp_buses_output()
         self._pp_lines_output()
         self._pp_ext_grids_output()
-        self._pp_loads_output()
+        self._pp_load_elements_output(element="load", symmetric=True)
+        self._pp_load_elements_output(element="ward", symmetric=True)
+        self._pp_load_elements_output(element="motor", symmetric=True)
         self._pp_shunts_output()
         self._pp_trafos_output()
         self._pp_sgens_output()
         self._pp_trafos3w_output()
-        self._pp_ward_output()
-        self._pp_motor_output()
         self._pp_asym_gens_output()
         self._pp_asym_loads_output()
         # Switches derive results from branches pp_output_data and pgm_output_data of links. Hence, placed in the end.
@@ -361,17 +361,15 @@ def _create_output_data_3ph(self):
         angle in degrees
         """
         # TODO create output_data_3ph for trafos3w
-        # Although Pandapower itself did not implmenet res_shunt_3ph
-        # Since results are avaiable in PGM output, these should be converted.
         self._pp_buses_output_3ph()
         self._pp_lines_output_3ph()
         self._pp_ext_grids_output_3ph()
-        self._pp_loads_output_3ph()
+        self._pp_load_elements_output(element="load", symmetric=False)
+        self._pp_load_elements_output(element="ward", symmetric=False)
+        self._pp_load_elements_output(element="motor", symmetric=False)
         self._pp_shunts_output_3ph()
         self._pp_trafos_output_3ph()
         self._pp_sgens_output_3ph()
-        self._pp_ward_output_3ph()
-        self._pp_motor_output_3ph()
         self._pp_asym_gens_output_3ph()
         self._pp_asym_loads_output_3ph()
 
@@ -1563,64 +1561,37 @@ def _pp_asym_gens_output(self):
 
         self.pp_output_data["res_asymmetric_sgen"] = pp_output_asym_gens
 
-    def _pp_loads_output(self):
+    def _pp_load_elements_output(self, element, symmetric):
         """
-        This function converts a power-grid-model Symmetrical Load output array to a Load Dataframe of PandaPower.
-
-        Returns:
-            a PandaPower Dataframe for the Load component
+        Utility function to convert output of elements represented as load
+        in power grid model.
+        element: "load", "motor" or "ward"
+        symmetric: True or False
         """
-        load_id_names = ["const_power", "const_impedance", "const_current"]
-        assert "res_load" not in self.pp_output_data
-
-        if (
-            ComponentType.sym_load not in self.pgm_output_data
-            or self.pgm_output_data[ComponentType.sym_load].size == 0
-            or ("load", load_id_names[0]) not in self.idx
-        ):
-            return
-
-        # Store the results, while assuring that we are not overwriting any data
-        assert "res_load" not in self.pp_output_data
-        self.pp_output_data["res_load"] = self._pp_load_result_accumulate(
-            pp_component_name="load", load_id_names=load_id_names
-        )
-
-    def _pp_ward_output(self):
-        load_id_names = ["ward_const_power_load", "ward_const_impedance_load"]
-        assert "res_ward" not in self.pp_output_data
-
-        if (
-            ComponentType.sym_load not in self.pgm_output_data
-            or self.pgm_output_data[ComponentType.sym_load].size == 0
-            or ("ward", load_id_names[0]) not in self.idx
-        ):
-            return
-
-        accumulated_loads = self._pp_load_result_accumulate(pp_component_name="ward", load_id_names=load_id_names)
-        # TODO Find a better way for mapping vm_pu from bus
-        # accumulated_loads["vm_pu"] = np.nan
-
-        # Store the results, while assuring that we are not overwriting any data
-        assert "res_ward" not in self.pp_output_data
-        self.pp_output_data["res_ward"] = accumulated_loads
+        if symmetric:
+            res_table = "res_" + element
+        else:
+            res_table = "res_" + element + "_3ph"
 
-    def _pp_motor_output(self):
-        load_id_names = ["motor_load"]
+        if element == "load":
+            load_id_names = ["const_power", "const_impedance", "const_current"]
+        elif element == "ward":
+            load_id_names = ["ward_const_power_load", "ward_const_impedance_load"]
+        elif element == "motor":
+            load_id_names = ["motor_load"]
 
-        assert "res_motor" not in self.pp_output_data
+        assert res_table not in self.pp_output_data
 
         if (
             ComponentType.sym_load not in self.pgm_output_data
             or self.pgm_output_data[ComponentType.sym_load].size == 0
-            or ("motor", load_id_names[0]) not in self.idx
+            or (element, load_id_names[0]) not in self.idx
         ):
             return
-
         # Store the results, while assuring that we are not overwriting any data
-        assert "res_motor" not in self.pp_output_data
-        self.pp_output_data["res_motor"] = self._pp_load_result_accumulate(
-            pp_component_name="motor", load_id_names=load_id_names
+        assert res_table not in self.pp_output_data
+        self.pp_output_data[res_table] = self._pp_load_result_accumulate(
+            pp_component_name=element, load_id_names=load_id_names
         )
 
     def _pp_load_result_accumulate(self, pp_component_name: str, load_id_names: List[str]) -> pd.DataFrame:
@@ -2134,64 +2105,6 @@ def _pp_trafos_output_3ph(self):  # pylint: disable=too-many-statements
         assert "res_trafo_3ph" not in self.pp_output_data
         self.pp_output_data["res_trafo_3ph"] = pp_output_trafos_3ph
 
-    def _pp_loads_output_3ph(self):
-        """
-        This function converts a power-grid-model Symmetrical Load output array to a Load Dataframe of PandaPower.
-
-        Returns:
-            a PandaPower Dataframe for the Load component
-        """
-        load_id_names = ["const_power", "const_impedance", "const_current"]
-        if (
-            ComponentType.sym_load not in self.pgm_output_data
-            or self.pgm_output_data[ComponentType.sym_load].size == 0
-            or ("load", load_id_names[0]) not in self.idx
-        ):
-            return
-
-        # Store the results, while assuring that we are not overwriting any data
-        assert "res_load_3ph" not in self.pp_output_data
-        self.pp_output_data["res_load_3ph"] = self._pp_load_result_accumulate(
-            pp_component_name="load", load_id_names=load_id_names
-        )
-
-    def _pp_ward_output_3ph(self):
-        # TODO: Create Unit tests
-        load_id_names = ["ward_const_power_load", "ward_const_impedance_load"]
-        if (
-            ComponentType.sym_load not in self.pgm_output_data
-            or self.pgm_output_data[ComponentType.sym_load].size == 0
-            or ("ward", load_id_names[0]) not in self.idx
-        ):
-            return
-
-        # TODO Find a better way for mapping vm_pu from bus
-        # accumulated_loads["vm_pu"] = np.nan
-        # Store the results, while assuring that we are not overwriting any data
-        assert "res_ward_3ph" not in self.pp_output_data
-        self.pp_output_data["res_ward_3ph"] = self._pp_load_result_accumulate(
-            pp_component_name="ward", load_id_names=load_id_names
-        )
-
-    def _pp_motor_output_3ph(self):
-        # TODO: Create unit tests
-        load_id_names = ["motor_load"]
-
-        assert "res_motor_3ph" not in self.pp_output_data
-
-        if (
-            ComponentType.sym_load not in self.pgm_output_data
-            or self.pgm_output_data[ComponentType.sym_load].size == 0
-            or ("motor", load_id_names[0]) not in self.idx
-        ):
-            return
-
-        # Store the results, while assuring that we are not overwriting any data
-        assert "res_motor" not in self.pp_output_data
-        self.pp_output_data["res_motor_3ph"] = self._pp_load_result_accumulate(
-            pp_component_name="motor", load_id_names=load_id_names
-        )
-
     def _pp_shunts_output_3ph(self):
         """
         This function converts a power-grid-model Shunt output array to a Shunt Dataframe of PandaPower.

tests/unit/converters/test_pandapower_converter_output.py

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: MPL-2.0
 
 from typing import Callable, List
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, MagicMock, call, patch
 
 import numpy as np
 import pandas as pd
@@ -38,11 +38,14 @@ def test_create_output_data():
     converter._pp_sgens_output.assert_called_once_with()
     converter._pp_trafos_output.assert_called_once_with()
     converter._pp_trafos3w_output.assert_called_once_with()
-    converter._pp_loads_output.assert_called_once_with()
+    expected_calls = [
+        call(element="load", symmetric=True),
+        call(element="ward", symmetric=True),
+        call(element="motor", symmetric=True),
+    ]
+    converter._pp_load_elements_output.assert_has_calls(expected_calls)
     converter._pp_asym_loads_output.assert_called_once_with()
     converter._pp_asym_gens_output.assert_called_once_with()
-    converter._pp_motor_output.assert_called_once_with()
-    converter._pp_ward_output.assert_called_once_with()
     converter._pp_switches_output.assert_called_once_with()
 
 
@@ -60,7 +63,12 @@ def test_create_output_data_3ph():
     converter._pp_ext_grids_output_3ph.assert_called_once_with()
     converter._pp_sgens_output_3ph.assert_called_once_with()
     converter._pp_trafos_output_3ph.assert_called_once_with()
-    converter._pp_loads_output_3ph.assert_called_once_with()
+    expected_calls = [
+        call(element="load", symmetric=False),
+        call(element="ward", symmetric=False),
+        call(element="motor", symmetric=False),
+    ]
+    converter._pp_load_elements_output.assert_has_calls(expected_calls)
     converter._pp_asym_loads_output_3ph.assert_called_once_with()
     converter._pp_asym_gens_output_3ph.assert_called_once_with()
 
@@ -75,18 +83,13 @@ def test_create_output_data_3ph():
         (PandaPowerConverter._pp_sgens_output, "sym_gen"),
         (PandaPowerConverter._pp_trafos_output, "transformer"),
         (PandaPowerConverter._pp_trafos3w_output, "three_winding_transformer"),
-        (PandaPowerConverter._pp_loads_output, "sym_load"),
-        (PandaPowerConverter._pp_asym_loads_output, "asym_load"),
         (PandaPowerConverter._pp_asym_gens_output, "asym_gen"),
-        (PandaPowerConverter._pp_ward_output, "ward"),
-        (PandaPowerConverter._pp_motor_output, "motor"),
         (PandaPowerConverter._pp_switches_output, "link"),
         (PandaPowerConverter._pp_buses_output_3ph, "node"),
         (PandaPowerConverter._pp_lines_output_3ph, "line"),
         (PandaPowerConverter._pp_ext_grids_output_3ph, "source"),
         (PandaPowerConverter._pp_sgens_output_3ph, "sym_gen"),
         (PandaPowerConverter._pp_trafos_output_3ph, "transformer"),
-        (PandaPowerConverter._pp_loads_output_3ph, "sym_load"),
         (PandaPowerConverter._pp_asym_loads_output_3ph, "asym_load"),
         (PandaPowerConverter._pp_asym_gens_output_3ph, "asym_gen"),
     ],
@@ -109,6 +112,34 @@ def test_create_pp_output_object__empty(create_fn: Callable[[PandaPowerConverter
         mock_df.assert_not_called()
 
 
+@pytest.mark.parametrize(
+    ("create_fn", "element", "symmetric", "table"),
+    [
+        (PandaPowerConverter._pp_load_elements_output, "load", True, "sym_load"),
+        (PandaPowerConverter._pp_load_elements_output, "ward", True, "ward"),
+        (PandaPowerConverter._pp_load_elements_output, "motor", True, "motor"),
+    ],
+)
+def test_create_pp_output_object_with_params__empty(
+    create_fn: Callable[[PandaPowerConverter, str, bool], None], element: str, symmetric: bool, table: str
+):
+    # Arrange: No table
+    converter = PandaPowerConverter()
+
+    # Act / Assert
+    with patch("power_grid_model_io.converters.pandapower_converter.pd.DataFrame") as mock_df:
+        create_fn(converter, element, symmetric)
+        mock_df.assert_not_called()
+
+    # Arrange: Empty table
+    converter.pgm_output_data[table] = np.array([])  # type: ignore
+
+    # Act / Assert
+    with patch("power_grid_model_io.converters.pandapower_converter.pd.DataFrame") as mock_df:
+        create_fn(converter, element, symmetric)
+        mock_df.assert_not_called()
+
+
 def test_output_bus(converter):
     # Arrange
     mock_pgm_array = MagicMock()
@@ -532,15 +563,34 @@ def test_pp_load_result_accumulate__asym():
 
 
 @pytest.mark.parametrize(
-    ("output_fn", "table", "load_id_names", "result_suffix"),
+    ("output_fn", "element", "symmetric", "table", "load_id_names", "result_suffix"),
     [
-        (PandaPowerConverter._pp_loads_output, "load", ["const_power", "const_impedance", "const_current"], ""),
-        (PandaPowerConverter._pp_motor_output, "motor", ["motor_load"], ""),
-        (PandaPowerConverter._pp_loads_output_3ph, "load", ["const_power", "const_impedance", "const_current"], "_3ph"),
+        (
+            PandaPowerConverter._pp_load_elements_output,
+            "load",
+            True,
+            "load",
+            ["const_power", "const_impedance", "const_current"],
+            "",
+        ),
+        (PandaPowerConverter._pp_load_elements_output, "motor", True, "motor", ["motor_load"], ""),
+        (
+            PandaPowerConverter._pp_load_elements_output,
+            "load",
+            False,
+            "load",
+            ["const_power", "const_impedance", "const_current"],
+            "_3ph",
+        ),
     ],
 )
 def test_output_load_types(
-    output_fn: Callable[[PandaPowerConverter], None], table: str, load_id_names: List[str], result_suffix: str
+    output_fn: Callable[[PandaPowerConverter, str, bool], None],
+    element: str,
+    symmetric: bool,
+    table: str,
+    load_id_names: List[str],
+    result_suffix: str,
 ):
     # Arrange
     converter = PandaPowerConverter()
@@ -551,7 +601,7 @@ def test_output_load_types(
     converter._pp_load_result_accumulate = MagicMock()  # type: ignore
 
     # Act
-    output_fn(converter)
+    output_fn(converter, element, symmetric)
 
     # Assert
     converter._pp_load_result_accumulate.assert_called_once_with(pp_component_name=table, load_id_names=load_id_names)
@@ -570,7 +620,7 @@ def test_output_load_ward():
     converter._pp_load_result_accumulate = MagicMock()
 
     # Act
-    converter._pp_ward_output()
+    converter._pp_load_elements_output(element="ward", symmetric=True)
 
     # Assert
     converter._pp_load_result_accumulate.assert_called_once_with(pp_component_name="ward", load_id_names=load_id_names)

tests/validation/converters/test_pandapower_converter_output.py

@@ -357,6 +357,7 @@ def run_pf_asym_with_pgm(net):
             net[table] = output_tables[table]
 
     net = pp_networks.ieee_european_lv_asymmetric()
+    pp.create_load(net, 50, 0.1, 0.05)
     run_pf_asym_with_pgm(net)
     s_ext_grid, s_load, s_loss = _get_total_powers_3ph(net)
     assert np.isclose(s_ext_grid, (s_load + s_loss))
@@ -430,6 +431,7 @@ def run_pf_sym_with_pgm(net):
             net[table] = output_tables[table]
 
     net = pp_networks.ieee_european_lv_asymmetric()
+    pp.create_load(net, 50, 0.1, 0.05)
     run_pf_sym_with_pgm(net)
     s_ext_grid, s_load, s_loss = _get_total_powers(net)
     assert np.isclose(s_ext_grid, (s_load + s_loss))