To integrate Monsoon PMC to Sony speaker relayandroid-o-iot-preview-5o-iot-preview-5

Bug: 64698476
Test: Manual run on local computer
Change-Id: I093ea7dd38ed0488a0d8d52b01cd7b381c0adb4b

1 files changed, 85 insertions, 33 deletions

diff --git a/acts/tests/google/bt/power/A2dpPowerTest.py b/acts/tests/google/bt/power/A2dpPowerTest.py
index 5b8d38799b..56b9850f6d 100644
--- a/acts/tests/google/bt/power/A2dpPowerTest.py
+++ b/acts/tests/google/bt/power/A2dpPowerTest.py
@@ -35,6 +35,7 @@ from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
 from acts.test_utils.bt.PowerBaseTest import PowerBaseTest
 from acts.test_utils.bt.bt_test_utils import bluetooth_enabled_check
 from acts.test_utils.bt.bt_test_utils import disable_bluetooth
+from acts.controllers.relay_lib.sony_xb2_speaker import SonyXB2Speaker
 
 
 def push_file_to_device(ad, file_path, device_path, config_path):
@@ -98,12 +99,8 @@ class A2dpPowerTest(PowerBaseTest):
     # LDAC playback quality constant for the codecs other than LDAC
     LDACBT_NONE = 0
 
-    def setup_class(self):
-
-        self.ad = self.android_devices[0]
-
+    def _pair_by_config(self):
         bt_device_address = self.user_params["bt_device_address"]
-
         # Push the bt_config.conf file to Android device
         # if it is specified in config file
         # so it can pair and connect automatically.
@@ -123,32 +120,12 @@ class A2dpPowerTest(PowerBaseTest):
             if not push_file_to_device(
                     self.ad, bt_config_path, bt_conf_path_dut,
                     self.user_params[Config.key_config_path]):
-                self.log.error("Unable to push file {} to DUT.".format(
-                    bt_config_path))
+                self.log.error(
+                    "Unable to push file {} to DUT.".format(bt_config_path))
 
             self.log.info("Reboot")
             self.ad.reboot()
 
-        # Add music files to the Android device
-        music_path_dut = "/sdcard/Music/"
-        self.cd_quality_music_file = self.user_params["cd_quality_music_file"]
-        self.log.info("Push CD quality music file {}".format(
-            self.cd_quality_music_file))
-        if not push_file_to_device(self.ad, self.cd_quality_music_file,
-                                   music_path_dut,
-                                   self.user_params[Config.key_config_path]):
-            self.log.error("Unable to push file {} to DUT.".format(
-                self.cd_quality_music_file))
-
-        self.hi_res_music_file = self.user_params["hi_res_music_file"]
-        self.log.info("Push Hi Res quality music file {}".format(
-            self.hi_res_music_file))
-        if not push_file_to_device(self.ad, self.hi_res_music_file,
-                                   music_path_dut,
-                                   self.user_params[Config.key_config_path]):
-            self.log.error("Unable to find file {}.".format(
-                self.hi_res_music_file))
-
         if not bluetooth_enabled_check(self.ad):
             self.log.error("Failed to enable Bluetooth on DUT")
             return False
@@ -165,21 +142,96 @@ class A2dpPowerTest(PowerBaseTest):
                 break
 
             try:
-                self.ad.droid.bluetoothConnectBonded(self.user_params[
-                    "bt_device_address"])
+                self.ad.droid.bluetoothConnectBonded(
+                    self.user_params["bt_device_address"])
             except Exception as err:
                 self.log.error(
                     "Failed to connect bonded device. Err: {}".format(err))
 
-        if not result:
+        if result is False:
             self.log.error("No headset is connected")
             return False
+        return True
+
+    def _discover_and_pair(self):
+        self.ad.droid.bluetoothStartDiscovery()
+        time.sleep(5)
+        self.ad.droid.bluetoothCancelDiscovery()
+        for device in self.ad.droid.bluetoothGetDiscoveredDevices():
+            if device['address'] == self.a2dp_speaker.mac_address:
+                self.ad.droid.bluetoothDiscoverAndBond(
+                    self.a2dp_speaker.mac_address)
+                end_time = time.time() + 20
+                self.log.info("Verifying devices are bonded")
+                while (time.time() < end_time):
+                    bonded_devices = self.ad.droid.bluetoothGetBondedDevices()
+                    for d in bonded_devices:
+                        if d['address'] == self.a2dp_speaker.mac_address:
+                            self.log.info("Successfully bonded to device.")
+                            self.log.info("XB2 Bonded devices:\n{}".format(
+                                bonded_devices))
+                            return True
+        return False
+
+    def setup_class(self):
+        self.ad = self.android_devices[0]
+        self.ad.droid.bluetoothFactoryReset()
+        # Factory reset requires a short delay to take effect
+        time.sleep(3)
+
+        # Determine if we have a relay-based device
+        self.a2dp_speaker = None
+        if self.relay_devices[0]:
+            self.a2dp_speaker = self.relay_devices[0]
+            self.a2dp_speaker.setup()
+            # The device may be on, enter pairing mode.
+            self.a2dp_speaker.enter_pairing_mode()
+            if self._discover_and_pair() is False:
+                # The device is probably off, turn it on
+                self.a2dp_speaker.power_on()
+                time.sleep(0.25)
+                self.a2dp_speaker.enter_pairing_mode()
+                if self._discover_and_pair() is False:
+                    self.log.info("Unable to pair to relay based device.")
+                    return False
+            if len(self.ad.droid.bluetoothGetConnectedDevices()) == 0:
+                self.log.info("Not connected to relay based device.")
+                return False
+        else:
+            # No relay-based device used config
+            if self._pair_by_config() is False:
+                return False
+
+        # Add music files to the Android device
+        music_path_dut = "/sdcard/Music/"
+        self.cd_quality_music_file = self.user_params["cd_quality_music_file"]
+        self.log.info(
+            "Push CD quality music file {}".format(self.cd_quality_music_file))
+        if not push_file_to_device(self.ad, self.cd_quality_music_file,
+                                   music_path_dut,
+                                   self.user_params[Config.key_config_path]):
+            self.log.error("Unable to push file {} to DUT.".format(
+                self.cd_quality_music_file))
+
+        self.hi_res_music_file = self.user_params["hi_res_music_file"]
+        self.log.info(
+            "Push Hi Res quality music file {}".format(self.hi_res_music_file))
+        if not push_file_to_device(self.ad, self.hi_res_music_file,
+                                   music_path_dut,
+                                   self.user_params[Config.key_config_path]):
+            self.log.error(
+                "Unable to find file {}.".format(self.hi_res_music_file))
 
         # Then do other power testing setup in the base class
         # including start PMC etc
         super(A2dpPowerTest, self).setup_class()
         return True
 
+    def teardown_class(self):
+        if self.a2dp_speaker is not None:
+            self.a2dp_speaker.power_off()
+            self.a2dp_speaker.clean_up()
+
     def _main_power_test_function_for_codec(self,
                                             codec_type,
                                             sample_rate,
@@ -220,8 +272,8 @@ class A2dpPowerTest(PowerBaseTest):
             # Get the base name of music file
             music_name = os.path.basename(music_file)
             self.music_url = "file:///sdcard/Music/{}".format(music_name)
-            play_time = self.MEASURE_TIME + self.DELAY_MEASURE_TIME
-            +self.DELAY_STOP_TIME
+            play_time = (self.MEASURE_TIME + self.DELAY_MEASURE_TIME +
+                         self.DELAY_STOP_TIME)
             play_msg = "%s --es MusicURL %s --es PlayTime %d" % (
                 self.PMC_BASE_CMD, self.music_url, play_time)
 
@@ -234,7 +286,7 @@ class A2dpPowerTest(PowerBaseTest):
                                                           bits_per_sample)
                 if codec_type == self.CODEC_LDAC:
                     msg = "%s --es LdacPlaybackQuality %d" % (
-                        play_msg, ldac_playback_quality)
+                        codec_msg, ldac_playback_quality)
                 else:
                     msg = codec_msg