7972695112

Committed 20 Feb 2024 11:24AM UTC coverage: 91.829% (+0.4%) from 91.479%

Build # 7972695112

Build Type

Pull #1317

github

Committed by

web-flow

Commit Message

Merge 7b329d06f into ab51d75ef

Pull Request Pull Request #1317: Remove sql_query functionality to prepare unified db backends

Run Details

12 of 12 new or added lines in 3 files covered. (100.0%)

86 existing lines in 10 files now uncovered.

8597 of 9362 relevant lines covered (91.83%)

1.66 hits per line

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

94.74

/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

    @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))

    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"])

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

pyiron / pyiron_base / 7972695112

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