18004573301

Committed 25 Sep 2025 10:26AM UTC coverage: 84.795% (+0.02%) from 84.776%

Build # 18004573301

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.8.3 → 1.8.4

Run Details

747 of 960 branches covered (77.81%)

Branch coverage included in aggregate %.

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

5 existing lines in 1 file now uncovered.

2744 of 3157 relevant lines covered (86.92%)

0.87 hits per line

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

90.82

/eitprocessing/datahandling/eitdata.py

from __future__ import annotations

import warnings
from dataclasses import InitVar, dataclass, field
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any, TypeVar

import numpy as np

from eitprocessing.datahandling import DataContainer
from eitprocessing.datahandling.continuousdata import ContinuousData
from eitprocessing.datahandling.mixins.slicing import SelectByTime

if TYPE_CHECKING:
    from typing_extensions import Self


T = TypeVar("T", bound="EITData")


@dataclass(eq=False)
class EITData(DataContainer, SelectByTime):
    """Container for EIT impedance data.

    This class holds the pixel impedance from an EIT measurement, as well as metadata describing the measurement. The
    class is meant to hold data from (part of) a singular continuous measurement.

    This class can't be initialized directly. Instead, use `load_eit_data(<path>, vendor=<vendor>)` to load data from
    disk.

    Args:
        path: The path of list of paths of the source from which data was derived.
        nframes: Number of frames.
        time: The time of each frame (since start measurement).
        sample_frequency: The (average) frequency at which the frames are collected, in Hz.
        vendor: The vendor of the device the data was collected with.
        label: Computer readable label identifying this dataset.
        name: Human readable name for the data.
        pixel_impedance: Impedance values for each pixel at each frame.
    """  # TODO: fix docstring

    path: str | Path | list[Path | str] = field(compare=False, repr=False)
    nframes: int = field(repr=False)
    time: np.ndarray = field(repr=False)
    sample_frequency: float = field(metadata={"check_equivalence": True}, repr=False)
    vendor: Vendor = field(metadata={"check_equivalence": True}, repr=False)
    label: str | None = field(default=None, compare=False, metadata={"check_equivalence": True})
    description: str = field(default="", compare=False, repr=False)
    name: str | None = field(default=None, compare=False, repr=False)
    pixel_impedance: np.ndarray = field(repr=False, kw_only=True)
    suppress_simulated_warning: InitVar[bool] = False

    def __post_init__(self, suppress_simulated_warning: bool) -> None:
        if not self.label:
            self.label = f"{self.__class__.__name__}_{id(self)}"

        self.path = self.ensure_path_list(self.path)
        if len(self.path) == 1:
            self.path = self.path[0]

        self.name = self.name or self.label

        if (lv := len(self.pixel_impedance)) != (lt := len(self.time)):
            msg = f"The number of time points ({lt}) does not match the number of pixel impedance values ({lv})."
            raise ValueError(msg)

        if not suppress_simulated_warning and self.vendor == Vendor.SIMULATED:
            warnings.warn(
                "The simulated vendor is used for testing purposes. "
                "It is not a real vendor and should not be used in production code.",
                UserWarning,
                stacklevel=2,
            )

    @property
    def framerate(self) -> float:
        """Deprecated alias to `sample_frequency`."""
        warnings.warn(
            "The `framerate` attribute has been deprecated. Use `sample_frequency` instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        return self.sample_frequency

    @staticmethod
    def ensure_path_list(
        path: str | Path | list[str | Path],
    ) -> list[Path]:
        """Return the path or paths as a list.

        The path of any EITData object can be a single str/Path or a list of str/Path objects. This method returns a
        list of Path objects given either a str/Path or list of str/Paths.
        """
        if isinstance(path, list):
            return [Path(p) for p in path]
        return [Path(path)]

    def __add__(self: Self, other: Self) -> Self:
        return self.concatenate(other)

    def concatenate(self: Self, other: Self, newlabel: str | None = None) -> Self:  # noqa: D102, will be moved to mixin in future
        # Check that data can be concatenated
        self.isequivalent(other, raise_=True)
        if np.min(other.time) <= np.max(self.time):
            msg = f"Concatenation failed. Second dataset ({other.name}) may not start before first ({self.name}) ends."
            raise ValueError(msg)

        self_path = self.ensure_path_list(self.path)
        other_path = self.ensure_path_list(other.path)
        newlabel = newlabel or f"Merge of <{self.label}> and <{other.label}>"

        return self.__class__(
            vendor=self.vendor,
            path=[*self_path, *other_path],
            label=self.label,  # TODO: using newlabel leads to errors
            sample_frequency=self.sample_frequency,
            nframes=self.nframes + other.nframes,
            time=np.concatenate((self.time, other.time)),
            pixel_impedance=np.concatenate((self.pixel_impedance, other.pixel_impedance), axis=0),
        )

    def _sliced_copy(
        self,
        start_index: int,
        end_index: int,
        newlabel: str,  # noqa: ARG002
    ) -> Self:
        cls = self.__class__
        time = np.copy(self.time[start_index:end_index])
        nframes = len(time)

        pixel_impedance = np.copy(self.pixel_impedance[start_index:end_index, :, :])

        return cls(
            path=self.path,
            nframes=nframes,
            vendor=self.vendor,
            time=time,
            sample_frequency=self.sample_frequency,
            label=self.label,  # newlabel gives errors
            pixel_impedance=pixel_impedance,
        )

    def __len__(self):
        return self.pixel_impedance.shape[0]

    def get_summed_impedance(self, *, return_label: str | None = None, **return_kwargs) -> ContinuousData:
        """Return a ContinuousData-object with the same time axis and summed pixel values over time.

        Args:
            return_label: The label of the returned object; defaults to 'summed <label>' where '<label>' is the label of
            the current object.
            **return_kwargs: Keyword arguments for the creation of the returned object.
        """
        summed_impedance = np.nansum(self.pixel_impedance, axis=(1, 2))

        if return_label is None:
            return_label = f"summed {self.label}"

        return_kwargs_: dict[str, Any] = {
            "name": return_label,
            "unit": "AU",
            "category": "impedance",
            "sample_frequency": self.sample_frequency,
        } | return_kwargs

        return ContinuousData(label=return_label, time=np.copy(self.time), values=summed_impedance, **return_kwargs_)

    def calculate_global_impedance(self) -> np.ndarray:
        """Return the global impedance, i.e. the sum of all included pixels at each frame."""
        return np.nansum(self.pixel_impedance, axis=(1, 2))


class Vendor(Enum):
    """Enum indicating the vendor (manufacturer) of the source EIT device.

    The enum values are all lowercase strings. For some manufacturers, multiple ways of wrinting are provided mapping to
    the same value, to prevent confusion over conversion of special characters. The "simulated" vendor is provided to
    indicate simulated data.
    """

    DRAEGER = "draeger"
    """Dräger (PulmoVista V500)"""

    TIMPEL = "timpel"
    """Timpel (Enlight 2100)"""

    SENTEC = "sentec"
    """Sentec (Lumon)"""

    DRAGER = DRAEGER
    """Synonym of DRAEGER"""

    DRÄGER = DRAEGER  # noqa: PIE796, PLC2401
    """Synonym of DRAEGER"""

    SIMULATED = "simulated"
    """Simulated data"""

1	from __future__ import annotations	1✔
2
3	import warnings	1✔
4	from dataclasses import InitVar, dataclass, field	1✔
5	from enum import Enum	1✔
6	from pathlib import Path	1✔
7	from typing import TYPE_CHECKING, Any, TypeVar	1✔
8
9	import numpy as np	1✔
10
11	from eitprocessing.datahandling import DataContainer	1✔
12	from eitprocessing.datahandling.continuousdata import ContinuousData	1✔
13	from eitprocessing.datahandling.mixins.slicing import SelectByTime	1✔
14
15	if TYPE_CHECKING:
16	from typing_extensions import Self
17
18
19	T = TypeVar("T", bound="EITData")	1✔
20
21
22	@dataclass(eq=False)	1✔
23	class EITData(DataContainer, SelectByTime):	1✔
24	"""Container for EIT impedance data.
25
26	This class holds the pixel impedance from an EIT measurement, as well as metadata describing the measurement. The
27	class is meant to hold data from (part of) a singular continuous measurement.
28
29	This class can't be initialized directly. Instead, use `load_eit_data(<path>, vendor=<vendor>)` to load data from
30	disk.
31
32	Args:
33	path: The path of list of paths of the source from which data was derived.
34	nframes: Number of frames.
35	time: The time of each frame (since start measurement).
36	sample_frequency: The (average) frequency at which the frames are collected, in Hz.
37	vendor: The vendor of the device the data was collected with.
38	label: Computer readable label identifying this dataset.
39	name: Human readable name for the data.
40	pixel_impedance: Impedance values for each pixel at each frame.
41	""" # TODO: fix docstring
42
43	path: str \| Path \| list[Path \| str] = field(compare=False, repr=False)	1✔
44	nframes: int = field(repr=False)	1✔
45	time: np.ndarray = field(repr=False)	1✔
46	sample_frequency: float = field(metadata={"check_equivalence": True}, repr=False)	1✔
47	vendor: Vendor = field(metadata={"check_equivalence": True}, repr=False)	1✔
48	label: str \| None = field(default=None, compare=False, metadata={"check_equivalence": True})	1✔
49	description: str = field(default="", compare=False, repr=False)	1✔
50	name: str \| None = field(default=None, compare=False, repr=False)	1✔
51	pixel_impedance: np.ndarray = field(repr=False, kw_only=True)	1✔
52	suppress_simulated_warning: InitVar[bool] = False	1✔
53
54	def __post_init__(self, suppress_simulated_warning: bool) -> None:	1✔
55	if not self.label:	1✔
56	self.label = f"{self.__class__.__name__}_{id(self)}"	1✔
57
58	self.path = self.ensure_path_list(self.path)	1✔
59	if len(self.path) == 1:	1✔
60	self.path = self.path[0]	1✔
61
62	self.name = self.name or self.label	1✔
63
64	if (lv := len(self.pixel_impedance)) != (lt := len(self.time)):	1!
UNCOV 65	msg = f"The number of time points ({lt}) does not match the number of pixel impedance values ({lv})."	×
66	raise ValueError(msg)	×
67
68	if not suppress_simulated_warning and self.vendor == Vendor.SIMULATED:	1!
UNCOV 69	warnings.warn(	×
70	"The simulated vendor is used for testing purposes. "
71	"It is not a real vendor and should not be used in production code.",
72	UserWarning,
73	stacklevel=2,
74	)
75
76	@property	1✔
77	def framerate(self) -> float:	1✔
78	"""Deprecated alias to `sample_frequency`."""
UNCOV 79	warnings.warn(	×
80	"The `framerate` attribute has been deprecated. Use `sample_frequency` instead.",
81	DeprecationWarning,
82	stacklevel=2,
83	)
UNCOV 84	return self.sample_frequency	×
85
86	@staticmethod	1✔
87	def ensure_path_list(	1✔
88	path: str \| Path \| list[str \| Path],
89	) -> list[Path]:
90	"""Return the path or paths as a list.
91
92	The path of any EITData object can be a single str/Path or a list of str/Path objects. This method returns a
93	list of Path objects given either a str/Path or list of str/Paths.
94	"""
95	if isinstance(path, list):	1✔
96	return [Path(p) for p in path]	1✔
97	return [Path(path)]	1✔
98
99	def __add__(self: Self, other: Self) -> Self:	1✔
UNCOV 100	return self.concatenate(other)	×
101
102	def concatenate(self: Self, other: Self, newlabel: str \| None = None) -> Self: # noqa: D102, will be moved to mixin in future	1✔
103	# Check that data can be concatenated
104	self.isequivalent(other, raise_=True)	1✔
105	if np.min(other.time) <= np.max(self.time):	1✔
106	msg = f"Concatenation failed. Second dataset ({other.name}) may not start before first ({self.name}) ends."	1✔
107	raise ValueError(msg)	1✔
108
109	self_path = self.ensure_path_list(self.path)	1✔
110	other_path = self.ensure_path_list(other.path)	1✔
111	newlabel = newlabel or f"Merge of <{self.label}> and <{other.label}>"	1✔
112
113	return self.__class__(	1✔
114	vendor=self.vendor,
115	path=[self_path, other_path],
116	label=self.label, # TODO: using newlabel leads to errors
117	sample_frequency=self.sample_frequency,
118	nframes=self.nframes + other.nframes,
119	time=np.concatenate((self.time, other.time)),
120	pixel_impedance=np.concatenate((self.pixel_impedance, other.pixel_impedance), axis=0),
121	)
122
123	def _sliced_copy(	1✔
124	self,
125	start_index: int,
126	end_index: int,
127	newlabel: str, # noqa: ARG002
128	) -> Self:
129	cls = self.__class__	1✔
130	time = np.copy(self.time[start_index:end_index])	1✔
131	nframes = len(time)	1✔
132
133	pixel_impedance = np.copy(self.pixel_impedance[start_index:end_index, :, :])	1✔
134
135	return cls(	1✔
136	path=self.path,
137	nframes=nframes,
138	vendor=self.vendor,
139	time=time,
140	sample_frequency=self.sample_frequency,
141	label=self.label, # newlabel gives errors
142	pixel_impedance=pixel_impedance,
143	)
144
145	def __len__(self):	1✔
146	return self.pixel_impedance.shape[0]	1✔
147
148	def get_summed_impedance(self, , return_label: str \| None = None, *return_kwargs) -> ContinuousData:	1✔
149	"""Return a ContinuousData-object with the same time axis and summed pixel values over time.
150
151	Args:
152	return_label: The label of the returned object; defaults to 'summed <label>' where '<label>' is the label of
153	the current object.
154	**return_kwargs: Keyword arguments for the creation of the returned object.
155	"""
156	summed_impedance = np.nansum(self.pixel_impedance, axis=(1, 2))	1✔
157
158	if return_label is None:	1!
159	return_label = f"summed {self.label}"	1✔
160
161	return_kwargs_: dict[str, Any] = {	1✔
162	"name": return_label,
163	"unit": "AU",
164	"category": "impedance",
165	"sample_frequency": self.sample_frequency,
166	} \| return_kwargs
167
168	return ContinuousData(label=return_label, time=np.copy(self.time), values=summed_impedance, **return_kwargs_)	1✔
169
170	def calculate_global_impedance(self) -> np.ndarray:	1✔
171	"""Return the global impedance, i.e. the sum of all included pixels at each frame."""
172	return np.nansum(self.pixel_impedance, axis=(1, 2))	1✔
173
174
175	class Vendor(Enum):	1✔
176	"""Enum indicating the vendor (manufacturer) of the source EIT device.
177
178	The enum values are all lowercase strings. For some manufacturers, multiple ways of wrinting are provided mapping to
179	the same value, to prevent confusion over conversion of special characters. The "simulated" vendor is provided to
180	indicate simulated data.
181	"""
182
183	DRAEGER = "draeger"	1✔
184	"""Dräger (PulmoVista V500)"""	1✔
185
186	TIMPEL = "timpel"	1✔
187	"""Timpel (Enlight 2100)"""	1✔
188
189	SENTEC = "sentec"	1✔
190	"""Sentec (Lumon)"""	1✔
191
192	DRAGER = DRAEGER	1✔
193	"""Synonym of DRAEGER"""	1✔
194
195	DRÄGER = DRAEGER # noqa: PIE796, PLC2401	1✔
196	"""Synonym of DRAEGER"""	1✔
197
198	SIMULATED = "simulated"	1✔
199	"""Simulated data"""	1✔

EIT-ALIVE / eitprocessing / 18004573301

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