#!/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. """