| #!/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 CellularFr2UplinkPowerSweepTest(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] |
| |
| def process_testclass_results(self): |
| pass |
| |
| def get_per_cell_power_sweeps(self, testcase_params): |
| 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': 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 _test_throughput_bler_sweep_ul_power(self, testcase_params): |
| """Test function to run cellular throughput and BLER measurements. |
| |
| The function runs BLER/throughput measurement after configuring the |
| callbox and DUT. The test supports running PHY or TCP/UDP layer traffic |
| in a variety of band/carrier/mcs/etc configurations. |
| |
| Args: |
| testcase_params: dict containing test-specific parameters |
| Returns: |
| result: dict containing throughput results and metadata |
| """ |
| # Prepare results dicts |
| testcase_params = self.compile_test_params(testcase_params) |
| testcase_params['nr_target_power_sweep'] = list( |
| numpy.arange(self.testclass_params['nr_target_power_start'], |
| self.testclass_params['nr_target_power_stop'], |
| self.testclass_params['nr_target_power_step'])) |
| |
| testcase_results = collections.OrderedDict() |
| testcase_results['testcase_params'] = testcase_params |
| testcase_results['results'] = [] |
| |
| # Setup ota chamber if needed |
| if hasattr(self, |
| 'keysight_chamber') and 'orientation' in testcase_params: |
| self.keysight_chamber.move_theta_phi_abs( |
| self.keysight_chamber.preset_orientations[ |
| testcase_params['orientation']]['theta'], |
| self.keysight_chamber.preset_orientations[ |
| testcase_params['orientation']]['phi']) |
| |
| # Setup tester and wait for DUT to connect |
| self.setup_tester(testcase_params) |
| |
| # Run throughput test loop |
| stop_counter = 0 |
| if testcase_params['endc_combo_config']['nr_cell_count']: |
| self.keysight_test_app.select_display_tab('NR5G', 1, 'BTHR', |
| 'OTAGRAPH') |
| else: |
| self.keysight_test_app.select_display_tab('LTE', 1, 'BTHR', |
| 'OTAGRAPH') |
| for power_idx in range(len(testcase_params['nr_target_power_sweep'])): |
| result = collections.OrderedDict() |
| # Check that cells are still connected |
| connected = 1 |
| for cell in testcase_params['endc_combo_config']['cell_list']: |
| if not self.keysight_test_app.wait_for_cell_status( |
| cell['cell_type'], cell['cell_number'], |
| ['ACT', 'CONN'], SHORT_SLEEP, SHORT_SLEEP): |
| connected = 0 |
| if not connected: |
| self.log.info('DUT lost connection to cells. Ending test.') |
| break |
| # Set DL cell power |
| current_target_power = testcase_params['nr_target_power_sweep'][ |
| power_idx] |
| self.log.info( |
| 'Setting target power to {}dBm'.format(current_target_power)) |
| for cell_idx, cell in enumerate( |
| testcase_params['endc_combo_config']['cell_list']): |
| self.keysight_test_app.set_cell_ul_power_control( |
| cell['cell_type'], cell['cell_number'], 'TARget', |
| current_target_power) |
| # Start BLER and throughput measurements |
| current_throughput = self.run_single_throughput_measurement( |
| testcase_params) |
| result['throughput_measurements'] = current_throughput |
| result['nr_target_power'] = current_target_power |
| self.print_throughput_result(current_throughput) |
| |
| tx_power = self.dut_utils.get_fr2_tx_power() |
| |
| testcase_results['results'].append(result) |
| if (('lte_bler_result' in result['throughput_measurements'] |
| and result['throughput_measurements']['lte_bler_result'] |
| ['total']['DL']['nack_ratio'] * 100 > 99) |
| or ('nr_bler_result' in result['throughput_measurements'] |
| and result['throughput_measurements']['nr_bler_result'] |
| ['total']['DL']['nack_ratio'] * 100 > 99)): |
| stop_counter = stop_counter + 1 |
| else: |
| stop_counter = 0 |
| if stop_counter == STOP_COUNTER_LIMIT: |
| break |
| |
| # Save results |
| self.testclass_results[self.current_test_name] = testcase_results |
| |
| def generate_test_cases(self, bands, channels, nr_mcs_pair_list, |
| num_dl_cells_list, num_ul_cells_list, |
| orientation_list, dl_mimo_config, ul_mimo_config, |
| **kwargs): |
| """Function that auto-generates test cases for a test class.""" |
| test_cases = [] |
| for orientation, band, channel, num_ul_cells, num_dl_cells, nr_mcs_pair in itertools.product( |
| orientation_list, 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 = { |
| 'nr_band': band, |
| 'nr_bandwidth': 'BW100', |
| 'nr_duplex_mode': 'TDD', |
| 'nr_cell_type': 'NSA', |
| 'nr_channel': cputils.PCC_PRESET_MAPPING[band][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_ul_power_sweep_{}_{}_{}_DL_{}CC_mcs{}_{}x{}_UL_{}CC_mcs{}_{}x{}'.format( |
| orientation, 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], |
| orientation=orientation, |
| **kwargs) |
| setattr( |
| self, test_name, |
| partial(self._test_throughput_bler_sweep_ul_power, |
| test_params)) |
| test_cases.append(test_name) |
| return test_cases |
| |
| |
| class CellularFr2CpOfdmUplinkPowerSweepTest(CellularFr2UplinkPowerSweepTest): |
| |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| self.testclass_params = self.user_params[ |
| 'fr2_uplink_power_sweep_test_params'] |
| self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], |
| ['low', 'mid', 'high'], |
| [(4, 16), (4, 25), (4, 27), |
| (4, 28)], [1], [1], |
| ['A_Plane', 'B_Plane'], |
| force_contiguous_nr_channel=True, |
| dl_mimo_config=2, |
| ul_mimo_config=2, |
| schedule_scenario="FULL_TPUT", |
| schedule_slot_ratio=80, |
| traffic_direction='UL', |
| transform_precoding=0, |
| lte_dl_mcs=4, |
| lte_dl_mcs_table='QAM64', |
| 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, 25), |
| (4, 27), |
| (4, 28)], [2], |
| [2], ['A_Plane', 'B_Plane'], |
| force_contiguous_nr_channel=True, |
| dl_mimo_config=2, |
| ul_mimo_config=2, |
| schedule_scenario="FULL_TPUT", |
| schedule_slot_ratio=80, |
| traffic_direction='UL', |
| transform_precoding=0, |
| lte_dl_mcs=4, |
| lte_dl_mcs_table='QAM64', |
| 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, 25), |
| (4, 27), |
| (4, 28)], [4], |
| [4], ['A_Plane', 'B_Plane'], |
| force_contiguous_nr_channel=True, |
| dl_mimo_config=2, |
| ul_mimo_config=2, |
| schedule_scenario="FULL_TPUT", |
| schedule_slot_ratio=80, |
| traffic_direction='UL', |
| transform_precoding=0, |
| lte_dl_mcs=4, |
| lte_dl_mcs_table='QAM64', |
| lte_ul_mcs=4, |
| lte_ul_mcs_table='QAM64')) |
| |
| |
| class CellularFr2DftsOfdmUplinkPowerSweepTest(CellularFr2UplinkPowerSweepTest): |
| |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| self.testclass_params = self.user_params[ |
| 'fr2_uplink_power_sweep_test_params'] |
| self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'], |
| ['low', 'mid', 'high'], |
| [(4, 16), (4, 25), (4, 27), |
| (4, 28)], [1], [1], |
| ['A_Plane', 'B_Plane'], |
| force_contiguous_nr_channel=True, |
| dl_mimo_config=2, |
| ul_mimo_config=1, |
| schedule_scenario="FULL_TPUT", |
| schedule_slot_ratio=80, |
| traffic_direction='UL', |
| transform_precoding=1, |
| lte_dl_mcs=4, |
| lte_dl_mcs_table='QAM64', |
| 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, 25), |
| (4, 27), |
| (4, 28)], [2], |
| [2], ['A_Plane', 'B_Plane'], |
| force_contiguous_nr_channel=True, |
| dl_mimo_config=2, |
| ul_mimo_config=2, |
| schedule_scenario="FULL_TPUT", |
| schedule_slot_ratio=80, |
| traffic_direction='UL', |
| transform_precoding=1, |
| lte_dl_mcs=4, |
| lte_dl_mcs_table='QAM64', |
| 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, 25), |
| (4, 27), |
| (4, 28)], [4], |
| [4], ['A_Plane', 'B_Plane'], |
| force_contiguous_nr_channel=True, |
| dl_mimo_config=2, |
| ul_mimo_config=2, |
| schedule_scenario="FULL_TPUT", |
| schedule_slot_ratio=80, |
| traffic_direction='UL', |
| transform_precoding=1, |
| lte_dl_mcs=4, |
| lte_dl_mcs_table='QAM64', |
| lte_ul_mcs=4, |
| lte_ul_mcs_table='QAM64')) |
