cleanup

Signed-off-by: Martijn Govers <[email protected]>

src/power_grid_model_io/converters/pandapower_converter.py

@@ -17,7 +17,7 @@
 from power_grid_model_io.converters.base_converter import BaseConverter
 from power_grid_model_io.data_types import ExtraInfo
 from power_grid_model_io.functions import get_winding
-from power_grid_model_io.utils.parsing import parse_node_ref, parse_trafo3_connection, parse_trafo_connection
+from power_grid_model_io.utils.parsing import is_node_ref, parse_trafo3_connection, parse_trafo_connection
 
 PandaPowerData = MutableMapping[str, pd.DataFrame]
 
@@ -170,7 +170,7 @@ def _fill_pgm_extra_info(self, extra_info: ExtraInfo):
         extra_cols = ["i_n"]
         for component_data in self.pgm_input_data.values():
             for attr_name in component_data.dtype.names:
-                if not parse_node_ref(attr_name) and attr_name not in extra_cols:
+                if not is_node_ref(attr_name) and attr_name not in extra_cols:
                     continue
                 for pgm_id, node_id in component_data[["id", attr_name]]:
                     if pgm_id not in extra_info:
@@ -248,7 +248,7 @@ def _extra_info_to_pgm_input_data(self, extra_info: ExtraInfo):  # pylint: disab
         all_other_cols = ["i_n"]
         for component, data in self.pgm_output_data.items():
             input_cols = power_grid_meta_data["input"][component].dtype.names
-            node_cols = [col for col in input_cols if parse_node_ref(col)]
+            node_cols = [col for col in input_cols if is_node_ref(col)]
             other_cols = [col for col in input_cols if col in all_other_cols]
             if not node_cols + other_cols:
                 continue
@@ -2291,8 +2291,6 @@ def get_trafo_winding_types(self) -> pd.DataFrame:
         @lru_cache
         def vector_group_to_winding_types(vector_group: str) -> pd.Series:
             trafo_connection = parse_trafo_connection(vector_group)
-            if not trafo_connection:
-                raise ValueError(f"Invalid transformer connection string: '{vector_group}'")
             winding_from = get_winding(trafo_connection["winding_from"]).value
             winding_to = get_winding(trafo_connection["winding_to"]).value
             return pd.Series([winding_from, winding_to])
@@ -2316,8 +2314,6 @@ def get_trafo3w_winding_types(self) -> pd.DataFrame:
         @lru_cache
         def vector_group_to_winding_types(vector_group: str) -> pd.Series:
             trafo_connection = parse_trafo3_connection(vector_group)
-            if not trafo_connection:
-                raise ValueError(f"Invalid transformer connection string: '{vector_group}'")
             winding_1 = get_winding(trafo_connection["winding_1"]).value
             winding_2 = get_winding(trafo_connection["winding_2"]).value
             winding_3 = get_winding(trafo_connection["winding_3"]).value

src/power_grid_model_io/functions/phase_to_phase.py

@@ -6,7 +6,6 @@
 """
 
 import math
-from typing import Tuple
 
 import structlog
 from power_grid_model import WindingType
@@ -76,68 +75,36 @@ def power_wind_speed(  # pylint: disable=too-many-arguments
     return 0.0
 
 
-def get_winding_from(conn_str: str, neutral_grounding: bool = True) -> WindingType:
-    """
-    Get the winding type, based on a textual encoding of the conn_str
-    """
-    winding_from, _, _ = _split_connection_string(conn_str)
-    return get_winding(winding=winding_from, neutral_grounding=neutral_grounding)
-
-
-def get_winding_to(conn_str: str, neutral_grounding: bool = True) -> WindingType:
-    """
-    Get the winding type, based on a textual encoding of the conn_str
-    """
-    _, winding_to, _ = _split_connection_string(conn_str)
-    return get_winding(winding=winding_to, neutral_grounding=neutral_grounding)
-
-
-def get_winding_1(conn_str: str, neutral_grounding: bool = True) -> WindingType:
-    """
-    Get the winding type, based on a textual encoding of the conn_str
-    """
-    winding_1, _, _, _, _ = _split_connection_string_3w(conn_str)
-    return get_winding(winding=winding_1, neutral_grounding=neutral_grounding)
-
-
-def get_winding_2(conn_str: str, neutral_grounding: bool = True) -> WindingType:
-    """
-    Get the winding type, based on a textual encoding of the conn_str
-    """
-    _, winding_2, _, _, _ = _split_connection_string_3w(conn_str)
-    return get_winding(winding=winding_2, neutral_grounding=neutral_grounding)
+def _get_winding(trafo_connection_parser, winding_ref: str):
+    def _get_winding_impl(conn_str: str, neutral_grounding: bool = True) -> WindingType:
+        """
+        Get the winding type, based on a textual encoding of the conn_str
+        """
+        return get_winding(trafo_connection_parser(conn_str)[winding_ref], neutral_grounding=neutral_grounding)
 
+    return _get_winding_impl
 
-def get_winding_3(conn_str: str, neutral_grounding: bool = True) -> WindingType:
-    """
-    Get the winding type, based on a textual encoding of the conn_str
-    """
-    _, _, _, winding_3, _ = _split_connection_string_3w(conn_str)
-    return get_winding(winding=winding_3, neutral_grounding=neutral_grounding)
 
+def _get_clock(trafo_connection_parser, clock_ref: str):
+    def _get_clock_impl(conn_str: str) -> int:
+        """
+        Extract the clock part of the conn_str
+        """
+        return int(trafo_connection_parser(conn_str)[clock_ref])
 
-def get_clock(conn_str: str) -> int:
-    """
-    Extract the clock part of the conn_str
-    """
-    _, _, clock = _split_connection_string(conn_str)
-    return clock
+    return _get_clock_impl
 
 
-def get_clock_12(conn_str: str) -> int:
-    """
-    Extract the clock part of the conn_str
-    """
-    _, _, clock_12, _, _ = _split_connection_string_3w(conn_str)
-    return clock_12
+get_winding_from = _get_winding(parse_trafo_connection, "winding_from")
+get_winding_to = _get_winding(parse_trafo_connection, "winding_to")
+get_winding_1 = _get_winding(parse_trafo3_connection, "winding_1")
+get_winding_2 = _get_winding(parse_trafo3_connection, "winding_2")
+get_winding_3 = _get_winding(parse_trafo3_connection, "winding_3")
 
 
-def get_clock_13(conn_str: str) -> int:
-    """
-    Extract the clock part of the conn_str
-    """
-    _, _, _, _, clock_13 = _split_connection_string_3w(conn_str)
-    return clock_13
+get_clock = _get_clock(parse_trafo_connection, "clock")
+get_clock_12 = _get_clock(parse_trafo3_connection, "clock_12")
+get_clock_13 = _get_clock(parse_trafo3_connection, "clock_13")
 
 
 def reactive_power_to_susceptance(q: float, u_nom: float) -> float:
@@ -147,49 +114,19 @@ def reactive_power_to_susceptance(q: float, u_nom: float) -> float:
     return q / u_nom / u_nom
 
 
-def _split_connection_string(conn_str: str) -> Tuple[str, str, int]:
-    """
-    Helper function to split the conn_str into three parts:
-     * winding_from
-     * winding_to
-     * clock
-    """
-    trafo_connection = parse_trafo_connection(conn_str)
-    if not trafo_connection:
-        raise ValueError(f"Invalid transformer connection string: '{conn_str}'")
-    return trafo_connection["winding_from"], trafo_connection["winding_to"], int(trafo_connection["clock_number"])
-
-
-def _split_connection_string_3w(conn_str: str) -> Tuple[str, str, int, str, int]:
-    """
-    Helper function to split the conn_str into three parts:
-     * winding_1
-     * winding_2
-     * clock 12
-     * winding_3
-     * clock 13
-    """
-    trafo_connection = parse_trafo3_connection(conn_str)
-    if not trafo_connection:
-        raise ValueError(f"Invalid three winding transformer connection string: '{conn_str}'")
-    return (
-        trafo_connection["winding_1"],
-        trafo_connection["winding_2"],
-        int(trafo_connection["clock_12"]),
-        trafo_connection["winding_3"],
-        int(trafo_connection["clock_13"]),
-    )
-
-
 def pvs_power_adjustment(p: float, efficiency_type: str) -> float:
     """
     Adjust power of PV for the default efficiency type of 97% or 95%. Defaults to 100 % for other custom types
     """
-    pvs_efficiency_type = parse_pvs_efficiency_type(efficiency_type)
-    if pvs_efficiency_type is not None:
-        _LOG.warning("PV approximation applied for efficiency type", efficiency_type=efficiency_type)
-        if pvs_efficiency_type == "97":
-            return p * 0.97
-        if pvs_efficiency_type == "95":
-            return p * 0.95
+    try:
+        pvs_efficiency_type = parse_pvs_efficiency_type(efficiency_type)
+    except ValueError:
+        return p
+
+    _LOG.warning("PV approximation applied for efficiency type", efficiency_type=efficiency_type)
+    if pvs_efficiency_type == "97":
+        return p * 0.97
+    if pvs_efficiency_type == "95":
+        return p * 0.95
+
     return p

src/power_grid_model_io/utils/parsing.py

@@ -6,13 +6,13 @@
 """
 
 import re
-from typing import Dict, Optional
+from typing import Dict
 
 _TRAFO_CONNECTION_RE = re.compile(r"^(Y|YN|D|Z|ZN)(y|yn|d|z|zn)(\d|1[0-2])?$")
 
 
-def parse_trafo_connection(string: str) -> Optional[Dict[str, str]]:
-    r"""Parse a trafo connection string if possible.
+def parse_trafo_connection(string: str) -> Dict[str, str]:
+    r"""Parse a trafo connection string.
 
     Matches the following regular expression to the winding_from and winding_to codes.
     Optionally checks the clock number:
@@ -26,21 +26,24 @@ def parse_trafo_connection(string: str) -> Optional[Dict[str, str]]:
     Args:
         string (str): The input string.
 
+    Raises:
+        ValueError: If the input is not a trafo connection string.
+
     Returns:
-        Optional[Dict[str, str]]: The parameters of the trafo connection if correct, else None.
+        Dict[str, str]: The parameters of the trafo connection.
     """
     match = _TRAFO_CONNECTION_RE.fullmatch(string)
     if not match:
-        return None
+        raise ValueError(f"Invalid transformer connection string: '{string}'")
 
-    return {"winding_from": match.group(1), "winding_to": match.group(2), "clock_number": match.group(3)}
+    return {"winding_from": match.group(1), "winding_to": match.group(2), "clock": match.group(3)}
 
 
 _TRAFO3_CONNECTION_RE = re.compile(r"^(Y|YN|D|Z|ZN)(y|yn|d|z|zn)(\d|1[0-2])?(y|yn|d|z|zn)(\d|1[0-2])?$")
 
 
-def parse_trafo3_connection(string: str) -> Optional[Dict[str, str]]:
-    r"""Parse a trafo connection string if possible.
+def parse_trafo3_connection(string: str) -> Dict[str, str]:
+    r"""Parse a trafo connection string.
 
     Matches the following regular expression to the winding_1, winding_2 and winding_3 codes.
     Optionally checks the clock numbers:
@@ -56,12 +59,15 @@ def parse_trafo3_connection(string: str) -> Optional[Dict[str, str]]:
     Args:
         string (str): The input string.
 
+    Raises:
+        ValueError: If the input is not a trafo connection string.
+
     Returns:
-        Optional[Dict[str, str]]: The parameters of the trafo connection if correct, else None.
+        Dict[str, str]: The parameters of the trafo connection.
     """
     match = _TRAFO3_CONNECTION_RE.fullmatch(string)
     if not match:
-        return None
+        raise ValueError(f"Invalid three winding transformer connection string: '{string}'")
 
     return {
         "winding_1": match.group(1),
@@ -72,8 +78,8 @@ def parse_trafo3_connection(string: str) -> Optional[Dict[str, str]]:
     }
 
 
-def parse_node_ref(string: str) -> Optional[Dict[str, str]]:
-    """Parse a node reference. string if possible.
+def parse_node_ref(string: str) -> Dict[str, str]:
+    """Parse a node reference string.
 
     Matches if the input is the word 'node' with an optional prefix or suffix. E.g.:
 
@@ -84,30 +90,55 @@ def parse_node_ref(string: str) -> Optional[Dict[str, str]]:
     Args:
         string (str): The input string.
 
+    Raises:
+        ValueError: If the input string is not a node reference.
+
     Returns:
-        Optional[Dict[str, str]]: The prefix and suffix if the input string is a reference to a node, else None.
+        Optional[Dict[str, str]]: The prefix and suffix (may be empty).
     """
+
+    def _raise():
+        raise ValueError(f"Invalid node reference string: '{string}'")
+
     if "node" not in string:
-        return None
+        _raise()
 
     prefix_and_suffix = string.split("node")
     if len(prefix_and_suffix) != 2:
-        return None
+        _raise()
 
     prefix, suffix = prefix_and_suffix
     if prefix and not prefix.endswith("_"):
-        return None
+        _raise()
     if suffix and not suffix.startswith("_"):
-        return None
+        _raise()
 
     return {"prefix": prefix, "suffix": suffix}
 
 
+def is_node_ref(string: str) -> bool:
+    """Return True if the string represents a node reference, else False.
+
+    Like parse_node_ref, but without exceptions and result data.
+
+    Args:
+        string (str): The input string.
+
+    Returns:
+        bool: True if the string represents a node reference, else False.
+    """
+    try:
+        parse_node_ref(string)
+        return True
+    except ValueError:
+        return False
+
+
 _PVS_EFFICIENCY_TYPE_RE = re.compile(r"[ ,.]1 pu: (95|97) %")
 
 
-def parse_pvs_efficiency_type(string: str) -> Optional[str]:
-    r"""Parse a PVS efficiency type string if possible.
+def parse_pvs_efficiency_type(string: str) -> str:
+    r"""Parse a PVS efficiency type string.
 
     Matches the following regex to the efficiency type percentage at 1 pu.
 
@@ -123,11 +154,14 @@ def parse_pvs_efficiency_type(string: str) -> Optional[str]:
     Args:
         string (str): The input string.
 
+    Raises:
+        ValueError: If the input string is not a PVS efficiency type.
+
     Returns:
-        Optional[str]: The efficiency type percentage string at 1 pu if correct, else None.
+        Optional[str]: The efficiency type percentage string at 1 pu.
     """
     match = _PVS_EFFICIENCY_TYPE_RE.search(string)
     if not match:
-        return None
+        raise ValueError(f"Invalid PVS efficiency type string: '{string}'")
 
     return match.group(1)

tests/unit/functions/test_phase_to_phase.py

@@ -2,7 +2,6 @@
 #
 # SPDX-License-Identifier: MPL-2.0
 import numpy as np
-import pytest
 from power_grid_model.enum import WindingType
 from pytest import approx, mark, param, raises