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

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:
            msg = f"'{key}' does not match label '{item.label}'. Keys to Collection items must match their labels."
            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.
            # To force an overwrite, use __setitem__ and set `overwrite=True`.
            msg = (
                f"Item with label {key} already exists and cannot be overwritten. Please use a different label instead."
            )
            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	msg = f"'{key}' does not match label '{item.label}'. Keys to Collection items must match their labels."	1✔
82	raise KeyError(msg)	1✔
83
84	if not key:	1✔
85	key = item.label	1✔
86
87	if not overwrite and key in self:	1✔
88	# Generally it is not expected one would want to overwrite existing data with different/derived data.
89	# One should probably change the label instead over overwriting existing data.
90	# To force an overwrite, use __setitem__ and set `overwrite=True`.
91	msg = (	1✔
92	f"Item with label {key} already exists and cannot be overwritten. Please use a different label instead."
93	)
94	raise KeyError(msg)	1✔
95
96	def get_loaded_data(self) -> dict[str, V]:	1✔
97	"""Return all data that was directly loaded from disk."""
98	return {k: v for k, v in self.items() if v.loaded}	1✔
99
100	def get_data_derived_from(self, obj: V) -> dict[str, V]:	1✔
101	"""Return all data that was derived from a specific source."""
102	return {k: v for k, v in self.items() if any(obj is item for item in v.derived_from)}	1✔
103
104	def get_derived_data(self) -> dict[str, V]:	1✔
105	"""Return all data that was derived from any source."""
106	return {k: v for k, v in self.items() if v.derived_from}	1✔
107
108	def concatenate(self: Self, other: Self) -> Self:	1✔
109	"""Concatenate this collection with an equivalent collection.
110
111	Each item of self of concatenated with the item of other with the same key.
112	"""
113	self.isequivalent(other, raise_=True)	1✔
114
115	concatenated = self.__class__(self.data_type)	1✔
116	for key in self:	1✔
117	concatenated.add(self[key].concatenate(other[key]))	1✔
118
119	return concatenated	1✔
120
121	def select_by_time(	1✔
122	self,
123	start_time: float \| None,
124	end_time: float \| None,
125	start_inclusive: bool = True,
126	end_inclusive: bool = False,
127	) -> DataCollection:
128	"""Return a DataCollection containing sliced copies of the items."""
129	if self.data_type is IntervalData:	1!
UNCOV 130	return DataCollection(	×
131	self.data_type,
132	**{
133	k: v.select_by_time(
134	start_time=start_time,
135	end_time=end_time,
136	)
137	for k, v in self.items()
138	},
139	)
140
141	return DataCollection(	1✔
142	self.data_type,
143	**{
144	k: v.select_by_time(
145	start_time=start_time,
146	end_time=end_time,
147	start_inclusive=start_inclusive,
148	end_inclusive=end_inclusive,
149	)
150	for k, v in self.items()
151	},
152	)

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