27054202885

Committed 06 Jun 2026 05:49AM UTC coverage: 97.114% (-0.7%) from 97.821%

Build # 27054202885

Build Type

Pull #69

github

Committed by

web-flow

Commit Message

Merge c1c4272d0 into 952a740e0

Pull Request Pull Request #69: aw shit it's mypy plugin time

Coverage Stats

376 of 403 new or added lines in 14 files covered. (93.3%)

4 existing lines in 1 file now uncovered.

1918 of 1975 relevant lines covered (97.11%)

6.78 hits per line

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

98.92

/src/numpydantic/interface/hdf5.py

"""
Interfaces for HDF5 Datasets

.. note::

    HDF5 arrays are accessed through a proxy class :class:`.H5Proxy` .
    Getting/setting values should work as normal, **except** that setting
    values on nested views is impossible -

    Specifically this doesn't work:

    .. code-block:: python

        my_model.array[0][0] = 1

    But this does work:

    .. code-block:: python

        my_model.array[0,0] = 1

    To have direct access to the hdf5 dataset, use the
    :meth:`.H5Proxy.open` method.

Datetimes
---------

Datetimes are supported as a dtype annotation, but currently they must be stored
as ``S32`` isoformatted byte strings (timezones optional) like:

.. code-block:: python

    import h5py
    from datetime import datetime
    import numpy as np
    data = np.array([datetime.now().isoformat().encode('utf-8')], dtype="S32")
    h5f = h5py.File('test.hdf5', 'w')
    h5f.create_dataset('data', data=data)

"""

import sys
from collections.abc import Iterable
from datetime import datetime
from pathlib import Path
from typing import Any, NamedTuple, TypeVar

import numpy as np
from pydantic import SerializationInfo

from numpydantic.interface.interface import Interface, JsonDict, Proxy
from numpydantic.types import DtypeType, NDArrayType

try:
    import h5py
except ImportError:  # pragma: no cover
    h5py = None

if sys.version_info.minor >= 10:
    from typing import TypeAlias
else:
    from typing import TypeAlias

H5Arraylike: TypeAlias = tuple[Path | str, str]

T = TypeVar("T")


class H5ArrayPath(NamedTuple):
    """Location specifier for arrays within an HDF5 file"""

    file: Path | str
    """Location of HDF5 file"""
    path: str
    """Path within the HDF5 file"""
    field: str | list[str] | None = None
    """Refer to a specific field within a compound dtype"""


class H5JsonDict(JsonDict):
    """Round-trip Json-able version of an HDF5 dataset"""

    file: str
    path: str
    field: str | None = None

    def to_array_input(self) -> H5ArrayPath:
        """Construct an :class:`.H5ArrayPath`"""
        return H5ArrayPath(
            **{k: v for k, v in self.model_dump().items() if k in H5ArrayPath._fields}
        )


class H5Proxy(Proxy):
    """
    Proxy class to mimic numpy-like array behavior with an HDF5 array

    The attribute and item access methods only open the file for the duration of the
    method, making it less perilous to share this object between threads and processes.

    This class attempts to be a passthrough class to a :class:`h5py.Dataset` object,
    including its attributes and item getters/setters.

    When using read-only methods, no locking is attempted (beyond the HDF5 defaults),
    but when using the write methods (setting an array value), try and use the
    ``locking`` methods of :class:`h5py.File` .

    Args:
        file (pathlib.Path | str): Location of hdf5 file on filesystem
        path (str): Path to array within hdf5 file
        field (str, list[str]): Optional - refer to a specific field within
            a compound dtype
        annotation_dtype (dtype): Optional - the dtype of our type annotation
    """

    def __init__(
        self,
        file: Path | str,
        path: str,
        field: str | list[str] | None = None,
        annotation_dtype: DtypeType | None = None,
    ):
        self._h5f = None
        self.file = Path(file).resolve()
        self.path = path
        self.field = field
        self._annotation_dtype = annotation_dtype
        self._h5arraypath = H5ArrayPath(self.file, self.path, self.field)

    @classmethod
    def proxy_for(cls) -> type[Interface]:
        """Declare this class as a proxy for the H5Interface"""
        return H5Interface

    def array_exists(self) -> bool:
        """
        Check that there is in fact an array at :attr:`.H5Proxy.path`
        within :attr:`.H5Proxy.file`
        """
        with h5py.File(self.file, "r") as h5f:
            obj = h5f.get(self.path)
            return obj is not None

    @classmethod
    def from_h5array(cls, h5array: H5ArrayPath) -> "H5Proxy":
        """Instantiate using :class:`.H5ArrayPath`"""
        return H5Proxy(file=h5array.file, path=h5array.path, field=h5array.field)

    @property
    def dtype(self) -> np.dtype:
        """
        Get dtype of array, using :attr:`.field` if present
        """
        with h5py.File(self.file, "r") as h5f:
            obj = h5f.get(self.path)
            if self.field is None:
                return obj.dtype
            else:
                return obj.dtype[self.field]

    def __array__(self) -> np.ndarray:
        """To a numpy array"""
        with h5py.File(self.file, "r") as h5f:
            obj = h5f.get(self.path)
            return obj[:]

    def __getattr__(self, item: str):
        if item == "__name__":
            # special case for H5Proxies that don't refer to a real file during testing
            return "H5Proxy"
        with h5py.File(self.file, "r") as h5f:
            obj = h5f.get(self.path)
            val = getattr(obj, item)
            return val

    def __getitem__(
        self, item: int | slice | tuple[int | slice, ...]
    ) -> np.ndarray | DtypeType:
        with h5py.File(self.file, "r") as h5f:
            obj = h5f.get(self.path)
            # handle compound dtypes
            if self.field is not None:
                # handle compound string dtype
                if encoding := h5py.h5t.check_string_dtype(obj.dtype[self.field]):
                    if isinstance(item, tuple):
                        item = (*item, self.field)
                    else:
                        item = (item, self.field)

                    try:
                        # single string
                        val = obj[item].decode(encoding.encoding)
                        if self._annotation_dtype is np.datetime64:
                            return np.datetime64(val)
                        else:
                            return val
                    except AttributeError:
                        # numpy array of bytes
                        val = np.char.decode(obj[item], encoding=encoding.encoding)
                        if self._annotation_dtype is np.datetime64:
                            return val.astype(np.datetime64)
                        else:
                            return val
                # normal compound type
                else:
                    obj = obj.fields(self.field)
            else:
                if h5py.h5t.check_string_dtype(obj.dtype):
                    obj = obj.asstr()

            val = obj[item]
            if self._annotation_dtype is np.datetime64:
                if isinstance(val, str):
                    return np.datetime64(val)
                else:
                    return val.astype(np.datetime64)
            else:
                return val

    def __setitem__(
        self,
        key: int | slice | tuple[int | slice, ...],
        value: int | float | datetime | np.ndarray,
    ):
        # TODO: Make a generalized value serdes system instead of ad-hoc type conversion
        value = self._serialize_datetime(value)
        with h5py.File(self.file, "r+", locking=True) as h5f:
            obj = h5f.get(self.path)
            if self.field is None:
                obj[key] = value
            else:
                if isinstance(key, tuple):
                    key = (*key, self.field)
                    obj[key] = value
                else:
                    obj[key, self.field] = value

    def __len__(self) -> int:
        """self.shape[0]"""
        return self.shape[0]

    def __eq__(self, other: "H5Proxy") -> bool:
        """
        Check that we are referring to the same hdf5 array
        """
        if isinstance(other, H5Proxy):
            return self._h5arraypath == other._h5arraypath
        else:
            raise ValueError("Can only compare equality of two H5Proxies")

    def open(self, mode: str = "r") -> "h5py.Dataset":
        """
        Return the opened :class:`h5py.Dataset` object

        You must remember to close the associated file with :meth:`.close`
        """
        if self._h5f is None:
            self._h5f = h5py.File(self.file, mode)
        return self._h5f.get(self.path)

    def close(self) -> None:
        """
        Close the :class:`h5py.File` object left open when returning the dataset with
        :meth:`.open`
        """
        if self._h5f is not None:
            self._h5f.close()
        self._h5f = None

    def _serialize_datetime(self, v: T | datetime) -> T | bytes:
        """
        Convert a datetime into a bytestring
        """
        if self._annotation_dtype is np.datetime64:
            if not isinstance(v, Iterable):
                v = [v]
            v = np.array(v).astype("S32")
        return v


class H5Interface(Interface):
    """
    Interface for Arrays stored as datasets within an HDF5 file.

    Takes a :class:`.H5ArrayPath` specifier to select a :class:`h5py.Dataset` from a
    :class:`h5py.File` and returns a :class:`.H5Proxy` class that acts like a
    passthrough numpy-like interface to the dataset.
    """

    name = "hdf5"
    input_types = (H5ArrayPath, H5Arraylike, H5Proxy)
    return_type = H5Proxy
    json_model = H5JsonDict

    @classmethod
    def enabled(cls) -> bool:
        """Check whether h5py can be imported"""
        return h5py is not None

    @classmethod
    def check(cls, array: H5ArrayPath | H5Arraylike) -> bool:
        """
        Check that the given array is a :class:`.H5ArrayPath` or something that
        resembles one.
        """
        if isinstance(array, (H5ArrayPath, H5Proxy)):
            return True

        if isinstance(array, dict):
            if array.get("type", False) == cls.name:
                return True
            # continue checking if dict contains an hdf5 file
            file = array.get("file", "")
            array = (file, "")

        if isinstance(array, (tuple, list)) and len(array) in (2, 3):
            # check that the first arg is an hdf5 file
            try:
                file = Path(array[0])
            except TypeError:
                # not a path, we don't apply.
                return False

            if not file.exists():
                return False

            # hdf5 files are commonly given odd suffixes,
            # so we just try and open it and see what happens
            try:
                with h5py.File(file, "r"):
                    # don't check that the array exists and raise here,
                    # this check is just for whether the validator applies or not.
                    pass
                return True
            except (FileNotFoundError, OSError):
                return False

        return False

    def before_validation(self, array: Any) -> NDArrayType:
        """Create an :class:`.H5Proxy` to use throughout validation"""
        if isinstance(array, H5ArrayPath):
            array = H5Proxy.from_h5array(h5array=array)
        elif isinstance(array, H5Proxy):
            # nothing to do, already proxied
            pass
        elif isinstance(array, (tuple, list)) and len(array) == 2:  # pragma: no cover
            array = H5Proxy(file=array[0], path=array[1])
        elif isinstance(array, (tuple, list)) and len(array) == 3:
            array = H5Proxy(file=array[0], path=array[1], field=array[2])
        else:  # pragma: no cover
            # this should never happen really since `check` confirms this before
            # we'd reach here, but just to complete the if else...
            raise ValueError(
                "Need to specify a file and a path within an HDF5 file to use the HDF5 "
                "Interface"
            )
        array._annotation_dtype = self.dtype

        if not array.array_exists():
            raise ValueError(
                f"HDF5 file located at {array.file}, "
                f"but no array found at {array.path}"
            )

        return array

    def get_dtype(self, array: NDArrayType) -> DtypeType:
        """
        Get the dtype from the input array

        Subclasses to correctly handle
        """
        if h5py.h5t.check_string_dtype(array.dtype):
            # check for datetimes
            try:
                if array[0].dtype.type is np.datetime64:
                    return np.datetime64
                else:
                    return str
            except (AttributeError, TypeError):  # pragma: no cover
                # it's not a datetime, but it is some kind of string
                return str
            except (IndexError, ValueError):
                # if the dataset is empty, we can't tell if something is a datetime
                # or not, so we just tell the validation method what it wants to hear
                if self.dtype in (np.datetime64, str):
                    return self.dtype
                else:
                    return str
        else:
            return array.dtype

    @classmethod
    def to_json(cls, array: H5Proxy, info: SerializationInfo | None = None) -> dict:
        """
        Render HDF5 array as JSON

        If ``round_trip == True``, we dump just the proxy info, a dictionary like:

        * ``file``: :attr:`.H5Proxy.file`
        * ``path``: :attr:`.H5Proxy.path`
        * ``attrs``: Any HDF5 attributes on the dataset
        * ``array``: The array as a list of lists

        Otherwise, we dump the array as a list of lists
        """
        if info.round_trip:
            as_json = {
                "type": cls.name,
            }
            as_json.update(array._h5arraypath._asdict())
        else:
            try:
                dset = array.open()
                as_json = dset[:].tolist()
            finally:
                array.close()

        return as_json

1	"""
2	Interfaces for HDF5 Datasets
3
4	.. note::
5
6	HDF5 arrays are accessed through a proxy class :class:`.H5Proxy` .
7	Getting/setting values should work as normal, except that setting
8	values on nested views is impossible -
9
10	Specifically this doesn't work:
11
12	.. code-block:: python
13
14	my_model.array[0][0] = 1
15
16	But this does work:
17
18	.. code-block:: python
19
20	my_model.array[0,0] = 1
21
22	To have direct access to the hdf5 dataset, use the
23	:meth:`.H5Proxy.open` method.
24
25	Datetimes
26	---------
27
28	Datetimes are supported as a dtype annotation, but currently they must be stored
29	as ``S32`` isoformatted byte strings (timezones optional) like:
30
31	.. code-block:: python
32
33	import h5py
34	from datetime import datetime
35	import numpy as np
36	data = np.array([datetime.now().isoformat().encode('utf-8')], dtype="S32")
37	h5f = h5py.File('test.hdf5', 'w')
38	h5f.create_dataset('data', data=data)
39
40	"""
41
42	import sys	7✔
43	from collections.abc import Iterable	7✔
44	from datetime import datetime	7✔
45	from pathlib import Path	7✔
46	from typing import Any, NamedTuple, TypeVar	7✔
47
48	import numpy as np	7✔
49	from pydantic import SerializationInfo	7✔
50
51	from numpydantic.interface.interface import Interface, JsonDict, Proxy	7✔
52	from numpydantic.types import DtypeType, NDArrayType	7✔
53
54	try:	7✔
55	import h5py	7✔
56	except ImportError: # pragma: no cover
57	h5py = None
58
59	if sys.version_info.minor >= 10:	7✔
60	from typing import TypeAlias	7✔
61	else:
62	from typing import TypeAlias	×
63
64	H5Arraylike: TypeAlias = tuple[Path \| str, str]	7✔
65
66	T = TypeVar("T")	7✔
67
68
69	class H5ArrayPath(NamedTuple):	7✔
70	"""Location specifier for arrays within an HDF5 file"""
71
72	file: Path \| str	7✔
73	"""Location of HDF5 file"""	7✔
74	path: str	7✔
75	"""Path within the HDF5 file"""	7✔
76	field: str \| list[str] \| None = None	7✔
77	"""Refer to a specific field within a compound dtype"""	7✔
78
79
80	class H5JsonDict(JsonDict):	7✔
81	"""Round-trip Json-able version of an HDF5 dataset"""
82
83	file: str	7✔
84	path: str	7✔
85	field: str \| None = None	7✔
86
87	def to_array_input(self) -> H5ArrayPath:	7✔
88	"""Construct an :class:`.H5ArrayPath`"""
89	return H5ArrayPath(	7✔
90	**{k: v for k, v in self.model_dump().items() if k in H5ArrayPath._fields}
91	)
92
93
94	class H5Proxy(Proxy):	7✔
95	"""
96	Proxy class to mimic numpy-like array behavior with an HDF5 array
97
98	The attribute and item access methods only open the file for the duration of the
99	method, making it less perilous to share this object between threads and processes.
100
101	This class attempts to be a passthrough class to a :class:`h5py.Dataset` object,
102	including its attributes and item getters/setters.
103
104	When using read-only methods, no locking is attempted (beyond the HDF5 defaults),
105	but when using the write methods (setting an array value), try and use the
106	``locking`` methods of :class:`h5py.File` .
107
108	Args:
109	file (pathlib.Path \| str): Location of hdf5 file on filesystem
110	path (str): Path to array within hdf5 file
111	field (str, list[str]): Optional - refer to a specific field within
112	a compound dtype
113	annotation_dtype (dtype): Optional - the dtype of our type annotation
114	"""
115
116	def __init__(	7✔
117	self,
118	file: Path \| str,
119	path: str,
120	field: str \| list[str] \| None = None,
121	annotation_dtype: DtypeType \| None = None,
122	):
123	self._h5f = None	7✔
124	self.file = Path(file).resolve()	7✔
125	self.path = path	7✔
126	self.field = field	7✔
127	self._annotation_dtype = annotation_dtype	7✔
128	self._h5arraypath = H5ArrayPath(self.file, self.path, self.field)	7✔
129
130	@classmethod	7✔
131	def proxy_for(cls) -> type[Interface]:	7✔
132	"""Declare this class as a proxy for the H5Interface"""
NEW 133	return H5Interface	×
134
135	def array_exists(self) -> bool:	7✔
136	"""
137	Check that there is in fact an array at :attr:`.H5Proxy.path`
138	within :attr:`.H5Proxy.file`
139	"""
140	with h5py.File(self.file, "r") as h5f:	7✔
141	obj = h5f.get(self.path)	7✔
142	return obj is not None	7✔
143
144	@classmethod	7✔
145	def from_h5array(cls, h5array: H5ArrayPath) -> "H5Proxy":	7✔
146	"""Instantiate using :class:`.H5ArrayPath`"""
147	return H5Proxy(file=h5array.file, path=h5array.path, field=h5array.field)	7✔
148
149	@property	7✔
150	def dtype(self) -> np.dtype:	7✔
151	"""
152	Get dtype of array, using :attr:`.field` if present
153	"""
154	with h5py.File(self.file, "r") as h5f:	7✔
155	obj = h5f.get(self.path)	7✔
156	if self.field is None:	7✔
157	return obj.dtype	7✔
158	else:
159	return obj.dtype[self.field]	7✔
160
161	def __array__(self) -> np.ndarray:	7✔
162	"""To a numpy array"""
163	with h5py.File(self.file, "r") as h5f:	7✔
164	obj = h5f.get(self.path)	7✔
165	return obj[:]	7✔
166
167	def __getattr__(self, item: str):	7✔
168	if item == "__name__":	7✔
169	# special case for H5Proxies that don't refer to a real file during testing
170	return "H5Proxy"	7✔
171	with h5py.File(self.file, "r") as h5f:	7✔
172	obj = h5f.get(self.path)	7✔
173	val = getattr(obj, item)	7✔
174	return val	7✔
175
176	def __getitem__(	7✔
177	self, item: int \| slice \| tuple[int \| slice, ...]
178	) -> np.ndarray \| DtypeType:
179	with h5py.File(self.file, "r") as h5f:	7✔
180	obj = h5f.get(self.path)	7✔
181	# handle compound dtypes
182	if self.field is not None:	7✔
183	# handle compound string dtype
184	if encoding := h5py.h5t.check_string_dtype(obj.dtype[self.field]):	7✔
185	if isinstance(item, tuple):	7✔
186	item = (*item, self.field)	7✔
187	else:
188	item = (item, self.field)	7✔
189
190	try:	7✔
191	# single string
192	val = obj[item].decode(encoding.encoding)	7✔
193	if self._annotation_dtype is np.datetime64:	7✔
194	return np.datetime64(val)	7✔
195	else:
196	return val	7✔
197	except AttributeError:	7✔
198	# numpy array of bytes
199	val = np.char.decode(obj[item], encoding=encoding.encoding)	7✔
200	if self._annotation_dtype is np.datetime64:	7✔
201	return val.astype(np.datetime64)	7✔
202	else:
203	return val	7✔
204	# normal compound type
205	else:
206	obj = obj.fields(self.field)	7✔
207	else:
208	if h5py.h5t.check_string_dtype(obj.dtype):	7✔
209	obj = obj.asstr()	7✔
210
211	val = obj[item]	7✔
212	if self._annotation_dtype is np.datetime64:	7✔
213	if isinstance(val, str):	7✔
214	return np.datetime64(val)	7✔
215	else:
216	return val.astype(np.datetime64)	7✔
217	else:
218	return val	7✔
219
220	def __setitem__(	7✔
221	self,
222	key: int \| slice \| tuple[int \| slice, ...],
223	value: int \| float \| datetime \| np.ndarray,
224	):
225	# TODO: Make a generalized value serdes system instead of ad-hoc type conversion
226	value = self._serialize_datetime(value)	7✔
227	with h5py.File(self.file, "r+", locking=True) as h5f:	7✔
228	obj = h5f.get(self.path)	7✔
229	if self.field is None:	7✔
230	obj[key] = value	7✔
231	else:
232	if isinstance(key, tuple):	7✔
233	key = (*key, self.field)	7✔
234	obj[key] = value	7✔
235	else:
236	obj[key, self.field] = value	7✔
237
238	def __len__(self) -> int:	7✔
239	"""self.shape[0]"""
240	return self.shape[0]	7✔
241
242	def __eq__(self, other: "H5Proxy") -> bool:	7✔
243	"""
244	Check that we are referring to the same hdf5 array
245	"""
246	if isinstance(other, H5Proxy):	7✔
247	return self._h5arraypath == other._h5arraypath	7✔
248	else:
249	raise ValueError("Can only compare equality of two H5Proxies")	7✔
250
251	def open(self, mode: str = "r") -> "h5py.Dataset":	7✔
252	"""
253	Return the opened :class:`h5py.Dataset` object
254
255	You must remember to close the associated file with :meth:`.close`
256	"""
257	if self._h5f is None:	7✔
258	self._h5f = h5py.File(self.file, mode)	7✔
259	return self._h5f.get(self.path)	7✔
260
261	def close(self) -> None:	7✔
262	"""
263	Close the :class:`h5py.File` object left open when returning the dataset with
264	:meth:`.open`
265	"""
266	if self._h5f is not None:	7✔
267	self._h5f.close()	7✔
268	self._h5f = None	7✔
269
270	def _serialize_datetime(self, v: T \| datetime) -> T \| bytes:	7✔
271	"""
272	Convert a datetime into a bytestring
273	"""
274	if self._annotation_dtype is np.datetime64:	7✔
275	if not isinstance(v, Iterable):	7✔
276	v = [v]	7✔
277	v = np.array(v).astype("S32")	7✔
278	return v	7✔
279
280
281	class H5Interface(Interface):	7✔
282	"""
283	Interface for Arrays stored as datasets within an HDF5 file.
284
285	Takes a :class:`.H5ArrayPath` specifier to select a :class:`h5py.Dataset` from a
286	:class:`h5py.File` and returns a :class:`.H5Proxy` class that acts like a
287	passthrough numpy-like interface to the dataset.
288	"""
289
290	name = "hdf5"	7✔
291	input_types = (H5ArrayPath, H5Arraylike, H5Proxy)	7✔
292	return_type = H5Proxy	7✔
293	json_model = H5JsonDict	7✔
294
295	@classmethod	7✔
296	def enabled(cls) -> bool:	7✔
297	"""Check whether h5py can be imported"""
298	return h5py is not None	7✔
299
300	@classmethod	7✔
301	def check(cls, array: H5ArrayPath \| H5Arraylike) -> bool:	7✔
302	"""
303	Check that the given array is a :class:`.H5ArrayPath` or something that
304	resembles one.
305	"""
306	if isinstance(array, (H5ArrayPath, H5Proxy)):	7✔
307	return True	7✔
308
309	if isinstance(array, dict):	7✔
310	if array.get("type", False) == cls.name:	7✔
311	return True	7✔
312	# continue checking if dict contains an hdf5 file
313	file = array.get("file", "")	7✔
314	array = (file, "")	7✔
315
316	if isinstance(array, (tuple, list)) and len(array) in (2, 3):	7✔
317	# check that the first arg is an hdf5 file
318	try:	7✔
319	file = Path(array[0])	7✔
320	except TypeError:	7✔
321	# not a path, we don't apply.
322	return False	7✔
323
324	if not file.exists():	7✔
325	return False	7✔
326
327	# hdf5 files are commonly given odd suffixes,
328	# so we just try and open it and see what happens
329	try:	7✔
330	with h5py.File(file, "r"):	7✔
331	# don't check that the array exists and raise here,
332	# this check is just for whether the validator applies or not.
333	pass	7✔
334	return True	7✔
335	except (FileNotFoundError, OSError):	7✔
336	return False	7✔
337
338	return False	7✔
339
340	def before_validation(self, array: Any) -> NDArrayType:	7✔
341	"""Create an :class:`.H5Proxy` to use throughout validation"""
342	if isinstance(array, H5ArrayPath):	7✔
343	array = H5Proxy.from_h5array(h5array=array)	7✔
344	elif isinstance(array, H5Proxy):	7✔
345	# nothing to do, already proxied
346	pass	7✔
347	elif isinstance(array, (tuple, list)) and len(array) == 2: # pragma: no cover
348	array = H5Proxy(file=array[0], path=array[1])
349	elif isinstance(array, (tuple, list)) and len(array) == 3:	7✔
350	array = H5Proxy(file=array[0], path=array[1], field=array[2])	7✔
351	else: # pragma: no cover
352	# this should never happen really since `check` confirms this before
353	# we'd reach here, but just to complete the if else...
354	raise ValueError(
355	"Need to specify a file and a path within an HDF5 file to use the HDF5 "
356	"Interface"
357	)
358	array._annotation_dtype = self.dtype	7✔
359
360	if not array.array_exists():	7✔
361	raise ValueError(	7✔
362	f"HDF5 file located at {array.file}, "
363	f"but no array found at {array.path}"
364	)
365
366	return array	7✔
367
368	def get_dtype(self, array: NDArrayType) -> DtypeType:	7✔
369	"""
370	Get the dtype from the input array
371
372	Subclasses to correctly handle
373	"""
374	if h5py.h5t.check_string_dtype(array.dtype):	7✔
375	# check for datetimes
376	try:	7✔
377	if array[0].dtype.type is np.datetime64:	7✔
378	return np.datetime64	7✔
379	else:
380	return str	7✔
381	except (AttributeError, TypeError): # pragma: no cover
382	# it's not a datetime, but it is some kind of string
383	return str
384	except (IndexError, ValueError):	7✔
385	# if the dataset is empty, we can't tell if something is a datetime
386	# or not, so we just tell the validation method what it wants to hear
387	if self.dtype in (np.datetime64, str):	7✔
388	return self.dtype	7✔
389	else:
390	return str	7✔
391	else:
392	return array.dtype	7✔
393
394	@classmethod	7✔
395	def to_json(cls, array: H5Proxy, info: SerializationInfo \| None = None) -> dict:	7✔
396	"""
397	Render HDF5 array as JSON
398
399	If ``round_trip == True``, we dump just the proxy info, a dictionary like:
400
401	* ``file``: :attr:`.H5Proxy.file`
402	* ``path``: :attr:`.H5Proxy.path`
403	* ``attrs``: Any HDF5 attributes on the dataset
404	* ``array``: The array as a list of lists
405
406	Otherwise, we dump the array as a list of lists
407	"""
408	if info.round_trip:	7✔
409	as_json = {	7✔
410	"type": cls.name,
411	}
412	as_json.update(array._h5arraypath._asdict())	7✔
413	else:
414	try:	7✔
415	dset = array.open()	7✔
416	as_json = dset[:].tolist()	7✔
417	finally:
418	array.close()	7✔
419
420	return as_json	7✔

p2p-ld / numpydantic / 27054202885

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