10230611952

Committed 03 Aug 2024 08:06PM UTC coverage: 94.363% (+0.006%) from 94.357%

Build # 10230611952

Build Type

Pull #385

github

Committed by

web-flow

Commit Message

Merge 589aa5ece into 4255d1c26

Pull Request Pull Request #385: Add docstrings and typehints with Github copilot

Run Details

42 of 43 new or added lines in 11 files covered. (97.67%)

4 existing lines in 2 files now uncovered.

904 of 958 relevant lines covered (94.36%)

0.94 hits per line

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

89.66

/executorlib/cache/hdf.py

from typing import Tuple

import cloudpickle
import h5io
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():
                h5io.write_hdf5(
                    fname=fname,
                    data=np.void(cloudpickle.dumps(data_value)),
                    overwrite="update",
                    title=group_dict[data_key],
                )


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(
                h5io.read_hdf5(fname=hdf, title="function", slash="ignore")
            )
        else:
            raise TypeError("Function not found in HDF5 file.")
        if "input_args" in hdf:
            data_dict["args"] = cloudpickle.loads(
                h5io.read_hdf5(fname=hdf, title="input_args", slash="ignore")
            )
        else:
            data_dict["args"] = ()
        if "input_kwargs" in hdf:
            data_dict["kwargs"] = cloudpickle.loads(
                h5io.read_hdf5(fname=hdf, title="input_kwargs", slash="ignore")
            )
        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(
                h5io.read_hdf5(fname=hdf, title="output", slash="ignore")
            )
        else:
            return False, None

1	from typing import Tuple	1✔
2
3	import cloudpickle	1✔
4	import h5io	1✔
5	import h5py	1✔
6	import numpy as np	1✔
7
8
9	def dump(file_name: str, data_dict: dict) -> None:	1✔
10	"""
11	Dump data dictionary into HDF5 file
12
13	Args:
14	file_name (str): file name of the HDF5 file as absolute path
15	data_dict (dict): dictionary containing the python function to be executed {"fn": ..., "args": (), "kwargs": {}}
16	"""
17	group_dict = {	1✔
18	"fn": "function",
19	"args": "input_args",
20	"kwargs": "input_kwargs",
21	"output": "output",
22	}
23	with h5py.File(file_name, "a") as fname:	1✔
24	for data_key, data_value in data_dict.items():	1✔
25	if data_key in group_dict.keys():	1✔
26	h5io.write_hdf5(	1✔
27	fname=fname,
28	data=np.void(cloudpickle.dumps(data_value)),
29	overwrite="update",
30	title=group_dict[data_key],
31	)
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(	1✔
48	h5io.read_hdf5(fname=hdf, title="function", slash="ignore")
49	)
50	else:
NEW 51	raise TypeError("Function not found in HDF5 file.")	×
52	if "input_args" in hdf:	1✔
53	data_dict["args"] = cloudpickle.loads(	1✔
54	h5io.read_hdf5(fname=hdf, title="input_args", slash="ignore")
55	)
56	else:
57	data_dict["args"] = ()	×
58	if "input_kwargs" in hdf:	1✔
59	data_dict["kwargs"] = cloudpickle.loads(	1✔
60	h5io.read_hdf5(fname=hdf, title="input_kwargs", slash="ignore")
61	)
62	else:
63	data_dict["kwargs"] = {}	1✔
64	return data_dict	1✔
65
66
67	def get_output(file_name: str) -> Tuple[bool, object]:	1✔
68	"""
69	Check if output is available in the HDF5 file
70
71	Args:
72	file_name (str): file name of the HDF5 file as absolute path
73
74	Returns:
75	Tuple[bool, object]: boolean flag indicating if output is available and the output object itself
76	"""
77	with h5py.File(file_name, "r") as hdf:	1✔
78	if "output" in hdf:	1✔
79	return True, cloudpickle.loads(	1✔
80	h5io.read_hdf5(fname=hdf, title="output", slash="ignore")
81	)
82	else:
83	return False, None	×

pyiron / executorlib / 10230611952

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