# Lint as: python2, python3
# Copyright 2015 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import threading
import time

from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros.network import interface
from autotest_lib.server.cros.network import expected_performance_results
from autotest_lib.client.common_lib.cros.network import ping_runner
from autotest_lib.server.cros.network import perf_test_manager as perf_manager
from autotest_lib.server.cros.network import wifi_cell_perf_test_base

from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import \
     BluetoothAdapterQuickTests
from autotest_lib.server.cros.bluetooth.bluetooth_adapter_audio_tests import \
     BluetoothAdapterAudioTests
from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import A2DP


class network_WiFi_BluetoothStreamPerf(
        wifi_cell_perf_test_base.WiFiCellPerfTestBase,
        BluetoothAdapterQuickTests, BluetoothAdapterAudioTests):
    """Test maximal achievable bandwidth on several channels per band.

    Conducts a performance test for a set of specified router configurations
    and reports results as keyval pairs.

    """

    test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator
    batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator

    version = 1

    PERF_TEST_TYPES = [
            perf_manager.PerfTestTypes.TEST_TYPE_TCP_TX,
            perf_manager.PerfTestTypes.TEST_TYPE_TCP_RX,
            perf_manager.PerfTestTypes.TEST_TYPE_TCP_BIDIRECTIONAL,
            perf_manager.PerfTestTypes.TEST_TYPE_UDP_TX,
            # TODO(b:251380100): investigate unstable results for UDP_RX
            # perf_manager.PerfTestTypes.TEST_TYPE_UDP_RX,
            perf_manager.PerfTestTypes.TEST_TYPE_UDP_BIDIRECTIONAL,
    ]

    BT_STREAMING_DURATION = 60

    def parse_additional_arguments(self, commandline_args, additional_params):
        """Hook into super class to take control files parameters.

        @param commandline_args dict of parsed parameters from the autotest.
        @param additional_params list of HostapConfig objects.

        """
        self._should_required = 'should' in commandline_args

        super(network_WiFi_BluetoothStreamPerf, self).parse_additional_arguments(
                commandline_args)

        self._ap_configs, self._use_iperf = additional_params

    def verify_result(self, result_drop, should_expected_drop,
                      must_expected_drop, board_expected_drop, test_type,
                      failed_test_types, ap_config_tag, bt_tag):
        """Verfiy that performance test result passes the must and should
        throughput requirements.

        @param result: the throughput result object
        @param must_expected_throughput: the min must expected throughput
        @param should_expected_throughput: the min should expected throughput
        @param test_type: the performance_test_types test type
        @param failed_test_types: a set of failed test_types
        @param power_save: powersaving configuration
        @param ap_config: the AP configuration
        @param bt_tag: string for BT operation

        """
        must_drop_failed = False
        should_drop_failed = False

        # If the must requirement is smaller than our maximum expectation for a
        # board, use the board specific maximum expectation instead of the must
        # requirement.
        if board_expected_drop and board_expected_drop > must_expected_drop:
            must_expected_drop = board_expected_drop

        if result_drop > must_expected_drop:
            logging.error(
                    'Throughput drop is too big for %s. Expected (must) %d %%, got %d.',
                    test_type, must_expected_drop, result_drop)
            must_drop_failed = True
        if result_drop > should_expected_drop:
            if self._should_required:
                logging.error(
                        'Throughput drop is too big for %s. Expected (should) %d %%, got %d.',
                        test_type, should_expected_drop,
                        result_drop)
                should_drop_failed = True
            else:
                logging.info(
                        'Throughput drop is bigger then expectation for %s. Expected (should) %d %%, got %d.',
                        test_type, should_expected_drop,
                        result_drop)
        if must_drop_failed or should_drop_failed:
            failed_test_type_list = [
                    '[test_type=%s' % test_type,
                    'ap_config_tag=%s' % ap_config_tag,
                    'bt_tag=%s' % bt_tag,
                    'measured_drop=%d' % result_drop
            ]
            if must_drop_failed:
                failed_test_type_list.append(
                        'must_expected_drop_failed=%d' %
                        must_expected_drop)
            elif should_drop_failed:
                failed_test_type_list.append(
                        'should_expected_drop_failed=%d' %
                        should_expected_drop)
            failed_test_types.add(', '.join(failed_test_type_list) + ']')

    def test_one(self, manager, session, config, test_type, failed_test_types, ap_config, ap_config_tag, bt_tag):
        """Run one iteration of wifi testing.

        @param manager: a PerfTestManager instance
        @param session: NetperfSession or IperfSession session
        @param config: NetperfConfig or IperfConfig config
        @param test_type: the performance_test_types test type
        @param failed_test_types: a set of failed test_types
        @param ap_config: the AP configuration
        @param ap_config_tag: string for AP configuration
        @param bt_tag: string for BT operation

        """
        get_ping_config = lambda period: ping_runner.PingConfig(
                self.context.get_wifi_addr(),
                interval=1,
                count=period,
                source_iface=self.context.client.wifi_if)

        logging.info('testing config %s, ap_config %s, BT:%s',
                     test_type, ap_config_tag, bt_tag)
        test_str = '_'.join([ap_config_tag, bt_tag])
        time.sleep(1)

        # Record the signal level.
        signal_level = self.context.client.wifi_signal_level
        signal_description = '_'.join(['signal', test_str])
        self.write_perf_keyval({signal_description: signal_level})

        # Run perf tool and log the results.
        results = session.run(config)
        if not results:
            logging.error('Failed to take measurement for %s',
                          config.test_type)
            return
        values = [result.throughput for result in results]
        self.output_perf_value(config.test_type + '_' + bt_tag,
                               values,
                               units='Mbps',
                               higher_is_better=True,
                               graph=ap_config_tag)
        result = manager.get_result(results, config)
        self.write_perf_keyval(
                result.get_keyval(
                        prefix='_'.join([config.test_type, test_str])))

        # Log the standard deviation
        throughput_dev = result.throughput_dev
        self.output_perf_value(config.test_type + '_' + bt_tag + '_dev',
                               throughput_dev,
                               units='Mbps',
                               higher_is_better=False,
                               graph=ap_config_tag + '_dev')
        self.write_perf_keyval(
            {'_'.join([config.test_type, test_str, 'dev']): throughput_dev})

        # Log the drop in throughput compared with the 'BT_disconnected'
        # baseline. Only positive values are valid. Report the drop as a
        # whole integer percentage of (base_through-through)/base_through.
        if bt_tag == BluetoothAdapterAudioTests.CONNECTION_STATE_DISCONNECTED:
            self.base_through = result.throughput
        elif self.base_through > 0:
            expected_drop = expected_performance_results.get_expected_wifibt_coex_throughput_drop(
                    test_type, ap_config, bt_tag)

            board_expected_drop = expected_performance_results.get_board_max_wifibt_coex_throughput_drop_expectation(
                test_type, self.context.client.board, ap_config, bt_tag)

            drop = int( (self.base_through - result.throughput) * 100 /
                        self.base_through)
            self.output_perf_value(test_type + '_' + bt_tag + '_drop',
                                   drop,
                                   units='percent_drop',
                                   higher_is_better=False,
                                   graph=ap_config_tag + '_drop')
            self.verify_result(drop, expected_drop[0],
                               expected_drop[1], board_expected_drop, test_type,
                               failed_test_types, ap_config_tag, bt_tag)
            self.write_perf_keyval(
                    {'_'.join([config.test_type, test_str, 'drop']): drop})
            logging.info('logging drop value as %d%%', drop)

        # Test latency with ping.
        result_ping = self.context.client.ping(get_ping_config(3))
        self.write_perf_keyval(
            { '_'.join(['ping', test_str]): result_ping.avg_latency })
        logging.info('Ping statistics with %s: %r', bt_tag, result_ping)

        return failed_test_types

    def pair_audio_device(self, device):
        """Pair an audio device pre-test to simplify later re-connection"""
        self.test_device_set_discoverable(device, True)
        self.test_discover_device(device.address)
        self.test_pairing(device.address, device.pin, trusted=True)
        device.SetTrustedByRemoteAddress(self.bluetooth_facade.address)
        self.test_disconnection_by_adapter(device.address)

    def do_audio_test(self, device):
        """Run the body of the audio test"""
        self.test_a2dp_sinewaves(device, A2DP, self.BT_STREAMING_DURATION)

    def do_run(self, ap_config, manager, power_save, governor):
        """Run a single set of perf tests, for a given AP and DUT config.

        @param ap_config: the AP configuration that is being used
        @param manager: a PerfTestManager instance
        @param power_save: whether or not to use power-save mode on the DUT
                           (boolean)
        @ return set of failed configs
        """
        governor_name = self.setup_governor(governor)
        # If CPU governor is already set to self._governor, don't
        # perform the run twice
        if governor_name == self._governor:
            return

        failed_test_types = set()

        self.context.client.powersave_switch(power_save)
        ps_tag = 'PS%s' % ('on' if power_save else 'off')
        governor_tag = 'governor-%s' % governor_name
        ap_config_tag = '_'.join([ap_config.perf_loggable_description,
                                  ps_tag, governor_tag])

        device = self.bt_device

        for test_type in self.PERF_TEST_TYPES:
            config = manager.get_config(test_type, self._is_openwrt)
            pcap_lan_iface = interface.Interface(self._pcap_lan_iface_name,
                                                 self.context.pcap_host.host)
            session = manager.get_session(test_type,
                                          self.context.client,
                                          self.context.pcap_host,
                                          peer_device_interface=pcap_lan_iface)

            session.MEASUREMENT_MAX_SAMPLES = int(self.BT_STREAMING_DURATION / 10)

            self.base_through = 0
            self.test_one(
                    manager, session, config, test_type, None, ap_config,
                    ap_config_tag,
                    BluetoothAdapterAudioTests.CONNECTION_STATE_DISCONNECTED)

            self.test_connection_by_device(device)
            failed_test_types.update(
                    self.test_one(
                            manager, session, config, test_type,
                            failed_test_types, ap_config, ap_config_tag,
                            BluetoothAdapterAudioTests.CONNECTION_STATE_CONNECTED))

            # Start playing audio in background
            audio_thread = threading.Thread(target=self.do_audio_test,
                                            args=(device, ))
            audio_thread.start()
            failed_test_types.update(
                    self.test_one(
                            manager, session, config, test_type,
                            failed_test_types, ap_config, ap_config_tag,
                            BluetoothAdapterAudioTests.CONNECTION_STATE_STREAMING))

            # Wait for audio thread to complete
            audio_thread.join()

            self.test_disconnection_by_adapter(device.address)
            failed_test_types.update(
                    self.test_one(
                            manager, session, config, test_type,
                            failed_test_types, ap_config, ap_config_tag,
                            BluetoothAdapterAudioTests.CONNECTION_STATE_DISCONNECTED_AGAIN))

        if governor:
            self.restore_scaling_governors()

        return failed_test_types

    @test_wrapper('Coex tests', devices={'BLUETOOTH_AUDIO': 1})
    def coex_test(self):
        """Test body."""
        start_time = time.time()
        self.bt_device = self.devices['BLUETOOTH_AUDIO'][0]
        self.initialize_bluetooth_audio(self.bt_device, A2DP)
        self.pair_audio_device(self.bt_device)

        high_drop_tests = self.configure_and_run_tests()

        self.cleanup_bluetooth_audio(self.bt_device, A2DP)
        end_time = time.time()
        logging.info('Running time %0.1f seconds.', end_time - start_time)
        if len(high_drop_tests) != 0:
            high_drop_tags = list(high_drop_tests)
            raise error.TestFail(
                    'Throughput drop level too high for test type(s): %s' %
                    ', '.join(high_drop_tags))

    @batch_wrapper('Coex batch')
    def coex_health_batch_run(self, num_iterations=1, test_name=None):
        """Run the bluetooth coex health test batch or a specific given test.

        @param num_iterations: how many iterations to run
        @param test_name: specific test to run otherwise None to run the
                whole batch
        """
        self.coex_test()

    def run_once(self, host, test_name=None, args_dict=None):
        self.host = host

        self.quick_test_init(host, use_btpeer=True, args_dict=args_dict)
        self.coex_health_batch_run(test_name=test_name)
        self.quick_test_cleanup()