xref: /tools/test/connectivity/acts/framework/acts/controllers/cellular_simulator.py

#!/usr/bin/env python3
#
#   Copyright 2019 - 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.
from acts import logger
from acts.controllers import cellular_lib


class AbstractCellularSimulator:
    """ A generic cellular simulator controller class that can be derived to
    implement equipment specific classes and allows the tests to be implemented
    without depending on a singular instrument model.

    This class defines the interface that every cellular simulator controller
    needs to implement and shouldn't be instantiated by itself. """

    # The maximum number of carriers that this simulator can support for LTE
    LTE_MAX_CARRIERS = None

    # The maximum power that the equipment is able to transmit
    MAX_DL_POWER = None

    def __init__(self):
        """ Initializes the cellular simulator. """
        self.log = logger.create_tagged_trace_logger('CellularSimulator')
        self.num_carriers = None

    def destroy(self):
        """ Sends finalization commands to the cellular equipment and closes
        the connection. """
        raise NotImplementedError()

    def setup_lte_scenario(self):
        """ Configures the equipment for an LTE simulation. """
        raise NotImplementedError()

    def set_band_combination(self, bands, mimo_modes):
        """ Prepares the test equipment for the indicated band/mimo combination.

        Args:
            bands: a list of bands represented as ints or strings
            mimo_modes: a list of LteSimulation.MimoMode to use for each antenna
        """
        raise NotImplementedError()

    def configure_bts(self, config, bts_index=0):
        """ Commands the equipment to setup a base station with the required
        configuration. This method applies configurations that are common to all
        RATs.

        Args:
            config: a BaseSimulation.BtsConfig object.
            bts_index: the base station number.
        """

        config_vars = vars(config)
        config_dict = {
            key: config_vars[key]
            for key in config_vars if config_vars[key]
        }
        self.log.info('The config for {} is {}'.format(bts_index, config_dict))

        if config.output_power:
            self.set_output_power(bts_index, config.output_power)

        if config.input_power:
            self.set_input_power(bts_index, config.input_power)

        if isinstance(config, cellular_lib.LteCellConfig.LteCellConfig):
            self.configure_lte_bts(config, bts_index)

        if isinstance(config, cellular_lib.NrCellConfig.NrCellConfig):
            self.configure_nr_bts(config, bts_index)

    def configure_bts_after_started(self, config, bts_index=0):
        """ Commands the equipment to setup a base station with the required
        configuration after simulation started. This method applies
        configurations for some simulator.

        Args:
            config: a BaseSimulation.BtsConfig object.
            bts_index: the base station number.
        """

        self.log.info('Configure after simulation started for some simulator')

        if isinstance(config, cellular_lib.LteCellConfig.LteCellConfig):
            self.configure_lte_bts_after_started(config, bts_index)

        if isinstance(config, cellular_lib.NrCellConfig.NrCellConfig):
            self.configure_nr_bts_after_started(config, bts_index)


    def configure_lte_bts_after_started(self, config, bts_index=0):
        """ Commands the equipment to further setup a lte base station.

        This is for some simulator where further setup is needed after the
           simulation started.

        Args:
            config: a BaseSimulation.BtsConfig object.
            bts_index: the base station number.
        """
        pass

    def configure_nr_bts_after_started(self, config, bts_index=0):
        """ Commands the equipment to further setup a lte base station.

        This is for some simulator where further setup is needed after the
           simulation started.

        Args:
            config: a BaseSimulation.BtsConfig object.
            bts_index: the base station number.
        """
        pass

    def configure_lte_bts_base(self, config, bts_index=0):
        """ Commands the equipment to setup an LTE base station with the
        required configuration.

        Args:
            config: an LteSimulation.BtsConfig object.
            bts_index: the base station number.
        """
        if config.tracking_area:
            self.set_tracking_area(bts_index, config.tracking_area)

        if config.band:
            self.set_band(bts_index, config.band)

        if config.dlul_config:
            self.set_tdd_config(bts_index, config.dlul_config)

        if config.ssf_config:
            self.set_ssf_config(bts_index, config.ssf_config)

        if config.bandwidth:
            self.set_bandwidth(bts_index, config.bandwidth)

        if config.dl_channel:
            self.set_downlink_channel_number(bts_index, config.dl_channel)

        if config.mimo_mode:
            self.set_mimo_mode(bts_index, config.mimo_mode)

        if config.transmission_mode:
            self.set_transmission_mode(bts_index, config.transmission_mode)

        # Modulation order should be set before set_scheduling_mode being
        # called.
        if config.dl_256_qam_enabled is not None:
            self.set_dl_256_qam_enabled(bts_index, config.dl_256_qam_enabled)

        if config.ul_64_qam_enabled is not None:
            self.set_ul_64_qam_enabled(bts_index, config.ul_64_qam_enabled)

        if config.scheduling_mode:

            if (config.scheduling_mode
                    == cellular_lib.LteSimulation.SchedulingMode.STATIC
                    and not (config.dl_rbs and config.ul_rbs and config.dl_mcs
                             and config.ul_mcs)):
                raise ValueError('When the scheduling mode is set to manual, '
                                 'the RB and MCS parameters are required.')

            # If scheduling mode is set to Dynamic, the RB and MCS parameters
            # will be ignored by set_scheduling_mode.
            self.set_scheduling_mode(bts_index, config.scheduling_mode,
                                     config.dl_mcs, config.ul_mcs,
                                     config.dl_rbs, config.ul_rbs)

        # This variable stores a boolean value so the following is needed to
        # differentiate False from None
        if config.mac_padding is not None:
            self.set_mac_padding(bts_index, config.mac_padding)

        if config.cfi:
            self.set_cfi(bts_index, config.cfi)

        if config.paging_cycle:
            self.set_paging_cycle(bts_index, config.paging_cycle)

        if config.phich:
            self.set_phich_resource(bts_index, config.phich)

    def configure_lte_bts(self, config, bts_index=0):
        """ Commands the equipment to setup an LTE base station with the
        required configuration.

        For some simulator (for example cmx500), cdrx has to be configured after
        similation started in some cases (for example nsa test). Also, it is
        fine for this simulator to configure cdrx after the cell starts. Thus,
        split the method of configure_lte_bts to two parts so that the cdrx
        configuration could be done after the simulation starts for some
        simulator.

        Args:
            config: an LteSimulation.BtsConfig object.
            bts_index: the base station number.
        """

        self.configure_lte_bts_base(config, bts_index=bts_index)
        if config.drx_connected_mode:
            self.set_drx_connected_mode(bts_index, config.drx_connected_mode)

            if config.drx_on_duration_timer:
                self.set_drx_on_duration_timer(bts_index,
                                               config.drx_on_duration_timer)

            if config.drx_inactivity_timer:
                self.set_drx_inactivity_timer(bts_index,
                                              config.drx_inactivity_timer)

            if config.drx_retransmission_timer:
                self.set_drx_retransmission_timer(
                    bts_index, config.drx_retransmission_timer)

            if config.drx_long_cycle:
                self.set_drx_long_cycle(bts_index, config.drx_long_cycle)

            if config.drx_long_cycle_offset is not None:
                self.set_drx_long_cycle_offset(bts_index,
                                               config.drx_long_cycle_offset)

    def configure_nr_bts(self, config, bts_index=1):
        """ Commands the equipment to setup an LTE base station with the
        required configuration.

        Args:
            config: an LteSimulation.BtsConfig object.
            bts_index: the base station number.
        """
        if config.tracking_area:
            self.set_tracking_area(bts_index, config.tracking_area)

        if config.band:
            self.set_band(bts_index, config.band)

        if config.nr_arfcn:
            self.set_downlink_channel_number(bts_index, config.nr_arfcn)

        if config.bandwidth:
            self.set_bandwidth(bts_index, config.bandwidth)

        if config.mimo_mode:
            self.set_mimo_mode(bts_index, config.mimo_mode)

        if config.scheduling_mode:

            if (config.scheduling_mode
                    == cellular_lib.LteSimulation.SchedulingMode.STATIC
                    and not (config.dl_rbs and config.ul_rbs and config.dl_mcs
                             and config.ul_mcs)):
                raise ValueError('When the scheduling mode is set to manual, '
                                 'the RB and MCS parameters are required.')

            # If scheduling mode is set to Dynamic, the RB and MCS parameters
            # will be ignored by set_scheduling_mode.
            self.set_scheduling_mode(bts_index, config.scheduling_mode,
                                     config.dl_mcs, config.ul_mcs,
                                     config.dl_rbs, config.ul_rbs)
        if config.mac_padding is not None:
            self.set_mac_padding(bts_index, config.mac_padding)

    def set_lte_rrc_state_change_timer(self, enabled, time=10):
        """ Configures the LTE RRC state change timer.

        Args:
            enabled: a boolean indicating if the timer should be on or off.
            time: time in seconds for the timer to expire
        """
        raise NotImplementedError()

    def set_band(self, bts_index, band):
        """ Sets the band for the indicated base station.

        Args:
            bts_index: the base station number
            band: the new band
        """
        raise NotImplementedError()

    def set_input_power(self, bts_index, input_power):
        """ Sets the input power for the indicated base station.

        Args:
            bts_index: the base station number
            input_power: the new input power
        """
        raise NotImplementedError()

    def set_output_power(self, bts_index, output_power):
        """ Sets the output power for the indicated base station.

        Args:
            bts_index: the base station number
            output_power: the new output power
        """
        raise NotImplementedError()

    def set_tdd_config(self, bts_index, tdd_config):
        """ Sets the tdd configuration number for the indicated base station.

        Args:
            bts_index: the base station number
            tdd_config: the new tdd configuration number
        """
        raise NotImplementedError()

    def set_ssf_config(self, bts_index, ssf_config):
        """ Sets the Special Sub-Frame config number for the indicated
        base station.

        Args:
            bts_index: the base station number
            ssf_config: the new ssf config number
        """
        raise NotImplementedError()

    def set_bandwidth(self, bts_index, bandwidth):
        """ Sets the bandwidth for the indicated base station.

        Args:
            bts_index: the base station number
            bandwidth: the new bandwidth
        """
        raise NotImplementedError()

    def set_downlink_channel_number(self, bts_index, channel_number):
        """ Sets the downlink channel number for the indicated base station.

        Args:
            bts_index: the base station number
            channel_number: the new channel number
        """
        raise NotImplementedError()

    def set_mimo_mode(self, bts_index, mimo_mode):
        """ Sets the mimo mode for the indicated base station.

        Args:
            bts_index: the base station number
            mimo_mode: the new mimo mode
        """
        raise NotImplementedError()

    def set_transmission_mode(self, bts_index, transmission_mode):
        """ Sets the transmission mode for the indicated base station.

        Args:
            bts_index: the base station number
            transmission_mode: the new transmission mode
        """
        raise NotImplementedError()

    def set_scheduling_mode(self, bts_index, scheduling_mode, mcs_dl, mcs_ul,
                            nrb_dl, nrb_ul):
        """ Sets the scheduling mode for the indicated base station.

        Args:
            bts_index: the base station number
            scheduling_mode: the new scheduling mode
            mcs_dl: Downlink MCS (only for STATIC scheduling)
            mcs_ul: Uplink MCS (only for STATIC scheduling)
            nrb_dl: Number of RBs for downlink (only for STATIC scheduling)
            nrb_ul: Number of RBs for uplink (only for STATIC scheduling)
        """
        raise NotImplementedError()

    def set_dl_256_qam_enabled(self, bts_index, enabled):
        """ Determines what MCS table should be used for the downlink.

        Args:
            bts_index: the base station number
            enabled: whether 256 QAM should be used
        """
        raise NotImplementedError()

    def set_ul_64_qam_enabled(self, bts_index, enabled):
        """ Determines what MCS table should be used for the uplink.

        Args:
            bts_index: the base station number
            enabled: whether 64 QAM should be used
        """
        raise NotImplementedError()

    def set_mac_padding(self, bts_index, mac_padding):
        """ Enables or disables MAC padding in the indicated base station.

        Args:
            bts_index: the base station number
            mac_padding: the new MAC padding setting
        """
        raise NotImplementedError()

    def set_cfi(self, bts_index, cfi):
        """ Sets the Channel Format Indicator for the indicated base station.

        Args:
            bts_index: the base station number
            cfi: the new CFI setting
        """
        raise NotImplementedError()

    def set_paging_cycle(self, bts_index, cycle_duration):
        """ Sets the paging cycle duration for the indicated base station.

        Args:
            bts_index: the base station number
            cycle_duration: the new paging cycle duration in milliseconds
        """
        raise NotImplementedError()

    def set_phich_resource(self, bts_index, phich):
        """ Sets the PHICH Resource setting for the indicated base station.

        Args:
            bts_index: the base station number
            phich: the new PHICH resource setting
        """
        raise NotImplementedError()

    def set_drx_connected_mode(self, bts_index, active):
        """ Sets the time interval to wait before entering DRX mode

        Args:
            bts_index: the base station number
            active: Boolean indicating whether cDRX mode
                is active
        """
        raise NotImplementedError()

    def set_drx_on_duration_timer(self, bts_index, timer):
        """ Sets the amount of PDCCH subframes to wait for data after
            waking up from a DRX cycle

        Args:
            bts_index: the base station number
            timer: Number of PDCCH subframes to wait and check for user data
                after waking from the DRX cycle
        """
        raise NotImplementedError()

    def set_drx_inactivity_timer(self, bts_index, timer):
        """ Sets the number of PDCCH subframes to wait before entering DRX mode

        Args:
            bts_index: the base station number
            timer: The amount of time to wait before entering DRX mode
        """
        raise NotImplementedError()

    def set_drx_retransmission_timer(self, bts_index, timer):
        """ Sets the number of consecutive PDCCH subframes to wait
        for retransmission

        Args:
            bts_index: the base station number
            timer: Number of PDCCH subframes to remain active

        """
        raise NotImplementedError()

    def set_drx_long_cycle(self, bts_index, cycle):
        """ Sets the amount of subframes representing a DRX long cycle.

        Args:
            bts_index: the base station number
            cycle: The amount of subframes representing one long DRX cycle.
                One cycle consists of DRX sleep + DRX on duration
        """
        raise NotImplementedError()

    def set_drx_long_cycle_offset(self, bts_index, offset):
        """ Sets the offset used to determine the subframe number
        to begin the long drx cycle

        Args:
            bts_index: the base station number
            offset: Number in range 0 to (long cycle - 1)
        """
        raise NotImplementedError()

    def set_tracking_area(self, bts_index, tac):
        """ Assigns the cell to a specific tracking area.

        Args:
            tac: the unique tac to assign the cell to.
        """
        raise NotImplementedError()

    def set_apn(self, apn):
        """ Configures the callbox network Access Point Name.

        Args:
            apn: the APN name
        """
        raise NotImplementedError()

    def set_ip_type(self, ip_type):
        """ Configures the callbox network IP type.

        Args:
            ip_type: the network type to use.
        """
        raise NotImplementedError()

    def set_mtu(self, mtu):
        """ Configures the callbox network Maximum Transmission Unit.

        Args:
            mtu: the MTU size.
        """
        raise NotImplementedError()

    def lte_attach_secondary_carriers(self, ue_capability_enquiry):
        """ Activates the secondary carriers for CA. Requires the DUT to be
        attached to the primary carrier first.

        Args:
            ue_capability_enquiry: UE capability enquiry message to be sent to
        the UE before starting carrier aggregation.
        """
        raise NotImplementedError()

    def wait_until_attached(self, timeout=120):
        """ Waits until the DUT is attached to the primary carrier.

        Args:
            timeout: after this amount of time the method will raise a
                CellularSimulatorError exception. Default is 120 seconds.
        """
        raise NotImplementedError()

    def wait_until_communication_state(self, timeout=120):
        """ Waits until the DUT is in Communication state.

        Args:
            timeout: after this amount of time the method will raise a
                CellularSimulatorError exception. Default is 120 seconds.
        """
        raise NotImplementedError()

    def wait_until_idle_state(self, timeout=120):
        """ Waits until the DUT is in Idle state.

        Args:
            timeout: after this amount of time the method will raise a
                CellularSimulatorError exception. Default is 120 seconds.
        """
        raise NotImplementedError()

    def wait_until_quiet(self, timeout=120):
        """Waits for all pending operations to finish on the simulator.

        Args:
            timeout: after this amount of time the method will raise a
                CellularSimulatorError exception. Default is 120 seconds.
        """
        raise NotImplementedError()

    def detach(self):
        """ Turns off all the base stations so the DUT loose connection."""
        raise NotImplementedError()

    def stop(self):
        """ Stops current simulation. After calling this method, the simulator
        will need to be set up again. """
        raise NotImplementedError()

    def start_data_traffic(self):
        """ Starts transmitting data from the instrument to the DUT. """
        raise NotImplementedError()

    def stop_data_traffic(self):
        """ Stops transmitting data from the instrument to the DUT. """
        raise NotImplementedError()

    def get_measured_pusch_power(self):
        """ Queries PUSCH power measured at the callbox.

        Returns:
            The PUSCH power in the primary input port.
        """
        raise NotImplementedError()

    def send_sms(self, message):
        """ Sends an SMS message to the DUT.

        Args:
            message: the SMS message to send.
        """
        raise NotImplementedError()


class CellularSimulatorError(Exception):
    """ Exceptions thrown when the cellular equipment is unreachable or it
    returns an error after receiving a command. """