1#!/usr/bin/env python3.4 2# 3# Copyright 2022 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the 'License'); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an 'AS IS' BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import collections 18import csv 19import numpy 20import json 21import re 22import os 23import time 24from acts import context 25from acts import base_test 26from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 27from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils 28from acts_contrib.test_utils.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest 29from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 30from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure 31from functools import partial 32 33VERY_SHORT_SLEEP = 0.1 34SHORT_SLEEP = 1 35MEDIUM_SLEEP = 5 36LONG_SLEEP = 10 37STOP_COUNTER_LIMIT = 3 38 39 40class CellularPageDecodeTest(CellularThroughputBaseTest): 41 """Class to test ENDC sensitivity""" 42 43 def __init__(self, controllers): 44 base_test.BaseTestClass.__init__(self, controllers) 45 self.testcase_metric_logger = ( 46 BlackboxMappedMetricLogger.for_test_case()) 47 self.testclass_metric_logger = ( 48 BlackboxMappedMetricLogger.for_test_class()) 49 self.publish_testcase_metrics = True 50 self.testclass_params = self.user_params['page_decode_test_params'] 51 self.tests = self.generate_test_cases() 52 53 def _test_page_decode(self, testcase_params): 54 """Test function to run cellular throughput and BLER measurements. 55 56 The function runs BLER/throughput measurement after configuring the 57 callbox and DUT. The test supports running PHY or TCP/UDP layer traffic 58 in a variety of band/carrier/mcs/etc configurations. 59 60 Args: 61 testcase_params: dict containing test-specific parameters 62 Returns: 63 result: dict containing throughput results and meta data 64 """ 65 # Prepare results dicts 66 testcase_params = self.compile_test_params(testcase_params) 67 testcase_results = collections.OrderedDict() 68 testcase_results['testcase_params'] = testcase_params 69 testcase_results['results'] = [] 70 71 # Setup ota chamber if needed 72 if hasattr(self, 73 'keysight_chamber') and 'orientation' in testcase_params: 74 self.keysight_chamber.move_theta_phi_abs( 75 self.keysight_chamber.preset_orientations[ 76 testcase_params['orientation']]['theta'], 77 self.keysight_chamber.preset_orientations[ 78 testcase_params['orientation']]['phi']) 79 80 # Setup tester and wait for DUT to connect 81 self.setup_tester(testcase_params) 82 test_cell = testcase_params['endc_combo_config']['cell_list'][0] 83 84 # Release RRC connection 85 self.keysight_test_app.release_rrc_connection(test_cell['cell_type'], 86 test_cell['cell_number']) 87 # Set tester to ignore RACH 88 self.keysight_test_app.enable_rach(test_cell['cell_type'], 89 test_cell['cell_number'], 90 enabled=0) 91 self.keysight_test_app.enable_preamble_report(test_cell['cell_type'], 92 1) 93 stop_counter = 0 94 for power_idx in range(len(testcase_params['cell_power_sweep'][0])): 95 result = collections.OrderedDict() 96 # Set DL cell power 97 for cell_idx, cell in enumerate( 98 testcase_params['endc_combo_config']['cell_list']): 99 cell_power_array = [] 100 current_cell_power = testcase_params['cell_power_sweep'][ 101 cell_idx][power_idx] 102 cell_power_array.append(current_cell_power) 103 self.keysight_test_app.set_cell_dl_power( 104 cell['cell_type'], cell['cell_number'], current_cell_power, 105 1) 106 result['cell_power'] = cell_power_array 107 # Start BLER and throughput measurements 108 decode_counter = 0 109 for idx in range(self.testclass_params['num_measurements']): 110 # Page device 111 self.keysight_test_app.send_rrc_paging( 112 test_cell['cell_type'], test_cell['cell_number']) 113 time.sleep(MEDIUM_SLEEP) 114 # Fetch page result 115 preamble_report = self.keysight_test_app.fetch_preamble_report( 116 test_cell['cell_type'], test_cell['cell_number']) 117 self.log.info(preamble_report) 118 # If rach attempted, increment decode counter. 119 if preamble_report: 120 decode_counter = decode_counter + 1 121 lte_rx_meas = self.dut_utils.get_rx_measurements('LTE') 122 nr_rx_meas = self.dut_utils.get_rx_measurements('NR5G') 123 result[ 124 'decode_probability'] = decode_counter / self.testclass_params[ 125 'num_measurements'] 126 127 if self.testclass_params.get('log_rsrp_metrics', 1): 128 result['lte_rx_measurements'] = lte_rx_meas 129 result['nr_rx_measurements'] = nr_rx_meas 130 self.log.info('LTE Rx Measurements: {}'.format(lte_rx_meas)) 131 self.log.info('NR Rx Measurements: {}'.format(nr_rx_meas)) 132 133 testcase_results['results'].append(result) 134 if result['decode_probability'] == 0: 135 stop_counter = stop_counter + 1 136 else: 137 stop_counter = 0 138 if stop_counter == STOP_COUNTER_LIMIT: 139 break 140 self.keysight_test_app.enable_rach(test_cell['cell_type'], 141 test_cell['cell_number'], 142 enabled=1) 143 144 # Save results 145 self.testclass_results[self.current_test_name] = testcase_results 146 147 def get_per_cell_power_sweeps(self, testcase_params): 148 # get reference test 149 nr_cell_index = testcase_params['endc_combo_config']['lte_cell_count'] 150 current_band = testcase_params['endc_combo_config']['cell_list'][ 151 nr_cell_index]['band'] 152 reference_test = None 153 reference_sensitivity = None 154 for testcase_name, testcase_data in self.testclass_results.items(): 155 if testcase_data['testcase_params']['endc_combo_config'][ 156 'cell_list'][nr_cell_index]['band'] == current_band: 157 reference_test = testcase_name 158 reference_sensitivity = testcase_data['sensitivity'] 159 if reference_test and reference_sensitivity and not self.retry_flag: 160 start_atten = reference_sensitivity + self.testclass_params[ 161 'adjacent_mcs_gap'] 162 self.log.info( 163 "Reference test {} found. Sensitivity {} dBm. Starting at {} dBm" 164 .format(reference_test, reference_sensitivity, start_atten)) 165 else: 166 start_atten = self.testclass_params['nr_cell_power_start'] 167 self.log.info( 168 "Reference test not found. Starting at {} dBm".format( 169 start_atten)) 170 # get current cell power start 171 nr_cell_sweep = list( 172 numpy.arange(start_atten, 173 self.testclass_params['nr_cell_power_stop'], 174 self.testclass_params['nr_cell_power_step'])) 175 lte_sweep = [self.testclass_params['lte_cell_power'] 176 ] * len(nr_cell_sweep) 177 if nr_cell_index == 0: 178 cell_power_sweeps = [nr_cell_sweep] 179 else: 180 cell_power_sweeps = [lte_sweep, nr_cell_sweep] 181 return cell_power_sweeps 182 183 def compile_test_params(self, testcase_params): 184 """Function that completes all test params based on the test name. 185 186 Args: 187 testcase_params: dict containing test-specific parameters 188 """ 189 # Cell power sweep 190 # TODO: Make this a function to support single power and sweep modes for each cell 191 testcase_params['cell_power_sweep'] = self.get_per_cell_power_sweeps( 192 testcase_params) 193 return testcase_params 194 195 def generate_test_cases(self, **kwargs): 196 test_cases = [] 197 with open(self.testclass_params['nr_single_cell_configs'], 198 'r') as csvfile: 199 test_configs = csv.DictReader(csvfile) 200 for test_config in test_configs: 201 if int(test_config['skip_test']): 202 continue 203 endc_combo_config = cputils.generate_endc_combo_config_from_csv_row( 204 test_config) 205 test_name = 'test_fr1_{}'.format(test_config['nr_band']) 206 test_params = collections.OrderedDict( 207 endc_combo_config=endc_combo_config, 208 lte_dl_mcs_table='QAM256', 209 lte_dl_mcs=4, 210 lte_ul_mcs_table='QAM256', 211 lte_ul_mcs=4, 212 nr_dl_mcs=4, 213 nr_ul_mcs=4, 214 transform_precoding=0, 215 # schedule_scenario='FULL_TPUT', 216 # schedule_slot_ratio=80 217 **kwargs) 218 setattr(self, test_name, 219 partial(self._test_page_decode, test_params)) 220 test_cases.append(test_name) 221 return test_cases 222