diff --git a/eitprocessing/datahandling/sequence.py b/eitprocessing/datahandling/sequence.py index 6358d8f5f..e52709721 100644 --- a/eitprocessing/datahandling/sequence.py +++ b/eitprocessing/datahandling/sequence.py @@ -186,6 +186,7 @@ def data(self) -> _DataAccess: `sequence.data.add(obj)`). Other dict-like behaviour is also supported: + - `label in sequence.data` to check whether an object with a label exists; - `del sequence.data[label]` to remove an object from the sequence based on the label; - `for label in sequence.data` to iterate over the labels; diff --git a/eitprocessing/features/pixel_breath.py b/eitprocessing/features/pixel_breath.py index 213118c6a..a7a6a65a9 100644 --- a/eitprocessing/features/pixel_breath.py +++ b/eitprocessing/features/pixel_breath.py @@ -1,8 +1,11 @@ import itertools +import warnings from collections.abc import Callable -from dataclasses import dataclass, field +from dataclasses import InitVar, dataclass, field +from typing import Final, Literal import numpy as np +from scipy import signal from eitprocessing.datahandling.breath import Breath from eitprocessing.datahandling.continuousdata import ContinuousData @@ -11,8 +14,17 @@ from eitprocessing.datahandling.sequence import Sequence from eitprocessing.features.breath_detection import BreathDetection +_SENTINEL_BREATH_DETECTION: Final = BreathDetection() +MAX_XCORR_LAG = 0.75 -@dataclass + +def _return_sentinel_breath_detection() -> BreathDetection: + # Returns a sential of a BreathDetection, which only exists to signal that the default value for breath_detection + # was used. + return _SENTINEL_BREATH_DETECTION + + +@dataclass(kw_only=True) class PixelBreath: """Algorithm for detecting timing of pixel breaths in pixel impedance data. @@ -21,6 +33,16 @@ class PixelBreath: of inspiration and expiration. These points are then used to find the start/end of pixel inspiration/expiration in pixel impedance data. + Some pixel breaths may be phase shifted (inflation starts and ends later compared to others, e.g., due to pendelluft + or late airway opening). Other pixel breaths may have a negative amplitude (impedance decreases during inspiration, + e.g., due to pleural effusion or reconstruction artifacts). It is not always possible to determine whether a pixel + is out of phase or has a negative amplitude. PixelBreath has three different phase correction modes. In 'negative + amplitude' mode (default), pixels that have a decrease in amplitude between the start and end of globally defined + inspiration, will have a negative amplitude and smaller phase shift. In 'phase shift' mode, all pixel breaths will + have positive amplitudes, but can have large phase shifts. In 'none'/`None` mode, all pixels are assumed to be + within rouglhy -90 to 90 degrees of phase. Note that the 'none' mode can lead to unexpected results, such as + ultra-short (down to 2 frames) or very long breaths. + Example: ``` >>> pi = PixelBreath() @@ -30,20 +52,30 @@ class PixelBreath: ``` Args: - breath_detection_kwargs (dict): A dictionary of keyword arguments for breath detection. - The available keyword arguments are: - minimum_duration: minimum expected duration of breaths, defaults to 2/3 of a second - averaging_window_duration: duration of window used for averaging the data, defaults to 15 seconds - averaging_window_function: function used to create a window for averaging the data, defaults to np.blackman - amplitude_cutoff_fraction: fraction of the median amplitude below which breaths are removed, defaults to 0.25 - invalid_data_removal_window_length: window around invalid data in which breaths are removed, defaults to 0.5 - invalid_data_removal_percentile: the nth percentile of values used to remove outliers, defaults to 5 - invalid_data_removal_multiplier: the multiplier used to remove outliers, defaults to 4 + breath_detection (BreathDetection): BreathDetection object to use for detecting breaths. + phase_correction_mode: How to resolve pixels that are out-of-phase. Defaults to "negative amplitude". """ - breath_detection_kwargs: dict = field(default_factory=dict) - - def find_pixel_breaths( + breath_detection: BreathDetection = field(default_factory=_return_sentinel_breath_detection) + breath_detection_kwargs: InitVar[dict | None] = None + phase_correction_mode: Literal["negative amplitude", "phase shift", "none"] | None = "negative amplitude" + + def __post_init__(self, breath_detection_kwargs: dict | None): + if breath_detection_kwargs is not None: + if self.breath_detection is not _SENTINEL_BREATH_DETECTION: + msg = ( + "`breath_detection_kwargs` is deprecated, and can't be used at the same time as `breath_detection`." + ) + raise TypeError(msg) + + self.breath_detection = BreathDetection(**breath_detection_kwargs) + warnings.warn( + "`breath_detection_kwargs` is deprecated and will be removed soon. " + "Replace with `breath_detection=BreathDetection(**breath_detection_kwargs)`.", + DeprecationWarning, + ) + + def find_pixel_breaths( # noqa: C901, PLR0912, PLR0915 self, eit_data: EITData, continuous_data: ContinuousData, @@ -53,36 +85,29 @@ def find_pixel_breaths( ) -> IntervalData: """Find pixel breaths in the data. - This method finds the pixel start/end of inspiration/expiration - based on the start/end of inspiration/expiration as detected - in the continuous data. - - If pixel impedance is in phase (within 180 degrees) with the continuous data, - the start of breath of that pixel is defined as the local minimum between - two end-inspiratory points in the continuous signal. - The end of expiration of that pixel is defined as the local minimum between two - consecutive end-inspiratory points in the continuous data. - The end of inspiration of that pixel is defined as the local maximum between - the start of inspiration and end of expiration of that pixel. - - If pixel impedance is out of phase with the continuous signal, - the start of inspiration of that pixel is defined as the local maximum between - two end-inspiration points in the continuous data. - The end of expiration of that pixel is defined as the local maximum between two - consecutive end-inspiratory points in the continuous data. - The end of inspiration of that pixel is defined as the local minimum between - the start of inspiration and end of expiration of that pixel. - - Pixel breaths are constructed as a valley-peak-valley combination, - representing the start of inspiration, the end of inspiration/start of - expiration, and end of expiration. + This method finds the pixel start/end of inspiration/expiration based on the start/end of inspiration/expiration + as detected in the continuous data. + + For most pixels, the start of a breath (start inspiration) is the valley between the middles (start of + expiration) of the globally defined breaths on either side. The end of a pixel breath is the start of the next + pixel breath. The middle of the pixel breath is the peak between the start and end of the pixel breath. + + If the pixel is out of phase or has negative amplitude, the definition of the breath depends on the phase + correction mode. In 'negative amplitude' mode, the start of a breath is the peak between the middles of the + globally defined breaths on either side, while the middle of the pixel breath is the valley of the start and end + of the pixel breath. In 'phase shift' mode, first the phase shift between the pixel impedance and global + impedance is determined as the highest crosscorrelation between the signals near a phase shift of 0. The start + of breath is the valley between the phase shifted middles of the globally defined breaths on either side. + + Pixel breaths are constructed as a valley-peak-valley combination, representing the start of inspiration, the + end of inspiration/start of expiration, and end of expiration. Args: - eit_data: EITData to apply the algorithm to - continuous_data: ContinuousData to use for global breath detection + eit_data: EITData to apply the algorithm to. + continuous_data: ContinuousData to use for global breath detection. result_label: label of the returned IntervalData object, defaults to `'pixel_breaths'`. - sequence: optional, Sequence that contains the object to detect pixel breaths in, - and/or to store the result in. + sequence: optional, Sequence that contains the object to detect pixel breaths in, and/or to store the result + in. store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided. Returns: @@ -103,9 +128,7 @@ def find_pixel_breaths( msg = "To store the result a Sequence dataclass must be provided." raise ValueError(msg) - bd_kwargs = self.breath_detection_kwargs.copy() - breath_detection = BreathDetection(**bd_kwargs) - continuous_breaths = breath_detection.find_breaths(continuous_data) + continuous_breaths = self.breath_detection.find_breaths(continuous_data) indices_breath_middles = np.searchsorted( eit_data.time, @@ -144,28 +167,73 @@ def find_pixel_breaths( pixel_breaths = np.full((len(continuous_breaths), n_rows, n_cols), None) + lags = signal.correlation_lags(len(continuous_data), len(continuous_data), mode="same") + + match self.phase_correction_mode: + case "negative amplitude": + allow_negative_amplitude = True + correct_for_phase_shift = None + case "phase shift": + allow_negative_amplitude = False + correct_for_phase_shift = True + case "none" | None: + allow_negative_amplitude = False + correct_for_phase_shift = False + case _: + msg = f"Unknown phase correction mode ({self.phase_correction_mode})." + raise ValueError(msg) + for row, col in itertools.product(range(n_rows), range(n_cols)): mean_tiv = mean_tiv_pixel[row, col] - if np.any(pixel_impedance[:, row, col] > 0): - if mean_tiv < 0: - start_func, middle_func = np.argmax, np.argmin - else: - start_func, middle_func = np.argmin, np.argmax - - outsides = self._find_extreme_indices(pixel_impedance, indices_breath_middles, row, col, start_func) - starts = outsides[:-1] - ends = outsides[1:] - middles = self._find_extreme_indices(pixel_impedance, outsides, row, col, middle_func) - # TODO discuss; this block of code is implemented to prevent noisy pixels from breaking the code. - # Quick solve is to make entire breath object None if any breath in a pixel does not have - # consecutive start, middle and end. - # However, this might cause problems elsewhere. - if (starts >= middles).any() or (middles >= ends).any(): - pixel_breath = None + if np.std(pixel_impedance[:, row, col]) == 0: + # pixel has no amplitude + continue + + if allow_negative_amplitude and mean_tiv < 0: + start_func, middle_func = np.argmax, np.argmin + lagged_indices_breath_middles = indices_breath_middles + else: + start_func, middle_func = np.argmin, np.argmax + + cd = np.copy(continuous_data.values) + cd -= np.nanmean(cd) + pi = np.copy(pixel_impedance[:, row, col]) + if not np.all(np.isnan(pi)): + pi -= np.nanmean(pixel_impedance[:, row, col]) + + if correct_for_phase_shift: + # search for maximum cross correlation within MAX_XCORR_LAG times the average + # duration of a breath + xcorr = signal.correlate(cd, pi, mode="same") + max_lag = MAX_XCORR_LAG * np.mean(np.diff(indices_breath_middles)) + lag_range = (lags > -max_lag) & (lags < max_lag) + # TODO: if this does not work, implement robust peak detection + + # positive lag: pixel inflates later than summed + lag = lags[lag_range][np.argmax(xcorr[lag_range])] + + # shift search area + lagged_indices_breath_middles = indices_breath_middles - lag + lagged_indices_breath_middles = lagged_indices_breath_middles[ + (lagged_indices_breath_middles >= 0) & (lagged_indices_breath_middles < len(cd)) + ] else: - pixel_breath = self._construct_breaths(starts, middles, ends, time) - pixel_breaths[:, row, col] = pixel_breath + lagged_indices_breath_middles = indices_breath_middles + + outsides = self._find_extreme_indices(pixel_impedance, lagged_indices_breath_middles, row, col, start_func) + starts = outsides[:-1] + ends = outsides[1:] + middles = self._find_extreme_indices(pixel_impedance, outsides, row, col, middle_func) + # TODO discuss; this block of code is implemented to prevent noisy pixels from breaking the code. + # Quick solve is to make entire breath object None if any breath in a pixel does not have + # consecutive start, middle and end. + # However, this might cause problems elsewhere. + if (starts >= middles).any() or (middles >= ends).any(): + pixel_breath = None + else: + pixel_breath = self._construct_breaths(starts, middles, ends, time) + pixel_breaths[:, row, col] = pixel_breath intervals = [(breath.start_time, breath.end_time) for breath in continuous_breaths.values] @@ -178,7 +246,6 @@ def find_pixel_breaths( values=list( pixel_breaths, ), ## TODO: change back to pixel_breaths array when IntervalData works with 3D array - parameters=self.breath_detection_kwargs, derived_from=[eit_data], ) if store: @@ -186,7 +253,7 @@ def find_pixel_breaths( return pixel_breaths_container - def _construct_breaths(self, start: list, middle: list, end: list, time: np.ndarray) -> list: + def _construct_breaths(self, start: list[int], middle: list[int], end: list[int], time: np.ndarray) -> list: breaths = [Breath(time[s], time[m], time[e]) for s, m, e in zip(start, middle, end, strict=True)] # First and last breath are not detected by definition (need two breaths to find one breath) return [None, *breaths, None] @@ -222,5 +289,5 @@ def _find_extreme_indices( are located for each time segment. """ return np.array( - [function(pixel_impedance[times[i] : times[i + 1], row, col]) + times[i] for i in range(len(times) - 1)], + [function(pixel_impedance[t1:t2, row, col]) + t1 for t1, t2 in itertools.pairwise(times)], ) diff --git a/eitprocessing/parameters/eeli.py b/eitprocessing/parameters/eeli.py index 6622760d2..abc6dbd99 100644 --- a/eitprocessing/parameters/eeli.py +++ b/eitprocessing/parameters/eeli.py @@ -1,5 +1,6 @@ -from dataclasses import dataclass, field -from typing import Literal, get_args +import warnings +from dataclasses import InitVar, dataclass, field +from typing import Final, Literal, get_args import numpy as np @@ -10,15 +11,38 @@ from eitprocessing.features.breath_detection import BreathDetection from eitprocessing.parameters import ParameterCalculation +_SENTINEL_BREATH_DETECTION: Final = BreathDetection() + + +def _sentinel_breath_detection() -> BreathDetection: + # Returns a sential of a BreathDetection, which only exists to signal that the default value for breath_detection + # was used. + return _SENTINEL_BREATH_DETECTION + @dataclass class EELI(ParameterCalculation): """Compute the end-expiratory lung impedance (EELI) per breath.""" + breath_detection: BreathDetection = field(default_factory=_sentinel_breath_detection) method: Literal["breath_detection"] = "breath_detection" - breath_detection_kwargs: dict = field(default_factory=dict) + breath_detection_kwargs: InitVar[dict | None] = None + + def __post_init__(self, breath_detection_kwargs: dict | None): + if breath_detection_kwargs is not None: + if self.breath_detection is not _SENTINEL_BREATH_DETECTION: + msg = ( + "`breath_detection_kwargs` is deprecated, and can't be used at the same time as `breath_detection`." + ) + raise TypeError(msg) + + self.breath_detection = BreathDetection(**breath_detection_kwargs) + warnings.warn( + "`breath_detection_kwargs` is deprecated and will be removed soon. " + "Replace with `breath_detection=BreathDetection(**breath_detection_kwargs)`.", + DeprecationWarning, + ) - def __post_init__(self): _methods = get_args(EELI.__dataclass_fields__["method"].type) if self.method not in _methods: msg = f"Method {self.method} is not valid. Use any of {', '.join(_methods)}" @@ -66,9 +90,7 @@ def compute_parameter( check_category(continuous_data, "impedance", raise_=True) - bd_kwargs = self.breath_detection_kwargs.copy() - breath_detection = BreathDetection(**bd_kwargs) - breaths = breath_detection.find_breaths(continuous_data) + breaths = self.breath_detection.find_breaths(continuous_data) if not len(breaths): time = np.array([], dtype=float) diff --git a/eitprocessing/parameters/tidal_impedance_variation.py b/eitprocessing/parameters/tidal_impedance_variation.py index 09119d29a..315df03b8 100644 --- a/eitprocessing/parameters/tidal_impedance_variation.py +++ b/eitprocessing/parameters/tidal_impedance_variation.py @@ -1,7 +1,9 @@ import itertools -from dataclasses import dataclass, field +import sys +import warnings +from dataclasses import InitVar, dataclass, field from functools import singledispatchmethod -from typing import Literal, NoReturn +from typing import Final, Literal, NoReturn import numpy as np @@ -15,19 +17,57 @@ from eitprocessing.features.pixel_breath import PixelBreath from eitprocessing.parameters import ParameterCalculation +_SENTINEL_PIXEL_BREATH: Final = PixelBreath() +_SENTINEL_BREATH_DETECTION: Final = BreathDetection() + + +def _sentinel_pixel_breath() -> PixelBreath: + # Returns a sential of a PixelBreath, which only exists to signal that the default value for pixel_breath was used. + return _SENTINEL_PIXEL_BREATH + + +def _sentinel_breath_detection() -> BreathDetection: + # Returns a sential of a BreathDetection, which only exists to signal that the default value for breath_detection + # was used. + return _SENTINEL_BREATH_DETECTION + @dataclass class TIV(ParameterCalculation): """Compute the tidal impedance variation (TIV) per breath.""" method: Literal["extremes"] = "extremes" - breath_detection_kwargs: dict = field(default_factory=dict) + breath_detection: BreathDetection = field(default_factory=_sentinel_breath_detection) + breath_detection_kwargs: InitVar[dict | None] = None - def __post_init__(self) -> None: + # The default is a sentinal that will be replaced in __post_init__ + pixel_breath: PixelBreath = field(default_factory=_sentinel_pixel_breath) + + def __post_init__(self, breath_detection_kwargs: dict | None) -> None: if self.method != "extremes": msg = f"Method {self.method} is not implemented. The method must be 'extremes'." raise NotImplementedError(msg) + if breath_detection_kwargs is not None: + if self.breath_detection is not _SENTINEL_BREATH_DETECTION: + msg = ( + "`breath_detection_kwargs` is deprecated, and can't be used at the same time as `breath_detection`." + ) + raise TypeError(msg) + + self.breath_detection = BreathDetection(**breath_detection_kwargs) + warnings.warn( + "`breath_detection_kwargs` is deprecated and will be removed soon. " + "Replace with `breath_detection=BreathDetection(**breath_detection_kwargs)`.", + DeprecationWarning, + ) + + if self.pixel_breath is _SENTINEL_PIXEL_BREATH: + # If no value was provided at initialization, PixelBreath should use the same BreathDetection object as TIV. + # However, a default factory cannot be used because it can't access self.breath_detection. The sentinal + # object is replaced here (only if pixel_breath was not provided) with the correct BreathDetection object. + self.pixel_breath = PixelBreath(breath_detection=self.breath_detection) + @singledispatchmethod def compute_parameter( self, @@ -101,7 +141,6 @@ def compute_continuous_parameter( category="impedance difference", time=[breath.middle_time for breath in breaths.values if breath is not None], description="Tidal impedance variation determined on continuous data", - parameters=self.breath_detection_kwargs, derived_from=[continuous_data], values=tiv_values, ) @@ -213,7 +252,6 @@ def compute_pixel_parameter( category="impedance difference", time=list(all_pixels_breath_timings), description="Tidal impedance variation determined on pixel impedance", - parameters=self.breath_detection_kwargs, derived_from=[eit_data], values=list(all_pixels_tiv_values.astype(float)), ) @@ -224,9 +262,7 @@ def compute_pixel_parameter( return tiv_container def _detect_breaths(self, data: ContinuousData) -> IntervalData: - bd_kwargs = self.breath_detection_kwargs.copy() - breath_detection = BreathDetection(**bd_kwargs) - return breath_detection.find_breaths(data) + return self.breath_detection.find_breaths(data) def _detect_pixel_breaths( self, @@ -235,9 +271,7 @@ def _detect_pixel_breaths( sequence: Sequence, store: bool, ) -> IntervalData: - bd_kwargs = self.breath_detection_kwargs.copy() - pi = PixelBreath(breath_detection_kwargs=bd_kwargs) - return pi.find_pixel_breaths( + return self.pixel_breath.find_pixel_breaths( eit_data, continuous_data, result_label="pixel breaths", @@ -274,6 +308,13 @@ def _calculate_tiv_values( mean_outer_values = data[[start_indices, end_indices]].mean(axis=0) end_inspiratory_values = data[middle_indices] tiv_values = end_inspiratory_values - mean_outer_values + else: + msg = f"`tiv_method` ({tiv_method}) not valid." + exc = ValueError(msg) + if sys.version_info >= (3, 11): + exc.add_note("Valid value for `tiv_method` are 'inspiratory', 'expiratory' and 'mean'.") + raise exc + if tiv_timing == "pixel": tiv_values = [None, *tiv_values, None] diff --git a/tests/test_eeli.py b/tests/test_eeli.py index 0c7e85f75..73f3e2b6b 100644 --- a/tests/test_eeli.py +++ b/tests/test_eeli.py @@ -86,13 +86,25 @@ def test_eeli_values(repeat_n: int): # noqa: ARG001 values=data, sample_frequency=sample_frequency, ) - eeli = EELI(breath_detection_kwargs={"minimum_duration": 0}) + with pytest.warns(DeprecationWarning): + eeli = EELI(breath_detection_kwargs={"minimum_duration": 0}) eeli_values = eeli.compute_parameter(cd).values assert len(eeli_values) == expected_n_breaths assert np.array_equal(eeli_values, valley_values[1:]) +def test_bd_init(): + with pytest.warns(DeprecationWarning): + assert EELI(breath_detection_kwargs={"minimum_duration": 0}) == EELI( + breath_detection=BreathDetection(minimum_duration=0) + ) + with pytest.warns(DeprecationWarning): + EELI(breath_detection_kwargs={"minimum_duration": 0}) + with pytest.raises(TypeError): + EELI(breath_detection_kwargs={"minimum_duration": 0}, breath_detection=BreathDetection(minimum_duration=0)) + + def test_with_data(draeger1: Sequence, pytestconfig: pytest.Config): if pytestconfig.getoption("--cov"): pytest.skip("Skip with option '--cov' so other tests can cover 100%.") diff --git a/tests/test_parameter_tiv.py b/tests/test_parameter_tiv.py index d6183fb5d..07c1b13b2 100644 --- a/tests/test_parameter_tiv.py +++ b/tests/test_parameter_tiv.py @@ -13,6 +13,7 @@ from eitprocessing.datahandling.sequence import Sequence from eitprocessing.datahandling.sparsedata import SparseData from eitprocessing.parameters.tidal_impedance_variation import TIV +from tests.test_breath_detection import BreathDetection environment = Path( os.environ.get( @@ -144,7 +145,20 @@ def test_tiv_initialization(): """Test that TIV initializes correctly with default parameters.""" tiv = TIV() assert tiv.method == "extremes" - assert tiv.breath_detection_kwargs == {} + assert tiv.breath_detection == BreathDetection() + + +def test_deprecated(): + with pytest.warns(DeprecationWarning): + _ = TIV(breath_detection_kwargs={}) + + with pytest.raises(TypeError): + _ = TIV(breath_detection=BreathDetection(), breath_detection_kwargs={}) + + bd_kwargs = {"minimum_duration": 10, "averaging_window_duration": 100.0} + + with pytest.warns(DeprecationWarning): + assert TIV(breath_detection_kwargs=bd_kwargs).breath_detection == BreathDetection(**bd_kwargs) def test_compute_parameter_type_error(): @@ -479,16 +493,13 @@ def test_with_data(draeger1: Sequence, timpel1: Sequence, pytestconfig: pytest.C # Iterate over both sequences (draeger1 and timpel1) for sequence in draeger1, timpel1: - # Select a subset of the sequence for testing (first 500 samples) - ssequence = sequence[0:500] - # Initialize the TIV object tiv = TIV() - eit_data = ssequence.eit_data["raw"] - cd = ssequence.continuous_data["global_impedance_(raw)"] + eit_data = sequence.eit_data["raw"] + cd = sequence.continuous_data["global_impedance_(raw)"] result_continuous = tiv.compute_continuous_parameter(cd, tiv_method="inspiratory") - result_pixel = tiv.compute_pixel_parameter(eit_data, cd, ssequence) + result_pixel = tiv.compute_pixel_parameter(eit_data, cd, sequence) arr_result_continuous = np.stack(result_continuous.values) arr_result_pixel = np.stack(result_pixel.values) @@ -508,23 +519,37 @@ def test_with_data(draeger1: Sequence, timpel1: Sequence, pytestconfig: pytest.C [ ({"amplitude_cutoff_fraction": 0.3, "minimum_duration": 5.0}, ValueError), # too long duration ({"amplitude_cutoff_fraction": 2, "minimum_duration": 5.0}, ValueError), # too high amplitude cutoff - ({"minimum_amplitude": 2, "minimum_duration": 5.0}, TypeError), # unexpected keyword minimum_amplitude ], ) def test_detect_pixel_breaths_with_invalid_bd_kwargs( bd_kwargs: dict, - expected_error: ValueError, + expected_error: type[Exception], mock_eit_data: EITData, mock_continuous_data: ContinuousData, mock_sequence: Sequence, ): """Test detect_pixel_breaths with invalid bd_kwargs that raise errors.""" - tiv = TIV(breath_detection_kwargs=bd_kwargs) + tiv = TIV(breath_detection=BreathDetection(**bd_kwargs)) with pytest.raises(expected_error): tiv._detect_pixel_breaths(mock_eit_data, mock_continuous_data, mock_sequence, store=False) +@pytest.mark.parametrize( + ("bd_kwargs", "expected_error"), + [ + ({"minimum_amplitude": 2, "minimum_duration": 5.0}, TypeError), # unexpected keyword minimum_amplitude + ], +) +def test_detect_pixel_breaths_with_invalid_bd_kwargs_( + bd_kwargs: dict, + expected_error: type[Exception], +): + """Test detect_pixel_breaths with invalid bd_kwargs that raise errors.""" + with pytest.raises(expected_error): + _ = TIV(breath_detection=BreathDetection(**bd_kwargs)) + + @pytest.mark.parametrize( ("bd_kwargs"), [ @@ -540,7 +565,7 @@ def test_detect_pixel_breaths_with_valid_bd_kwargs( mock_sequence: Sequence, ): """Test detect_pixel_breaths with valid bd_kwargs that return expected results.""" - tiv = TIV(breath_detection_kwargs=bd_kwargs) + tiv = TIV(breath_detection=BreathDetection(**bd_kwargs)) result = tiv._detect_pixel_breaths(mock_eit_data, mock_continuous_data, mock_sequence, store=False) test_result = np.stack(result.values) diff --git a/tests/test_pixel_breath.py b/tests/test_pixel_breath.py index ac3aebca3..0fe3d2855 100644 --- a/tests/test_pixel_breath.py +++ b/tests/test_pixel_breath.py @@ -15,6 +15,7 @@ from eitprocessing.datahandling.sequence import Sequence from eitprocessing.datahandling.sparsedata import SparseData from eitprocessing.features.pixel_breath import PixelBreath +from tests.test_breath_detection import BreathDetection environment = Path( os.environ.get( @@ -167,7 +168,7 @@ def none_sequence(): def mock_compute_pixel_parameter(mean: int): - def _mock(*_args, **_kwargs) -> np.ndarray: + def _mock(*_args, **_kwargs) -> SparseData: return SparseData( label="mock_sparse_data", name="Tidal impedance variation", @@ -183,6 +184,18 @@ def _mock(*_args, **_kwargs) -> np.ndarray: return _mock +def test_deprecated(): + with pytest.warns(DeprecationWarning): + _ = PixelBreath(breath_detection_kwargs={}) + + with pytest.raises(TypeError): + _ = PixelBreath(breath_detection=BreathDetection(), breath_detection_kwargs={}) + + bd_kwargs = {"minimum_duration": 10, "averaging_window_duration": 100.0} + with pytest.warns(DeprecationWarning): + assert PixelBreath(breath_detection_kwargs=bd_kwargs).breath_detection == BreathDetection(**bd_kwargs) + + def test__compute_breaths(): """Test _compute_breaths helper function.""" time = np.array([0, 1, 2, 3, 4]) @@ -241,10 +254,10 @@ def test_store_result_with_errors( request: pytest.FixtureRequest, store_input: bool, sequence_fixture: str, - expected_exception: ValueError | RuntimeError, + expected_exception: type[ValueError | RuntimeError], ): """Test storing results when errors are expected.""" - pi = PixelBreath(breath_detection_kwargs={"minimum_duration": 0.01}) # Ensure that breaths are detected + pi = PixelBreath(breath_detection=BreathDetection(minimum_duration=0.01)) # Ensure that breaths are detected sequence = request.getfixturevalue(sequence_fixture) @@ -271,7 +284,7 @@ def test_store_result_success( sequence_fixture: str, ): """Test storing results when no errors are expected.""" - pi = PixelBreath(breath_detection_kwargs={"minimum_duration": 0.01}) # Ensure that breaths are detected + pi = PixelBreath(breath_detection=BreathDetection(minimum_duration=0.01)) # Ensure that breaths are detected sequence = request.getfixturevalue(sequence_fixture) @@ -303,7 +316,7 @@ def test_with_custom_mean_pixel_tiv( "eitprocessing.parameters.tidal_impedance_variation.TIV.compute_pixel_parameter", side_effect=mock_function, ): - pi = PixelBreath(breath_detection_kwargs={"minimum_duration": 0.01}) + pi = PixelBreath(breath_detection=BreathDetection(minimum_duration=0.01)) result = pi.find_pixel_breaths(mock_eit_data, mock_continuous_data) @@ -321,7 +334,7 @@ def test_with_custom_mean_pixel_tiv( def test_with_zero_impedance(mock_zero_eit_data: EITData, mock_continuous_data: ContinuousData): - pi = PixelBreath(breath_detection_kwargs={"minimum_duration": 0.01}) + pi = PixelBreath(breath_detection=BreathDetection(minimum_duration=0.01)) breath_container = pi.find_pixel_breaths(mock_zero_eit_data, mock_continuous_data) test_result = np.stack(breath_container.values) assert np.all(np.vectorize(lambda x: x is None)(test_result[:, 1, 1])) @@ -335,12 +348,13 @@ def test_with_data(draeger1: Sequence, timpel1: Sequence, pytestconfig: pytest.C draeger1 = copy.deepcopy(draeger1) timpel1 = copy.deepcopy(timpel1) for sequence in draeger1, timpel1: - ssequence = sequence[0:500] + ssequence = sequence pi = PixelBreath() eit_data = ssequence.eit_data["raw"] cd = ssequence.continuous_data["global_impedance_(raw)"] pixel_breaths = pi.find_pixel_breaths(eit_data, cd) test_result = np.stack(pixel_breaths.values) + assert not np.all(test_result == None) # noqa: E711 _, n_rows, n_cols = test_result.shape for row, col in itertools.product(range(n_rows), range(n_cols)): @@ -367,3 +381,44 @@ def test_with_data(draeger1: Sequence, timpel1: Sequence, pytestconfig: pytest.C for breath in filtered_values: # Test whether the indices are in the proper order within a breath assert breath.start_time < breath.middle_time < breath.end_time + + +def test_phase_modes(draeger1: Sequence, pytestconfig: pytest.Config): + if pytestconfig.getoption("--cov"): + pytest.skip("Skip with option '--cov' so other tests can cover 100%.") + + ssequence = draeger1 + eit_data = ssequence.eit_data["raw"] + + # reduce the pixel set to middly 'well-behaved' pixels with positive TIV + eit_data.pixel_impedance = eit_data.pixel_impedance[:, 10:23, 10:23] + + # flip a single pixel, so the differences between algorithms becomes predictable + eit_data.pixel_impedance[:, 6, 6] = -eit_data.pixel_impedance[:, 6, 6] + + cd = ssequence.continuous_data["global_impedance_(raw)"] + + # replace the 'global' data with the sum of the middly pixels + cd.values = np.sum(eit_data.pixel_impedance, axis=(1, 2)) + + pb_negative_amplitude = PixelBreath(phase_correction_mode="negative amplitude").find_pixel_breaths(eit_data, cd) + pb_phase_shift = PixelBreath(phase_correction_mode="phase shift").find_pixel_breaths(eit_data, cd) + + # results are not compared, other than for length; just make sure it runs + pb_none = PixelBreath(phase_correction_mode="none").find_pixel_breaths(eit_data, cd) + + assert len(pb_negative_amplitude) == len(pb_phase_shift) == len(pb_none) + + # all breaths, except for the first and last, should have been detected + assert not np.any(np.array(pb_negative_amplitude.values)[1:-1] == None) # noqa: E711 + assert not np.any(np.array(pb_phase_shift.values)[1:-1] == None) # noqa: E711 + + same_pixel_timing = np.array(pb_negative_amplitude.values) == np.array(pb_phase_shift.values) + assert not np.all(same_pixel_timing) + assert not np.any(same_pixel_timing[1:-1, 6, 6]) # the single flipped pixel + assert np.all(same_pixel_timing[1:-1, :6, :]) # all pixels in the rows above match + assert np.all(same_pixel_timing[1:-1, 7:, :]) # all pixels in the rows below match + assert np.all(same_pixel_timing[1:-1, :, :6]) # all pixels in the columns to the left match + assert np.all(same_pixel_timing[1:-1, :, 7:]) # all pixels in the columns to the right match + assert np.all(same_pixel_timing[0, :, :]) # all first values match, because they are all None + assert np.all(same_pixel_timing[-1, :, :]) # all last values match, because they are all None diff --git a/tests/test_sequence_data.py b/tests/test_sequence_data.py index 38b905341..deea26952 100644 --- a/tests/test_sequence_data.py +++ b/tests/test_sequence_data.py @@ -17,6 +17,7 @@ def create_continuous_data_object(): category="other", time=np.array([]), values=np.array([]), + sample_frequency=20, )