Merge "Adding audio analysis to a2dp tests." am: 0e09a33199 am: 7e6f18e984
am: 7d65003e73
Change-Id: Ie4a9fc93a208780e26912ace20e739f1b9ecaa1c
diff --git a/acts/framework/acts/test_utils/coex/CoexBaseTest.py b/acts/framework/acts/test_utils/coex/CoexBaseTest.py
index b209fe4..a37995a 100644
--- a/acts/framework/acts/test_utils/coex/CoexBaseTest.py
+++ b/acts/framework/acts/test_utils/coex/CoexBaseTest.py
@@ -29,7 +29,6 @@
from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test
from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
from acts.test_utils.coex.coex_test_utils import A2dpDumpsysParser
-from acts.test_utils.coex.audio_test_utils import SshAudioCapture
from acts.test_utils.coex.coex_test_utils import check_wifi_status
from acts.test_utils.coex.coex_test_utils import (
collect_bluetooth_manager_dumpsys_logs)
@@ -116,7 +115,7 @@
if self.audio_params["music_file"]:
self.music_file_to_play = push_music_to_android_device(
self.pri_ad, self.audio_params)
- if not self.music_file_play:
+ if not self.music_file_to_play:
self.log.error("Music file push failed.")
return False
else:
@@ -131,9 +130,10 @@
self.dev_list = {}
self.iperf_variables = self.IperfVariables(self.current_test_name)
self.a2dp_dumpsys = A2dpDumpsysParser()
- log_path = os.path.join(self.pri_ad.log_path, self.current_test_name)
- create_dir(log_path)
- self.json_file = os.path.join(log_path, "test_results.json")
+ self.log_path = os.path.join(self.pri_ad.log_path,
+ self.current_test_name)
+ create_dir(self.log_path)
+ self.json_file = os.path.join(self.log_path, "test_results.json")
for a in self.android_devices:
a.ed.clear_all_events()
if not wifi_connection_check(self.pri_ad, self.network["SSID"]):
@@ -143,18 +143,14 @@
self.log.error("Failed to enable bluetooth")
return False
if hasattr(self, "required_devices"):
- if "discovery" in self.current_test_name or (
+ if ("discovery" in self.current_test_name or
"inquiry" in self.current_test_name or
- ("ble" in self.current_test_name)):
+ "ble" in self.current_test_name):
self.create_android_relay_object()
else:
self.log.warning("required_devices is not given in config file")
- if "a2dp_streaming" in self.current_test_name:
- self.audio = SshAudioCapture(self.audio_params, log_path)
def teardown_test(self):
- if "a2dp_streaming" in self.current_test_name:
- self.audio.terminate_and_store_audio_results()
self.parsing_results()
self.teardown_result()
with open(self.json_file, 'a') as results_file:
@@ -192,9 +188,9 @@
"""Destroys android device object and relay device object if required
devices has android device and relay device."""
if hasattr(self, "required_devices"):
- if "discovery" in self.current_test_name or (
+ if ("discovery" in self.current_test_name or
"inquiry" in self.current_test_name or
- ("ble" in self.current_test_name)):
+ "ble" in self.current_test_name):
if hasattr(self, "inquiry_devices"):
for device in range(len(self.inquiry_devices)):
inquiry_device = self.inquiry_devices[device]
@@ -217,6 +213,8 @@
self.pri_ad, self.current_test_name)
self.result["a2dp_packet_drop"] = (
self.a2dp_dumpsys.parse(file_path))
+ if self.result["a2dp_packet_drop"] == 0:
+ self.result["a2dp_packet_drop"] = None
def start_iperf_server_on_shell(self, server_port):
"""Starts iperf server on android device with specified.
diff --git a/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py b/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py
index 506caec..7293533 100644
--- a/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py
+++ b/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py
@@ -20,7 +20,7 @@
from acts.test_utils.bt.bt_test_utils import disable_bluetooth
from acts.test_utils.coex.CoexBaseTest import CoexBaseTest
-from acts.test_utils.coex.coex_test_utils import bokeh_plot
+from acts.test_utils.coex.coex_test_utils import bokeh_chart_plot
from acts.test_utils.coex.coex_test_utils import (
collect_bluetooth_manager_dumpsys_logs)
from acts.test_utils.coex.coex_test_utils import multithread_func
@@ -72,8 +72,6 @@
super().setup_test()
def teardown_test(self):
- if "a2dp_streaming" in self.current_test_name:
- self.audio.terminate_and_store_audio_results()
self.performance_baseline_check()
self.attenuators[self.num_atten - 1].set_atten(0)
for i in range(self.num_atten - 1):
@@ -114,10 +112,12 @@
if "a2dp_streaming" in self.current_test_name:
(self.rvr[bt_atten]["throughput_received"],
self.rvr[bt_atten]["a2dp_packet_drop"]) = (
- self.rvr_throughput(called_func))
+ self.rvr_throughput(bt_atten, called_func))
+ if all(x <= 0 for x in self.a2dp_dropped_list):
+ self.rvr[bt_atten]["a2dp_packet_drop"] = []
else:
self.rvr[bt_atten]["throughput_received"] = (
- self.rvr_throughput(called_func))
+ self.rvr_throughput(bt_atten, called_func))
self.rvr[bt_atten]["fixed_attenuation"] = (
self.test_params["fixed_attenuation"][str(
self.network["channel"])])
@@ -125,10 +125,11 @@
self.rvr["bt_attenuation"] = list(self.bt_attenuation_range)
return True
- def rvr_throughput(self, called_func=None):
+ def rvr_throughput(self, bt_atten, called_func=None):
"""Sets attenuation and runs the function passed.
Args:
+ bt_atten: Bluetooth attenuation.
called_func: Functions object to run parallely.
Returns:
@@ -137,6 +138,7 @@
self.iperf_received = []
self.iperf_variables.received = []
self.a2dp_dropped_list = []
+ self.rvr[bt_atten]["audio_artifacts"] = {}
for atten in self.attenuation_range:
self.log.info("Setting attenuation = {}".format(atten))
for i in range(self.num_atten - 1):
@@ -151,8 +153,11 @@
else:
self.run_iperf_and_get_result()
if "a2dp_streaming" in self.current_test_name:
+ analysis_path = self.audio.audio_quality_analysis(self.log_path)
+ with open(analysis_path) as f:
+ self.rvr[bt_atten]["audio_artifacts"][atten] = f.readline()
file_path = collect_bluetooth_manager_dumpsys_logs(
- self.pri_ad, self.current_test_name)
+ self.pri_ad, self.current_test_name)
self.a2dp_dropped_list.append(
self.a2dp_dumpsys.parse(file_path))
self.teardown_result()
@@ -196,7 +201,8 @@
"""
data_sets = {}
test_name = self.current_test_name
- x_label = 'Attenuation (dB)'
+ x_label = 'WIFI Attenuation (dB)'
+ y_label = []
legends = {}
fig_property = {
"title": test_name,
@@ -204,14 +210,14 @@
"linewidth": 3,
"markersize": 10
}
- y_label = []
+
for bt_atten in self.bt_attenuation_range:
data_sets[bt_atten] = {}
legends[bt_atten] = []
total_atten = self.total_attenuation(self.rvr[bt_atten])
y_label.insert(0, 'Throughput (Mbps)')
legends[bt_atten].insert(
- 0, str(self.current_test_name + (" @ {}dB".format(bt_atten))))
+ 0, str("BT Attenuation @ %sdB" % bt_atten))
data_sets[bt_atten]["attenuation"] = total_atten
data_sets[bt_atten]["throughput_received"] = (
self.rvr[bt_atten]["throughput_received"])
@@ -267,7 +273,7 @@
fig_property["y_label"] = y_label
output_file_path = os.path.join(self.pri_ad.log_path, test_name,
"attenuation_plot.html")
- bokeh_plot(
+ bokeh_chart_plot(
list(self.bt_attenuation_range),
data_sets,
legends,
diff --git a/acts/framework/acts/test_utils/coex/coex_test_utils.py b/acts/framework/acts/test_utils/coex/coex_test_utils.py
index 3949ee1..cafbdd5 100644
--- a/acts/framework/acts/test_utils/coex/coex_test_utils.py
+++ b/acts/framework/acts/test_utils/coex/coex_test_utils.py
@@ -18,11 +18,9 @@
import logging
import math
import os
-import pyaudio
import re
import subprocess
import time
-import wave
import xlsxwriter
from acts.controllers.ap_lib import hostapd_config
@@ -56,6 +54,7 @@
from acts.test_utils.tel.tel_test_utils import run_multithread_func
from acts.test_utils.tel.tel_test_utils import setup_droid_properties
from acts.test_utils.tel.tel_test_utils import wait_and_answer_call
+from acts.test_utils.wifi.wifi_power_test_utils import bokeh_plot
from acts.test_utils.wifi.wifi_test_utils import reset_wifi
from acts.test_utils.wifi.wifi_test_utils import wifi_connect
from acts.test_utils.wifi.wifi_test_utils import wifi_test_device_init
@@ -1150,7 +1149,7 @@
worksheet.write(row + 1, col + i, cell)
-def bokeh_plot(bt_attenuation_range,
+def bokeh_chart_plot(bt_attenuation_range,
data_sets,
legends,
fig_property,
@@ -1177,6 +1176,8 @@
'yellow', 'darkred', 'goldenrod'
]
plot = []
+ data = [[], []]
+ legend = []
for i in bt_attenuation_range:
if "Packet drop" in legends[i][0]:
plot_info = {0: "A2dp_packet_drop_plot", 1: "throughput_plot"}
@@ -1184,31 +1185,35 @@
plot_info = {0: "throughput_plot"}
for j in plot_info:
if "Packet drops" in legends[i][j]:
- plot_i_j = figure(
- plot_width=1000,
- plot_height=500,
- title=fig_property['title'],
- tools=TOOLS)
- plot_i_j.add_tools(
- bokeh_tools.WheelZoomTool(dimensions="width"))
- plot_i_j.add_tools(
- bokeh_tools.WheelZoomTool(dimensions="height"))
- plot_i_j.xaxis.axis_label = fig_property['x_label']
- plot_i_j.yaxis.axis_label = fig_property['y_label'][j]
- plot_i_j.legend.location = "top_right"
- plot_i_j.legend.click_policy = "hide"
- plot_i_j.title.text_font_size = {'value': '15pt'}
- plot_i_j.line(
- data_sets[i]["a2dp_attenuation"],
- data_sets[i]["a2dp_packet_drops"],
- legend=legends[i][j],
- line_width=3,
- color=colors[j])
- plot_i_j.circle(
- data_sets[i]["a2dp_attenuation"],
- data_sets[i]["a2dp_packet_drops"],
- legend=str(legends[i][j]),
- fill_color=colors[j])
+ if data_sets[i]["a2dp_packet_drops"]:
+ plot_i_j = figure(
+ plot_width=1000,
+ plot_height=500,
+ title=fig_property['title'],
+ tools=TOOLS)
+
+ plot_i_j.add_tools(
+ bokeh_tools.WheelZoomTool(dimensions="width"))
+ plot_i_j.add_tools(
+ bokeh_tools.WheelZoomTool(dimensions="height"))
+ plot_i_j.xaxis.axis_label = fig_property['x_label']
+ plot_i_j.yaxis.axis_label = fig_property['y_label'][j]
+ plot_i_j.legend.location = "top_right"
+ plot_i_j.legend.click_policy = "hide"
+ plot_i_j.title.text_font_size = {'value': '15pt'}
+
+ plot_i_j.line(
+ data_sets[i]["a2dp_attenuation"],
+ data_sets[i]["a2dp_packet_drops"],
+ legend=legends[i][j],
+ line_width=3,
+ color=colors[j])
+ plot_i_j.circle(
+ data_sets[i]["a2dp_attenuation"],
+ data_sets[i]["a2dp_packet_drops"],
+ legend=str(legends[i][j]),
+ fill_color=colors[j])
+ plot.append(plot_i_j)
elif "Performance Results" in legends[i][j]:
plot_i_j = figure(
plot_width=1000,
@@ -1224,7 +1229,9 @@
plot_i_j.legend.location = "top_right"
plot_i_j.legend.click_policy = "hide"
plot_i_j.title.text_font_size = {'value': '15pt'}
-
+ data[0].insert(0, data_sets[i]["attenuation"])
+ data[1].insert(0, data_sets[i]["throughput_received"])
+ legend.insert(0, legends[i][j + 1])
plot_i_j.line(
data_sets[i]["user_attenuation"],
data_sets[i]["user_throughput"],
@@ -1258,6 +1265,7 @@
color='#7570B3',
line_alpha=0.1,
fill_alpha=0.1)
+ plot.append(plot_i_j)
else:
plot_i_j = figure(
plot_width=1000,
@@ -1273,7 +1281,9 @@
plot_i_j.legend.location = "top_right"
plot_i_j.legend.click_policy = "hide"
plot_i_j.title.text_font_size = {'value': '15pt'}
-
+ data[0].insert(0, data_sets[i]["attenuation"])
+ data[1].insert(0, data_sets[i]["throughput_received"])
+ legend.insert(0, legends[i][j])
plot_i_j.line(
data_sets[i]["attenuation"],
data_sets[i]["throughput_received"],
@@ -1285,7 +1295,11 @@
data_sets[i]["throughput_received"],
legend=str(legends[i][j]),
fill_color=colors[j])
- plot.append(plot_i_j)
+ plot.append(plot_i_j)
+ fig_property['y_label'] = "Throughput (Mbps)"
+ all_plot = bokeh_plot(data, legend, fig_property, shaded_region=None,
+ output_file_path=None)
+ plot.insert(0, all_plot)
if output_file_path is not None:
output_file(output_file_path)
save(column(plot))
diff --git a/acts/tests/google/coex/functionality_tests/WlanWithA2dpFunctionalityTest.py b/acts/tests/google/coex/functionality_tests/WlanWithA2dpFunctionalityTest.py
index 7ec63f1..27f190b 100644
--- a/acts/tests/google/coex/functionality_tests/WlanWithA2dpFunctionalityTest.py
+++ b/acts/tests/google/coex/functionality_tests/WlanWithA2dpFunctionalityTest.py
@@ -26,6 +26,7 @@
from acts.test_utils.bt import BtEnum
from acts.test_utils.bt.bt_test_utils import clear_bonded_devices
+from acts.test_utils.coex.audio_test_utils import SshAudioCapture
from acts.test_utils.coex.CoexBaseTest import CoexBaseTest
from acts.test_utils.coex.coex_test_utils import avrcp_actions
from acts.test_utils.coex.coex_test_utils import connect_ble
@@ -38,6 +39,7 @@
from acts.test_utils.coex.coex_test_utils import start_fping
from acts.test_utils.coex.coex_test_utils import toggle_screen_state
+
BLUETOOTH_WAIT_TIME = 2
@@ -53,6 +55,8 @@
def setup_test(self):
super().setup_test()
+ if "a2dp_streaming" in self.current_test_name:
+ self.audio = SshAudioCapture(self.audio_params, self.log_path)
self.audio_receiver.enter_pairing_mode()
time.sleep(5) #Wait until device goes into pairing mode.
if not pair_and_connect_headset(
@@ -64,6 +68,11 @@
def teardown_test(self):
clear_bonded_devices(self.pri_ad)
self.audio_receiver.clean_up()
+ if "a2dp_streaming" in self.current_test_name:
+ analysis_path = self.audio.audio_quality_analysis(self.log_path)
+ with open(analysis_path) as f:
+ self.result["audio_artifacts"] = f.readline()
+ self.audio.terminate_and_store_audio_results()
super().teardown_test()
def connect_disconnect_a2dp_headset(self):
diff --git a/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py b/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py
index 0d17a7b..294b737 100644
--- a/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py
+++ b/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py
@@ -26,6 +26,7 @@
from acts.test_utils.bt import BtEnum
from acts.test_utils.bt.bt_test_utils import clear_bonded_devices
from acts.test_utils.coex.CoexPerformanceBaseTest import CoexPerformanceBaseTest
+from acts.test_utils.coex.audio_test_utils import SshAudioCapture
from acts.test_utils.coex.coex_test_utils import avrcp_actions
from acts.test_utils.coex.coex_test_utils import music_play_and_check
from acts.test_utils.coex.coex_test_utils import pair_and_connect_headset
@@ -39,6 +40,8 @@
def setup_test(self):
super().setup_test()
+ if "a2dp_streaming" in self.current_test_name:
+ self.audio = SshAudioCapture(self.audio_params, self.log_path)
self.audio_receiver.enter_pairing_mode()
time.sleep(5)
if not pair_and_connect_headset(
@@ -48,6 +51,8 @@
return False
def teardown_test(self):
+ if "a2dp_streaming" in self.current_test_name:
+ self.audio.terminate_and_store_audio_results()
clear_bonded_devices(self.pri_ad)
self.audio_receiver.clean_up()
super().teardown_test()
diff --git a/acts/tests/google/coex/stress_tests/CoexA2dpStressTest.py b/acts/tests/google/coex/stress_tests/CoexA2dpStressTest.py
index 41b2f95..6bdc4e2 100644
--- a/acts/tests/google/coex/stress_tests/CoexA2dpStressTest.py
+++ b/acts/tests/google/coex/stress_tests/CoexA2dpStressTest.py
@@ -25,6 +25,7 @@
from acts.test_utils.bt import BtEnum
from acts.test_utils.bt.bt_test_utils import clear_bonded_devices
+from acts.test_utils.coex.audio_test_utils import SshAudioCapture
from acts.test_utils.coex.CoexBaseTest import CoexBaseTest
from acts.test_utils.coex.coex_test_utils import connect_dev_to_headset
from acts.test_utils.coex.coex_test_utils import disconnect_headset_from_dev
@@ -45,6 +46,8 @@
def setup_test(self):
super().setup_test()
+ if "a2dp_streaming" in self.current_test_name:
+ self.audio = SshAudioCapture(self.audio_params, self.log_path)
self.audio_receiver.pairing_mode()
time.sleep(5) #Wait time until headset goes into pairing mode.
if not pair_and_connect_headset(
@@ -56,6 +59,11 @@
def teardown_test(self):
clear_bonded_devices(self.pri_ad)
self.audio_receiver.clean_up()
+ if "a2dp_streaming" in self.current_test_name:
+ analysis_path = self.audio.audio_quality_analysis(self.log_path)
+ with open(analysis_path) as f:
+ self.result["audio_artifacts"] = f.readline()
+ self.audio.terminate_and_store_audio_results()
super().teardown_test()
def connect_disconnect_headset(self):