diff --git a/src/pyrecest/filters/measurement_reliability.py b/src/pyrecest/filters/measurement_reliability.py index c4efaa577..d895e37da 100644 --- a/src/pyrecest/filters/measurement_reliability.py +++ b/src/pyrecest/filters/measurement_reliability.py @@ -27,9 +27,25 @@ class MeasurementReliabilitySelection: active_measurement_indices: list[int] +def _contains_masked_value(value: Any) -> bool: + """Return whether *value* contains genuinely masked NumPy entries.""" + + if np.ma.is_masked(value): + return True + if isinstance(value, np.ndarray): + if value.dtype != object: + return False + return any(_contains_masked_value(item) for item in value.reshape(-1)) + if isinstance(value, (list, tuple)): + return any(_contains_masked_value(item) for item in value) + return False + + def _normalize_integer_count( value: Any, name: str, *, minimum: int, message: str ) -> int: + if _contains_masked_value(value): + raise ValueError(message) try: value_array = np.asarray(value) except (TypeError, ValueError) as exc: @@ -197,6 +213,8 @@ def normalize_measurement_weights(measurement_weights, n_measurements: int): n_measurements = _normalize_nonnegative_integer(n_measurements, "n_measurements") if measurement_weights is None: return ones(n_measurements) + if _contains_masked_value(measurement_weights): + raise ValueError("measurement_weights must not contain masked values") weights = array(measurement_weights) _raise_if_not_real_numeric_weights(weights) @@ -242,6 +260,8 @@ def normalize_active_measurement_mask( n_measurements = _normalize_nonnegative_integer(n_measurements, "n_measurements") if active_measurement_mask is None: return [True] * n_measurements + if _contains_masked_value(active_measurement_mask): + raise ValueError("active_measurement_mask must not contain masked values") mask = array(active_measurement_mask) if not _has_boolean_dtype(mask): @@ -297,6 +317,8 @@ def normalize_measurement_noise_covariances( n_measurements = _normalize_nonnegative_integer(n_measurements, "n_measurements") measurement_dim = _normalize_positive_integer(measurement_dim, "measurement_dim") + if _contains_masked_value(measurement_noise): + raise ValueError(f"{name} must not contain masked values") noise = array(measurement_noise) empty_shape = (0, measurement_dim, measurement_dim) diff --git a/tests/filters/test_measurement_reliability_masked_inputs.py b/tests/filters/test_measurement_reliability_masked_inputs.py new file mode 100644 index 000000000..b744e3fed --- /dev/null +++ b/tests/filters/test_measurement_reliability_masked_inputs.py @@ -0,0 +1,107 @@ +import unittest + +import numpy as np +from pyrecest.backend import array, eye, to_numpy +from pyrecest.filters import ( + normalize_active_measurement_mask, + normalize_measurement_noise_covariances, + normalize_measurement_reliability, + normalize_measurement_weights, +) + + +def _as_covariance_matrix(value, dim, name): + matrix = array(value) + if matrix.ndim == 0: + matrix = matrix * eye(dim) + if matrix.shape != (dim, dim): + raise ValueError(f"{name} must have shape ({dim}, {dim})") + return matrix + + +class TestMeasurementReliabilityMaskedInputs(unittest.TestCase): + def test_masked_counts_are_rejected_before_payload_conversion(self): + masked_count = np.ma.array(2, mask=True) + with self.assertRaisesRegex(ValueError, "n_measurements"): + normalize_measurement_weights(None, masked_count) + with self.assertRaisesRegex(ValueError, "n_measurements"): + normalize_measurement_reliability(None, None, masked_count) + + masked_dim = np.ma.array(1, mask=True) + with self.assertRaisesRegex(ValueError, "measurement_dim"): + normalize_measurement_noise_covariances( + 1.0, + 1, + masked_dim, + as_covariance_matrix=_as_covariance_matrix, + ) + + def test_masked_measurement_weights_are_rejected(self): + invalid_weights = ( + np.ma.array(0.5, mask=True), + np.ma.array([1.0, 0.5], mask=[False, True]), + [1.0, np.ma.array(0.5, mask=True)], + ) + for weights in invalid_weights: + with self.subTest(weights=weights): + with self.assertRaisesRegex(ValueError, "masked values"): + normalize_measurement_weights(weights, 2) + + def test_masked_active_measurement_flags_are_rejected(self): + invalid_masks = ( + np.ma.array(True, mask=True), + np.ma.array([True, False], mask=[False, True]), + [True, np.ma.array(False, mask=True)], + ) + for active_mask in invalid_masks: + with self.subTest(active_mask=active_mask): + with self.assertRaisesRegex(ValueError, "masked values"): + normalize_active_measurement_mask(active_mask, 2) + + def test_masked_measurement_noise_is_rejected(self): + shared_noise = np.ma.array([[1.0]], mask=[[True]]) + with self.assertRaisesRegex(ValueError, "R must not contain masked values"): + normalize_measurement_noise_covariances( + shared_noise, + 1, + 1, + as_covariance_matrix=_as_covariance_matrix, + ) + + batched_noise = np.ma.array( + [[[1.0]], [[2.0]]], + mask=[[[False]], [[True]]], + ) + with self.assertRaisesRegex(ValueError, "noise must not contain masked values"): + normalize_measurement_noise_covariances( + batched_noise, + 2, + 1, + as_covariance_matrix=_as_covariance_matrix, + name="noise", + ) + + def test_fully_unmasked_masked_arrays_remain_supported(self): + weights = normalize_measurement_weights( + np.ma.array([1.0, 0.5], mask=False), + 2, + ) + np.testing.assert_allclose(to_numpy(weights), np.array([1.0, 0.5])) + + active_mask = normalize_active_measurement_mask( + np.ma.array([True, False], mask=False), + 2, + ) + self.assertEqual(active_mask, [True, False]) + + noise = normalize_measurement_noise_covariances( + np.ma.array([[2.0]], mask=False), + 2, + 1, + as_covariance_matrix=_as_covariance_matrix, + ) + np.testing.assert_allclose(to_numpy(noise), np.array([[[2.0]], [[2.0]]])) + + +if __name__ == "__main__": + unittest.main()