Module for processing response from conductor for slice selection
"""
-from osdf.logging.osdf_logging import debug_log
-
-
-SLICE_PROFILE_FIELDS = ["latency", "max_number_of_ues", "coverage_area_ta_list",
- "ue_mobility_level", "resource_sharing_level", "exp_data_rate_ul",
- "exp_data_rate_dl", "area_traffic_cap_ul", "area_traffic_cap_dl",
- "activity_factor", "e2e_latency", "jitter", "survival_time",
- "exp_data_rate", "payload_size", "traffic_density", "conn_density",
- "reliability", "service_area_dimension", "cs_availability"]
-
-
-def conductor_response_processor(overall_recommendations, nst_info_map, request_info):
- """Process conductor response to form the response for the API request
- :param overall_recommendations: recommendations from conductor
- :param nst_info_map: NST info from the request
- :param request_info: request info
- :return: response json as a dictionary
- """
- shared_nsi_solutions = list()
- new_nsi_solutions = list()
-
- for nst_name, recommendations in overall_recommendations.items():
- for recommendation in recommendations:
- nsi_set = set(values['candidate']['nsi_id'] for key, values in recommendation.items())
- if len(nsi_set) == 1:
- nsi_id = nsi_set.pop()
- candidate = list(recommendation.values())[0]['candidate']
- debug_log.debug("The NSSIs in the solution belongs to the same NSI {}"
- .format(nsi_id))
- shared_nsi_solution = dict()
- shared_nsi_solution["NSIId"] = nsi_id
- shared_nsi_solution["NSIName"] = candidate.get('nsi_name')
- shared_nsi_solution["UUID"] = candidate.get('nsi_model_version_id')
- shared_nsi_solution["invariantUUID"] = candidate.get('nsi_model_invariant_id')
- shared_nsi_solutions.append(shared_nsi_solution)
- else:
- nssi_solutions = get_nssi_solutions(recommendation)
- new_nsi_solution = dict()
- new_nsi_solution['matchLevel'] = ""
- new_nsi_solution['NSTInfo'] = nst_info_map.get(nst_name)
- new_nsi_solution['NSSISolutions'] = nssi_solutions
- new_nsi_solutions.append(new_nsi_solution)
-
- solutions = dict()
- solutions['sharedNSISolutions'] = shared_nsi_solutions
- solutions['newNSISolutions'] = new_nsi_solutions
- return get_nsi_selection_response(request_info, solutions)
-
-
-def conductor_error_response_processor(request_info, error_message):
- """Form response message from the error message
- :param request_info: request info
- :param error_message: error message while processing the request
- :return: response json as dictionary
- """
- return {'requestId': request_info['requestId'],
- 'transactionId': request_info['transactionId'],
- 'requestStatus': 'error',
- 'statusMessage': error_message}
-
-
-def get_nssi_solutions(recommendation):
- """Get nssi solutions from recommendation
- :param recommendation: recommendation from conductor
- :return: new nssi solutions list
- """
- nssi_solutions = list()
-
- for nsst_name, nsst_rec in recommendation.items():
- candidate = nsst_rec['candidate']
- nssi_info, slice_profile = get_solution_from_candidate(candidate)
- nsst_info = {"NSSTName": nsst_name}
- nssi_solution = {"sliceProfile": slice_profile,
- "NSSTInfo": nsst_info,
- "NSSISolution": nssi_info}
- nssi_solutions.append(nssi_solution)
- return nssi_solutions
-
-
-def get_solution_from_candidate(candidate):
- """Get nssi info from candidate
- :param candidate: Candidate from the recommendation
- :return: nssi_info and slice profile derived from candidate
- """
- slice_profile = dict()
- nssi_info = {"NSSIName": candidate['instance_name'],
- "NSSIId": candidate['candidate_id']}
-
- for field in SLICE_PROFILE_FIELDS:
- if candidate[field]:
- slice_profile[field] = candidate[field]
-
- return nssi_info, slice_profile
-
-
-def get_nsi_selection_response(request_info, solutions):
- """Get NSI selection response from final solution
- :param request_info: request info
- :param solutions: final solutions
- :return: NSI selection response to send back as dictionary
- """
- return {'requestId': request_info['requestId'],
- 'transactionId': request_info['transactionId'],
- 'requestStatus': 'completed',
- 'statusMessage': '',
- 'solutions': solutions}
+import re
+
+
+class ResponseProcessor(object):
+ def __init__(self, request_info, slice_config):
+ self.request_info = request_info
+ self.slice_config = slice_config
+
+ def process_response(self, recommendations, model_info, subnets, model_type):
+ """Process conductor response to form the response for the API request
+
+ :param recommendations: recommendations from conductor
+ :param model_info: model info from the request
+ :param subnets: list of subnets
+ :param model_type: NSI or NSSI
+ :return: response json as a dictionary
+ """
+ if not recommendations:
+ return self.get_slice_selection_response([])
+ model_name = model_info['name']
+ solutions = [self.get_solution_from_candidate(rec[model_name]['candidate'], model_info, subnets, model_type)
+ for rec in recommendations]
+ return self.get_slice_selection_response(solutions)
+
+ def get_solution_from_candidate(self, candidate, model_info, subnets, model_type):
+ if candidate['inventory_type'] == 'slice_profiles':
+ return {
+ 'existingNSI': False,
+ 'newNSISolution': {
+ 'sliceProfiles': self.get_slice_profiles_from_candidate(candidate, subnets)
+ }
+ }
+ elif model_type == 'NSSI':
+ return {
+ 'UUID': model_info['UUID'],
+ 'invariantUUID': model_info['invariantUUID'],
+ 'NSSIName': candidate['instance_name'],
+ 'NSSIId': candidate['instance_id']
+ }
+
+ elif model_type == 'NSI':
+ return {
+ 'existingNSI': True,
+ 'sharedNSISolution': {
+ 'UUID': model_info['UUID'],
+ 'invariantUUID': model_info['invariantUUID'],
+ 'NSIName': candidate['instance_name'],
+ 'NSIId': candidate['instance_id']
+ }
+ }
+
+ def get_slice_profiles_from_candidate(self, candidate, subnets):
+ slice_profiles = []
+ for subnet in subnets:
+ slice_profile = {self.get_profile_attribute(k, subnet): v for k, v in candidate.items()
+ if k.startswith(subnet)}
+ slice_profile['domainType'] = subnet
+ slice_profiles.append(slice_profile)
+ return slice_profiles
+
+ def get_profile_attribute(self, attribute, subnet):
+ snake_to_camel = self.slice_config['attribute_mapping']['snake_to_camel']
+ return snake_to_camel[re.sub(f'^{subnet}_', '', attribute)]
+
+ def process_error_response(self, error_message):
+ """Form response message from the error message
+
+ :param error_message: error message while processing the request
+ :return: response json as dictionary
+ """
+ return {'requestId': self.request_info['requestId'],
+ 'transactionId': self.request_info['transactionId'],
+ 'requestStatus': 'error',
+ 'statusMessage': error_message}
+
+ def get_slice_selection_response(self, solutions):
+ """Get NSI selection response from final solution
+
+ :param solutions: final solutions
+ :return: NSI selection response to send back as dictionary
+ """
+ return {'requestId': self.request_info['requestId'],
+ 'transactionId': self.request_info['transactionId'],
+ 'requestStatus': 'completed',
+ 'statusMessage': '',
+ 'solutions': solutions}
Code Review / optf / osdf.git / blobdiff