16616223548

Committed 30 Jul 2025 07:27AM UTC coverage: 98.307% (+0.004%) from 98.303%

Build # 16616223548

Build Type

Pull #53

github

Committed by

web-flow

Commit Message

Merge af88e2559 into 1632c3ee7

Pull Request Pull Request #53: Dtype docs, fix scalar serialization

Run Details

21 of 21 new or added lines in 5 files covered. (100.0%)

3 existing lines in 2 files now uncovered.

1510 of 1536 relevant lines covered (98.31%)

17.41 hits per line

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

98.53

/src/numpydantic/testing/interfaces.py

from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Tuple

import cv2
import dask.array as da
import h5py
import numpy as np
import zarr
from pydantic import BaseModel

from numpydantic.interface import (
    DaskInterface,
    H5ArrayPath,
    H5Interface,
    NumpyInterface,
    VideoInterface,
    ZarrArrayPath,
    ZarrInterface,
)
from numpydantic.testing.helpers import InterfaceCase
from numpydantic.types import DtypeType, NDArrayType


class NumpyCase(InterfaceCase):
    """In-memory numpy array"""

    interface = NumpyInterface

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> np.ndarray:
        if array is not None:
            return np.array(array, dtype=dtype)
        elif issubclass(dtype, BaseModel):
            return np.full(shape=shape, fill_value=dtype(x=1))
        else:
            return np.zeros(shape=shape, dtype=dtype)


class _HDF5MetaCase(InterfaceCase):
    """Base case for hdf5 cases"""

    interface = H5Interface

    @classmethod
    def skip(cls, shape: Tuple[int, ...], dtype: DtypeType) -> bool:
        return issubclass(dtype, BaseModel)


class HDF5Case(_HDF5MetaCase):
    """HDF5 Array"""

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> Optional[H5ArrayPath]:
        if cls.skip(shape, dtype):  # pragma: no cover
            return None

        hdf5_file = path / "h5f.h5"
        array_path = "/" + "_".join([str(s) for s in shape]) + "__" + dtype.__name__
        generator = np.random.default_rng()

        if array is not None:
            data = np.array(array, dtype=dtype)
        elif dtype is str:
            data = generator.random(shape).astype(bytes)
        elif dtype is np.str_:
            data = generator.random(shape).astype("S32")
        elif dtype is datetime:
            data = np.empty(shape, dtype="S32")
            data.fill(datetime.now(timezone.utc).isoformat().encode("utf-8"))
        else:
            data = generator.random(shape).astype(dtype)

        h5path = H5ArrayPath(hdf5_file, array_path)

        with h5py.File(hdf5_file, "w") as h5f:
            _ = h5f.create_dataset(array_path, data=data)
        return h5path


class HDF5CompoundCase(_HDF5MetaCase):
    """HDF5 Array with a fake compound dtype"""

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> Optional[H5ArrayPath]:
        if cls.skip(shape, dtype):  # pragma: no cover
            return None

        hdf5_file = path / "h5f.h5"
        array_path = "/" + "_".join([str(s) for s in shape]) + "__" + dtype.__name__
        if array is not None:
            data = np.array(array, dtype=dtype)
        elif dtype in (str, np.str_):
            dt = np.dtype([("data", np.dtype("S10")), ("extra", "i8")])
            data = np.array([("hey", 0)] * np.prod(shape), dtype=dt).reshape(shape)
        elif dtype is datetime:
            dt = np.dtype([("data", np.dtype("S32")), ("extra", "i8")])
            data = np.array(
                [(datetime.now(timezone.utc).isoformat().encode("utf-8"), 0)]
                * np.prod(shape),
                dtype=dt,
            ).reshape(shape)
        else:
            dt = np.dtype([("data", dtype), ("extra", "i8")])
            data = np.zeros(shape, dtype=dt)
        h5path = H5ArrayPath(hdf5_file, array_path, "data")

        with h5py.File(hdf5_file, "w") as h5f:
            _ = h5f.create_dataset(array_path, data=data)
        return h5path


class DaskCase(InterfaceCase):
    """In-memory dask array"""

    interface = DaskInterface

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> da.Array:
        if array is not None:
            return da.array(array, dtype=dtype)
        if issubclass(dtype, BaseModel):
            return da.full(shape=shape, fill_value=dtype(x=1), chunks=-1)
        else:
            return da.zeros(shape=shape, dtype=dtype, chunks=10)


class _ZarrMetaCase(InterfaceCase):
    """Shared classmethods for zarr cases"""

    interface = ZarrInterface

    @classmethod
    def skip(cls, shape: Tuple[int, ...], dtype: DtypeType) -> bool:
        return issubclass(dtype, BaseModel) or dtype is str


class ZarrCase(_ZarrMetaCase):
    """In-memory zarr array"""

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> Optional[zarr.Array]:
        if array is not None:
            return zarr.array(array, dtype=dtype, chunks=-1)
        else:
            return zarr.zeros(shape=shape, dtype=dtype)


class ZarrDirCase(_ZarrMetaCase):
    """On-disk zarr array"""

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> Optional[zarr.Array]:
        store = zarr.DirectoryStore(str(path / "array.zarr"))
        if array is not None:
            return zarr.array(array, dtype=dtype, store=store, chunks=-1)
        else:
            return zarr.zeros(shape=shape, dtype=dtype, store=store)


class ZarrZipCase(_ZarrMetaCase):
    """Zarr zip store"""

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> Optional[zarr.Array]:
        store = zarr.ZipStore(str(path / "array.zarr"), mode="w")
        if array is not None:
            return zarr.array(array, dtype=dtype, store=store, chunks=-1)
        else:
            return zarr.zeros(shape=shape, dtype=dtype, store=store)


class ZarrNestedCase(_ZarrMetaCase):
    """Nested zarr array"""

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10),
        dtype: DtypeType = float,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> ZarrArrayPath:
        file = str(path / "nested.zarr")
        root = zarr.open(file, mode="w")
        subpath = "a/b/c"
        if array is not None:
            _ = root.array(subpath, array, dtype=dtype)
        else:
            _ = root.zeros(subpath, shape=shape, dtype=dtype)
        return ZarrArrayPath(file=file, path=subpath)


class VideoCase(InterfaceCase):
    """AVI video"""

    interface = VideoInterface

    @classmethod
    def make_array(
        cls,
        shape: Tuple[int, ...] = (10, 10, 10, 3),
        dtype: DtypeType = np.uint8,
        path: Optional[Path] = None,
        array: Optional[NDArrayType] = None,
    ) -> Optional[Path]:
        if cls.skip(shape, dtype):  # pragma: no cover
            return None

        if array is not None:
            array = np.array(array, dtype=np.uint8)
            shape = array.shape

        is_color = len(shape) == 4
        frames = shape[0]
        frame_shape = shape[1:]

        video_path = path / "test.avi"
        writer = cv2.VideoWriter(
            str(video_path),
            cv2.VideoWriter_fourcc(*"RGBA"),  # raw video for testing purposes
            30,
            (frame_shape[1], frame_shape[0]),
            is_color,
        )
        for i in range(frames):
            if array is not None:
                frame = array[i]
            else:
                # make fresh array every time bc opencv eats them
                frame = np.full(frame_shape, fill_value=i, dtype=np.uint8)
            writer.write(frame)
        writer.release()
        return video_path

    @classmethod
    def skip(cls, shape: Tuple[int, ...], dtype: DtypeType) -> bool:
        """
        We really can only handle 4 dimensional cases in 8-bit rn lol

        .. todo::

            Fix shape/writing for grayscale videos

        """
        if len(shape) != 4:
            return True

        # if len(shape) < 3 or len(shape) > 4:
        #     return True
        if dtype not in (int, np.uint8):
            return True
        # if we have a color video (ie. shape == 4, needs to be RGB)
        if len(shape) == 4 and shape[3] != 3:
            return True

1	from datetime import datetime, timezone	18✔
2	from pathlib import Path	18✔
3	from typing import Optional, Tuple	18✔
4
5	import cv2	18✔
6	import dask.array as da	18✔
7	import h5py	18✔
8	import numpy as np	18✔
9	import zarr	18✔
10	from pydantic import BaseModel	18✔
11
12	from numpydantic.interface import (	18✔
13	DaskInterface,
14	H5ArrayPath,
15	H5Interface,
16	NumpyInterface,
17	VideoInterface,
18	ZarrArrayPath,
19	ZarrInterface,
20	)
21	from numpydantic.testing.helpers import InterfaceCase	18✔
22	from numpydantic.types import DtypeType, NDArrayType	18✔
23
24
25	class NumpyCase(InterfaceCase):	18✔
26	"""In-memory numpy array"""
27
28	interface = NumpyInterface	18✔
29
30	@classmethod	18✔
31	def make_array(	18✔
32	cls,
33	shape: Tuple[int, ...] = (10, 10),
34	dtype: DtypeType = float,
35	path: Optional[Path] = None,
36	array: Optional[NDArrayType] = None,
37	) -> np.ndarray:
38	if array is not None:	18✔
39	return np.array(array, dtype=dtype)	18✔
40	elif issubclass(dtype, BaseModel):	18✔
41	return np.full(shape=shape, fill_value=dtype(x=1))	18✔
42	else:
43	return np.zeros(shape=shape, dtype=dtype)	18✔
44
45
46	class _HDF5MetaCase(InterfaceCase):	18✔
47	"""Base case for hdf5 cases"""
48
49	interface = H5Interface	18✔
50
51	@classmethod	18✔
52	def skip(cls, shape: Tuple[int, ...], dtype: DtypeType) -> bool:	18✔
53	return issubclass(dtype, BaseModel)	18✔
54
55
56	class HDF5Case(_HDF5MetaCase):	18✔
57	"""HDF5 Array"""
58
59	@classmethod	18✔
60	def make_array(	18✔
61	cls,
62	shape: Tuple[int, ...] = (10, 10),
63	dtype: DtypeType = float,
64	path: Optional[Path] = None,
65	array: Optional[NDArrayType] = None,
66	) -> Optional[H5ArrayPath]:
67	if cls.skip(shape, dtype): # pragma: no cover
68	return None
69
70	hdf5_file = path / "h5f.h5"	18✔
71	array_path = "/" + "_".join([str(s) for s in shape]) + "__" + dtype.__name__	18✔
72	generator = np.random.default_rng()	18✔
73
74	if array is not None:	18✔
UNCOV 75	data = np.array(array, dtype=dtype)	×
76	elif dtype is str:	18✔
77	data = generator.random(shape).astype(bytes)	18✔
78	elif dtype is np.str_:	18✔
79	data = generator.random(shape).astype("S32")	18✔
80	elif dtype is datetime:	18✔
81	data = np.empty(shape, dtype="S32")	18✔
82	data.fill(datetime.now(timezone.utc).isoformat().encode("utf-8"))	18✔
83	else:
84	data = generator.random(shape).astype(dtype)	18✔
85
86	h5path = H5ArrayPath(hdf5_file, array_path)	18✔
87
88	with h5py.File(hdf5_file, "w") as h5f:	18✔
89	_ = h5f.create_dataset(array_path, data=data)	18✔
90	return h5path	18✔
91
92
93	class HDF5CompoundCase(_HDF5MetaCase):	18✔
94	"""HDF5 Array with a fake compound dtype"""
95
96	@classmethod	18✔
97	def make_array(	18✔
98	cls,
99	shape: Tuple[int, ...] = (10, 10),
100	dtype: DtypeType = float,
101	path: Optional[Path] = None,
102	array: Optional[NDArrayType] = None,
103	) -> Optional[H5ArrayPath]:
104	if cls.skip(shape, dtype): # pragma: no cover
105	return None
106
107	hdf5_file = path / "h5f.h5"	18✔
108	array_path = "/" + "_".join([str(s) for s in shape]) + "__" + dtype.__name__	18✔
109	if array is not None:	18✔
UNCOV 110	data = np.array(array, dtype=dtype)	×
111	elif dtype in (str, np.str_):	18✔
112	dt = np.dtype([("data", np.dtype("S10")), ("extra", "i8")])	18✔
113	data = np.array([("hey", 0)] * np.prod(shape), dtype=dt).reshape(shape)	18✔
114	elif dtype is datetime:	18✔
115	dt = np.dtype([("data", np.dtype("S32")), ("extra", "i8")])	18✔
116	data = np.array(	18✔
117	[(datetime.now(timezone.utc).isoformat().encode("utf-8"), 0)]
118	* np.prod(shape),
119	dtype=dt,
120	).reshape(shape)
121	else:
122	dt = np.dtype([("data", dtype), ("extra", "i8")])	18✔
123	data = np.zeros(shape, dtype=dt)	18✔
124	h5path = H5ArrayPath(hdf5_file, array_path, "data")	18✔
125
126	with h5py.File(hdf5_file, "w") as h5f:	18✔
127	_ = h5f.create_dataset(array_path, data=data)	18✔
128	return h5path	18✔
129
130
131	class DaskCase(InterfaceCase):	18✔
132	"""In-memory dask array"""
133
134	interface = DaskInterface	18✔
135
136	@classmethod	18✔
137	def make_array(	18✔
138	cls,
139	shape: Tuple[int, ...] = (10, 10),
140	dtype: DtypeType = float,
141	path: Optional[Path] = None,
142	array: Optional[NDArrayType] = None,
143	) -> da.Array:
144	if array is not None:	18✔
145	return da.array(array, dtype=dtype)	18✔
146	if issubclass(dtype, BaseModel):	18✔
147	return da.full(shape=shape, fill_value=dtype(x=1), chunks=-1)	18✔
148	else:
149	return da.zeros(shape=shape, dtype=dtype, chunks=10)	18✔
150
151
152	class _ZarrMetaCase(InterfaceCase):	18✔
153	"""Shared classmethods for zarr cases"""
154
155	interface = ZarrInterface	18✔
156
157	@classmethod	18✔
158	def skip(cls, shape: Tuple[int, ...], dtype: DtypeType) -> bool:	18✔
159	return issubclass(dtype, BaseModel) or dtype is str	18✔
160
161
162	class ZarrCase(_ZarrMetaCase):	18✔
163	"""In-memory zarr array"""
164
165	@classmethod	18✔
166	def make_array(	18✔
167	cls,
168	shape: Tuple[int, ...] = (10, 10),
169	dtype: DtypeType = float,
170	path: Optional[Path] = None,
171	array: Optional[NDArrayType] = None,
172	) -> Optional[zarr.Array]:
173	if array is not None:	18✔
174	return zarr.array(array, dtype=dtype, chunks=-1)	18✔
175	else:
176	return zarr.zeros(shape=shape, dtype=dtype)	18✔
177
178
179	class ZarrDirCase(_ZarrMetaCase):	18✔
180	"""On-disk zarr array"""
181
182	@classmethod	18✔
183	def make_array(	18✔
184	cls,
185	shape: Tuple[int, ...] = (10, 10),
186	dtype: DtypeType = float,
187	path: Optional[Path] = None,
188	array: Optional[NDArrayType] = None,
189	) -> Optional[zarr.Array]:
190	store = zarr.DirectoryStore(str(path / "array.zarr"))	18✔
191	if array is not None:	18✔
192	return zarr.array(array, dtype=dtype, store=store, chunks=-1)	18✔
193	else:
194	return zarr.zeros(shape=shape, dtype=dtype, store=store)	18✔
195
196
197	class ZarrZipCase(_ZarrMetaCase):	18✔
198	"""Zarr zip store"""
199
200	@classmethod	18✔
201	def make_array(	18✔
202	cls,
203	shape: Tuple[int, ...] = (10, 10),
204	dtype: DtypeType = float,
205	path: Optional[Path] = None,
206	array: Optional[NDArrayType] = None,
207	) -> Optional[zarr.Array]:
208	store = zarr.ZipStore(str(path / "array.zarr"), mode="w")	18✔
209	if array is not None:	18✔
210	return zarr.array(array, dtype=dtype, store=store, chunks=-1)	18✔
211	else:
212	return zarr.zeros(shape=shape, dtype=dtype, store=store)	18✔
213
214
215	class ZarrNestedCase(_ZarrMetaCase):	18✔
216	"""Nested zarr array"""
217
218	@classmethod	18✔
219	def make_array(	18✔
220	cls,
221	shape: Tuple[int, ...] = (10, 10),
222	dtype: DtypeType = float,
223	path: Optional[Path] = None,
224	array: Optional[NDArrayType] = None,
225	) -> ZarrArrayPath:
226	file = str(path / "nested.zarr")	18✔
227	root = zarr.open(file, mode="w")	18✔
228	subpath = "a/b/c"	18✔
229	if array is not None:	18✔
230	_ = root.array(subpath, array, dtype=dtype)	18✔
231	else:
232	_ = root.zeros(subpath, shape=shape, dtype=dtype)	18✔
233	return ZarrArrayPath(file=file, path=subpath)	18✔
234
235
236	class VideoCase(InterfaceCase):	18✔
237	"""AVI video"""
238
239	interface = VideoInterface	18✔
240
241	@classmethod	18✔
242	def make_array(	18✔
243	cls,
244	shape: Tuple[int, ...] = (10, 10, 10, 3),
245	dtype: DtypeType = np.uint8,
246	path: Optional[Path] = None,
247	array: Optional[NDArrayType] = None,
248	) -> Optional[Path]:
249	if cls.skip(shape, dtype): # pragma: no cover
250	return None
251
252	if array is not None:	18✔
253	array = np.array(array, dtype=np.uint8)	18✔
254	shape = array.shape	18✔
255
256	is_color = len(shape) == 4	18✔
257	frames = shape[0]	18✔
258	frame_shape = shape[1:]	18✔
259
260	video_path = path / "test.avi"	18✔
261	writer = cv2.VideoWriter(	18✔
262	str(video_path),
263	cv2.VideoWriter_fourcc(*"RGBA"), # raw video for testing purposes
264	30,
265	(frame_shape[1], frame_shape[0]),
266	is_color,
267	)
268	for i in range(frames):	18✔
269	if array is not None:	18✔
270	frame = array[i]	18✔
271	else:
272	# make fresh array every time bc opencv eats them
273	frame = np.full(frame_shape, fill_value=i, dtype=np.uint8)	18✔
274	writer.write(frame)	18✔
275	writer.release()	18✔
276	return video_path	18✔
277
278	@classmethod	18✔
279	def skip(cls, shape: Tuple[int, ...], dtype: DtypeType) -> bool:	18✔
280	"""
281	We really can only handle 4 dimensional cases in 8-bit rn lol
282
283	.. todo::
284
285	Fix shape/writing for grayscale videos
286
287	"""
288	if len(shape) != 4:	18✔
289	return True	18✔
290
291	# if len(shape) < 3 or len(shape) > 4:
292	# return True
293	if dtype not in (int, np.uint8):	18✔
294	return True	18✔
295	# if we have a color video (ie. shape == 4, needs to be RGB)
296	if len(shape) == 4 and shape[3] != 3:	18✔
297	return True	18✔

p2p-ld / numpydantic / 16616223548

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