cpdbench.task.DatasetFetchTask 

 1import inspect
 2
 3from cpdbench.exception.ValidationException import DatasetValidationException, InputValidationException
 4from cpdbench.dataset import CPDDataset
 5from cpdbench.task.Task import Task
 6
 7from cpdbench.utils.Utils import get_name_of_function
 8
 9
10class DatasetFetchTask(Task):
11    def __init__(self, function, counter, param_dict=None):
12        super().__init__(function, counter, param_dict)
13
14    def validate_task(self) -> None:
15        # Check number of args
16        full_arg_spec = inspect.getfullargspec(self._function)
17        if len(full_arg_spec.args) > 0:
18            # Wrong number of arguments
19            function_name = get_name_of_function(self._function)
20            raise InputValidationException("The number of arguments for the dataset task '{0}' is {1} but should be 0."
21                                           .format(function_name, len(full_arg_spec.args)))
22
23    def validate_input(self, *args) -> CPDDataset:
24        try:
25            dataset: CPDDataset = self._function()
26            dataset.init()
27        except Exception as e:
28            raise DatasetValidationException(f"The validation of {get_name_of_function(self._function)} failed.") \
29                from e
30        else:
31            return dataset
32
33    def execute(self) -> CPDDataset:
34        dataset: CPDDataset = self._function()
35        dataset.init()
36        return dataset
37
38    def get_task_name(self) -> str:
39        return f"dataset:{self._task_name}"
class DatasetFetchTask(cpdbench.task.Task.Task): 
11class DatasetFetchTask(Task):
12    def __init__(self, function, counter, param_dict=None):
13        super().__init__(function, counter, param_dict)
14
15    def validate_task(self) -> None:
16        # Check number of args
17        full_arg_spec = inspect.getfullargspec(self._function)
18        if len(full_arg_spec.args) > 0:
19            # Wrong number of arguments
20            function_name = get_name_of_function(self._function)
21            raise InputValidationException("The number of arguments for the dataset task '{0}' is {1} but should be 0."
22                                           .format(function_name, len(full_arg_spec.args)))
23
24    def validate_input(self, *args) -> CPDDataset:
25        try:
26            dataset: CPDDataset = self._function()
27            dataset.init()
28        except Exception as e:
29            raise DatasetValidationException(f"The validation of {get_name_of_function(self._function)} failed.") \
30                from e
31        else:
32            return dataset
33
34    def execute(self) -> CPDDataset:
35        dataset: CPDDataset = self._function()
36        dataset.init()
37        return dataset
38
39    def get_task_name(self) -> str:
40        return f"dataset:{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.

DatasetFetchTask(function, counter, param_dict=None) 
12    def __init__(self, function, counter, param_dict=None):
13        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: 
15    def validate_task(self) -> None:
16        # Check number of args
17        full_arg_spec = inspect.getfullargspec(self._function)
18        if len(full_arg_spec.args) > 0:
19            # Wrong number of arguments
20            function_name = get_name_of_function(self._function)
21            raise InputValidationException("The number of arguments for the dataset task '{0}' is {1} but should be 0."
22                                           .format(function_name, len(full_arg_spec.args)))

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

def validate_input( self, *args) -> <module 'cpdbench.dataset.CPDDataset' from '/Users/dominik/Documents/Projects/CPD-Bench/src/cpdbench/dataset/CPDDataset.py'>: 
24    def validate_input(self, *args) -> CPDDataset:
25        try:
26            dataset: CPDDataset = self._function()
27            dataset.init()
28        except Exception as e:
29            raise DatasetValidationException(f"The validation of {get_name_of_function(self._function)} failed.") \
30                from e
31        else:
32            return dataset

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

def execute( self) -> <module 'cpdbench.dataset.CPDDataset' from '/Users/dominik/Documents/Projects/CPD-Bench/src/cpdbench/dataset/CPDDataset.py'>: 
34    def execute(self) -> CPDDataset:
35        dataset: CPDDataset = self._function()
36        dataset.init()
37        return dataset

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

def get_task_name(self) -> str: 
39    def get_task_name(self) -> str:
40        return f"dataset:{self._task_name}"

Returns a descriptive name for the task.

Returns

task name as string

Inherited Members
cpdbench.task.Task.Task
get_param_dict