Merge Android 14 QPR3 to AOSP main
Bug: 346855327
Merged-In: Ieaafba65d89342b3b45f3b7b74e04aa3861923e9
Change-Id: I606b02a2f3d33bd96563e92c69e75ef33d0fa2b1
diff --git a/acts_tests/acts_contrib/test_utils/cellular/keysight_5g_testapp.py b/acts_tests/acts_contrib/test_utils/cellular/keysight_5g_testapp.py
index f4ba74f..967cace 100644
--- a/acts_tests/acts_contrib/test_utils/cellular/keysight_5g_testapp.py
+++ b/acts_tests/acts_contrib/test_utils/cellular/keysight_5g_testapp.py
@@ -16,13 +16,16 @@
import collections
import itertools
+
import pyvisa
import time
from acts import logger
+from acts import asserts as acts_asserts
from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils
SHORT_SLEEP = 1
VERY_SHORT_SLEEP = 0.1
+MEDIUM_SLEEP = 5
SUBFRAME_DURATION = 0.001
VISA_QUERY_DELAY = 0.01
@@ -129,12 +132,20 @@
if 'No error' not in error:
self.log.warning("Command: {}. Error: {}".format(
command, error))
- self.send_cmd('*OPC?', 1)
+ self.send_cmd('*OPC?', 1, check_errors)
time.sleep(VISA_QUERY_DELAY)
except:
raise RuntimeError('Lost connection to test app.')
return None
+ def check_error(self):
+ error = self.test_app.query('SYSTem:ERRor?')
+ if 'No error' not in error:
+ self.log.warning("Error: {}".format(error))
+ return True
+ else:
+ return False
+
def import_scpi_file(self, file_name, check_last_loaded=0):
"""Function to import SCPI file specified in file_name.
@@ -438,11 +449,12 @@
or subcarrier
"""
if full_bw:
- self.send_cmd('BSE:CONFIG:{}:{}:DL:POWer:CHANnel {}'.format(
+ self.send_cmd('BSE:CONFig:{}:{}:DL:POWer:CHANnel {}'.format(
cell_type, Keysight5GTestApp._format_cells(cell), power))
else:
- self.send_cmd('BSE:CONFIG:{}:{}:DL:POWer:EPRE {}'.format(
+ self.send_cmd('BSE:CONFig:{}:{}:DL:POWer:EPRE {}'.format(
cell_type, Keysight5GTestApp._format_cells(cell), power))
+ time.sleep(VERY_SHORT_SLEEP)
self.send_cmd('BSE:CONFig:{}:APPLY'.format(cell_type))
def set_cell_ul_power_control(self, cell_type, cell, mode, target_power=0):
@@ -475,8 +487,22 @@
cell: cell/carrier number
power: expected input power
"""
- self.send_cmd('BSE:CONFIG:{}:{}:MANual:POWer {}'.format(
- cell_type, Keysight5GTestApp._format_cells(cell), power))
+ if power == "AUTO" and cell_type == "LTE":
+ self.send_cmd('BSE:CONFig:{}:{}:CONTrol:POWer:AUTO ON'.format(
+ cell_type, Keysight5GTestApp._format_cells(cell)))
+ elif cell_type == "LTE":
+ self.send_cmd('BSE:CONFig:{}:{}:CONTrol:POWer:AUTO OFF'.format(
+ cell_type, Keysight5GTestApp._format_cells(cell)))
+ self.send_cmd('BSE:CONFig:{}:{}:MANual:POWer {}'.format(
+ cell_type, Keysight5GTestApp._format_cells(cell), power))
+ if power == "AUTO" and cell_type == "NR5G":
+ self.send_cmd('BSE:CONFig:{}:UL:EIP:AUTO ON'.format(
+ cell_type))
+ elif cell_type == "NR5G":
+ self.send_cmd('BSE:CONFig:{}:UL:EIP:AUTO OFF'.format(
+ cell_type))
+ self.send_cmd('BSE:CONFig:{}:{}:MANual:POWer {}'.format(
+ cell_type, Keysight5GTestApp._format_cells(cell), power))
self.send_cmd('BSE:CONFig:{}:APPLY'.format(cell_type))
def set_cell_duplex_mode(self, cell_type, cell, duplex_mode):
@@ -688,6 +714,20 @@
'BSE:CONFig:NR5G:UL:{}:CLPControl:TARGet:POWer:ALL {}'.format(
channel, target))
+ def configure_channel_emulator(self, cell_type, cell, fading_model):
+ if cell_type == 'LTE':
+ self.send_cmd('BSE:CONFig:{}:{}:CMODel {}'.format(cell_type, Keysight5GTestApp._format_cells(cell), fading_model['channel_model']))
+ self.send_cmd('BSE:CONFig:{}:{}:CMATrix {}'.format(cell_type, Keysight5GTestApp._format_cells(cell), fading_model['correlation_matrix']))
+ self.send_cmd('BSE:CONFig:{}:{}:MDSHift {}'.format(cell_type, Keysight5GTestApp._format_cells(cell), fading_model['max_doppler']))
+ elif cell_type == 'NR5G':
+ #TODO: check that this is FR1
+ self.send_cmd('BSE:CONFig:{}:{}:FRANge1:CMODel {}'.format(cell_type, Keysight5GTestApp._format_cells(cell), fading_model['channel_model']))
+ self.send_cmd('BSE:CONFig:{}:{}:FRANge1:CMATrix {}'.format(cell_type, Keysight5GTestApp._format_cells(cell), fading_model['correlation_matrix']))
+ self.send_cmd('BSE:CONFig:{}:{}:FRANge1:MDSHift {}'.format(cell_type, Keysight5GTestApp._format_cells(cell), fading_model['max_doppler']))
+
+ def set_channel_emulator_state(self, state):
+ self.send_cmd('BSE:CONFig:FADing:ENABle {}'.format(int(state)))
+
def apply_lte_carrier_agg(self, cells):
"""Function to start LTE carrier aggregation on already configured cells"""
if self.wait_for_cell_status('LTE', 'CELL1', 'CONN', 60):
@@ -700,11 +740,16 @@
def apply_carrier_agg(self):
"""Function to start carrier aggregation on already configured cells"""
- if self.wait_for_cell_status('LTE', 'CELL1', 'CONN', 60):
- self.send_cmd(
- 'BSE:CONFig:LTE:CELL1:CAGGregation:AGGRegate:NRCC:APPly')
- else:
+ if not self.wait_for_cell_status('LTE', 'CELL1', 'CONN', 60):
raise RuntimeError('LTE must be connected to start aggregation.')
+ # Continue if LTE connected
+ self.send_cmd(
+ 'BSE:CONFig:LTE:CELL1:CAGGregation:AGGRegate:NRCC:APPly', 0, 0)
+ time.sleep(MEDIUM_SLEEP)
+ error = self.check_error()
+ if error:
+ acts_asserts.fail('Failed to apply NR carrier aggregation.')
+
def get_ip_throughput(self, cell_type):
"""Function to query IP layer throughput on LTE or NR
@@ -728,8 +773,8 @@
"""Helper function to get PHY layer throughput on single cell"""
if cell_type == 'LTE':
tput_response = self.send_cmd(
- 'BSE:MEASure:LTE:{}:BTHRoughput:{}:THRoughput:OTA:{}?'.format(
- Keysight5GTestApp._format_cells(cell), link,
+ 'BSE:MEASure:LTE:BTHRoughput:{}:THRoughput:OTA:{}?'.format(
+ link,
Keysight5GTestApp._format_cells(cell)), 1)
elif cell_type == 'NR5G':
# Tester reply format
@@ -747,7 +792,7 @@
}
return tput_result
- def get_throughput(self, cell_type, cells):
+ def get_throughput(self, cell_type, dl_cells, ul_cells):
"""Function to get PHY layer throughput on on or more cells
This function returns the throughput data on the requested cells
@@ -760,15 +805,18 @@
Returns:
tput_result: dict containing all throughput statistics in Mbps
"""
- if not isinstance(cells, list):
- cells = [cells]
+ if not isinstance(dl_cells, list):
+ dl_cells = [dl_cells]
+ if not isinstance(ul_cells, list):
+ ul_cells = [ul_cells]
tput_result = collections.OrderedDict()
- for cell in cells:
- tput_result[cell] = {
- 'DL': self._get_throughput(cell_type, 'DL', cell),
- 'UL': self._get_throughput(cell_type, 'UL', cell)
- }
+ for cell in dl_cells:
+ tput_result.setdefault(cell, {})
+ tput_result[cell]['DL'] = self._get_throughput(cell_type, 'DL', cell)
frame_count = tput_result[cell]['DL']['frame_count']
+ for cell in ul_cells:
+ tput_result.setdefault(cell, {})
+ tput_result[cell]['UL'] = self._get_throughput(cell_type, 'UL', cell)
agg_tput = {
'DL': {
'frame_count': frame_count,
@@ -841,33 +889,35 @@
self._configure_bler_measurement(cell_type, 0, length)
self._set_bler_measurement_state(cell_type, 1)
time.sleep(0.1)
- bler_check = self.get_bler_result(cell_type, cells, length, 0)
- if bler_check['total']['DL']['frame_count'] == 0:
- self.log.warning('BLER measurement did not start. Retrying')
- self.start_bler_measurement(cell_type, cells, length)
+ #bler_check = self.get_bler_result(cell_type, cells, length, 0)
+ #if bler_check['total']['DL']['frame_count'] == 0:
+ # self.log.warning('BLER measurement did not start. Retrying')
+ # self.start_bler_measurement(cell_type, cells, length)
def _get_bler(self, cell_type, link, cell):
"""Helper function to get single-cell BLER measurement results."""
if cell_type == 'LTE':
bler_response = self.send_cmd(
- 'BSE:MEASure:LTE:CELL1:BTHRoughput:{}:BLER:CELL1?'.format(
- link), 1)
+ 'BSE:MEASure:LTE:BTHRoughput:{}:BLER:{}?'.format(
+ link, Keysight5GTestApp._format_cells(cell)), 1)
+ bler_items = ['frame_count','ack_count','ack_ratio','nack_count','nack_ratio',
+ 'statDtx_count','statDtx_ratio','nackStatDtx_count','nackStatDtx_ratio',
+ 'pdschBler_count','pdschBler_ratio','any_count','any_ratio']
+ bler_result = {bler_items[x] : bler_response[x] for x in range(len(bler_response))}
elif cell_type == 'NR5G':
bler_response = self.send_cmd(
'BSE:MEASure:NR5G:BTHRoughput:{}:BLER:{}?'.format(
link, Keysight5GTestApp._format_cells(cell)), 1)
- bler_result = {
- 'frame_count': bler_response[0],
- 'ack_count': bler_response[1],
- 'ack_ratio': bler_response[2],
- 'nack_count': bler_response[3],
- 'nack_ratio': bler_response[4]
- }
+ bler_items = ['frame_count','ack_count','ack_ratio','nack_count','nack_ratio',
+ 'statDtx_count','statDtx_ratio', 'pdschBler_count','pdschBler_ratio','pdschTputRatio']
+
+ bler_result = {bler_items[x]: bler_response[x] for x in range(len(bler_response))}
return bler_result
def get_bler_result(self,
cell_type,
- cells,
+ dl_cells,
+ ul_cells,
length,
wait_for_length=1,
polling_interval=SHORT_SLEEP):
@@ -888,22 +938,24 @@
Returns:
bler_result: dict containing per-cell and aggregate BLER results
"""
-
- if not isinstance(cells, list):
- cells = [cells]
+ if not isinstance(dl_cells, list):
+ dl_cells = [dl_cells]
+ if not isinstance(ul_cells, list):
+ ul_cells = [ul_cells]
while wait_for_length:
- dl_bler = self._get_bler(cell_type, 'DL', cells[0])
+ dl_bler = self._get_bler(cell_type, 'DL', dl_cells[0])
if dl_bler['frame_count'] < length:
time.sleep(polling_interval)
else:
break
bler_result = collections.OrderedDict()
- for cell in cells:
- bler_result[cell] = {
- 'DL': self._get_bler(cell_type, 'DL', cell),
- 'UL': self._get_bler(cell_type, 'UL', cell)
- }
+ for cell in dl_cells:
+ bler_result.setdefault(cell, {})
+ bler_result[cell]['DL'] = self._get_bler(cell_type, 'DL', cell)
+ for cell in ul_cells:
+ bler_result.setdefault(cell, {})
+ bler_result[cell]['UL'] = self._get_bler(cell_type, 'UL', cell)
agg_bler = {
'DL': {
'frame_count': length,
diff --git a/acts_tests/acts_contrib/test_utils/cellular/keysight_catr_chamber.py b/acts_tests/acts_contrib/test_utils/cellular/keysight_chamber.py
similarity index 91%
rename from acts_tests/acts_contrib/test_utils/cellular/keysight_catr_chamber.py
rename to acts_tests/acts_contrib/test_utils/cellular/keysight_chamber.py
index bd7540d..079db99 100644
--- a/acts_tests/acts_contrib/test_utils/cellular/keysight_catr_chamber.py
+++ b/acts_tests/acts_contrib/test_utils/cellular/keysight_chamber.py
@@ -4,10 +4,8 @@
import pyvisa
import time
from acts import logger
-from ota_chamber import Chamber
-
-class Chamber(Chamber):
+class KeysightChamber(object):
"""Base class implementation for signal generators.
Base class provides functions whose implementation is shared by all
@@ -15,14 +13,15 @@
"""
CHAMBER_SLEEP = 10
+ VISA_LOCATION = '/opt/keysight/iolibs/libktvisa32.so'
+
def __init__(self, config):
self.config = config
self.log = logger.create_tagged_trace_logger("{}{}".format(
self.config['brand'], self.config['model']))
- self.chamber_resource = pyvisa.ResourceManager()
+ self.chamber_resource = pyvisa.ResourceManager(self.VISA_LOCATION)
self.chamber_inst = self.chamber_resource.open_resource(
- '{}::{}::{}::INSTR'.format(self.config['network_id'],
- self.config['ip_address'],
+ 'TCPIP0::{}::{}::INSTR'.format(self.config['ip_address'],
self.config['hislip_interface']))
self.chamber_inst.timeout = 200000
self.chamber_inst.write_termination = '\n'
@@ -41,6 +40,7 @@
self.log.warning(
'Reset home set to false. Assumed [0,0]. Chamber angles may not be as expected.'
)
+ self.preset_orientations = self.config['preset_orientations']
def id_check(self, config):
""" Checks Chamber ID."""
@@ -87,6 +87,10 @@
self.move_to_azim_roll(self.current_azim, theta)
def move_theta_phi_abs(self, theta, phi):
+ self.log.info("Moving to Theta={}, Phi={}".format(theta, phi))
+ self.move_to_azim_roll(phi, theta)
+
+ def move_theta_phi_abs(self, theta, phi):
self.log.info("Moving chamber to [{}, {}]".format(theta, phi))
self.move_to_azim_roll(phi, theta)
@@ -159,7 +163,6 @@
self.chamber_inst.write("POS:SWE:CONT 1")
def sweep_init(self):
-
def query_float_list(inst, scpi):
resp = inst.query(scpi)
return list(map(float, resp.split(',')))
@@ -170,12 +173,4 @@
rolls = query_float_list(self.chamber_inst, "FETC:ROLL?")
phis = query_float_list(self.chamber_inst, "FETC:DUT:PHI?")
thetas = query_float_list(self.chamber_inst, "FETC:DUT:THET?")
- return zip(azims, rolls, phis, thetas)
-
- def configure_positioner(self, pos_name, pos_visa_addr):
- select = "True"
- simulate = "False"
- options = ""
- data = f"'{pos_name}~{select}~{simulate}~{pos_visa_addr}~{options}'"
- self.chamber_inst.write(f"EQU:CONF {data}")
- self.chamber_inst.write("EQU:UPD")
+ return zip(azims, rolls, phis, thetas)
\ No newline at end of file
diff --git a/acts_tests/acts_contrib/test_utils/cellular/performance/CellularThroughputBaseTest.py b/acts_tests/acts_contrib/test_utils/cellular/performance/CellularThroughputBaseTest.py
index fd2d794..2ebcb25 100644
--- a/acts_tests/acts_contrib/test_utils/cellular/performance/CellularThroughputBaseTest.py
+++ b/acts_tests/acts_contrib/test_utils/cellular/performance/CellularThroughputBaseTest.py
@@ -32,6 +32,7 @@
from acts.controllers.android_lib.tel import tel_utils
from acts.controllers import iperf_server as ipf
from acts_contrib.test_utils.cellular.keysight_5g_testapp import Keysight5GTestApp
+from acts_contrib.test_utils.cellular.keysight_chamber import KeysightChamber
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 functools import partial
@@ -40,6 +41,7 @@
MEDIUM_SLEEP = 2
IPERF_TIMEOUT = 10
SHORT_SLEEP = 1
+VERY_SHORT_SLEEP = 0.1
SUBFRAME_LENGTH = 0.001
STOP_COUNTER_LIMIT = 3
@@ -70,6 +72,9 @@
self.dut = self.android_devices[-1]
self.keysight_test_app = Keysight5GTestApp(
self.user_params['Keysight5GTestApp'])
+ if 'KeysightChamber' in self.user_params:
+ self.keysight_chamber = KeysightChamber(
+ self.user_params['KeysightChamber'])
self.iperf_server = self.iperf_servers[0]
self.iperf_client = self.iperf_clients[0]
self.remote_server = ssh.connection.SshConnection(
@@ -245,7 +250,7 @@
testcase_params['bler_measurement_length'])
if testcase_params['endc_combo_config']['lte_cell_count']:
self.keysight_test_app.start_bler_measurement(
- 'LTE', testcase_params['endc_combo_config']['lte_carriers'][0],
+ 'LTE', testcase_params['endc_combo_config']['lte_dl_carriers'][0],
testcase_params['bler_measurement_length'])
if self.testclass_params['traffic_type'] != 'PHY':
@@ -255,64 +260,59 @@
if testcase_params['endc_combo_config']['nr_cell_count']:
result['nr_bler_result'] = self.keysight_test_app.get_bler_result(
'NR5G', testcase_params['endc_combo_config']['nr_dl_carriers'],
+ testcase_params['endc_combo_config']['nr_ul_carriers'],
testcase_params['bler_measurement_length'])
result['nr_tput_result'] = self.keysight_test_app.get_throughput(
- 'NR5G', testcase_params['endc_combo_config']['nr_dl_carriers'])
+ 'NR5G', testcase_params['endc_combo_config']['nr_dl_carriers'],
+ testcase_params['endc_combo_config']['nr_ul_carriers'])
if testcase_params['endc_combo_config']['lte_cell_count']:
result['lte_bler_result'] = self.keysight_test_app.get_bler_result(
- 'LTE', testcase_params['endc_combo_config']['lte_carriers'],
- testcase_params['bler_measurement_length'])
+ cell_type='LTE', dl_cells=testcase_params['endc_combo_config']['lte_dl_carriers'],
+ ul_cells=testcase_params['endc_combo_config']['lte_ul_carriers'],
+ length=testcase_params['bler_measurement_length'])
result['lte_tput_result'] = self.keysight_test_app.get_throughput(
- 'LTE', testcase_params['endc_combo_config']['lte_carriers'])
+ 'LTE', testcase_params['endc_combo_config']['lte_dl_carriers'],
+ testcase_params['endc_combo_config']['lte_ul_carriers'])
return result
def print_throughput_result(self, result):
# Print Test Summary
if 'nr_tput_result' in result:
+
self.log.info(
- "----NR5G STATS-------NR5G STATS-------NR5G STATS---")
- self.log.info(
- "DL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}"
+ "NR5G DL PHY Tput (Mbps) (Min/Avg/Max/Th): {:.2f} / {:.2f} / {:.2f} / {:.2f}\tBLER: {:.2f}"
.format(
result['nr_tput_result']['total']['DL']['min_tput'],
result['nr_tput_result']['total']['DL']['average_tput'],
result['nr_tput_result']['total']['DL']['max_tput'],
- result['nr_tput_result']['total']['DL']
- ['theoretical_tput']))
+ result['nr_tput_result']['total']['DL']['theoretical_tput'],
+ result['nr_bler_result']['total']['DL']['nack_ratio'] * 100))
self.log.info(
- "UL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}"
+ "NR5G UL PHY Tput (Mbps) (Min/Avg/Max/Th): {:.2f} / {:.2f} / {:.2f} / {:.2f}\tBLER: {:.2f}"
.format(
result['nr_tput_result']['total']['UL']['min_tput'],
result['nr_tput_result']['total']['UL']['average_tput'],
result['nr_tput_result']['total']['UL']['max_tput'],
- result['nr_tput_result']['total']['UL']
- ['theoretical_tput']))
- self.log.info("DL BLER: {:.2f}%\tUL BLER: {:.2f}%".format(
- result['nr_bler_result']['total']['DL']['nack_ratio'] * 100,
- result['nr_bler_result']['total']['UL']['nack_ratio'] * 100))
+ result['nr_tput_result']['total']['UL']['theoretical_tput'],
+ result['nr_bler_result']['total']['UL']['nack_ratio'] * 100))
if 'lte_tput_result' in result:
- self.log.info("----LTE STATS-------LTE STATS-------LTE STATS---")
self.log.info(
- "DL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}"
+ "LTE DL PHY Tput (Mbps) (Min/Avg/Max/Th): {:.2f} / {:.2f} / {:.2f} / {:.2f}\tBLER: {:.2f}"
.format(
result['lte_tput_result']['total']['DL']['min_tput'],
result['lte_tput_result']['total']['DL']['average_tput'],
result['lte_tput_result']['total']['DL']['max_tput'],
- result['lte_tput_result']['total']['DL']
- ['theoretical_tput']))
+ result['lte_tput_result']['total']['DL']['theoretical_tput'],
+ result['lte_bler_result']['total']['DL']['nack_ratio'] * 100))
if self.testclass_params['lte_ul_mac_padding']:
self.log.info(
- "UL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}"
+ "LTE UL PHY Tput (Mbps) (Min/Avg/Max/Th): {:.2f} / {:.2f} / {:.2f} / {:.2f}\tBLER: {:.2f}"
.format(
result['lte_tput_result']['total']['UL']['min_tput'],
- result['lte_tput_result']['total']['UL']
- ['average_tput'],
+ result['lte_tput_result']['total']['UL']['average_tput'],
result['lte_tput_result']['total']['UL']['max_tput'],
- result['lte_tput_result']['total']['UL']
- ['theoretical_tput']))
- self.log.info("DL BLER: {:.2f}%\tUL BLER: {:.2f}%".format(
- result['lte_bler_result']['total']['DL']['nack_ratio'] * 100,
- result['lte_bler_result']['total']['UL']['nack_ratio'] * 100))
+ result['lte_tput_result']['total']['UL']['theoretical_tput'],
+ result['lte_bler_result']['total']['UL']['nack_ratio'] * 100))
if self.testclass_params['traffic_type'] != 'PHY':
self.log.info("{} Tput: {:.2f} Mbps".format(
self.testclass_params['traffic_type'],
@@ -337,15 +337,18 @@
self.keysight_test_app.set_cell_input_power(
cell['cell_type'], cell['cell_number'],
self.testclass_params['input_power'][cell['cell_type']])
- self.keysight_test_app.set_cell_ul_power_control(
- cell['cell_type'], cell['cell_number'],
- self.testclass_params['ul_power_control_mode'],
- self.testclass_params.get('ul_power_control_target',0)
- )
+ if cell['cell_type'] == 'LTE' and cell['pcc'] == 0:
+ pass
+ else:
+ self.keysight_test_app.set_cell_ul_power_control(
+ cell['cell_type'], cell['cell_number'],
+ self.testclass_params['ul_power_control_mode'],
+ self.testclass_params.get('ul_power_control_target',0)
+ )
if cell['cell_type'] == 'NR5G':
self.keysight_test_app.set_nr_subcarrier_spacing(
cell['cell_number'], cell['subcarrier_spacing'])
- if 'channel' in cell:
+ if 'channel' in cell and cell['channel'] is not None:
self.keysight_test_app.set_cell_channel(
cell['cell_type'], cell['cell_number'], cell['channel'])
self.keysight_test_app.set_cell_bandwidth(cell['cell_type'],
@@ -363,6 +366,10 @@
self.keysight_test_app.set_cell_mimo_config(
cell['cell_type'], cell['cell_number'], 'UL',
cell['ul_mimo_config'])
+ if 'fading_scenario' in self.testclass_params:
+ self.keysight_test_app.configure_channel_emulator(
+ cell['cell_type'], cell['cell_number'],
+ self.testclass_params['fading_scenario'][cell['cell_type']])
if testcase_params.get('force_contiguous_nr_channel', False):
self.keysight_test_app.toggle_contiguous_nr_channels(1)
@@ -377,7 +384,6 @@
self.testclass_params['lte_ul_mac_padding'])
if testcase_params['endc_combo_config']['nr_cell_count']:
-
if 'schedule_scenario' in testcase_params:
self.keysight_test_app.set_nr_cell_schedule_scenario(
'CELL1',
@@ -406,18 +412,12 @@
cell['cell_number']))
self.keysight_test_app.set_cell_state(cell['cell_type'],
cell['cell_number'], 1)
- # Activate LTE aggregation if applicable
- if testcase_params['endc_combo_config']['lte_scc_list']:
- self.keysight_test_app.apply_lte_carrier_agg(
- testcase_params['endc_combo_config']['lte_scc_list'])
self.log.info('Waiting for LTE connections')
# Turn airplane mode off
num_apm_toggles = 10
for idx in range(num_apm_toggles):
self.log.info('Turning off airplane mode')
- #asserts.assert_true(utils.force_airplane_mode(self.dut, False),
- # 'Can not turn off airplane mode.')
- tel_utils.toggle_airplane_mode(self.log, self.dut, False)
+ cputils.toggle_airplane_mode(self.log, self.dut, False, False, idx)
if self.keysight_test_app.wait_for_cell_status(
'LTE', 'CELL1', 'CONN', 10*(idx+1)):
self.log.info('Connected! Waiting for {} seconds.'.format(LONG_SLEEP))
@@ -425,12 +425,14 @@
break
elif idx < num_apm_toggles - 1:
self.log.info('Turning on airplane mode')
- # asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- # 'Can not turn on airplane mode.')
- tel_utils.toggle_airplane_mode(self.log, self.dut, True)
+ cputils.toggle_airplane_mode(self.log, self.dut, True, False, idx)
time.sleep(MEDIUM_SLEEP)
else:
asserts.fail('DUT did not connect to LTE.')
+ # Activate LTE aggregation if applicable
+ if testcase_params['endc_combo_config']['lte_scc_list']:
+ self.keysight_test_app.apply_lte_carrier_agg(
+ testcase_params['endc_combo_config']['lte_scc_list'])
if testcase_params['endc_combo_config']['nr_cell_count']:
self.keysight_test_app.apply_carrier_agg()
@@ -454,9 +456,7 @@
num_apm_toggles = 10
for idx in range(num_apm_toggles):
self.log.info('Turning off airplane mode now.')
- #asserts.assert_true(utils.force_airplane_mode(self.dut, False),
- # 'Can not turn off airplane mode.')
- tel_utils.toggle_airplane_mode(self.log, self.dut, False)
+ cputils.toggle_airplane_mode(self.log, self.dut, False, False, idx)
if self.keysight_test_app.wait_for_cell_status(
'NR5G', 'CELL1', 'CONN', 10*(idx+1)):
self.log.info('Connected! Waiting for {} seconds.'.format(LONG_SLEEP))
@@ -464,13 +464,17 @@
break
elif idx < num_apm_toggles - 1:
self.log.info('Turning on airplane mode now.')
- # asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- # 'Can not turn on airplane mode.')
- tel_utils.toggle_airplane_mode(self.log, self.dut, True)
+ cputils.toggle_airplane_mode(self.log, self.dut, True, False, idx)
time.sleep(MEDIUM_SLEEP)
else:
asserts.fail('DUT did not connect to NR.')
+ if 'fading_scenario' in self.testclass_params and self.testclass_params['fading_scenario']['enable']:
+ self.log.info('Enabling fading.')
+ self.keysight_test_app.set_channel_emulator_state(self.testclass_params['fading_scenario']['enable'])
+ else:
+ self.keysight_test_app.set_channel_emulator_state(0)
+
def _test_throughput_bler(self, testcase_params):
"""Test function to run cellular throughput and BLER measurements.
@@ -489,6 +493,12 @@
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)
@@ -502,6 +512,16 @@
'OTAGRAPH')
for power_idx in range(len(testcase_params['cell_power_sweep'][0])):
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'], VERY_SHORT_SLEEP,VERY_SHORT_SLEEP):
+ connected = 0
+ if not connected:
+ self.log.info('DUT lost connection to cells. Ending test.')
+ break
# Set DL cell power
for cell_idx, cell in enumerate(
testcase_params['endc_combo_config']['cell_list']):
diff --git a/acts_tests/acts_contrib/test_utils/cellular/performance/cellular_performance_test_utils.py b/acts_tests/acts_contrib/test_utils/cellular/performance/cellular_performance_test_utils.py
index 2a72590..0132693 100644
--- a/acts_tests/acts_contrib/test_utils/cellular/performance/cellular_performance_test_utils.py
+++ b/acts_tests/acts_contrib/test_utils/cellular/performance/cellular_performance_test_utils.py
@@ -19,6 +19,8 @@
import os
import re
import time
+from queue import Empty
+from acts.controllers.android_lib.tel import tel_utils
PCC_PRESET_MAPPING = {
'N257': {
@@ -66,6 +68,7 @@
},
}
+LONG_SLEEP = 10
def extract_test_id(testcase_params, id_fields):
test_id = collections.OrderedDict(
@@ -249,3 +252,293 @@
sinr_values = [float(x) for x in re.findall(sinr_regex, rx_meas)]
return {'rsrp': rsrp_values, 'rsrq': rsrq_values, 'rssi': rssi_values, 'sinr': sinr_values}
+def toggle_airplane_mode(log, ad, new_state=None, strict_checking=True, try_index=0):
+ """ Toggle the state of airplane mode.
+
+ Args:
+ log: log handler.
+ ad: android_device object.
+ new_state: Airplane mode state to set to.
+ If None, opposite of the current state.
+ strict_checking: Whether to turn on strict checking that checks all features.
+ try_index: index of apm toggle
+
+ Returns:
+ result: True if operation succeed. False if error happens.
+ """
+ if try_index % 2 == 0:
+ log.info('Toggling airplane mode {} by adb.'.format(new_state))
+ return tel_utils.toggle_airplane_mode_by_adb(log, ad, new_state)
+ else:
+ log.info('Toggling airplane mode {} by msim.'.format(new_state))
+ return toggle_airplane_mode_msim(
+ log, ad, new_state, strict_checking=strict_checking)
+
+def toggle_airplane_mode_msim(log, ad, new_state=None, strict_checking=True):
+ """ Toggle the state of airplane mode.
+
+ Args:
+ log: log handler.
+ ad: android_device object.
+ new_state: Airplane mode state to set to.
+ If None, opposite of the current state.
+ strict_checking: Whether to turn on strict checking that checks all features.
+
+ Returns:
+ result: True if operation succeed. False if error happens.
+ """
+
+ cur_state = ad.droid.connectivityCheckAirplaneMode()
+ if cur_state == new_state:
+ ad.log.info("Airplane mode already in %s", new_state)
+ return True
+ elif new_state is None:
+ new_state = not cur_state
+ ad.log.info("Toggle APM mode, from current tate %s to %s", cur_state,
+ new_state)
+ sub_id_list = []
+ active_sub_info = ad.droid.subscriptionGetAllSubInfoList()
+ if active_sub_info:
+ for info in active_sub_info:
+ sub_id_list.append(info['subscriptionId'])
+
+ ad.ed.clear_all_events()
+ time.sleep(0.1)
+ service_state_list = []
+ if new_state:
+ service_state_list.append(tel_utils.SERVICE_STATE_POWER_OFF)
+ ad.log.info("Turn on airplane mode")
+
+ else:
+ # If either one of these 3 events show up, it should be OK.
+ # Normal SIM, phone in service
+ service_state_list.append(tel_utils.SERVICE_STATE_IN_SERVICE)
+ # NO SIM, or Dead SIM, or no Roaming coverage.
+ service_state_list.append(tel_utils.SERVICE_STATE_OUT_OF_SERVICE)
+ service_state_list.append(tel_utils.SERVICE_STATE_EMERGENCY_ONLY)
+ ad.log.info("Turn off airplane mode")
+
+ for sub_id in sub_id_list:
+ ad.droid.telephonyStartTrackingServiceStateChangeForSubscription(
+ sub_id)
+
+ timeout_time = time.time() + LONG_SLEEP
+ ad.droid.connectivityToggleAirplaneMode(new_state)
+
+ try:
+ try:
+ event = ad.ed.wait_for_event(
+ tel_utils.EVENT_SERVICE_STATE_CHANGED,
+ tel_utils.is_event_match_for_list,
+ timeout= LONG_SLEEP,
+ field=tel_utils.ServiceStateContainer.SERVICE_STATE,
+ value_list=service_state_list)
+ ad.log.info("Got event %s", event)
+ except Empty:
+ ad.log.warning("Did not get expected service state change to %s",
+ service_state_list)
+ finally:
+ for sub_id in sub_id_list:
+ ad.droid.telephonyStopTrackingServiceStateChangeForSubscription(
+ sub_id)
+ except Exception as e:
+ ad.log.error(e)
+
+ # APM on (new_state=True) will turn off bluetooth but may not turn it on
+ try:
+ if new_state and not tel_utils._wait_for_bluetooth_in_state(
+ log, ad, False, timeout_time - time.time()):
+ ad.log.error(
+ "Failed waiting for bluetooth during airplane mode toggle")
+ if strict_checking: return False
+ except Exception as e:
+ ad.log.error("Failed to check bluetooth state due to %s", e)
+ if strict_checking:
+ raise
+
+ # APM on (new_state=True) will turn off wifi but may not turn it on
+ if new_state and not tel_utils._wait_for_wifi_in_state(log, ad, False,
+ timeout_time - time.time()):
+ ad.log.error("Failed waiting for wifi during airplane mode toggle on")
+ if strict_checking: return False
+
+ if ad.droid.connectivityCheckAirplaneMode() != new_state:
+ ad.log.error("Set airplane mode to %s failed", new_state)
+ return False
+ return True
+
+def generate_endc_combo_config_from_string(endc_combo_str):
+ """Function to generate ENDC combo config from combo string
+
+ Args:
+ endc_combo_str: ENDC combo descriptor (e.g. B48A[4];A[1]+N5A[2];A[1])
+ Returns:
+ endc_combo_config: dictionary with all ENDC combo settings
+ """
+ endc_combo_config = collections.OrderedDict()
+ endc_combo_config['endc_combo_name']=endc_combo_str
+ endc_combo_str = endc_combo_str.replace(' ', '')
+ endc_combo_list = endc_combo_str.split('+')
+ cell_config_list = list()
+ lte_cell_count = 0
+ nr_cell_count = 0
+ lte_scc_list = []
+ nr_dl_carriers = []
+ nr_ul_carriers = []
+ lte_dl_carriers = []
+ lte_ul_carriers = []
+
+ cell_config_regex = re.compile(
+ r'(?P<cell_type>[B,N])(?P<band>[0-9]+)(?P<bandwidth_class>[A-Z])\[bw=(?P<dl_bandwidth>[0-9]+)\]'
+ r'(\[ch=)?(?P<channel>[0-9]+)?\]?\[ant=(?P<dl_mimo_config>[0-9]+),?(?P<transmission_mode>[TM0-9]+)?\];?'
+ r'(?P<ul_bandwidth_class>[A-Z])?(\[ant=)?(?P<ul_mimo_config>[0-9])?(\])?'
+ )
+ for cell_string in endc_combo_list:
+ cell_config = re.match(cell_config_regex, cell_string).groupdict()
+ if cell_config['cell_type'] == 'B':
+ # Configure LTE specific parameters
+ cell_config['cell_type'] = 'LTE'
+ lte_cell_count = lte_cell_count + 1
+ cell_config['cell_number'] = lte_cell_count
+ if cell_config['cell_number'] == 1:
+ cell_config['pcc'] = 1
+ endc_combo_config['lte_pcc'] = cell_config['cell_number']
+ else:
+ cell_config['pcc'] = 0
+ lte_scc_list.append(cell_config['cell_number'])
+ cell_config['duplex_mode'] = 'FDD' if int(
+ cell_config['band']
+ ) in DUPLEX_MODE_TO_BAND_MAPPING['LTE'][
+ 'FDD'] else 'TDD'
+ cell_config['dl_mimo_config'] = 'D{nss}U{nss}'.format(
+ nss=cell_config['dl_mimo_config'])
+ lte_dl_carriers.append(cell_config['cell_number'])
+ else:
+ # Configure NR specific parameters
+ cell_config['cell_type'] = 'NR5G'
+ nr_cell_count = nr_cell_count + 1
+ cell_config['cell_number'] = nr_cell_count
+ nr_dl_carriers.append(cell_config['cell_number'])
+ #TODO: fix NSA/SA indicator
+ cell_config['nr_cell_type'] = 'NSA'
+ cell_config['band'] = 'N' + cell_config['band']
+ cell_config['duplex_mode'] = 'FDD' if cell_config[
+ 'band'] in DUPLEX_MODE_TO_BAND_MAPPING['NR5G'][
+ 'FDD'] else 'TDD'
+ cell_config['subcarrier_spacing'] = 'MU0' if cell_config[
+ 'duplex_mode'] == 'FDD' else 'MU1'
+ cell_config['dl_mimo_config'] = 'N{nss}X{nss}'.format(
+ nss=cell_config['dl_mimo_config'])
+
+ cell_config['dl_bandwidth_class'] = cell_config['bandwidth_class']
+ cell_config['dl_bandwidth'] = 'BW'+ cell_config['dl_bandwidth']
+ cell_config['ul_enabled'] = 1 if cell_config['ul_bandwidth_class'] else 0
+ if cell_config['ul_enabled']:
+ cell_config['ul_mimo_config'] = 'N{nss}X{nss}'.format(
+ nss=cell_config['ul_mimo_config'])
+ if cell_config['cell_type'] == 'LTE':
+ lte_ul_carriers.append(cell_config['cell_number'])
+ elif cell_config['cell_type'] == 'NR5G':
+ nr_ul_carriers.append(cell_config['cell_number'])
+ cell_config_list.append(cell_config)
+ 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_dl_carriers
+ endc_combo_config['lte_ul_carriers'] = lte_ul_carriers
+ return endc_combo_config
+
+def generate_endc_combo_config_from_csv_row(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()
+ lte_cell_count = 0
+ nr_cell_count = 0
+ lte_scc_list = []
+ nr_dl_carriers = []
+ nr_ul_carriers = []
+ lte_dl_carriers = []
+ lte_ul_carriers = []
+
+ cell_config_list = []
+ if test_config['lte_band']:
+ 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'])
+ }
+ if int(test_config['lte_dl_mimo_config']) == 1:
+ lte_cell['transmission_mode'] = 'TM1'
+ elif int(test_config['lte_dl_mimo_config']) == 2:
+ lte_cell['transmission_mode'] = 'TM2'
+ else:
+ lte_cell['transmission_mode'] = 'TM3'
+ cell_config_list.append(lte_cell)
+ endc_combo_config['lte_pcc'] = 1
+ lte_cell_count = 1
+ lte_dl_carriers = [1]
+ lte_ul_carriers = [1]
+
+ if test_config['nr_band']:
+ nr_cell = {
+ 'cell_type':
+ 'NR5G',
+ 'cell_number':
+ 1,
+ 'band':
+ test_config['nr_band'],
+ 'nr_cell_type': test_config['nr_cell_type'],
+ 'duplex_mode':
+ test_config['nr_duplex_mode'],
+ '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,
+ 'ul_bandwidth_class':
+ 'A',
+ 'ul_mimo_config':
+ 'N{nss}X{nss}'.format(nss=test_config['nr_ul_mimo_config']),
+ 'subcarrier_spacing':
+ 'MU0' if test_config['nr_scs'] == '15' else 'MU1'
+ }
+ cell_config_list.append(nr_cell)
+ nr_cell_count = 1
+ nr_dl_carriers = [1]
+ nr_ul_carriers = [1]
+
+ 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_dl_carriers
+ endc_combo_config['lte_ul_carriers'] = lte_ul_carriers
+ return endc_combo_config
diff --git a/acts_tests/acts_contrib/test_utils/wifi/wifi_performance_test_utils/brcm_utils.py b/acts_tests/acts_contrib/test_utils/wifi/wifi_performance_test_utils/brcm_utils.py
index d944db5..b9e6d28 100644
--- a/acts_tests/acts_contrib/test_utils/wifi/wifi_performance_test_utils/brcm_utils.py
+++ b/acts_tests/acts_contrib/test_utils/wifi/wifi_performance_test_utils/brcm_utils.py
@@ -269,7 +269,7 @@
for key, val in connected_rssi.copy().items():
if 'data' not in val:
continue
- filtered_rssi_values = [x for x in val['data'] if not math.isnan(x)]
+ filtered_rssi_values = [x for x in val['data'] if not (math.isnan(x) or math.isinf(x))]
if len(filtered_rssi_values) > ignore_samples:
filtered_rssi_values = filtered_rssi_values[ignore_samples:]
if filtered_rssi_values:
diff --git a/acts_tests/tests/google/cellular/performance/Cellular5GFR2SensitivityTest.py b/acts_tests/tests/google/cellular/performance/Cellular5GFR2SensitivityTest.py
deleted file mode 100644
index f3e49b8..0000000
--- a/acts_tests/tests/google/cellular/performance/Cellular5GFR2SensitivityTest.py
+++ /dev/null
@@ -1,235 +0,0 @@
-#!/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 itertools
-import json
-import numpy
-import os
-from functools import partial
-from acts import asserts
-from acts import context
-from acts import base_test
-from acts import utils
-from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger
-from acts.controllers.utils_lib import ssh
-from acts_contrib.test_utils.cellular.keysight_5g_testapp import Keysight5GTestApp
-from acts_contrib.test_utils.cellular.performance import cellular_performance_test_utils as cputils
-from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure
-from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils
-from Cellular5GFR2ThroughputTest import Cellular5GFR2ThroughputTest
-
-
-class Cellular5GFR2SensitivityTest(Cellular5GFR2ThroughputTest):
- """Class to test cellular throughput
-
- This class implements cellular throughput tests on a lab/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):
- 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 setup_class(self):
- """Initializes common test hardware and parameters.
-
- This function initializes hardwares and compiles parameters that are
- common to all tests in this class.
- """
- self.dut = self.android_devices[-1]
- self.testclass_params = self.user_params['sensitivity_test_params']
- self.keysight_test_app = Keysight5GTestApp(
- self.user_params['Keysight5GTestApp'])
- self.testclass_results = collections.OrderedDict()
- self.iperf_server = self.iperf_servers[0]
- self.iperf_client = self.iperf_clients[0]
- self.remote_server = ssh.connection.SshConnection(
- ssh.settings.from_config(
- self.user_params['RemoteServer']['ssh_config']))
- if self.testclass_params.get('reload_scpi', 1):
- self.keysight_test_app.import_scpi_file(
- self.testclass_params['scpi_file'])
- # Configure test retries
- self.user_params['retry_tests'] = [self.__class__.__name__]
-
- # Turn Airplane mode on
- asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- 'Can not turn on airplane mode.')
-
- def process_testcase_results(self):
- if self.current_test_name not in self.testclass_results:
- return
- testcase_results = self.testclass_results[self.current_test_name]
- cell_power_list = [
- result['cell_power'] for result in testcase_results['results']
- ]
- dl_bler_list = [
- result['bler_result']['total']['DL']['nack_ratio']
- for result in testcase_results['results']
- ]
- bler_above_threshold = [
- x > self.testclass_params['bler_threshold'] for x in dl_bler_list
- ]
- for idx in range(len(bler_above_threshold)):
- if all(bler_above_threshold[idx:]):
- sensitivity_index = max(idx, 1) - 1
- cell_power_at_sensitivity = cell_power_list[sensitivity_index]
- break
- else:
- sensitivity_index = -1
- cell_power_at_sensitivity = float('nan')
- if min(dl_bler_list) < 0.05:
- testcase_results['sensitivity'] = cell_power_at_sensitivity
- else:
- testcase_results['sensitivity'] = float('nan')
-
- testcase_results['cell_power_list'] = cell_power_list
- testcase_results['dl_bler_list'] = dl_bler_list
-
- 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_results),
- results_file,
- indent=4)
-
- result_string = ('DL {}CC MCS {} Sensitivity = {}dBm.'.format(
- testcase_results['testcase_params']['num_dl_cells'],
- testcase_results['testcase_params']['dl_mcs'],
- testcase_results['sensitivity']))
- if min(dl_bler_list) < 0.05:
- self.log.info('Test Passed. {}'.format(result_string))
- else:
- self.log.info('Result unreliable. {}'.format(result_string))
-
- def process_testclass_results(self):
- Cellular5GFR2ThroughputTest.process_testclass_results(self)
-
- plots = collections.OrderedDict()
- id_fields = ['band', 'num_dl_cells']
- for testcase, testcase_data in self.testclass_results.items():
- testcase_params = testcase_data['testcase_params']
- plot_id = cputils.extract_test_id(testcase_params, id_fields)
- plot_id = tuple(plot_id.items())
- if plot_id not in plots:
- plots[plot_id] = BokehFigure(title='{} {}CC'.format(
- testcase_params['band'], testcase_params['num_dl_cells']),
- x_label='Cell Power (dBm)',
- primary_y_label='BLER (%)')
- plots[plot_id].add_line(
- testcase_data['cell_power_list'],
- testcase_data['dl_bler_list'],
- 'Channel {}, MCS {}'.format(testcase_params['channel'],
- testcase_params['dl_mcs']))
- 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)
-
- def generate_test_cases(self, bands, channels, mcs_pair_list,
- num_dl_cells_list, num_ul_cells_list, **kwargs):
- """Function that auto-generates test cases for a test class."""
- test_cases = []
-
- for band, channel, num_ul_cells, num_dl_cells, mcs_pair in itertools.product(
- bands, channels, num_ul_cells_list, num_dl_cells_list,
- mcs_pair_list):
- if num_ul_cells > num_dl_cells:
- continue
- test_name = 'test_nr_sensitivity_{}_{}_DL_{}CC_mcs{}'.format(
- band, channel, num_dl_cells, mcs_pair[0])
- test_params = collections.OrderedDict(
- band=band,
- channel=channel,
- dl_mcs=mcs_pair[0],
- ul_mcs=mcs_pair[1],
- num_dl_cells=num_dl_cells,
- num_ul_cells=num_ul_cells,
- dl_cell_list=list(range(1, num_dl_cells + 1)),
- ul_cell_list=list(range(1, num_ul_cells + 1)),
- **kwargs)
- setattr(self, test_name,
- partial(self._test_nr_throughput_bler, test_params))
- test_cases.append(test_name)
- return test_cases
-
-
-class Cellular5GFR2_AllBands_SensitivityTest(Cellular5GFR2SensitivityTest):
-
- def __init__(self, controllers):
- super().__init__(controllers)
- self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(16, 4), (27, 4)],
- list(range(1, 9)), [1],
- schedule_scenario="FULL_TPUT",
- traffic_direction='DL',
- transform_precoding=0)
-
-
-class Cellular5GFR2_FrequencySweep_SensitivityTest(Cellular5GFR2SensitivityTest
- ):
-
- def __init__(self, controllers):
- super().__init__(controllers)
- frequency_sweep_params = self.user_params['sensitivity_test_params'][
- 'frequency_sweep']
- self.tests = self.generate_test_cases(frequency_sweep_params,
- [(16, 4), (27, 4)],
- schedule_scenario="FULL_TPUT",
- traffic_direction='DL',
- transform_precoding=0)
-
- def generate_test_cases(self, dl_frequency_sweep_params, mcs_pair_list,
- **kwargs):
- """Function that auto-generates test cases for a test class."""
- test_cases = ['test_load_scpi']
-
- for band, band_config in dl_frequency_sweep_params.items():
- for num_dl_cells_str, sweep_config in band_config.items():
- num_dl_cells = int(num_dl_cells_str[0])
- num_ul_cells = 1
- freq_vector = numpy.arange(sweep_config[0], sweep_config[1],
- sweep_config[2])
- for freq in freq_vector:
- for mcs_pair in mcs_pair_list:
- test_name = 'test_nr_sensitivity_{}_{}_DL_{}CC_mcs{}'.format(
- band, freq, num_dl_cells, mcs_pair[0])
- test_params = collections.OrderedDict(
- band=band,
- channel=freq,
- dl_mcs=mcs_pair[0],
- ul_mcs=mcs_pair[1],
- num_dl_cells=num_dl_cells,
- num_ul_cells=num_ul_cells,
- dl_cell_list=list(range(1, num_dl_cells + 1)),
- ul_cell_list=list(range(1, num_ul_cells + 1)),
- **kwargs)
- setattr(
- self, test_name,
- partial(self._test_nr_throughput_bler,
- test_params))
- test_cases.append(test_name)
- return test_cases
diff --git a/acts_tests/tests/google/cellular/performance/Cellular5GFR2ThroughputTest.py b/acts_tests/tests/google/cellular/performance/Cellular5GFR2ThroughputTest.py
deleted file mode 100644
index 9e848a3..0000000
--- a/acts_tests/tests/google/cellular/performance/Cellular5GFR2ThroughputTest.py
+++ /dev/null
@@ -1,664 +0,0 @@
-#!/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 numpy
-import os
-import time
-from acts import asserts
-from acts import context
-from acts import base_test
-from acts import utils
-from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger
-from acts.controllers.utils_lib import ssh
-from acts.controllers import iperf_server as ipf
-from acts_contrib.test_utils.cellular.keysight_5g_testapp import Keysight5GTestApp
-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 functools import partial
-
-LONG_SLEEP = 10
-MEDIUM_SLEEP = 2
-IPERF_TIMEOUT = 10
-SHORT_SLEEP = 1
-SUBFRAME_LENGTH = 0.001
-STOP_COUNTER_LIMIT = 3
-
-
-class Cellular5GFR2ThroughputTest(base_test.BaseTestClass):
- """Class to test cellular throughput
-
- This class implements cellular throughput tests on a lab/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):
- 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 setup_class(self):
- """Initializes common test hardware and parameters.
-
- This function initializes hardwares and compiles parameters that are
- common to all tests in this class.
- """
- self.dut = self.android_devices[-1]
- self.testclass_params = self.user_params['throughput_test_params']
- self.keysight_test_app = Keysight5GTestApp(
- self.user_params['Keysight5GTestApp'])
- self.testclass_results = collections.OrderedDict()
- self.iperf_server = self.iperf_servers[0]
- self.iperf_client = self.iperf_clients[0]
- self.remote_server = ssh.connection.SshConnection(
- ssh.settings.from_config(
- self.user_params['RemoteServer']['ssh_config']))
- if self.testclass_params.get('reload_scpi', 1):
- self.keysight_test_app.import_scpi_file(
- self.testclass_params['scpi_file'])
- # Configure test retries
- self.user_params['retry_tests'] = [self.__class__.__name__]
-
- # Turn Airplane mode on
- asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- 'Can not turn on airplane mode.')
-
- def teardown_class(self):
- self.log.info('Turning airplane mode on')
- try:
- asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- 'Can not turn on airplane mode.')
- except:
- self.log.warning('Cannot perform teardown operations on DUT.')
- try:
- self.keysight_test_app.set_cell_state('LTE', 1, 0)
- self.keysight_test_app.destroy()
- except:
- self.log.warning('Cannot perform teardown operations on tester.')
- self.process_testclass_results()
-
- def setup_test(self):
- if self.testclass_params['enable_pixel_logs']:
- cputils.start_pixel_logger(self.dut)
-
- def on_retry(self):
- """Function to control test logic on retried tests.
-
- This function is automatically executed on tests that are being
- retried. In this case the function resets wifi, toggles it off and on
- and sets a retry_flag to enable further tweaking the test logic on
- second attempts.
- """
- asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- 'Can not turn on airplane mode.')
- if self.keysight_test_app.get_cell_state('LTE', 'CELL1'):
- self.log.info('Turning LTE off.')
- self.keysight_test_app.set_cell_state('LTE', 'CELL1', 0)
-
- def teardown_test(self):
- self.log.info('Turing airplane mode on')
- asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- 'Can not turn on airplane mode.')
- log_path = os.path.join(
- context.get_current_context().get_full_output_path(), 'pixel_logs')
- os.makedirs(self.log_path, exist_ok=True)
- if self.testclass_params['enable_pixel_logs']:
- cputils.stop_pixel_logger(self.dut, log_path)
- self.process_testcase_results()
- self.pass_fail_check()
-
- 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]
- 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 = {
- 'min_dl_tput':
- testcase_result['tput_result']['total']['DL']['min_tput'],
- 'max_dl_tput':
- testcase_result['tput_result']['total']['DL']['max_tput'],
- 'avg_dl_tput':
- testcase_result['tput_result']['total']['DL']['average_tput'],
- 'theoretical_dl_tput':
- testcase_result['tput_result']['total']['DL']['theoretical_tput'],
- 'dl_bler':
- testcase_result['bler_result']['total']['DL']['nack_ratio'] * 100,
- 'min_dl_tput':
- testcase_result['tput_result']['total']['UL']['min_tput'],
- 'max_dl_tput':
- testcase_result['tput_result']['total']['UL']['max_tput'],
- 'avg_dl_tput':
- testcase_result['tput_result']['total']['UL']['average_tput'],
- 'theoretical_dl_tput':
- testcase_result['tput_result']['total']['UL']['theoretical_tput'],
- 'ul_bler':
- testcase_result['bler_result']['total']['UL']['nack_ratio'] * 100,
- 'tcp_udp_tput':
- testcase_result.get('iperf_throughput', float('nan'))
- }
- 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 pass_fail_check(self):
- pass
-
- 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 = [
- 'Band', 'Channel', 'DL Carriers', 'UL Carriers', 'DL MCS',
- 'DL MIMO', 'UL MCS', 'UL MIMO', 'Cell Power',
- 'DL Min. Throughput', 'DL Max. Throughput',
- 'DL Avg. Throughput', 'DL Theoretical Throughput',
- 'UL Min. Throughput', 'UL Max. Throughput',
- 'UL Avg. Throughput', 'UL Theoretical Throughput',
- 'DL BLER (%)', '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']:
- writer.writerow({
- 'Band':
- testcase_results['testcase_params']['band'],
- 'Channel':
- testcase_results['testcase_params']['channel'],
- 'DL Carriers':
- testcase_results['testcase_params']['num_dl_cells'],
- 'UL Carriers':
- testcase_results['testcase_params']['num_ul_cells'],
- 'DL MCS':
- testcase_results['testcase_params']['dl_mcs'],
- 'DL MIMO':
- testcase_results['testcase_params']['dl_mimo_config'],
- 'UL MCS':
- testcase_results['testcase_params']['ul_mcs'],
- 'UL MIMO':
- testcase_results['testcase_params']['ul_mimo_config'],
- 'Cell Power':
- result['cell_power'],
- 'DL Min. Throughput':
- result['tput_result']['total']['DL']['min_tput'],
- 'DL Max. Throughput':
- result['tput_result']['total']['DL']['max_tput'],
- 'DL Avg. Throughput':
- result['tput_result']['total']['DL']['average_tput'],
- 'DL Theoretical Throughput':
- result['tput_result']['total']['DL']
- ['theoretical_tput'],
- 'UL Min. Throughput':
- result['tput_result']['total']['UL']['min_tput'],
- 'UL Max. Throughput':
- result['tput_result']['total']['UL']['max_tput'],
- 'UL Avg. Throughput':
- result['tput_result']['total']['UL']['average_tput'],
- 'UL Theoretical Throughput':
- result['tput_result']['total']['UL']
- ['theoretical_tput'],
- 'DL BLER (%)':
- result['bler_result']['total']['DL']['nack_ratio'] *
- 100,
- 'UL BLER (%)':
- result['bler_result']['total']['UL']['nack_ratio'] *
- 100,
- 'TCP/UDP Throughput':
- result.get('iperf_throughput', 0)
- })
-
- def setup_tester(self, testcase_params):
- if not self.keysight_test_app.get_cell_state('LTE', 'CELL1'):
- self.log.info('Turning LTE on.')
- self.keysight_test_app.set_cell_state('LTE', 'CELL1', 1)
- self.log.info('Turning off airplane mode')
- asserts.assert_true(utils.force_airplane_mode(self.dut, False),
- 'Can not turn on airplane mode.')
- for cell in testcase_params['dl_cell_list']:
- self.keysight_test_app.set_cell_band('NR5G', cell,
- testcase_params['band'])
- self.keysight_test_app.set_cell_mimo_config(
- 'NR5G', cell, 'DL', testcase_params['dl_mimo_config'])
- self.keysight_test_app.set_cell_dl_power(
- 'NR5G', cell, testcase_params['cell_power_list'][0], 1)
- for cell in testcase_params['ul_cell_list']:
- self.keysight_test_app.set_cell_mimo_config(
- 'NR5G', cell, 'UL', testcase_params['ul_mimo_config'])
- self.keysight_test_app.configure_contiguous_nr_channels(
- testcase_params['dl_cell_list'][0], testcase_params['band'],
- testcase_params['channel'])
- # Consider configuring schedule quick config
- self.keysight_test_app.set_nr_cell_schedule_scenario(
- testcase_params['dl_cell_list'][0],
- testcase_params['schedule_scenario'])
- self.keysight_test_app.set_nr_ul_dft_precoding(
- testcase_params['dl_cell_list'][0],
- testcase_params['transform_precoding'])
- self.keysight_test_app.set_nr_cell_mcs(
- testcase_params['dl_cell_list'][0], testcase_params['dl_mcs'],
- testcase_params['ul_mcs'])
- self.keysight_test_app.set_dl_carriers(testcase_params['dl_cell_list'])
- self.keysight_test_app.set_ul_carriers(testcase_params['ul_cell_list'])
- self.log.info('Waiting for LTE and applying aggregation')
- if not self.keysight_test_app.wait_for_cell_status(
- 'LTE', 'CELL1', 'CONN', 60):
- asserts.fail('DUT did not connect to LTE.')
- self.keysight_test_app.apply_carrier_agg()
- self.log.info('Waiting for 5G connection')
- connected = self.keysight_test_app.wait_for_cell_status(
- 'NR5G', testcase_params['dl_cell_list'][-1], ['ACT', 'CONN'], 60)
- if not connected:
- asserts.fail('DUT did not connect to NR.')
- time.sleep(SHORT_SLEEP)
-
- def run_iperf_traffic(self, testcase_params):
- self.iperf_server.start(tag=0)
- dut_ip = self.dut.droid.connectivityGetIPv4Addresses('rmnet0')[0]
- if 'iperf_server_address' in self.testclass_params:
- iperf_server_address = self.testclass_params[
- 'iperf_server_address']
- elif isinstance(self.iperf_server, ipf.IPerfServerOverAdb):
- iperf_server_address = dut_ip
- else:
- iperf_server_address = wputils.get_server_address(
- self.remote_server, dut_ip, '255.255.255.0')
- client_output_path = self.iperf_client.start(
- iperf_server_address, testcase_params['iperf_args'], 0,
- self.testclass_params['traffic_duration'] + IPERF_TIMEOUT)
- server_output_path = self.iperf_server.stop()
- # Parse and log result
- if testcase_params['use_client_output']:
- iperf_file = client_output_path
- else:
- iperf_file = server_output_path
- try:
- iperf_result = ipf.IPerfResult(iperf_file)
- current_throughput = numpy.mean(iperf_result.instantaneous_rates[
- self.testclass_params['iperf_ignored_interval']:-1]) * 8 * (
- 1.024**2)
- except:
- self.log.warning(
- 'ValueError: Cannot get iperf result. Setting to 0')
- current_throughput = 0
- return current_throughput
-
- def _test_nr_throughput_bler(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 meta data
- """
- testcase_params = self.compile_test_params(testcase_params)
- testcase_results = collections.OrderedDict()
- testcase_results['testcase_params'] = testcase_params
- testcase_results['results'] = []
- # Setup tester and wait for DUT to connect
- self.setup_tester(testcase_params)
- # Run test
- stop_counter = 0
- for cell_power in testcase_params['cell_power_list']:
- result = collections.OrderedDict()
- result['cell_power'] = cell_power
- # Set DL cell power
- for cell in testcase_params['dl_cell_list']:
- self.keysight_test_app.set_cell_dl_power(
- 'NR5G', cell, result['cell_power'], 1)
- self.keysight_test_app.select_display_tab(
- 'NR5G', testcase_params['dl_cell_list'][0], 'BTHR', 'OTAGRAPH')
- time.sleep(SHORT_SLEEP)
- # Start BLER and throughput measurements
- self.keysight_test_app.start_bler_measurement(
- 'NR5G', testcase_params['dl_cell_list'],
- testcase_params['bler_measurement_length'])
- if self.testclass_params['traffic_type'] != 'PHY':
- result['iperf_throughput'] = self.run_iperf_traffic(
- testcase_params)
- if self.testclass_params['log_power_metrics']:
- if testcase_params[
- 'bler_measurement_length'] >= 5000 and self.testclass_params[
- 'traffic_type'] == 'PHY':
- time.sleep(testcase_params['bler_measurement_length'] /
- 1000 - 5)
- cputils.log_system_power_metrics(self.dut, verbose=0)
- else:
- self.log.warning('Test too short to log metrics')
-
- result['bler_result'] = self.keysight_test_app.get_bler_result(
- 'NR5G', testcase_params['dl_cell_list'],
- testcase_params['bler_measurement_length'])
- result['tput_result'] = self.keysight_test_app.get_throughput(
- 'NR5G', testcase_params['dl_cell_list'])
-
- # Print Test Summary
- self.log.info("Cell Power: {}dBm".format(cell_power))
- self.log.info(
- "DL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}"
- .format(
- result['tput_result']['total']['DL']['min_tput'],
- result['tput_result']['total']['DL']['average_tput'],
- result['tput_result']['total']['DL']['max_tput'],
- result['tput_result']['total']['DL']['theoretical_tput']))
- self.log.info(
- "UL PHY Tput (Mbps):\tMin: {:.2f},\tAvg: {:.2f},\tMax: {:.2f},\tTheoretical: {:.2f}"
- .format(
- result['tput_result']['total']['UL']['min_tput'],
- result['tput_result']['total']['UL']['average_tput'],
- result['tput_result']['total']['UL']['max_tput'],
- result['tput_result']['total']['UL']['theoretical_tput']))
- self.log.info("DL BLER: {:.2f}%\tUL BLER: {:.2f}%".format(
- result['bler_result']['total']['DL']['nack_ratio'] * 100,
- result['bler_result']['total']['UL']['nack_ratio'] * 100))
- testcase_results['results'].append(result)
- if self.testclass_params['traffic_type'] != 'PHY':
- self.log.info("{} {} Tput: {:.2f} Mbps".format(
- self.testclass_params['traffic_type'],
- testcase_params['traffic_direction'],
- result['iperf_throughput']))
-
- if result['bler_result']['total']['DL']['nack_ratio'] * 100 > 99:
- stop_counter = stop_counter + 1
- else:
- stop_counter = 0
- if stop_counter == STOP_COUNTER_LIMIT:
- break
- # Turn off NR cells
- for cell in testcase_params['dl_cell_list'][::-1]:
- self.keysight_test_app.set_cell_state('NR5G', cell, 0)
- asserts.assert_true(utils.force_airplane_mode(self.dut, True),
- 'Can not turn on airplane mode.')
-
- # Save results
- self.testclass_results[self.current_test_name] = testcase_results
-
- def compile_test_params(self, testcase_params):
- """Function that completes all test params based on the test name.
-
- Args:
- testcase_params: dict containing test-specific parameters
- """
- testcase_params['bler_measurement_length'] = int(
- self.testclass_params['traffic_duration'] / SUBFRAME_LENGTH)
- testcase_params['cell_power_list'] = numpy.arange(
- self.testclass_params['cell_power_start'],
- self.testclass_params['cell_power_stop'],
- self.testclass_params['cell_power_step'])
- if self.testclass_params['traffic_type'] == 'PHY':
- return testcase_params
- if self.testclass_params['traffic_type'] == 'TCP':
- testcase_params['iperf_socket_size'] = self.testclass_params.get(
- 'tcp_socket_size', None)
- testcase_params['iperf_processes'] = self.testclass_params.get(
- 'tcp_processes', 1)
- elif self.testclass_params['traffic_type'] == 'UDP':
- testcase_params['iperf_socket_size'] = self.testclass_params.get(
- 'udp_socket_size', None)
- testcase_params['iperf_processes'] = self.testclass_params.get(
- 'udp_processes', 1)
- if (testcase_params['traffic_direction'] == 'DL'
- and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb)
- ) or (testcase_params['traffic_direction'] == 'UL'
- and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)):
- testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
- duration=self.testclass_params['traffic_duration'],
- reverse_direction=1,
- traffic_type=self.testclass_params['traffic_type'],
- socket_size=testcase_params['iperf_socket_size'],
- num_processes=testcase_params['iperf_processes'],
- udp_throughput=self.testclass_params['UDP_rates'].get(
- testcase_params['num_dl_cells'],
- self.testclass_params['UDP_rates']["default"]),
- udp_length=1440)
- testcase_params['use_client_output'] = True
- elif (testcase_params['traffic_direction'] == 'UL'
- and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb)
- ) or (testcase_params['traffic_direction'] == 'DL'
- and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)):
- testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
- duration=self.testclass_params['traffic_duration'],
- reverse_direction=0,
- traffic_type=self.testclass_params['traffic_type'],
- socket_size=testcase_params['iperf_socket_size'],
- num_processes=testcase_params['iperf_processes'],
- udp_throughput=self.testclass_params['UDP_rates'].get(
- testcase_params['num_dl_cells'],
- self.testclass_params['UDP_rates']["default"]),
- udp_length=1440)
- testcase_params['use_client_output'] = False
- return testcase_params
-
- def generate_test_cases(self, bands, channels, 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 = ['test_load_scpi']
-
- for band, channel, num_ul_cells, num_dl_cells, mcs_pair in itertools.product(
- bands, channels, num_ul_cells_list, num_dl_cells_list,
- mcs_pair_list):
- if num_ul_cells > num_dl_cells:
- continue
- if channel not in cputils.PCC_PRESET_MAPPING[band]:
- continue
- test_name = 'test_nr_throughput_bler_{}_{}_DL_{}CC_mcs{}_{}_UL_{}CC_mcs{}_{}'.format(
- band, channel, num_dl_cells, mcs_pair[0], dl_mimo_config,
- num_ul_cells, mcs_pair[1], ul_mimo_config)
- test_params = collections.OrderedDict(
- band=band,
- channel=channel,
- dl_mcs=mcs_pair[0],
- ul_mcs=mcs_pair[1],
- num_dl_cells=num_dl_cells,
- num_ul_cells=num_ul_cells,
- dl_mimo_config=dl_mimo_config,
- ul_mimo_config=ul_mimo_config,
- dl_cell_list=list(range(1, num_dl_cells + 1)),
- ul_cell_list=list(range(1, num_ul_cells + 1)),
- **kwargs)
- setattr(self, test_name,
- partial(self._test_nr_throughput_bler, test_params))
- test_cases.append(test_name)
- return test_cases
-
-
-class Cellular5GFR2_DL_ThroughputTest(Cellular5GFR2ThroughputTest):
-
- def __init__(self, controllers):
- super().__init__(controllers)
- 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)),
- dl_mimo_config='N2X2',
- ul_mimo_config='N1X1',
- schedule_scenario="FULL_TPUT",
- traffic_direction='DL',
- transform_precoding=0)
-
-
-class Cellular5GFR2_CP_UL_ThroughputTest(Cellular5GFR2ThroughputTest):
-
- def __init__(self, controllers):
- super().__init__(controllers)
- self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [1], [1],
- dl_mimo_config='N2X2',
- ul_mimo_config='N1X1',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=0)
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [1], [1],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=0))
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [2], [2],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=0))
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [3], [3],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="UL_RMC",
- traffic_direction='UL',
- transform_precoding=0))
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [4], [4],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=0))
-
-
-class Cellular5GFR2_DFTS_UL_ThroughputTest(Cellular5GFR2ThroughputTest):
-
- def __init__(self, controllers):
- super().__init__(controllers)
- self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [1], [1],
- dl_mimo_config='N2X2',
- ul_mimo_config='N1X1',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=1)
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [1], [1],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=1))
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [2], [2],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=1))
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [3], [3],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=1))
- self.tests.extend(
- self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
- ['low', 'mid', 'high'],
- [(4, 16), (4, 27)], [4], [4],
- dl_mimo_config='N2X2',
- ul_mimo_config='N2X2',
- schedule_scenario="FULL_TPUT",
- traffic_direction='UL',
- transform_precoding=1))
-
-
-class Cellular5GFR2_DL_FrequecySweep_ThroughputTest(Cellular5GFR2ThroughputTest
- ):
-
- def __init__(self, controllers):
- super().__init__(controllers)
- dl_frequency_sweep_params = self.user_params['throughput_test_params'][
- 'dl_frequency_sweep']
- self.tests = self.generate_test_cases(dl_frequency_sweep_params,
- [(16, 4), (27, 4)],
- schedule_scenario="FULL_TPUT",
- traffic_direction='DL',
- transform_precoding=0,
- dl_mimo_config='N2X2',
- ul_mimo_config='N1X1')
-
- def generate_test_cases(self, dl_frequency_sweep_params, mcs_pair_list,
- **kwargs):
- """Function that auto-generates test cases for a test class."""
- test_cases = ['test_load_scpi']
-
- for band, band_config in dl_frequency_sweep_params.items():
- for num_dl_cells_str, sweep_config in band_config.items():
- num_dl_cells = int(num_dl_cells_str[0])
- num_ul_cells = 1
- freq_vector = numpy.arange(sweep_config[0], sweep_config[1],
- sweep_config[2])
- for freq in freq_vector:
- for mcs_pair in mcs_pair_list:
- test_name = 'test_nr_throughput_bler_{}_{}MHz_DL_{}CC_mcs{}_UL_{}CC_mcs{}'.format(
- band, freq, num_dl_cells, mcs_pair[0],
- num_ul_cells, mcs_pair[1])
- test_params = collections.OrderedDict(
- band=band,
- channel=freq,
- dl_mcs=mcs_pair[0],
- ul_mcs=mcs_pair[1],
- num_dl_cells=num_dl_cells,
- num_ul_cells=num_ul_cells,
- dl_cell_list=list(range(1, num_dl_cells + 1)),
- ul_cell_list=list(range(1, num_ul_cells + 1)),
- **kwargs)
- setattr(
- self, test_name,
- partial(self._test_nr_throughput_bler,
- test_params))
- test_cases.append(test_name)
- return test_cases
diff --git a/acts_tests/tests/google/cellular/performance/CellularFr1RvRTest.py b/acts_tests/tests/google/cellular/performance/CellularFr1RvRTest.py
index 96b2aa0..605580d 100644
--- a/acts_tests/tests/google/cellular/performance/CellularFr1RvRTest.py
+++ b/acts_tests/tests/google/cellular/performance/CellularFr1RvRTest.py
@@ -23,14 +23,15 @@
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 acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure
-from CellularLtePlusFr1PeakThroughputTest import CellularFr1SingleCellPeakThroughputTest
from functools import partial
-class CellularFr1RvrTest(CellularFr1SingleCellPeakThroughputTest):
+class CellularFr1RvrTest(CellularThroughputBaseTest):
"""Class to test single cell FR1 NSA sensitivity"""
def __init__(self, controllers):
@@ -129,7 +130,7 @@
test_configs, channel_list):
if int(test_config['skip_test']):
continue
- endc_combo_config = self.generate_endc_combo_config(
+ endc_combo_config = cputils.generate_endc_combo_config_from_csv_row(
test_config)
test_name = 'test_fr1_{}_{}'.format(
test_config['nr_band'], channel.lower())
diff --git a/acts_tests/tests/google/cellular/performance/CellularFr1SensitivityTest.py b/acts_tests/tests/google/cellular/performance/CellularFr1SensitivityTest.py
index 2e515cb..3057cfb 100644
--- a/acts_tests/tests/google/cellular/performance/CellularFr1SensitivityTest.py
+++ b/acts_tests/tests/google/cellular/performance/CellularFr1SensitivityTest.py
@@ -23,14 +23,15 @@
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 CellularLtePlusFr1PeakThroughputTest import CellularFr1SingleCellPeakThroughputTest
+from acts_contrib.test_utils.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest
from functools import partial
-class CellularFr1SensitivityTest(CellularFr1SingleCellPeakThroughputTest):
+class CellularFr1SensitivityTest(CellularThroughputBaseTest):
"""Class to test single cell FR1 NSA sensitivity"""
def __init__(self, controllers):
@@ -49,7 +50,10 @@
lte_dl_mcs=4,
lte_ul_mcs_table='QAM256',
lte_ul_mcs=4,
- transform_precoding=0)
+ transform_precoding=0,
+ schedule_scenario='FULL_TPUT',
+ schedule_slot_ratio=80
+ )
def process_testclass_results(self):
# Plot individual test id results raw data and compile metrics
@@ -98,7 +102,6 @@
width=1,
style='dashed')
- # Compute average RvRs and compute metrics over orientations
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'], [])))
@@ -121,17 +124,29 @@
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]
- 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)
bler_list = []
average_throughput_list = []
@@ -176,6 +191,14 @@
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
nr_cell_index = testcase_params['endc_combo_config']['lte_cell_count']
@@ -221,7 +244,7 @@
test_configs, channel_list, dl_mcs_list):
if int(test_config['skip_test']):
continue
- endc_combo_config = self.generate_endc_combo_config(
+ endc_combo_config = cputils.generate_endc_combo_config_from_csv_row(
test_config)
test_name = 'test_fr1_{}_{}_dl_mcs{}'.format(
test_config['nr_band'], channel.lower(), nr_dl_mcs)
@@ -233,3 +256,27 @@
partial(self._test_throughput_bler, test_params))
test_cases.append(test_name)
return test_cases
+
+class CellularFr1Sensitivity_SampleMCS_Test(CellularFr1SensitivityTest):
+ """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['nr_sensitivity_test_params']
+ self.tests = self.generate_test_cases(
+ channel_list=['LOW'],
+ dl_mcs_list=[27, 25, 16, 9],
+ nr_ul_mcs=4,
+ lte_dl_mcs_table='QAM256',
+ lte_dl_mcs=4,
+ lte_ul_mcs_table='QAM256',
+ lte_ul_mcs=4,
+ transform_precoding=0,
+ schedule_scenario='FULL_TPUT',
+ schedule_slot_ratio=80
+ )
\ No newline at end of file
diff --git a/acts_tests/tests/google/cellular/performance/CellularFr2PeakThroughputTest.py b/acts_tests/tests/google/cellular/performance/CellularFr2PeakThroughputTest.py
index 848832e..6fbb4d9 100644
--- a/acts_tests/tests/google/cellular/performance/CellularFr2PeakThroughputTest.py
+++ b/acts_tests/tests/google/cellular/performance/CellularFr2PeakThroughputTest.py
@@ -76,60 +76,60 @@
'nr_cell_count']:
metric_map.update({
'nr_min_dl_tput':
- testcase_result['nr_tput_result']['total']['DL']['min_tput'],
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['DL']['min_tput'],
'nr_max_dl_tput':
- testcase_result['nr_tput_result']['total']['DL']['max_tput'],
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['DL']['max_tput'],
'nr_avg_dl_tput':
- testcase_result['nr_tput_result']['total']['DL']
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['DL']
['average_tput'],
'nr_theoretical_dl_tput':
- testcase_result['nr_tput_result']['total']['DL']
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['DL']
['theoretical_tput'],
'nr_dl_bler':
- testcase_result['nr_bler_result']['total']['DL']['nack_ratio']
+ testcase_result['throughput_measurements']['nr_bler_result']['total']['DL']['nack_ratio']
* 100,
'nr_min_dl_tput':
- testcase_result['nr_tput_result']['total']['UL']['min_tput'],
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['UL']['min_tput'],
'nr_max_dl_tput':
- testcase_result['nr_tput_result']['total']['UL']['max_tput'],
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['UL']['max_tput'],
'nr_avg_dl_tput':
- testcase_result['nr_tput_result']['total']['UL']
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['UL']
['average_tput'],
'nr_theoretical_dl_tput':
- testcase_result['nr_tput_result']['total']['UL']
+ testcase_result['throughput_measurements']['nr_tput_result']['total']['UL']
['theoretical_tput'],
'nr_ul_bler':
- testcase_result['nr_bler_result']['total']['UL']['nack_ratio']
+ testcase_result['throughput_measurements']['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'],
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['DL']['min_tput'],
'lte_max_dl_tput':
- testcase_result['lte_tput_result']['total']['DL']['max_tput'],
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['DL']['max_tput'],
'lte_avg_dl_tput':
- testcase_result['lte_tput_result']['total']['DL']
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['DL']
['average_tput'],
'lte_theoretical_dl_tput':
- testcase_result['lte_tput_result']['total']['DL']
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['DL']
['theoretical_tput'],
'lte_dl_bler':
- testcase_result['lte_bler_result']['total']['DL']['nack_ratio']
+ testcase_result['throughput_measurements']['lte_bler_result']['total']['DL']['nack_ratio']
* 100,
'lte_min_dl_tput':
- testcase_result['lte_tput_result']['total']['UL']['min_tput'],
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['UL']['min_tput'],
'lte_max_dl_tput':
- testcase_result['lte_tput_result']['total']['UL']['max_tput'],
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['UL']['max_tput'],
'lte_avg_dl_tput':
- testcase_result['lte_tput_result']['total']['UL']
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['UL']
['average_tput'],
'lte_theoretical_dl_tput':
- testcase_result['lte_tput_result']['total']['UL']
+ testcase_result['throughput_measurements']['lte_tput_result']['total']['UL']
['theoretical_tput'],
'lte_ul_bler':
- testcase_result['lte_bler_result']['total']['UL']['nack_ratio']
+ testcase_result['throughput_measurements']['lte_bler_result']['total']['UL']['nack_ratio']
* 100
})
if self.publish_testcase_metrics:
@@ -170,68 +170,68 @@
'endc_combo_config']['nr_cell_count']:
row_dict.update({
'NR DL Min. Throughput':
- result['nr_tput_result']['total']['DL']
+ result['throughput_measurements']['nr_tput_result']['total']['DL']
['min_tput'],
'NR DL Max. Throughput':
- result['nr_tput_result']['total']['DL']
+ result['throughput_measurements']['nr_tput_result']['total']['DL']
['max_tput'],
'NR DL Avg. Throughput':
- result['nr_tput_result']['total']['DL']
+ result['throughput_measurements']['nr_tput_result']['total']['DL']
['average_tput'],
'NR DL Theoretical Throughput':
- result['nr_tput_result']['total']['DL']
+ result['throughput_measurements']['nr_tput_result']['total']['DL']
['theoretical_tput'],
'NR UL Min. Throughput':
- result['nr_tput_result']['total']['UL']
+ result['throughput_measurements']['nr_tput_result']['total']['UL']
['min_tput'],
'NR UL Max. Throughput':
- result['nr_tput_result']['total']['UL']
+ result['throughput_measurements']['nr_tput_result']['total']['UL']
['max_tput'],
'NR UL Avg. Throughput':
- result['nr_tput_result']['total']['UL']
+ result['throughput_measurements']['nr_tput_result']['total']['UL']
['average_tput'],
'NR UL Theoretical Throughput':
- result['nr_tput_result']['total']['UL']
+ result['throughput_measurements']['nr_tput_result']['total']['UL']
['theoretical_tput'],
'NR DL BLER (%)':
- result['nr_bler_result']['total']['DL']
+ result['throughput_measurements']['nr_bler_result']['total']['DL']
['nack_ratio'] * 100,
'NR UL BLER (%)':
- result['nr_bler_result']['total']['UL']
+ result['throughput_measurements']['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']
+ result['throughput_measurements']['lte_tput_result']['total']['DL']
['min_tput'],
'LTE DL Max. Throughput':
- result['lte_tput_result']['total']['DL']
+ result['throughput_measurements']['lte_tput_result']['total']['DL']
['max_tput'],
'LTE DL Avg. Throughput':
- result['lte_tput_result']['total']['DL']
+ result['throughput_measurements']['lte_tput_result']['total']['DL']
['average_tput'],
'LTE DL Theoretical Throughput':
- result['lte_tput_result']['total']['DL']
+ result['throughput_measurements']['lte_tput_result']['total']['DL']
['theoretical_tput'],
'LTE UL Min. Throughput':
- result['lte_tput_result']['total']['UL']
+ result['throughput_measurements']['lte_tput_result']['total']['UL']
['min_tput'],
'LTE UL Max. Throughput':
- result['lte_tput_result']['total']['UL']
+ result['throughput_measurements']['lte_tput_result']['total']['UL']
['max_tput'],
'LTE UL Avg. Throughput':
- result['lte_tput_result']['total']['UL']
+ result['throughput_measurements']['lte_tput_result']['total']['UL']
['average_tput'],
'LTE UL Theoretical Throughput':
- result['lte_tput_result']['total']['UL']
+ result['throughput_measurements']['lte_tput_result']['total']['UL']
['theoretical_tput'],
'LTE DL BLER (%)':
- result['lte_bler_result']['total']['DL']
+ result['throughput_measurements']['lte_bler_result']['total']['DL']
['nack_ratio'] * 100,
'LTE UL BLER (%)':
- result['lte_bler_result']['total']['UL']
+ result['throughput_measurements']['lte_bler_result']['total']['UL']
['nack_ratio'] * 100
})
writer.writerow(row_dict)
@@ -301,6 +301,7 @@
'NR5G',
'cell_number':
nr_cell_idx,
+ 'nr_cell_type': 'NSA',
'band':
test_config['nr_band'],
'duplex_mode':
@@ -334,16 +335,17 @@
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
+ 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, bands, channels, nr_mcs_pair_list,
- num_dl_cells_list, num_ul_cells_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 band, channel, num_ul_cells, num_dl_cells, nr_mcs_pair in itertools.product(
- bands, channels, num_ul_cells_list, num_dl_cells_list,
+ 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
@@ -358,6 +360,7 @@
'nr_band': band,
'nr_bandwidth': 'BW100',
'nr_duplex_mode': 'TDD',
+ 'nr_cell_type': 'NSA',
'nr_channel': channel,
'num_dl_cells': num_dl_cells,
'num_ul_cells': num_ul_cells,
@@ -365,14 +368,15 @@
'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,
+ test_name = 'test_fr2_{}_{}_{}_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, test_params))
@@ -390,16 +394,18 @@
def __init__(self, controllers):
super().__init__(controllers)
- self.testclass_params = self.user_params['throughput_test_params']
+ self.testclass_params = self.user_params['fr2_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)),
+ ['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='DL',
transform_precoding=0,
lte_dl_mcs=4,
@@ -416,10 +422,12 @@
self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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=0,
lte_dl_mcs=4,
@@ -430,10 +438,12 @@
self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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,
@@ -444,10 +454,12 @@
self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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,
@@ -458,10 +470,12 @@
self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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,
@@ -478,10 +492,12 @@
self.tests = self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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,
@@ -492,10 +508,12 @@
self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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=1,
lte_dl_mcs=4,
@@ -506,10 +524,12 @@
self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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,
@@ -520,10 +540,12 @@
self.generate_test_cases(['N257', 'N258', 'N260', 'N261'],
['low', 'mid', 'high'],
[(4, 16), (4, 27)], [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,
@@ -548,10 +570,12 @@
['N257', 'N258', 'N260', 'N261'],
self.user_params['throughput_test_params']['frequency_sweep'],
[(16, 4), (27, 4)],
+ ['A_Plane', 'B_Plane'],
force_contiguous_nr_channel=False,
dl_mimo_config=2,
ul_mimo_config=1,
schedule_scenario="FULL_TPUT",
+ schedule_slot_ratio=80,
traffic_direction='DL',
transform_precoding=0,
lte_dl_mcs=4,
diff --git a/acts_tests/tests/google/cellular/performance/CellularFr2SensitivityTest.py b/acts_tests/tests/google/cellular/performance/CellularFr2SensitivityTest.py
index 4176a62..8f92667 100644
--- a/acts_tests/tests/google/cellular/performance/CellularFr2SensitivityTest.py
+++ b/acts_tests/tests/google/cellular/performance/CellularFr2SensitivityTest.py
@@ -26,12 +26,12 @@
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 CellularFr2PeakThroughputTest import CellularFr2PeakThroughputTest
+from acts_contrib.test_utils.cellular.performance.CellularThroughputBaseTest import CellularThroughputBaseTest
from functools import partial
-class CellularFr2SensitivityTest(CellularFr2PeakThroughputTest):
+class CellularFr2SensitivityTest(CellularThroughputBaseTest):
"""Class to test single cell FR1 NSA sensitivity"""
def __init__(self, controllers):
@@ -41,13 +41,14 @@
self.testclass_metric_logger = (
BlackboxMappedMetricLogger.for_test_class())
self.publish_testcase_metrics = True
- self.testclass_params = self.user_params['nr_sensitivity_test_params']
+ 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(27, -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',
@@ -55,6 +56,7 @@
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)
@@ -63,8 +65,10 @@
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'][1]
+ '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
@@ -103,7 +107,6 @@
width=1,
style='dashed')
- # Compute average RvRs and compute metrics over orientations
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'], [])))
@@ -126,30 +129,46 @@
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]
- 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)
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'][
- 1]
+ nr_cell_index]
for result in testcase_data['results']:
- bler_list.append(
- result['nr_bler_result']['total']['DL']['nack_ratio'])
+ bler_list.append(result['throughput_measurements']
+ ['nr_bler_result']['total']['DL']['nack_ratio'])
average_throughput_list.append(
- result['nr_tput_result']['total']['DL']['average_tput'])
+ result['throughput_measurements']['nr_tput_result']['total']
+ ['DL']['average_tput'])
theoretical_throughput_list.append(
- result['nr_tput_result']['total']['DL']['theoretical_tput'])
+ 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)
@@ -167,8 +186,8 @@
sensitivity = cell_power_list[sensitivity_idx]
self.log.info('NR Band {} MCS {} Sensitivity = {}dBm'.format(
testcase_data['testcase_params']['endc_combo_config']['cell_list']
- [1]['band'], testcase_data['testcase_params']['nr_dl_mcs'],
- sensitivity))
+ [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
@@ -177,6 +196,14 @@
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][
@@ -211,13 +238,98 @@
[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':
+ 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,
+ '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, **kwargs):
+ num_dl_cells_list, dl_mimo_config, orientation_list, **kwargs):
"""Function that auto-generates test cases for a test class."""
test_cases = []
- for band, channel, num_dl_cells, nr_dl_mcs in itertools.product(
- band_list, channel_list, num_dl_cells_list, dl_mcs_list):
+ 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 = {
@@ -242,9 +354,9 @@
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)
- self.log.info(test_cases)
return test_cases
diff --git a/acts_tests/tests/google/cellular/performance/CellularLteFr1EndcSensitivityTest.py b/acts_tests/tests/google/cellular/performance/CellularLteFr1EndcSensitivityTest.py
new file mode 100644
index 0000000..3d347c0
--- /dev/null
+++ b/acts_tests/tests/google/cellular/performance/CellularLteFr1EndcSensitivityTest.py
@@ -0,0 +1,249 @@
+#!/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 re
+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 acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure
+from functools import partial
+
+
+class CellularLteFr1EndcSensitivityTest(CellularThroughputBaseTest):
+ """Class to test ENDC 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['endc_sensitivity_test_params']
+ self.tests = self.generate_test_cases(lte_dl_mcs_list=list(numpy.arange(27,0,-1)),
+ lte_dl_mcs_table='QAM256',
+ lte_ul_mcs_table='QAM256',
+ lte_ul_mcs=4,
+ nr_dl_mcs_list=list(numpy.arange(27,0,-1)),
+ nr_ul_mcs=4,
+ transform_precoding=0,
+ schedule_scenario='FULL_TPUT',
+ schedule_slot_ratio=80)
+
+ 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', '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 = []
+ test_cell_idx = testcase_data['testcase_params']['test_cell_idx']
+ test_cell_config = testcase_data['testcase_params']['endc_combo_config']['cell_list'][test_cell_idx]
+ cell_power_list = testcase_data['testcase_params']['cell_power_sweep'][
+ test_cell_idx]
+
+ for result in testcase_data['results']:
+ if test_cell_config['cell_type'] == 'LTE':
+ bler_list.append(1-result['throughput_measurements']
+ ['lte_bler_result'][test_cell_config['cell_number']]['DL']['ack_ratio'])
+ average_throughput_list.append(
+ result['throughput_measurements']['lte_tput_result'][test_cell_config['cell_number']]
+ ['DL']['average_tput'])
+ theoretical_throughput_list.append(
+ result['throughput_measurements']['lte_tput_result'][test_cell_config['cell_number']]
+ ['DL']['theoretical_tput'])
+ else:
+ bler_list.append(1-result['throughput_measurements']
+ ['nr_bler_result'][test_cell_config['cell_number']]['DL']['ack_ratio'])
+ average_throughput_list.append(
+ result['throughput_measurements']['nr_tput_result'][test_cell_config['cell_number']]
+ ['DL']['average_tput'])
+ theoretical_throughput_list.append(
+ result['throughput_measurements']['nr_tput_result'][test_cell_config['cell_number']]
+ ['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
+ sensitivity = cell_power_list[sensitivity_idx]
+ break
+ else:
+ sensitivity = float('nan')
+
+
+ if test_cell_config['cell_type'] == 'LTE':
+ test_mcs = testcase_data['testcase_params']['lte_dl_mcs']
+ else:
+ test_mcs = testcase_data['testcase_params']['nr_dl_mcs']
+ self.log.info('{} Band {} MCS {} Sensitivity = {}dBm'.format(
+ test_cell_config['cell_type'],
+ test_cell_config['band'],
+ test_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):
+ cell_power_sweeps = []
+ # Construct test cell sweep
+ test_cell = testcase_params['endc_combo_config']['cell_list'][testcase_params['test_cell_idx']]
+ if test_cell['cell_type'] == 'LTE':
+ test_cell_sweep = list(
+ numpy.arange(self.testclass_params['lte_cell_power_start'],
+ self.testclass_params['lte_cell_power_stop'],
+ self.testclass_params['lte_cell_power_step']))
+ else:
+ test_cell_sweep = list(
+ numpy.arange(self.testclass_params['nr_cell_power_start'],
+ self.testclass_params['nr_cell_power_stop'],
+ self.testclass_params['nr_cell_power_step']))
+
+ for cell_idx, cell_config in enumerate(testcase_params['endc_combo_config']['cell_list']):
+ if cell_idx == testcase_params['test_cell_idx']:
+ cell_power_sweeps.append(test_cell_sweep)
+ elif cell_config['cell_type'] == 'LTE':
+ lte_sweep = [self.testclass_params['lte_cell_power_start']
+ ] * len(test_cell_sweep)
+ cell_power_sweeps.append(lte_sweep)
+ elif cell_config['cell_type'] == 'NR5G':
+ nr_sweep = [self.testclass_params['nr_cell_power_start']
+ ] * len(test_cell_sweep)
+ cell_power_sweeps.append(nr_sweep)
+ return cell_power_sweeps
+
+ def generate_test_cases(self, lte_dl_mcs_list, lte_dl_mcs_table,
+ lte_ul_mcs_table, lte_ul_mcs, nr_dl_mcs_list,
+ nr_ul_mcs, **kwargs):
+ test_cases = []
+ with open(self.testclass_params['endc_combo_file'],
+ 'r') as endc_combos:
+ for endc_combo_str in endc_combos:
+ if endc_combo_str[0] == '#':
+ continue
+ endc_combo_config = cputils.generate_endc_combo_config_from_string(
+ endc_combo_str)
+ special_chars = '+[]=;,\n'
+ for char in special_chars:
+ endc_combo_str = endc_combo_str.replace(char, '_')
+ endc_combo_str = endc_combo_str.replace('__', '_')
+ endc_combo_str = endc_combo_str.strip('_')
+ for cell_idx, cell_config in enumerate(endc_combo_config['cell_list']):
+ if cell_config['cell_type'] == 'LTE':
+ dl_mcs_list = lte_dl_mcs_list
+ else:
+ dl_mcs_list = nr_dl_mcs_list
+ for dl_mcs in dl_mcs_list:
+ test_name = 'test_sensitivity_{}_cell_{}_mcs{}'.format(
+ endc_combo_str, cell_idx, dl_mcs)
+ if cell_config['cell_type'] == 'LTE':
+ test_params = collections.OrderedDict(
+ endc_combo_config=endc_combo_config,
+ test_cell_idx=cell_idx,
+ lte_dl_mcs_table=lte_dl_mcs_table,
+ lte_dl_mcs=dl_mcs,
+ lte_ul_mcs_table=lte_ul_mcs_table,
+ lte_ul_mcs=lte_ul_mcs,
+ nr_dl_mcs=4,
+ nr_ul_mcs=nr_ul_mcs,
+ **kwargs)
+ else:
+ test_params = collections.OrderedDict(
+ endc_combo_config=endc_combo_config,
+ test_cell_idx=cell_idx,
+ lte_dl_mcs_table=lte_dl_mcs_table,
+ lte_dl_mcs=4,
+ lte_ul_mcs_table=lte_ul_mcs_table,
+ lte_ul_mcs=lte_ul_mcs,
+ nr_dl_mcs=dl_mcs,
+ nr_ul_mcs=nr_ul_mcs,
+ **kwargs)
+ setattr(self, test_name,
+ partial(self._test_throughput_bler, test_params))
+ test_cases.append(test_name)
+ return test_cases
+
+
+class CellularLteFr1EndcSensitivity_SampleMCS_Test(CellularLteFr1EndcSensitivityTest):
+ """Class to test single cell LTE 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['endc_sensitivity_test_params']
+ self.tests = self.generate_test_cases(lte_dl_mcs_list=[27,25,16,9],
+ lte_dl_mcs_table='QAM256',
+ lte_ul_mcs_table='QAM256',
+ lte_ul_mcs=4,
+ nr_dl_mcs_list=[27,25,16,9],
+ nr_ul_mcs=4,
+ transform_precoding=0,
+ schedule_scenario='FULL_TPUT',
+ schedule_slot_ratio=80)
\ No newline at end of file
diff --git a/acts_tests/tests/google/cellular/performance/CellularLtePlusFr1PeakThroughputTest.py b/acts_tests/tests/google/cellular/performance/CellularLtePlusFr1PeakThroughputTest.py
index 65fef93..1a90e9e 100644
--- a/acts_tests/tests/google/cellular/performance/CellularLtePlusFr1PeakThroughputTest.py
+++ b/acts_tests/tests/google/cellular/performance/CellularLtePlusFr1PeakThroughputTest.py
@@ -259,98 +259,9 @@
self.tests = self.generate_test_cases([(27, 4), (4, 27)],
lte_dl_mcs_table='QAM256',
lte_ul_mcs_table='QAM256',
- transform_precoding=0)
-
- def generate_endc_combo_config(self, endc_combo_str):
- """Function to generate ENDC combo config from combo string
-
- Args:
- endc_combo_str: ENDC combo descriptor (e.g. B48A[4];A[1]+N5A[2];A[1])
- Returns:
- endc_combo_config: dictionary with all ENDC combo settings
- """
- endc_combo_str = endc_combo_str.replace(' ', '')
- endc_combo_list = endc_combo_str.split('+')
- endc_combo_list = [combo.split(';') for combo in endc_combo_list]
- endc_combo_config = collections.OrderedDict()
- cell_config_list = list()
- lte_cell_count = 0
- nr_cell_count = 0
- lte_scc_list = []
- nr_dl_carriers = []
- nr_ul_carriers = []
- lte_carriers = []
-
- for cell in endc_combo_list:
- cell_config = {}
- dl_config_str = cell[0]
- dl_config_regex = re.compile(
- r'(?P<cell_type>[B,N])(?P<band>[0-9]+)(?P<bandwidth_class>[A-Z])\[(?P<mimo_config>[0-9])\]'
- )
- dl_config_match = re.match(dl_config_regex, dl_config_str)
- if dl_config_match.group('cell_type') == 'B':
- cell_config['cell_type'] = 'LTE'
- lte_cell_count = lte_cell_count + 1
- cell_config['cell_number'] = lte_cell_count
- if cell_config['cell_number'] == 1:
- cell_config['pcc'] = 1
- endc_combo_config['lte_pcc'] = cell_config['cell_number']
- else:
- cell_config['pcc'] = 0
- lte_scc_list.append(cell_config['cell_number'])
- cell_config['band'] = dl_config_match.group('band')
- cell_config['duplex_mode'] = 'FDD' if int(
- cell_config['band']
- ) in cputils.DUPLEX_MODE_TO_BAND_MAPPING['LTE'][
- 'FDD'] else 'TDD'
- cell_config['dl_mimo_config'] = 'D{nss}U{nss}'.format(
- nss=dl_config_match.group('mimo_config'))
- if int(dl_config_match.group('mimo_config')) == 1:
- cell_config['transmission_mode'] = 'TM1'
- elif int(dl_config_match.group('mimo_config')) == 2:
- cell_config['transmission_mode'] = 'TM2'
- else:
- cell_config['transmission_mode'] = 'TM3'
- lte_carriers.append(cell_config['cell_number'])
- else:
- cell_config['cell_type'] = 'NR5G'
- nr_cell_count = nr_cell_count + 1
- cell_config['cell_number'] = nr_cell_count
- nr_dl_carriers.append(cell_config['cell_number']),
- cell_config['nr_cell_type'] = 'NSA',
- cell_config['band'] = 'N' + dl_config_match.group('band')
- cell_config['duplex_mode'] = 'FDD' if cell_config[
- 'band'] in cputils.DUPLEX_MODE_TO_BAND_MAPPING['NR5G'][
- 'FDD'] else 'TDD'
- cell_config['subcarrier_spacing'] = 'MU0' if cell_config[
- 'duplex_mode'] == 'FDD' else 'MU1'
- cell_config['dl_mimo_config'] = 'N{nss}X{nss}'.format(
- nss=dl_config_match.group('mimo_config'))
-
- cell_config['dl_bandwidth_class'] = dl_config_match.group(
- 'bandwidth_class')
- cell_config['dl_bandwidth'] = 'BW20'
- cell_config['ul_enabled'] = len(cell) > 1
- if cell_config['ul_enabled']:
- ul_config_str = cell[1]
- ul_config_regex = re.compile(
- r'(?P<bandwidth_class>[A-Z])\[(?P<mimo_config>[0-9])\]')
- ul_config_match = re.match(ul_config_regex, ul_config_str)
- cell_config['ul_bandwidth_class'] = ul_config_match.group(
- 'bandwidth_class')
- cell_config['ul_mimo_config'] = 'N{nss}X{nss}'.format(
- nss=ul_config_match.group('mimo_config'))
- if cell_config['cell_type'] == 'NR5G':
- nr_ul_carriers.append(cell_config['cell_number'])
- cell_config_list.append(cell_config)
- 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
+ transform_precoding=0,
+ schedule_scenario='FULL_TPUT',
+ schedule_slot_ratio=80)
def generate_test_cases(self, mcs_pair_list, **kwargs):
test_cases = []
@@ -360,9 +271,9 @@
for endc_combo_str in endc_combos:
if endc_combo_str[0] == '#':
continue
- endc_combo_config = self.generate_endc_combo_config(
+ endc_combo_config = cputils.generate_endc_combo_config_from_string(
endc_combo_str)
- special_chars = '+[];\n'
+ special_chars = '+[]=;,\n'
for char in special_chars:
endc_combo_str = endc_combo_str.replace(char, '_')
endc_combo_str = endc_combo_str.replace('__', '_')
@@ -383,100 +294,7 @@
return test_cases
-class CellularSingleCellThroughputTest(CellularLtePlusFr1PeakThroughputTest):
- """Base Class to test single cell LTE or LTE/FR1"""
-
- 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()
- lte_cell_count = 0
- nr_cell_count = 0
- lte_scc_list = []
- nr_dl_carriers = []
- nr_ul_carriers = []
- lte_carriers = []
-
- cell_config_list = []
- if test_config['lte_band']:
- 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'])
- }
- if int(test_config['lte_dl_mimo_config']) == 1:
- lte_cell['transmission_mode'] = 'TM1'
- elif int(test_config['lte_dl_mimo_config']) == 2:
- lte_cell['transmission_mode'] = 'TM2'
- else:
- lte_cell['transmission_mode'] = 'TM3'
- cell_config_list.append(lte_cell)
- endc_combo_config['lte_pcc'] = 1
- lte_cell_count = 1
- lte_carriers = [1]
-
- if test_config['nr_band']:
- nr_cell = {
- 'cell_type':
- 'NR5G',
- 'cell_number':
- 1,
- 'band':
- test_config['nr_band'],
- 'nr_cell_type': test_config['nr_cell_type'],
- 'duplex_mode':
- test_config['nr_duplex_mode'],
- '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,
- 'ul_bandwidth_class':
- 'A',
- 'ul_mimo_config':
- 'N{nss}X{nss}'.format(nss=test_config['nr_ul_mimo_config']),
- 'subcarrier_spacing':
- 'MU0' if test_config['nr_scs'] == '15' else 'MU1'
- }
- cell_config_list.append(nr_cell)
- nr_cell_count = 1
- nr_dl_carriers = [1]
- nr_ul_carriers = [1]
-
- 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
-
-
-class CellularFr1SingleCellPeakThroughputTest(CellularSingleCellThroughputTest
+class CellularFr1SingleCellPeakThroughputTest(CellularLtePlusFr1PeakThroughputTest
):
"""Class to test single cell FR1 NSA mode"""
@@ -509,7 +327,7 @@
test_configs, nr_channel_list, nr_mcs_pair_list):
if int(test_config['skip_test']):
continue
- endc_combo_config = self.generate_endc_combo_config(
+ endc_combo_config = cputils.generate_endc_combo_config_from_csv_row(
test_config)
endc_combo_config['cell_list'][endc_combo_config['lte_cell_count']]['channel'] = nr_channel
test_name = 'test_fr1_{}_{}_dl_mcs{}_ul_mcs{}'.format(
@@ -526,7 +344,7 @@
return test_cases
-class CellularLteSingleCellPeakThroughputTest(CellularSingleCellThroughputTest
+class CellularLteSingleCellPeakThroughputTest(CellularLtePlusFr1PeakThroughputTest
):
"""Class to test single cell LTE"""
@@ -552,7 +370,7 @@
test_configs, lte_mcs_pair_list):
if int(test_config['skip_test']):
continue
- endc_combo_config = self.generate_endc_combo_config(
+ endc_combo_config = cputils.generate_endc_combo_config_from_csv_row(
test_config)
test_name = 'test_lte_B{}_dl_{}_mcs{}_ul_{}_mcs{}'.format(
test_config['lte_band'], lte_mcs_pair[0][0],
diff --git a/acts_tests/tests/google/cellular/performance/CellularLteRvrTest.py b/acts_tests/tests/google/cellular/performance/CellularLteRvrTest.py
index d444bca..16e6e95 100644
--- a/acts_tests/tests/google/cellular/performance/CellularLteRvrTest.py
+++ b/acts_tests/tests/google/cellular/performance/CellularLteRvrTest.py
@@ -24,14 +24,15 @@
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 acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure
-from CellularLtePlusFr1PeakThroughputTest import CellularLteSingleCellPeakThroughputTest
from functools import partial
-class CellularLteRvrTest(CellularLteSingleCellPeakThroughputTest):
+class CellularLteRvrTest(CellularThroughputBaseTest):
"""Class to test single cell LTE sensitivity"""
def __init__(self, controllers):
@@ -41,7 +42,7 @@
self.testclass_metric_logger = (
BlackboxMappedMetricLogger.for_test_class())
self.publish_testcase_metrics = True
- self.testclass_params = self.user_params['lte_sensitivity_test_params']
+ self.testclass_params = self.user_params['lte_rvr_test_params']
self.tests = self.generate_test_cases(lte_dl_mcs_table='QAM256',
lte_ul_mcs_table='QAM256',
lte_ul_mcs=4,
@@ -194,7 +195,7 @@
for test_config in test_configs:
if int(test_config['skip_test']):
continue
- endc_combo_config = self.generate_endc_combo_config(
+ endc_combo_config = cputils.generate_endc_combo_config_from_csv_row(
test_config)
test_name = 'test_lte_B{}_dl_{}'.format(
test_config['lte_band'], lte_dl_mcs_table)
diff --git a/acts_tests/tests/google/cellular/performance/CellularLteSensitivityTest.py b/acts_tests/tests/google/cellular/performance/CellularLteSensitivityTest.py
index 22e436b..f3b5afe 100644
--- a/acts_tests/tests/google/cellular/performance/CellularLteSensitivityTest.py
+++ b/acts_tests/tests/google/cellular/performance/CellularLteSensitivityTest.py
@@ -24,14 +24,14 @@
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 acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure
-from CellularLtePlusFr1PeakThroughputTest import CellularLteSingleCellPeakThroughputTest
-
from functools import partial
-class CellularLteSensitivityTest(CellularLteSingleCellPeakThroughputTest):
+class CellularLteSensitivityTest(CellularThroughputBaseTest):
"""Class to test single cell LTE sensitivity"""
def __init__(self, controllers):
@@ -100,7 +100,6 @@
width=1,
style='dashed')
- # Compute average RvRs and compute metrics over orientations
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'], [])))
@@ -123,17 +122,29 @@
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]
- 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)
bler_list = []
average_throughput_list = []
@@ -176,6 +187,14 @@
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'][0][
@@ -217,7 +236,7 @@
test_configs, dl_mcs_list):
if int(test_config['skip_test']):
continue
- endc_combo_config = self.generate_endc_combo_config(
+ endc_combo_config = cputils.generate_endc_combo_config_from_csv_row(
test_config)
test_name = 'test_lte_B{}_dl_{}_mcs{}'.format(
test_config['lte_band'], lte_dl_mcs_table, lte_dl_mcs)
@@ -232,3 +251,21 @@
partial(self._test_throughput_bler, test_params))
test_cases.append(test_name)
return test_cases
+
+
+class CellularLteSensitivity_SampleMCS_Test(CellularLteSensitivityTest):
+ """Class to test single cell LTE 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['lte_sensitivity_test_params']
+ self.tests = self.generate_test_cases(dl_mcs_list=[27,25,16,9],
+ lte_dl_mcs_table='QAM256',
+ lte_ul_mcs_table='QAM256',
+ lte_ul_mcs=4,
+ transform_precoding=0)
\ No newline at end of file
diff --git a/acts_tests/tests/google/wifi/WifiRvrTest.py b/acts_tests/tests/google/wifi/WifiRvrTest.py
index cb188fd..731807c 100644
--- a/acts_tests/tests/google/wifi/WifiRvrTest.py
+++ b/acts_tests/tests/google/wifi/WifiRvrTest.py
@@ -1103,13 +1103,13 @@
WifiOtaRvrTest.__init__(self, controllers)
self.tests = self.generate_test_cases([6], ['bw20'],
list(range(0, 360, 45)), ['UDP'],
- ['DL'])
+ ['DL', 'UL'])
self.tests.extend(
self.generate_test_cases([36, 149], ['bw80', 'bw160'],
- list(range(0, 360, 45)), ['UDP'], ['DL']))
+ list(range(0, 360, 45)), ['UDP'], ['DL', 'UL']))
self.tests.extend(
self.generate_test_cases(['6g37'], ['bw160'],
- list(range(0, 360, 45)), ['UDP'], ['DL']))
+ list(range(0, 360, 45)), ['UDP'], ['DL', 'UL']))
class WifiOtaRvr_SingleOrientation_Test(WifiOtaRvrTest):
