15254007841

Committed 15 May 2025 05:21PM UTC coverage: 83.253% (+0.5%) from 82.774%

Build # 15254007841

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.7.0 → 1.7.1

Run Details

391 of 518 branches covered (75.48%)

Branch coverage included in aggregate %.

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

11 existing lines in 2 files now uncovered.

1503 of 1757 relevant lines covered (85.54%)

0.86 hits per line

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

82.35

/eitprocessing/features/pixel_breath.py

import itertools
import warnings
from collections.abc import Callable
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
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

_SENTINEL_BREATH_DETECTION: Final = BreathDetection()
MAX_XCORR_LAG = 0.75


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.

    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.

    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()
    >>> 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 (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: 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,
        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.

        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.
            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)

        continuous_breaths = self.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)

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

        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
            derived_from=[eit_data],
        )
        if store:
            sequence.interval_data.add(pixel_breaths_container)

        return pixel_breaths_container

    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]

    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[t1:t2, row, col]) + t1 for t1, t2 in itertools.pairwise(times)],
        )

1	import itertools	1✔
2	import warnings	1✔
3	from collections.abc import Callable	1✔
4	from dataclasses import InitVar, dataclass, field	1✔
5	from typing import Final, Literal	1✔
6
7	import numpy as np	1✔
8	from scipy import signal	1✔
9
10	from eitprocessing.datahandling.breath import Breath	1✔
11	from eitprocessing.datahandling.continuousdata import ContinuousData	1✔
12	from eitprocessing.datahandling.eitdata import EITData	1✔
13	from eitprocessing.datahandling.intervaldata import IntervalData	1✔
14	from eitprocessing.datahandling.sequence import Sequence	1✔
15	from eitprocessing.features.breath_detection import BreathDetection	1✔
16
17	_SENTINEL_BREATH_DETECTION: Final = BreathDetection()	1✔
18	MAX_XCORR_LAG = 0.75	1✔
19
20
21	def _return_sentinel_breath_detection() -> BreathDetection:	1✔
22	# Returns a sential of a BreathDetection, which only exists to signal that the default value for breath_detection
23	# was used.
24	return _SENTINEL_BREATH_DETECTION	1✔
25
26
27	@dataclass(kw_only=True)	1✔
28	class PixelBreath:	1✔
29	"""Algorithm for detecting timing of pixel breaths in pixel impedance data.
30
31	This algorithm detects the position of start of inspiration, end of inspiration and
32	end of expiration in pixel impedance data. It uses BreathDetection to find the global start and end
33	of inspiration and expiration. These points are then used to find the start/end of pixel
34	inspiration/expiration in pixel impedance data.
35
36	Some pixel breaths may be phase shifted (inflation starts and ends later compared to others, e.g., due to pendelluft
37	or late airway opening). Other pixel breaths may have a negative amplitude (impedance decreases during inspiration,
38	e.g., due to pleural effusion or reconstruction artifacts). It is not always possible to determine whether a pixel
39	is out of phase or has a negative amplitude. PixelBreath has three different phase correction modes. In 'negative
40	amplitude' mode (default), pixels that have a decrease in amplitude between the start and end of globally defined
41	inspiration, will have a negative amplitude and smaller phase shift. In 'phase shift' mode, all pixel breaths will
42	have positive amplitudes, but can have large phase shifts. In 'none'/`None` mode, all pixels are assumed to be
43	within rouglhy -90 to 90 degrees of phase. Note that the 'none' mode can lead to unexpected results, such as
44	ultra-short (down to 2 frames) or very long breaths.
45
46	Example:
47	```
48	>>> pi = PixelBreath()
49	>>> eit_data = sequence.eit_data['raw']
50	>>> continuous_data = sequence.continuous_data['global_impedance_(raw)']
51	>>> pixel_breaths = pi.find_pixel_breaths(eit_data, continuous_data, sequence)
52	```
53
54	Args:
55	breath_detection (BreathDetection): BreathDetection object to use for detecting breaths.
56	phase_correction_mode: How to resolve pixels that are out-of-phase. Defaults to "negative amplitude".
57	"""
58
59	breath_detection: BreathDetection = field(default_factory=_return_sentinel_breath_detection)	1✔
60	breath_detection_kwargs: InitVar[dict \| None] = None	1✔
61	phase_correction_mode: Literal["negative amplitude", "phase shift", "none"] \| None = "negative amplitude"	1✔
62
63	def __post_init__(self, breath_detection_kwargs: dict \| None):	1✔
64	if breath_detection_kwargs is not None:	1✔
65	if self.breath_detection is not _SENTINEL_BREATH_DETECTION:	1✔
66	msg = (	1✔
67	"`breath_detection_kwargs` is deprecated, and can't be used at the same time as `breath_detection`."
68	)
69	raise TypeError(msg)	1✔
70
71	self.breath_detection = BreathDetection(**breath_detection_kwargs)	1✔
72	warnings.warn(	1✔
73	"`breath_detection_kwargs` is deprecated and will be removed soon. "
74	"Replace with `breath_detection=BreathDetection(**breath_detection_kwargs)`.",
75	DeprecationWarning,
76	)
77
78	def find_pixel_breaths( # noqa: C901, PLR0912, PLR0915	1✔
79	self,
80	eit_data: EITData,
81	continuous_data: ContinuousData,
82	sequence: Sequence \| None = None,
83	store: bool \| None = None,
84	result_label: str = "pixel_breaths",
85	) -> IntervalData:
86	"""Find pixel breaths in the data.
87
88	This method finds the pixel start/end of inspiration/expiration based on the start/end of inspiration/expiration
89	as detected in the continuous data.
90
91	For most pixels, the start of a breath (start inspiration) is the valley between the middles (start of
92	expiration) of the globally defined breaths on either side. The end of a pixel breath is the start of the next
93	pixel breath. The middle of the pixel breath is the peak between the start and end of the pixel breath.
94
95	If the pixel is out of phase or has negative amplitude, the definition of the breath depends on the phase
96	correction mode. In 'negative amplitude' mode, the start of a breath is the peak between the middles of the
97	globally defined breaths on either side, while the middle of the pixel breath is the valley of the start and end
98	of the pixel breath. In 'phase shift' mode, first the phase shift between the pixel impedance and global
99	impedance is determined as the highest crosscorrelation between the signals near a phase shift of 0. The start
100	of breath is the valley between the phase shifted middles of the globally defined breaths on either side.
101
102	Pixel breaths are constructed as a valley-peak-valley combination, representing the start of inspiration, the
103	end of inspiration/start of expiration, and end of expiration.
104
105	Args:
106	eit_data: EITData to apply the algorithm to.
107	continuous_data: ContinuousData to use for global breath detection.
108	result_label: label of the returned IntervalData object, defaults to `'pixel_breaths'`.
109	sequence: optional, Sequence that contains the object to detect pixel breaths in, and/or to store the result
110	in.
111	store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.
112
113	Returns:
114	An IntervalData object containing Breath objects.
115
116	Raises:
117	RuntimeError: If store is set to true but no sequence is provided.
118	ValueError: If the provided sequence is not an instance of the Sequence dataclass.
119	"""
120	if store is None and isinstance(sequence, Sequence):	1✔
121	store = True	1✔
122
123	if store and sequence is None:	1✔
124	msg = "Can't store the result if no Sequence is provided."	1✔
125	raise RuntimeError(msg)	1✔
126
127	if store and not isinstance(sequence, Sequence):	1✔
128	msg = "To store the result a Sequence dataclass must be provided."	1✔
129	raise ValueError(msg)	1✔
130
131	continuous_breaths = self.breath_detection.find_breaths(continuous_data)	1✔
132
133	indices_breath_middles = np.searchsorted(	1✔
134	eit_data.time,
135	[breath.middle_time for breath in continuous_breaths.values],
136	)
137
138	_, n_rows, n_cols = eit_data.pixel_impedance.shape	1✔
139
140	from eitprocessing.parameters.tidal_impedance_variation import TIV	1✔
141
142	pixel_tivs = TIV().compute_pixel_parameter(	1✔
143	eit_data,
144	continuous_data,
145	sequence,
146	tiv_method="inspiratory",
147	tiv_timing="continuous",
148	store=False,
149	) # Set store to false as to not save these pixel tivs as SparseData.
150
151	pixel_tiv_with_continuous_data_timing = (	1✔
152	np.empty((0, n_rows, n_cols)) if not len(pixel_tivs.values) else np.stack(pixel_tivs.values)
153	)
154
155	# Create a mask to detect slices that are entirely NaN
156	all_nan_mask = np.isnan(pixel_tiv_with_continuous_data_timing).all(axis=0)	1✔
157
158	# Initialize the mean_tiv_pixel array with NaNs
159	mean_tiv_pixel = np.full((n_rows, n_cols), np.nan)	1✔
160
161	# Only compute nanmean for slices that are not entirely NaN
162	if not all_nan_mask.all(): # Check if there are any valid (non-all-NaN) slices	1✔
163	mean_tiv_pixel[~all_nan_mask] = np.nanmean(pixel_tiv_with_continuous_data_timing[:, ~all_nan_mask], axis=0)	1✔
164
165	time = eit_data.time	1✔
166	pixel_impedance = eit_data.pixel_impedance	1✔
167
168	pixel_breaths = np.full((len(continuous_breaths), n_rows, n_cols), None)	1✔
169
170	lags = signal.correlation_lags(len(continuous_data), len(continuous_data), mode="same")	1✔
171
172	match self.phase_correction_mode:	1✔
173	case "negative amplitude":	1!
174	allow_negative_amplitude = True	1✔
175	correct_for_phase_shift = None	1✔
UNCOV 176	case "phase shift":	×
UNCOV 177	allow_negative_amplitude = False	×
UNCOV 178	correct_for_phase_shift = True	×
UNCOV 179	case "none" \| None:	×
UNCOV 180	allow_negative_amplitude = False	×
181	correct_for_phase_shift = False	×
182	case _:	×
183	msg = f"Unknown phase correction mode ({self.phase_correction_mode})."	×
184	raise ValueError(msg)	×
185
186	for row, col in itertools.product(range(n_rows), range(n_cols)):	1✔
187	mean_tiv = mean_tiv_pixel[row, col]	1✔
188
189	if np.std(pixel_impedance[:, row, col]) == 0:	1✔
190	# pixel has no amplitude
191	continue	1✔
192
193	if allow_negative_amplitude and mean_tiv < 0:	1✔
194	start_func, middle_func = np.argmax, np.argmin	1✔
195	lagged_indices_breath_middles = indices_breath_middles	1✔
196	else:
197	start_func, middle_func = np.argmin, np.argmax	1✔
198
199	cd = np.copy(continuous_data.values)	1✔
200	cd -= np.nanmean(cd)	1✔
201	pi = np.copy(pixel_impedance[:, row, col])	1✔
202	if not np.all(np.isnan(pi)):	1!
203	pi -= np.nanmean(pixel_impedance[:, row, col])	1✔
204
205	if correct_for_phase_shift:	1!
206	# search for maximum cross correlation within MAX_XCORR_LAG times the average
207	# duration of a breath
UNCOV 208	xcorr = signal.correlate(cd, pi, mode="same")	×
UNCOV 209	max_lag = MAX_XCORR_LAG * np.mean(np.diff(indices_breath_middles))	×
UNCOV 210	lag_range = (lags > -max_lag) & (lags < max_lag)	×
211	# TODO: if this does not work, implement robust peak detection
212
213	# positive lag: pixel inflates later than summed
214	lag = lags[lag_range][np.argmax(xcorr[lag_range])]	×
215
216	# shift search area
UNCOV 217	lagged_indices_breath_middles = indices_breath_middles - lag	×
218	lagged_indices_breath_middles = lagged_indices_breath_middles[	×
219	(lagged_indices_breath_middles >= 0) & (lagged_indices_breath_middles < len(cd))
220	]
221	else:
222	lagged_indices_breath_middles = indices_breath_middles	1✔
223
224	outsides = self._find_extreme_indices(pixel_impedance, lagged_indices_breath_middles, row, col, start_func)	1✔
225	starts = outsides[:-1]	1✔
226	ends = outsides[1:]	1✔
227	middles = self._find_extreme_indices(pixel_impedance, outsides, row, col, middle_func)	1✔
228	# TODO discuss; this block of code is implemented to prevent noisy pixels from breaking the code.
229	# Quick solve is to make entire breath object None if any breath in a pixel does not have
230	# consecutive start, middle and end.
231	# However, this might cause problems elsewhere.
232	if (starts >= middles).any() or (middles >= ends).any():	1!
233	pixel_breath = None	×
234	else:
235	pixel_breath = self._construct_breaths(starts, middles, ends, time)	1✔
236	pixel_breaths[:, row, col] = pixel_breath	1✔
237
238	intervals = [(breath.start_time, breath.end_time) for breath in continuous_breaths.values]	1✔
239
240	pixel_breaths_container = IntervalData(	1✔
241	label=result_label,
242	name="Pixel in- and deflation timing as determined by Pixelbreath",
243	unit=None,
244	category="breath",
245	intervals=intervals,
246	values=list(
247	pixel_breaths,
248	), ## TODO: change back to pixel_breaths array when IntervalData works with 3D array
249	derived_from=[eit_data],
250	)
251	if store:	1✔
252	sequence.interval_data.add(pixel_breaths_container)	1✔
253
254	return pixel_breaths_container	1✔
255
256	def _construct_breaths(self, start: list[int], middle: list[int], end: list[int], time: np.ndarray) -> list:	1✔
257	breaths = [Breath(time[s], time[m], time[e]) for s, m, e in zip(start, middle, end, strict=True)]	1✔
258	# First and last breath are not detected by definition (need two breaths to find one breath)
259	return [None, *breaths, None]	1✔
260
261	def _find_extreme_indices(	1✔
262	self,
263	pixel_impedance: np.ndarray,
264	times: np.ndarray,
265	row: int,
266	col: int,
267	function: Callable,
268	) -> np.ndarray:
269	"""Finds extreme indices in pixel impedance.
270
271	This method divides the pixel impedance for a single pixel (selected using row and col) into smaller segments
272	based on the `times` array. The times array consists of indices to divide these segments. The function iterates
273	over each index in the times array to select consecutive segments of pixel impedance.
274	For each segment, the method applies the `function` (either np.argmax or np.argmin) to extract an extreme value
275	(local maximum or minimum).
276
277	Args:
278	pixel_impedance (np.ndarray): The pixel impedance array from which the function will extract values.
279	Assumed to be 3-dimensional (e.g., time, rows, and columns).
280	times (np.ndarray): 1D array of time indices. These times define the start
281	and end of each segment in the pixel impedance.
282	row (int): The row index in the pixel impedance
283	col (int): The column index in the pixel impedance
284	function (Callable): A function that is applied to each segment of data to find
285	an extreme value (np.argmax or np.argmin)
286
287	Returns:
288	np.ndarray: An array of indices where the extreme values (based on the function)
289	are located for each time segment.
290	"""
291	return np.array(	1✔
292	[function(pixel_impedance[t1:t2, row, col]) + t1 for t1, t2 in itertools.pairwise(times)],
293	)

EIT-ALIVE / eitprocessing / 15254007841

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