11766741741

Committed 10 Nov 2024 04:26PM UTC coverage: 94.813% (-0.3%) from 95.089%

Build # 11766741741

Build Type

Pull #491

github

Committed by

web-flow

Commit Message

Merge e34b5de8a into 78062fd9b

Pull Request Pull Request #491: Empty cache

Run Details

9 of 12 new or added lines in 2 files covered. (75.0%)

914 of 964 relevant lines covered (94.81%)

0.95 hits per line

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

93.55

/executorlib/standalone/hdf.py

from typing import Tuple

import cloudpickle
import h5py
import numpy as np


def dump(file_name: str, data_dict: dict) -> None:
    """
    Dump data dictionary into HDF5 file

    Args:
        file_name (str): file name of the HDF5 file as absolute path
        data_dict (dict): dictionary containing the python function to be executed {"fn": ..., "args": (), "kwargs": {}}
    """
    group_dict = {
        "fn": "function",
        "args": "input_args",
        "kwargs": "input_kwargs",
        "output": "output",
    }
    with h5py.File(file_name, "a") as fname:
        for data_key, data_value in data_dict.items():
            if data_key in group_dict.keys():
                try:
                    fname.create_dataset(
                        name="/" + group_dict[data_key],
                        data=np.void(cloudpickle.dumps(data_value)),
                    )
                except ValueError:
                    pass


def load(file_name: str) -> dict:
    """
    Load data dictionary from HDF5 file

    Args:
        file_name (str): file name of the HDF5 file as absolute path

    Returns:
        dict: dictionary containing the python function to be executed {"fn": ..., "args": (), "kwargs": {}}
    """
    with h5py.File(file_name, "r") as hdf:
        data_dict = {}
        if "function" in hdf:
            data_dict["fn"] = cloudpickle.loads(np.void(hdf["/function"]))
        else:
            raise TypeError("Function not found in HDF5 file.")
        if "input_args" in hdf:
            data_dict["args"] = cloudpickle.loads(np.void(hdf["/input_args"]))
        else:
            data_dict["args"] = ()
        if "input_kwargs" in hdf:
            data_dict["kwargs"] = cloudpickle.loads(np.void(hdf["/input_kwargs"]))
        else:
            data_dict["kwargs"] = {}
        return data_dict


def get_output(file_name: str) -> Tuple[bool, object]:
    """
    Check if output is available in the HDF5 file

    Args:
        file_name (str): file name of the HDF5 file as absolute path

    Returns:
        Tuple[bool, object]: boolean flag indicating if output is available and the output object itself
    """
    with h5py.File(file_name, "r") as hdf:
        if "output" in hdf:
            return True, cloudpickle.loads(np.void(hdf["/output"]))
        else:
            return False, None

1	from typing import Tuple	1✔
2
3	import cloudpickle	1✔
4	import h5py	1✔
5	import numpy as np	1✔
6
7
8	def dump(file_name: str, data_dict: dict) -> None:	1✔
9	"""
10	Dump data dictionary into HDF5 file
11
12	Args:
13	file_name (str): file name of the HDF5 file as absolute path
14	data_dict (dict): dictionary containing the python function to be executed {"fn": ..., "args": (), "kwargs": {}}
15	"""
16	group_dict = {	1✔
17	"fn": "function",
18	"args": "input_args",
19	"kwargs": "input_kwargs",
20	"output": "output",
21	}
22	with h5py.File(file_name, "a") as fname:	1✔
23	for data_key, data_value in data_dict.items():	1✔
24	if data_key in group_dict.keys():	1✔
25	try:	1✔
26	fname.create_dataset(	1✔
27	name="/" + group_dict[data_key],
28	data=np.void(cloudpickle.dumps(data_value)),
29	)
30	except ValueError:	1✔
31	pass	1✔
32
33
34	def load(file_name: str) -> dict:	1✔
35	"""
36	Load data dictionary from HDF5 file
37
38	Args:
39	file_name (str): file name of the HDF5 file as absolute path
40
41	Returns:
42	dict: dictionary containing the python function to be executed {"fn": ..., "args": (), "kwargs": {}}
43	"""
44	with h5py.File(file_name, "r") as hdf:	1✔
45	data_dict = {}	1✔
46	if "function" in hdf:	1✔
47	data_dict["fn"] = cloudpickle.loads(np.void(hdf["/function"]))	1✔
48	else:
49	raise TypeError("Function not found in HDF5 file.")	×
50	if "input_args" in hdf:	1✔
51	data_dict["args"] = cloudpickle.loads(np.void(hdf["/input_args"]))	1✔
52	else:
53	data_dict["args"] = ()	×
54	if "input_kwargs" in hdf:	1✔
55	data_dict["kwargs"] = cloudpickle.loads(np.void(hdf["/input_kwargs"]))	1✔
56	else:
57	data_dict["kwargs"] = {}	1✔
58	return data_dict	1✔
59
60
61	def get_output(file_name: str) -> Tuple[bool, object]:	1✔
62	"""
63	Check if output is available in the HDF5 file
64
65	Args:
66	file_name (str): file name of the HDF5 file as absolute path
67
68	Returns:
69	Tuple[bool, object]: boolean flag indicating if output is available and the output object itself
70	"""
71	with h5py.File(file_name, "r") as hdf:	1✔
72	if "output" in hdf:	1✔
73	return True, cloudpickle.loads(np.void(hdf["/output"]))	1✔
74	else:
75	return False, None	1✔

pyiron / executorlib / 11766741741

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