Merge pull request #634 from PowerGridModel/feature/validation-check-…

…control-side-tap

Feature/validation check control side tap

src/power_grid_model/validation/errors.py

@@ -467,3 +467,55 @@ class FaultPhaseError(MultiFieldValidationError):
     """
 
     _message = "The fault phase is not applicable to the corresponding fault type for {n} {objects}."
+
+
+class InvalidAssociatedEnumValueError(MultiFieldValidationError):
+    """
+    The value is not a valid value in combination with the other specified attributes.
+    E.g. When a transformer tap regulator has a branch3 control side but regulates a transformer.
+    """
+
+    _message = "The combination of fields {field} results in invalid {enum} values for {n} {objects}."
+    enum: Union[Type[Enum], List[Type[Enum]]]
+
+    def __init__(
+        self,
+        component: str,
+        fields: List[str],
+        ids: List[int],
+        enum: Union[Type[Enum], List[Type[Enum]]],
+    ):
+        """
+        Args:
+            component: Component name
+            fields: List of field names
+            ids: List of component IDs (not row indices)
+            enum: The supported enum values
+        """
+        super().__init__(component, fields, ids)
+        self.enum = enum
+
+    @property
+    def enum_str(self) -> str:
+        """
+        A string representation of the field to which this error applies.
+        """
+        if isinstance(self.enum, list):
+            return ",".join(e.__name__ for e in self.enum)
+
+        return self.enum.__name__
+
+    def __eq__(self, other):
+        return super().__eq__(other) and self.enum == other.enum
+
+
+class UnsupportedTransformerRegulationError(MultiFieldValidationError):
+    """
+    The control side of a branch regulator is not supported for the regulated object
+    """
+
+    _message = (
+        "Unsupported control side for {n} {objects}. "
+        "The control side cannot be on the same side as the tap side. "
+        "The node at the control side must have the same or lower u_rated as the node at the tap side."
+    )

src/power_grid_model/validation/rules.py

@@ -46,6 +46,7 @@
     FaultPhaseError,
     IdNotInDatasetError,
     InfinityError,
+    InvalidAssociatedEnumValueError,
     InvalidEnumValueError,
     InvalidIdError,
     MissingValueError,
@@ -63,9 +64,17 @@
     SameValueError,
     TransformerClockError,
     TwoValuesZeroError,
+    UnsupportedTransformerRegulationError,
     ValidationError,
 )
-from power_grid_model.validation.utils import eval_expression, nan_type, set_default_value
+from power_grid_model.validation.utils import (
+    eval_expression,
+    get_indexer,
+    get_mask,
+    get_valid_ids,
+    nan_type,
+    set_default_value,
+)
 
 Error = TypeVar("Error", bound=ValidationError)
 CompError = TypeVar("CompError", bound=ComparisonError)
@@ -481,7 +490,7 @@ def all_valid_enum_values(
         data (SingleDataset): The input/update data set for all components
         component (str): The component of interest
         field (str): The field of interest
-        enum (Type[Enum]): The enum type to validate against, or a list of such enum types
+        enum (Type[Enum] | List[Type[Enum]]): The enum type to validate against, or a list of such enum types
 
     Returns:
         A list containing zero or one InvalidEnumValueError, listing all ids where the value in the field of interest
@@ -500,6 +509,48 @@ def all_valid_enum_values(
     return []
 
 
+def all_valid_associated_enum_values(  # pylint: disable=too-many-arguments
+    data: SingleDataset,
+    component: str,
+    field: str,
+    ref_object_id_field: str,
+    ref_components: List[str],
+    enum: Union[Type[Enum], List[Type[Enum]]],
+    **filters: Any,
+) -> List[InvalidAssociatedEnumValueError]:
+    """
+    Args:
+        data (SingleDataset): The input/update data set for all components
+        component (str): The component of interest
+        field (str): The field of interest
+        ref_object_id_field (str): The field that contains the referenced component ids
+        ref_components (List[str]): The component or components in which we want to look for ids
+        enum (Type[Enum] | List[Type[Enum]]): The enum type to validate against, or a list of such enum types
+        **filters: One or more filters on the dataset. E.g. regulated_object="transformer".
+
+    Returns:
+        A list containing zero or one InvalidAssociatedEnumValueError, listing all ids where the value in the field
+        of interest was not a valid value in the supplied enum type.
+    """
+    enums: List[Type[Enum]] = enum if isinstance(enum, list) else [enum]
+
+    valid_ids = get_valid_ids(data=data, ref_components=ref_components)
+    mask = np.logical_and(
+        get_mask(data=data, component=component, field=field, **filters),
+        np.isin(data[component][ref_object_id_field], valid_ids),
+    )
+
+    valid = {nan_type(component, field)}
+    for enum_type in enums:
+        valid.update(list(enum_type))
+
+    invalid = np.isin(data[component][field][mask], np.array(list(valid), dtype=np.int8), invert=True)
+    if invalid.any():
+        ids = data[component]["id"][mask][invalid].flatten().tolist()
+        return [InvalidAssociatedEnumValueError(component, [field, ref_object_id_field], ids, enum)]
+    return []
+
+
 def all_valid_ids(
     data: SingleDataset, component: str, field: str, ref_components: Union[str, List[str]], **filters: Any
 ) -> List[InvalidIdError]:
@@ -518,30 +569,11 @@ def all_valid_ids(
         A list containing zero or one InvalidIdError, listing all ids where the value in the field of interest
         was not a valid object identifier.
     """
-    # For convenience, ref_component may be a string and we'll convert it to a 'list' containing that string as it's
-    # single element.
-    if isinstance(ref_components, str):
-        ref_components = [ref_components]
-
-    # Create a set of ids by chaining the ids of all ref_components
-    valid_ids = set()
-    for ref_component in ref_components:
-        if ref_component in data:
-            nan = nan_type(ref_component, "id")
-            if np.isnan(nan):
-                mask = ~np.isnan(data[ref_component]["id"])
-            else:
-                mask = np.not_equal(data[ref_component]["id"], nan)
-            valid_ids.update(data[ref_component]["id"][mask])
-
-    # Apply the filters (e.g. to select only records with a certain MeasuredTerminalType)
-    values = data[component][field]
-    mask = np.ones(shape=values.shape, dtype=bool)
-    for filter_field, filter_value in filters.items():
-        mask = np.logical_and(mask, data[component][filter_field] == filter_value)
+    valid_ids = get_valid_ids(data=data, ref_components=ref_components)
+    mask = get_mask(data=data, component=component, field=field, **filters)
 
     # Find any values that can't be found in the set of ids
-    invalid = np.logical_and(mask, np.isin(values, list(valid_ids), invert=True))
+    invalid = np.logical_and(mask, np.isin(data[component][field], valid_ids, invert=True))
     if invalid.any():
         ids = data[component]["id"][invalid].flatten().tolist()
         return [InvalidIdError(component, field, ids, ref_components, filters)]
@@ -830,3 +862,47 @@ def _fault_phase_supported(fault_type: FaultType, fault_phase: FaultPhase):
             )
         ]
     return []
+
+
+def all_supported_tap_control_side(  # pylint: disable=too-many-arguments
+    data: SingleDataset,
+    component: str,
+    control_side_field: str,
+    regulated_object_field: str,
+    tap_side_fields: List[Tuple[str, str]],
+    **filters: Any,
+) -> List[UnsupportedTransformerRegulationError]:
+    """
+    Args:
+        data (SingleDataset): The input/update data set for all components
+        component (str): The component of interest
+        control_side_field (str): The field of interest
+        regulated_object_field (str): The field that contains the regulated component ids
+        tap_side_fields (List[Tuple[str, str]]): The fields of interest per regulated component,
+            formatted as [(component_1, field_1), (component_2, field_2)]
+        **filters: One or more filters on the dataset. E.g. regulated_object="transformer".
+
+    Returns:
+        A list containing zero or more InvalidAssociatedEnumValueErrors; listing all the ids
+        of components where the field of interest was invalid, given the referenced object's field.
+    """
+    mask = get_mask(data=data, component=component, field=control_side_field, **filters)
+    values = data[component][control_side_field][mask]
+
+    invalid = np.zeros_like(mask)
+
+    for ref_component, ref_field in tap_side_fields:
+        indices = get_indexer(data[ref_component]["id"], data[component][regulated_object_field], default_value=-1)
+        found = indices != -1
+        ref_comp_values = data[ref_component][ref_field][indices[found]]
+        invalid[found] = np.logical_or(invalid[found], values[found] == ref_comp_values)
+
+    if invalid.any():
+        return [
+            UnsupportedTransformerRegulationError(
+                component=component,
+                fields=[control_side_field, regulated_object_field],
+                ids=data[component]["id"][invalid].flatten().tolist(),
+            )
+        ]
+    return []

src/power_grid_model/validation/utils.py

@@ -6,7 +6,7 @@
 Utilities used for validation. Only errors_to_string() is intended for end users.
 """
 import re
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import numpy as np
 
@@ -168,26 +168,40 @@ def nan_type(component: str, field: str, data_type="input"):
     return power_grid_meta_data[data_type][component].nans[field]
 
 
-def get_indexer(input_ids: np.ndarray, update_ids: np.ndarray) -> np.ndarray:
+def get_indexer(source: np.ndarray, target: np.ndarray, default_value: Optional[int] = None) -> np.ndarray:
     """
-    Given array of ids from input and update dataset.
-    Find the posision of each id in the update dataset in the context of input dataset.
+    Given array of values from a source and a target dataset.
+    Find the position of each value in the target dataset in the context of the source dataset.
     This is needed to update values in the dataset by id lookup.
     Internally this is done by sorting the input ids, then using binary search lookup.
 
+    E.g.: Find the position of each id in an update (target) dataset in the input (source) dataset
+
+    >>> input_ids = [1, 2, 3, 4, 5]
+    >>> update_ids = [3]
+    >>> assert get_indexer(input_ids, update_ids) == np.array([2])
+
     Args:
-        input_ids: array of ids in the input dataset
-        update_ids: array of ids in the update dataset
+        source: array of values in the source dataset
+        target: array of values in the target dataset
+        default_value: Optional. the default index to provide for target values not in source
 
     Returns:
-        np.ndarray: array of positions of the ids from update dataset in the input dataset
-            the following should hold
-            input_ids[result] == update_ids
+        np.ndarray: array of positions of the values from target dataset in the source dataset
+            if default_value is None, (source[result] == target)
+            else, ((source[result] == target) | (source[result] == default_value))
+
+    Raises:
+        IndexError: if default_value is None and there were values in target that were not in source
     """
-    permutation_sort = np.argsort(input_ids)  # complexity O(N_input * logN_input)
-    return permutation_sort[
-        np.searchsorted(input_ids, update_ids, sorter=permutation_sort)
-    ]  # complexity O(N_update * logN_input)
+    permutation_sort = np.argsort(source)  # complexity O(N_input * logN_input)
+    indices = np.searchsorted(source, target, sorter=permutation_sort)  # complexity O(N_update * logN_input)
+
+    if default_value is None:
+        return permutation_sort[indices]
+
+    clipped_indices = np.take(permutation_sort, indices, mode="clip")
+    return np.where(source[clipped_indices] == target, permutation_sort[clipped_indices], default_value)
 
 
 def set_default_value(data: SingleDataset, component: str, field: str, default_value: Union[int, float, np.ndarray]):
@@ -214,3 +228,55 @@ def set_default_value(data: SingleDataset, component: str, field: str, default_v
         data[component][field][mask] = default_value[mask]
     else:
         data[component][field][mask] = default_value
+
+
+def get_valid_ids(data: SingleDataset, ref_components: Union[str, List[str]]) -> List[int]:
+    """
+    This function returns the valid IDs specified by all ref_components
+
+    Args:
+        data: The input/update data set for all components
+        ref_components: The component or components in which we want to look for ids
+
+    Returns:
+        List[int]: the list of valid IDs
+    """
+    # For convenience, ref_component may be a string and we'll convert it to a 'list' containing that string as it's
+    # single element.
+    if isinstance(ref_components, str):
+        ref_components = [ref_components]
+
+    # Create a set of ids by chaining the ids of all ref_components
+    valid_ids = set()
+    for ref_component in ref_components:
+        if ref_component in data:
+            nan = nan_type(ref_component, "id")
+            if np.isnan(nan):
+                mask = ~np.isnan(data[ref_component]["id"])
+            else:
+                mask = np.not_equal(data[ref_component]["id"], nan)
+            valid_ids.update(data[ref_component]["id"][mask])
+
+    return list(valid_ids)
+
+
+def get_mask(data: SingleDataset, component: str, field: str, **filters: Any) -> np.ndarray:
+    """
+    Get a mask based on the specified filters. E.g. measured_terminal_type=MeasuredTerminalType.source.
+
+    Args:
+        data: The input/update data set for all components
+        component: The component of interest
+        field: The field of interest
+        ref_components: The component or components in which we want to look for ids
+        **filters: One or more filters on the dataset. E.
+
+    Returns:
+        np.ndarray: the mask
+    """
+    values = data[component][field]
+    mask = np.ones(shape=values.shape, dtype=bool)
+    for filter_field, filter_value in filters.items():
+        mask = np.logical_and(mask, data[component][filter_field] == filter_value)
+
+    return mask

src/power_grid_model/validation/validation.py

@@ -48,7 +48,9 @@
     all_less_than,
     all_not_two_values_equal,
     all_not_two_values_zero,
+    all_supported_tap_control_side,
     all_unique,
+    all_valid_associated_enum_values,
     all_valid_clocks,
     all_valid_enum_values,
     all_valid_fault_phases,
@@ -833,8 +835,26 @@ def validate_transformer_tap_regulator(data: SingleDataset) -> List[ValidationEr
     errors = validate_regulator(data, "transformer_tap_regulator")
     errors += all_boolean(data, "transformer_tap_regulator", "status")
     errors += all_valid_enum_values(data, "transformer_tap_regulator", "control_side", [BranchSide, Branch3Side])
+    errors += all_valid_associated_enum_values(
+        data, "transformer_tap_regulator", "control_side", "regulated_object", ["transformer"], [BranchSide]
+    )
+    errors += all_valid_associated_enum_values(
+        data,
+        "transformer_tap_regulator",
+        "control_side",
+        "regulated_object",
+        ["three_winding_transformer"],
+        [Branch3Side],
+    )
     errors += all_greater_than_or_equal_to_zero(data, "transformer_tap_regulator", "u_set")
     errors += all_greater_than_zero(data, "transformer_tap_regulator", "u_band")
     errors += all_greater_than_or_equal_to_zero(data, "transformer_tap_regulator", "line_drop_compensation_r", 0.0)
     errors += all_greater_than_or_equal_to_zero(data, "transformer_tap_regulator", "line_drop_compensation_x", 0.0)
+    errors += all_supported_tap_control_side(
+        data,
+        "transformer_tap_regulator",
+        "control_side",
+        "regulated_object",
+        [("transformer", "tap_side"), ("three_winding_transformer", "tap_side")],
+    )
     return errors