12633263274

Committed 06 Jan 2025 01:20PM UTC coverage: 83.698% (+2.1%) from 81.598%

Build # 12633263274

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.5.1 → 1.5.2

Run Details

347 of 454 branches covered (76.43%)

Branch coverage included in aggregate %.

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

9 existing lines in 4 files now uncovered.

1373 of 1601 relevant lines covered (85.76%)

0.86 hits per line

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

92.11

/eitprocessing/datahandling/sparsedata.py

import copy
from dataclasses import dataclass, field
from typing import Any, TypeVar

import numpy as np
from typing_extensions import Self

from eitprocessing.datahandling import DataContainer
from eitprocessing.datahandling.mixins.slicing import SelectByTime

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


@dataclass(eq=False)
class SparseData(DataContainer, SelectByTime):
    """Container for data related to individual time points.

    Sparse data is data for which the time points are not necessarily evenly spaced. Data can consist time-value pairs
    or only time points.

    Sparse data differs from interval data in that each data points is associated with a single time point rather than a
    time range.

    Examples are data points at end of inspiration/end of expiration (e.g. tidal volume, end-expiratoy lung impedance)
    or detected time points (e.g. QRS complexes).

    Args:
        label: Computer readable name.
        name: Human readable name.
        unit: Unit of the data, if applicable.
        category: Category the data falls into, e.g. 'detected r peak'.
        description: Human readable extended description of the data.
        parameters: Parameters used to derive the data.
        derived_from: Traceback of intermediates from which the current data was derived.
        values: List or array of values. These van be numeric data, text or Python objects.
    """

    label: str = field(compare=False)
    name: str = field(compare=False, repr=False)
    unit: str | None = field(metadata={"check_equivalence": True}, repr=False)
    category: str = field(metadata={"check_equivalence": True}, repr=False)
    time: np.ndarray = field(repr=False)
    description: str = field(compare=False, default="", repr=False)
    parameters: dict[str, Any] = field(default_factory=dict, metadata={"check_equivalence": True}, repr=False)
    derived_from: list[Any] = field(default_factory=list, compare=False, repr=False)
    values: Any | None = None

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}('{self.label}')"

    def __len__(self) -> int:
        return len(self.time)

    def __post_init__(self):
        if self.has_values and (lv := len(self.values)) != (lt := len(self.time)):
            msg = f"The number of time points ({lt}) does not match the number of values ({lv})."
            raise ValueError(msg)

    @property
    def has_values(self) -> bool:
        """True if the SparseData has values, False otherwise."""
        return self.values is not None

    def _sliced_copy(
        self,
        start_index: int,
        end_index: int,
        newlabel: str,
    ) -> Self:
        cls = self.__class__
        time = self.time[start_index:end_index]
        values = self.values[start_index:end_index] if self.has_values else None
        description = f"Slice ({start_index}-{end_index}) of <{self.description}>"

        return cls(
            label=newlabel,
            name=self.name,
            unit=self.unit,
            category=self.category,
            description=description,
            derived_from=[*self.derived_from, self],
            time=time,
            values=values,
        )

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

    def concatenate(self: T, other: T, newlabel: str | None = None) -> T:  # noqa: D102, will be moved to mixin in future
        self.isequivalent(other, raise_=True)

        # TODO: make proper copy functions
        if not len(self):
            return copy.deepcopy(other)
        if not len(other):
            return copy.deepcopy(self)

        if np.min(other.time) <= np.max(self.time):
            msg = f"{other} (b) starts before {self} (a) ends."
            raise ValueError(msg)

        cls = type(self)
        newlabel = newlabel or self.label

        if type(self.values) is not type(other.values):
            msg = "Concatenation failed because value types are non-identical."
            raise TypeError(msg)
        if not self.has_values:
            new_values = None
        elif isinstance(self.values, np.ndarray):
            new_values = np.concatenate((self.values, other.values))
        else:
            try:
                new_values = self.values + other.values
            except TypeError as e:
                msg = "Concatenation failed because values could not be concatenated."
                raise TypeError(msg) from e

        return cls(
            label=newlabel,
            name=self.name,
            unit=self.unit,
            category=self.category,
            description=self.description,
            derived_from=[*self.derived_from, *other.derived_from, self, other],
            time=np.concatenate((self.time, other.time)),
            values=new_values,
        )

1	import copy	1✔
2	from dataclasses import dataclass, field	1✔
3	from typing import Any, TypeVar	1✔
4
5	import numpy as np	1✔
6	from typing_extensions import Self	1✔
7
8	from eitprocessing.datahandling import DataContainer	1✔
9	from eitprocessing.datahandling.mixins.slicing import SelectByTime	1✔
10
11	T = TypeVar("T", bound="SparseData")	1✔
12
13
14	@dataclass(eq=False)	1✔
15	class SparseData(DataContainer, SelectByTime):	1✔
16	"""Container for data related to individual time points.
17
18	Sparse data is data for which the time points are not necessarily evenly spaced. Data can consist time-value pairs
19	or only time points.
20
21	Sparse data differs from interval data in that each data points is associated with a single time point rather than a
22	time range.
23
24	Examples are data points at end of inspiration/end of expiration (e.g. tidal volume, end-expiratoy lung impedance)
25	or detected time points (e.g. QRS complexes).
26
27	Args:
28	label: Computer readable name.
29	name: Human readable name.
30	unit: Unit of the data, if applicable.
31	category: Category the data falls into, e.g. 'detected r peak'.
32	description: Human readable extended description of the data.
33	parameters: Parameters used to derive the data.
34	derived_from: Traceback of intermediates from which the current data was derived.
35	values: List or array of values. These van be numeric data, text or Python objects.
36	"""
37
38	label: str = field(compare=False)	1✔
39	name: str = field(compare=False, repr=False)	1✔
40	unit: str \| None = field(metadata={"check_equivalence": True}, repr=False)	1✔
41	category: str = field(metadata={"check_equivalence": True}, repr=False)	1✔
42	time: np.ndarray = field(repr=False)	1✔
43	description: str = field(compare=False, default="", repr=False)	1✔
44	parameters: dict[str, Any] = field(default_factory=dict, metadata={"check_equivalence": True}, repr=False)	1✔
45	derived_from: list[Any] = field(default_factory=list, compare=False, repr=False)	1✔
46	values: Any \| None = None	1✔
47
48	def __repr__(self) -> str:	1✔
49	return f"{self.__class__.__name__}('{self.label}')"	1✔
50
51	def __len__(self) -> int:	1✔
52	return len(self.time)	1✔
53
54	def __post_init__(self):	1✔
55	if self.has_values and (lv := len(self.values)) != (lt := len(self.time)):	1!
UNCOV 56	msg = f"The number of time points ({lt}) does not match the number of values ({lv})."	×
57	raise ValueError(msg)	×
58
59	@property	1✔
60	def has_values(self) -> bool:	1✔
61	"""True if the SparseData has values, False otherwise."""
62	return self.values is not None	1✔
63
64	def _sliced_copy(	1✔
65	self,
66	start_index: int,
67	end_index: int,
68	newlabel: str,
69	) -> Self:
70	cls = self.__class__	1✔
71	time = self.time[start_index:end_index]	1✔
72	values = self.values[start_index:end_index] if self.has_values else None	1✔
73	description = f"Slice ({start_index}-{end_index}) of <{self.description}>"	1✔
74
75	return cls(	1✔
76	label=newlabel,
77	name=self.name,
78	unit=self.unit,
79	category=self.category,
80	description=description,
81	derived_from=[*self.derived_from, self],
82	time=time,
83	values=values,
84	)
85
86	def __add__(self: Self, other: Self) -> Self:	1✔
87	return self.concatenate(other)	1✔
88
89	def concatenate(self: T, other: T, newlabel: str \| None = None) -> T: # noqa: D102, will be moved to mixin in future	1✔
90	self.isequivalent(other, raise_=True)	1✔
91
92	# TODO: make proper copy functions
93	if not len(self):	1✔
94	return copy.deepcopy(other)	1✔
95	if not len(other):	1✔
96	return copy.deepcopy(self)	1✔
97
98	if np.min(other.time) <= np.max(self.time):	1✔
99	msg = f"{other} (b) starts before {self} (a) ends."	1✔
100	raise ValueError(msg)	1✔
101
102	cls = type(self)	1✔
103	newlabel = newlabel or self.label	1✔
104
105	if type(self.values) is not type(other.values):	1✔
106	msg = "Concatenation failed because value types are non-identical."	1✔
107	raise TypeError(msg)	1✔
108	if not self.has_values:	1✔
109	new_values = None	1✔
110	elif isinstance(self.values, np.ndarray):	1✔
111	new_values = np.concatenate((self.values, other.values))	1✔
112	else:
113	try:	1✔
114	new_values = self.values + other.values	1✔
UNCOV 115	except TypeError as e:	×
116	msg = "Concatenation failed because values could not be concatenated."	×
117	raise TypeError(msg) from e	×
118
119	return cls(	1✔
120	label=newlabel,
121	name=self.name,
122	unit=self.unit,
123	category=self.category,
124	description=self.description,
125	derived_from=[self.derived_from, other.derived_from, self, other],
126	time=np.concatenate((self.time, other.time)),
127	values=new_values,
128	)

EIT-ALIVE / eitprocessing / 12633263274

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