8024371140

Committed 23 Feb 2024 07:56PM UTC coverage: 91.949% (+0.03%) from 91.924%

Build # 8024371140

Build Type

Pull #1356

github

Committed by

web-flow

Commit Message

Merge d6f11b1f4 into dec8f0931

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

Run Details

13 of 15 new or added lines in 1 file covered. (86.67%)

8623 of 9378 relevant lines covered (91.95%)

1.66 hits per line

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

95.24

/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

    @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):
        hash_suffix = get_hash(
            binary=cloudpickle.dumps(
                {"fn": self._function, "kwargs": self.input.to_builtin()}
            )
        )

        job_name = self._function.__name__
        if self._mangle_name_on_save:
            job_name += hash_suffix

        self.job_name = job_name
        if 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	2✔
48
49	@property	2✔
50	def python_function(self):	2✔
51	return self._function	×
52
53	@python_function.setter	2✔
54	def python_function(self, funct):	2✔
55	self.input.update(get_function_parameter_dict(funct=funct))	2✔
56	self._function = funct	2✔
57
58	def __call__(self, args, *kwargs):	2✔
59	self.input.update(	2✔
60	inspect.signature(self._function).bind(args, *kwargs).arguments	1✔
61	)
62	self.run()	2✔
63	return self.output["result"]	2✔
64
65	def to_hdf(self, hdf=None, group_name=None):	2✔
66	super().to_hdf(hdf=hdf, group_name=group_name)	2✔
67	self.project_hdf5["function"] = np.void(cloudpickle.dumps(self._function))	2✔
68	self.project_hdf5["_mangle_name_on_save"] = self._mangle_name_on_save	2✔
69
70	def from_hdf(self, hdf=None, group_name=None):	2✔
71	super().from_hdf(hdf=hdf, group_name=group_name)	2✔
72	self._function = cloudpickle.loads(self.project_hdf5["function"])	2✔
73	self._mangle_name_on_save = bool(self.project_hdf5["_mangle_name_on_save"])	2✔
74
75	def save(self):	2✔
76	hash_suffix = get_hash(	2✔
77	binary=cloudpickle.dumps(	1✔
78	{"fn": self._function, "kwargs": self.input.to_builtin()}	1✔
79	)
80	)
81
82	job_name = self._function.__name__	2✔
83	if self._mangle_name_on_save:	2✔
84	job_name += hash_suffix	2✔
85
86	self.job_name = job_name	2✔
87	if job_name in self.project.list_nodes():	2✔
NEW 88	self.from_hdf()	×
NEW 89	self.status.finished = True	×
90	else:
91	super().save()	2✔
92
93	def run_static(self):	2✔
94	if (	1✔
95	self._executor_type is not None	1✔
96	and "executor" in inspect.signature(self._function).parameters.keys()	1✔
97	):
98	input_dict = self.input.to_builtin()	2✔
99	del input_dict["executor"]	2✔
100	output = self._function(	2✔
101	**input_dict, executor=self._get_executor(max_workers=self.server.cores)	1✔
102	)
103	else:
104	output = self._function(**self.input.to_builtin())	2✔
105	self.output.update({"result": output})	2✔
106	self.to_hdf()	2✔
107	self.status.finished = True	2✔

pyiron / pyiron_base / 8024371140

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