#!/usr/bin/env python3.4 # # Copyright 2022 - The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the 'License'); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an 'AS IS' BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import collections import csv import itertools import numpy import json import os from acts import context from acts import base_test from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure from acts_contrib.test_utils.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest from functools import partial class CellularFr2SensitivityTest(CellularThroughputBaseTest): """Class to test single cell FR1 NSA sensitivity""" def __init__(self, controllers): base_test.BaseTestClass.__init__(self, controllers) self.testcase_metric_logger = ( BlackboxMappedMetricLogger.for_test_case()) self.testclass_metric_logger = ( BlackboxMappedMetricLogger.for_test_class()) self.publish_testcase_metrics = True self.testclass_params = self.user_params['fr2_sensitivity_test_params'] self.log.info('Hello') self.tests = self.generate_test_cases( band_list=['N257', 'N258', 'N260', 'N261'], channel_list=['low', 'mid', 'high'], dl_mcs_list=list(numpy.arange(28, -1, -1)), num_dl_cells_list=[1, 2, 4, 8], orientation_list=['A_Plane', 'B_Plane'], dl_mimo_config=2, nr_ul_mcs=4, lte_dl_mcs_table='QAM256', lte_dl_mcs=4, lte_ul_mcs_table='QAM256', lte_ul_mcs=4, schedule_scenario="FULL_TPUT", schedule_slot_ratio=80, force_contiguous_nr_channel=True, transform_precoding=0) def process_testclass_results(self): # Plot individual test id results raw data and compile metrics plots = collections.OrderedDict() compiled_data = collections.OrderedDict() for testcase_name, testcase_data in self.testclass_results.items(): nr_cell_index = testcase_data['testcase_params'][ 'endc_combo_config']['lte_cell_count'] cell_config = testcase_data['testcase_params'][ 'endc_combo_config']['cell_list'][nr_cell_index] test_id = tuple(('band', cell_config['band'])) if test_id not in plots: # Initialize test id data when not present compiled_data[test_id] = { 'mcs': [], 'average_throughput': [], 'theoretical_throughput': [], 'cell_power': [], } plots[test_id] = BokehFigure( title='Band {} - BLER Curves'.format(cell_config['band']), x_label='Cell Power (dBm)', primary_y_label='BLER (Mbps)') test_id_rvr = test_id + tuple('RvR') plots[test_id_rvr] = BokehFigure( title='Band {} - RvR'.format(cell_config['band']), x_label='Cell Power (dBm)', primary_y_label='PHY Rate (Mbps)') # Compile test id data and metrics compiled_data[test_id]['average_throughput'].append( testcase_data['average_throughput_list']) compiled_data[test_id]['cell_power'].append( testcase_data['cell_power_list']) compiled_data[test_id]['mcs'].append( testcase_data['testcase_params']['nr_dl_mcs']) # Add test id to plots plots[test_id].add_line( testcase_data['cell_power_list'], testcase_data['bler_list'], 'MCS {}'.format(testcase_data['testcase_params']['nr_dl_mcs']), width=1) plots[test_id_rvr].add_line( testcase_data['cell_power_list'], testcase_data['average_throughput_list'], 'MCS {}'.format(testcase_data['testcase_params']['nr_dl_mcs']), width=1, style='dashed') for test_id, test_data in compiled_data.items(): test_id_rvr = test_id + tuple('RvR') cell_power_interp = sorted(set(sum(test_data['cell_power'], []))) average_throughput_interp = [] for mcs, cell_power, throughput in zip( test_data['mcs'], test_data['cell_power'], test_data['average_throughput']): throughput_interp = numpy.interp(cell_power_interp, cell_power[::-1], throughput[::-1]) average_throughput_interp.append(throughput_interp) rvr = numpy.max(average_throughput_interp, 0) plots[test_id_rvr].add_line(cell_power_interp, rvr, 'Rate vs. Range') figure_list = [] for plot_id, plot in plots.items(): plot.generate_figure() figure_list.append(plot) output_file_path = os.path.join(self.log_path, 'results.html') BokehFigure.save_figures(figure_list, output_file_path) """Saves CSV with all test results to enable comparison.""" results_file_path = os.path.join( context.get_current_context().get_full_output_path(), 'results.csv') with open(results_file_path, 'w', newline='') as csvfile: field_names = ['Test Name', 'Sensitivity'] writer = csv.DictWriter(csvfile, fieldnames=field_names) writer.writeheader() for testcase_name, testcase_results in self.testclass_results.items( ): row_dict = { 'Test Name': testcase_name, 'Sensitivity': testcase_results['sensitivity'] } writer.writerow(row_dict) def process_testcase_results(self): if self.current_test_name not in self.testclass_results: return testcase_data = self.testclass_results[self.current_test_name] bler_list = [] average_throughput_list = [] theoretical_throughput_list = [] nr_cell_index = testcase_data['testcase_params']['endc_combo_config'][ 'lte_cell_count'] cell_power_list = testcase_data['testcase_params']['cell_power_sweep'][ nr_cell_index] for result in testcase_data['results']: bler_list.append(result['throughput_measurements'] ['nr_bler_result']['total']['DL']['nack_ratio']) average_throughput_list.append( result['throughput_measurements']['nr_tput_result']['total'] ['DL']['average_tput']) theoretical_throughput_list.append( result['throughput_measurements']['nr_tput_result']['total'] ['DL']['theoretical_tput']) padding_len = len(cell_power_list) - len(average_throughput_list) average_throughput_list.extend([0] * padding_len) theoretical_throughput_list.extend([0] * padding_len) bler_above_threshold = [ bler > self.testclass_params['bler_threshold'] for bler in bler_list ] for idx in range(len(bler_above_threshold)): if all(bler_above_threshold[idx:]): sensitivity_idx = max(idx, 1) - 1 break else: sensitivity_idx = -1 sensitivity = cell_power_list[sensitivity_idx] self.log.info('NR Band {} MCS {} Sensitivity = {}dBm'.format( testcase_data['testcase_params']['endc_combo_config']['cell_list'] [nr_cell_index]['band'], testcase_data['testcase_params']['nr_dl_mcs'], sensitivity)) testcase_data['bler_list'] = bler_list testcase_data['average_throughput_list'] = average_throughput_list testcase_data[ 'theoretical_throughput_list'] = theoretical_throughput_list testcase_data['cell_power_list'] = cell_power_list testcase_data['sensitivity'] = sensitivity results_file_path = os.path.join( context.get_current_context().get_full_output_path(), '{}.json'.format(self.current_test_name)) with open(results_file_path, 'w') as results_file: json.dump(wputils.serialize_dict(testcase_data), results_file, indent=4) def get_per_cell_power_sweeps(self, testcase_params): # get reference test current_band = testcase_params['endc_combo_config']['cell_list'][1][ 'band'] reference_test = None reference_sensitivity = None for testcase_name, testcase_data in self.testclass_results.items(): if testcase_data['testcase_params']['endc_combo_config'][ 'cell_list'][1]['band'] == current_band: reference_test = testcase_name reference_sensitivity = testcase_data['sensitivity'] if reference_test and reference_sensitivity and not self.retry_flag: start_atten = reference_sensitivity + self.testclass_params[ 'adjacent_mcs_gap'] self.log.info( "Reference test {} found. Sensitivity {} dBm. Starting at {} dBm" .format(reference_test, reference_sensitivity, start_atten)) else: start_atten = self.testclass_params['nr_cell_power_start'] self.log.info( "Reference test not found. Starting at {} dBm".format( start_atten)) # get current cell power start nr_cell_sweep = list( numpy.arange(start_atten, self.testclass_params['nr_cell_power_stop'], self.testclass_params['nr_cell_power_step'])) lte_sweep = [self.testclass_params['lte_cell_power'] ] * len(nr_cell_sweep) cell_power_sweeps = [lte_sweep] cell_power_sweeps.extend( [nr_cell_sweep] * testcase_params['endc_combo_config']['nr_cell_count']) return cell_power_sweeps def generate_endc_combo_config(self, test_config): """Function to generate ENDC combo config from CSV test config Args: test_config: dict containing ENDC combo config from CSV Returns: endc_combo_config: dictionary with all ENDC combo settings """ endc_combo_config = collections.OrderedDict() cell_config_list = [] lte_cell_count = 1 lte_carriers = [1] lte_scc_list = [] endc_combo_config['lte_pcc'] = 1 lte_cell = { 'cell_type': 'LTE', 'cell_number': 1, 'pcc': 1, 'band': self.testclass_params['lte_anchor_band'], 'dl_bandwidth': self.testclass_params['lte_anchor_bandwidth'], 'ul_enabled': 1, 'duplex_mode': self.testclass_params['lte_anchor_duplex_mode'], 'dl_mimo_config': 'D{nss}U{nss}'.format(nss=1), 'ul_mimo_config': 'D{nss}U{nss}'.format(nss=1), 'transmission_mode': 'TM1', 'num_codewords': 1, 'num_layers': 1, 'dl_subframe_allocation': [1] * 10, } cell_config_list.append(lte_cell) nr_cell_count = 0 nr_dl_carriers = [] nr_ul_carriers = [] for nr_cell_idx in range(1, test_config['num_dl_cells'] + 1): nr_cell = { 'cell_type': 'NR5G', 'cell_number': nr_cell_idx, 'nr_cell_type': 'NSA', 'band': test_config['nr_band'], 'duplex_mode': test_config['nr_duplex_mode'], 'channel': test_config['nr_channel'], 'dl_mimo_config': 'N{nss}X{nss}'.format(nss=test_config['nr_dl_mimo_config']), 'dl_bandwidth_class': 'A', 'dl_bandwidth': test_config['nr_bandwidth'], 'ul_enabled': 1 if nr_cell_idx <= test_config['num_ul_cells'] else 0, 'ul_bandwidth_class': 'A', 'ul_mimo_config': 'N{nss}X{nss}'.format(nss=test_config['nr_ul_mimo_config']), 'subcarrier_spacing': 'MU3' } cell_config_list.append(nr_cell) nr_cell_count = nr_cell_count + 1 nr_dl_carriers.append(nr_cell_idx) if nr_cell_idx <= test_config['num_ul_cells']: nr_ul_carriers.append(nr_cell_idx) endc_combo_config['lte_cell_count'] = lte_cell_count endc_combo_config['nr_cell_count'] = nr_cell_count endc_combo_config['nr_dl_carriers'] = nr_dl_carriers endc_combo_config['nr_ul_carriers'] = nr_ul_carriers endc_combo_config['cell_list'] = cell_config_list endc_combo_config['lte_scc_list'] = lte_scc_list endc_combo_config['lte_dl_carriers'] = lte_carriers endc_combo_config['lte_ul_carriers'] = lte_carriers return endc_combo_config def generate_test_cases(self, band_list, channel_list, dl_mcs_list, num_dl_cells_list, dl_mimo_config, orientation_list, **kwargs): """Function that auto-generates test cases for a test class.""" test_cases = [] for orientation, band, channel, num_dl_cells, nr_dl_mcs in itertools.product( orientation_list, band_list, channel_list, num_dl_cells_list, dl_mcs_list): if channel not in cputils.PCC_PRESET_MAPPING[band]: continue test_config = { 'nr_band': band, 'nr_bandwidth': 'BW100', 'nr_duplex_mode': 'TDD', 'nr_channel': channel, 'num_dl_cells': num_dl_cells, 'num_ul_cells': 1, 'nr_dl_mimo_config': dl_mimo_config, 'nr_ul_mimo_config': 1 } endc_combo_config = self.generate_endc_combo_config(test_config) test_name = 'test_fr2_{}_{}_{}CC_mcs{}_{}x{}'.format( band, channel.lower(), num_dl_cells, nr_dl_mcs, dl_mimo_config, dl_mimo_config) test_params = collections.OrderedDict( endc_combo_config=endc_combo_config, nr_dl_mcs=nr_dl_mcs, orientation=orientation, **kwargs) setattr(self, test_name, partial(self._test_throughput_bler, test_params)) test_cases.append(test_name) return test_cases