15051278226

Committed 15 May 2025 05:21PM UTC coverage: 83.253% (-0.6%) from 83.838%

Build # 15051278226

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.7.0 → 1.7.1

Run Details

391 of 518 branches covered (75.48%)

Branch coverage included in aggregate %.

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

31 existing lines in 4 files now uncovered.

1503 of 1757 relevant lines covered (85.54%)

0.86 hits per line

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

82.83

/eitprocessing/datahandling/sequence.py

from __future__ import annotations

import itertools
import sys
from dataclasses import MISSING, dataclass, field
from typing import TYPE_CHECKING, Any, TypeVar, overload

from eitprocessing.datahandling.continuousdata import ContinuousData
from eitprocessing.datahandling.datacollection import DataCollection
from eitprocessing.datahandling.eitdata import EITData
from eitprocessing.datahandling.intervaldata import IntervalData
from eitprocessing.datahandling.mixins.equality import Equivalence
from eitprocessing.datahandling.mixins.slicing import SelectByTime
from eitprocessing.datahandling.sparsedata import SparseData

if TYPE_CHECKING:
    from collections.abc import Iterator

    import numpy as np
    from typing_extensions import Self

    from eitprocessing.parameters import DataContainer

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


@dataclass(eq=False)
class Sequence(Equivalence, SelectByTime):
    """Sequence of timepoints containing respiratory data.

    A Sequence object is a representation of data points over time. These data can consist of any combination of EIT
    frames (`EITData`), waveform data (`ContinuousData`) from different sources, or individual events (`SparseData`)
    occurring at any given timepoint. A Sequence can consist of an entire measurement, a section of a measurement, a
    single breath, or even a portion of a breath. A Sequence can consist of multiple sets of each type of data from the
    same time-points or can be a single measurement from just one source.

    A Sequence can be split up into separate sections of a measurement or multiple (similar) Sequence objects can be
    merged together to form a single Sequence.

    Args:
        label: Computer readable naming of the instance.
        name: Human readable naming of the instance.
        description: Human readable extended description of the data.
        eit_data: Collection of one or more sets of EIT data frames.
        continuous_data: Collection of one or more sets of continuous data points.
        sparse_data: Collection of one or more sets of individual data points.
    """  # TODO: check that docstring is up to date

    label: str | None = field(default=None, compare=False)
    name: str | None = field(default=None, compare=False, repr=False)
    description: str = field(default="", compare=False, repr=False)
    eit_data: DataCollection[EITData] = field(default_factory=lambda: DataCollection(EITData), repr=False)
    continuous_data: DataCollection[ContinuousData] = field(
        default_factory=lambda: DataCollection(ContinuousData),
        repr=False,
    )
    sparse_data: DataCollection[SparseData] = field(default_factory=lambda: DataCollection(SparseData), repr=False)
    interval_data: DataCollection[IntervalData] = field(
        default_factory=lambda: DataCollection(IntervalData),
        repr=False,
    )

    def __post_init__(self):
        if not self.label:
            self.label = f"Sequence_{id(self)}"
        self.name = self.name or self.label

    @property
    def time(self) -> np.ndarray:
        """Time axis from either EITData or ContinuousData."""
        if len(self.eit_data):
            return self.eit_data["raw"].time
        if len(self.continuous_data):
            return next(iter(self.continuous_data.values())).time

        msg = "Sequence has no timed data"
        raise AttributeError(msg)

    def __len__(self):
        return len(self.time)

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

    @classmethod  # TODO: why is this a class method? In other cases it's instance method
    def concatenate(
        cls,
        a: Sequence,
        b: Sequence,
        newlabel: str | None = None,
    ) -> Sequence:
        """Create a merge of two Sequence objects."""
        # TODO: rewrite

        concat_eit = a.eit_data.concatenate(b.eit_data)
        concat_continuous = a.continuous_data.concatenate(b.continuous_data)
        concat_sparse = a.sparse_data.concatenate(b.sparse_data)
        concat_interval = a.interval_data.concatenate(b.interval_data)

        newlabel = newlabel or a.label
        # TODO: add concatenation of other attached objects

        return a.__class__(
            eit_data=concat_eit,
            continuous_data=concat_continuous,
            sparse_data=concat_sparse,
            interval_data=concat_interval,
            label=newlabel,
        )

    def _sliced_copy(self, start_index: int, end_index: int, newlabel: str) -> Self:  # noqa: ARG002
        if start_index >= len(self.time):
            msg = "start_index larger than length of time axis"
            raise ValueError(msg)
        time = self.time[start_index:end_index]

        sliced_eit = DataCollection(EITData)
        for value in self.eit_data.values():
            sliced_eit.add(value[start_index:end_index])

        sliced_continuous = DataCollection(ContinuousData)
        for value in self.continuous_data.values():
            sliced_continuous.add(value[start_index:end_index])

        sliced_sparse = DataCollection(SparseData)
        for value in self.sparse_data.values():
            sliced_sparse.add(value.t[time[0] : time[-1]])

        sliced_interval = DataCollection(IntervalData)
        for value in self.interval_data.values():
            sliced_interval.add(value.t[time[0] : time[-1]])

        return self.__class__(
            label=self.label,  # newlabel gives errors
            name=f"Sliced copy of <{self.name}>",
            description=f"Sliced copy of <{self.description}>",
            eit_data=sliced_eit,
            continuous_data=sliced_continuous,
            sparse_data=sliced_sparse,
            interval_data=sliced_interval,
        )

    def select_by_time(
        self,
        start_time: float | None = None,
        end_time: float | None = None,
        start_inclusive: bool = True,
        end_inclusive: bool = False,
        label: str | None = None,
        name: str | None = None,
        description: str = "",
    ) -> Self:
        """Return a sliced version of the Sequence.

        See `SelectByTime.select_by_time()`.
        """
        if not label:
            label = f"copy_of_<{self.label}>"
        if not name:
            f"Sliced copy of <{self.name}>"

        return self.__class__(
            label=label,
            name=name,
            description=description,
            # perform select_by_time() on all four data types
            **{
                key: getattr(self, key).select_by_time(
                    start_time=start_time,
                    end_time=end_time,
                    start_inclusive=start_inclusive,
                    end_inclusive=end_inclusive,
                )
                for key in ("eit_data", "continuous_data", "sparse_data", "interval_data")
            },
        )

    @property
    def data(self) -> _DataAccess:
        """Shortcut access to data stored in collections inside a sequence.

        This allows all data objects stored in a collection inside a sequence to be accessed.
        Instead of `sequence.continuous_data["global_impedance"]` you can use
        `sequence.data["global_impedance"]`. This works for getting (`sequence.data["label"]` or
        `sequence.data.get("label")`) and adding data (`sequence.data["label"] = obj` or
        `sequence.data.add(obj)`).

        Other dict-like behaviour is also supported:

        - `label in sequence.data` to check whether an object with a label exists;
        - `del sequence.data[label]` to remove an object from the sequence based on the label;
        - `for label in sequence.data` to iterate over the labels;
        - `sequence.data.items()` to retrieve a list of (label, object) pairs, especially useful for iteration;
        - `sequence.data.labels()` or `sequence.data.keys()` to get a list of data labels;
        - `sequence.data.objects()` or `sequence.data.values()` to get a list of data objects.

        This interface only works if the labels are unique among the data collections. An attempt
        to add a data object with an exiting label will result in a KeyError.
        """
        return _DataAccess(self)


@dataclass
class _DataAccess:
    _sequence: Sequence

    def __post_init__(self):
        for a, b in itertools.combinations(self._collections, 2):
            if duplicates := set(a) & set(b):
                msg = f"Duplicate labels ({', '.join(sorted(duplicates))}) found in {a} and {b}."
                exc = KeyError(msg)
                if sys.version_info >= (3, 11):
                    exc.add_note(
                        "You can't use the `data` interface with duplicate labels. "
                        "Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
                        "`interval_data`) instead."
                    )
                raise exc

    @property
    def _collections(self) -> tuple[DataCollection, ...]:
        return (
            self._sequence.continuous_data,
            self._sequence.interval_data,
            self._sequence.sparse_data,
            self._sequence.eit_data,
        )

    @overload
    def get(self, label: str) -> DataContainer: ...

    @overload
    def get(self, label: str, default: T) -> DataContainer | T: ...

    def get(self, label: str, default: object = MISSING) -> DataContainer | object:
        """Get a DataContainer object by label.

        Example:
        ```
        if filtered_data := sequence.data.get("filtered data", None):
            print(filtered_data.values.mean())
        else:
            print("No filtered data was found.")

        ```

        Args:
            label (str): label of the object to retrieve.
            default (optional): a default value that is returned if the object is not found.
                Defaults to MISSING.

        Raises:
            KeyError: if the object is not found, and no default was set.

        Returns:
            DataContainer: the requested DataContainer.
        """
        for collection in self._collections:
            if label in collection:
                return collection[label]

        if default is not MISSING:
            return default

        msg = f"No object with label {label} was found."
        raise KeyError(msg)

    def __getitem__(self, key: str) -> DataContainer:
        return self.get(key)

    def add(self, *obj: DataContainer) -> None:
        """Add a DataContainer object to the sequence.

        Adds the object to the appropriate data collection. The label of the object must be unique
        among all data collections, otherwise a KeyError is raised.

        Args:
            obj (DataContainer): the object to add to the Sequence.

        Raises:
            KeyError: if the label of the object already exists in any of the data collections.
        """
        for object_ in obj:
            if self.get(object_.label, None):
                msg = f"An object with the label {object_.label} already exists in this sequence."
                exc = KeyError(msg)
                if sys.version_info >= (3, 11):
                    exc.add_note(
                        "You can't add an object with the same label through the `data` interface. "
                        "Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
                        "`interval_data`) instead."
                    )
                raise exc

            match object_:
                case ContinuousData():
                    self._sequence.continuous_data.add(object_)
                case IntervalData():
                    self._sequence.interval_data.add(object_)
                case SparseData():
                    self._sequence.sparse_data.add(object_)
                case EITData():
                    self._sequence.eit_data.add(object_)

    def __setitem__(self, label: str, obj: DataContainer):
        if obj.label != label:
            msg = f"Label {label} does not match object label {obj.label}."
            raise KeyError(msg)
        return self.add(obj)

    def __contains__(self, label: str) -> bool:
        return any(label in container for container in self._collections)

    def __delitem__(self, label: str) -> None:
        for container in self._collections:
            if label in container:
                del container[label]
                return

        msg = f"Object with label {label} was not found."
        raise KeyError(msg)

    def __iter__(self) -> Iterator[str]:
        return itertools.chain(*[collection.keys() for collection in self._collections])

    def items(self) -> list[tuple[str, DataContainer]]:
        """Return all data items (`(label, object)` pairs)."""
        return list(itertools.chain(*[collection.items() for collection in self._collections]))

    def keys(self) -> list[str]:
        """Return a list of all labels."""
        return list(self.__iter__())

    labels = keys

    def values(self) -> list[DataContainer]:
        """Return all data objects."""
        return list(itertools.chain(*[collection.values() for collection in self._collections]))

    objects = values

1	from __future__ import annotations	1✔
2
3	import itertools	1✔
4	import sys	1✔
5	from dataclasses import MISSING, dataclass, field	1✔
6	from typing import TYPE_CHECKING, Any, TypeVar, overload	1✔
7
8	from eitprocessing.datahandling.continuousdata import ContinuousData	1✔
9	from eitprocessing.datahandling.datacollection import DataCollection	1✔
10	from eitprocessing.datahandling.eitdata import EITData	1✔
11	from eitprocessing.datahandling.intervaldata import IntervalData	1✔
12	from eitprocessing.datahandling.mixins.equality import Equivalence	1✔
13	from eitprocessing.datahandling.mixins.slicing import SelectByTime	1✔
14	from eitprocessing.datahandling.sparsedata import SparseData	1✔
15
16	if TYPE_CHECKING:
17	from collections.abc import Iterator
18
19	import numpy as np
20	from typing_extensions import Self
21
22	from eitprocessing.parameters import DataContainer
23
24	T = TypeVar("T", bound=Any)	1✔
25
26
27	@dataclass(eq=False)	1✔
28	class Sequence(Equivalence, SelectByTime):	1✔
29	"""Sequence of timepoints containing respiratory data.
30
31	A Sequence object is a representation of data points over time. These data can consist of any combination of EIT
32	frames (`EITData`), waveform data (`ContinuousData`) from different sources, or individual events (`SparseData`)
33	occurring at any given timepoint. A Sequence can consist of an entire measurement, a section of a measurement, a
34	single breath, or even a portion of a breath. A Sequence can consist of multiple sets of each type of data from the
35	same time-points or can be a single measurement from just one source.
36
37	A Sequence can be split up into separate sections of a measurement or multiple (similar) Sequence objects can be
38	merged together to form a single Sequence.
39
40	Args:
41	label: Computer readable naming of the instance.
42	name: Human readable naming of the instance.
43	description: Human readable extended description of the data.
44	eit_data: Collection of one or more sets of EIT data frames.
45	continuous_data: Collection of one or more sets of continuous data points.
46	sparse_data: Collection of one or more sets of individual data points.
47	""" # TODO: check that docstring is up to date
48
49	label: str \| None = field(default=None, compare=False)	1✔
50	name: str \| None = field(default=None, compare=False, repr=False)	1✔
51	description: str = field(default="", compare=False, repr=False)	1✔
52	eit_data: DataCollection[EITData] = field(default_factory=lambda: DataCollection(EITData), repr=False)	1✔
53	continuous_data: DataCollection[ContinuousData] = field(	1✔
54	default_factory=lambda: DataCollection(ContinuousData),
55	repr=False,
56	)
57	sparse_data: DataCollection[SparseData] = field(default_factory=lambda: DataCollection(SparseData), repr=False)	1✔
58	interval_data: DataCollection[IntervalData] = field(	1✔
59	default_factory=lambda: DataCollection(IntervalData),
60	repr=False,
61	)
62
63	def __post_init__(self):	1✔
64	if not self.label:	1✔
65	self.label = f"Sequence_{id(self)}"	1✔
66	self.name = self.name or self.label	1✔
67
68	@property	1✔
69	def time(self) -> np.ndarray:	1✔
70	"""Time axis from either EITData or ContinuousData."""
71	if len(self.eit_data):	1✔
72	return self.eit_data["raw"].time	1✔
73	if len(self.continuous_data):	1!
74	return next(iter(self.continuous_data.values())).time	1✔
75
76	msg = "Sequence has no timed data"	×
77	raise AttributeError(msg)	×
78
79	def __len__(self):	1✔
80	return len(self.time)	1✔
81
82	def __add__(self, other: Sequence) -> Sequence:	1✔
83	return self.concatenate(self, other)	1✔
84
85	@classmethod # TODO: why is this a class method? In other cases it's instance method	1✔
86	def concatenate(	1✔
87	cls,
88	a: Sequence,
89	b: Sequence,
90	newlabel: str \| None = None,
91	) -> Sequence:
92	"""Create a merge of two Sequence objects."""
93	# TODO: rewrite
94
95	concat_eit = a.eit_data.concatenate(b.eit_data)	1✔
96	concat_continuous = a.continuous_data.concatenate(b.continuous_data)	1✔
97	concat_sparse = a.sparse_data.concatenate(b.sparse_data)	1✔
98	concat_interval = a.interval_data.concatenate(b.interval_data)	1✔
99
100	newlabel = newlabel or a.label	1✔
101	# TODO: add concatenation of other attached objects
102
103	return a.__class__(	1✔
104	eit_data=concat_eit,
105	continuous_data=concat_continuous,
106	sparse_data=concat_sparse,
107	interval_data=concat_interval,
108	label=newlabel,
109	)
110
111	def _sliced_copy(self, start_index: int, end_index: int, newlabel: str) -> Self: # noqa: ARG002	1✔
112	if start_index >= len(self.time):	1!
113	msg = "start_index larger than length of time axis"	×
114	raise ValueError(msg)	×
115	time = self.time[start_index:end_index]	1✔
116
117	sliced_eit = DataCollection(EITData)	1✔
118	for value in self.eit_data.values():	1✔
119	sliced_eit.add(value[start_index:end_index])	1✔
120
121	sliced_continuous = DataCollection(ContinuousData)	1✔
122	for value in self.continuous_data.values():	1✔
123	sliced_continuous.add(value[start_index:end_index])	1✔
124
125	sliced_sparse = DataCollection(SparseData)	1✔
126	for value in self.sparse_data.values():	1✔
127	sliced_sparse.add(value.t[time[0] : time[-1]])	1✔
128
129	sliced_interval = DataCollection(IntervalData)	1✔
130	for value in self.interval_data.values():	1✔
131	sliced_interval.add(value.t[time[0] : time[-1]])	1✔
132
133	return self.__class__(	1✔
134	label=self.label, # newlabel gives errors
135	name=f"Sliced copy of <{self.name}>",
136	description=f"Sliced copy of <{self.description}>",
137	eit_data=sliced_eit,
138	continuous_data=sliced_continuous,
139	sparse_data=sliced_sparse,
140	interval_data=sliced_interval,
141	)
142
143	def select_by_time(	1✔
144	self,
145	start_time: float \| None = None,
146	end_time: float \| None = None,
147	start_inclusive: bool = True,
148	end_inclusive: bool = False,
149	label: str \| None = None,
150	name: str \| None = None,
151	description: str = "",
152	) -> Self:
153	"""Return a sliced version of the Sequence.
154
155	See `SelectByTime.select_by_time()`.
156	"""
157	if not label:	×
158	label = f"copy_of_<{self.label}>"	×
159	if not name:	×
160	f"Sliced copy of <{self.name}>"	×
161
162	return self.__class__(	×
163	label=label,
164	name=name,
165	description=description,
166	# perform select_by_time() on all four data types
167	**{
168	key: getattr(self, key).select_by_time(
169	start_time=start_time,
170	end_time=end_time,
171	start_inclusive=start_inclusive,
172	end_inclusive=end_inclusive,
173	)
174	for key in ("eit_data", "continuous_data", "sparse_data", "interval_data")
175	},
176	)
177
178	@property	1✔
179	def data(self) -> _DataAccess:	1✔
180	"""Shortcut access to data stored in collections inside a sequence.
181
182	This allows all data objects stored in a collection inside a sequence to be accessed.
183	Instead of `sequence.continuous_data["global_impedance"]` you can use
184	`sequence.data["global_impedance"]`. This works for getting (`sequence.data["label"]` or
185	`sequence.data.get("label")`) and adding data (`sequence.data["label"] = obj` or
186	`sequence.data.add(obj)`).
187
188	Other dict-like behaviour is also supported:
189
190	- `label in sequence.data` to check whether an object with a label exists;
191	- `del sequence.data[label]` to remove an object from the sequence based on the label;
192	- `for label in sequence.data` to iterate over the labels;
193	- `sequence.data.items()` to retrieve a list of (label, object) pairs, especially useful for iteration;
194	- `sequence.data.labels()` or `sequence.data.keys()` to get a list of data labels;
195	- `sequence.data.objects()` or `sequence.data.values()` to get a list of data objects.
196
197	This interface only works if the labels are unique among the data collections. An attempt
198	to add a data object with an exiting label will result in a KeyError.
199	"""
200	return _DataAccess(self)	1✔
201
202
203	@dataclass	1✔
204	class _DataAccess:	1✔
205	_sequence: Sequence	1✔
206
207	def __post_init__(self):	1✔
208	for a, b in itertools.combinations(self._collections, 2):	1✔
209	if duplicates := set(a) & set(b):	1✔
210	msg = f"Duplicate labels ({', '.join(sorted(duplicates))}) found in {a} and {b}."	1✔
211	exc = KeyError(msg)	1✔
212	if sys.version_info >= (3, 11):	1!
UNCOV 213	exc.add_note(	×
214	"You can't use the `data` interface with duplicate labels. "
215	"Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
216	"`interval_data`) instead."
217	)
218	raise exc	1✔
219
220	@property	1✔
221	def _collections(self) -> tuple[DataCollection, ...]:	1✔
222	return (	1✔
223	self._sequence.continuous_data,
224	self._sequence.interval_data,
225	self._sequence.sparse_data,
226	self._sequence.eit_data,
227	)
228
229	@overload	1✔
230	def get(self, label: str) -> DataContainer: ...	1!
231
232	@overload	1✔
233	def get(self, label: str, default: T) -> DataContainer \| T: ...	1!
234
235	def get(self, label: str, default: object = MISSING) -> DataContainer \| object:	1✔
236	"""Get a DataContainer object by label.
237
238	Example:
239	```
240	if filtered_data := sequence.data.get("filtered data", None):
241	print(filtered_data.values.mean())
242	else:
243	print("No filtered data was found.")
244
245	```
246
247	Args:
248	label (str): label of the object to retrieve.
249	default (optional): a default value that is returned if the object is not found.
250	Defaults to MISSING.
251
252	Raises:
253	KeyError: if the object is not found, and no default was set.
254
255	Returns:
256	DataContainer: the requested DataContainer.
257	"""
258	for collection in self._collections:	1✔
259	if label in collection:	1✔
260	return collection[label]	1✔
261
262	if default is not MISSING:	1✔
263	return default	1✔
264
265	msg = f"No object with label {label} was found."	1✔
266	raise KeyError(msg)	1✔
267
268	def __getitem__(self, key: str) -> DataContainer:	1✔
269	return self.get(key)	1✔
270
271	def add(self, *obj: DataContainer) -> None:	1✔
272	"""Add a DataContainer object to the sequence.
273
274	Adds the object to the appropriate data collection. The label of the object must be unique
275	among all data collections, otherwise a KeyError is raised.
276
277	Args:
278	obj (DataContainer): the object to add to the Sequence.
279
280	Raises:
281	KeyError: if the label of the object already exists in any of the data collections.
282	"""
283	for object_ in obj:	1✔
284	if self.get(object_.label, None):	1✔
285	msg = f"An object with the label {object_.label} already exists in this sequence."	1✔
286	exc = KeyError(msg)	1✔
287	if sys.version_info >= (3, 11):	1!
UNCOV 288	exc.add_note(	×
289	"You can't add an object with the same label through the `data` interface. "
290	"Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
291	"`interval_data`) instead."
292	)
293	raise exc	1✔
294
295	match object_:	1✔
296	case ContinuousData():	1✔
297	self._sequence.continuous_data.add(object_)	1✔
298	case IntervalData():	1!
299	self._sequence.interval_data.add(object_)	1✔
300	case SparseData():	×
301	self._sequence.sparse_data.add(object_)	×
302	case EITData():	×
UNCOV 303	self._sequence.eit_data.add(object_)	×
304
305	def __setitem__(self, label: str, obj: DataContainer):	1✔
306	if obj.label != label:	1✔
307	msg = f"Label {label} does not match object label {obj.label}."	1✔
308	raise KeyError(msg)	1✔
309	return self.add(obj)	1✔
310
311	def __contains__(self, label: str) -> bool:	1✔
312	return any(label in container for container in self._collections)	1✔
313
314	def __delitem__(self, label: str) -> None:	1✔
315	for container in self._collections:	1!
316	if label in container:	1!
317	del container[label]	1✔
318	return	1✔
319
320	msg = f"Object with label {label} was not found."	×
UNCOV 321	raise KeyError(msg)	×
322
323	def __iter__(self) -> Iterator[str]:	1✔
324	return itertools.chain(*[collection.keys() for collection in self._collections])	1✔
325
326	def items(self) -> list[tuple[str, DataContainer]]:	1✔
327	"""Return all data items (`(label, object)` pairs)."""
328	return list(itertools.chain(*[collection.items() for collection in self._collections]))	1✔
329
330	def keys(self) -> list[str]:	1✔
331	"""Return a list of all labels."""
332	return list(self.__iter__())	1✔
333
334	labels = keys	1✔
335
336	def values(self) -> list[DataContainer]:	1✔
337	"""Return all data objects."""
338	return list(itertools.chain(*[collection.values() for collection in self._collections]))	1✔
339
340	objects = values	1✔

EIT-ALIVE / eitprocessing / 15051278226

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