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.