Catch duplicate ids within a single update scenario

docs/user_manual/data-validator.md

@@ -62,6 +62,13 @@ In such cases, the latter is leading when only running batch calculations.
 Running single calculations on an incomplete input data set is, of course, unsupported.
 ```
 
+```{note}
+Within a single scenario, every component may be updated at most once: an update is applied by id lookup, so a
+duplicate id would silently result in only the last of the duplicated records being applied. `validate_batch_data`
+reports such duplicates as a `NotUniqueError` for the scenario in question. Updating the same component again in
+another scenario of the same batch is of course perfectly valid.
+```
+
 Validating a Cartesian product of datasets used in PGM's `update_data` via providing it with `list[BatchDataset]`is
 done by validating each individual `BatchDataset` that conforms the Cartesian product.
 

src/power_grid_model/validation/_rules.py

@@ -746,6 +746,31 @@ def ids_valid_in_update_data_set(
     return []
 
 
+def ids_unique_in_update_data_set(update_data: SingleDataset, component: ComponentType) -> list[NotUniqueError]:
+    """
+    Check that for all records of a particular type of component, the ids are unique within a single scenario.
+
+    An id may be updated at most once per scenario: the update of a component is applied by an id lookup, so a
+    duplicate id would silently result in only the last of the duplicated records being applied. Note that the same
+    id may of course be updated again in any of the other scenarios of a batch.
+
+    Args:
+        update_data: A single scenario of the update data set for all components
+        component: The component of interest
+
+    Returns:
+        A list containing zero or one NotUniqueError, listing all ids that occur more than once within the scenario.
+    """
+    ids = update_data[component][AttributeType.id]
+    # Unset ids are not actual ids: they either mark an update-by-position scenario, or they are already reported
+    # by ids_valid_in_update_data_set. Either way they are not duplicates and are excluded from this check.
+    ids = ids[~is_nan_or_default(ids)]
+    _, inverse, counts = np.unique(ids, return_inverse=True, return_counts=True)
+    if np.any(counts != 1):
+        return [NotUniqueError(component, AttributeType.id, ids[(counts != 1)[inverse]].flatten().tolist())]
+    return []
+
+
 def all_finite(
     data: SingleDataset, exceptions: dict[ComponentType, list[AttributeType]] | None = None
 ) -> list[InfinityError]:

src/power_grid_model/validation/_validation.py

@@ -65,6 +65,7 @@
     all_valid_fault_phases as _all_valid_fault_phases,
     all_valid_ids as _all_valid_ids,
     any_voltage_angle_measurement_if_global_current_measurement as _any_voltage_angle_measurement_if_global_current_measurement,  # noqa: E501
+    ids_unique_in_update_data_set as _ids_unique_in_update_data_set,
     ids_valid_in_update_data_set as _ids_valid_in_update_data_set,
     no_strict_subset_missing as _no_strict_subset_missing,
     none_missing as _none_missing,
@@ -77,6 +78,7 @@
     InvalidVoltageRegulationError,
     MissingValueError,
     MultiComponentNotUniqueError,
+    NotUniqueError,
     ValidationError,
 )
 from power_grid_model.validation.utils import _update_input_data
@@ -135,7 +137,8 @@ def validate_batch_data(
 
     For each batch the update data is validated:
         3. Is the update data structure correct? (checking data types and numpy array shapes)
-        4. Are all update ID's valid? (checking object identifiers across update and input data)
+        4. Are all update ID's valid? (checking object identifiers across update and input data, and checking that
+           each object is updated at most once within the scenario)
 
     Then (for each batch independently) the input dataset is updated with the batch's update data and validated:
         5. Are all required values provided? (checking NaNs)
@@ -178,7 +181,10 @@ def validate_batch_data(
             for component, component_data in row_update_data.items()
             if _get_comp_size(component_data) > 0
         }
-        id_errors: list[IdNotInDatasetError | InvalidIdError] = validate_ids(row_update_data, input_data_copy)
+        id_errors: list[IdNotInDatasetError | InvalidIdError | NotUniqueError] = list(
+            validate_ids(row_update_data, input_data_copy)
+        )
+        id_errors += validate_unique_ids_in_scenario(row_update_data)
 
         batch_errors = input_errors + id_errors
 
@@ -273,6 +279,28 @@ def validate_ids(update_data: SingleDataset, input_data: SingleDataset) -> list[
     return list(chain(*errors))
 
 
+def validate_unique_ids_in_scenario(update_data: SingleDataset) -> list[NotUniqueError]:
+    """
+    Checks if all ids of the components in a single update scenario are unique within that scenario.
+
+    An update is applied by id lookup, so a duplicate id within a scenario would silently result in only the last of
+    the duplicated records being applied. Reusing the same id in another scenario of the batch is perfectly valid and
+    is not reported here.
+
+    This function should be called for every update dataset in a batch set
+
+    Args:
+        update_data: A single update dataset
+
+    Returns:
+        An empty list if all ids within the scenario are unique, or a list of NotUniqueErrors for all update
+        components that contain duplicate ids
+
+    """
+    errors = (_ids_unique_in_update_data_set(update_data, component) for component in update_data)
+    return list(chain(*errors))
+
+
 def validate_required_values(  # noqa: PLR0915
     data: SingleDataset, calculation_type: CalculationType | None = None, symmetric: bool = True
 ) -> list[MissingValueError]:

tests/unit/test_optional_ids.py

@@ -74,7 +74,9 @@ def input_data(request):
 
 @pytest.fixture
 def update_sym_load_r():
-    sym_load = initialize_array(DatasetType.update, CT.sym_load, (2, 2))
+    # one object per scenario, mirroring update_sym_load_c; a (2, 2) array would broadcast the values and
+    # thereby update the same id twice per scenario, which is not a valid update
+    sym_load = initialize_array(DatasetType.update, CT.sym_load, (2, 1))
     sym_load[AT.id] = [[4], [7]]
     sym_load[AT.p_specified] = [[30e6], [15e6]]
     return sym_load

tests/unit/validation/test_batch_validation.py

@@ -15,6 +15,7 @@
     MultiComponentNotUniqueError,
     NotBetweenOrAtError,
     NotBooleanError,
+    NotUniqueError,
 )
 
 
@@ -170,6 +171,50 @@ def test_validate_batch_data_update_error(input_data, batch_data):
     assert errors[2] == [NotBooleanError(CT.line, AT.from_status, [5, 7])]
 
 
+def test_validate_batch_data_duplicate_ids_in_scenario(input_data, batch_data):
+    """An object may be updated at most once per scenario; a duplicate id silently drops all but the last update."""
+    batch_data[CT.line][AT.id][0] = [5, 5]
+    errors = validate_batch_data(input_data, batch_data)
+
+    assert errors is not None
+    # only the offending scenario is reported; id 5 is also updated in scenario 2, which stays valid
+    assert list(errors) == [0]
+    assert errors[0] == [NotUniqueError(CT.line, AT.id, [5, 5])]
+
+    with pytest.raises(ValidationException):
+        assert_valid_batch_data(input_data, batch_data)
+
+
+def test_validate_batch_data_duplicate_ids_in_multiple_scenarios(input_data, batch_data):
+    """Every scenario is checked independently."""
+    batch_data[CT.line][AT.id][0] = [5, 5]
+    batch_data[CT.asym_load][AT.id][2] = [10, 10]
+    errors = validate_batch_data(input_data, batch_data)
+
+    assert errors is not None
+    assert sorted(errors) == [0, 2]
+    assert errors[0] == [NotUniqueError(CT.line, AT.id, [5, 5])]
+    assert errors[2] == [NotUniqueError(CT.asym_load, AT.id, [10, 10])]
+
+
+def test_validate_batch_data_same_id_in_different_scenarios(input_data, batch_data):
+    """Updating the same objects again in another scenario is normal and must remain valid."""
+    batch_data[CT.line][AT.id][:] = [[5, 6], [5, 6], [5, 6]]
+    batch_data[CT.asym_load][AT.id][:] = [[9, 10], [9, 10], [9, 10]]
+
+    assert validate_batch_data(input_data, batch_data) is None
+    assert_valid_batch_data(input_data, batch_data)
+
+
+def test_validate_batch_data_unset_ids_are_not_duplicates(input_data):
+    """Unset ids mean 'update by position'; the repeated 'not available' value is not a duplicate id."""
+    line = initialize_array(DatasetType.update, CT.line, (2, 4))
+    line[AT.from_status] = 1
+
+    assert np.all(line[AT.id] == np.iinfo(line[AT.id].dtype).min)
+    assert validate_batch_data(input_data, {CT.line: line}) is None
+
+
 @pytest.mark.parametrize("columnar_input", [False, True])
 @pytest.mark.parametrize("columnar_update", [False, True])
 @pytest.mark.parametrize(

tests/unit/validation/test_validation_functions.py

@@ -33,6 +33,7 @@
     validate_input_data,
     validate_required_values,
     validate_unique_ids_across_components,
+    validate_unique_ids_in_scenario,
     validate_values,
 )
 from power_grid_model.validation.errors import (
@@ -205,6 +206,52 @@ def test_validate_ids():
     assert IdNotInDatasetError(CT.sym_load, [7], DatasetType.update) in invalid_ids
 
 
+def test_validate_unique_ids_in_scenario():
+    source_update = initialize_array(DatasetType.update, CT.source, 4)
+    source_update[AT.id] = [1, 2, 1, 1]
+    source_update[AT.u_ref] = [1.0, 2.0, 3.0, 4.0]
+
+    sym_load_update = initialize_array(DatasetType.update, CT.sym_load, 3)
+    sym_load_update[AT.id] = [4, 5, 6]
+    sym_load_update[AT.p_specified] = [4.0, 5.0, 6.0]
+
+    not_unique_ids = validate_unique_ids_in_scenario({CT.source: source_update, CT.sym_load: sym_load_update})
+
+    # the id is repeated as often as it occurs, to maintain object counts
+    assert not_unique_ids == [NotUniqueError(CT.source, AT.id, [1, 1, 1])]
+
+
+def test_validate_unique_ids_in_scenario_all_unique():
+    source_update = initialize_array(DatasetType.update, CT.source, 3)
+    source_update[AT.id] = [1, 2, 3]
+    source_update[AT.u_ref] = [1.0, 2.0, 3.0]
+
+    assert validate_unique_ids_in_scenario({CT.source: source_update}) == []
+
+
+def test_validate_unique_ids_in_scenario_optional_ids():
+    """Unset ids indicate an update by position; the repeated 'not available' value is not a duplicate id."""
+    source_update = initialize_array(DatasetType.update, CT.source, 3)
+    source_update[AT.u_ref] = [1.0, 2.0, 3.0]
+
+    assert np.all(source_update[AT.id] == NaN)
+    assert validate_unique_ids_in_scenario({CT.source: source_update}) == []
+
+
+def test_validate_unique_ids_in_scenario_columnar():
+    source_update = initialize_array(DatasetType.update, CT.source, 3)
+    source_update[AT.id] = [1, 1, 2]
+    source_update[AT.u_ref] = [1.0, 2.0, 3.0]
+
+    update_data_col = compatibility_convert_row_columnar_dataset(
+        data={CT.source: source_update},
+        data_filter=ComponentAttributeFilterOptions.relevant,
+        dataset_type=DatasetType.update,
+    )
+
+    assert validate_unique_ids_in_scenario(update_data_col) == [NotUniqueError(CT.source, AT.id, [1, 1])]
+
+
 @pytest.mark.parametrize(
     "calculation_type",
     [