11745726274

Committed 08 Nov 2024 04:20PM UTC coverage: 82.129% (+2.1%) from 80.0%

Build # 11745726274

Build Type

push

github

Committed by

actions-user

Commit Message

Bump version: 1.4.4 → 1.4.5

Run Details

336 of 452 branches covered (74.34%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

30 existing lines in 5 files now uncovered.

1346 of 1596 relevant lines covered (84.34%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.47

/eitprocessing/features/pixel_breath.py

import itertools
from collections.abc import Callable
from dataclasses import dataclass, field

import numpy as np

from eitprocessing.datahandling.breath import Breath
from eitprocessing.datahandling.continuousdata import ContinuousData
from eitprocessing.datahandling.eitdata import EITData
from eitprocessing.datahandling.intervaldata import IntervalData
from eitprocessing.datahandling.sequence import Sequence
from eitprocessing.features.breath_detection import BreathDetection


@dataclass
class PixelBreath:
    """Algorithm for detecting timing of pixel breaths in pixel impedance data.

    This algorithm detects the position of start of inspiration, end of inspiration and
    end of expiration in pixel impedance data. It uses BreathDetection to find the global start and end
    of inspiration and expiration. These points are then used to find the start/end of pixel
    inspiration/expiration in pixel impedance data.

    Example:
    ```
    >>> pi = PixelBreath()
    >>> eit_data = sequence.eit_data['raw']
    >>> continuous_data = sequence.continuous_data['global_impedance_(raw)']
    >>> pixel_breaths = pi.find_pixel_breaths(eit_data, continuous_data, sequence)
    ```

    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_kwargs: dict = field(default_factory=dict)

    def find_pixel_breaths(
        self,
        eit_data: EITData,
        continuous_data: ContinuousData,
        sequence: Sequence | None = None,
        store: bool | None = None,
        result_label: str = "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.

        Args:
            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.
            store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.

        Returns:
            An IntervalData object containing Breath objects.

        Raises:
            RuntimeError: If store is set to true but no sequence is provided.
            ValueError: If the provided sequence is not an instance of the Sequence dataclass.
        """
        if store is None and isinstance(sequence, Sequence):
            store = True

        if store and sequence is None:
            msg = "Can't store the result if no Sequence is provided."
            raise RuntimeError(msg)

        if store and not isinstance(sequence, Sequence):
            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)

        indices_breath_middles = np.searchsorted(
            eit_data.time,
            [breath.middle_time for breath in continuous_breaths.values],
        )

        _, n_rows, n_cols = eit_data.pixel_impedance.shape

        from eitprocessing.parameters.tidal_impedance_variation import TIV

        pixel_tivs = TIV().compute_pixel_parameter(
            eit_data,
            continuous_data,
            sequence,
            tiv_method="inspiratory",
            tiv_timing="continuous",
            store=False,
        )  # Set store to false as to not save these pixel tivs as SparseData.

        pixel_tiv_with_continuous_data_timing = (
            np.empty((0, n_rows, n_cols)) if not len(pixel_tivs.values) else np.stack(pixel_tivs.values)
        )

        # Create a mask to detect slices that are entirely NaN
        all_nan_mask = np.isnan(pixel_tiv_with_continuous_data_timing).all(axis=0)

        # Initialize the mean_tiv_pixel array with NaNs
        mean_tiv_pixel = np.full((n_rows, n_cols), np.nan)

        # Only compute nanmean for slices that are not entirely NaN
        if not all_nan_mask.all():  # Check if there are any valid (non-all-NaN) slices
            mean_tiv_pixel[~all_nan_mask] = np.nanmean(pixel_tiv_with_continuous_data_timing[:, ~all_nan_mask], axis=0)

        time = eit_data.time
        pixel_impedance = eit_data.pixel_impedance

        pixel_breaths = np.full((len(continuous_breaths), n_rows, n_cols), None)

        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
                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]

        pixel_breaths_container = IntervalData(
            label=result_label,
            name="Pixel in- and deflation timing as determined by Pixelbreath",
            unit=None,
            category="breath",
            intervals=intervals,
            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:
            sequence.interval_data.add(pixel_breaths_container)

        return pixel_breaths_container

    def _construct_breaths(self, start: list, middle: list, end: list, 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]

    def _find_extreme_indices(
        self,
        pixel_impedance: np.ndarray,
        times: np.ndarray,
        row: int,
        col: int,
        function: Callable,
    ) -> np.ndarray:
        """Finds extreme indices in pixel impedance.

        This method divides the pixel impedance for a single pixel (selected using row and col) into smaller segments
        based on the `times` array. The times array consists of indices to divide these segments. The function iterates
        over each index in the times array to select consecutive segments of pixel impedance.
        For each segment, the method applies the `function` (either np.argmax or np.argmin) to extract an extreme value
        (local maximum or minimum).

        Args:
            pixel_impedance (np.ndarray): The pixel impedance array from which the function will extract values.
                Assumed to be 3-dimensional (e.g., time, rows, and columns).
            times (np.ndarray): 1D array of time indices. These times define the start
                and end of each segment in the pixel impedance.
            row (int): The row index in the pixel impedance
            col (int): The column index in the pixel impedance
            function (Callable): A function that is applied to each segment of data to find
                an extreme value (np.argmax or np.argmin)

        Returns:
            np.ndarray: An array of indices where the extreme values (based on the function)
            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)],
        )

1	import itertools	1✔
2	from collections.abc import Callable	1✔
3	from dataclasses import dataclass, field	1✔
4
5	import numpy as np	1✔
6
7	from eitprocessing.datahandling.breath import Breath	1✔
8	from eitprocessing.datahandling.continuousdata import ContinuousData	1✔
9	from eitprocessing.datahandling.eitdata import EITData	1✔
10	from eitprocessing.datahandling.intervaldata import IntervalData	1✔
11	from eitprocessing.datahandling.sequence import Sequence	1✔
12	from eitprocessing.features.breath_detection import BreathDetection	1✔
13
14
15	@dataclass	1✔
16	class PixelBreath:	1✔
17	"""Algorithm for detecting timing of pixel breaths in pixel impedance data.
18
19	This algorithm detects the position of start of inspiration, end of inspiration and
20	end of expiration in pixel impedance data. It uses BreathDetection to find the global start and end
21	of inspiration and expiration. These points are then used to find the start/end of pixel
22	inspiration/expiration in pixel impedance data.
23
24	Example:
25	```
26	>>> pi = PixelBreath()
27	>>> eit_data = sequence.eit_data['raw']
28	>>> continuous_data = sequence.continuous_data['global_impedance_(raw)']
29	>>> pixel_breaths = pi.find_pixel_breaths(eit_data, continuous_data, sequence)
30	```
31
32	Args:
33	breath_detection_kwargs (dict): A dictionary of keyword arguments for breath detection.
34	The available keyword arguments are:
35	minimum_duration: minimum expected duration of breaths, defaults to 2/3 of a second
36	averaging_window_duration: duration of window used for averaging the data, defaults to 15 seconds
37	averaging_window_function: function used to create a window for averaging the data, defaults to np.blackman
38	amplitude_cutoff_fraction: fraction of the median amplitude below which breaths are removed, defaults to 0.25
39	invalid_data_removal_window_length: window around invalid data in which breaths are removed, defaults to 0.5
40	invalid_data_removal_percentile: the nth percentile of values used to remove outliers, defaults to 5
41	invalid_data_removal_multiplier: the multiplier used to remove outliers, defaults to 4
42	"""
43
44	breath_detection_kwargs: dict = field(default_factory=dict)	1✔
45
46	def find_pixel_breaths(	1✔
47	self,
48	eit_data: EITData,
49	continuous_data: ContinuousData,
50	sequence: Sequence \| None = None,
51	store: bool \| None = None,
52	result_label: str = "pixel_breaths",
53	) -> IntervalData:
54	"""Find pixel breaths in the data.
55
56	This method finds the pixel start/end of inspiration/expiration
57	based on the start/end of inspiration/expiration as detected
58	in the continuous data.
59
60	If pixel impedance is in phase (within 180 degrees) with the continuous data,
61	the start of breath of that pixel is defined as the local minimum between
62	two end-inspiratory points in the continuous signal.
63	The end of expiration of that pixel is defined as the local minimum between two
64	consecutive end-inspiratory points in the continuous data.
65	The end of inspiration of that pixel is defined as the local maximum between
66	the start of inspiration and end of expiration of that pixel.
67
68	If pixel impedance is out of phase with the continuous signal,
69	the start of inspiration of that pixel is defined as the local maximum between
70	two end-inspiration points in the continuous data.
71	The end of expiration of that pixel is defined as the local maximum between two
72	consecutive end-inspiratory points in the continuous data.
73	The end of inspiration of that pixel is defined as the local minimum between
74	the start of inspiration and end of expiration of that pixel.
75
76	Pixel breaths are constructed as a valley-peak-valley combination,
77	representing the start of inspiration, the end of inspiration/start of
78	expiration, and end of expiration.
79
80	Args:
81	eit_data: EITData to apply the algorithm to
82	continuous_data: ContinuousData to use for global breath detection
83	result_label: label of the returned IntervalData object, defaults to `'pixel_breaths'`.
84	sequence: optional, Sequence that contains the object to detect pixel breaths in,
85	and/or to store the result in.
86	store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.
87
88	Returns:
89	An IntervalData object containing Breath objects.
90
91	Raises:
92	RuntimeError: If store is set to true but no sequence is provided.
93	ValueError: If the provided sequence is not an instance of the Sequence dataclass.
94	"""
95	if store is None and isinstance(sequence, Sequence):	1✔
96	store = True	1✔
97
98	if store and sequence is None:	1✔
99	msg = "Can't store the result if no Sequence is provided."	1✔
100	raise RuntimeError(msg)	1✔
101
102	if store and not isinstance(sequence, Sequence):	1✔
103	msg = "To store the result a Sequence dataclass must be provided."	1✔
104	raise ValueError(msg)	1✔
105
106	bd_kwargs = self.breath_detection_kwargs.copy()	1✔
107	breath_detection = BreathDetection(**bd_kwargs)	1✔
108	continuous_breaths = breath_detection.find_breaths(continuous_data)	1✔
109
110	indices_breath_middles = np.searchsorted(	1✔
111	eit_data.time,
112	[breath.middle_time for breath in continuous_breaths.values],
113	)
114
115	_, n_rows, n_cols = eit_data.pixel_impedance.shape	1✔
116
117	from eitprocessing.parameters.tidal_impedance_variation import TIV	1✔
118
119	pixel_tivs = TIV().compute_pixel_parameter(	1✔
120	eit_data,
121	continuous_data,
122	sequence,
123	tiv_method="inspiratory",
124	tiv_timing="continuous",
125	store=False,
126	) # Set store to false as to not save these pixel tivs as SparseData.
127
128	pixel_tiv_with_continuous_data_timing = (	1✔
129	np.empty((0, n_rows, n_cols)) if not len(pixel_tivs.values) else np.stack(pixel_tivs.values)
130	)
131
132	# Create a mask to detect slices that are entirely NaN
133	all_nan_mask = np.isnan(pixel_tiv_with_continuous_data_timing).all(axis=0)	1✔
134
135	# Initialize the mean_tiv_pixel array with NaNs
136	mean_tiv_pixel = np.full((n_rows, n_cols), np.nan)	1✔
137
138	# Only compute nanmean for slices that are not entirely NaN
139	if not all_nan_mask.all(): # Check if there are any valid (non-all-NaN) slices	1✔
140	mean_tiv_pixel[~all_nan_mask] = np.nanmean(pixel_tiv_with_continuous_data_timing[:, ~all_nan_mask], axis=0)	1✔
141
142	time = eit_data.time	1✔
143	pixel_impedance = eit_data.pixel_impedance	1✔
144
145	pixel_breaths = np.full((len(continuous_breaths), n_rows, n_cols), None)	1✔
146
147	for row, col in itertools.product(range(n_rows), range(n_cols)):	1✔
148	mean_tiv = mean_tiv_pixel[row, col]	1✔
149
150	if np.any(pixel_impedance[:, row, col] > 0):	1✔
151	if mean_tiv < 0:	1✔
152	start_func, middle_func = np.argmax, np.argmin	1✔
153	else:
154	start_func, middle_func = np.argmin, np.argmax	1✔
155
156	outsides = self._find_extreme_indices(pixel_impedance, indices_breath_middles, row, col, start_func)	1✔
157	starts = outsides[:-1]	1✔
158	ends = outsides[1:]	1✔
159	middles = self._find_extreme_indices(pixel_impedance, outsides, row, col, middle_func)	1✔
160	# TODO discuss; this block of code is implemented to prevent noisy pixels from breaking the code.
161	# Quick solve is to make entire breath object None if any breath in a pixel does not have
162	# consecutive start, middle and end.
163	# However, this might cause problems elsewhere.
164	if (starts >= middles).any() or (middles >= ends).any():	1!
UNCOV 165	pixel_breath = None	×
166	else:
167	pixel_breath = self._construct_breaths(starts, middles, ends, time)	1✔
168	pixel_breaths[:, row, col] = pixel_breath	1✔
169
170	intervals = [(breath.start_time, breath.end_time) for breath in continuous_breaths.values]	1✔
171
172	pixel_breaths_container = IntervalData(	1✔
173	label=result_label,
174	name="Pixel in- and deflation timing as determined by Pixelbreath",
175	unit=None,
176	category="breath",
177	intervals=intervals,
178	values=list(
179	pixel_breaths,
180	), ## TODO: change back to pixel_breaths array when IntervalData works with 3D array
181	parameters=self.breath_detection_kwargs,
182	derived_from=[eit_data],
183	)
184	if store:	1✔
185	sequence.interval_data.add(pixel_breaths_container)	1✔
186
187	return pixel_breaths_container	1✔
188
189	def _construct_breaths(self, start: list, middle: list, end: list, time: np.ndarray) -> list:	1✔
190	breaths = [Breath(time[s], time[m], time[e]) for s, m, e in zip(start, middle, end, strict=True)]	1✔
191	# First and last breath are not detected by definition (need two breaths to find one breath)
192	return [None, *breaths, None]	1✔
193
194	def _find_extreme_indices(	1✔
195	self,
196	pixel_impedance: np.ndarray,
197	times: np.ndarray,
198	row: int,
199	col: int,
200	function: Callable,
201	) -> np.ndarray:
202	"""Finds extreme indices in pixel impedance.
203
204	This method divides the pixel impedance for a single pixel (selected using row and col) into smaller segments
205	based on the `times` array. The times array consists of indices to divide these segments. The function iterates
206	over each index in the times array to select consecutive segments of pixel impedance.
207	For each segment, the method applies the `function` (either np.argmax or np.argmin) to extract an extreme value
208	(local maximum or minimum).
209
210	Args:
211	pixel_impedance (np.ndarray): The pixel impedance array from which the function will extract values.
212	Assumed to be 3-dimensional (e.g., time, rows, and columns).
213	times (np.ndarray): 1D array of time indices. These times define the start
214	and end of each segment in the pixel impedance.
215	row (int): The row index in the pixel impedance
216	col (int): The column index in the pixel impedance
217	function (Callable): A function that is applied to each segment of data to find
218	an extreme value (np.argmax or np.argmin)
219
220	Returns:
221	np.ndarray: An array of indices where the extreme values (based on the function)
222	are located for each time segment.
223	"""
224	return np.array(	1✔
225	[function(pixel_impedance[times[i] : times[i + 1], row, col]) + times[i] for i in range(len(times) - 1)],
226	)

EIT-ALIVE / eitprocessing / 11745726274

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous