bench_generator.generator

This module holds the Generator class which is responsible for generating a benchmark testbed. All features of this tool can be accessed through the Generator class, other classes should not be used directly.

  1#!/usr/bin/env python3
  2
  3"""
  4This module holds the Generator class which is responsible for generating a
  5benchmark testbed. All features of this tool can be accessed through the
  6Generator class, other classes should not be used directly.
  7"""
  8
  9import os
 10import sys
 11import json
 12import importlib
 13import inspect
 14import jsonschema
 15from typing import List, Dict, Any
 16from bench_generator.logger import Logger
 17
 18SCHEMA_FILE = 'metadata.schema'
 19DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
 20
 21
 22class Generator:
 23    """
 24    Generator class generates a benchmark testbed.
 25    """
 26
 27    def __init__(self, main_directory: str, verbose: bool = False):
 28        """Create an instance of the Benchmark class.
 29
 30        Parameters
 31        ----------
 32        main_directory : str
 33            The root directory of all the cases to execute.
 34        verbose : bool
 35            Enables verbose logs.
 36        """
 37        self._main_directory = os.path.abspath(main_directory)
 38        self._resources: List[Dict[str, Any]] = []
 39        self._class_module_mapping: Dict[str, Any] = {}
 40        self._verbose = verbose
 41        self._logger = Logger(__name__, self._main_directory, self._verbose)
 42        self._schema = {}
 43
 44        with open(os.path.join(DATA_DIR, SCHEMA_FILE)) as f:
 45            self._schema = json.load(f)
 46
 47        self._init_resources()
 48
 49    @property
 50    def main_directory(self) -> str:
 51        """The main directory of all the cases.
 52
 53        Returns
 54        -------
 55        main_directory : str
 56            The path to the main directory of the cases.
 57        """
 58        return self._main_directory
 59
 60    def _init_resources(self) -> None:
 61        """Initialize resources of a case
 62
 63        Resources are discovered automatically by analyzing Python modules.
 64        """
 65
 66        # Discover all modules to import
 67        sys.path.append(os.path.dirname(__file__))
 68        self._modules = list(filter(lambda x: x.endswith('.py')
 69                                    and '__init__' not in x
 70                                    and '__pycache__' not in x,
 71                                    os.listdir(os.path.dirname(__file__))))
 72
 73        # Discover all classes in each module
 74        for m in self._modules:
 75            module_name = os.path.splitext(m)[0]
 76            parent_module = os.path.split(os.path.dirname(__file__))[-1]
 77            import_name = '.'.join([parent_module, module_name])
 78            imported_module = importlib.import_module(import_name)
 79            for name, cls in inspect.getmembers(imported_module,
 80                                                inspect.isclass):
 81                parent_cls = cls.__base__
 82
 83                # Skip hidden classes
 84                if name.startswith('_') or name[0].islower():
 85                    continue
 86
 87                # Skip classes which do not inherit from Scenario
 88                if parent_cls is None or parent_cls.__name__ != 'Scenario':
 89                    continue
 90
 91                # Store class-module mapping for reverse look-up
 92                self._class_module_mapping[name] = imported_module
 93
 94                # Discover all methods and their parameters in each class
 95                methods: Dict[str, List[Dict[str, str]]] = {}
 96                filt = inspect.getmembers(cls, inspect.isfunction)
 97                for method_name, method in filt:
 98                    parameters = inspect.signature(method).parameters
 99                    methods[method_name] = []
100                    for key in parameters.keys():
101                        if key == 'self':
102                            continue
103                        p = parameters[key]
104                        required = (p.default == inspect.Parameter.empty)
105                        methods[method_name].append({'name': p.name,
106                                                     'required': required})
107
108                if name not in list(filter(lambda x: x['name'],
109                                           self._resources)):
110                    self._resources.append({'name': name, 'commands': methods})
111
112    def _resources_all_names(self) -> list:
113        """Retrieve all resources' name in a case.
114
115        Returns
116        -------
117        names : list
118            List of all resources' name in a case.
119        """
120        names = []
121        for r in self._resources:
122            names.append(r['name'])
123
124        return names
125
126    def _resources_all_commands_by_name(self, name: str) -> list:
127        """Retrieve all resources' commands.
128
129        Parameters
130        ----------
131        name : str
132            The resource's name.
133
134        Returns
135        -------
136        commands : list
137            List of commands for the resource.
138        """
139        commands = []
140        for r in filter(lambda x: x['name'] == name, self._resources):
141            commands += list(r['commands'].keys())  # type: ignore
142
143        return commands
144
145    def _resources_all_parameters_by_command(self, name: str,
146                                             command: str,
147                                             required_only=False) -> list:
148        """Retrieve all parameters of a command of a resource.
149
150        Parameters
151        ----------
152        name : str
153            The resource's name.
154        command : str
155            The command's name.
156        required_only : bool
157            Only return the required parameters of a command. Default all
158            parameters are returned.
159
160        Returns
161        -------
162        parameters : list
163            List of parameters of the resource's command. None if failed.
164
165        Raises
166        ------
167        KeyError : Exception
168            If the command cannot be found for the resource.
169        """
170        parameters = []
171        for r in filter(lambda x: x['name'] == name, self._resources):
172            try:
173                for p in r['commands'][command]:
174                    if required_only:
175                        if p['required']:
176                            parameters.append(p['name'])
177                    else:
178                        parameters.append(p['name'])
179            except KeyError as e:
180                self._logger.error(f'Command "{command}" not found for '
181                                   f'resource "{name}": {e}')
182                raise e
183
184        return parameters
185
186    def _validate_instance(self, generator: str, parameters: dict) -> bool:
187        # Verify if we have a generator implementation available
188        if generator not in self._resources_all_names():
189            return False
190
191        init_params = self._resources_all_parameters_by_command(generator,
192                                                                '__init__',
193                                                                True)
194        for p in init_params:
195            if p == 'main_directory' or p == 'verbose':
196                continue
197            if p not in parameters:
198                print(f'❌ Parameter "{p}" is missing for generator'
199                      f' "{generator}"', file=sys.stderr)
200                return False
201        return True
202
203    def list(self, scenario) -> list:
204        """List all scenarios in file.
205
206        Returns
207        -------
208        scenarios : list
209            List of scenarios in file.
210        """
211        scenarios: list = []
212        path = os.path.join(self._main_directory, scenario)
213
214        if not os.path.exists(path):
215            print(f'❌ "{scenario}" does not exist', file=sys.stderr)
216            return []
217
218        with open(path, 'r') as f:
219            data = json.load(f)
220            for s in data['instances']:
221                scenarios.append(s['name'])
222
223        return scenarios
224
225    def _generate_scenario(self, scenario: dict) -> bool:
226        """Generate a scenario.
227
228        Parameters
229        ----------
230        scenario : dict
231            Scenario description.
232
233        Returns
234        -------
235        success : bool
236            True if successfull, otherwise false.
237        """
238        generator: str = scenario['generator']
239        parameters: dict = scenario['parameters']
240
241        module = self._class_module_mapping[generator]
242        cls = getattr(module, generator)(self._main_directory, self._verbose,
243                                         **parameters)
244        getattr(cls, 'generate')()
245
246        return True
247
248    def generate(self, scenario: str) -> bool:
249        """Generate scenarios listed in supplied file.
250
251        Parameters
252        ----------
253        scenario : str
254            Path to file with scenarios to generate.
255
256        Returns
257        -------
258        success : bool
259            True if successfully generated, otherwise false.
260        """
261        if not os.path.exists(os.path.join(self._main_directory, scenario)):
262            print(f'"{scenario}" does not exist', file=sys.stderr)
263            return False
264
265        path = os.path.join(self._main_directory, scenario)
266        success: bool = True
267        with open(path, 'r') as f:
268            data = json.load(f)
269
270        # Validate scenario instances
271        try:
272            jsonschema.validate(data, self._schema)
273            for s in data['instances']:
274                generator = s['generator']
275                parameters = s['parameters']
276                if not self._validate_instance(generator, parameters):
277                    return False
278            self._logger.debug('Scenario instances valid')
279        except jsonschema.ValidationError:
280            msg = f'{path}: JSON schema violation'
281            self._logger.error(msg)
282            return False
283
284        # Generate instances
285        self._logger.debug(f'Scenario: {data["name"]}')
286        print('Generating scenario\'s instances:')
287        for s in data['instances']:
288            if not self._generate_scenario(s):
289                print('    ❌ Failed to generate scenario'
290                      f' "{s["name"]}"',
291                      file=sys.stderr)
292                success = False
293            else:
294                print(f'    ✅ {s["name"]}')
295
296        return success
SCHEMA_FILE = 'metadata.schema'
DATA_DIR = '/home/dylan/Projects/KROWN/data-generator/bench_generator/data'
class Generator:
 23class Generator:
 24    """
 25    Generator class generates a benchmark testbed.
 26    """
 27
 28    def __init__(self, main_directory: str, verbose: bool = False):
 29        """Create an instance of the Benchmark class.
 30
 31        Parameters
 32        ----------
 33        main_directory : str
 34            The root directory of all the cases to execute.
 35        verbose : bool
 36            Enables verbose logs.
 37        """
 38        self._main_directory = os.path.abspath(main_directory)
 39        self._resources: List[Dict[str, Any]] = []
 40        self._class_module_mapping: Dict[str, Any] = {}
 41        self._verbose = verbose
 42        self._logger = Logger(__name__, self._main_directory, self._verbose)
 43        self._schema = {}
 44
 45        with open(os.path.join(DATA_DIR, SCHEMA_FILE)) as f:
 46            self._schema = json.load(f)
 47
 48        self._init_resources()
 49
 50    @property
 51    def main_directory(self) -> str:
 52        """The main directory of all the cases.
 53
 54        Returns
 55        -------
 56        main_directory : str
 57            The path to the main directory of the cases.
 58        """
 59        return self._main_directory
 60
 61    def _init_resources(self) -> None:
 62        """Initialize resources of a case
 63
 64        Resources are discovered automatically by analyzing Python modules.
 65        """
 66
 67        # Discover all modules to import
 68        sys.path.append(os.path.dirname(__file__))
 69        self._modules = list(filter(lambda x: x.endswith('.py')
 70                                    and '__init__' not in x
 71                                    and '__pycache__' not in x,
 72                                    os.listdir(os.path.dirname(__file__))))
 73
 74        # Discover all classes in each module
 75        for m in self._modules:
 76            module_name = os.path.splitext(m)[0]
 77            parent_module = os.path.split(os.path.dirname(__file__))[-1]
 78            import_name = '.'.join([parent_module, module_name])
 79            imported_module = importlib.import_module(import_name)
 80            for name, cls in inspect.getmembers(imported_module,
 81                                                inspect.isclass):
 82                parent_cls = cls.__base__
 83
 84                # Skip hidden classes
 85                if name.startswith('_') or name[0].islower():
 86                    continue
 87
 88                # Skip classes which do not inherit from Scenario
 89                if parent_cls is None or parent_cls.__name__ != 'Scenario':
 90                    continue
 91
 92                # Store class-module mapping for reverse look-up
 93                self._class_module_mapping[name] = imported_module
 94
 95                # Discover all methods and their parameters in each class
 96                methods: Dict[str, List[Dict[str, str]]] = {}
 97                filt = inspect.getmembers(cls, inspect.isfunction)
 98                for method_name, method in filt:
 99                    parameters = inspect.signature(method).parameters
100                    methods[method_name] = []
101                    for key in parameters.keys():
102                        if key == 'self':
103                            continue
104                        p = parameters[key]
105                        required = (p.default == inspect.Parameter.empty)
106                        methods[method_name].append({'name': p.name,
107                                                     'required': required})
108
109                if name not in list(filter(lambda x: x['name'],
110                                           self._resources)):
111                    self._resources.append({'name': name, 'commands': methods})
112
113    def _resources_all_names(self) -> list:
114        """Retrieve all resources' name in a case.
115
116        Returns
117        -------
118        names : list
119            List of all resources' name in a case.
120        """
121        names = []
122        for r in self._resources:
123            names.append(r['name'])
124
125        return names
126
127    def _resources_all_commands_by_name(self, name: str) -> list:
128        """Retrieve all resources' commands.
129
130        Parameters
131        ----------
132        name : str
133            The resource's name.
134
135        Returns
136        -------
137        commands : list
138            List of commands for the resource.
139        """
140        commands = []
141        for r in filter(lambda x: x['name'] == name, self._resources):
142            commands += list(r['commands'].keys())  # type: ignore
143
144        return commands
145
146    def _resources_all_parameters_by_command(self, name: str,
147                                             command: str,
148                                             required_only=False) -> list:
149        """Retrieve all parameters of a command of a resource.
150
151        Parameters
152        ----------
153        name : str
154            The resource's name.
155        command : str
156            The command's name.
157        required_only : bool
158            Only return the required parameters of a command. Default all
159            parameters are returned.
160
161        Returns
162        -------
163        parameters : list
164            List of parameters of the resource's command. None if failed.
165
166        Raises
167        ------
168        KeyError : Exception
169            If the command cannot be found for the resource.
170        """
171        parameters = []
172        for r in filter(lambda x: x['name'] == name, self._resources):
173            try:
174                for p in r['commands'][command]:
175                    if required_only:
176                        if p['required']:
177                            parameters.append(p['name'])
178                    else:
179                        parameters.append(p['name'])
180            except KeyError as e:
181                self._logger.error(f'Command "{command}" not found for '
182                                   f'resource "{name}": {e}')
183                raise e
184
185        return parameters
186
187    def _validate_instance(self, generator: str, parameters: dict) -> bool:
188        # Verify if we have a generator implementation available
189        if generator not in self._resources_all_names():
190            return False
191
192        init_params = self._resources_all_parameters_by_command(generator,
193                                                                '__init__',
194                                                                True)
195        for p in init_params:
196            if p == 'main_directory' or p == 'verbose':
197                continue
198            if p not in parameters:
199                print(f'❌ Parameter "{p}" is missing for generator'
200                      f' "{generator}"', file=sys.stderr)
201                return False
202        return True
203
204    def list(self, scenario) -> list:
205        """List all scenarios in file.
206
207        Returns
208        -------
209        scenarios : list
210            List of scenarios in file.
211        """
212        scenarios: list = []
213        path = os.path.join(self._main_directory, scenario)
214
215        if not os.path.exists(path):
216            print(f'❌ "{scenario}" does not exist', file=sys.stderr)
217            return []
218
219        with open(path, 'r') as f:
220            data = json.load(f)
221            for s in data['instances']:
222                scenarios.append(s['name'])
223
224        return scenarios
225
226    def _generate_scenario(self, scenario: dict) -> bool:
227        """Generate a scenario.
228
229        Parameters
230        ----------
231        scenario : dict
232            Scenario description.
233
234        Returns
235        -------
236        success : bool
237            True if successfull, otherwise false.
238        """
239        generator: str = scenario['generator']
240        parameters: dict = scenario['parameters']
241
242        module = self._class_module_mapping[generator]
243        cls = getattr(module, generator)(self._main_directory, self._verbose,
244                                         **parameters)
245        getattr(cls, 'generate')()
246
247        return True
248
249    def generate(self, scenario: str) -> bool:
250        """Generate scenarios listed in supplied file.
251
252        Parameters
253        ----------
254        scenario : str
255            Path to file with scenarios to generate.
256
257        Returns
258        -------
259        success : bool
260            True if successfully generated, otherwise false.
261        """
262        if not os.path.exists(os.path.join(self._main_directory, scenario)):
263            print(f'"{scenario}" does not exist', file=sys.stderr)
264            return False
265
266        path = os.path.join(self._main_directory, scenario)
267        success: bool = True
268        with open(path, 'r') as f:
269            data = json.load(f)
270
271        # Validate scenario instances
272        try:
273            jsonschema.validate(data, self._schema)
274            for s in data['instances']:
275                generator = s['generator']
276                parameters = s['parameters']
277                if not self._validate_instance(generator, parameters):
278                    return False
279            self._logger.debug('Scenario instances valid')
280        except jsonschema.ValidationError:
281            msg = f'{path}: JSON schema violation'
282            self._logger.error(msg)
283            return False
284
285        # Generate instances
286        self._logger.debug(f'Scenario: {data["name"]}')
287        print('Generating scenario\'s instances:')
288        for s in data['instances']:
289            if not self._generate_scenario(s):
290                print('    ❌ Failed to generate scenario'
291                      f' "{s["name"]}"',
292                      file=sys.stderr)
293                success = False
294            else:
295                print(f'    ✅ {s["name"]}')
296
297        return success

Generator class generates a benchmark testbed.

Generator(main_directory: str, verbose: bool = False)
28    def __init__(self, main_directory: str, verbose: bool = False):
29        """Create an instance of the Benchmark class.
30
31        Parameters
32        ----------
33        main_directory : str
34            The root directory of all the cases to execute.
35        verbose : bool
36            Enables verbose logs.
37        """
38        self._main_directory = os.path.abspath(main_directory)
39        self._resources: List[Dict[str, Any]] = []
40        self._class_module_mapping: Dict[str, Any] = {}
41        self._verbose = verbose
42        self._logger = Logger(__name__, self._main_directory, self._verbose)
43        self._schema = {}
44
45        with open(os.path.join(DATA_DIR, SCHEMA_FILE)) as f:
46            self._schema = json.load(f)
47
48        self._init_resources()

Create an instance of the Benchmark class.

Parameters
  • main_directory (str): The root directory of all the cases to execute.
  • verbose (bool): Enables verbose logs.
main_directory: str
50    @property
51    def main_directory(self) -> str:
52        """The main directory of all the cases.
53
54        Returns
55        -------
56        main_directory : str
57            The path to the main directory of the cases.
58        """
59        return self._main_directory

The main directory of all the cases.

Returns
  • main_directory (str): The path to the main directory of the cases.
def list(self, scenario) -> list:
204    def list(self, scenario) -> list:
205        """List all scenarios in file.
206
207        Returns
208        -------
209        scenarios : list
210            List of scenarios in file.
211        """
212        scenarios: list = []
213        path = os.path.join(self._main_directory, scenario)
214
215        if not os.path.exists(path):
216            print(f'❌ "{scenario}" does not exist', file=sys.stderr)
217            return []
218
219        with open(path, 'r') as f:
220            data = json.load(f)
221            for s in data['instances']:
222                scenarios.append(s['name'])
223
224        return scenarios

List all scenarios in file.

Returns
  • scenarios (list): List of scenarios in file.
def generate(self, scenario: str) -> bool:
249    def generate(self, scenario: str) -> bool:
250        """Generate scenarios listed in supplied file.
251
252        Parameters
253        ----------
254        scenario : str
255            Path to file with scenarios to generate.
256
257        Returns
258        -------
259        success : bool
260            True if successfully generated, otherwise false.
261        """
262        if not os.path.exists(os.path.join(self._main_directory, scenario)):
263            print(f'"{scenario}" does not exist', file=sys.stderr)
264            return False
265
266        path = os.path.join(self._main_directory, scenario)
267        success: bool = True
268        with open(path, 'r') as f:
269            data = json.load(f)
270
271        # Validate scenario instances
272        try:
273            jsonschema.validate(data, self._schema)
274            for s in data['instances']:
275                generator = s['generator']
276                parameters = s['parameters']
277                if not self._validate_instance(generator, parameters):
278                    return False
279            self._logger.debug('Scenario instances valid')
280        except jsonschema.ValidationError:
281            msg = f'{path}: JSON schema violation'
282            self._logger.error(msg)
283            return False
284
285        # Generate instances
286        self._logger.debug(f'Scenario: {data["name"]}')
287        print('Generating scenario\'s instances:')
288        for s in data['instances']:
289            if not self._generate_scenario(s):
290                print('    ❌ Failed to generate scenario'
291                      f' "{s["name"]}"',
292                      file=sys.stderr)
293                success = False
294            else:
295                print(f'    ✅ {s["name"]}')
296
297        return success

Generate scenarios listed in supplied file.

Parameters
  • scenario (str): Path to file with scenarios to generate.
Returns
  • success (bool): True if successfully generated, otherwise false.