15051278226

Committed 15 May 2025 05:21PM UTC coverage: 83.253% (-0.6%) from 83.838%

Build # 15051278226

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

31 existing lines in 4 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

85.07

/eitprocessing/parameters/eeli.py

import warnings
from dataclasses import InitVar, dataclass, field
from typing import Final, Literal, get_args

import numpy as np

from eitprocessing.categories import check_category
from eitprocessing.datahandling.continuousdata import ContinuousData
from eitprocessing.datahandling.sequence import Sequence
from eitprocessing.datahandling.sparsedata import SparseData
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: 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,
            )

        _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)}"
            raise ValueError(msg)

    def compute_parameter(
        self,
        continuous_data: ContinuousData,
        sequence: Sequence | None = None,
        store: bool | None = None,
        result_label: str = "continuous_eelis",
    ) -> SparseData:
        """Compute the EELI for each breath in the impedance data.

        Example:
        ```
        >>> global_impedance = sequence.continuous_data["global_impedance_(raw)"]
        >>> eeli_data = EELI().compute_parameter(global_impedance)
        ```

        Args:
            continuous_data: a ContinuousData object containing impedance data.
            sequence: optional, Sequence 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_eelis'`.

        Returns:
            A SparseData object with the end-expiratory values of all breaths in the impedance data.

        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)

        check_category(continuous_data, "impedance", raise_=True)

        breaths = self.breath_detection.find_breaths(continuous_data)

        if not len(breaths):
            time = np.array([], dtype=float)
            values = np.array([], dtype=float)
        else:
            _, _, end_expiratory_times = zip(*breaths.values, strict=True)
            end_expiratory_indices = np.flatnonzero(np.isin(continuous_data.time, end_expiratory_times))
            time = [breath.end_time for breath in breaths.values if breath is not None]
            values = continuous_data.values[end_expiratory_indices]

        eeli_container = SparseData(
            label=result_label,
            name="End-expiratory lung impedance (EELI)",
            unit=None,
            category="impedance",
            time=time,
            description="End-expiratory lung impedance (EELI) determined on continuous data",
            parameters=self.breath_detection_kwargs,
            derived_from=[continuous_data],
            values=values,
        )
        if store:
            sequence.sparse_data.add(eeli_container)

        return eeli_container

1	import warnings	1✔
2	from dataclasses import InitVar, dataclass, field	1✔
3	from typing import Final, Literal, get_args	1✔
4
5	import numpy as np	1✔
6
7	from eitprocessing.categories import check_category	1✔
8	from eitprocessing.datahandling.continuousdata import ContinuousData	1✔
9	from eitprocessing.datahandling.sequence import Sequence	1✔
10	from eitprocessing.datahandling.sparsedata import SparseData	1✔
11	from eitprocessing.features.breath_detection import BreathDetection	1✔
12	from eitprocessing.parameters import ParameterCalculation	1✔
13
14	_SENTINEL_BREATH_DETECTION: Final = BreathDetection()	1✔
15
16
17	def _sentinel_breath_detection() -> BreathDetection:	1✔
18	# Returns a sential of a BreathDetection, which only exists to signal that the default value for breath_detection
19	# was used.
20	return _SENTINEL_BREATH_DETECTION	1✔
21
22
23	@dataclass	1✔
24	class EELI(ParameterCalculation):	1✔
25	"""Compute the end-expiratory lung impedance (EELI) per breath."""
26
27	breath_detection: BreathDetection = field(default_factory=_sentinel_breath_detection)	1✔
28	method: Literal["breath_detection"] = "breath_detection"	1✔
29	breath_detection_kwargs: InitVar[dict \| None] = None	1✔
30
31	def __post_init__(self, breath_detection_kwargs: dict \| None):	1✔
32	if breath_detection_kwargs is not None:	1✔
33	if self.breath_detection is not _SENTINEL_BREATH_DETECTION:	1✔
34	msg = (	1✔
35	"`breath_detection_kwargs` is deprecated, and can't be used at the same time as `breath_detection`."
36	)
37	raise TypeError(msg)	1✔
38
39	self.breath_detection = BreathDetection(**breath_detection_kwargs)	1✔
40	warnings.warn(	1✔
41	"`breath_detection_kwargs` is deprecated and will be removed soon. "
42	"Replace with `breath_detection=BreathDetection(**breath_detection_kwargs)`.",
43	DeprecationWarning,
44	)
45
46	_methods = get_args(EELI.__dataclass_fields__["method"].type)	1✔
47	if self.method not in _methods:	1✔
48	msg = f"Method {self.method} is not valid. Use any of {', '.join(_methods)}"	1✔
49	raise ValueError(msg)	1✔
50
51	def compute_parameter(	1✔
52	self,
53	continuous_data: ContinuousData,
54	sequence: Sequence \| None = None,
55	store: bool \| None = None,
56	result_label: str = "continuous_eelis",
57	) -> SparseData:
58	"""Compute the EELI for each breath in the impedance data.
59
60	Example:
61	```
62	>>> global_impedance = sequence.continuous_data["global_impedance_(raw)"]
63	>>> eeli_data = EELI().compute_parameter(global_impedance)
64	```
65
66	Args:
67	continuous_data: a ContinuousData object containing impedance data.
68	sequence: optional, Sequence to store the result in.
69	store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.
70	result_label: label of the returned SparseData object, defaults to `'continuous_eelis'`.
71
72	Returns:
73	A SparseData object with the end-expiratory values of all breaths in the impedance data.
74
75	Raises:
76	RuntimeError: If store is set to true but no sequence is provided.
77	ValueError: If the provided sequence is not an instance of the Sequence dataclass.
78	ValueError: If tiv_method is not one of 'inspiratory', 'expiratory', or 'mean'.
79	"""
80	if store is None and isinstance(sequence, Sequence):	1!
UNCOV 81	store = True	×
82
83	if store and sequence is None:	1!
UNCOV 84	msg = "Can't store the result if no Sequence is provided."	×
UNCOV 85	raise RuntimeError(msg)	×
86
87	if store and not isinstance(sequence, Sequence):	1!
UNCOV 88	msg = "To store the result a Sequence dataclass must be provided."	×
UNCOV 89	raise ValueError(msg)	×
90
91	check_category(continuous_data, "impedance", raise_=True)	1✔
92
93	breaths = self.breath_detection.find_breaths(continuous_data)	1✔
94
95	if not len(breaths):	1✔
96	time = np.array([], dtype=float)	1✔
97	values = np.array([], dtype=float)	1✔
98	else:
99	_, _, end_expiratory_times = zip(*breaths.values, strict=True)	1✔
100	end_expiratory_indices = np.flatnonzero(np.isin(continuous_data.time, end_expiratory_times))	1✔
101	time = [breath.end_time for breath in breaths.values if breath is not None]	1✔
102	values = continuous_data.values[end_expiratory_indices]	1✔
103
104	eeli_container = SparseData(	1✔
105	label=result_label,
106	name="End-expiratory lung impedance (EELI)",
107	unit=None,
108	category="impedance",
109	time=time,
110	description="End-expiratory lung impedance (EELI) determined on continuous data",
111	parameters=self.breath_detection_kwargs,
112	derived_from=[continuous_data],
113	values=values,
114	)
115	if store:	1!
UNCOV 116	sequence.sparse_data.add(eeli_container)	×
117
118	return eeli_container	1✔

EIT-ALIVE / eitprocessing / 15051278226

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