24373313656

Committed 14 Apr 2026 12:00AM UTC coverage: 97.821% (-0.5%) from 98.345%

Build # 24373313656

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #67 from p2p-ld/ndarray-schema-proxies

allow proxies to be used with ndarray schema as input

Coverage Stats

15 of 17 new or added lines in 4 files covered. (88.24%)

23 existing lines in 11 files now uncovered.

1571 of 1606 relevant lines covered (97.82%)

9.78 hits per line

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

98.54

/src/numpydantic/testing/interfaces.py

from datetime import datetime, timezone
from pathlib import Path

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

p2p-ld / numpydantic / 24373313656

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