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

90.32

/eitprocessing/datahandling/eitdata.py

from __future__ import annotations

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

import numpy as np
from strenum import LowercaseStrEnum  # TODO: EOL 3.10: replace with native StrEnum

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(LowercaseStrEnum):
    """Enum indicating the vendor (manufacturer) of the source EIT device."""

    DRAEGER = auto()
    TIMPEL = auto()
    SENTEC = auto()
    DRAGER = DRAEGER
    DRÄGER = DRAEGER  # noqa: PLC2401
    SIMULATED = auto()

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