19311527649

Committed 12 Nov 2025 08:51PM UTC coverage: 86.671% (+1.9%) from 84.795%

Build # 19311527649

Build Type

Pull #458

github

Committed by

web-flow

Commit Message

Merge pull request #455 from EIT-ALIVE/tests/330-open-source-test-data

Migrate to open source test data and fix revealed issues

Pull Request Pull Request #458: Release 1.8.5

Run Details

781 of 976 branches covered (80.02%)

Branch coverage included in aggregate %.

73 of 82 new or added lines in 4 files covered. (89.02%)

9 existing lines in 2 files now uncovered.

2828 of 3188 relevant lines covered (88.71%)

0.89 hits per line

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

88.57

/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
        old_sample_frequency = self.sample_frequency
        self.sample_frequency = float(self.sample_frequency)
        if self.sample_frequency != old_sample_frequency:
            msg = (
                "Sample frequency could not be correctly converted from "
                f"{old_sample_frequency} ({type(old_sample_frequency)}) to "
                f"{self.sample_frequency:.1f} (float)."
            )
            raise TypeError(msg)

        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	old_sample_frequency = self.sample_frequency	1✔
64	self.sample_frequency = float(self.sample_frequency)	1✔
65	if self.sample_frequency != old_sample_frequency:	1!
NEW 66	msg = (	×
67	"Sample frequency could not be correctly converted from "
68	f"{old_sample_frequency} ({type(old_sample_frequency)}) to "
69	f"{self.sample_frequency:.1f} (float)."
70	)
NEW 71	raise TypeError(msg)	×
72
73	if (lv := len(self.pixel_impedance)) != (lt := len(self.time)):	1!
74	msg = f"The number of time points ({lt}) does not match the number of pixel impedance values ({lv})."	×
75	raise ValueError(msg)	×
76
77	if not suppress_simulated_warning and self.vendor == Vendor.SIMULATED:	1!
78	warnings.warn(	×
79	"The simulated vendor is used for testing purposes. "
80	"It is not a real vendor and should not be used in production code.",
81	UserWarning,
82	stacklevel=2,
83	)
84
85	@property	1✔
86	def framerate(self) -> float:	1✔
87	"""Deprecated alias to `sample_frequency`."""
88	warnings.warn(	×
89	"The `framerate` attribute has been deprecated. Use `sample_frequency` instead.",
90	DeprecationWarning,
91	stacklevel=2,
92	)
93	return self.sample_frequency	×
94
95	@staticmethod	1✔
96	def ensure_path_list(	1✔
97	path: str \| Path \| list[str \| Path],
98	) -> list[Path]:
99	"""Return the path or paths as a list.
100
101	The path of any EITData object can be a single str/Path or a list of str/Path objects. This method returns a
102	list of Path objects given either a str/Path or list of str/Paths.
103	"""
104	if isinstance(path, list):	1✔
105	return [Path(p) for p in path]	1✔
106	return [Path(path)]	1✔
107
108	def __add__(self: Self, other: Self) -> Self:	1✔
109	return self.concatenate(other)	×
110
111	def concatenate(self: Self, other: Self, newlabel: str \| None = None) -> Self: # noqa: D102, will be moved to mixin in future	1✔
112	# Check that data can be concatenated
113	self.isequivalent(other, raise_=True)	1✔
114	if np.min(other.time) <= np.max(self.time):	1✔
115	msg = f"Concatenation failed. Second dataset ({other.name}) may not start before first ({self.name}) ends."	1✔
116	raise ValueError(msg)	1✔
117
118	self_path = self.ensure_path_list(self.path)	1✔
119	other_path = self.ensure_path_list(other.path)	1✔
120	newlabel = newlabel or f"Merge of <{self.label}> and <{other.label}>"	1✔
121
122	return self.__class__(	1✔
123	vendor=self.vendor,
124	path=[self_path, other_path],
125	label=self.label, # TODO: using newlabel leads to errors
126	sample_frequency=self.sample_frequency,
127	nframes=self.nframes + other.nframes,
128	time=np.concatenate((self.time, other.time)),
129	pixel_impedance=np.concatenate((self.pixel_impedance, other.pixel_impedance), axis=0),
130	)
131
132	def _sliced_copy(	1✔
133	self,
134	start_index: int,
135	end_index: int,
136	newlabel: str, # noqa: ARG002
137	) -> Self:
138	cls = self.__class__	1✔
139	time = np.copy(self.time[start_index:end_index])	1✔
140	nframes = len(time)	1✔
141
142	pixel_impedance = np.copy(self.pixel_impedance[start_index:end_index, :, :])	1✔
143
144	return cls(	1✔
145	path=self.path,
146	nframes=nframes,
147	vendor=self.vendor,
148	time=time,
149	sample_frequency=self.sample_frequency,
150	label=self.label, # newlabel gives errors
151	pixel_impedance=pixel_impedance,
152	)
153
154	def __len__(self):	1✔
155	return self.pixel_impedance.shape[0]	1✔
156
157	def get_summed_impedance(self, , return_label: str \| None = None, *return_kwargs) -> ContinuousData:	1✔
158	"""Return a ContinuousData-object with the same time axis and summed pixel values over time.
159
160	Args:
161	return_label: The label of the returned object; defaults to 'summed <label>' where '<label>' is the label of
162	the current object.
163	**return_kwargs: Keyword arguments for the creation of the returned object.
164	"""
165	summed_impedance = np.nansum(self.pixel_impedance, axis=(1, 2))	1✔
166
167	if return_label is None:	1!
168	return_label = f"summed {self.label}"	1✔
169
170	return_kwargs_: dict[str, Any] = {	1✔
171	"name": return_label,
172	"unit": "AU",
173	"category": "impedance",
174	"sample_frequency": self.sample_frequency,
175	} \| return_kwargs
176
177	return ContinuousData(label=return_label, time=np.copy(self.time), values=summed_impedance, **return_kwargs_)	1✔
178
179	def calculate_global_impedance(self) -> np.ndarray:	1✔
180	"""Return the global impedance, i.e. the sum of all included pixels at each frame."""
181	return np.nansum(self.pixel_impedance, axis=(1, 2))	1✔
182
183
184	class Vendor(Enum):	1✔
185	"""Enum indicating the vendor (manufacturer) of the source EIT device.
186
187	The enum values are all lowercase strings. For some manufacturers, multiple ways of wrinting are provided mapping to
188	the same value, to prevent confusion over conversion of special characters. The "simulated" vendor is provided to
189	indicate simulated data.
190	"""
191
192	DRAEGER = "draeger"	1✔
193	"""Dräger (PulmoVista V500)"""	1✔
194
195	TIMPEL = "timpel"	1✔
196	"""Timpel (Enlight 2100)"""	1✔
197
198	SENTEC = "sentec"	1✔
199	"""Sentec (Lumon)"""	1✔
200
201	DRAGER = DRAEGER	1✔
202	"""Synonym of DRAEGER"""	1✔
203
204	DRÄGER = DRAEGER # noqa: PIE796, PLC2401	1✔
205	"""Synonym of DRAEGER"""	1✔
206
207	SIMULATED = "simulated"	1✔
208	"""Simulated data"""	1✔

EIT-ALIVE / eitprocessing / 19311527649

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