1#!/usr/bin/env python3 2# 3# Copyright (C) 2018 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); you may not 6# use this file except in compliance with the License. You may obtain a copy of 7# the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 14# License for the specific language governing permissions and limitations under 15# the License 16 17import json 18import logging 19import math 20import os 21import re 22import time 23 24from acts import asserts 25from acts.controllers.ap_lib import hostapd_config 26from acts.controllers.ap_lib import hostapd_constants 27from acts.controllers.ap_lib import hostapd_security 28from acts.controllers.utils_lib.ssh import connection 29from acts.controllers.utils_lib.ssh import settings 30from acts.controllers.iperf_server import IPerfResult 31from acts.libs.proc import job 32from acts_contrib.test_utils.bt.bt_constants import ( 33 bluetooth_profile_connection_state_changed) 34from acts_contrib.test_utils.bt.bt_constants import bt_default_timeout 35from acts_contrib.test_utils.bt.bt_constants import bt_profile_constants 36from acts_contrib.test_utils.bt.bt_constants import bt_profile_states 37from acts_contrib.test_utils.bt.bt_constants import bt_scan_mode_types 38from acts_contrib.test_utils.bt.bt_gatt_utils import GattTestUtilsError 39from acts_contrib.test_utils.bt.bt_gatt_utils import orchestrate_gatt_connection 40from acts_contrib.test_utils.bt.bt_test_utils import disable_bluetooth 41from acts_contrib.test_utils.bt.bt_test_utils import enable_bluetooth 42from acts_contrib.test_utils.bt.bt_test_utils import is_a2dp_src_device_connected 43from acts_contrib.test_utils.bt.bt_test_utils import is_a2dp_snk_device_connected 44from acts_contrib.test_utils.bt.bt_test_utils import is_hfp_client_device_connected 45from acts_contrib.test_utils.bt.bt_test_utils import is_map_mce_device_connected 46from acts_contrib.test_utils.bt.bt_test_utils import is_map_mse_device_connected 47from acts_contrib.test_utils.bt.bt_test_utils import set_bt_scan_mode 48from acts_contrib.test_utils.car.car_telecom_utils import wait_for_active 49from acts_contrib.test_utils.car.car_telecom_utils import wait_for_dialing 50from acts_contrib.test_utils.car.car_telecom_utils import wait_for_not_in_call 51from acts_contrib.test_utils.car.car_telecom_utils import wait_for_ringing 52from acts_contrib.test_utils.tel.tel_test_utils import get_phone_number 53from acts_contrib.test_utils.tel.tel_test_utils import hangup_call 54from acts_contrib.test_utils.tel.tel_test_utils import initiate_call 55from acts_contrib.test_utils.tel.tel_test_utils import run_multithread_func 56from acts_contrib.test_utils.tel.tel_test_utils import setup_droid_properties 57from acts_contrib.test_utils.tel.tel_test_utils import wait_and_answer_call 58from acts_contrib.test_utils.wifi.wifi_power_test_utils import get_phone_ip 59from acts_contrib.test_utils.wifi.wifi_test_utils import reset_wifi 60from acts_contrib.test_utils.wifi.wifi_test_utils import wifi_connect 61from acts_contrib.test_utils.wifi.wifi_test_utils import wifi_test_device_init 62from acts_contrib.test_utils.wifi.wifi_test_utils import wifi_toggle_state 63from acts.utils import exe_cmd 64from bokeh.layouts import column 65from bokeh.models import tools as bokeh_tools 66from bokeh.plotting import figure, output_file, save 67 68THROUGHPUT_THRESHOLD = 100 69AP_START_TIME = 10 70DISCOVERY_TIME = 10 71BLUETOOTH_WAIT_TIME = 2 72AVRCP_WAIT_TIME = 3 73 74 75def avrcp_actions(pri_ad, bt_device): 76 """Performs avrcp controls like volume up, volume down, skip next and 77 skip previous. 78 79 Args: 80 pri_ad: Android device. 81 bt_device: bt device instance. 82 83 Returns: 84 True if successful, otherwise False. 85 """ 86 current_volume = pri_ad.droid.getMediaVolume() 87 for _ in range(5): 88 bt_device.volume_up() 89 time.sleep(AVRCP_WAIT_TIME) 90 if current_volume == pri_ad.droid.getMediaVolume(): 91 pri_ad.log.error("Increase volume failed") 92 return False 93 time.sleep(AVRCP_WAIT_TIME) 94 current_volume = pri_ad.droid.getMediaVolume() 95 for _ in range(5): 96 bt_device.volume_down() 97 time.sleep(AVRCP_WAIT_TIME) 98 if current_volume == pri_ad.droid.getMediaVolume(): 99 pri_ad.log.error("Decrease volume failed") 100 return False 101 102 #TODO: (sairamganesh) validate next and previous calls. 103 bt_device.next() 104 time.sleep(AVRCP_WAIT_TIME) 105 bt_device.previous() 106 time.sleep(AVRCP_WAIT_TIME) 107 return True 108 109 110def connect_ble(pri_ad, sec_ad): 111 """Connect BLE device from DUT. 112 113 Args: 114 pri_ad: An android device object. 115 sec_ad: An android device object. 116 117 Returns: 118 True if successful, otherwise False. 119 """ 120 adv_instances = [] 121 gatt_server_list = [] 122 bluetooth_gatt_list = [] 123 pri_ad.droid.bluetoothEnableBLE() 124 sec_ad.droid.bluetoothEnableBLE() 125 gatt_server_cb = sec_ad.droid.gattServerCreateGattServerCallback() 126 gatt_server = sec_ad.droid.gattServerOpenGattServer(gatt_server_cb) 127 gatt_server_list.append(gatt_server) 128 try: 129 bluetooth_gatt, gatt_callback, adv_callback = ( 130 orchestrate_gatt_connection(pri_ad, sec_ad)) 131 bluetooth_gatt_list.append(bluetooth_gatt) 132 except GattTestUtilsError as err: 133 pri_ad.log.error(err) 134 return False 135 adv_instances.append(adv_callback) 136 connected_devices = sec_ad.droid.gattServerGetConnectedDevices(gatt_server) 137 pri_ad.log.debug("Connected device = {}".format(connected_devices)) 138 return True 139 140 141def collect_bluetooth_manager_dumpsys_logs(pri_ad, test_name): 142 """Collect "adb shell dumpsys bluetooth_manager" logs. 143 144 Args: 145 pri_ad: An android device. 146 test_name: Current test case name. 147 148 Returns: 149 Dumpsys file path. 150 """ 151 dump_counter = 0 152 dumpsys_path = os.path.join(pri_ad.log_path, test_name, "BluetoothDumpsys") 153 os.makedirs(dumpsys_path, exist_ok=True) 154 while os.path.exists( 155 os.path.join(dumpsys_path, 156 "bluetooth_dumpsys_%s.txt" % dump_counter)): 157 dump_counter += 1 158 out_file = "bluetooth_dumpsys_%s.txt" % dump_counter 159 cmd = "adb -s {} shell dumpsys bluetooth_manager > {}/{}".format( 160 pri_ad.serial, dumpsys_path, out_file) 161 exe_cmd(cmd) 162 file_path = os.path.join(dumpsys_path, out_file) 163 return file_path 164 165 166def configure_and_start_ap(ap, network): 167 """Configure hostapd parameters and starts access point. 168 169 Args: 170 ap: An access point object. 171 network: A dictionary with wifi network details. 172 """ 173 hostapd_sec = None 174 if network["security"] == "wpa2": 175 hostapd_sec = hostapd_security.Security( 176 security_mode=network["security"], password=network["password"]) 177 178 config = hostapd_config.HostapdConfig( 179 n_capabilities=[hostapd_constants.N_CAPABILITY_HT40_MINUS], 180 mode=hostapd_constants.MODE_11N_PURE, 181 channel=network["channel"], 182 ssid=network["SSID"], 183 security=hostapd_sec) 184 ap.start_ap(config) 185 time.sleep(AP_START_TIME) 186 187 188def connect_dev_to_headset(pri_droid, dev_to_connect, profiles_set): 189 """Connects primary android device to headset. 190 191 Args: 192 pri_droid: Android device initiating connection. 193 dev_to_connect: Third party headset mac address. 194 profiles_set: Profiles to be connected. 195 196 Returns: 197 True if Pass 198 False if Fail 199 """ 200 supported_profiles = bt_profile_constants.values() 201 for profile in profiles_set: 202 if profile not in supported_profiles: 203 pri_droid.log.info("Profile {} is not supported list {}".format( 204 profile, supported_profiles)) 205 return False 206 207 paired = False 208 for paired_device in pri_droid.droid.bluetoothGetBondedDevices(): 209 if paired_device['address'] == dev_to_connect: 210 paired = True 211 break 212 213 if not paired: 214 pri_droid.log.info("{} not paired to {}".format(pri_droid.serial, 215 dev_to_connect)) 216 return False 217 218 end_time = time.time() + 10 219 profile_connected = set() 220 sec_addr = dev_to_connect 221 pri_droid.log.info("Profiles to be connected {}".format(profiles_set)) 222 223 while (time.time() < end_time and 224 not profile_connected.issuperset(profiles_set)): 225 if (bt_profile_constants['headset_client'] not in profile_connected and 226 bt_profile_constants['headset_client'] in profiles_set): 227 if is_hfp_client_device_connected(pri_droid, sec_addr): 228 profile_connected.add(bt_profile_constants['headset_client']) 229 if (bt_profile_constants['headset'] not in profile_connected and 230 bt_profile_constants['headset'] in profiles_set): 231 profile_connected.add(bt_profile_constants['headset']) 232 if (bt_profile_constants['a2dp'] not in profile_connected and 233 bt_profile_constants['a2dp'] in profiles_set): 234 if is_a2dp_src_device_connected(pri_droid, sec_addr): 235 profile_connected.add(bt_profile_constants['a2dp']) 236 if (bt_profile_constants['a2dp_sink'] not in profile_connected and 237 bt_profile_constants['a2dp_sink'] in profiles_set): 238 if is_a2dp_snk_device_connected(pri_droid, sec_addr): 239 profile_connected.add(bt_profile_constants['a2dp_sink']) 240 if (bt_profile_constants['map_mce'] not in profile_connected and 241 bt_profile_constants['map_mce'] in profiles_set): 242 if is_map_mce_device_connected(pri_droid, sec_addr): 243 profile_connected.add(bt_profile_constants['map_mce']) 244 if (bt_profile_constants['map'] not in profile_connected and 245 bt_profile_constants['map'] in profiles_set): 246 if is_map_mse_device_connected(pri_droid, sec_addr): 247 profile_connected.add(bt_profile_constants['map']) 248 time.sleep(0.1) 249 250 while not profile_connected.issuperset(profiles_set): 251 try: 252 time.sleep(10) 253 profile_event = pri_droid.ed.pop_event( 254 bluetooth_profile_connection_state_changed, 255 bt_default_timeout + 10) 256 pri_droid.log.info("Got event {}".format(profile_event)) 257 except Exception: 258 pri_droid.log.error("Did not get {} profiles left {}".format( 259 bluetooth_profile_connection_state_changed, profile_connected)) 260 return False 261 profile = profile_event['data']['profile'] 262 state = profile_event['data']['state'] 263 device_addr = profile_event['data']['addr'] 264 if state == bt_profile_states['connected'] and ( 265 device_addr == dev_to_connect): 266 profile_connected.add(profile) 267 pri_droid.log.info( 268 "Profiles connected until now {}".format(profile_connected)) 269 return True 270 271 272def device_discoverable(pri_ad, sec_ad): 273 """Verifies whether the device is discoverable or not. 274 275 Args: 276 pri_ad: An primary android device object. 277 sec_ad: An secondary android device object. 278 279 Returns: 280 True if the device found, False otherwise. 281 """ 282 pri_ad.droid.bluetoothMakeDiscoverable() 283 scan_mode = pri_ad.droid.bluetoothGetScanMode() 284 if scan_mode == bt_scan_mode_types['connectable_discoverable']: 285 pri_ad.log.info("Primary device scan mode is " 286 "SCAN_MODE_CONNECTABLE_DISCOVERABLE.") 287 else: 288 pri_ad.log.info("Primary device scan mode is not " 289 "SCAN_MODE_CONNECTABLE_DISCOVERABLE.") 290 return False 291 if sec_ad.droid.bluetoothStartDiscovery(): 292 time.sleep(DISCOVERY_TIME) 293 droid_name = pri_ad.droid.bluetoothGetLocalName() 294 droid_address = pri_ad.droid.bluetoothGetLocalAddress() 295 get_discovered_devices = sec_ad.droid.bluetoothGetDiscoveredDevices() 296 find_flag = False 297 298 if get_discovered_devices: 299 for device in get_discovered_devices: 300 if 'name' in device and device['name'] == droid_name or ( 301 'address' in device and 302 device["address"] == droid_address): 303 pri_ad.log.info("Primary device is in the discovery " 304 "list of secondary device.") 305 find_flag = True 306 break 307 else: 308 pri_ad.log.info("Secondary device get all the discovered devices " 309 "list is empty") 310 return False 311 else: 312 pri_ad.log.info("Secondary device start discovery process error.") 313 return False 314 if not find_flag: 315 return False 316 return True 317 318 319def device_discoverability(required_devices): 320 """Wrapper function to keep required_devices in discoverable mode. 321 322 Args: 323 required_devices: List of devices to be discovered. 324 325 Returns: 326 discovered_devices: List of BD_ADDR of devices in discoverable mode. 327 """ 328 discovered_devices = [] 329 if "AndroidDevice" in required_devices: 330 discovered_devices.extend( 331 android_device_discoverability(required_devices["AndroidDevice"])) 332 if "RelayDevice" in required_devices: 333 discovered_devices.extend( 334 relay_device_discoverability(required_devices["RelayDevice"])) 335 return discovered_devices 336 337 338def android_device_discoverability(droid_dev): 339 """To keep android devices in discoverable mode. 340 341 Args: 342 droid_dev: Android device object. 343 344 Returns: 345 device_list: List of device discovered. 346 """ 347 device_list = [] 348 for device in range(len(droid_dev)): 349 inquiry_device = droid_dev[device] 350 if enable_bluetooth(inquiry_device.droid, inquiry_device.ed): 351 if set_bt_scan_mode(inquiry_device, 352 bt_scan_mode_types['connectable_discoverable']): 353 device_list.append( 354 inquiry_device.droid.bluetoothGetLocalAddress()) 355 else: 356 droid_dev.log.error( 357 "Device {} scan mode is not in" 358 "SCAN_MODE_CONNECTABLE_DISCOVERABLE.".format( 359 inquiry_device.droid.bluetoothGetLocalAddress())) 360 return device_list 361 362 363def relay_device_discoverability(relay_devices): 364 """To keep relay controlled devices in discoverable mode. 365 366 Args: 367 relay_devices: Relay object. 368 369 Returns: 370 mac_address: Mac address of relay controlled device. 371 """ 372 relay_device = relay_devices[0] 373 relay_device.power_on() 374 relay_device.enter_pairing_mode() 375 return relay_device.mac_address 376 377 378def disconnect_headset_from_dev(pri_ad, sec_ad, profiles_list): 379 """Disconnect primary from secondary on a specific set of profiles 380 381 Args: 382 pri_ad: Primary android_device initiating disconnection 383 sec_ad: Secondary android droid (sl4a interface to keep the 384 method signature the same connect_pri_to_sec above) 385 profiles_list: List of profiles we want to disconnect from 386 387 Returns: 388 True on Success 389 False on Failure 390 """ 391 supported_profiles = bt_profile_constants.values() 392 for profile in profiles_list: 393 if profile not in supported_profiles: 394 pri_ad.log.info("Profile {} is not in supported list {}".format( 395 profile, supported_profiles)) 396 return False 397 398 pri_ad.log.info(pri_ad.droid.bluetoothGetBondedDevices()) 399 400 try: 401 pri_ad.droid.bluetoothDisconnectConnectedProfile(sec_ad, profiles_list) 402 except Exception as err: 403 pri_ad.log.error( 404 "Exception while trying to disconnect profile(s) {}: {}".format( 405 profiles_list, err)) 406 return False 407 408 profile_disconnected = set() 409 pri_ad.log.info("Disconnecting from profiles: {}".format(profiles_list)) 410 411 while not profile_disconnected.issuperset(profiles_list): 412 try: 413 profile_event = pri_ad.ed.pop_event( 414 bluetooth_profile_connection_state_changed, bt_default_timeout) 415 pri_ad.log.info("Got event {}".format(profile_event)) 416 except Exception: 417 pri_ad.log.warning("Did not disconnect from Profiles") 418 return True 419 420 profile = profile_event['data']['profile'] 421 state = profile_event['data']['state'] 422 device_addr = profile_event['data']['addr'] 423 424 if state == bt_profile_states['disconnected'] and ( 425 device_addr == sec_ad): 426 profile_disconnected.add(profile) 427 pri_ad.log.info( 428 "Profiles disconnected so far {}".format(profile_disconnected)) 429 430 return True 431 432 433def initiate_disconnect_from_hf(audio_receiver, pri_ad, sec_ad, duration): 434 """Initiates call and disconnect call on primary device. 435 436 Steps: 437 1. Initiate call from HF. 438 2. Wait for dialing state at DUT and wait for ringing at secondary device. 439 3. Accepts call from secondary device. 440 4. Wait for call active state at primary and secondary device. 441 5. Sleeps until given duration. 442 6. Disconnect call from primary device. 443 7. Wait for call is not present state. 444 445 Args: 446 audio_receiver: An relay device object. 447 pri_ad: An android device to disconnect call. 448 sec_ad: An android device accepting call. 449 duration: Duration of call in seconds. 450 451 Returns: 452 True if successful, False otherwise. 453 """ 454 audio_receiver.press_initiate_call() 455 time.sleep(2) 456 flag = True 457 flag &= wait_for_dialing(logging, pri_ad) 458 flag &= wait_for_ringing(logging, sec_ad) 459 if not flag: 460 pri_ad.log.error("Outgoing call did not get established") 461 return False 462 463 if not wait_and_answer_call(logging, sec_ad): 464 pri_ad.log.error("Failed to answer call in second device.") 465 return False 466 if not wait_for_active(logging, pri_ad): 467 pri_ad.log.error("AG not in Active state.") 468 return False 469 if not wait_for_active(logging, sec_ad): 470 pri_ad.log.error("RE not in Active state.") 471 return False 472 time.sleep(duration) 473 if not hangup_call(logging, pri_ad): 474 pri_ad.log.error("Failed to hangup call.") 475 return False 476 flag = True 477 flag &= wait_for_not_in_call(logging, pri_ad) 478 flag &= wait_for_not_in_call(logging, sec_ad) 479 return flag 480 481 482def initiate_disconnect_call_dut(pri_ad, sec_ad, duration, callee_number): 483 """Initiates call and disconnect call on primary device. 484 485 Steps: 486 1. Initiate call from DUT. 487 2. Wait for dialing state at DUT and wait for ringing at secondary device. 488 3. Accepts call from secondary device. 489 4. Wait for call active state at primary and secondary device. 490 5. Sleeps until given duration. 491 6. Disconnect call from primary device. 492 7. Wait for call is not present state. 493 494 Args: 495 pri_ad: An android device to disconnect call. 496 sec_ad: An android device accepting call. 497 duration: Duration of call in seconds. 498 callee_number: Secondary device's phone number. 499 500 Returns: 501 True if successful, False otherwise. 502 """ 503 if not initiate_call(logging, pri_ad, callee_number): 504 pri_ad.log.error("Failed to initiate call") 505 return False 506 time.sleep(2) 507 508 flag = True 509 flag &= wait_for_dialing(logging, pri_ad) 510 flag &= wait_for_ringing(logging, sec_ad) 511 if not flag: 512 pri_ad.log.error("Outgoing call did not get established") 513 return False 514 515 if not wait_and_answer_call(logging, sec_ad): 516 pri_ad.log.error("Failed to answer call in second device.") 517 return False 518 # Wait for AG, RE to go into an Active state. 519 if not wait_for_active(logging, pri_ad): 520 pri_ad.log.error("AG not in Active state.") 521 return False 522 if not wait_for_active(logging, sec_ad): 523 pri_ad.log.error("RE not in Active state.") 524 return False 525 time.sleep(duration) 526 if not hangup_call(logging, pri_ad): 527 pri_ad.log.error("Failed to hangup call.") 528 return False 529 flag = True 530 flag &= wait_for_not_in_call(logging, pri_ad) 531 flag &= wait_for_not_in_call(logging, sec_ad) 532 533 return flag 534 535 536def check_wifi_status(pri_ad, network, ssh_config=None): 537 """Function to check existence of wifi connection. 538 539 Args: 540 pri_ad: An android device. 541 network: network ssid. 542 ssh_config: ssh config for iperf client. 543 """ 544 time.sleep(5) 545 proc = job.run("pgrep -f 'iperf3 -c'") 546 pid_list = proc.stdout.split() 547 548 while True: 549 iperf_proc = job.run(["pgrep", "-f", "iperf3"]) 550 process_list = iperf_proc.stdout.split() 551 if not wifi_connection_check(pri_ad, network["SSID"]): 552 pri_ad.adb.shell("killall iperf3") 553 if ssh_config: 554 time.sleep(5) 555 ssh_settings = settings.from_config(ssh_config) 556 ssh_session = connection.SshConnection(ssh_settings) 557 result = ssh_session.run("pgrep iperf3") 558 res = result.stdout.split("\n") 559 for pid in res: 560 try: 561 ssh_session.run("kill -9 %s" % pid) 562 except Exception as e: 563 logging.warning("No such process: %s" % e) 564 for pid in pid_list[:-1]: 565 job.run(["kill", " -9", " %s" % pid], ignore_status=True) 566 else: 567 job.run(["killall", " iperf3"], ignore_status=True) 568 break 569 elif pid_list[0] not in process_list: 570 break 571 572 573def iperf_result(log, protocol, result): 574 """Accepts the iperf result in json format and parse the output to 575 get throughput value. 576 577 Args: 578 log: Logger object. 579 protocol : TCP or UDP protocol. 580 result: iperf result's filepath. 581 582 Returns: 583 rx_rate: Data received from client. 584 """ 585 if os.path.exists(result): 586 ip_cl = IPerfResult(result) 587 588 if protocol == "udp": 589 rx_rate = (math.fsum(ip_cl.instantaneous_rates) / 590 len(ip_cl.instantaneous_rates))*8 591 else: 592 rx_rate = ip_cl.avg_receive_rate * 8 593 return rx_rate 594 else: 595 log.error("IPerf file not found") 596 return False 597 598 599def is_a2dp_connected(pri_ad, headset_mac_address): 600 """Convenience Function to see if the 2 devices are connected on A2DP. 601 602 Args: 603 pri_ad : An android device. 604 headset_mac_address : Mac address of headset. 605 606 Returns: 607 True:If A2DP connection exists, False otherwise. 608 """ 609 devices = pri_ad.droid.bluetoothA2dpGetConnectedDevices() 610 for device in devices: 611 pri_ad.log.debug("A2dp Connected device {}".format(device["name"])) 612 if device["address"] == headset_mac_address: 613 return True 614 return False 615 616 617def media_stream_check(pri_ad, duration, headset_mac_address): 618 """Checks whether A2DP connecion is active or not for given duration of 619 time. 620 621 Args: 622 pri_ad : An android device. 623 duration : No of seconds to check if a2dp streaming is alive. 624 headset_mac_address : Headset mac address. 625 626 Returns: 627 True: If A2dp connection is active for entire duration. 628 False: If A2dp connection is not active. 629 """ 630 while time.time() < duration: 631 if not is_a2dp_connected(pri_ad, headset_mac_address): 632 pri_ad.log.error('A2dp connection not active at %s', pri_ad.serial) 633 return False 634 time.sleep(1) 635 return True 636 637 638def multithread_func(log, tasks): 639 """Multi-thread function wrapper. 640 641 Args: 642 log: log object. 643 tasks: tasks to be executed in parallel. 644 645 Returns: 646 List of results of tasks 647 """ 648 results = run_multithread_func(log, tasks) 649 for res in results: 650 if not res: 651 return False 652 return True 653 654 655def music_play_and_check(pri_ad, headset_mac_address, music_to_play, duration): 656 """Starts playing media and checks if media plays for n seconds. 657 658 Steps: 659 1. Starts media player on android device. 660 2. Checks if music streaming is ongoing for n seconds. 661 3. Stops media player. 662 4. Collect dumpsys logs. 663 664 Args: 665 pri_ad: An android device. 666 headset_mac_address: Mac address of third party headset. 667 music_to_play: Indicates the music file to play. 668 duration: Time in secs to indicate music time to play. 669 670 Returns: 671 True if successful, False otherwise. 672 """ 673 pri_ad.droid.setMediaVolume(pri_ad.droid.getMaxMediaVolume() - 1) 674 pri_ad.log.info("current volume = {}".format(pri_ad.droid.getMediaVolume())) 675 pri_ad.log.debug("In music play and check") 676 if not start_media_play(pri_ad, music_to_play): 677 pri_ad.log.error("Start media play failed.") 678 return False 679 stream_time = time.time() + duration 680 if not media_stream_check(pri_ad, stream_time, headset_mac_address): 681 pri_ad.log.error("A2DP Connection check failed.") 682 pri_ad.droid.mediaPlayStop() 683 return False 684 pri_ad.droid.mediaPlayStop() 685 return True 686 687 688def music_play_and_check_via_app(pri_ad, headset_mac_address, duration=5): 689 """Starts google music player and check for A2DP connection. 690 691 Steps: 692 1. Starts Google music player on android device. 693 2. Checks for A2DP connection. 694 695 Args: 696 pri_ad: An android device. 697 headset_mac_address: Mac address of third party headset. 698 duration: Total time of music streaming. 699 700 Returns: 701 True if successful, False otherwise. 702 """ 703 pri_ad.adb.shell("am start com.google.android.music") 704 time.sleep(3) 705 pri_ad.adb.shell("input keyevent 85") 706 stream_time = time.time() + duration 707 try: 708 if not media_stream_check(pri_ad, stream_time, headset_mac_address): 709 pri_ad.log.error("A2dp connection not active at %s", pri_ad.serial) 710 return False 711 finally: 712 pri_ad.adb.shell("am force-stop com.google.android.music") 713 return True 714 715 716def pair_dev_to_headset(pri_ad, dev_to_pair): 717 """Pairs primary android device with headset. 718 719 Args: 720 pri_ad: Android device initiating connection 721 dev_to_pair: Third party headset mac address. 722 723 Returns: 724 True if Pass 725 False if Fail 726 """ 727 bonded_devices = pri_ad.droid.bluetoothGetBondedDevices() 728 for d in bonded_devices: 729 if d['address'] == dev_to_pair: 730 pri_ad.log.info("Successfully bonded to device {}".format( 731 dev_to_pair)) 732 return True 733 pri_ad.droid.bluetoothStartDiscovery() 734 time.sleep(10) # Wait until device gets discovered 735 pri_ad.droid.bluetoothCancelDiscovery() 736 pri_ad.log.debug("Discovered bluetooth devices: {}".format( 737 pri_ad.droid.bluetoothGetDiscoveredDevices())) 738 for device in pri_ad.droid.bluetoothGetDiscoveredDevices(): 739 if device['address'] == dev_to_pair: 740 741 result = pri_ad.droid.bluetoothDiscoverAndBond(dev_to_pair) 742 pri_ad.log.info(result) 743 end_time = time.time() + bt_default_timeout 744 pri_ad.log.info("Verifying if device bonded with {}".format( 745 dev_to_pair)) 746 time.sleep(5) # Wait time until device gets paired. 747 while time.time() < end_time: 748 bonded_devices = pri_ad.droid.bluetoothGetBondedDevices() 749 for d in bonded_devices: 750 if d['address'] == dev_to_pair: 751 pri_ad.log.info( 752 "Successfully bonded to device {}".format( 753 dev_to_pair)) 754 return True 755 pri_ad.log.error("Failed to bond with {}".format(dev_to_pair)) 756 return False 757 758 759def pair_and_connect_headset(pri_ad, headset_mac_address, profile_to_connect, retry=5): 760 """Pair and connect android device with third party headset. 761 762 Args: 763 pri_ad: An android device. 764 headset_mac_address: Mac address of third party headset. 765 profile_to_connect: Profile to be connected with headset. 766 retry: Number of times pair and connection should happen. 767 768 Returns: 769 True if pair and connect to headset successful, or raises exception 770 on failure. 771 """ 772 773 paired = False 774 for i in range(1, retry): 775 if pair_dev_to_headset(pri_ad, headset_mac_address): 776 paired = True 777 break 778 else: 779 pri_ad.log.error("Attempt {} out of {}, Failed to pair, " 780 "Retrying.".format(i, retry)) 781 782 if paired: 783 for i in range(1, retry): 784 if connect_dev_to_headset(pri_ad, headset_mac_address, 785 profile_to_connect): 786 return True 787 else: 788 pri_ad.log.error("Attempt {} out of {}, Failed to connect, " 789 "Retrying.".format(i, retry)) 790 else: 791 asserts.fail("Failed to pair and connect with {}".format( 792 headset_mac_address)) 793 794 795def perform_classic_discovery(pri_ad, duration, file_name, dev_list=None): 796 """Convenience function to start and stop device discovery. 797 798 Args: 799 pri_ad: An android device. 800 duration: iperf duration of the test. 801 file_name: Json file to which result is dumped 802 dev_list: List of devices to be discoverable mode. 803 804 Returns: 805 True start and stop discovery is successful, False otherwise. 806 """ 807 if dev_list: 808 devices_required = device_discoverability(dev_list) 809 else: 810 devices_required = None 811 iteration = 0 812 result = {} 813 result["discovered_devices"] = {} 814 discover_result = [] 815 start_time = time.time() 816 while time.time() < start_time + duration: 817 if not pri_ad.droid.bluetoothStartDiscovery(): 818 pri_ad.log.error("Failed to start discovery") 819 return False 820 time.sleep(DISCOVERY_TIME) 821 if not pri_ad.droid.bluetoothCancelDiscovery(): 822 pri_ad.log.error("Failed to cancel discovery") 823 return False 824 pri_ad.log.info("Discovered device list {}".format( 825 pri_ad.droid.bluetoothGetDiscoveredDevices())) 826 if devices_required is not None: 827 result["discovered_devices"][iteration] = [] 828 devices_name = { 829 element.get('name', element['address']) 830 for element in pri_ad.droid.bluetoothGetDiscoveredDevices() 831 if element["address"] in devices_required 832 } 833 result["discovered_devices"][iteration] = list(devices_name) 834 discover_result.extend([len(devices_name) == len(devices_required)]) 835 iteration += 1 836 with open(file_name, 'a') as results_file: 837 json.dump(result, results_file, indent=4) 838 if False in discover_result: 839 return False 840 else: 841 pri_ad.log.warning("No devices are kept in discoverable mode") 842 return True 843 844 845def connect_wlan_profile(pri_ad, network): 846 """Disconnect and Connect to AP. 847 848 Args: 849 pri_ad: An android device. 850 network: Network to which AP to be connected. 851 852 Returns: 853 True if successful, False otherwise. 854 """ 855 reset_wifi(pri_ad) 856 wifi_toggle_state(pri_ad, False) 857 wifi_test_device_init(pri_ad) 858 wifi_connect(pri_ad, network) 859 if not wifi_connection_check(pri_ad, network["SSID"]): 860 pri_ad.log.error("Wifi connection does not exist.") 861 return False 862 return True 863 864 865def toggle_bluetooth(pri_ad, duration): 866 """Toggles bluetooth on/off for N iterations. 867 868 Args: 869 pri_ad: An android device object. 870 duration: Iperf duration of the test. 871 872 Returns: 873 True if successful, False otherwise. 874 """ 875 start_time = time.time() 876 while time.time() < start_time + duration: 877 if not enable_bluetooth(pri_ad.droid, pri_ad.ed): 878 pri_ad.log.error("Failed to enable bluetooth") 879 return False 880 time.sleep(BLUETOOTH_WAIT_TIME) 881 if not disable_bluetooth(pri_ad.droid): 882 pri_ad.log.error("Failed to turn off bluetooth") 883 return False 884 time.sleep(BLUETOOTH_WAIT_TIME) 885 return True 886 887 888def toggle_screen_state(pri_ad, duration): 889 """Toggles the screen state to on or off.. 890 891 Args: 892 pri_ad: Android device. 893 duration: Iperf duration of the test. 894 895 Returns: 896 True if successful, False otherwise. 897 """ 898 start_time = time.time() 899 while time.time() < start_time + duration: 900 if not pri_ad.ensure_screen_on(): 901 pri_ad.log.error("User window cannot come up") 902 return False 903 if not pri_ad.go_to_sleep(): 904 pri_ad.log.info("Screen off") 905 return True 906 907 908def setup_tel_config(pri_ad, sec_ad, sim_conf_file): 909 """Sets tel properties for primary device and secondary devices 910 911 Args: 912 pri_ad: An android device object. 913 sec_ad: An android device object. 914 sim_conf_file: Sim card map. 915 916 Returns: 917 pri_ad_num: Phone number of primary device. 918 sec_ad_num: Phone number of secondary device. 919 """ 920 setup_droid_properties(logging, pri_ad, sim_conf_file) 921 pri_ad_num = get_phone_number(logging, pri_ad) 922 setup_droid_properties(logging, sec_ad, sim_conf_file) 923 sec_ad_num = get_phone_number(logging, sec_ad) 924 return pri_ad_num, sec_ad_num 925 926 927def start_fping(pri_ad, duration, fping_params): 928 """Starts fping to ping for DUT's ip address. 929 930 Steps: 931 1. Run fping command to check DUT's IP is alive or not. 932 933 Args: 934 pri_ad: An android device object. 935 duration: Duration of fping in seconds. 936 fping_params: List of parameters for fping to run. 937 938 Returns: 939 True if successful, False otherwise. 940 """ 941 counter = 0 942 fping_path = ''.join((pri_ad.log_path, "/Fping")) 943 os.makedirs(fping_path, exist_ok=True) 944 while os.path.isfile(fping_path + "/fping_%s.txt" % counter): 945 counter += 1 946 out_file_name = "{}".format("fping_%s.txt" % counter) 947 948 full_out_path = os.path.join(fping_path, out_file_name) 949 cmd = "fping {} -D -c {}".format(get_phone_ip(pri_ad), duration) 950 if fping_params["ssh_config"]: 951 ssh_settings = settings.from_config(fping_params["ssh_config"]) 952 ssh_session = connection.SshConnection(ssh_settings) 953 try: 954 with open(full_out_path, 'w') as outfile: 955 job_result = ssh_session.run(cmd) 956 outfile.write(job_result.stdout) 957 outfile.write("\n") 958 outfile.writelines(job_result.stderr) 959 except Exception as err: 960 pri_ad.log.error("Fping run has been failed. = {}".format(err)) 961 return False 962 else: 963 cmd = cmd.split() 964 with open(full_out_path, "w") as f: 965 job.run(cmd) 966 result = parse_fping_results(fping_params["fping_drop_tolerance"], 967 full_out_path) 968 return bool(result) 969 970 971def parse_fping_results(failure_rate, full_out_path): 972 """Calculates fping results. 973 974 Steps: 975 1. Read the file and calculate the results. 976 977 Args: 978 failure_rate: Fping packet drop tolerance value. 979 full_out_path: path where the fping results has been stored. 980 981 Returns: 982 loss_percent: loss percentage of fping packet. 983 """ 984 try: 985 result_file = open(full_out_path, "r") 986 lines = result_file.readlines() 987 res_line = lines[-1] 988 # Ex: res_line = "192.168.186.224 : xmt/rcv/%loss = 10/10/0%, 989 # min/avg/max = 36.7/251/1272" 990 loss_percent = re.search("[0-9]+%", res_line) 991 if int(loss_percent.group().strip("%")) > failure_rate: 992 logging.error("Packet drop observed") 993 return False 994 return loss_percent.group() 995 except Exception as e: 996 logging.error("Error in parsing fping results : %s" %(e)) 997 return False 998 999 1000def start_media_play(pri_ad, music_file_to_play): 1001 """Starts media player on device. 1002 1003 Args: 1004 pri_ad : An android device. 1005 music_file_to_play : An audio file to play. 1006 1007 Returns: 1008 True:If media player start music, False otherwise. 1009 """ 1010 if not pri_ad.droid.mediaPlayOpen( 1011 "file:///sdcard/Music/{}".format(music_file_to_play)): 1012 pri_ad.log.error("Failed to play music") 1013 return False 1014 1015 pri_ad.droid.mediaPlaySetLooping(True) 1016 pri_ad.log.info("Music is now playing on device {}".format(pri_ad.serial)) 1017 return True 1018 1019 1020def wifi_connection_check(pri_ad, ssid): 1021 """Function to check existence of wifi connection. 1022 1023 Args: 1024 pri_ad : An android device. 1025 ssid : wifi ssid to check. 1026 1027 Returns: 1028 True if wifi connection exists, False otherwise. 1029 """ 1030 wifi_info = pri_ad.droid.wifiGetConnectionInfo() 1031 if (wifi_info["SSID"] == ssid and 1032 wifi_info["supplicant_state"] == "completed"): 1033 return True 1034 pri_ad.log.error("Wifi Connection check failed : {}".format(wifi_info)) 1035 return False 1036 1037 1038def push_music_to_android_device(ad, audio_params): 1039 """Add music to Android device as specified by the test config 1040 1041 Args: 1042 ad: Android device 1043 audio_params: Music file to push. 1044 1045 Returns: 1046 True on success, False on failure 1047 """ 1048 ad.log.info("Pushing music to the Android device") 1049 android_music_path = "/sdcard/Music/" 1050 music_path = audio_params["music_file"] 1051 if type(music_path) is list: 1052 ad.log.info("Media ready to push as is.") 1053 for item in music_path: 1054 music_file_to_play = item 1055 ad.adb.push(item, android_music_path) 1056 return music_file_to_play 1057 else: 1058 music_file_to_play = audio_params["music_file"] 1059 ad.adb.push("{} {}".format(music_file_to_play, android_music_path)) 1060 return (os.path.basename(music_file_to_play)) 1061 1062def bokeh_plot(data_sets, 1063 legends, 1064 fig_property, 1065 shaded_region=None, 1066 output_file_path=None): 1067 """Plot bokeh figs. 1068 Args: 1069 data_sets: data sets including lists of x_data and lists of y_data 1070 ex: [[[x_data1], [x_data2]], [[y_data1],[y_data2]]] 1071 legends: list of legend for each curve 1072 fig_property: dict containing the plot property, including title, 1073 labels, linewidth, circle size, etc. 1074 shaded_region: optional dict containing data for plot shading 1075 output_file_path: optional path at which to save figure 1076 Returns: 1077 plot: bokeh plot figure object 1078 """ 1079 tools = 'box_zoom,box_select,pan,crosshair,redo,undo,reset,hover,save' 1080 plot = figure(plot_width=1300, 1081 plot_height=700, 1082 title=fig_property['title'], 1083 tools=tools, 1084 output_backend="webgl") 1085 plot.add_tools(bokeh_tools.WheelZoomTool(dimensions="width")) 1086 plot.add_tools(bokeh_tools.WheelZoomTool(dimensions="height")) 1087 colors = [ 1088 'red', 'green', 'blue', 'olive', 'orange', 'salmon', 'black', 'navy', 1089 'yellow', 'darkred', 'goldenrod' 1090 ] 1091 if shaded_region: 1092 band_x = shaded_region["x_vector"] 1093 band_x.extend(shaded_region["x_vector"][::-1]) 1094 band_y = shaded_region["lower_limit"] 1095 band_y.extend(shaded_region["upper_limit"][::-1]) 1096 plot.patch(band_x, 1097 band_y, 1098 color='#7570B3', 1099 line_alpha=0.1, 1100 fill_alpha=0.1) 1101 1102 for x_data, y_data, legend in zip(data_sets[0], data_sets[1], legends): 1103 index_now = legends.index(legend) 1104 color = colors[index_now % len(colors)] 1105 plot.line(x_data, 1106 y_data, 1107 legend=str(legend), 1108 line_width=fig_property['linewidth'], 1109 color=color) 1110 plot.circle(x_data, 1111 y_data, 1112 size=fig_property['markersize'], 1113 legend=str(legend), 1114 fill_color=color) 1115 1116 # Plot properties 1117 plot.xaxis.axis_label = fig_property['x_label'] 1118 plot.yaxis.axis_label = fig_property['y_label'] 1119 plot.legend.location = "top_right" 1120 plot.legend.click_policy = "hide" 1121 plot.title.text_font_size = {'value': '15pt'} 1122 if output_file_path is not None: 1123 output_file(output_file_path) 1124 save(plot) 1125 return plot 1126 1127def bokeh_chart_plot(bt_attenuation_range, 1128 data_sets, 1129 legends, 1130 fig_property, 1131 shaded_region=None, 1132 output_file_path=None): 1133 """Plot bokeh figs. 1134 1135 Args: 1136 bt_attenuation_range: range of BT attenuation. 1137 data_sets: data sets including lists of x_data and lists of y_data 1138 ex: [[[x_data1], [x_data2]], [[y_data1],[y_data2]]] 1139 legends: list of legend for each curve 1140 fig_property: dict containing the plot property, including title, 1141 labels, linewidth, circle size, etc. 1142 shaded_region: optional dict containing data for plot shading 1143 output_file_path: optional path at which to save figure 1144 1145 Returns: 1146 plot: bokeh plot figure object 1147 """ 1148 TOOLS = ('box_zoom,box_select,pan,crosshair,redo,undo,reset,hover,save') 1149 colors = [ 1150 'red', 'green', 'blue', 'olive', 'orange', 'salmon', 'black', 'navy', 1151 'yellow', 'darkred', 'goldenrod' 1152 ] 1153 plot = [] 1154 data = [[], []] 1155 legend = [] 1156 for i in bt_attenuation_range: 1157 if "Packet drop" in legends[i][0]: 1158 plot_info = {0: "A2dp_packet_drop_plot", 1: "throughput_plot"} 1159 else: 1160 plot_info = {0: "throughput_plot"} 1161 for j in plot_info: 1162 if "Packet drops" in legends[i][j]: 1163 if data_sets[i]["a2dp_packet_drops"]: 1164 plot_i_j = figure( 1165 plot_width=1000, 1166 plot_height=500, 1167 title=fig_property['title'], 1168 tools=TOOLS) 1169 1170 plot_i_j.add_tools( 1171 bokeh_tools.WheelZoomTool(dimensions="width")) 1172 plot_i_j.add_tools( 1173 bokeh_tools.WheelZoomTool(dimensions="height")) 1174 plot_i_j.xaxis.axis_label = fig_property['x_label'] 1175 plot_i_j.yaxis.axis_label = fig_property['y_label'][j] 1176 plot_i_j.legend.location = "top_right" 1177 plot_i_j.legend.click_policy = "hide" 1178 plot_i_j.title.text_font_size = {'value': '15pt'} 1179 1180 plot_i_j.line( 1181 data_sets[i]["a2dp_attenuation"], 1182 data_sets[i]["a2dp_packet_drops"], 1183 legend=legends[i][j], 1184 line_width=3, 1185 color=colors[j]) 1186 plot_i_j.circle( 1187 data_sets[i]["a2dp_attenuation"], 1188 data_sets[i]["a2dp_packet_drops"], 1189 legend=str(legends[i][j]), 1190 fill_color=colors[j]) 1191 plot.append(plot_i_j) 1192 elif "Performance Results" in legends[i][j]: 1193 plot_i_j = figure( 1194 plot_width=1000, 1195 plot_height=500, 1196 title=fig_property['title'], 1197 tools=TOOLS) 1198 plot_i_j.add_tools( 1199 bokeh_tools.WheelZoomTool(dimensions="width")) 1200 plot_i_j.add_tools( 1201 bokeh_tools.WheelZoomTool(dimensions="height")) 1202 plot_i_j.xaxis.axis_label = fig_property['x_label'] 1203 plot_i_j.yaxis.axis_label = fig_property['y_label'][j] 1204 plot_i_j.legend.location = "top_right" 1205 plot_i_j.legend.click_policy = "hide" 1206 plot_i_j.title.text_font_size = {'value': '15pt'} 1207 data[0].insert(0, data_sets[i]["attenuation"]) 1208 data[1].insert(0, data_sets[i]["throughput_received"]) 1209 legend.insert(0, legends[i][j + 1]) 1210 plot_i_j.line( 1211 data_sets[i]["user_attenuation"], 1212 data_sets[i]["user_throughput"], 1213 legend=legends[i][j], 1214 line_width=3, 1215 color=colors[j]) 1216 plot_i_j.circle( 1217 data_sets[i]["user_attenuation"], 1218 data_sets[i]["user_throughput"], 1219 legend=str(legends[i][j]), 1220 fill_color=colors[j]) 1221 plot_i_j.line( 1222 data_sets[i]["attenuation"], 1223 data_sets[i]["throughput_received"], 1224 legend=legends[i][j + 1], 1225 line_width=3, 1226 color=colors[j]) 1227 plot_i_j.circle( 1228 data_sets[i]["attenuation"], 1229 data_sets[i]["throughput_received"], 1230 legend=str(legends[i][j + 1]), 1231 fill_color=colors[j]) 1232 if shaded_region: 1233 band_x = shaded_region[i]["x_vector"] 1234 band_x.extend(shaded_region[i]["x_vector"][::-1]) 1235 band_y = shaded_region[i]["lower_limit"] 1236 band_y.extend(shaded_region[i]["upper_limit"][::-1]) 1237 plot_i_j.patch( 1238 band_x, 1239 band_y, 1240 color='#7570B3', 1241 line_alpha=0.1, 1242 fill_alpha=0.1) 1243 plot.append(plot_i_j) 1244 else: 1245 plot_i_j = figure( 1246 plot_width=1000, 1247 plot_height=500, 1248 title=fig_property['title'], 1249 tools=TOOLS) 1250 plot_i_j.add_tools( 1251 bokeh_tools.WheelZoomTool(dimensions="width")) 1252 plot_i_j.add_tools( 1253 bokeh_tools.WheelZoomTool(dimensions="height")) 1254 plot_i_j.xaxis.axis_label = fig_property['x_label'] 1255 plot_i_j.yaxis.axis_label = fig_property['y_label'][j] 1256 plot_i_j.legend.location = "top_right" 1257 plot_i_j.legend.click_policy = "hide" 1258 plot_i_j.title.text_font_size = {'value': '15pt'} 1259 data[0].insert(0, data_sets[i]["attenuation"]) 1260 data[1].insert(0, data_sets[i]["throughput_received"]) 1261 legend.insert(0, legends[i][j]) 1262 plot_i_j.line( 1263 data_sets[i]["attenuation"], 1264 data_sets[i]["throughput_received"], 1265 legend=legends[i][j], 1266 line_width=3, 1267 color=colors[j]) 1268 plot_i_j.circle( 1269 data_sets[i]["attenuation"], 1270 data_sets[i]["throughput_received"], 1271 legend=str(legends[i][j]), 1272 fill_color=colors[j]) 1273 plot.append(plot_i_j) 1274 fig_property['y_label'] = "Throughput (Mbps)" 1275 all_plot = bokeh_plot(data, legend, fig_property, shaded_region=None, 1276 output_file_path=None) 1277 plot.insert(0, all_plot) 1278 if output_file_path is not None: 1279 output_file(output_file_path) 1280 save(column(plot)) 1281 return plot 1282 1283 1284class A2dpDumpsysParser(): 1285 1286 def __init__(self): 1287 self.count_list = [] 1288 self.frame_list = [] 1289 self.dropped_count = None 1290 1291 def parse(self, file_path): 1292 """Convenience function to parse a2dp dumpsys logs. 1293 1294 Args: 1295 file_path: Path of dumpsys logs. 1296 1297 Returns: 1298 dropped_list containing packet drop count for every iteration. 1299 drop containing list of all packets dropped for test suite. 1300 """ 1301 a2dp_dumpsys_info = [] 1302 with open(file_path) as dumpsys_file: 1303 for line in dumpsys_file: 1304 if "A2DP State:" in line: 1305 a2dp_dumpsys_info.append(line) 1306 elif "Counts (max dropped)" not in line and len( 1307 a2dp_dumpsys_info) > 0: 1308 a2dp_dumpsys_info.append(line) 1309 elif "Counts (max dropped)" in line: 1310 a2dp_dumpsys_info = ''.join(a2dp_dumpsys_info) 1311 a2dp_info = a2dp_dumpsys_info.split("\n") 1312 # Ex: Frames per packet (total/max/ave) : 5034 / 1 / 0 1313 frames = int(re.split("[':/()]", str(a2dp_info[-3]))[-3]) 1314 self.frame_list.append(frames) 1315 # Ex : Counts (flushed/dropped/dropouts) : 0 / 4 / 0 1316 count = int(re.split("[':/()]", str(a2dp_info[-2]))[-2]) 1317 if count > 0: 1318 for i in range(len(self.count_list)): 1319 count = count - self.count_list[i] 1320 self.count_list.append(count) 1321 if len(self.frame_list) > 1: 1322 last_frame = self.frame_list[-1] - self.frame_list[ 1323 -2] 1324 self.dropped_count = (count / last_frame) * 100 1325 else: 1326 self.dropped_count = ( 1327 count / self.frame_list[-1]) * 100 1328 else: 1329 self.dropped_count = count 1330 logging.info(a2dp_dumpsys_info) 1331 return self.dropped_count 1332