#!/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 json import re import numpy 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.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils from functools import partial LONG_SLEEP = 10 MEDIUM_SLEEP = 2 IPERF_TIMEOUT = 10 SHORT_SLEEP = 1 SUBFRAME_LENGTH = 0.001 STOP_COUNTER_LIMIT = 3 class CellularFr2PeakThroughputTest(CellularThroughputBaseTest): """Base class to test cellular FR2 throughput This class implements cellular FR2 throughput tests on a callbox setup. The class setups up the callbox in the desired configurations, configures and connects the phone, and runs traffic/iperf throughput. """ def __init__(self, controllers): super().__init__(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 def process_testcase_results(self): """Publish test case metrics and save results""" if self.current_test_name not in self.testclass_results: return testcase_data = self.testclass_results[self.current_test_name] 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) testcase_result = testcase_data['results'][0] metric_map = { 'tcp_udp_tput': testcase_result.get('iperf_throughput', float('nan')) } if testcase_data['testcase_params']['endc_combo_config'][ 'nr_cell_count']: metric_map.update({ 'nr_min_dl_tput': testcase_result['nr_tput_result']['total']['DL']['min_tput'], 'nr_max_dl_tput': testcase_result['nr_tput_result']['total']['DL']['max_tput'], 'nr_avg_dl_tput': testcase_result['nr_tput_result']['total']['DL'] ['average_tput'], 'nr_theoretical_dl_tput': testcase_result['nr_tput_result']['total']['DL'] ['theoretical_tput'], 'nr_dl_bler': testcase_result['nr_bler_result']['total']['DL']['nack_ratio'] * 100, 'nr_min_dl_tput': testcase_result['nr_tput_result']['total']['UL']['min_tput'], 'nr_max_dl_tput': testcase_result['nr_tput_result']['total']['UL']['max_tput'], 'nr_avg_dl_tput': testcase_result['nr_tput_result']['total']['UL'] ['average_tput'], 'nr_theoretical_dl_tput': testcase_result['nr_tput_result']['total']['UL'] ['theoretical_tput'], 'nr_ul_bler': testcase_result['nr_bler_result']['total']['UL']['nack_ratio'] * 100 }) if testcase_data['testcase_params']['endc_combo_config'][ 'lte_cell_count']: metric_map.update({ 'lte_min_dl_tput': testcase_result['lte_tput_result']['total']['DL']['min_tput'], 'lte_max_dl_tput': testcase_result['lte_tput_result']['total']['DL']['max_tput'], 'lte_avg_dl_tput': testcase_result['lte_tput_result']['total']['DL'] ['average_tput'], 'lte_theoretical_dl_tput': testcase_result['lte_tput_result']['total']['DL'] ['theoretical_tput'], 'lte_dl_bler': testcase_result['lte_bler_result']['total']['DL']['nack_ratio'] * 100, 'lte_min_dl_tput': testcase_result['lte_tput_result']['total']['UL']['min_tput'], 'lte_max_dl_tput': testcase_result['lte_tput_result']['total']['UL']['max_tput'], 'lte_avg_dl_tput': testcase_result['lte_tput_result']['total']['UL'] ['average_tput'], 'lte_theoretical_dl_tput': testcase_result['lte_tput_result']['total']['UL'] ['theoretical_tput'], 'lte_ul_bler': testcase_result['lte_bler_result']['total']['UL']['nack_ratio'] * 100 }) if self.publish_testcase_metrics: for metric_name, metric_value in metric_map.items(): self.testcase_metric_logger.add_metric(metric_name, metric_value) def process_testclass_results(self): """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', 'NR DL Min. Throughput', 'NR DL Max. Throughput', 'NR DL Avg. Throughput', 'NR DL Theoretical Throughput', 'NR UL Min. Throughput', 'NR UL Max. Throughput', 'NR UL Avg. Throughput', 'NR UL Theoretical Throughput', 'NR DL BLER (%)', 'NR UL BLER (%)', 'LTE DL Min. Throughput', 'LTE DL Max. Throughput', 'LTE DL Avg. Throughput', 'LTE DL Theoretical Throughput', 'LTE UL Min. Throughput', 'LTE UL Max. Throughput', 'LTE UL Avg. Throughput', 'LTE UL Theoretical Throughput', 'LTE DL BLER (%)', 'LTE UL BLER (%)', 'TCP/UDP Throughput' ] writer = csv.DictWriter(csvfile, fieldnames=field_names) writer.writeheader() for testcase_name, testcase_results in self.testclass_results.items( ): for result in testcase_results['results']: row_dict = { 'Test Name': testcase_name, 'TCP/UDP Throughput': result.get('iperf_throughput', 0) } if testcase_results['testcase_params'][ 'endc_combo_config']['nr_cell_count']: row_dict.update({ 'NR DL Min. Throughput': result['nr_tput_result']['total']['DL'] ['min_tput'], 'NR DL Max. Throughput': result['nr_tput_result']['total']['DL'] ['max_tput'], 'NR DL Avg. Throughput': result['nr_tput_result']['total']['DL'] ['average_tput'], 'NR DL Theoretical Throughput': result['nr_tput_result']['total']['DL'] ['theoretical_tput'], 'NR UL Min. Throughput': result['nr_tput_result']['total']['UL'] ['min_tput'], 'NR UL Max. Throughput': result['nr_tput_result']['total']['UL'] ['max_tput'], 'NR UL Avg. Throughput': result['nr_tput_result']['total']['UL'] ['average_tput'], 'NR UL Theoretical Throughput': result['nr_tput_result']['total']['UL'] ['theoretical_tput'], 'NR DL BLER (%)': result['nr_bler_result']['total']['DL'] ['nack_ratio'] * 100, 'NR UL BLER (%)': result['nr_bler_result']['total']['UL'] ['nack_ratio'] * 100 }) if testcase_results['testcase_params'][ 'endc_combo_config']['lte_cell_count']: row_dict.update({ 'LTE DL Min. Throughput': result['lte_tput_result']['total']['DL'] ['min_tput'], 'LTE DL Max. Throughput': result['lte_tput_result']['total']['DL'] ['max_tput'], 'LTE DL Avg. Throughput': result['lte_tput_result']['total']['DL'] ['average_tput'], 'LTE DL Theoretical Throughput': result['lte_tput_result']['total']['DL'] ['theoretical_tput'], 'LTE UL Min. Throughput': result['lte_tput_result']['total']['UL'] ['min_tput'], 'LTE UL Max. Throughput': result['lte_tput_result']['total']['UL'] ['max_tput'], 'LTE UL Avg. Throughput': result['lte_tput_result']['total']['UL'] ['average_tput'], 'LTE UL Theoretical Throughput': result['lte_tput_result']['total']['UL'] ['theoretical_tput'], 'LTE DL BLER (%)': result['lte_bler_result']['total']['DL'] ['nack_ratio'] * 100, 'LTE UL BLER (%)': result['lte_bler_result']['total']['UL'] ['nack_ratio'] * 100 }) writer.writerow(row_dict) def get_per_cell_power_sweeps(self, testcase_params): """Function to get per cell power sweep lists Args: testcase_params: dict containing all test case params Returns: cell_power_sweeps: list of cell power sweeps for each cell under test """ cell_power_sweeps = [] for cell in testcase_params['endc_combo_config']['cell_list']: if cell['cell_type'] == 'LTE': sweep = [self.testclass_params['lte_cell_power']] else: sweep = [self.testclass_params['nr_cell_power']] cell_power_sweeps.append(sweep) 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': test_config['lte_band'], 'dl_bandwidth': test_config['lte_bandwidth'], 'ul_enabled': 1, 'duplex_mode': test_config['lte_duplex_mode'], 'dl_mimo_config': 'D{nss}U{nss}'.format(nss=test_config['lte_dl_mimo_config']), 'ul_mimo_config': 'D{nss}U{nss}'.format(nss=test_config['lte_ul_mimo_config']), 'transmission_mode': 'TM1' } 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, '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_carriers'] = lte_carriers return endc_combo_config def generate_test_cases(self, bands, channels, nr_mcs_pair_list, num_dl_cells_list, num_ul_cells_list, dl_mimo_config, ul_mimo_config, **kwargs): """Function that auto-generates test cases for a test class.""" test_cases = [] for band, channel, num_ul_cells, num_dl_cells, nr_mcs_pair in itertools.product( bands, channels, num_ul_cells_list, num_dl_cells_list, nr_mcs_pair_list): if num_ul_cells > num_dl_cells: continue if channel not in cputils.PCC_PRESET_MAPPING[band]: continue test_config = { 'lte_band': 2, 'lte_bandwidth': 'BW20', 'lte_duplex_mode': 'FDD', 'lte_dl_mimo_config': 1, 'lte_ul_mimo_config': 1, 'nr_band': band, 'nr_bandwidth': 'BW100', 'nr_duplex_mode': 'TDD', 'nr_channel': channel, 'num_dl_cells': num_dl_cells, 'num_ul_cells': num_ul_cells, 'nr_dl_mimo_config': dl_mimo_config, 'nr_ul_mimo_config': ul_mimo_config } endc_combo_config = self.generate_endc_combo_config(test_config) test_name = 'test_fr2_{}_{}_DL_{}CC_mcs{}_{}x{}_UL_{}CC_mcs{}_{}x{}'.format( band, channel, num_dl_cells, nr_mcs_pair[0], dl_mimo_config, dl_mimo_config, num_ul_cells, nr_mcs_pair[1], ul_mimo_config, ul_mimo_config) test_params = collections.OrderedDict( endc_combo_config=endc_combo_config, nr_dl_mcs=nr_mcs_pair[0], nr_ul_mcs=nr_mcs_pair[1], **kwargs) setattr(self, test_name, partial(self._test_throughput_bler, test_params)) test_cases.append(test_name) return test_cases class CellularFr2DlPeakThroughputTest(CellularFr2PeakThroughputTest): """Base class to test cellular FR2 throughput This class implements cellular FR2 throughput tests on a callbox setup. The class setups up the callbox in the desired configurations, configures and connects the phone, and runs traffic/iperf throughput. """ def __init__(self, controllers): super().__init__(controllers) self.testclass_params = self.user_params['throughput_test_params'] self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(16, 4), (27, 4)], list(range(1, 9)), list(range(1, 3)), force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=1, schedule_scenario="FULL_TPUT", traffic_direction='DL', transform_precoding=0, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64') class CellularFr2CpOfdmUlPeakThroughputTest(CellularFr2PeakThroughputTest): def __init__(self, controllers): super().__init__(controllers) self.testclass_params = self.user_params['throughput_test_params'] self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [1], [1], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=1, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=0, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64') self.tests.extend( self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [1], [1], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=2, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=0, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64')) self.tests.extend( self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [2], [2], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=2, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=0, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64')) self.tests.extend( self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [4], [4], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=2, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=0, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64')) class CellularFr2DftsOfdmUlPeakThroughputTest(CellularFr2PeakThroughputTest): def __init__(self, controllers): super().__init__(controllers) self.testclass_params = self.user_params['throughput_test_params'] self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [1], [1], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=1, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=1, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64') self.tests.extend( self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [1], [1], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=2, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=1, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64')) self.tests.extend( self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [2], [2], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=2, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=1, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64')) self.tests.extend( self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], ['low', 'mid', 'high'], [(4, 16), (4, 27)], [4], [4], force_contiguous_nr_channel=True, dl_mimo_config=2, ul_mimo_config=2, schedule_scenario="FULL_TPUT", traffic_direction='UL', transform_precoding=1, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64')) class CellularFr2DlFrequencySweepPeakThroughputTest( CellularFr2PeakThroughputTest): """Base class to test cellular FR2 throughput This class implements cellular FR2 throughput tests on a callbox setup. The class setups up the callbox in the desired configurations, configures and connects the phone, and runs traffic/iperf throughput. """ def __init__(self, controllers): super().__init__(controllers) self.testclass_params = self.user_params['throughput_test_params'] self.tests = self.generate_test_cases( ['N257', 'N258', 'N260', 'N261'], self.user_params['throughput_test_params']['frequency_sweep'], [(16, 4), (27, 4)], force_contiguous_nr_channel=False, dl_mimo_config=2, ul_mimo_config=1, schedule_scenario="FULL_TPUT", traffic_direction='DL', transform_precoding=0, lte_dl_mcs=4, lte_dl_mcs_table='QAM256', lte_ul_mcs=4, lte_ul_mcs_table='QAM64') def generate_test_cases(self, bands, channels, nr_mcs_pair_list, num_dl_cells_list, num_ul_cells_list, dl_mimo_config, ul_mimo_config, **kwargs): """Function that auto-generates test cases for a test class.""" test_cases = [] for band, channel, num_ul_cells, num_dl_cells, nr_mcs_pair in itertools.product( bands, channels, num_ul_cells_list, num_dl_cells_list, nr_mcs_pair_list): if num_ul_cells > num_dl_cells: continue if channel not in cputils.PCC_PRESET_MAPPING[band]: continue test_config = { 'lte_band': 2, 'lte_bandwidth': 'BW20', 'lte_duplex_mode': 'FDD', 'lte_dl_mimo_config': 1, 'lte_ul_mimo_config': 1, 'nr_band': band, 'nr_bandwidth': 'BW100', 'nr_duplex_mode': 'TDD', 'nr_channel': channel, 'num_dl_cells': num_dl_cells, 'num_ul_cells': num_ul_cells, 'nr_dl_mimo_config': dl_mimo_config, 'nr_ul_mimo_config': ul_mimo_config } endc_combo_config = self.generate_endc_combo_config(test_config) test_name = 'test_fr2_{}_{}_DL_{}CC_mcs{}_{}x{}_UL_{}CC_mcs{}_{}x{}'.format( band, channel, num_dl_cells, nr_mcs_pair[0], dl_mimo_config, dl_mimo_config, num_ul_cells, nr_mcs_pair[1], ul_mimo_config, ul_mimo_config) test_params = collections.OrderedDict( endc_combo_config=endc_combo_config, nr_dl_mcs=nr_mcs_pair[0], nr_ul_mcs=nr_mcs_pair[1], **kwargs) setattr(self, test_name, partial(self._test_throughput_bler, test_params)) test_cases.append(test_name) return test_cases