12120635114

Committed 02 Dec 2024 01:44PM UTC coverage: 81.489% (-0.6%) from 82.129%

Build # 12120635114

Build Type

push

github

Committed by

actions-user

Commit Message

Bump version: 1.4.6 → 1.4.7

Run Details

344 of 468 branches covered (73.5%)

Branch coverage included in aggregate %.

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

36 existing lines in 5 files now uncovered.

1364 of 1628 relevant lines covered (83.78%)

0.84 hits per line

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

81.48

/eitprocessing/parameters/eeli.py

from dataclasses import dataclass, field
from typing import 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


@dataclass
class EELI(ParameterCalculation):
    """Compute the end-expiratory lung impedance (EELI) per breath."""

    method: Literal["breath_detection"] = "breath_detection"
    breath_detection_kwargs: dict = field(default_factory=dict)

    def __post_init__(self):
        _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)

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

EIT-ALIVE / eitprocessing / 12120635114

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