15254232358

Committed 26 May 2025 12:41PM UTC coverage: 82.774% (-0.5%) from 83.253%

Build # 15254232358

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.7.1 → 1.7.2

Run Details

389 of 520 branches covered (74.81%)

Branch coverage included in aggregate %.

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

21 existing lines in 2 files now uncovered.

1509 of 1773 relevant lines covered (85.11%)

0.85 hits per line

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

94.8

/eitprocessing/parameters/tidal_impedance_variation.py

import itertools
import sys
import warnings
from dataclasses import InitVar, dataclass, field
from functools import singledispatchmethod
from typing import Final, Literal, NoReturn

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.datahandling.sparsedata import SparseData
from eitprocessing.features.breath_detection import BreathDetection
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: BreathDetection = field(default_factory=_sentinel_breath_detection)
    breath_detection_kwargs: InitVar[dict | None] = 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,
        data: ContinuousData | EITData,
    ) -> NoReturn:
        """Compute the tidal impedance variation per breath on either ContinuousData or EITData, depending on the input.

        Args:
            data: either continuous_data or eit_data to compute TIV on.
        """
        msg = f"This method is implemented for ContinuousData or EITData, not {type(data)}."
        raise TypeError(msg)

    @compute_parameter.register(ContinuousData)
    def compute_continuous_parameter(
        self,
        continuous_data: ContinuousData,
        tiv_method: Literal["inspiratory", "expiratory", "mean"] = "inspiratory",
        sequence: Sequence | None = None,
        store: bool | None = None,
        result_label: str = "continuous_tivs",
    ) -> SparseData:
        """Compute the tidal impedance variation per breath.

        Args:
            continuous_data: The ContinuousData to compute the TIV on.
            tiv_method: The label of which part of the breath the TIV
                should be determined on (inspiratory, expiratory, or mean).
                Defaults to 'inspiratory'.
            sequence: optional, Sequence that contains the object to detect TIV on,
            and/or to store the result in.
            store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.
            result_label: label of the returned SparseData object, defaults to `'continuous_tivs'`.

        Returns:
            A SparseData object with the computed TIV values.

        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.
            ValueError: If tiv_method is not one of 'inspiratory', 'expiratory', or 'mean'.
        """
        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)

        if tiv_method not in {"inspiratory", "expiratory", "mean"}:
            msg = f"Invalid tiv_method: {tiv_method}. Must be one of 'inspiratory', 'expiratory', or 'mean'."
            raise ValueError(msg)

        breaths = self._detect_breaths(continuous_data)

        tiv_values = self._calculate_tiv_values(
            continuous_data.values,
            continuous_data.time,
            breaths.values,
            tiv_method,
            tiv_timing="continuous",
        )
        tiv_container = SparseData(
            label=result_label,
            name="Continuous tidal impedance variation",
            unit=None,
            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",
            derived_from=[continuous_data],
            values=tiv_values,
        )
        if store:
            sequence.sparse_data.add(tiv_container)

        return tiv_container

    @compute_parameter.register(EITData)
    def compute_pixel_parameter(
        self,
        eit_data: EITData,
        continuous_data: ContinuousData,
        sequence: Sequence,
        tiv_method: Literal["inspiratory", "expiratory", "mean"] = "inspiratory",
        tiv_timing: Literal["pixel", "continuous"] = "pixel",
        store: bool | None = None,
        result_label: str = "pixel_tivs",
    ) -> SparseData:
        """Compute the tidal impedance variation per breath on pixel level.

        Args:
            sequence: The sequence containing the data.
            eit_data: The eit pixel level data to determine the TIV of.
            continuous_data: The continuous data to determine the continuous data breaths or pixel level breaths.
            tiv_method: The label of which part of the breath the TIV should be determined on
                        (inspiratory, expiratory or mean). Defaults to 'inspiratory'.
            tiv_timing: The label of which timing should be used to compute the TIV, either based on breaths
                        detected in continuous data ('continuous') or based on pixel breaths ('pixel').
                        Defaults to 'pixel'.
            result_label: label of the returned IntervalData object, defaults to `'pixel_tivs'`.
            store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.

        Returns:
            A SparseData object with the computed TIV values.

        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.
            ValueError: If tiv_method is not one of 'inspiratory', 'expiratory', or 'mean'.
            ValueError: If tiv_timing is not one of 'continuous' or 'pixel'.
        """
        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)

        if tiv_method not in ["inspiratory", "expiratory", "mean"]:
            msg = f"Invalid {tiv_method}. The tiv_method must be either 'inspiratory', 'expiratory' or 'mean'."
            raise ValueError(msg)

        if tiv_timing not in ["continuous", "pixel"]:
            msg = f"Invalid {tiv_timing}. The tiv_timing must be either 'continuous' or 'pixel'."
            raise ValueError(msg)

        data = eit_data.pixel_impedance
        _, n_rows, n_cols = data.shape

        if tiv_timing == "pixel":
            pixel_breaths = self._detect_pixel_breaths(
                eit_data,
                continuous_data,
                sequence,
                store=False,
            )  # Set store to false as to not save these pixel breaths as IntervalData.
            # Check if pixel_breaths.values is empty
            breath_data = (
                np.empty((0, n_rows, n_cols)) if not len(pixel_breaths.values) else np.stack(pixel_breaths.values)
            )
            ## TODO: replace with breath_data = pixel_breaths.values when IntervalData works with 3D array
        else:  # tiv_timing == "continuous"
            global_breaths = self._detect_breaths(
                continuous_data,
            )
            breath_data = global_breaths.values

        number_of_breaths = len(breath_data)
        all_pixels_tiv_values = np.full((number_of_breaths, n_rows, n_cols), None, dtype=object)
        all_pixels_breath_timings = np.full((number_of_breaths, n_rows, n_cols), None, dtype=object)

        for row, col in itertools.product(range(n_rows), range(n_cols)):
            time_series = data[:, row, col]
            breaths = breath_data[:, row, col] if tiv_timing == "pixel" else breath_data
            pixel_tiv_values = self._calculate_tiv_values(
                time_series,
                eit_data.time,
                breaths,
                tiv_method,
                tiv_timing,
            )
            # Get the middle times of each breath where breath is not None
            pixel_breath_timings = [breath.middle_time for breath in breaths if breath is not None]

            # Store these in all_pixels_breath_timings, ensuring they match the expected shape
            all_pixels_breath_timings[: len(pixel_breath_timings), row, col] = pixel_breath_timings

            all_pixels_tiv_values[:, row, col] = pixel_tiv_values

        tiv_container = SparseData(
            label=result_label,
            name="Pixel tidal impedance variation",
            unit=None,
            category="impedance difference",
            time=list(all_pixels_breath_timings),
            description="Tidal impedance variation determined on pixel impedance",
            derived_from=[eit_data],
            values=list(all_pixels_tiv_values.astype(float)),
        )

        if store:
            sequence.sparse_data.add(tiv_container)

        return tiv_container

    def _detect_breaths(self, data: ContinuousData) -> IntervalData:
        return self.breath_detection.find_breaths(data)

    def _detect_pixel_breaths(
        self,
        eit_data: EITData,
        continuous_data: ContinuousData,
        sequence: Sequence,
        store: bool,
    ) -> IntervalData:
        return self.pixel_breath.find_pixel_breaths(
            eit_data,
            continuous_data,
            result_label="pixel breaths",
            sequence=sequence,
            store=store,
        )

    def _calculate_tiv_values(
        self,
        data: np.ndarray,
        time: np.ndarray,
        breaths: list[Breath],
        tiv_method: str,
        tiv_timing: str,  # noqa: ARG002 # remove when restructuring
    ) -> list:
        # Filter out None breaths
        breaths = np.array(breaths)
        valid_breath_indices = np.flatnonzero([breath is not None for breath in breaths])
        valid_breaths = breaths[valid_breath_indices]

        # If there are no valid breaths, return a list of None with the same length as the number of breaths
        if not len(valid_breaths):
            return np.full(len(breaths), np.nan)

        start_indices = np.searchsorted(time, [breath.start_time for breath in valid_breaths])
        middle_indices = np.searchsorted(time, [breath.middle_time for breath in valid_breaths])
        end_indices = np.searchsorted(time, [breath.end_time for breath in valid_breaths])

        if tiv_method == "inspiratory":
            tiv_values = np.squeeze(np.diff(data[[start_indices, middle_indices]], axis=0), axis=0)

        elif tiv_method == "expiratory":
            tiv_values = np.squeeze(np.diff(data[[end_indices, middle_indices]], axis=0), axis=0)

        elif tiv_method == "mean":
            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

        tiv_values_ = np.full(len(breaths), np.nan)
        tiv_values_[valid_breath_indices] = tiv_values

        return tiv_values_

1	import itertools	1✔
2	import sys	1✔
3	import warnings	1✔
4	from dataclasses import InitVar, dataclass, field	1✔
5	from functools import singledispatchmethod	1✔
6	from typing import Final, Literal, NoReturn	1✔
7
8	import numpy as np	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.datahandling.sparsedata import SparseData	1✔
16	from eitprocessing.features.breath_detection import BreathDetection	1✔
17	from eitprocessing.features.pixel_breath import PixelBreath	1✔
18	from eitprocessing.parameters import ParameterCalculation	1✔
19
20	_SENTINEL_PIXEL_BREATH: Final = PixelBreath()	1✔
21	_SENTINEL_BREATH_DETECTION: Final = BreathDetection()	1✔
22
23
24	def _sentinel_pixel_breath() -> PixelBreath:	1✔
25	# Returns a sential of a PixelBreath, which only exists to signal that the default value for pixel_breath was used.
26	return _SENTINEL_PIXEL_BREATH	1✔
27
28
29	def _sentinel_breath_detection() -> BreathDetection:	1✔
30	# Returns a sential of a BreathDetection, which only exists to signal that the default value for breath_detection
31	# was used.
32	return _SENTINEL_BREATH_DETECTION	1✔
33
34
35	@dataclass	1✔
36	class TIV(ParameterCalculation):	1✔
37	"""Compute the tidal impedance variation (TIV) per breath."""
38
39	method: Literal["extremes"] = "extremes"	1✔
40	breath_detection: BreathDetection = field(default_factory=_sentinel_breath_detection)	1✔
41	breath_detection_kwargs: InitVar[dict \| None] = None	1✔
42
43	# The default is a sentinal that will be replaced in __post_init__
44	pixel_breath: PixelBreath = field(default_factory=_sentinel_pixel_breath)	1✔
45
46	def __post_init__(self, breath_detection_kwargs: dict \| None) -> None:	1✔
47	if self.method != "extremes":	1✔
48	msg = f"Method {self.method} is not implemented. The method must be 'extremes'."	1✔
49	raise NotImplementedError(msg)	1✔
50
51	if breath_detection_kwargs is not None:	1✔
52	if self.breath_detection is not _SENTINEL_BREATH_DETECTION:	1✔
53	msg = (	1✔
54	"`breath_detection_kwargs` is deprecated, and can't be used at the same time as `breath_detection`."
55	)
56	raise TypeError(msg)	1✔
57
58	self.breath_detection = BreathDetection(**breath_detection_kwargs)	1✔
59	warnings.warn(	1✔
60	"`breath_detection_kwargs` is deprecated and will be removed soon. "
61	"Replace with `breath_detection=BreathDetection(**breath_detection_kwargs)`.",
62	DeprecationWarning,
63	)
64
65	if self.pixel_breath is _SENTINEL_PIXEL_BREATH:	1!
66	# If no value was provided at initialization, PixelBreath should use the same BreathDetection object as TIV.
67	# However, a default factory cannot be used because it can't access self.breath_detection. The sentinal
68	# object is replaced here (only if pixel_breath was not provided) with the correct BreathDetection object.
69	self.pixel_breath = PixelBreath(breath_detection=self.breath_detection)	1✔
70
71	@singledispatchmethod	1✔
72	def compute_parameter(	1✔
73	self,
74	data: ContinuousData \| EITData,
75	) -> NoReturn:
76	"""Compute the tidal impedance variation per breath on either ContinuousData or EITData, depending on the input.
77
78	Args:
79	data: either continuous_data or eit_data to compute TIV on.
80	"""
81	msg = f"This method is implemented for ContinuousData or EITData, not {type(data)}."	1✔
82	raise TypeError(msg)	1✔
83
84	@compute_parameter.register(ContinuousData)	1✔
85	def compute_continuous_parameter(	1✔
86	self,
87	continuous_data: ContinuousData,
88	tiv_method: Literal["inspiratory", "expiratory", "mean"] = "inspiratory",
89	sequence: Sequence \| None = None,
90	store: bool \| None = None,
91	result_label: str = "continuous_tivs",
92	) -> SparseData:
93	"""Compute the tidal impedance variation per breath.
94
95	Args:
96	continuous_data: The ContinuousData to compute the TIV on.
97	tiv_method: The label of which part of the breath the TIV
98	should be determined on (inspiratory, expiratory, or mean).
99	Defaults to 'inspiratory'.
100	sequence: optional, Sequence that contains the object to detect TIV on,
101	and/or to store the result in.
102	store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.
103	result_label: label of the returned SparseData object, defaults to `'continuous_tivs'`.
104
105	Returns:
106	A SparseData object with the computed TIV values.
107
108	Raises:
109	RuntimeError: If store is set to true but no sequence is provided.
110	ValueError: If the provided sequence is not an instance of the Sequence dataclass.
111	ValueError: If tiv_method is not one of 'inspiratory', 'expiratory', or 'mean'.
112	"""
113	if store is None and isinstance(sequence, Sequence):	1✔
114	store = True	1✔
115
116	if store and sequence is None:	1✔
117	msg = "Can't store the result if no Sequence is provided."	1✔
118	raise RuntimeError(msg)	1✔
119
120	if store and not isinstance(sequence, Sequence):	1✔
121	msg = "To store the result a Sequence dataclass must be provided."	1✔
122	raise ValueError(msg)	1✔
123
124	if tiv_method not in {"inspiratory", "expiratory", "mean"}:	1✔
125	msg = f"Invalid tiv_method: {tiv_method}. Must be one of 'inspiratory', 'expiratory', or 'mean'."	1✔
126	raise ValueError(msg)	1✔
127
128	breaths = self._detect_breaths(continuous_data)	1✔
129
130	tiv_values = self._calculate_tiv_values(	1✔
131	continuous_data.values,
132	continuous_data.time,
133	breaths.values,
134	tiv_method,
135	tiv_timing="continuous",
136	)
137	tiv_container = SparseData(	1✔
138	label=result_label,
139	name="Continuous tidal impedance variation",
140	unit=None,
141	category="impedance difference",
142	time=[breath.middle_time for breath in breaths.values if breath is not None],
143	description="Tidal impedance variation determined on continuous data",
144	derived_from=[continuous_data],
145	values=tiv_values,
146	)
147	if store:	1✔
148	sequence.sparse_data.add(tiv_container)	1✔
149
150	return tiv_container	1✔
151
152	@compute_parameter.register(EITData)	1✔
153	def compute_pixel_parameter(	1✔
154	self,
155	eit_data: EITData,
156	continuous_data: ContinuousData,
157	sequence: Sequence,
158	tiv_method: Literal["inspiratory", "expiratory", "mean"] = "inspiratory",
159	tiv_timing: Literal["pixel", "continuous"] = "pixel",
160	store: bool \| None = None,
161	result_label: str = "pixel_tivs",
162	) -> SparseData:
163	"""Compute the tidal impedance variation per breath on pixel level.
164
165	Args:
166	sequence: The sequence containing the data.
167	eit_data: The eit pixel level data to determine the TIV of.
168	continuous_data: The continuous data to determine the continuous data breaths or pixel level breaths.
169	tiv_method: The label of which part of the breath the TIV should be determined on
170	(inspiratory, expiratory or mean). Defaults to 'inspiratory'.
171	tiv_timing: The label of which timing should be used to compute the TIV, either based on breaths
172	detected in continuous data ('continuous') or based on pixel breaths ('pixel').
173	Defaults to 'pixel'.
174	result_label: label of the returned IntervalData object, defaults to `'pixel_tivs'`.
175	store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.
176
177	Returns:
178	A SparseData object with the computed TIV values.
179
180	Raises:
181	RuntimeError: If store is set to true but no sequence is provided.
182	ValueError: If the provided sequence is not an instance of the Sequence dataclass.
183	ValueError: If tiv_method is not one of 'inspiratory', 'expiratory', or 'mean'.
184	ValueError: If tiv_timing is not one of 'continuous' or 'pixel'.
185	"""
186	if store is None and isinstance(sequence, Sequence):	1✔
187	store = True	1✔
188
189	if store and sequence is None:	1✔
190	msg = "Can't store the result if no Sequence is provided."	1✔
191	raise RuntimeError(msg)	1✔
192
193	if store and not isinstance(sequence, Sequence):	1✔
194	msg = "To store the result a Sequence dataclass must be provided."	1✔
195	raise ValueError(msg)	1✔
196
197	if tiv_method not in ["inspiratory", "expiratory", "mean"]:	1✔
198	msg = f"Invalid {tiv_method}. The tiv_method must be either 'inspiratory', 'expiratory' or 'mean'."	1✔
199	raise ValueError(msg)	1✔
200
201	if tiv_timing not in ["continuous", "pixel"]:	1✔
202	msg = f"Invalid {tiv_timing}. The tiv_timing must be either 'continuous' or 'pixel'."	1✔
203	raise ValueError(msg)	1✔
204
205	data = eit_data.pixel_impedance	1✔
206	_, n_rows, n_cols = data.shape	1✔
207
208	if tiv_timing == "pixel":	1✔
209	pixel_breaths = self._detect_pixel_breaths(	1✔
210	eit_data,
211	continuous_data,
212	sequence,
213	store=False,
214	) # Set store to false as to not save these pixel breaths as IntervalData.
215	# Check if pixel_breaths.values is empty
216	breath_data = (	1✔
217	np.empty((0, n_rows, n_cols)) if not len(pixel_breaths.values) else np.stack(pixel_breaths.values)
218	)
219	## TODO: replace with breath_data = pixel_breaths.values when IntervalData works with 3D array
220	else: # tiv_timing == "continuous"
221	global_breaths = self._detect_breaths(	1✔
222	continuous_data,
223	)
224	breath_data = global_breaths.values	1✔
225
226	number_of_breaths = len(breath_data)	1✔
227	all_pixels_tiv_values = np.full((number_of_breaths, n_rows, n_cols), None, dtype=object)	1✔
228	all_pixels_breath_timings = np.full((number_of_breaths, n_rows, n_cols), None, dtype=object)	1✔
229
230	for row, col in itertools.product(range(n_rows), range(n_cols)):	1✔
231	time_series = data[:, row, col]	1✔
232	breaths = breath_data[:, row, col] if tiv_timing == "pixel" else breath_data	1✔
233	pixel_tiv_values = self._calculate_tiv_values(	1✔
234	time_series,
235	eit_data.time,
236	breaths,
237	tiv_method,
238	tiv_timing,
239	)
240	# Get the middle times of each breath where breath is not None
241	pixel_breath_timings = [breath.middle_time for breath in breaths if breath is not None]	1✔
242
243	# Store these in all_pixels_breath_timings, ensuring they match the expected shape
244	all_pixels_breath_timings[: len(pixel_breath_timings), row, col] = pixel_breath_timings	1✔
245
246	all_pixels_tiv_values[:, row, col] = pixel_tiv_values	1✔
247
248	tiv_container = SparseData(	1✔
249	label=result_label,
250	name="Pixel tidal impedance variation",
251	unit=None,
252	category="impedance difference",
253	time=list(all_pixels_breath_timings),
254	description="Tidal impedance variation determined on pixel impedance",
255	derived_from=[eit_data],
256	values=list(all_pixels_tiv_values.astype(float)),
257	)
258
259	if store:	1✔
260	sequence.sparse_data.add(tiv_container)	1✔
261
262	return tiv_container	1✔
263
264	def _detect_breaths(self, data: ContinuousData) -> IntervalData:	1✔
265	return self.breath_detection.find_breaths(data)	1✔
266
267	def _detect_pixel_breaths(	1✔
268	self,
269	eit_data: EITData,
270	continuous_data: ContinuousData,
271	sequence: Sequence,
272	store: bool,
273	) -> IntervalData:
274	return self.pixel_breath.find_pixel_breaths(	1✔
275	eit_data,
276	continuous_data,
277	result_label="pixel breaths",
278	sequence=sequence,
279	store=store,
280	)
281
282	def _calculate_tiv_values(	1✔
283	self,
284	data: np.ndarray,
285	time: np.ndarray,
286	breaths: list[Breath],
287	tiv_method: str,
288	tiv_timing: str, # noqa: ARG002 # remove when restructuring
289	) -> list:
290	# Filter out None breaths
291	breaths = np.array(breaths)	1✔
292	valid_breath_indices = np.flatnonzero([breath is not None for breath in breaths])	1✔
293	valid_breaths = breaths[valid_breath_indices]	1✔
294
295	# If there are no valid breaths, return a list of None with the same length as the number of breaths
296	if not len(valid_breaths):	1✔
297	return np.full(len(breaths), np.nan)	1✔
298
299	start_indices = np.searchsorted(time, [breath.start_time for breath in valid_breaths])	1✔
300	middle_indices = np.searchsorted(time, [breath.middle_time for breath in valid_breaths])	1✔
301	end_indices = np.searchsorted(time, [breath.end_time for breath in valid_breaths])	1✔
302
303	if tiv_method == "inspiratory":	1✔
304	tiv_values = np.squeeze(np.diff(data[[start_indices, middle_indices]], axis=0), axis=0)	1✔
305
306	elif tiv_method == "expiratory":	1✔
307	tiv_values = np.squeeze(np.diff(data[[end_indices, middle_indices]], axis=0), axis=0)	1✔
308
309	elif tiv_method == "mean":	1!
310	mean_outer_values = data[[start_indices, end_indices]].mean(axis=0)	1✔
311	end_inspiratory_values = data[middle_indices]	1✔
312	tiv_values = end_inspiratory_values - mean_outer_values	1✔
313	else:
314	msg = f"`tiv_method` ({tiv_method}) not valid."	×
315	exc = ValueError(msg)	×
316	if sys.version_info >= (3, 11):	×
UNCOV 317	exc.add_note("Valid value for `tiv_method` are 'inspiratory', 'expiratory' and 'mean'.")	×
UNCOV 318	raise exc	×
319
320	tiv_values_ = np.full(len(breaths), np.nan)	1✔
321	tiv_values_[valid_breath_indices] = tiv_values	1✔
322
323	return tiv_values_	1✔

EIT-ALIVE / eitprocessing / 15254232358

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