cpdbench.task.AlgorithmExecutionTask 

 1from collections.abc import Iterable
 2
 3from numpy import ndarray
 4
 5from cpdbench.exception.ValidationException import InputValidationException, AlgorithmValidationException
 6from cpdbench.task.Task import Task
 7
 8import inspect
 9
10from cpdbench.utils.Utils import get_name_of_function
11
12
13class AlgorithmExecutionTask(Task):
14    def __init__(self, function, counter, param_dict=None):
15        super().__init__(function, counter, param_dict)
16
17    def validate_task(self) -> None:
18        # Check number of args
19        full_arg_spec = inspect.getfullargspec(self._function)
20        if len(full_arg_spec.args) != 1:
21            # Wrong number of arguments
22            function_name = get_name_of_function(self._function)
23            raise InputValidationException(f"The number of arguments for the algorithm task '{function_name}' "
24                                           f"is {len(full_arg_spec.args)} but should be "
25                                           "1: (signal)")
26
27    def validate_input(self, data: ndarray) -> tuple[Iterable, Iterable]:
28        try:
29            alg_res_index, alg_res_scores = self._function(data)
30        except Exception as e:
31            raise AlgorithmValidationException(f"The validation of {get_name_of_function(self._function)} failed.") \
32                from e
33        else:
34            return alg_res_index, alg_res_scores
35
36    def execute(self, data: ndarray) -> tuple[Iterable, Iterable]:
37        alg_res_index, alg_res_scores = self._function(data)
38        return alg_res_index, alg_res_scores
39
40    def get_task_name(self) -> str:
41        return f"algorithm:{self._task_name}"
class AlgorithmExecutionTask(cpdbench.task.Task.Task): 
14class AlgorithmExecutionTask(Task):
15    def __init__(self, function, counter, param_dict=None):
16        super().__init__(function, counter, param_dict)
17
18    def validate_task(self) -> None:
19        # Check number of args
20        full_arg_spec = inspect.getfullargspec(self._function)
21        if len(full_arg_spec.args) != 1:
22            # Wrong number of arguments
23            function_name = get_name_of_function(self._function)
24            raise InputValidationException(f"The number of arguments for the algorithm task '{function_name}' "
25                                           f"is {len(full_arg_spec.args)} but should be "
26                                           "1: (signal)")
27
28    def validate_input(self, data: ndarray) -> tuple[Iterable, Iterable]:
29        try:
30            alg_res_index, alg_res_scores = self._function(data)
31        except Exception as e:
32            raise AlgorithmValidationException(f"The validation of {get_name_of_function(self._function)} failed.") \
33                from e
34        else:
35            return alg_res_index, alg_res_scores
36
37    def execute(self, data: ndarray) -> tuple[Iterable, Iterable]:
38        alg_res_index, alg_res_scores = self._function(data)
39        return alg_res_index, alg_res_scores
40
41    def get_task_name(self) -> str:
42        return f"algorithm:{self._task_name}"

Abstract class for a Task object which is a work package to be executed by the framework. A task has a name, can be validated, and executed, and can have some parameters.

AlgorithmExecutionTask(function, counter, param_dict=None) 
15    def __init__(self, function, counter, param_dict=None):
16        super().__init__(function, counter, param_dict)

General constructor for all task objects.

Parameters
  • function: The function handle to be executed as task content
  • counter: A number which is appended to the task name. Useful if multiple tasks with the same name exist.
  • param_dict: An optional parameter dictionary for the task
def validate_task(self) -> None: 
18    def validate_task(self) -> None:
19        # Check number of args
20        full_arg_spec = inspect.getfullargspec(self._function)
21        if len(full_arg_spec.args) != 1:
22            # Wrong number of arguments
23            function_name = get_name_of_function(self._function)
24            raise InputValidationException(f"The number of arguments for the algorithm task '{function_name}' "
25                                           f"is {len(full_arg_spec.args)} but should be "
26                                           "1: (signal)")

Validates the task statically by checking task details before running it. Throws an exception if the validation fails.

def validate_input( self, data: numpy.ndarray) -> tuple[collections.abc.Iterable, collections.abc.Iterable]: 
28    def validate_input(self, data: ndarray) -> tuple[Iterable, Iterable]:
29        try:
30            alg_res_index, alg_res_scores = self._function(data)
31        except Exception as e:
32            raise AlgorithmValidationException(f"The validation of {get_name_of_function(self._function)} failed.") \
33                from e
34        else:
35            return alg_res_index, alg_res_scores

Validates the task in combination with some input arguments. Throws an exception if the validation fails.

def execute( self, data: numpy.ndarray) -> tuple[collections.abc.Iterable, collections.abc.Iterable]: 
37    def execute(self, data: ndarray) -> tuple[Iterable, Iterable]:
38        alg_res_index, alg_res_scores = self._function(data)
39        return alg_res_index, alg_res_scores

Executes the task. Can take an arbitrary number of arguments and can produce any result.

def get_task_name(self) -> str: 
41    def get_task_name(self) -> str:
42        return f"algorithm:{self._task_name}"

Returns a descriptive name for the task.

Returns

task name as string

Inherited Members
cpdbench.task.Task.Task
get_param_dict