17213080321

Committed 25 Aug 2025 03:19PM UTC coverage: 84.761% (+2.0%) from 82.774%

Build # 17213080321

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.7.3 → 1.8.0

Run Details

745 of 958 branches covered (77.77%)

Branch coverage included in aggregate %.

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

37 existing lines in 9 files now uncovered.

2737 of 3150 relevant lines covered (86.89%)

0.87 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,
                stacklevel=2,
            )

        _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	stacklevel=2,
45	)
46
47	_methods = get_args(EELI.__dataclass_fields__["method"].type)	1✔
48	if self.method not in _methods:	1✔
49	msg = f"Method {self.method} is not valid. Use any of {', '.join(_methods)}"	1✔
50	raise ValueError(msg)	1✔
51
52	def compute_parameter(	1✔
53	self,
54	continuous_data: ContinuousData,
55	sequence: Sequence \| None = None,
56	store: bool \| None = None,
57	result_label: str = "continuous_eelis",
58	) -> SparseData:
59	"""Compute the EELI for each breath in the impedance data.
60
61	Example:
62	```
63	>>> global_impedance = sequence.continuous_data["global_impedance_(raw)"]
64	>>> eeli_data = EELI().compute_parameter(global_impedance)
65	```
66
67	Args:
68	continuous_data: a ContinuousData object containing impedance data.
69	sequence: optional, Sequence to store the result in.
70	store: whether to store the result in the sequence, defaults to `True` if a Sequence if provided.
71	result_label: label of the returned SparseData object, defaults to `'continuous_eelis'`.
72
73	Returns:
74	A SparseData object with the end-expiratory values of all breaths in the impedance data.
75
76	Raises:
77	RuntimeError: If store is set to true but no sequence is provided.
78	ValueError: If the provided sequence is not an instance of the Sequence dataclass.
79	ValueError: If tiv_method is not one of 'inspiratory', 'expiratory', or 'mean'.
80	"""
81	if store is None and isinstance(sequence, Sequence):	1!
UNCOV 82	store = True	×
83
84	if store and sequence is None:	1!
85	msg = "Can't store the result if no Sequence is provided."	×
UNCOV 86	raise RuntimeError(msg)	×
87
88	if store and not isinstance(sequence, Sequence):	1!
89	msg = "To store the result a Sequence dataclass must be provided."	×
UNCOV 90	raise ValueError(msg)	×
91
92	check_category(continuous_data, "impedance", raise_=True)	1✔
93
94	breaths = self.breath_detection.find_breaths(continuous_data)	1✔
95
96	if not len(breaths):	1✔
97	time = np.array([], dtype=float)	1✔
98	values = np.array([], dtype=float)	1✔
99	else:
100	_, _, end_expiratory_times = zip(*breaths.values, strict=True)	1✔
101	end_expiratory_indices = np.flatnonzero(np.isin(continuous_data.time, end_expiratory_times))	1✔
102	time = [breath.end_time for breath in breaths.values if breath is not None]	1✔
103	values = continuous_data.values[end_expiratory_indices]	1✔
104
105	eeli_container = SparseData(	1✔
106	label=result_label,
107	name="End-expiratory lung impedance (EELI)",
108	unit=None,
109	category="impedance",
110	time=time,
111	description="End-expiratory lung impedance (EELI) determined on continuous data",
112	parameters=self.breath_detection_kwargs,
113	derived_from=[continuous_data],
114	values=values,
115	)
116	if store:	1!
UNCOV 117	sequence.sparse_data.add(eeli_container)	×
118
119	return eeli_container	1✔

EIT-ALIVE / eitprocessing / 17213080321

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