xref: /tools/test/connectivity/acts/framework/acts/test_utils/power/tel_simulations/BaseSimulation.py

#!/usr/bin/env python3.4
#
#   Copyright 2018 - 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 time
from enum import Enum

import numpy as np

from acts.controllers.anritsu_lib._anritsu_utils import AnritsuError
from acts.controllers.anritsu_lib.md8475a import BtsNumber
from acts.test_utils.tel.tel_test_utils import get_telephony_signal_strength
from acts.test_utils.tel.tel_test_utils import toggle_airplane_mode
from acts.test_utils.tel.tel_test_utils import toggle_cell_data_roaming


class BaseSimulation():
    """ Base class for an Anritsu Simulation abstraction.

    Classes that inherit from this base class implement different simulation
    setups. The base class contains methods that are common to all simulation
    configurations.

    """

    NUM_UL_CAL_READS = 3
    NUM_DL_CAL_READS = 5
    DL_CAL_TARGET_POWER = {'A': -15.0, 'B': -35.0}
    MAX_BTS_INPUT_POWER = 30
    MAX_PHONE_OUTPUT_POWER = 23
    DL_MAX_POWER = {'A': -10.0, 'B': -30.0}
    UL_MIN_POWER = -60.0

    # Key to read the calibration setting from the test_config dictionary.
    KEY_CALIBRATION = "calibration"

    # Time in seconds to wait for the phone to settle
    # after attaching to the base station.
    SETTLING_TIME = 10

    # Time in seconds to wait for the phone to attach
    # to the basestation after toggling airplane mode.
    ATTACH_WAITING_TIME = 120

    # Max retries before giving up attaching the phone
    ATTACH_MAX_RETRIES = 3

    def __init__(self, anritsu, log, dut, test_config, calibration_table):
        """ Initializes the Simulation object.

        Keeps a reference to the callbox, log and dut handlers and
        initializes the class attributes.

        Args:
            anritsu: the Anritsu callbox controller
            log: a logger handle
            dut: the android device handler
            test_config: test configuration obtained from the config file
            calibration_table: a dictionary containing path losses for
                different bands.
        """

        self.anritsu = anritsu
        self.log = log
        self.dut = dut
        self.calibration_table = calibration_table

        # Turn calibration on or off depending on the test config value. If the
        # key is not present, set to False by default
        if self.KEY_CALIBRATION not in test_config:
            self.log.warning("The '{} 'key is not set in the testbed "
                             "parameters. Setting to off by default. To "
                             "turn calibration on, include the key with "
                             "a true/false value.".format(
                                 self.KEY_CALIBRATION))

        self.calibration_required = test_config.get(self.KEY_CALIBRATION,
                                                    False)

        # Gets BTS1 since this sim only has 1 BTS
        self.bts1 = self.anritsu.get_BTS(BtsNumber.BTS1)

        # Store the current calibrated band
        self.current_calibrated_band = None

        # Path loss measured during calibration
        self.dl_path_loss = None
        self.ul_path_loss = None

        # Target signal levels obtained during configuration
        self.sim_dl_power = None
        self.sim_ul_power = None

        # Set to default APN
        log.info("Setting preferred APN to anritsu1.com.")
        dut.droid.telephonySetAPN("anritsu1.com", "anritsu1.com")

        # Enable roaming on the phone
        toggle_cell_data_roaming(self.dut, True)

    def start(self):
        """ Start simulation.

        Starts the simulation in the Anritsu Callbox.

        """

        # Make sure airplane mode is on so the phone won't attach right away
        toggle_airplane_mode(self.log, self.dut, True)

        # Wait for airplane mode setting to propagate
        time.sleep(2)

        # Start simulation if it wasn't started
        self.anritsu.start_simulation()

    def attach(self):
        """ Attach the phone to the basestation.

        Sets a good signal level, toggles airplane mode
        and waits for the phone to attach.

        Returns:
            True if the phone was able to attach, False if not.
        """

        # Turn on airplane mode
        toggle_airplane_mode(self.log, self.dut, True)

        # Wait for airplane mode setting to propagate
        time.sleep(2)

        # Provide a good signal power for the phone to attach easily
        self.bts1.input_level = -10
        time.sleep(2)
        self.bts1.output_level = -30

        # Try to attach the phone.
        for i in range(self.ATTACH_MAX_RETRIES):

            try:

                # Turn off airplane mode
                toggle_airplane_mode(self.log, self.dut, False)

                # Wait for the phone to attach.
                self.anritsu.wait_for_registration_state(
                    time_to_wait=self.ATTACH_WAITING_TIME)

            except AnritsuError as e:

                # The phone failed to attach
                self.log.info(
                    "UE failed to attach on attempt number {}.".format(i + 1))
                self.log.info("Error message: {}".format(str(e)))

                # Turn airplane mode on to prepare the phone for a retry.
                toggle_airplane_mode(self.log, self.dut, True)

                # Wait for APM to propagate
                time.sleep(3)

                # Retry
                if i < self.ATTACH_MAX_RETRIES - 1:
                    # Retry
                    continue
                else:
                    # No more retries left. Return False.
                    return False

            else:
                # The phone attached successfully.
                time.sleep(self.SETTLING_TIME)
                self.log.info("UE attached to the callbox.")
                break

        # Set signal levels obtained from the test parameters
        if self.sim_dl_power:
            self.set_downlink_rx_power(self.bts1, self.sim_dl_power)
            time.sleep(2)
        if self.sim_ul_power:
            self.set_uplink_tx_power(self.bts1, self.sim_ul_power)
            time.sleep(2)

        return True

    def detach(self):
        """ Detach the phone from the basestation.

        Turns airplane mode and resets basestation.
        """

        # Set the DUT to airplane mode so it doesn't see the
        # cellular network going off
        toggle_airplane_mode(self.log, self.dut, True)

        # Wait for APM to propagate
        time.sleep(2)

        # Power off basestation
        self.anritsu.set_simulation_state_to_poweroff()

    def stop(self):
        """  Detach phone from the basestation by stopping the simulation.

        Send stop command to anritsu and turn on airplane mode.

        """

        # Set the DUT to airplane mode so it doesn't see the
        # cellular network going off
        toggle_airplane_mode(self.log, self.dut, True)

        # Wait for APM to propagate
        time.sleep(2)

        # Stop the simulation
        self.anritsu.stop_simulation()

    def parse_parameters(self, parameters):
        """ Configures simulation using a list of parameters.

        Consumes parameters from a list.
        Children classes need to call this method first.

        Args:
            parameters: list of parameters
        """

        pass

    def consume_parameter(self, parameters, parameter_name, num_values=0):
        """ Parses a parameter from a list.

        Allows to parse the parameter list. Will delete parameters from the
        list after consuming them to ensure that they are not used twice.

        Args:
            parameters: list of parameters
            parameter_name: keyword to look up in the list
            num_values: number of arguments following the
                parameter name in the list
        Returns:
            A list containing the parameter name and the following num_values
            arguments
        """

        try:
            i = parameters.index(parameter_name)
        except ValueError:
            # parameter_name is not set
            return []

        return_list = []

        try:
            for j in range(num_values + 1):
                return_list.append(parameters.pop(i))
        except IndexError:
            raise ValueError(
                "Parameter {} has to be followed by {} values.".format(
                    parameter_name, num_values))

        return return_list

    def set_downlink_rx_power(self, bts, signal_level):
        """ Sets downlink rx power using calibration if available

        Args:
            bts: the base station in which to change the signal level
            signal_level: desired downlink received power, can be either a
            key value pair, an int or a float
        """

        # Obtain power value if the provided signal_level is a key value pair
        if isinstance(signal_level, Enum):
            power = signal_level.value
        else:
            power = signal_level

        # Try to use measured path loss value. If this was not set, it will
        # throw an TypeError exception
        try:
            calibrated_power = round(power + self.dl_path_loss)
            if (calibrated_power >
                    self.DL_MAX_POWER[self.anritsu._md8475_version]):
                self.log.warning(
                    "Cannot achieve phone DL Rx power of {} dBm. Requested TX "
                    "power of {} dBm exceeds callbox limit!".format(
                        power, calibrated_power))
                calibrated_power = self.DL_MAX_POWER[
                    self.anritsu._md8475_version]
                self.log.warning(
                    "Setting callbox Tx power to max possible ({} dBm)".format(
                        calibrated_power))

            self.log.info(
                "Requested phone DL Rx power of {} dBm, setting callbox Tx "
                "power at {} dBm".format(power, calibrated_power))
            bts.output_level = calibrated_power
            time.sleep(2)
            # Power has to be a natural number so calibration wont be exact.
            # Inform the actual received power after rounding.
            self.log.info(
                "Phone downlink received power is {0:.2f} dBm".format(
                    calibrated_power - self.dl_path_loss))
        except TypeError:
            bts.output_level = round(power)
            self.log.info("Phone downlink received power set to {} (link is "
                          "uncalibrated).".format(round(power)))

    def set_uplink_tx_power(self, bts, signal_level):
        """ Sets uplink tx power using calibration if available

        Args:
            bts: the base station in which to change the signal level
            signal_level: desired uplink transmitted power, can be either a
            key value pair, an int or a float
        """

        # Obtain power value if the provided signal_level is a key value pair
        if isinstance(signal_level, Enum):
            power = signal_level.value
        else:
            power = signal_level

        # Try to use measured path loss value. If this was not set, it will
        # throw an TypeError exception
        try:
            calibrated_power = round(power - self.ul_path_loss)
            if calibrated_power < self.UL_MIN_POWER:
                self.log.warning(
                    "Cannot achieve phone UL Tx power of {} dBm. Requested UL "
                    "power of {} dBm exceeds callbox limit!".format(
                        power, calibrated_power))
                calibrated_power = self.UL_MIN_POWER
                self.log.warning(
                    "Setting UL Tx power to min possible ({} dBm)".format(
                        calibrated_power))

            self.log.info(
                "Requested phone UL Tx power of {} dBm, setting callbox Rx "
                "power at {} dBm".format(power, calibrated_power))
            bts.input_level = calibrated_power
            time.sleep(2)
            # Power has to be a natural number so calibration wont be exact.
            # Inform the actual transmitted power after rounding.
            self.log.info(
                "Phone uplink transmitted power is {0:.2f} dBm".format(
                    calibrated_power + self.ul_path_loss))
        except TypeError:
            bts.input_level = round(power)
            self.log.info("Phone uplink transmitted power set to {} (link is "
                          "uncalibrated).".format(round(power)))

    def calibrate(self):
        """ Calculates UL and DL path loss if it wasn't done before.

        """
        # SET TBS pattern for calibration
        self.bts1.tbs_pattern = "FULLALLOCATION" if self.tbs_pattern_on else "OFF"

        if self.dl_path_loss and self.ul_path_loss:
            self.log.info("Measurements are already calibrated.")

        # Attach the phone to the base station
        if not self.attach():
            self.log.info(
                "Skipping calibration because the phone failed to attach.")
            return

        # If downlink or uplink were not yet calibrated, do it now
        if not self.dl_path_loss:
            self.dl_path_loss = self.downlink_calibration(self.bts1)
        if not self.ul_path_loss:
            self.ul_path_loss = self.uplink_calibration(self.bts1)

        # Detach after calibrating
        self.detach()
        time.sleep(2)

    def downlink_calibration(self,
                             bts,
                             rat=None,
                             power_units_conversion_func=None):
        """ Computes downlink path loss and returns the calibration value

        The bts needs to be set at the desired config (bandwidth, mode, etc)
        before running the calibration. The phone also needs to be attached
        to the desired basesation for calibration

        Args:
            bts: basestation handle
            rat: desired RAT to calibrate (matching the label reported by
                the phone)
            power_units_conversion_func: a function to convert the units
                reported by the phone to dBm. needs to take two arguments: the
                reported signal level and bts. use None if no conversion is
                needed.
        Returns:
            Dowlink calibration value and measured DL power.
        """

        # Check if this parameter was set. Child classes may need to override
        # this class passing the necessary parameters.
        if not rat:
            raise ValueError(
                "The parameter 'rat' has to indicate the RAT being used as "
                "reported by the phone.")

        # Set BTS to a good output level to minimize measurement error
        init_output_level = bts.output_level
        initial_screen_timeout = self.dut.droid.getScreenTimeout()
        bts.output_level = self.DL_CAL_TARGET_POWER[
            self.anritsu._md8475_version]

        # Set phone sleep time out
        self.dut.droid.setScreenTimeout(1800)
        self.dut.droid.goToSleepNow()
        time.sleep(2)

        # Starting first the IP traffic (UDP): Using always APN 1
        if not self.tbs_pattern_on:
            try:
                cmd = 'OPERATEIPTRAFFIC START,1'
                self.anritsu.send_command(cmd)
            except AnritsuError as inst:
                self.log.warning(
                    "{}\n".format(inst))  # Typically RUNNING already
            time.sleep(4)

        down_power_measured = []
        for i in range(0, self.NUM_DL_CAL_READS):
            # For some reason, the RSRP gets updated on Screen ON event
            self.dut.droid.wakeUpNow()
            time.sleep(4)
            signal_strength = get_telephony_signal_strength(self.dut)
            down_power_measured.append(signal_strength[rat])
            self.dut.droid.goToSleepNow()
            time.sleep(5)

        # Stop the IP traffic (UDP)
        if not self.tbs_pattern_on:
            try:
                cmd = 'OPERATEIPTRAFFIC STOP,1'
                self.anritsu.send_command(cmd)
            except AnritsuError as inst:
                self.log.warning(
                    "{}\n".format(inst))  # Typically STOPPED already
            time.sleep(2)

        # Reset phone and bts to original settings
        self.dut.droid.goToSleepNow()
        self.dut.droid.setScreenTimeout(initial_screen_timeout)
        bts.output_level = init_output_level
        time.sleep(2)

        # Calculate the mean of the measurements
        reported_asu_power = np.nanmean(down_power_measured)

        # Convert from RSRP to signal power
        if power_units_conversion_func:
            avg_down_power = power_units_conversion_func(
                reported_asu_power, bts)
        else:
            avg_down_power = reported_asu_power

        # Calculate Path Loss
        dl_target_power = self.DL_CAL_TARGET_POWER[
            self.anritsu._md8475_version]
        down_call_path_loss = dl_target_power - avg_down_power

        # Validate the result
        if not 0 < down_call_path_loss < 100:
            raise RuntimeError(
                "Downlink calibration failed. The calculated path loss value "
                "was {} dBm.".format(down_call_path_loss))

        self.log.info(
            "Measured downlink path loss: {} dB".format(down_call_path_loss))

        return down_call_path_loss

    def uplink_calibration(self, bts):
        """ Computes uplink path loss and returns the calibration value

        The bts needs to be set at the desired config (bandwidth, mode, etc)
        before running the calibration. The phone also neeeds to be attached
        to the desired basesation for calibration

        Args:
            bts: basestation handle

        Returns:
            Uplink calibration value and measured UL power
        """

        # Set BTS1 to maximum input allowed in order to perform
        # uplink calibration
        target_power = self.MAX_PHONE_OUTPUT_POWER
        initial_input_level = bts.input_level
        initial_screen_timeout = self.dut.droid.getScreenTimeout()
        bts.input_level = self.MAX_BTS_INPUT_POWER

        # Set phone sleep time out
        self.dut.droid.setScreenTimeout(1800)
        self.dut.droid.wakeUpNow()
        time.sleep(2)

        # Starting first the IP traffic (UDP): Using always APN 1
        if not self.tbs_pattern_on:
            try:
                cmd = 'OPERATEIPTRAFFIC START,1'
                self.anritsu.send_command(cmd)
            except AnritsuError as inst:
                self.log.warning(
                    "{}\n".format(inst))  # Typically RUNNING already
            time.sleep(4)

        up_power_per_chain = []
        # Get the number of chains
        cmd = 'MONITOR? UL_PUSCH'
        uplink_meas_power = self.anritsu.send_query(cmd)
        str_power_chain = uplink_meas_power.split(',')
        num_chains = len(str_power_chain)
        for ichain in range(0, num_chains):
            up_power_per_chain.append([])

        for i in range(0, self.NUM_UL_CAL_READS):
            uplink_meas_power = self.anritsu.send_query(cmd)
            str_power_chain = uplink_meas_power.split(',')

            for ichain in range(0, num_chains):
                if (str_power_chain[ichain] == 'DEACTIVE'):
                    up_power_per_chain[ichain].append(float('nan'))
                else:
                    up_power_per_chain[ichain].append(
                        float(str_power_chain[ichain]))

            time.sleep(3)

        # Stop the IP traffic (UDP)
        if not self.tbs_pattern_on:
            try:
                cmd = 'OPERATEIPTRAFFIC STOP,1'
                self.anritsu.send_command(cmd)
            except AnritsuError as inst:
                self.log.warning(
                    "{}\n".format(inst))  # Typically STOPPED already
            time.sleep(2)

        # Reset phone and bts to original settings
        self.dut.droid.goToSleepNow()
        self.dut.droid.setScreenTimeout(initial_screen_timeout)
        bts.input_level = initial_input_level
        time.sleep(2)

        # Phone only supports 1x1 Uplink so always chain 0
        avg_up_power = np.nanmean(up_power_per_chain[0])
        if np.isnan(avg_up_power):
            raise RuntimeError(
                "Calibration failed because the callbox reported the chain to "
                "be deactive.")

        up_call_path_loss = target_power - avg_up_power

        # Validate the result
        if not 0 < up_call_path_loss < 100:
            raise RuntimeError(
                "Uplink calibration failed. The calculated path loss value "
                "was {} dBm.".format(up_call_path_loss))

        self.log.info(
            "Measured uplink path loss: {} dB".format(up_call_path_loss))

        return up_call_path_loss

    def set_band(self, bts, band, calibrate_if_necessary=True):
        """ Sets the band used for communication.

        When moving to a new band, recalibrate the link.

        Args:
            bts: basestation handle
            band: desired band
            calibrate_if_necessary: if False calibration will be skipped
        """

        bts.band = band
        time.sleep(5)  # It takes some time to propagate the new band

        # Invalidate the calibration values
        self.dl_path_loss = None
        self.ul_path_loss = None

        # Only calibrate when required.
        if self.calibration_required and calibrate_if_necessary:
            # Try loading the path loss values from the calibration table. If
            # they are not available, use the automated calibration procedure.
            try:
                self.dl_path_loss = self.calibration_table[band]["dl"]
                self.ul_path_loss = self.calibration_table[band]["ul"]
            except KeyError:
                self.calibrate()

            # Complete the calibration table with the new values to be used in
            # the next tests.
            if band not in self.calibration_table:
                self.calibration_table[band] = {}

            if "dl" not in self.calibration_table[band] and self.dl_path_loss:
                self.calibration_table[band]["dl"] = self.dl_path_loss

            if "ul" not in self.calibration_table[band] and self.ul_path_loss:
                self.calibration_table[band]["ul"] = self.ul_path_loss

    def maximum_downlink_throughput(self):
        """ Calculates maximum achievable downlink throughput in the current
        simulation state.

        Because thoughput is dependent on the RAT, this method needs to be
        implemented by children classes.

        Returns:
            Maximum throughput in mbps
        """
        raise NotImplementedError()

    def maximum_uplink_throughput(self):
        """ Calculates maximum achievable downlink throughput in the current
        simulation state.

        Because thoughput is dependent on the RAT, this method needs to be
        implemented by children classes.

        Returns:
            Maximum throughput in mbps
        """
        raise NotImplementedError()

    def start_test_case(self):
        """ Starts a test case in the current simulation.

        Requires the phone to be attached.
        """

        pass