8024459492

Committed 23 Feb 2024 08:04PM UTC coverage: 91.948% (+0.02%) from 91.924%

Build # 8024459492

Build Type

Pull #1356

github

Committed by

web-flow

Commit Message

Merge fc2572d23 into dec8f0931

Pull Request Pull Request #1356: Introduce a flag for _not_ mangling the name of wrapped function jobs

Run Details

11 of 13 new or added lines in 1 file covered. (84.62%)

1 existing line in 1 file now uncovered.

8621 of 9376 relevant lines covered (91.95%)

1.66 hits per line

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

95.08

/pyiron_base/jobs/flex/pythonfunctioncontainer.py

import inspect
import hashlib
import re
import cloudpickle
import numpy as np
from pyiron_base.jobs.job.template import PythonTemplateJob


def get_function_parameter_dict(funct):
    return {
        k: None if v.default == inspect._empty else v.default
        for k, v in inspect.signature(funct).parameters.items()
    }


def get_hash(binary):
    # Remove specification of jupyter kernel from hash to be deterministic
    binary_no_ipykernel = re.sub(b"(?<=/ipykernel_)(.*)(?=/)", b"", binary)
    return str(hashlib.md5(binary_no_ipykernel).hexdigest())


class PythonFunctionContainerJob(PythonTemplateJob):
    """
    The PythonFunctionContainerJob is designed to wrap any kind of python function into a pyiron job object

    Example:

    >>> def test_function(a, b=8):
    >>>     return a+b
    >>>
    >>> from pyiron_base import Project
    >>> pr = Project("test")
    >>> job = pr.wrap_python_function(test_function)
    >>> job.input["a"] = 4
    >>> job.input["b"] = 5
    >>> job.run()
    >>> job.output
    >>>
    >>> test_function_wrapped = pr.wrap_python_function(test_function)
    >>> test_function_wrapped(4, b=6)
    """

    def __init__(self, project, job_name):
        super().__init__(project, job_name)
        self._function = None
        self._executor_type = None
        self._mangle_name_on_save = True  # Automatically rename job using function and
        # input values

    @property
    def python_function(self):
        return self._function

    @python_function.setter
    def python_function(self, funct):
        self.input.update(get_function_parameter_dict(funct=funct))
        self._function = funct

    def __call__(self, *args, **kwargs):
        self.input.update(
            inspect.signature(self._function).bind(*args, **kwargs).arguments
        )
        self.run()
        return self.output["result"]

    def to_hdf(self, hdf=None, group_name=None):
        super().to_hdf(hdf=hdf, group_name=group_name)
        self.project_hdf5["function"] = np.void(cloudpickle.dumps(self._function))
        self.project_hdf5["_mangle_name_on_save"] = self._mangle_name_on_save

    def from_hdf(self, hdf=None, group_name=None):
        super().from_hdf(hdf=hdf, group_name=group_name)
        self._function = cloudpickle.loads(self.project_hdf5["function"])
        self._mangle_name_on_save = bool(self.project_hdf5["_mangle_name_on_save"])

    def save(self):
        if self._mangle_name_on_save:
            job_name = self._function.__name__ + get_hash(
                binary=cloudpickle.dumps(
                    {"fn": self._function, "kwargs": self.input.to_builtin()}
                )
            )

            self.job_name = job_name

        if self.job_name in self.project.list_nodes():
            self.from_hdf()
            self.status.finished = True
        else:
            super().save()

    def run_static(self):
        if (
            self._executor_type is not None
            and "executor" in inspect.signature(self._function).parameters.keys()
        ):
            input_dict = self.input.to_builtin()
            del input_dict["executor"]
            output = self._function(
                **input_dict, executor=self._get_executor(max_workers=self.server.cores)
            )
        else:
            output = self._function(**self.input.to_builtin())
        self.output.update({"result": output})
        self.to_hdf()
        self.status.finished = True

1	import inspect	2✔
2	import hashlib	2✔
3	import re	2✔
4	import cloudpickle	2✔
5	import numpy as np	2✔
6	from pyiron_base.jobs.job.template import PythonTemplateJob	2✔
7
8
9	def get_function_parameter_dict(funct):	2✔
10	return {	2✔
11	k: None if v.default == inspect._empty else v.default	1✔
12	for k, v in inspect.signature(funct).parameters.items()	1✔
13	}
14
15
16	def get_hash(binary):	2✔
17	# Remove specification of jupyter kernel from hash to be deterministic
18	binary_no_ipykernel = re.sub(b"(?<=/ipykernel_)(.*)(?=/)", b"", binary)	2✔
19	return str(hashlib.md5(binary_no_ipykernel).hexdigest())	2✔
20
21
22	class PythonFunctionContainerJob(PythonTemplateJob):	2✔
23	"""
24	The PythonFunctionContainerJob is designed to wrap any kind of python function into a pyiron job object
25
26	Example:
27
28	>>> def test_function(a, b=8):
29	>>> return a+b
30	>>>
31	>>> from pyiron_base import Project
32	>>> pr = Project("test")
33	>>> job = pr.wrap_python_function(test_function)
34	>>> job.input["a"] = 4
35	>>> job.input["b"] = 5
36	>>> job.run()
37	>>> job.output
38	>>>
39	>>> test_function_wrapped = pr.wrap_python_function(test_function)
40	>>> test_function_wrapped(4, b=6)
41	"""
42
43	def __init__(self, project, job_name):	2✔
44	super().__init__(project, job_name)	2✔
45	self._function = None	2✔
46	self._executor_type = None	2✔
47	self._mangle_name_on_save = True # Automatically rename job using function and	2✔
48	# input values
49
50	@property	2✔
51	def python_function(self):	2✔
UNCOV 52	return self._function	×
53
54	@python_function.setter	2✔
55	def python_function(self, funct):	2✔
56	self.input.update(get_function_parameter_dict(funct=funct))	2✔
57	self._function = funct	2✔
58
59	def __call__(self, args, *kwargs):	2✔
60	self.input.update(	2✔
61	inspect.signature(self._function).bind(args, *kwargs).arguments	1✔
62	)
63	self.run()	2✔
64	return self.output["result"]	2✔
65
66	def to_hdf(self, hdf=None, group_name=None):	2✔
67	super().to_hdf(hdf=hdf, group_name=group_name)	2✔
68	self.project_hdf5["function"] = np.void(cloudpickle.dumps(self._function))	2✔
69	self.project_hdf5["_mangle_name_on_save"] = self._mangle_name_on_save	2✔
70
71	def from_hdf(self, hdf=None, group_name=None):	2✔
72	super().from_hdf(hdf=hdf, group_name=group_name)	2✔
73	self._function = cloudpickle.loads(self.project_hdf5["function"])	2✔
74	self._mangle_name_on_save = bool(self.project_hdf5["_mangle_name_on_save"])	2✔
75
76	def save(self):	2✔
77	if self._mangle_name_on_save:	2✔
78	job_name = self._function.__name__ + get_hash(	2✔
79	binary=cloudpickle.dumps(	1✔
80	{"fn": self._function, "kwargs": self.input.to_builtin()}	1✔
81	)
82	)
83
84	self.job_name = job_name	2✔
85
86	if self.job_name in self.project.list_nodes():	2✔
NEW 87	self.from_hdf()	×
NEW 88	self.status.finished = True	×
89	else:
90	super().save()	2✔
91
92	def run_static(self):	2✔
93	if (	1✔
94	self._executor_type is not None	1✔
95	and "executor" in inspect.signature(self._function).parameters.keys()	1✔
96	):
97	input_dict = self.input.to_builtin()	2✔
98	del input_dict["executor"]	2✔
99	output = self._function(	2✔
100	**input_dict, executor=self._get_executor(max_workers=self.server.cores)	1✔
101	)
102	else:
103	output = self._function(**self.input.to_builtin())	2✔
104	self.output.update({"result": output})	2✔
105	self.to_hdf()	2✔
106	self.status.finished = True	2✔

pyiron / pyiron_base / 8024459492

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