11745726274

Committed 08 Nov 2024 04:20PM UTC coverage: 82.129% (+2.1%) from 80.0%

Build # 11745726274

Build Type

push

github

Committed by

actions-user

Commit Message

Bump version: 1.4.4 → 1.4.5

Run Details

336 of 452 branches covered (74.34%)

Branch coverage included in aggregate %.

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

30 existing lines in 5 files now uncovered.

1346 of 1596 relevant lines covered (84.34%)

0.84 hits per line

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

97.14

/eitprocessing/datahandling/datacollection.py

from __future__ import annotations

from collections import UserDict
from typing import TYPE_CHECKING, Generic, TypeVar

from eitprocessing.datahandling.continuousdata import ContinuousData
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 HasTimeIndexer
from eitprocessing.datahandling.sparsedata import SparseData

if TYPE_CHECKING:
    from typing_extensions import Self


V = TypeVar("V", EITData, ContinuousData, SparseData, IntervalData)
V_classes = V.__constraints__


class DataCollection(Equivalence, UserDict, HasTimeIndexer, Generic[V]):
    """A collection of a single type of data with unique labels.

    A DataCollection functions largely as a dictionary, but requires a data_type argument, which must be one of the data
    containers existing in this package. When adding an item to the collection, the type of the value must match the
    data_type of the collection. Furthermore, the key has to match the attribute 'label' attached to the value.

    The convenience method `add()` adds an item by setting the key to `value.label`.

    Args:
        data_type: the data container stored in this collection.
    """

    data_type: type

    def __init__(self, data_type: type[V], *args, **kwargs):
        if not any(issubclass(data_type, cls) for cls in V_classes):
            msg = f"Type {data_type} not expected to be stored in a DataCollection."
            raise TypeError(msg)
        self.data_type = data_type
        super().__init__(*args, **kwargs)

    def __setitem__(self, __key: str, __value: V) -> None:
        self._check_item(__value, key=__key)
        return super().__setitem__(__key, __value)

    def add(self, *item: V, overwrite: bool = False) -> None:
        """Add one or multiple item(s) to the collection using the item label as the key."""
        for item_ in item:
            self._check_item(item_, overwrite=overwrite)
            super().__setitem__(item_.label, item_)

    def _check_item(
        self,
        item: V,
        key: str | None = None,
        overwrite: bool = False,
    ) -> None:
        """Check whether the item can be added to the collection.

        In order to be added to the collection, the data type of the item has to match the data type set in the
        collection. They key that is used to store the item in the collection has to match the label of the item itself.
        By default, existing keys can not be overridden.

        Args:
            item: Object to be added to the collection.
            key: Key of the item. Has to match `item.label`.
            overwrite: If False, the key can not already exist in the collection. Set to True to allow overwriting an
            existing object in the collection.

        Raises:
            TypeError: If the type of the item does not match the type set in the collection.
            KeyError: If the key does not match `item.label`, or when the key already exists in de collection and
            overwrite is set to False.
        """
        if not isinstance(item, self.data_type):
            msg = f"Type of `data` is {type(item)}, not '{self.data_type}'"
            raise TypeError(msg)

        if key and key != item.label:
            # It is expected that an item in this collection has a key equal to the label of the value.
            msg = f"'{key}' does not match label '{item.label}'."
            raise KeyError(msg)

        if not key:
            key = item.label

        if not overwrite and key in self:
            # Generally it is not expected one would want to overwrite existing data with different/derived data. One
            # should probably change the label instead over overwriting existing data.
            msg = f"Item with label {key} already exists. Use `overwrite=True` to overwrite."
            raise KeyError(msg)

    def get_loaded_data(self) -> dict[str, V]:
        """Return all data that was directly loaded from disk."""
        return {k: v for k, v in self.items() if v.loaded}

    def get_data_derived_from(self, obj: V) -> dict[str, V]:
        """Return all data that was derived from a specific source."""
        return {k: v for k, v in self.items() if any(obj is item for item in v.derived_from)}

    def get_derived_data(self) -> dict[str, V]:
        """Return all data that was derived from any source."""
        return {k: v for k, v in self.items() if v.derived_from}

    def concatenate(self: Self, other: Self) -> Self:
        """Concatenate this collection with an equivalent collection.

        Each item of self of concatenated with the item of other with the same key.
        """
        self.isequivalent(other, raise_=True)

        concatenated = self.__class__(self.data_type)
        for key in self:
            concatenated.add(self[key].concatenate(other[key]))

        return concatenated

    def select_by_time(
        self,
        start_time: float | None,
        end_time: float | None,
        start_inclusive: bool = True,
        end_inclusive: bool = False,
    ) -> DataCollection:
        """Return a DataCollection containing sliced copies of the items."""
        if self.data_type is IntervalData:
            return DataCollection(
                self.data_type,
                **{
                    k: v.select_by_time(
                        start_time=start_time,
                        end_time=end_time,
                    )
                    for k, v in self.items()
                },
            )

        return DataCollection(
            self.data_type,
            **{
                k: v.select_by_time(
                    start_time=start_time,
                    end_time=end_time,
                    start_inclusive=start_inclusive,
                    end_inclusive=end_inclusive,
                )
                for k, v in self.items()
            },
        )

1	from __future__ import annotations	1✔
2
3	from collections import UserDict	1✔
4	from typing import TYPE_CHECKING, Generic, TypeVar	1✔
5
6	from eitprocessing.datahandling.continuousdata import ContinuousData	1✔
7	from eitprocessing.datahandling.eitdata import EITData	1✔
8	from eitprocessing.datahandling.intervaldata import IntervalData	1✔
9	from eitprocessing.datahandling.mixins.equality import Equivalence	1✔
10	from eitprocessing.datahandling.mixins.slicing import HasTimeIndexer	1✔
11	from eitprocessing.datahandling.sparsedata import SparseData	1✔
12
13	if TYPE_CHECKING:
14	from typing_extensions import Self
15
16
17	V = TypeVar("V", EITData, ContinuousData, SparseData, IntervalData)	1✔
18	V_classes = V.__constraints__	1✔
19
20
21	class DataCollection(Equivalence, UserDict, HasTimeIndexer, Generic[V]):	1✔
22	"""A collection of a single type of data with unique labels.
23
24	A DataCollection functions largely as a dictionary, but requires a data_type argument, which must be one of the data
25	containers existing in this package. When adding an item to the collection, the type of the value must match the
26	data_type of the collection. Furthermore, the key has to match the attribute 'label' attached to the value.
27
28	The convenience method `add()` adds an item by setting the key to `value.label`.
29
30	Args:
31	data_type: the data container stored in this collection.
32	"""
33
34	data_type: type	1✔
35
36	def __init__(self, data_type: type[V], args, *kwargs):	1✔
37	if not any(issubclass(data_type, cls) for cls in V_classes):	1✔
38	msg = f"Type {data_type} not expected to be stored in a DataCollection."	1✔
39	raise TypeError(msg)	1✔
40	self.data_type = data_type	1✔
41	super().__init__(args, *kwargs)	1✔
42
43	def __setitem__(self, __key: str, __value: V) -> None:	1✔
44	self._check_item(__value, key=__key)	1✔
45	return super().__setitem__(__key, __value)	1✔
46
47	def add(self, *item: V, overwrite: bool = False) -> None:	1✔
48	"""Add one or multiple item(s) to the collection using the item label as the key."""
49	for item_ in item:	1✔
50	self._check_item(item_, overwrite=overwrite)	1✔
51	super().__setitem__(item_.label, item_)	1✔
52
53	def _check_item(	1✔
54	self,
55	item: V,
56	key: str \| None = None,
57	overwrite: bool = False,
58	) -> None:
59	"""Check whether the item can be added to the collection.
60
61	In order to be added to the collection, the data type of the item has to match the data type set in the
62	collection. They key that is used to store the item in the collection has to match the label of the item itself.
63	By default, existing keys can not be overridden.
64
65	Args:
66	item: Object to be added to the collection.
67	key: Key of the item. Has to match `item.label`.
68	overwrite: If False, the key can not already exist in the collection. Set to True to allow overwriting an
69	existing object in the collection.
70
71	Raises:
72	TypeError: If the type of the item does not match the type set in the collection.
73	KeyError: If the key does not match `item.label`, or when the key already exists in de collection and
74	overwrite is set to False.
75	"""
76	if not isinstance(item, self.data_type):	1✔
77	msg = f"Type of `data` is {type(item)}, not '{self.data_type}'"	1✔
78	raise TypeError(msg)	1✔
79
80	if key and key != item.label:	1✔
81	# It is expected that an item in this collection has a key equal to the label of the value.
82	msg = f"'{key}' does not match label '{item.label}'."	1✔
83	raise KeyError(msg)	1✔
84
85	if not key:	1✔
86	key = item.label	1✔
87
88	if not overwrite and key in self:	1✔
89	# Generally it is not expected one would want to overwrite existing data with different/derived data. One
90	# should probably change the label instead over overwriting existing data.
91	msg = f"Item with label {key} already exists. Use `overwrite=True` to overwrite."	1✔
92	raise KeyError(msg)	1✔
93
94	def get_loaded_data(self) -> dict[str, V]:	1✔
95	"""Return all data that was directly loaded from disk."""
96	return {k: v for k, v in self.items() if v.loaded}	1✔
97
98	def get_data_derived_from(self, obj: V) -> dict[str, V]:	1✔
99	"""Return all data that was derived from a specific source."""
100	return {k: v for k, v in self.items() if any(obj is item for item in v.derived_from)}	1✔
101
102	def get_derived_data(self) -> dict[str, V]:	1✔
103	"""Return all data that was derived from any source."""
104	return {k: v for k, v in self.items() if v.derived_from}	1✔
105
106	def concatenate(self: Self, other: Self) -> Self:	1✔
107	"""Concatenate this collection with an equivalent collection.
108
109	Each item of self of concatenated with the item of other with the same key.
110	"""
111	self.isequivalent(other, raise_=True)	1✔
112
113	concatenated = self.__class__(self.data_type)	1✔
114	for key in self:	1✔
115	concatenated.add(self[key].concatenate(other[key]))	1✔
116
117	return concatenated	1✔
118
119	def select_by_time(	1✔
120	self,
121	start_time: float \| None,
122	end_time: float \| None,
123	start_inclusive: bool = True,
124	end_inclusive: bool = False,
125	) -> DataCollection:
126	"""Return a DataCollection containing sliced copies of the items."""
127	if self.data_type is IntervalData:	1!
UNCOV 128	return DataCollection(	×
129	self.data_type,
130	**{
131	k: v.select_by_time(
132	start_time=start_time,
133	end_time=end_time,
134	)
135	for k, v in self.items()
136	},
137	)
138
139	return DataCollection(	1✔
140	self.data_type,
141	**{
142	k: v.select_by_time(
143	start_time=start_time,
144	end_time=end_time,
145	start_inclusive=start_inclusive,
146	end_inclusive=end_inclusive,
147	)
148	for k, v in self.items()
149	},
150	)

EIT-ALIVE / eitprocessing / 11745726274

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