• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

EIT-ALIVE / eitprocessing / 14716370906

28 Apr 2025 07:38PM UTC coverage: 83.838% (+0.06%) from 83.777%
14716370906

push

github

psomhorst
Bump version: 1.6.0 → 1.7.0

375 of 492 branches covered (76.22%)

Branch coverage included in aggregate %.

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

18 existing lines in 2 files now uncovered.

1451 of 1686 relevant lines covered (86.06%)

0.86 hits per line

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

82.83
/eitprocessing/datahandling/sequence.py
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

UNCOV
76
        msg = "Sequence has no timed data"
×
UNCOV
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!
UNCOV
113
            msg = "start_index larger than length of time axis"
×
UNCOV
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
        """
UNCOV
157
        if not label:
×
UNCOV
158
            label = f"copy_of_<{self.label}>"
×
UNCOV
159
        if not name:
×
UNCOV
160
            f"Sliced copy of <{self.name}>"
×
161

UNCOV
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
        - `label in sequence.data` to check whether an object with a label exists;
190
        - `del sequence.data[label]` to remove an object from the sequence based on the label;
191
        - `for label in sequence.data` to iterate over the labels;
192
        - `sequence.data.items()` to retrieve a list of (label, object) pairs, especially useful for iteration;
193
        - `sequence.data.labels()` or `sequence.data.keys()` to get a list of data labels;
194
        - `sequence.data.objects()` or `sequence.data.values()` to get a list of data objects.
195

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

201

202
@dataclass
1✔
203
class _DataAccess:
1✔
204
    _sequence: Sequence
1✔
205

206
    def __post_init__(self):
1✔
207
        for a, b in itertools.combinations(self._collections, 2):
1✔
208
            if duplicates := set(a) & set(b):
1✔
209
                msg = f"Duplicate labels ({', '.join(sorted(duplicates))}) found in {a} and {b}."
1✔
210
                exc = KeyError(msg)
1✔
211
                if sys.version_info >= (3, 11):
1!
UNCOV
212
                    exc.add_note(
×
213
                        "You can't use the `data` interface with duplicate labels. "
214
                        "Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
215
                        "`interval_data`) instead."
216
                    )
217
                raise exc
1✔
218

219
    @property
1✔
220
    def _collections(self) -> tuple[DataCollection, ...]:
1✔
221
        return (
1✔
222
            self._sequence.continuous_data,
223
            self._sequence.interval_data,
224
            self._sequence.sparse_data,
225
            self._sequence.eit_data,
226
        )
227

228
    @overload
1✔
229
    def get(self, label: str) -> DataContainer: ...
1!
230

231
    @overload
1✔
232
    def get(self, label: str, default: T) -> DataContainer | T: ...
1!
233

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

237
        Example:
238
        ```
239
        if filtered_data := sequence.data.get("filtered data", None):
240
            print(filtered_data.values.mean())
241
        else:
242
            print("No filtered data was found.")
243

244
        ```
245

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

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

254
        Returns:
255
            DataContainer: the requested DataContainer.
256
        """
257
        for collection in self._collections:
1✔
258
            if label in collection:
1✔
259
                return collection[label]
1✔
260

261
        if default is not MISSING:
1✔
262
            return default
1✔
263

264
        msg = f"No object with label {label} was found."
1✔
265
        raise KeyError(msg)
1✔
266

267
    def __getitem__(self, key: str) -> DataContainer:
1✔
268
        return self.get(key)
1✔
269

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

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

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

279
        Raises:
280
            KeyError: if the label of the object already exists in any of the data collections.
281
        """
282
        for object_ in obj:
1✔
283
            if self.get(object_.label, None):
1✔
284
                msg = f"An object with the label {object_.label} already exists in this sequence."
1✔
285
                exc = KeyError(msg)
1✔
286
                if sys.version_info >= (3, 11):
1!
UNCOV
287
                    exc.add_note(
×
288
                        "You can't add an object with the same label through the `data` interface. "
289
                        "Use the explicit data collections (`eit_data`, `continuous_data`, `sparse_data`, "
290
                        "`interval_data`) instead."
291
                    )
292
                raise exc
1✔
293

294
            match object_:
1✔
295
                case ContinuousData():
1✔
296
                    self._sequence.continuous_data.add(object_)
1✔
297
                case IntervalData():
1!
298
                    self._sequence.interval_data.add(object_)
1✔
UNCOV
299
                case SparseData():
×
UNCOV
300
                    self._sequence.sparse_data.add(object_)
×
UNCOV
301
                case EITData():
×
UNCOV
302
                    self._sequence.eit_data.add(object_)
×
303

304
    def __setitem__(self, label: str, obj: DataContainer):
1✔
305
        if obj.label != label:
1✔
306
            msg = f"Label {label} does not match object label {obj.label}."
1✔
307
            raise KeyError(msg)
1✔
308
        return self.add(obj)
1✔
309

310
    def __contains__(self, label: str) -> bool:
1✔
311
        return any(label in container for container in self._collections)
1✔
312

313
    def __delitem__(self, label: str) -> None:
1✔
314
        for container in self._collections:
1!
315
            if label in container:
1!
316
                del container[label]
1✔
317
                return
1✔
318

UNCOV
319
        msg = f"Object with label {label} was not found."
×
UNCOV
320
        raise KeyError(msg)
×
321

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

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

329
    def keys(self) -> list[str]:
1✔
330
        """Return a list of all labels."""
331
        return list(self.__iter__())
1✔
332

333
    labels = keys
1✔
334

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

339
    objects = values
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc