9766197345

Committed 02 Jul 2024 06:31PM UTC coverage: 71.367% (-0.07%) from 71.438%

Build # 9766197345

Build Type

Pull #1508

github

Committed by

web-flow

Commit Message

Merge 1ced485b4 into 51ba6c28a

Pull Request Pull Request #1508: Delayed execution in pyiron_base - for wrap_python_function() only

Run Details

126 of 177 new or added lines in 5 files covered. (71.19%)

177 existing lines in 3 files now uncovered.

7318 of 10254 relevant lines covered (71.37%)

0.71 hits per line

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

98.33

/pyiron_base/jobs/flex/pythonfunctioncontainer.py

import inspect
from typing import Tuple
import cloudpickle
import numpy as np
from pyiron_base.jobs.job.template import PythonTemplateJob
from pyiron_base.jobs.job.generic import get_executor
from pyiron_base.project.delayed import get_function_parameter_dict, get_hash


class PythonCalculateFunctionCaller:
    __slots__ = ("_function", "_executor_type", "_cores")

    def __init__(
        self,
        funct: callable = None,
        executor_type: str = None,
        cores: int = 1,
    ):
        self._function = funct
        self._executor_type = executor_type
        self._cores = cores

    def __call__(
        self,
        *args,
        **kwargs,
    ) -> Tuple[str, dict, bool]:
        """
        Generic calculate function, which writes the input files into the working_directory, executes the
        executable_script and parses the output using the output_parameter_dict.

        Args:
            Arguments of the user defined function

        Returns:
            str, dict, bool: Tuple consisting of the shell output (str), the parsed output (dict) and a boolean flag if
                             the execution raised an accepted error.
        """
        if (
            self._executor_type is not None
            and "executor" in inspect.signature(self._function).parameters.keys()
        ):
            if "executor" in kwargs.keys():
                del kwargs["executor"]
            with get_executor(
                executor_type=self._executor_type, max_workers=self._cores
            ) as exe:
                result = self._function(*args, executor=exe, **kwargs)
        else:
            result = self._function(*args, **kwargs)
        return None, {"result": result}, False


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._automatically_rename_on_save_using_input = False
        # Automatically rename job using function and input values at save time
        # This is useful for the edge case where these jobs are created from a wrapper
        # and automatically assigned a name based on the function name, but multiple
        # jobs are created from the same function (and thus distinguished only by their
        # input)

    @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

    @property
    def calculate_kwargs(self) -> dict:
        """
        Generate keyword arguments for the calculate() function.

        Example:

        >>> calculate_function = job.get_calculate_function()
        >>> shell_output, parsed_output, job_crashed = calculate_function(**job.calculate_kwargs)
        >>> job.save_output(output_dict=parsed_output, shell_output=shell_output)

        Returns:
            dict: keyword arguments for the calculate() function
        """
        return self.input.to_builtin()

    def get_calculate_function(self):
        """
        Generate calculate() function

        Example:

        >>> calculate_function = job.get_calculate_function()
        >>> shell_output, parsed_output, job_crashed = calculate_function(**job.calculate_kwargs)
        >>> job.save_output(output_dict=parsed_output, shell_output=shell_output)

        Returns:
            callable: calculate() functione
        """
        return PythonCalculateFunctionCaller(
            funct=self._function,
            executor_type=self._executor_type,
            cores=self.server.cores,
        )

    def to_dict(self):
        job_dict = super().to_dict()
        job_dict["function"] = np.void(cloudpickle.dumps(self._function))
        job_dict["_automatically_rename_on_save_using_input"] = (
            self._automatically_rename_on_save_using_input
        )
        return job_dict

    def from_dict(self, job_dict):
        super().from_dict(job_dict=job_dict)
        self._function = cloudpickle.loads(job_dict["function"])
        self._automatically_rename_on_save_using_input = bool(
            job_dict["_automatically_rename_on_save_using_input"]
        )

    def save(self):
        """
        Automatically rename job using function and input values at save time. This is useful for the edge case where
        these jobs are created from a wrapper and automatically assigned a name based on the function name, but multiple
        jobs are created from the same function (and thus distinguished only by their input).
        """
        if self._automatically_rename_on_save_using_input:
            self.job_name = self.job_name + get_hash(
                binary=cloudpickle.dumps(
                    {"fn": self._function, "kwargs": self.input.to_builtin()}
                )
            )

        if self.job_name in self.project.list_nodes():
            self.from_hdf()
            self.status.finished = True
            return  # Without saving
        super().save()

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

1	import inspect	1✔
2	from typing import Tuple	1✔
3	import cloudpickle	1✔
4	import numpy as np	1✔
5	from pyiron_base.jobs.job.template import PythonTemplateJob	1✔
6	from pyiron_base.jobs.job.generic import get_executor	1✔
7	from pyiron_base.project.delayed import get_function_parameter_dict, get_hash	1✔
8
9
10	class PythonCalculateFunctionCaller:	1✔
11	__slots__ = ("_function", "_executor_type", "_cores")	1✔
12
13	def __init__(	1✔
14	self,
15	funct: callable = None,
16	executor_type: str = None,
17	cores: int = 1,
18	):
19	self._function = funct	1✔
20	self._executor_type = executor_type	1✔
21	self._cores = cores	1✔
22
23	def __call__(	1✔
24	self,
25	*args,
26	**kwargs,
27	) -> Tuple[str, dict, bool]:
28	"""
29	Generic calculate function, which writes the input files into the working_directory, executes the
30	executable_script and parses the output using the output_parameter_dict.
31
32	Args:
33	Arguments of the user defined function
34
35	Returns:
36	str, dict, bool: Tuple consisting of the shell output (str), the parsed output (dict) and a boolean flag if
37	the execution raised an accepted error.
38	"""
39	if (	1✔
40	self._executor_type is not None
41	and "executor" in inspect.signature(self._function).parameters.keys()
42	):
43	if "executor" in kwargs.keys():	1✔
44	del kwargs["executor"]	1✔
45	with get_executor(	1✔
46	executor_type=self._executor_type, max_workers=self._cores
47	) as exe:
48	result = self._function(args, executor=exe, *kwargs)	1✔
49	else:
50	result = self._function(args, *kwargs)	1✔
51	return None, {"result": result}, False	1✔
52
53
54	class PythonFunctionContainerJob(PythonTemplateJob):	1✔
55	"""
56	The PythonFunctionContainerJob is designed to wrap any kind of python function into a pyiron job object
57
58	Example:
59
60	>>> def test_function(a, b=8):
61	>>> return a+b
62	>>>
63	>>> from pyiron_base import Project
64	>>> pr = Project("test")
65	>>> job = pr.wrap_python_function(test_function)
66	>>> job.input["a"] = 4
67	>>> job.input["b"] = 5
68	>>> job.run()
69	>>> job.output
70	>>>
71	>>> test_function_wrapped = pr.wrap_python_function(test_function)
72	>>> test_function_wrapped(4, b=6)
73	"""
74
75	def __init__(self, project, job_name):	1✔
76	super().__init__(project, job_name)	1✔
77	self._function = None	1✔
78	self._executor_type = None	1✔
79	self._automatically_rename_on_save_using_input = False	1✔
80	# Automatically rename job using function and input values at save time
81	# This is useful for the edge case where these jobs are created from a wrapper
82	# and automatically assigned a name based on the function name, but multiple
83	# jobs are created from the same function (and thus distinguished only by their
84	# input)
85
86	@property	1✔
87	def python_function(self):	1✔
UNCOV 88	return self._function	×
89
90	@python_function.setter	1✔
91	def python_function(self, funct):	1✔
92	self.input.update(get_function_parameter_dict(funct=funct))	1✔
93	self._function = funct	1✔
94
95	@property	1✔
96	def calculate_kwargs(self) -> dict:	1✔
97	"""
98	Generate keyword arguments for the calculate() function.
99
100	Example:
101
102	>>> calculate_function = job.get_calculate_function()
103	>>> shell_output, parsed_output, job_crashed = calculate_function(**job.calculate_kwargs)
104	>>> job.save_output(output_dict=parsed_output, shell_output=shell_output)
105
106	Returns:
107	dict: keyword arguments for the calculate() function
108	"""
109	return self.input.to_builtin()	1✔
110
111	def get_calculate_function(self):	1✔
112	"""
113	Generate calculate() function
114
115	Example:
116
117	>>> calculate_function = job.get_calculate_function()
118	>>> shell_output, parsed_output, job_crashed = calculate_function(**job.calculate_kwargs)
119	>>> job.save_output(output_dict=parsed_output, shell_output=shell_output)
120
121	Returns:
122	callable: calculate() functione
123	"""
124	return PythonCalculateFunctionCaller(	1✔
125	funct=self._function,
126	executor_type=self._executor_type,
127	cores=self.server.cores,
128	)
129
130	def to_dict(self):	1✔
131	job_dict = super().to_dict()	1✔
132	job_dict["function"] = np.void(cloudpickle.dumps(self._function))	1✔
133	job_dict["_automatically_rename_on_save_using_input"] = (	1✔
134	self._automatically_rename_on_save_using_input
135	)
136	return job_dict	1✔
137
138	def from_dict(self, job_dict):	1✔
139	super().from_dict(job_dict=job_dict)	1✔
140	self._function = cloudpickle.loads(job_dict["function"])	1✔
141	self._automatically_rename_on_save_using_input = bool(	1✔
142	job_dict["_automatically_rename_on_save_using_input"]
143	)
144
145	def save(self):	1✔
146	"""
147	Automatically rename job using function and input values at save time. This is useful for the edge case where
148	these jobs are created from a wrapper and automatically assigned a name based on the function name, but multiple
149	jobs are created from the same function (and thus distinguished only by their input).
150	"""
151	if self._automatically_rename_on_save_using_input:	1✔
152	self.job_name = self.job_name + get_hash(	1✔
153	binary=cloudpickle.dumps(
154	{"fn": self._function, "kwargs": self.input.to_builtin()}
155	)
156	)
157
158	if self.job_name in self.project.list_nodes():	1✔
159	self.from_hdf()	1✔
160	self.status.finished = True	1✔
161	return # Without saving	1✔
162	super().save()	1✔
163
164	def __call__(self, args, *kwargs):	1✔
165	self.input.update(	1✔
166	inspect.signature(self._function).bind(args, *kwargs).arguments
167	)
168	self.run()	1✔
169	return self.output["result"]	1✔

pyiron / pyiron_base / 9766197345

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