add error handling, finalize tests

Signed-off-by: Laurynas Jagutis <[email protected]>

Author: Laurynas-Jagutis

setup.py

@@ -214,7 +214,7 @@ def generate_build_ext(pkg_dir: Path, pkg_name: str):
     # list of extensions
     exts = [
         CTypesExtension(
-            name="power_grid_model_io_native._core.power_grid_model_io_core",
+            name="power_grid_model_io_native._core._power_grid_model_io_core",
             sources=sources,
             include_dirs=include_dirs,
             library_dirs=library_dirs,

src/power_grid_model_io_native/_core/error_handling.py

@@ -0,0 +1,187 @@
+# SPDX-FileCopyrightText: Contributors to the Power Grid Model project <[email protected]>
+#
+# SPDX-License-Identifier: MPL-2.0
+
+"""
+Error handling
+"""
+
+import re
+
+from power_grid_model.errors import (
+    AutomaticTapCalculationError,
+    ConflictID,
+    ConflictVoltage,
+    IDNotFound,
+    IDWrongType,
+    InvalidArguments,
+    InvalidBranch,
+    InvalidBranch3,
+    InvalidCalculationMethod,
+    InvalidMeasuredObject,
+    InvalidRegulatedObject,
+    InvalidShortCircuitPhaseOrType,
+    InvalidTransformerClock,
+    IterationDiverge,
+    MaxIterationReached,
+    MissingCaseForEnumError,
+    NotObservableError,
+    PowerGridBatchError,
+    PowerGridDatasetError,
+    PowerGridError,
+    PowerGridNotImplementedError,
+    PowerGridSerializationError,
+    PowerGridUnreachableHitError,
+    SparseMatrixError,
+)
+
+from power_grid_model_io_native._core.power_grid_model_io_core import pgm_io_core as pgmic
+
+VALIDATOR_MSG = "\nTry validate_input_data() or validate_batch_data() to validate your data.\n"
+# error codes
+PGM_NO_ERROR = 0
+PGM_REGULAR_ERROR = 1
+PGM_BATCH_ERROR = 2
+PGM_SERIALIZATION_ERROR = 3
+
+_MISSING_CASE_FOR_ENUM_RE = re.compile(r" is not implemented for (.+) #(-?\d+)!\n")
+_INVALID_ARGUMENTS_RE = re.compile(r" is not implemented for ")  # multiple different flavors
+_CONFLICT_VOLTAGE_RE = re.compile(
+    r"Conflicting voltage for line (-?\d+)\n voltage at from node (-?\d+) is (.*)\n"
+    r" voltage at to node (-?\d+) is (.*)\n"
+)
+_INVALID_BRANCH_RE = re.compile(r"Branch (-?\d+) has the same from- and to-node (-?\d+),\n This is not allowed!\n")
+_INVALID_BRANCH3_RE = re.compile(
+    r"Branch3 (-?\d+) is connected to the same node at least twice. Node 1\/2\/3: (-?\d+)\/(-?\d+)\/(-?\d+),\n"
+    r" This is not allowed!\n"
+)
+_INVALID_TRANSFORMER_CLOCK_RE = re.compile(r"Invalid clock for transformer (-?\d+), clock (-?\d+)\n")
+_SPARSE_MATRIX_ERROR_RE = re.compile(r"Sparse matrix error")  # multiple different flavors
+_NOT_OBSERVABLE_ERROR_RE = re.compile(r"Not enough measurements available for state estimation.\n")
+_ITERATION_DIVERGE_RE = re.compile(r"Iteration failed to converge")  # potentially multiple different flavors
+_MAX_ITERATION_REACHED_RE = re.compile(r"Maximum number of iterations reached")
+_CONFLICT_ID_RE = re.compile(r"Conflicting id detected: (-?\d+)\n")
+_ID_NOT_FOUND_RE = re.compile(r"The id cannot be found: (-?\d+)\n")
+_INVALID_MEASURED_OBJECT_RE = re.compile(r"(\w+) measurement is not supported for object of type (\w+)")
+_INVALID_REGULATED_OBJECT_RE = re.compile(
+    r"(\w+) regulator is not supported for object "
+)  # potentially multiple different flavors
+_AUTOMATIC_TAP_CALCULATION_ERROR_RE = re.compile(
+    r"Automatic tap changing regulator with tap_side at LV side is not supported. Found at id (-?\d+)\n"
+)
+_ID_WRONG_TYPE_RE = re.compile(r"Wrong type for object with id (-?\d+)\n")
+_INVALID_CALCULATION_METHOD_RE = re.compile(r"The calculation method is invalid for this calculation!")
+_INVALID_SHORT_CIRCUIT_PHASE_OR_TYPE_RE = re.compile(r"short circuit type")  # multiple different flavors
+_POWER_GRID_DATASET_ERROR_RE = re.compile(r"Dataset error: ")  # multiple different flavors
+_POWER_GRID_UNREACHABLE_HIT_RE = re.compile(r"Unreachable code hit when executing ")  # multiple different flavors
+_POWER_GRID_SEARCH_OPT_INCMPT_RE = re.compile(r"Search method is incompatible with optimization strategy: ")
+_POWER_GRID_NOT_IMPLEMENTED_ERROR_RE = re.compile(r"The functionality is either not supported or not yet implemented!")
+
+_ERROR_MESSAGE_PATTERNS = {
+    _MISSING_CASE_FOR_ENUM_RE: MissingCaseForEnumError,
+    _INVALID_ARGUMENTS_RE: InvalidArguments,
+    _CONFLICT_VOLTAGE_RE: ConflictVoltage,
+    _INVALID_BRANCH_RE: InvalidBranch,
+    _INVALID_BRANCH3_RE: InvalidBranch3,
+    _INVALID_TRANSFORMER_CLOCK_RE: InvalidTransformerClock,
+    _SPARSE_MATRIX_ERROR_RE: SparseMatrixError,
+    _NOT_OBSERVABLE_ERROR_RE: NotObservableError,
+    _ITERATION_DIVERGE_RE: IterationDiverge,
+    _MAX_ITERATION_REACHED_RE: MaxIterationReached,
+    _CONFLICT_ID_RE: ConflictID,
+    _ID_NOT_FOUND_RE: IDNotFound,
+    _INVALID_MEASURED_OBJECT_RE: InvalidMeasuredObject,
+    _INVALID_REGULATED_OBJECT_RE: InvalidRegulatedObject,
+    _AUTOMATIC_TAP_CALCULATION_ERROR_RE: AutomaticTapCalculationError,
+    _ID_WRONG_TYPE_RE: IDWrongType,
+    _INVALID_CALCULATION_METHOD_RE: InvalidCalculationMethod,
+    _INVALID_SHORT_CIRCUIT_PHASE_OR_TYPE_RE: InvalidShortCircuitPhaseOrType,
+    _POWER_GRID_DATASET_ERROR_RE: PowerGridDatasetError,
+    _POWER_GRID_UNREACHABLE_HIT_RE: PowerGridUnreachableHitError,
+    _POWER_GRID_SEARCH_OPT_INCMPT_RE: PowerGridUnreachableHitError,
+    _POWER_GRID_NOT_IMPLEMENTED_ERROR_RE: PowerGridNotImplementedError,
+}
+
+
+def _interpret_error(message: str, decode_error: bool = True) -> PowerGridError:
+    if decode_error:
+        for pattern, type_ in _ERROR_MESSAGE_PATTERNS.items():
+            if pattern.search(message) is not None:
+                return type_(message)
+
+    return PowerGridError(message)
+
+
+def find_error(batch_size: int = 1, decode_error: bool = True) -> RuntimeError | None:
+    """
+    Check if there is an error and return it
+
+    Args:
+        batch_size: (int, optional): Size of batch. Defaults to 1.
+        decode_error (bool, optional): Decode the error message(s) to derived error classes. Defaults to True
+
+    Returns: error object, can be none
+
+    """
+    error_code: int = pgmic.error_code()
+    if error_code == PGM_NO_ERROR:
+        return None
+    if error_code == PGM_REGULAR_ERROR:
+        error_message = pgmic.error_message()
+        error_message += VALIDATOR_MSG
+        return _interpret_error(error_message, decode_error=decode_error)
+    if error_code == PGM_BATCH_ERROR:
+        _ = batch_size
+        error_message = "There are errors in the batch calculation." + VALIDATOR_MSG
+        error = PowerGridBatchError(error_message)
+        return error
+    if error_code == PGM_SERIALIZATION_ERROR:
+        return PowerGridSerializationError(pgmic.error_message())
+    return RuntimeError("Unknown error!")
+
+
+def assert_no_error(batch_size: int = 1, decode_error: bool = True):
+    """
+    Assert there is no error in the last operation
+    If there is an error, raise it
+
+    Args:
+        batch_size (int, optional): Size of batch. Defaults to 1.
+        decode_error (bool, optional): Decode the error message(s) to derived error classes. Defaults to True
+
+    Returns:
+
+    """
+    error = find_error(batch_size=batch_size, decode_error=decode_error)
+    if error is not None:
+        raise error
+
+
+def handle_errors(
+    continue_on_batch_error: bool, batch_size: int = 1, decode_error: bool = True
+) -> PowerGridBatchError | None:
+    """
+    Handle any errors in the way that is specified.
+
+    Args:
+        continue_on_batch_error (bool): Return the error when the error type is a batch error instead of reraising it.
+        batch_size (int, optional): Size of batch. Defaults to 1.
+        decode_error (bool, optional): Decode the error message(s) to derived error classes. Defaults to True
+
+    Raises:
+        error: Any errors previously encountered, unless it was a batch error and continue_on_batch_error was True.
+
+    Returns:
+        PowerGridBatchError | None: None if there were no errors, or the previously encountered
+                                    error if it was a batch error and continue_on_batch_error was True.
+    """
+    error: RuntimeError | None = find_error(batch_size=batch_size, decode_error=decode_error)
+    if error is None:
+        return None
+
+    if continue_on_batch_error and isinstance(error, PowerGridBatchError):
+        # continue on batch error
+        return error
+
+    # raise normal error
+    raise error

src/power_grid_model_io_native/_core/power_grid_model_io_core.py

@@ -9,7 +9,7 @@
 from inspect import signature
 from itertools import chain
 from pathlib import Path
-from typing import Callable, Optional
+from typing import Callable
 
 from power_grid_model._core.power_grid_core import CharPtr, CStr, IdxC
 
@@ -40,9 +40,9 @@ def _load_core() -> CDLL:
 
     """
     if platform.system() == "Windows":
-        dll_file = "power_grid_model_io_core.dll"
+        dll_file = "_power_grid_model_io_core.dll"
     else:
-        dll_file = "power_grid_model_io_core.so"
+        dll_file = "_power_grid_model_io_core.so"
     cdll = CDLL(str(Path(__file__).parent / dll_file))
     # assign return types
     # handle
@@ -118,9 +118,9 @@ class PowerGridModelIoCore:
     """
 
     _handle: HandlePtr
-    _instance: Optional["PowerGridModelIoCore"] = None
+    _instance: "PowerGridModelIoCore | None" = None
 
-    # singleton of power grid core
+    # singleton of power grid model io core
     def __new__(cls, *args, **kwargs):
         if cls._instance is None:
             cls._instance = super().__new__(cls, *args, **kwargs)

src/power_grid_model_io_native/_core/vnf_converter.py

@@ -6,8 +6,7 @@
 Power grid model io native converter for vnf files
 """
 
-from power_grid_model._core.error_handling import assert_no_error
-
+from power_grid_model_io_native._core.error_handling import assert_no_error
 from power_grid_model_io_native._core.power_grid_model_io_core import PgmVnfConverterPtr, pgm_io_core as pgmic
 
 

tests/unit/test_vnf_converter.py

@@ -4,20 +4,20 @@
 
 import pytest
 
+from power_grid_model_io_native._core.error_handling import InvalidArguments, assert_no_error
 from power_grid_model_io_native._core.vnf_converter import PgmVnfConverter
 
 
 def test_pgmvnfconverter_constructor_without_experimental_features():
     """A test case for creating pgmvnfconverter without experimental features"""
-    converter = PgmVnfConverter("", 0)
-    with pytest.raises(OSError):
-        _ = converter.get_pgm_input_data()
+    with pytest.raises(InvalidArguments):
+        _ = PgmVnfConverter("", 0)
 
 
 def test_pgmvnfconverter_constructor_with_experimental_features():
     """A test case for creating pgmvnfconverter with experimental features"""
-    converter = PgmVnfConverter("", 1)
-    _ = converter.get_pgm_input_data()
+    _ = PgmVnfConverter("", 1)
+    assert_no_error()
 
 
 def test_get_pgm_input_data():