19312054818

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

Build # 19312054818

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.8.4 → 1.8.5

Run Details

781 of 976 branches covered (80.02%)

Branch coverage included in aggregate %.

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

34 existing lines in 5 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

84.53

/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(other)

    def concatenate(
        self: Sequence,
        other: Sequence,
        newlabel: str | None = None,
    ) -> Sequence:
        """Create a merge of two Sequence objects."""
        # TODO: rewrite

        concat_eit = self.eit_data.concatenate(other.eit_data)
        concat_continuous = self.continuous_data.concatenate(other.continuous_data)
        concat_sparse = self.sparse_data.concatenate(other.sparse_data)
        concat_interval = self.interval_data.concatenate(other.interval_data)

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

        return self.__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(other)	1✔
84
85	def concatenate(	1✔
86	self: Sequence,
87	other: Sequence,
88	newlabel: str \| None = None,
89	) -> Sequence:
90	"""Create a merge of two Sequence objects."""
91	# TODO: rewrite
92
93	concat_eit = self.eit_data.concatenate(other.eit_data)	1✔
94	concat_continuous = self.continuous_data.concatenate(other.continuous_data)	1✔
95	concat_sparse = self.sparse_data.concatenate(other.sparse_data)	1✔
96	concat_interval = self.interval_data.concatenate(other.interval_data)	1✔
97
98	newlabel = newlabel or self.label	1✔
99	# TODO: add concatenation of other attached objects
100
101	return self.__class__(	1✔
102	eit_data=concat_eit,
103	continuous_data=concat_continuous,
104	sparse_data=concat_sparse,
105	interval_data=concat_interval,
106	label=newlabel,
107	)
108
109	def _sliced_copy(self, start_index: int, end_index: int, newlabel: str) -> Self: # noqa: ARG002	1✔
110	if start_index >= len(self.time):	1!
UNCOV 111	msg = "start_index larger than length of time axis"	×
UNCOV 112	raise ValueError(msg)	×
113	time = self.time[start_index:end_index]	1✔
114
115	sliced_eit = DataCollection(EITData)	1✔
116	for value in self.eit_data.values():	1✔
117	sliced_eit.add(value[start_index:end_index])	1✔
118
119	sliced_continuous = DataCollection(ContinuousData)	1✔
120	for value in self.continuous_data.values():	1✔
121	sliced_continuous.add(value[start_index:end_index])	1✔
122
123	sliced_sparse = DataCollection(SparseData)	1✔
124	for value in self.sparse_data.values():	1✔
125	sliced_sparse.add(value.t[time[0] : time[-1]])	1✔
126
127	sliced_interval = DataCollection(IntervalData)	1✔
128	for value in self.interval_data.values():	1✔
129	sliced_interval.add(value.t[time[0] : time[-1]])	1✔
130
131	return self.__class__(	1✔
132	label=self.label, # newlabel gives errors
133	name=f"Sliced copy of <{self.name}>",
134	description=f"Sliced copy of <{self.description}>",
135	eit_data=sliced_eit,
136	continuous_data=sliced_continuous,
137	sparse_data=sliced_sparse,
138	interval_data=sliced_interval,
139	)
140
141	def select_by_time(	1✔
142	self,
143	start_time: float \| None = None,
144	end_time: float \| None = None,
145	start_inclusive: bool = True,
146	end_inclusive: bool = False,
147	label: str \| None = None,
148	name: str \| None = None,
149	description: str = "",
150	) -> Self:
151	"""Return a sliced version of the Sequence.
152
153	See `SelectByTime.select_by_time()`.
154	"""
UNCOV 155	if not label:	×
UNCOV 156	label = f"copy_of_<{self.label}>"	×
157	if not name:	×
158	f"Sliced copy of <{self.name}>"	×
159
160	return self.__class__(	×
161	label=label,
162	name=name,
163	description=description,
164	# perform select_by_time() on all four data types
165	**{
166	key: getattr(self, key).select_by_time(
167	start_time=start_time,
168	end_time=end_time,
169	start_inclusive=start_inclusive,
170	end_inclusive=end_inclusive,
171	)
172	for key in ("eit_data", "continuous_data", "sparse_data", "interval_data")
173	},
174	)
175
176	@property	1✔
177	def data(self) -> _DataAccess:	1✔
178	"""Shortcut access to data stored in collections inside a sequence.
179
180	This allows all data objects stored in a collection inside a sequence to be accessed.
181	Instead of `sequence.continuous_data["global_impedance"]` you can use
182	`sequence.data["global_impedance"]`. This works for getting (`sequence.data["label"]` or
183	`sequence.data.get("label")`) and adding data (`sequence.data["label"] = obj` or
184	`sequence.data.add(obj)`).
185
186	Other dict-like behaviour is also supported:
187
188	- `label in sequence.data` to check whether an object with a label exists;
189	- `del sequence.data[label]` to remove an object from the sequence based on the label;
190	- `for label in sequence.data` to iterate over the labels;
191	- `sequence.data.items()` to retrieve a list of (label, object) pairs, especially useful for iteration;
192	- `sequence.data.labels()` or `sequence.data.keys()` to get a list of data labels;
193	- `sequence.data.objects()` or `sequence.data.values()` to get a list of data objects.
194
195	This interface only works if the labels are unique among the data collections. An attempt
196	to add a data object with an exiting label will result in a KeyError.
197	"""
198	return _DataAccess(self)	1✔
199
200
201	@dataclass	1✔
202	class _DataAccess:	1✔
203	_sequence: Sequence	1✔
204
205	def __post_init__(self):	1✔
206	for a, b in itertools.combinations(self._collections, 2):	1✔
207	if duplicates := set(a) & set(b):	1✔
208	msg = f"Duplicate labels ({', '.join(sorted(duplicates))}) found in {a} and {b}."	1✔
209	exc = KeyError(msg)	1✔
210	if sys.version_info >= (3, 11):
211	exc.add_note(
212	"You can't use the `data` interface with duplicate labels. "
213	"Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
214	"`interval_data`) instead."
215	)
216	raise exc	1✔
217
218	@property	1✔
219	def _collections(self) -> tuple[DataCollection, ...]:	1✔
220	return (	1✔
221	self._sequence.continuous_data,
222	self._sequence.interval_data,
223	self._sequence.sparse_data,
224	self._sequence.eit_data,
225	)
226
227	@overload
228	def get(self, label: str) -> DataContainer: ...
229
230	@overload
231	def get(self, label: str, default: T) -> DataContainer \| T: ...
232
233	def get(self, label: str, default: object = MISSING) -> DataContainer \| object:	1✔
234	"""Get a DataContainer object by label.
235
236	Example:
237	```
238	if filtered_data := sequence.data.get("filtered data", None):
239	print(filtered_data.values.mean())
240	else:
241	print("No filtered data was found.")
242
243	```
244
245	Args:
246	label (str): label of the object to retrieve.
247	default (optional): a default value that is returned if the object is not found.
248	Defaults to MISSING.
249
250	Raises:
251	KeyError: if the object is not found, and no default was set.
252
253	Returns:
254	DataContainer: the requested DataContainer.
255	"""
256	for collection in self._collections:	1✔
257	if label in collection:	1✔
258	return collection[label]	1✔
259
260	if default is not MISSING:	1✔
261	return default	1✔
262
263	msg = f"No object with label {label} was found."	1✔
264	raise KeyError(msg)	1✔
265
266	def __getitem__(self, key: str) -> DataContainer:	1✔
267	return self.get(key)	1✔
268
269	def add(self, *obj: DataContainer) -> None:	1✔
270	"""Add a DataContainer object to the sequence.
271
272	Adds the object to the appropriate data collection. The label of the object must be unique
273	among all data collections, otherwise a KeyError is raised.
274
275	Args:
276	obj (DataContainer): the object to add to the Sequence.
277
278	Raises:
279	KeyError: if the label of the object already exists in any of the data collections.
280	"""
281	for object_ in obj:	1✔
282	if self.get(object_.label, None):	1✔
283	msg = f"An object with the label {object_.label} already exists in this sequence."	1✔
284	exc = KeyError(msg)	1✔
285	if sys.version_info >= (3, 11):
286	exc.add_note(
287	"You can't add an object with the same label through the `data` interface. "
288	"Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
289	"`interval_data`) instead."
290	)
291	raise exc	1✔
292
293	match object_:	1✔
294	case ContinuousData():	1✔
295	self._sequence.continuous_data.add(object_)	1✔
296	case IntervalData():	1!
297	self._sequence.interval_data.add(object_)	1✔
UNCOV 298	case SparseData():	×
UNCOV 299	self._sequence.sparse_data.add(object_)	×
300	case EITData():	×
301	self._sequence.eit_data.add(object_)	×
302
303	def __setitem__(self, label: str, obj: DataContainer):	1✔
304	if obj.label != label:	1✔
305	msg = f"Label {label} does not match object label {obj.label}."	1✔
306	raise KeyError(msg)	1✔
307	return self.add(obj)	1✔
308
309	def __contains__(self, label: str) -> bool:	1✔
310	return any(label in container for container in self._collections)	1✔
311
312	def __delitem__(self, label: str) -> None:	1✔
313	for container in self._collections:	1!
314	if label in container:	1!
315	del container[label]	1✔
316	return	1✔
317
UNCOV 318	msg = f"Object with label {label} was not found."	×
UNCOV 319	raise KeyError(msg)	×
320
321	def __iter__(self) -> Iterator[str]:	1✔
322	return itertools.chain(*[collection.keys() for collection in self._collections])	1✔
323
324	def items(self) -> list[tuple[str, DataContainer]]:	1✔
325	"""Return all data items (`(label, object)` pairs)."""
326	return list(itertools.chain(*[collection.items() for collection in self._collections]))	1✔
327
328	def keys(self) -> list[str]:	1✔
329	"""Return a list of all labels."""
330	return list(self.__iter__())	1✔
331
332	labels = keys	1✔
333
334	def values(self) -> list[DataContainer]:	1✔
335	"""Return all data objects."""
336	return list(itertools.chain(*[collection.values() for collection in self._collections]))	1✔
337
338	objects = values	1✔

EIT-ALIVE / eitprocessing / 19312054818

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