cpdbench.task.TaskFactory

 1from typing import Callable
 2import inspect
 3
 4from cpdbench.exception.UserParameterDoesNotExistException import UserParameterDoesNotExistException
 5from cpdbench.task.AlgorithmExecutionTask import AlgorithmExecutionTask
 6from cpdbench.task.DatasetFetchTask import DatasetFetchTask
 7from cpdbench.task.MetricExecutionTask import MetricExecutionTask
 8from cpdbench.task.Task import TaskType, Task
 9import cpdbench.utils.BenchConfig as BenchConfig
10from cpdbench.utils import Logger
11from cpdbench.utils.Utils import get_name_of_function
12from functools import partial
13
14
15class TaskFactory:
16    """Abstract factory for creating task objects"""
17
18    def __init__(self):
19        self._user_config = BenchConfig.get_user_config()
20        self._logger = Logger.get_application_logger()
21        self._task_counter = 0
22
23    def create_task_from_function(self, function: Callable, task_type: TaskType) -> Task:
24        """Creates one task for an unparametrized function.
25        :param function: the function to be executed as task
26        :param task_type: the type of the task to be created
27        :return: the constructed task object
28        """
29        return self._generate_task_object(function, {}, task_type)
30
31    def create_tasks_with_parameters(self, function: Callable, task_type: TaskType) -> list[Task]:
32        """Creates correct task objects based on the given task type and the needed parameters defined in the user
33        config. Because of this, this method can potentially output multiple tasks with the same function
34        but different parameters.
35        :param function: the function to be executed as task
36        :param task_type: the type of the task to be created
37        :return: the constructed task objects
38        """
39        all_params = [param.name for param in inspect.signature(function).parameters.values() if param.kind ==
40                      param.KEYWORD_ONLY]
41        try:
42            global_params = [param for param in all_params if self._user_config.check_if_global_param(param)]
43        except Exception as e:
44            if "Parameter not found" in str(e):
45                raise UserParameterDoesNotExistException(str(e).split()[-1], get_name_of_function(function))
46
47        if all_params is None or len(all_params) == 0:
48            # Easy case: no parameter
49            task = self._generate_task_object(function,{}, task_type)
50            self._logger.info(f"Created task {task.get_task_name()}")
51            self._task_counter += 1
52            return [task]
53        global_params.sort()
54        all_params.sort()
55        if global_params == all_params:
56            # Easy case: only global params
57            param_values = [{}]
58        else:
59            param_values = [{} for _ in range(self._user_config.get_number_of_executions(task_type))]
60        if len(param_values) == 0:
61            param_values = [{}]
62
63        for param in all_params:
64            try:
65                vals = self._user_config.get_user_param(param, task_type)
66            except Exception as e:
67                if str(e) == "Parameter not found":
68                    raise UserParameterDoesNotExistException(param, get_name_of_function(function))
69                else:
70                    raise e
71            else:
72                for i in range(len(param_values)):
73                    if param in global_params:
74                        param_values[i].update({param: vals[0]})  # global param
75                    else:
76                        param_values[i].update({param: vals[i]})  # execution param
77
78        tasks = []
79        for param_dict in param_values:
80            function_with_params = partial(function, **param_dict)
81            task = self._generate_task_object(function_with_params, param_dict, task_type)
82            self._task_counter += 1
83            self._logger.info(f"Created task {task.get_task_name()} with following parameters: {str(param_dict)}")
84            tasks.append(task)
85        return tasks
86
87    def _generate_task_object(self, function: Callable, param_dict: dict, task_type: TaskType):
88        if task_type == TaskType.DATASET_FETCH:
89            return DatasetFetchTask(function, self._task_counter, param_dict)
90        elif task_type == TaskType.ALGORITHM_EXECUTION:
91            return AlgorithmExecutionTask(function, self._task_counter, param_dict)
92        elif task_type == TaskType.METRIC_EXECUTION:
93            return MetricExecutionTask(function, self._task_counter, param_dict)