9136204652

Committed 18 May 2024 01:16AM UTC coverage: 100.0% (+0.4%) from 99.64%

Build # 9136204652

Build Type

push

github

Committed by

sneakers-the-rat

Commit Message

don't do coverage for stuff that shouldn't happen during testing

Run Details

554 of 554 relevant lines covered (100.0%)

2.98 hits per line

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

100.0

/src/numpydantic/interface/interface.py

"""
Base Interface metaclass
"""

from abc import ABC, abstractmethod
from operator import attrgetter
from typing import Any, Generic, Optional, Tuple, Type, TypeVar, Union

import numpy as np
from nptyping.shape_expression import check_shape
from pydantic import SerializationInfo

from numpydantic.exceptions import DtypeError, ShapeError
from numpydantic.types import DtypeType, NDArrayType, ShapeType

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


class Interface(ABC, Generic[T]):
    """
    Abstract parent class for interfaces to different array formats
    """

    input_types: Tuple[Any, ...]
    return_type: Type[T]
    priority: int = 0

    def __init__(self, shape: ShapeType, dtype: DtypeType) -> None:
        self.shape = shape
        self.dtype = dtype

    def validate(self, array: Any) -> T:
        """
        Validate input, returning final array type
        """
        array = self.before_validation(array)
        array = self.validate_dtype(array)
        array = self.validate_shape(array)
        array = self.after_validation(array)
        return array

    def before_validation(self, array: Any) -> NDArrayType:
        """
        Optional step pre-validation that coerces the input into a type that can be
        validated for shape and dtype

        Default method is a no-op
        """
        return array

    def validate_dtype(self, array: NDArrayType) -> NDArrayType:
        """
        Validate the dtype of the given array, returning it unmutated.

        Raises:
            :class:`~numpydantic.exceptions.DtypeError`
        """
        if self.dtype is Any:
            return array

        if isinstance(self.dtype, tuple):
            valid = array.dtype in self.dtype
        else:
            valid = array.dtype == self.dtype

        if not valid:
            raise DtypeError(f"Invalid dtype! expected {self.dtype}, got {array.dtype}")
        return array

    def validate_shape(self, array: NDArrayType) -> NDArrayType:
        """
        Validate the shape of the given array, returning it unmutated

        Raises:
            :class:`~numpydantic.exceptions.ShapeError`
        """
        if self.shape is Any:
            return array
        if not check_shape(array.shape, self.shape):
            raise ShapeError(
                f"Invalid shape! expected shape {self.shape.prepared_args}, "
                f"got shape {array.shape}"
            )
        return array

    def after_validation(self, array: NDArrayType) -> T:
        """
        Optional step post-validation that coerces the intermediate array type into the
        return type

        Default method is a no-op
        """
        return array

    @classmethod
    @abstractmethod
    def check(cls, array: Any) -> bool:
        """
        Method to check whether a given input applies to this interface
        """

    @classmethod
    @abstractmethod
    def enabled(cls) -> bool:
        """
        Check whether this array interface can be used (eg. its dependent packages are
        installed, etc.)
        """

    @classmethod
    def to_json(
        cls, array: Type[T], info: Optional[SerializationInfo] = None
    ) -> Union[list, dict]:
        """
        Convert an array of :attr:`.return_type` to a JSON-compatible format using
        base python types
        """
        if not isinstance(array, np.ndarray):  # pragma: no cover
            array = np.array(array)
        return array.tolist()

    @classmethod
    def interfaces(cls) -> Tuple[Type["Interface"], ...]:
        """
        Enabled interface subclasses
        """
        return tuple(
            sorted(
                [i for i in Interface.__subclasses__() if i.enabled()],
                key=attrgetter("priority"),
                reverse=True,
            )
        )

    @classmethod
    def return_types(cls) -> Tuple[NDArrayType, ...]:
        """Return types for all enabled interfaces"""
        return tuple([i.return_type for i in cls.interfaces()])

    @classmethod
    def input_types(cls) -> Tuple[Any, ...]:
        """Input types for all enabled interfaces"""
        in_types = []
        for iface in cls.interfaces():
            if isinstance(iface.input_types, (tuple, list)):
                in_types.extend(iface.input_types)
            else:  # pragma: no cover
                in_types.append(iface.input_types)

        return tuple(in_types)

    @classmethod
    def match(cls, array: Any) -> Type["Interface"]:
        """
        Find the interface that should be used for this array based on its input type
        """
        # first try and find a non-numpy interface, since the numpy interface
        # will try and load the array into memory in its check method
        interfaces = cls.interfaces()
        non_np_interfaces = [i for i in interfaces if i.__name__ != "NumpyInterface"]
        np_interface = [i for i in interfaces if i.__name__ == "NumpyInterface"][0]

        matches = [i for i in non_np_interfaces if i.check(array)]
        if len(matches) > 1:
            msg = f"More than one interface matches input {array}:\n"
            msg += "\n".join([f"  - {i}" for i in matches])
            raise ValueError(msg)
        elif len(matches) == 0:
            # now try the numpy interface
            if np_interface.check(array):
                return np_interface
            else:
                raise ValueError(f"No matching interfaces found for input {array}")
        else:
            return matches[0]

    @classmethod
    def match_output(cls, array: Any) -> Type["Interface"]:
        """
        Find the interface that should be used based on the output type -
        in the case that the output type differs from the input type, eg.
        the HDF5 interface, match an instantiated array for purposes of
        serialization to json, etc.
        """
        matches = [i for i in cls.interfaces() if isinstance(array, i.return_type)]
        if len(matches) > 1:
            msg = f"More than one interface matches output {array}:\n"
            msg += "\n".join([f"  - {i}" for i in matches])
            raise ValueError(msg)
        elif len(matches) == 0:
            raise ValueError(f"No matching interfaces found for output {array}")
        else:
            return matches[0]

1	"""
2	Base Interface metaclass
3	"""
4
5	from abc import ABC, abstractmethod	3✔
6	from operator import attrgetter	3✔
7	from typing import Any, Generic, Optional, Tuple, Type, TypeVar, Union	3✔
8
9	import numpy as np	3✔
10	from nptyping.shape_expression import check_shape	3✔
11	from pydantic import SerializationInfo	3✔
12
13	from numpydantic.exceptions import DtypeError, ShapeError	3✔
14	from numpydantic.types import DtypeType, NDArrayType, ShapeType	3✔
15
16	T = TypeVar("T", bound=NDArrayType)	3✔
17
18
19	class Interface(ABC, Generic[T]):	3✔
20	"""
21	Abstract parent class for interfaces to different array formats
22	"""
23
24	input_types: Tuple[Any, ...]	3✔
25	return_type: Type[T]	3✔
26	priority: int = 0	3✔
27
28	def __init__(self, shape: ShapeType, dtype: DtypeType) -> None:	3✔
29	self.shape = shape	3✔
30	self.dtype = dtype	3✔
31
32	def validate(self, array: Any) -> T:	3✔
33	"""
34	Validate input, returning final array type
35	"""
36	array = self.before_validation(array)	3✔
37	array = self.validate_dtype(array)	3✔
38	array = self.validate_shape(array)	3✔
39	array = self.after_validation(array)	3✔
40	return array	3✔
41
42	def before_validation(self, array: Any) -> NDArrayType:	3✔
43	"""
44	Optional step pre-validation that coerces the input into a type that can be
45	validated for shape and dtype
46
47	Default method is a no-op
48	"""
49	return array	3✔
50
51	def validate_dtype(self, array: NDArrayType) -> NDArrayType:	3✔
52	"""
53	Validate the dtype of the given array, returning it unmutated.
54
55	Raises:
56	:class:`~numpydantic.exceptions.DtypeError`
57	"""
58	if self.dtype is Any:	3✔
59	return array	3✔
60
61	if isinstance(self.dtype, tuple):	3✔
62	valid = array.dtype in self.dtype	3✔
63	else:
64	valid = array.dtype == self.dtype	3✔
65
66	if not valid:	3✔
67	raise DtypeError(f"Invalid dtype! expected {self.dtype}, got {array.dtype}")	3✔
68	return array	3✔
69
70	def validate_shape(self, array: NDArrayType) -> NDArrayType:	3✔
71	"""
72	Validate the shape of the given array, returning it unmutated
73
74	Raises:
75	:class:`~numpydantic.exceptions.ShapeError`
76	"""
77	if self.shape is Any:	3✔
78	return array	3✔
79	if not check_shape(array.shape, self.shape):	3✔
80	raise ShapeError(	3✔
81	f"Invalid shape! expected shape {self.shape.prepared_args}, "
82	f"got shape {array.shape}"
83	)
84	return array	3✔
85
86	def after_validation(self, array: NDArrayType) -> T:	3✔
87	"""
88	Optional step post-validation that coerces the intermediate array type into the
89	return type
90
91	Default method is a no-op
92	"""
93	return array	3✔
94
95	@classmethod	3✔
96	@abstractmethod	3✔
97	def check(cls, array: Any) -> bool:	3✔
98	"""
99	Method to check whether a given input applies to this interface
100	"""
101
102	@classmethod	3✔
103	@abstractmethod	3✔
104	def enabled(cls) -> bool:	3✔
105	"""
106	Check whether this array interface can be used (eg. its dependent packages are
107	installed, etc.)
108	"""
109
110	@classmethod	3✔
111	def to_json(	3✔
112	cls, array: Type[T], info: Optional[SerializationInfo] = None
113	) -> Union[list, dict]:
114	"""
115	Convert an array of :attr:`.return_type` to a JSON-compatible format using
116	base python types
117	"""
118	if not isinstance(array, np.ndarray): # pragma: no cover
119	array = np.array(array)
120	return array.tolist()	3✔
121
122	@classmethod	3✔
123	def interfaces(cls) -> Tuple[Type["Interface"], ...]:	3✔
124	"""
125	Enabled interface subclasses
126	"""
127	return tuple(	3✔
128	sorted(
129	[i for i in Interface.__subclasses__() if i.enabled()],
130	key=attrgetter("priority"),
131	reverse=True,
132	)
133	)
134
135	@classmethod	3✔
136	def return_types(cls) -> Tuple[NDArrayType, ...]:	3✔
137	"""Return types for all enabled interfaces"""
138	return tuple([i.return_type for i in cls.interfaces()])	3✔
139
140	@classmethod	3✔
141	def input_types(cls) -> Tuple[Any, ...]:	3✔
142	"""Input types for all enabled interfaces"""
143	in_types = []	3✔
144	for iface in cls.interfaces():	3✔
145	if isinstance(iface.input_types, (tuple, list)):	3✔
146	in_types.extend(iface.input_types)	3✔
147	else: # pragma: no cover
148	in_types.append(iface.input_types)
149
150	return tuple(in_types)	3✔
151
152	@classmethod	3✔
153	def match(cls, array: Any) -> Type["Interface"]:	3✔
154	"""
155	Find the interface that should be used for this array based on its input type
156	"""
157	# first try and find a non-numpy interface, since the numpy interface
158	# will try and load the array into memory in its check method
159	interfaces = cls.interfaces()	3✔
160	non_np_interfaces = [i for i in interfaces if i.__name__ != "NumpyInterface"]	3✔
161	np_interface = [i for i in interfaces if i.__name__ == "NumpyInterface"][0]	3✔
162
163	matches = [i for i in non_np_interfaces if i.check(array)]	3✔
164	if len(matches) > 1:	3✔
165	msg = f"More than one interface matches input {array}:\n"	3✔
166	msg += "\n".join([f" - {i}" for i in matches])	3✔
167	raise ValueError(msg)	3✔
168	elif len(matches) == 0:	3✔
169	# now try the numpy interface
170	if np_interface.check(array):	3✔
171	return np_interface	3✔
172	else:
173	raise ValueError(f"No matching interfaces found for input {array}")	3✔
174	else:
175	return matches[0]	3✔
176
177	@classmethod	3✔
178	def match_output(cls, array: Any) -> Type["Interface"]:	3✔
179	"""
180	Find the interface that should be used based on the output type -
181	in the case that the output type differs from the input type, eg.
182	the HDF5 interface, match an instantiated array for purposes of
183	serialization to json, etc.
184	"""
185	matches = [i for i in cls.interfaces() if isinstance(array, i.return_type)]	3✔
186	if len(matches) > 1:	3✔
187	msg = f"More than one interface matches output {array}:\n"	3✔
188	msg += "\n".join([f" - {i}" for i in matches])	3✔
189	raise ValueError(msg)	3✔
190	elif len(matches) == 0:	3✔
191	raise ValueError(f"No matching interfaces found for output {array}")	3✔
192	else:
193	return matches[0]	3✔

p2p-ld / numpydantic / 9136204652

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