12120635114

Committed 02 Dec 2024 01:44PM UTC coverage: 81.489% (-0.6%) from 82.129%

Build # 12120635114

Build Type

push

github

Committed by

actions-user

Commit Message

Bump version: 1.4.6 → 1.4.7

Run Details

344 of 468 branches covered (73.5%)

Branch coverage included in aggregate %.

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

36 existing lines in 5 files now uncovered.

1364 of 1628 relevant lines covered (83.78%)

0.84 hits per line

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

73.79

/eitprocessing/datahandling/continuousdata.py

from __future__ import annotations

import warnings
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, TypeVar

import numpy as np

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

if TYPE_CHECKING:
    from collections.abc import Callable

    from typing_extensions import Any, Self

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


@dataclass(eq=False)
class ContinuousData(DataContainer, SelectByTime):
    """Container for data with a continuous time axis.

    Continuous data is assumed to be sequential (i.e. a single data point at each time point, sorted by time) and
    continuously measured/created at a fixed sampling frequency. However, a fixed interval between consecutive time
    points is not enforced to account for floating point arithmetic, devices with imperfect sampling frequencies, and
    other sources of variation.

    Args:
        label: Computer readable naming of the instance.
        name: Human readable naming of the instance.
        unit: Unit of the data, if applicable.
        category: Category the data falls into, e.g. 'airway pressure'.
        description: Human readable extended description of the data.
        parameters: Parameters used to derive this data.
        derived_from: Traceback of intermediates from which the current data was derived.
        values: Data points.
    """

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

    def __post_init__(self) -> None:
        if self.loaded:
            self.lock()
        self.lock("time")

        if self.sample_frequency is None:
            msg = (
                "`sample_frequency` is set to `None`. This will not be supported in future versions. "
                "Provide a sample frequency when creating a ContinuousData object."
            )
            warnings.warn(msg, DeprecationWarning)

        if (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)

    def __setattr__(self, attr: str, value: Any):  # noqa: ANN401
        try:
            old_value = getattr(self, attr)
        except AttributeError:
            pass
        else:
            if isinstance(old_value, np.ndarray) and old_value.flags["WRITEABLE"] is False:
                msg = f"Attribute '{attr}' is locked and can't be overwritten."
                raise AttributeError(msg)
        super().__setattr__(attr, value)

    def copy(
        self,
        label: str,
        *,
        name: str | None = None,
        unit: str | None = None,
        description: str | None = None,
        parameters: dict | None = None,
    ) -> Self:
        """Create a copy.

        Whenever data is altered, it should probably be copied first. The alterations should then be made in the copy.
        """
        obj = self.__class__(
            label=label,
            name=name or label,
            unit=unit or self.unit,
            description=description or f"Derived from {self.name}",
            parameters=self.parameters | (parameters or {}),
            derived_from=[*self.derived_from, self],
            category=self.category,
            # copying data can become inefficient with large datasets if the
            # data is not directly edited afer copying but overridden instead;
            # consider creating a view and locking it, requiring the user to
            # make a copy if they want to edit the data directly
            time=np.copy(self.time),
            values=np.copy(self.values),
            sample_frequency=self.sample_frequency,
        )
        obj.unlock()
        return obj

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

    def concatenate(self: T, other: T, newlabel: str | None = None) -> T:  # noqa: D102, will be removed soon
        # TODO: compare both concatenate methods and check what is needed from both and merge into one
        # Check that data can be concatenated
        self.isequivalent(other, raise_=True)
        if np.min(other.time) <= np.max(self.time):
            msg = f"{other} (b) starts before {self} (a) ends."
            raise ValueError(msg)

        cls = self.__class__
        newlabel = newlabel or self.label

        return cls(
            name=self.name,
            label=self.label,  # TODO: using newlabel leads to errors
            unit=self.unit,
            category=self.category,
            time=np.concatenate((self.time, other.time)),
            values=np.concatenate((self.values, other.values)),
            derived_from=[*self.derived_from, *other.derived_from, self, other],
            sample_frequency=self.sample_frequency,
        )

    def derive(
        self,
        label: str,
        function: Callable,
        func_args: dict | None = None,
        **kwargs,
    ) -> Self:
        """Create a copy deriving data from values attribute.

        Args:
            label: New label for the derived object.
            function: Function that takes the values and returns the derived values.
            func_args: Arguments to pass to function, if any.
            **kwargs: Values for changed attributes of derived object.

        Example:
        ```
        def convert_data(x, add=None, subtract=None, multiply=None, divide=None):
            if add:
                x += add
            if subtract:
                x -= subtract
            if multiply:
                x *= multiply
            if divide:
                x /= divide
            return x


        data = ContinuousData(
            name="Lung volume (in mL)", label="volume_mL", unit="mL", category="volume", values=some_loaded_data
        )
        derived = data.derive("volume_L", convert_data, {"divide": 1000}, name="Lung volume (in L)", unit="L")
        ```
        """
        if func_args is None:
            func_args = {}
        copy = self.copy(label, **kwargs)
        copy.values = function(copy.values, **func_args)
        return copy

    def lock(self, *attr: str) -> None:
        """Lock attributes, essentially rendering them read-only.

        Locked attributes cannot be overwritten. Attributes can be unlocked using `unlock()`.

        Args:
            *attr: any number of attributes can be passed here, all of which will be locked. Defaults to "values".

        Examples:
            >>> # lock the `values` attribute of `data`
            >>> data.lock()
            >>> data.values = [1, 2, 3] # will result in an AttributeError
            >>> data.values[0] = 1      # will result in a RuntimeError
        """
        if not len(attr):
            # default values are not allowed when using *attr, so set a default here if none is supplied
            attr = ("values",)
        for attr_ in attr:
            getattr(self, attr_).flags["WRITEABLE"] = False

    def unlock(self, *attr: str) -> None:
        """Unlock attributes, rendering them editable.

        Locked attributes cannot be overwritten, but can be unlocked with this function to make them editable.

        Args:
            *attr: any number of attributes can be passed here, all of which will be unlocked. Defaults to "values".

        Examples:
            >>> # lock the `values` attribute of `data`
            >>> data.lock()
            >>> data.values = [1, 2, 3] # will result in an AttributeError
            >>> data.values[0] = 1      # will result in a RuntimeError
            >>> data.unlock()
            >>> data.values = [1, 2, 3]
            >>> print(data.values)
            [1,2,3]
            >>> data.values[0] = 1      # will result in a RuntimeError
            >>> print(data.values)
            1
        """
        if not len(attr):
            # default values are not allowed when using *attr, so set a default here if none is supplied
            attr = ("values",)
        for attr_ in attr:
            getattr(self, attr_).flags["WRITEABLE"] = True

    @property
    def locked(self) -> bool:
        """Return whether the values attribute is locked.

        See lock().
        """
        return not self.values.flags["WRITEABLE"]

    @property
    def loaded(self) -> bool:
        """Return whether the data was loaded from disk, or derived from elsewhere."""
        return len(self.derived_from) == 0

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

    def _sliced_copy(
        self,
        start_index: int,
        end_index: int,
        newlabel: str,  # noqa: ARG002
    ) -> Self:
        # TODO: check correct implementation
        cls = self.__class__
        time = np.copy(self.time[start_index:end_index])
        values = np.copy(self.values[start_index:end_index])
        description = f"Slice ({start_index}-{end_index}) of <{self.description}>"

        return cls(
            label=self.label,  # TODO: newlabel gives errors
            name=self.name,
            unit=self.unit,
            category=self.category,
            description=description,
            derived_from=[*self.derived_from, self],
            time=time,
            values=values,
            sample_frequency=self.sample_frequency,
        )

1	from __future__ import annotations	1✔
2
3	import warnings	1✔
4	from dataclasses import dataclass, field	1✔
5	from typing import TYPE_CHECKING, TypeVar	1✔
6
7	import numpy as np	1✔
8
9	from eitprocessing.datahandling import DataContainer	1✔
10	from eitprocessing.datahandling.mixins.slicing import SelectByTime	1✔
11
12	if TYPE_CHECKING:
13	from collections.abc import Callable
14
15	from typing_extensions import Any, Self
16
17	T = TypeVar("T", bound="ContinuousData")	1✔
18
19
20	@dataclass(eq=False)	1✔
21	class ContinuousData(DataContainer, SelectByTime):	1✔
22	"""Container for data with a continuous time axis.
23
24	Continuous data is assumed to be sequential (i.e. a single data point at each time point, sorted by time) and
25	continuously measured/created at a fixed sampling frequency. However, a fixed interval between consecutive time
26	points is not enforced to account for floating point arithmetic, devices with imperfect sampling frequencies, and
27	other sources of variation.
28
29	Args:
30	label: Computer readable naming of the instance.
31	name: Human readable naming of the instance.
32	unit: Unit of the data, if applicable.
33	category: Category the data falls into, e.g. 'airway pressure'.
34	description: Human readable extended description of the data.
35	parameters: Parameters used to derive this data.
36	derived_from: Traceback of intermediates from which the current data was derived.
37	values: Data points.
38	"""
39
40	label: str = field(compare=False)	1✔
41	name: str = field(compare=False, repr=False)	1✔
42	unit: str = field(metadata={"check_equivalence": True}, repr=False)	1✔
43	category: str = field(metadata={"check_equivalence": True}, repr=False)	1✔
44	description: str = field(default="", compare=False, repr=False)	1✔
45	parameters: dict[str, Any] = field(default_factory=dict, repr=False, metadata={"check_equivalence": True})	1✔
46	derived_from: Any \| list[Any] = field(default_factory=list, repr=False, compare=False)	1✔
47	time: np.ndarray = field(kw_only=True, repr=False)	1✔
48	values: np.ndarray = field(kw_only=True, repr=False)	1✔
49	sample_frequency: float \| None = field(kw_only=True, repr=False, metadata={"check_equivalence": True}, default=None)	1✔
50
51	def __post_init__(self) -> None:	1✔
52	if self.loaded:	1✔
53	self.lock()	1✔
54	self.lock("time")	1✔
55
56	if self.sample_frequency is None:	1✔
57	msg = (	1✔
58	"`sample_frequency` is set to `None`. This will not be supported in future versions. "
59	"Provide a sample frequency when creating a ContinuousData object."
60	)
61	warnings.warn(msg, DeprecationWarning)	1✔
62
63	if (lv := len(self.values)) != (lt := len(self.time)):	1!
UNCOV 64	msg = f"The number of time points ({lt}) does not match the number of values ({lv})."	×
UNCOV 65	raise ValueError(msg)	×
66
67	def __setattr__(self, attr: str, value: Any): # noqa: ANN401	1✔
68	try:	1✔
69	old_value = getattr(self, attr)	1✔
70	except AttributeError:	1✔
71	pass	1✔
72	else:
73	if isinstance(old_value, np.ndarray) and old_value.flags["WRITEABLE"] is False:	1!
UNCOV 74	msg = f"Attribute '{attr}' is locked and can't be overwritten."	×
UNCOV 75	raise AttributeError(msg)	×
76	super().__setattr__(attr, value)	1✔
77
78	def copy(	1✔
79	self,
80	label: str,
81	*,
82	name: str \| None = None,
83	unit: str \| None = None,
84	description: str \| None = None,
85	parameters: dict \| None = None,
86	) -> Self:
87	"""Create a copy.
88
89	Whenever data is altered, it should probably be copied first. The alterations should then be made in the copy.
90	"""
UNCOV 91	obj = self.__class__(	×
92	label=label,
93	name=name or label,
94	unit=unit or self.unit,
95	description=description or f"Derived from {self.name}",
96	parameters=self.parameters \| (parameters or {}),
97	derived_from=[*self.derived_from, self],
98	category=self.category,
99	# copying data can become inefficient with large datasets if the
100	# data is not directly edited afer copying but overridden instead;
101	# consider creating a view and locking it, requiring the user to
102	# make a copy if they want to edit the data directly
103	time=np.copy(self.time),
104	values=np.copy(self.values),
105	sample_frequency=self.sample_frequency,
106	)
107	obj.unlock()	×
UNCOV 108	return obj	×
109
110	def __add__(self: T, other: T) -> T:	1✔
UNCOV 111	return self.concatenate(other)	×
112
113	def concatenate(self: T, other: T, newlabel: str \| None = None) -> T: # noqa: D102, will be removed soon	1✔
114	# TODO: compare both concatenate methods and check what is needed from both and merge into one
115	# Check that data can be concatenated
116	self.isequivalent(other, raise_=True)	1✔
117	if np.min(other.time) <= np.max(self.time):	1✔
118	msg = f"{other} (b) starts before {self} (a) ends."	1✔
119	raise ValueError(msg)	1✔
120
121	cls = self.__class__	1✔
122	newlabel = newlabel or self.label	1✔
123
124	return cls(	1✔
125	name=self.name,
126	label=self.label, # TODO: using newlabel leads to errors
127	unit=self.unit,
128	category=self.category,
129	time=np.concatenate((self.time, other.time)),
130	values=np.concatenate((self.values, other.values)),
131	derived_from=[self.derived_from, other.derived_from, self, other],
132	sample_frequency=self.sample_frequency,
133	)
134
135	def derive(	1✔
136	self,
137	label: str,
138	function: Callable,
139	func_args: dict \| None = None,
140	**kwargs,
141	) -> Self:
142	"""Create a copy deriving data from values attribute.
143
144	Args:
145	label: New label for the derived object.
146	function: Function that takes the values and returns the derived values.
147	func_args: Arguments to pass to function, if any.
148	**kwargs: Values for changed attributes of derived object.
149
150	Example:
151	```
152	def convert_data(x, add=None, subtract=None, multiply=None, divide=None):
153	if add:
154	x += add
155	if subtract:
156	x -= subtract
157	if multiply:
158	x *= multiply
159	if divide:
160	x /= divide
161	return x
162
163
164	data = ContinuousData(
165	name="Lung volume (in mL)", label="volume_mL", unit="mL", category="volume", values=some_loaded_data
166	)
167	derived = data.derive("volume_L", convert_data, {"divide": 1000}, name="Lung volume (in L)", unit="L")
168	```
169	"""
170	if func_args is None:	×
UNCOV 171	func_args = {}	×
UNCOV 172	copy = self.copy(label, **kwargs)	×
UNCOV 173	copy.values = function(copy.values, **func_args)	×
UNCOV 174	return copy	×
175
176	def lock(self, *attr: str) -> None:	1✔
177	"""Lock attributes, essentially rendering them read-only.
178
179	Locked attributes cannot be overwritten. Attributes can be unlocked using `unlock()`.
180
181	Args:
182	*attr: any number of attributes can be passed here, all of which will be locked. Defaults to "values".
183
184	Examples:
185	>>> # lock the `values` attribute of `data`
186	>>> data.lock()
187	>>> data.values = [1, 2, 3] # will result in an AttributeError
188	>>> data.values[0] = 1 # will result in a RuntimeError
189	"""
190	if not len(attr):	1✔
191	# default values are not allowed when using *attr, so set a default here if none is supplied
192	attr = ("values",)	1✔
193	for attr_ in attr:	1✔
194	getattr(self, attr_).flags["WRITEABLE"] = False	1✔
195
196	def unlock(self, *attr: str) -> None:	1✔
197	"""Unlock attributes, rendering them editable.
198
199	Locked attributes cannot be overwritten, but can be unlocked with this function to make them editable.
200
201	Args:
202	*attr: any number of attributes can be passed here, all of which will be unlocked. Defaults to "values".
203
204	Examples:
205	>>> # lock the `values` attribute of `data`
206	>>> data.lock()
207	>>> data.values = [1, 2, 3] # will result in an AttributeError
208	>>> data.values[0] = 1 # will result in a RuntimeError
209	>>> data.unlock()
210	>>> data.values = [1, 2, 3]
211	>>> print(data.values)
212	[1,2,3]
213	>>> data.values[0] = 1 # will result in a RuntimeError
214	>>> print(data.values)
215	1
216	"""
217	if not len(attr):	×
218	# default values are not allowed when using *attr, so set a default here if none is supplied
UNCOV 219	attr = ("values",)	×
UNCOV 220	for attr_ in attr:	×
UNCOV 221	getattr(self, attr_).flags["WRITEABLE"] = True	×
222
223	@property	1✔
224	def locked(self) -> bool:	1✔
225	"""Return whether the values attribute is locked.
226
227	See lock().
228	"""
UNCOV 229	return not self.values.flags["WRITEABLE"]	×
230
231	@property	1✔
232	def loaded(self) -> bool:	1✔
233	"""Return whether the data was loaded from disk, or derived from elsewhere."""
234	return len(self.derived_from) == 0	1✔
235
236	def __len__(self):	1✔
UNCOV 237	return len(self.time)	×
238
239	def _sliced_copy(	1✔
240	self,
241	start_index: int,
242	end_index: int,
243	newlabel: str, # noqa: ARG002
244	) -> Self:
245	# TODO: check correct implementation
246	cls = self.__class__	1✔
247	time = np.copy(self.time[start_index:end_index])	1✔
248	values = np.copy(self.values[start_index:end_index])	1✔
249	description = f"Slice ({start_index}-{end_index}) of <{self.description}>"	1✔
250
251	return cls(	1✔
252	label=self.label, # TODO: newlabel gives errors
253	name=self.name,
254	unit=self.unit,
255	category=self.category,
256	description=description,
257	derived_from=[*self.derived_from, self],
258	time=time,
259	values=values,
260	sample_frequency=self.sample_frequency,
261	)

EIT-ALIVE / eitprocessing / 12120635114

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