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

81.33

/eitprocessing/categories.py

import collections
import copy
import itertools
from collections.abc import Sequence
from functools import lru_cache
from importlib import resources

import yaml
from anytree import Node
from anytree.importer import DictImporter
from anytree.search import find_by_attr
from typing_extensions import Self

from eitprocessing.datahandling import DataContainer

COMPACT_YAML_FILE_MODULE = "eitprocessing.config"
COMPACT_YAML_FILE_NAME = "categories-compact.yaml"


class Category(Node):
    """Data category indicating what type of information is saved in an object.

    Categories are nested, where more specific categories are nested inside more general categories. The root category
    is simply named 'category'. Categories have a unique name within the entire tree.

    To check the existence of a category with name <name> within a category tree, either as subcategory or
    subsub(...)category, `category.has_subcategory("<name>")` can be used. The keyword `in` can be used as a shorthand.

    Example:
    ```
    >>> "tea" in category  # is the same as:
    True
    >>> category.has_subcategory("tea")
    True
    ```

    To select a subcategory, `category["<name>"]` can be used. You can select multiple categories at once. This will
    create a new tree with a temporary root, containing only the selected categories.

    Example:
    ```
    >>> foobar = categories["foo", "bar"]
    >>> print(foobar)
    Category('/temporary root')
    >>> print(foobar.children)
    (Category('/temporary root/foo'), Category('/temporary root/bar'))
    ```

    Categories can be hand-crafted, created from a dictionary or a YAML string. See [`anytree.DictionaryImporter`
    documentation](https://anytree.readthedocs.io/en/latest/importer/dictimporter.html) for more info on the dictionary
    format. [anytree documentation on YAML import/export](https://anytree.readthedocs.io/en/latest/tricks/yaml.html)
    shows the relevant structure of a normal YAML string.

    Categories also supports a compact YAML format, where each category containing a subcategory is a sequence.
    Categories without subcategories are strings in those sequences.

    ```yaml
    root:
    - sub 1 (without subcategories)
    - sub 2 (with subcategories):
      - sub a (without subcategories)
    ```

    Categories are read-only by default, as they should not be edited by the end-user during runtime. Consider editing
    the config file instead.

    Each type of data that is attached to an eitprocessing object should be categorized as one of the available types of
    data. This allows algorithms to check whether it can apply itself to the provided data, preventing misuse of
    algorithms.

    Example:
    ```
    >>> categories = get_default_categories()
    >>> print(categories)
    Category('/category')
    >>> print("pressure" in categories)
    True
    >>> categories["pressure"]
    Category('/category/physical measurement/pressure')
    ```
    """

    readonly = True

    def has_subcategory(self, subcategory: str) -> bool:
        """Check whether this category contains a subcategory.

        Returns True if the category and subcategory both exist. Returns False if the category exists, but the
        subcategory does not. Raises a ValueError

        Attr:
            category: the category to be checked as an ancestor of the subcategory. This category should exist.
            subcategory: the subcategory to be checked as a descendent of the category.

        Returns:
            bool: whether subcategory exists as a descendent of category.

        Raises:
            ValueError: if category does not exist.
        """
        if isinstance(subcategory, type(self)):
            return subcategory is self or subcategory in self.descendants

        return bool(find_by_attr(self, subcategory, name="name"))

    def __init__(self, name: str, parent: Self | None = None) -> None:
        super().__init__(name=name)
        with _IgnoreReadonly(self):
            self.parent = parent

    def __getitem__(self, name: str | tuple[str]):
        if isinstance(name, str):
            node = find_by_attr(self, name, name="name")
            if not node:
                msg = f"Category {name} does not exist."
                raise ValueError(msg)
            return node

        temporary_root = Category(name="temporary root")

        child_categories = [copy.deepcopy(self[name_]) for name_ in name]

        with _IgnoreReadonly(child_categories):
            temporary_root.children = child_categories

        return temporary_root

    def __contains__(self, item: str | Self):
        return self.has_subcategory(item)

    @classmethod
    def from_yaml(cls, string: str) -> Self:
        """Load categories from YAML file."""
        dict_ = yaml.load(string, Loader=yaml.SafeLoader)
        return cls.from_dict(dict_)

    @classmethod
    def from_compact_yaml(cls, string: str) -> Self:
        """Load categories from compact YAML file."""

        def parse_node(node: str | dict) -> Category:
            if isinstance(node, str):
                return Category(name=node)

            if isinstance(node, dict):
                if len(node) > 1:
                    msg = "Category data is malformed."
                    raise ValueError(msg)
                key = next(iter(node.keys()))
                category = Category(name=key)
                child_categories = [parse_node(child_node) for child_node in node[key]]

                with _IgnoreReadonly(child_categories):
                    category.children = child_categories

                return category

            msg = f"Supplied node should be str or dict, not {type(node)}."
            raise TypeError(msg)

        data = yaml.load(string, Loader=yaml.SafeLoader)
        return parse_node(data)

    @classmethod
    def from_dict(cls, dictionary: dict) -> Self:
        """Create categories from dictionary."""
        return DictImporter(nodecls=Category).import_(dictionary)

    def _pre_attach_children(self, children: list[Self]) -> None:
        """Checks for non-unique categories before adding them to an existing category tree."""
        for child in children:
            for node in [child, *child.descendants]:
                # Checks whether the names of children to be added and their descendents don't already exist in the
                # tree.
                if node.name in self.root:
                    msg = f"Can't add non-unique category {node.name}"
                    raise ValueError(msg)

        for child_a, child_b in itertools.permutations(children, 2):
            for child in [child_a, *child_a.descendants]:
                if child.name in child_b:
                    # Checks whether any child or their descendents exist in other children to be added
                    msg = f"Can't add non-unique category '{child.name}'"
                    raise ValueError(msg)

    def _pre_attach(self, parent: Self) -> None:  # noqa: ARG002
        if self.readonly:
            msg = "Can't attach read-only Category to another Category."
            raise RuntimeError(msg)

    def _pre_detach(self, parent: Self) -> None:  # noqa: ARG002
        if self.readonly:
            msg = "Can't detach read-only Category from another Category."
            raise RuntimeError(msg)

    def _check_unique(self, raise_: bool = False) -> bool:
        names = [self.name, *(node.name for node in self.descendants)]

        if len(names) == len(set(names)):
            return True

        if not raise_:
            return False

        count = collections.Counter(names)
        non_unique_names = [name for name, count in count.items() if count > 1]
        joined_names = ", ".join(f"'{name}'" for name in non_unique_names)
        msg = f"Some nodes have non-unique names: {joined_names}."
        raise ValueError(msg)


@lru_cache
def get_default_categories() -> Category:
    """Loads the default categories from file.

    This returns the categories used in the eitprocessing package. The root category is simply called 'root'. All other
    categories are subdivided into physical measurements, calculated values and others.

    This function is cached, meaning it only loads the data once, and returns the same object every time afterwards.
    """
    yaml_file_path = resources.files(COMPACT_YAML_FILE_MODULE).joinpath(COMPACT_YAML_FILE_NAME)
    with yaml_file_path.open("r") as fh:
        return Category.from_compact_yaml(fh.read())


def check_category(data: DataContainer, category: str, *, raise_: bool = False) -> bool:
    """Check whether the category of a dataset is a given category or one of it's subcategories.

    Example:
    >>> data = ContinuousData(..., category="impedance", ...)
    >>> check_category(data, "impedance")  # True
    >>> check_category(data, "pressure")  # False
    >>> check_category(data, "pressure", raise_=True)  # raises ValueError
    >>> check_category(data, "does not exist", raise_=False)  # raises ValueError

    Args:
        data: DataContainer object with a `category` attribute.
        category: Category to match the data category against. The data category will match this and all subcategories.
        raise_: Keyword only. Whether to raise an exception if the data is not a (sub)category.

    Returns:
        bool: Whether the data category matches.

    Raises:
        ValueError: If the provided category does not exist.
        ValueError: If the data category does not match the provided category.
    """
    categories = get_default_categories()

    if category not in categories:
        msg = f"Category '{category}' does not exist in the default categories."
        raise ValueError(msg)

    if data.category in categories[category]:
        return True

    if raise_:
        msg = f"`This method will only work on '{category}' data, not '{data.category}'."
        raise ValueError(msg)

    return False


class _IgnoreReadonly:
    """Context manager allowing temporarily ignoring the read-only attribute.

    For internal use only.

    Example:
    >>> foo = categories["foo"]
    >>> foo.parent = None  # raises RuntimeError
    >>> with _IgnoreReadonly(foo):
    >>>    foo.parent = None  # does not raise RuntimeError
    """

    items: Sequence[Category]

    def __init__(self, items: Category | Sequence[Category]):
        if not isinstance(items, Sequence):
            items = (items,)

        self.items = items

    def __enter__(self) -> None:
        for item in self.items:
            item.readonly = False

    def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
        for item in self.items:
            item.readonly = True

1	import collections	1✔
2	import copy	1✔
3	import itertools	1✔
4	from collections.abc import Sequence	1✔
5	from functools import lru_cache	1✔
6	from importlib import resources	1✔
7
8	import yaml	1✔
9	from anytree import Node	1✔
10	from anytree.importer import DictImporter	1✔
11	from anytree.search import find_by_attr	1✔
12	from typing_extensions import Self	1✔
13
14	from eitprocessing.datahandling import DataContainer	1✔
15
16	COMPACT_YAML_FILE_MODULE = "eitprocessing.config"	1✔
17	COMPACT_YAML_FILE_NAME = "categories-compact.yaml"	1✔
18
19
20	class Category(Node):	1✔
21	"""Data category indicating what type of information is saved in an object.
22
23	Categories are nested, where more specific categories are nested inside more general categories. The root category
24	is simply named 'category'. Categories have a unique name within the entire tree.
25
26	To check the existence of a category with name <name> within a category tree, either as subcategory or
27	subsub(...)category, `category.has_subcategory("<name>")` can be used. The keyword `in` can be used as a shorthand.
28
29	Example:
30	```
31	>>> "tea" in category # is the same as:
32	True
33	>>> category.has_subcategory("tea")
34	True
35	```
36
37	To select a subcategory, `category["<name>"]` can be used. You can select multiple categories at once. This will
38	create a new tree with a temporary root, containing only the selected categories.
39
40	Example:
41	```
42	>>> foobar = categories["foo", "bar"]
43	>>> print(foobar)
44	Category('/temporary root')
45	>>> print(foobar.children)
46	(Category('/temporary root/foo'), Category('/temporary root/bar'))
47	```
48
49	Categories can be hand-crafted, created from a dictionary or a YAML string. See [`anytree.DictionaryImporter`
50	documentation](https://anytree.readthedocs.io/en/latest/importer/dictimporter.html) for more info on the dictionary
51	format. [anytree documentation on YAML import/export](https://anytree.readthedocs.io/en/latest/tricks/yaml.html)
52	shows the relevant structure of a normal YAML string.
53
54	Categories also supports a compact YAML format, where each category containing a subcategory is a sequence.
55	Categories without subcategories are strings in those sequences.
56
57	```yaml
58	root:
59	- sub 1 (without subcategories)
60	- sub 2 (with subcategories):
61	- sub a (without subcategories)
62	```
63
64	Categories are read-only by default, as they should not be edited by the end-user during runtime. Consider editing
65	the config file instead.
66
67	Each type of data that is attached to an eitprocessing object should be categorized as one of the available types of
68	data. This allows algorithms to check whether it can apply itself to the provided data, preventing misuse of
69	algorithms.
70
71	Example:
72	```
73	>>> categories = get_default_categories()
74	>>> print(categories)
75	Category('/category')
76	>>> print("pressure" in categories)
77	True
78	>>> categories["pressure"]
79	Category('/category/physical measurement/pressure')
80	```
81	"""
82
83	readonly = True	1✔
84
85	def has_subcategory(self, subcategory: str) -> bool:	1✔
86	"""Check whether this category contains a subcategory.
87
88	Returns True if the category and subcategory both exist. Returns False if the category exists, but the
89	subcategory does not. Raises a ValueError
90
91	Attr:
92	category: the category to be checked as an ancestor of the subcategory. This category should exist.
93	subcategory: the subcategory to be checked as a descendent of the category.
94
95	Returns:
96	bool: whether subcategory exists as a descendent of category.
97
98	Raises:
99	ValueError: if category does not exist.
100	"""
101	if isinstance(subcategory, type(self)):	1✔
102	return subcategory is self or subcategory in self.descendants	1✔
103
104	return bool(find_by_attr(self, subcategory, name="name"))	1✔
105
106	def __init__(self, name: str, parent: Self \| None = None) -> None:	1✔
107	super().__init__(name=name)	1✔
108	with _IgnoreReadonly(self):	1✔
109	self.parent = parent	1✔
110
111	def __getitem__(self, name: str \| tuple[str]):	1✔
112	if isinstance(name, str):	1✔
113	node = find_by_attr(self, name, name="name")	1✔
114	if not node:	1✔
115	msg = f"Category {name} does not exist."	1✔
116	raise ValueError(msg)	1✔
117	return node	1✔
118
119	temporary_root = Category(name="temporary root")	1✔
120
121	child_categories = [copy.deepcopy(self[name_]) for name_ in name]	1✔
122
123	with _IgnoreReadonly(child_categories):	1✔
124	temporary_root.children = child_categories	1✔
125
126	return temporary_root	1✔
127
128	def __contains__(self, item: str \| Self):	1✔
129	return self.has_subcategory(item)	1✔
130
131	@classmethod	1✔
132	def from_yaml(cls, string: str) -> Self:	1✔
133	"""Load categories from YAML file."""
134	dict_ = yaml.load(string, Loader=yaml.SafeLoader)	1✔
135	return cls.from_dict(dict_)	1✔
136
137	@classmethod	1✔
138	def from_compact_yaml(cls, string: str) -> Self:	1✔
139	"""Load categories from compact YAML file."""
140
141	def parse_node(node: str \| dict) -> Category:	1✔
142	if isinstance(node, str):	1✔
143	return Category(name=node)	1✔
144
145	if isinstance(node, dict):	1!
146	if len(node) > 1:	1!
UNCOV 147	msg = "Category data is malformed."	×
UNCOV 148	raise ValueError(msg)	×
149	key = next(iter(node.keys()))	1✔
150	category = Category(name=key)	1✔
151	child_categories = [parse_node(child_node) for child_node in node[key]]	1✔
152
153	with _IgnoreReadonly(child_categories):	1✔
154	category.children = child_categories	1✔
155
156	return category	1✔
157
UNCOV 158	msg = f"Supplied node should be str or dict, not {type(node)}."	×
UNCOV 159	raise TypeError(msg)	×
160
161	data = yaml.load(string, Loader=yaml.SafeLoader)	1✔
162	return parse_node(data)	1✔
163
164	@classmethod	1✔
165	def from_dict(cls, dictionary: dict) -> Self:	1✔
166	"""Create categories from dictionary."""
167	return DictImporter(nodecls=Category).import_(dictionary)	1✔
168
169	def _pre_attach_children(self, children: list[Self]) -> None:	1✔
170	"""Checks for non-unique categories before adding them to an existing category tree."""
171	for child in children:	1✔
172	for node in [child, *child.descendants]:	1✔
173	# Checks whether the names of children to be added and their descendents don't already exist in the
174	# tree.
175	if node.name in self.root:	1!
176	msg = f"Can't add non-unique category {node.name}"	×
UNCOV 177	raise ValueError(msg)	×
178
179	for child_a, child_b in itertools.permutations(children, 2):	1✔
180	for child in [child_a, *child_a.descendants]:	1✔
181	if child.name in child_b:	1✔
182	# Checks whether any child or their descendents exist in other children to be added
183	msg = f"Can't add non-unique category '{child.name}'"	1✔
184	raise ValueError(msg)	1✔
185
186	def _pre_attach(self, parent: Self) -> None: # noqa: ARG002	1✔
187	if self.readonly:	1!
UNCOV 188	msg = "Can't attach read-only Category to another Category."	×
189	raise RuntimeError(msg)	×
190
191	def _pre_detach(self, parent: Self) -> None: # noqa: ARG002	1✔
192	if self.readonly:	1✔
193	msg = "Can't detach read-only Category from another Category."	1✔
194	raise RuntimeError(msg)	1✔
195
196	def _check_unique(self, raise_: bool = False) -> bool:	1✔
UNCOV 197	names = [self.name, *(node.name for node in self.descendants)]	×
198
UNCOV 199	if len(names) == len(set(names)):	×
UNCOV 200	return True	×
201
UNCOV 202	if not raise_:	×
UNCOV 203	return False	×
204
UNCOV 205	count = collections.Counter(names)	×
UNCOV 206	non_unique_names = [name for name, count in count.items() if count > 1]	×
UNCOV 207	joined_names = ", ".join(f"'{name}'" for name in non_unique_names)	×
UNCOV 208	msg = f"Some nodes have non-unique names: {joined_names}."	×
UNCOV 209	raise ValueError(msg)	×
210
211
212	@lru_cache	1✔
213	def get_default_categories() -> Category:	1✔
214	"""Loads the default categories from file.
215
216	This returns the categories used in the eitprocessing package. The root category is simply called 'root'. All other
217	categories are subdivided into physical measurements, calculated values and others.
218
219	This function is cached, meaning it only loads the data once, and returns the same object every time afterwards.
220	"""
221	yaml_file_path = resources.files(COMPACT_YAML_FILE_MODULE).joinpath(COMPACT_YAML_FILE_NAME)	1✔
222	with yaml_file_path.open("r") as fh:	1✔
223	return Category.from_compact_yaml(fh.read())	1✔
224
225
226	def check_category(data: DataContainer, category: str, *, raise_: bool = False) -> bool:	1✔
227	"""Check whether the category of a dataset is a given category or one of it's subcategories.
228
229	Example:
230	>>> data = ContinuousData(..., category="impedance", ...)
231	>>> check_category(data, "impedance") # True
232	>>> check_category(data, "pressure") # False
233	>>> check_category(data, "pressure", raise_=True) # raises ValueError
234	>>> check_category(data, "does not exist", raise_=False) # raises ValueError
235
236	Args:
237	data: DataContainer object with a `category` attribute.
238	category: Category to match the data category against. The data category will match this and all subcategories.
239	raise_: Keyword only. Whether to raise an exception if the data is not a (sub)category.
240
241	Returns:
242	bool: Whether the data category matches.
243
244	Raises:
245	ValueError: If the provided category does not exist.
246	ValueError: If the data category does not match the provided category.
247	"""
248	categories = get_default_categories()	1✔
249
250	if category not in categories:	1!
UNCOV 251	msg = f"Category '{category}' does not exist in the default categories."	×
UNCOV 252	raise ValueError(msg)	×
253
254	if data.category in categories[category]:	1✔
255	return True	1✔
256
257	if raise_:	1!
258	msg = f"`This method will only work on '{category}' data, not '{data.category}'."	1✔
259	raise ValueError(msg)	1✔
260
UNCOV 261	return False	×
262
263
264	class _IgnoreReadonly:	1✔
265	"""Context manager allowing temporarily ignoring the read-only attribute.
266
267	For internal use only.
268
269	Example:
270	>>> foo = categories["foo"]
271	>>> foo.parent = None # raises RuntimeError
272	>>> with _IgnoreReadonly(foo):
273	>>> foo.parent = None # does not raise RuntimeError
274	"""
275
276	items: Sequence[Category]	1✔
277
278	def __init__(self, items: Category \| Sequence[Category]):	1✔
279	if not isinstance(items, Sequence):	1✔
280	items = (items,)	1✔
281
282	self.items = items	1✔
283
284	def __enter__(self) -> None:	1✔
285	for item in self.items:	1✔
286	item.readonly = False	1✔
287
288	def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:	1✔
289	for item in self.items:	1✔
290	item.readonly = True	1✔

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