cpdbench.control.TestbenchController 

 1import json
 2import logging
 3from enum import Enum
 4
 5import numpy as np
 6
 7from cpdbench.control.TestrunController import TestrunController
 8from cpdbench.control.ValidationRunController import ValidationRunController
 9from cpdbench.utils import BenchConfig, Logger
10
11
12class TestrunType(Enum):
13    """Types of predefined run configurations"""
14    NORMAL_RUN = 1,
15    VALIDATION_RUN = 2
16
17
18class ExtendedEncoder(json.JSONEncoder):
19    def default(self, obj):
20        if isinstance(obj, np.integer):
21            return int(obj)
22        if isinstance(obj, np.floating):
23            return float(obj)
24        if isinstance(obj, np.ndarray):
25            return obj.tolist()
26        return super(ExtendedEncoder, self).default(obj)
27
28
29class TestbenchController:
30    """Main controller for starting different types of test runs.
31    This is the basic class which is executed after initialization of the framework.
32    This controller then calls an ExecutionController implementation, which contains
33    the actual logic of a run.
34    """
35
36    def execute_testrun(self, runtype: TestrunType, datasets: list, algorithms: list, metrics: list) -> None:
37        """Prepares and runs the required testrun
38        :param runtype: Type of testrun to run
39        :param datasets: list of dataset functions
40        :param algorithms: list of algorithm functions
41        :param metrics: list of metric functions
42        """
43        if runtype == TestrunType.NORMAL_RUN:
44            controller = TestrunController()
45        else:
46            controller = ValidationRunController()
47
48        function_map = {
49            "datasets": datasets,
50            "algorithms": algorithms,
51            "metrics": metrics
52        }
53        try:
54            result = controller.execute_run(function_map)
55        except Exception as e:
56            Logger.get_application_logger().critical("Unexpected error occurred during bench run. CPDBench was aborted.")
57            Logger.get_application_logger().critical(e)
58            logging.shutdown()
59        else:
60            self._output_result(result.get_result_as_dict())
61            Logger.get_application_logger().info("CDPBench has finished. The following config has been used:")
62            Logger.get_application_logger().info(BenchConfig.get_complete_config())
63            logging.shutdown()
64
65    def _output_result(self, result_dict: dict) -> None:
66        """Outputs a result dict correctly on console and in a file
67        :param result_dict: the to be printed dict
68        """
69        json_string = json.dumps(result_dict, indent=4, cls=ExtendedEncoder)
70
71        # file output
72        with open(BenchConfig.result_file_name, 'w') as file:
73            file.write(json_string)
74
75        # console output
76        print(json_string)
class TestrunType(enum.Enum): 
13class TestrunType(Enum):
14    """Types of predefined run configurations"""
15    NORMAL_RUN = 1,
16    VALIDATION_RUN = 2

Types of predefined run configurations

NORMAL_RUN = <TestrunType.NORMAL_RUN: (1,)>
VALIDATION_RUN = <TestrunType.VALIDATION_RUN: 2>
Inherited Members
enum.Enum
name
value
class ExtendedEncoder(json.encoder.JSONEncoder): 
19class ExtendedEncoder(json.JSONEncoder):
20    def default(self, obj):
21        if isinstance(obj, np.integer):
22            return int(obj)
23        if isinstance(obj, np.floating):
24            return float(obj)
25        if isinstance(obj, np.ndarray):
26            return obj.tolist()
27        return super(ExtendedEncoder, self).default(obj)

Extensible JSON http://json.org encoder for Python data structures.

Supports the following objects and types by default:

+-------------------+---------------+ | Python | JSON | +===================+===============+ | dict | object | +-------------------+---------------+ | list, tuple | array | +-------------------+---------------+ | str | string | +-------------------+---------------+ | int, float | number | +-------------------+---------------+ | True | true | +-------------------+---------------+ | False | false | +-------------------+---------------+ | None | null | +-------------------+---------------+

To extend this to recognize other objects, subclass and implement a .default() method with another method that returns a serializable object for o if possible, otherwise it should call the superclass implementation (to raise TypeError).

def default(self, obj): 
20    def default(self, obj):
21        if isinstance(obj, np.integer):
22            return int(obj)
23        if isinstance(obj, np.floating):
24            return float(obj)
25        if isinstance(obj, np.ndarray):
26            return obj.tolist()
27        return super(ExtendedEncoder, self).default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this::

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
Inherited Members
json.encoder.JSONEncoder
JSONEncoder
item_separator
key_separator
skipkeys
ensure_ascii
check_circular
allow_nan
sort_keys
indent
encode
iterencode
class TestbenchController: 
30class TestbenchController:
31    """Main controller for starting different types of test runs.
32    This is the basic class which is executed after initialization of the framework.
33    This controller then calls an ExecutionController implementation, which contains
34    the actual logic of a run.
35    """
36
37    def execute_testrun(self, runtype: TestrunType, datasets: list, algorithms: list, metrics: list) -> None:
38        """Prepares and runs the required testrun
39        :param runtype: Type of testrun to run
40        :param datasets: list of dataset functions
41        :param algorithms: list of algorithm functions
42        :param metrics: list of metric functions
43        """
44        if runtype == TestrunType.NORMAL_RUN:
45            controller = TestrunController()
46        else:
47            controller = ValidationRunController()
48
49        function_map = {
50            "datasets": datasets,
51            "algorithms": algorithms,
52            "metrics": metrics
53        }
54        try:
55            result = controller.execute_run(function_map)
56        except Exception as e:
57            Logger.get_application_logger().critical("Unexpected error occurred during bench run. CPDBench was aborted.")
58            Logger.get_application_logger().critical(e)
59            logging.shutdown()
60        else:
61            self._output_result(result.get_result_as_dict())
62            Logger.get_application_logger().info("CDPBench has finished. The following config has been used:")
63            Logger.get_application_logger().info(BenchConfig.get_complete_config())
64            logging.shutdown()
65
66    def _output_result(self, result_dict: dict) -> None:
67        """Outputs a result dict correctly on console and in a file
68        :param result_dict: the to be printed dict
69        """
70        json_string = json.dumps(result_dict, indent=4, cls=ExtendedEncoder)
71
72        # file output
73        with open(BenchConfig.result_file_name, 'w') as file:
74            file.write(json_string)
75
76        # console output
77        print(json_string)

Main controller for starting different types of test runs. This is the basic class which is executed after initialization of the framework. This controller then calls an ExecutionController implementation, which contains the actual logic of a run.

def execute_testrun( self, runtype: TestrunType, datasets: list, algorithms: list, metrics: list) -> None: 
37    def execute_testrun(self, runtype: TestrunType, datasets: list, algorithms: list, metrics: list) -> None:
38        """Prepares and runs the required testrun
39        :param runtype: Type of testrun to run
40        :param datasets: list of dataset functions
41        :param algorithms: list of algorithm functions
42        :param metrics: list of metric functions
43        """
44        if runtype == TestrunType.NORMAL_RUN:
45            controller = TestrunController()
46        else:
47            controller = ValidationRunController()
48
49        function_map = {
50            "datasets": datasets,
51            "algorithms": algorithms,
52            "metrics": metrics
53        }
54        try:
55            result = controller.execute_run(function_map)
56        except Exception as e:
57            Logger.get_application_logger().critical("Unexpected error occurred during bench run. CPDBench was aborted.")
58            Logger.get_application_logger().critical(e)
59            logging.shutdown()
60        else:
61            self._output_result(result.get_result_as_dict())
62            Logger.get_application_logger().info("CDPBench has finished. The following config has been used:")
63            Logger.get_application_logger().info(BenchConfig.get_complete_config())
64            logging.shutdown()

Prepares and runs the required testrun

Parameters
  • runtype: Type of testrun to run
  • datasets: list of dataset functions
  • algorithms: list of algorithm functions
  • metrics: list of metric functions