Merge "PTS Cmd Line Tool for Ble" into oc-dev

2 files changed, 299 insertions, 0 deletions

diff --git a/acts/tests/google/bt/pts/ble_lib.py b/acts/tests/google/bt/pts/ble_lib.py
new file mode 100644
index 0000000000..9390dff2af
--- /dev/null
+++ b/acts/tests/google/bt/pts/ble_lib.py
@@ -0,0 +1,184 @@
+#/usr/bin/env python3.4
+#
+# Copyright (C) 2016 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.
+"""
+Ble libraries
+"""
+
+from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseMode
+from acts.test_utils.bt.BleEnum import AdvertiseSettingsAdvertiseTxPower
+from acts.test_utils.bt.BleEnum import ScanSettingsScanMode
+from acts.test_utils.bt.bt_test_utils import TIMEOUT_SMALL
+from acts.test_utils.bt.bt_test_utils import adv_fail
+from acts.test_utils.bt.bt_test_utils import adv_succ
+from acts.test_utils.bt.bt_test_utils import advertising_set_on_own_address_read
+from acts.test_utils.bt.bt_test_utils import advertising_set_started
+from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects
+
+import time
+import os
+
+
+class BleLib():
+    def __init__(self, log, mac_addr, dut):
+        self.advertisement_list = []
+        self.dut = dut
+        self.log = log
+        self.mac_addr = mac_addr
+        self.default_timeout = 5
+        self.set_advertisement_list = []
+        self.generic_uuid = "0000{}-0000-1000-8000-00805f9b34fb"
+
+    def _verify_ble_adv_started(self, advertise_callback):
+        """Helper for verifying if an advertisment started or not"""
+        regex = "({}|{})".format(
+            adv_succ.format(advertise_callback),
+            adv_fail.format(advertise_callback))
+        try:
+            event = self.dut.ed.pop_events(regex, 5, TIMEOUT_SMALL)
+        except Empty:
+            self.dut.log.error("Failed to get success or failed event.")
+            return
+        if event[0]["name"] == adv_succ.format(advertise_callback):
+            self.dut.log.info("Advertisement started successfully.")
+            return True
+        else:
+            self.dut.log.info("Advertisement failed to start.")
+            return False
+
+    def start_generic_connectable_advertisement(self, line):
+        """Start a connectable LE advertisement"""
+        scan_response = None
+        if line:
+            scan_response = bool(line)
+        self.dut.droid.bleSetAdvertiseSettingsAdvertiseMode(
+            AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY.value)
+        self.dut.droid.bleSetAdvertiseSettingsIsConnectable(True)
+        advertise_callback, advertise_data, advertise_settings = (
+            generate_ble_advertise_objects(self.dut.droid))
+        if scan_response:
+            self.dut.droid.bleStartBleAdvertisingWithScanResponse(
+                advertise_callback, advertise_data, advertise_settings,
+                advertise_data)
+        else:
+            self.dut.droid.bleStartBleAdvertising(
+                advertise_callback, advertise_data, advertise_settings)
+        if self._verify_ble_adv_started(advertise_callback):
+            self.log.info(
+                "Tracking Callback ID: {}".format(advertise_callback))
+            self.advertisement_list.append(advertise_callback)
+            self.log.info(self.advertisement_list)
+
+    def start_connectable_advertisement_set(self, line):
+        """Start Connectable Advertisement Set"""
+        adv_callback = self.dut.droid.bleAdvSetGenCallback()
+        adv_data = {
+            "includeDeviceName": True,
+        }
+        self.dut.droid.bleAdvSetStartAdvertisingSet({
+            "connectable":
+            True,
+            "legacyMode":
+            False,
+            "primaryPhy":
+            "PHY_LE_1M",
+            "secondaryPhy":
+            "PHY_LE_1M",
+            "interval":
+            320
+        }, adv_data, None, None, None, 0, 0, adv_callback)
+        evt = self.dut.ed.pop_event(
+            advertising_set_started.format(adv_callback), self.default_timeout)
+        set_id = evt['data']['setId']
+        self.log.error("did not receive the set started event!")
+        evt = self.dut.ed.pop_event(
+            advertising_set_on_own_address_read.format(set_id),
+            self.default_timeout)
+        address = evt['data']['address']
+        self.log.info("Advertiser address is: {}".format(str(address)))
+        self.set_advertisement_list.append(adv_callback)
+
+    def stop_all_advertisement_set(self, line):
+        """Stop all Advertisement Sets"""
+        for adv in self.set_advertisement_list:
+            try:
+                self.dut.droid.bleAdvSetStopAdvertisingSet(adv)
+            except Exception as err:
+                self.log.error("Failed to stop advertisement: {}".format(err))
+
+    def adv_add_service_uuid_list(self, line):
+        """Add service UUID to the LE advertisement inputs:
+         [uuid1 uuid2 ... uuidN]"""
+        uuids = line.split()
+        uuid_list = []
+        for uuid in uuids:
+            if len(uuid) == 4:
+                uuid = self.generic_uuid.format(line)
+            uuid_list.append(service.format(uuid))
+        self.dut.droid.bleSetAdvertiseDataSetServiceUuids(uuid_list)
+
+    def adv_data_include_local_name(self, is_included):
+        """Include local name in the advertisement. inputs: [true|false]"""
+        self.dut.droid.bleSetAdvertiseDataIncludeDeviceName(bool(is_included))
+
+    def adv_data_include_tx_power_level(self, is_included):
+        """Include tx power level in the advertisement. inputs: [true|false]"""
+        self.dut.droid.bleSetAdvertiseDataIncludeTxPowerLevel(
+            bool(is_included))
+
+    def adv_data_add_manufacturer_data(self, line):
+        """Include manufacturer id and data to the advertisment:
+        [id data1 data2 ... dataN]"""
+        info = line.split()
+        manu_id = int(info[0])
+        manu_data = []
+        for data in info[1:]:
+            manu_data.append(int(data))
+        self.dut.droid.bleAddAdvertiseDataManufacturerId(manu_id, manu_data)
+
+    def start_generic_nonconnectable_advertisement(self, line):
+        """Start a nonconnectable LE advertisement"""
+        self.dut.droid.bleSetAdvertiseSettingsAdvertiseMode(
+            AdvertiseSettingsAdvertiseMode.ADVERTISE_MODE_LOW_LATENCY.value)
+        self.dut.droid.bleSetAdvertiseSettingsIsConnectable(False)
+        advertise_callback, advertise_data, advertise_settings = (
+            generate_ble_advertise_objects(self.dut.droid))
+        self.dut.droid.bleStartBleAdvertising(
+            advertise_callback, advertise_data, advertise_settings)
+        if self._verify_ble_adv_started(advertise_callback):
+            self.log.info(
+                "Tracking Callback ID: {}".format(advertise_callback))
+            self.advertisement_list.append(advertise_callback)
+            self.log.info(self.advertisement_list)
+
+    def stop_all_advertisements(self, line):
+        """Stop all LE advertisements"""
+        for callback_id in self.advertisement_list:
+            self.log.info("Stopping Advertisement {}".format(callback_id))
+            self.dut.droid.bleStopBleAdvertising(callback_id)
+            time.sleep(1)
+        self.advertisement_list = []
+
+    def do_ble_stop_advertisement(self, callback_id):
+        """Stop an LE advertisement"""
+        if not callback_id:
+            self.log.info("Need a callback ID")
+            return
+        callback_id = int(callback_id)
+        if callback_id not in self.advertisement_list:
+            self.log.info("Callback not in list of advertisements.")
+            return
+        self.dut.droid.bleStopBleAdvertising(callback_id)
+        self.advertisement_list.remove(callback_id)
diff --git a/acts/tests/google/bt/pts/cmd_input.py b/acts/tests/google/bt/pts/cmd_input.py
index e14ef06f6b..2217f76448 100644
--- a/acts/tests/google/bt/pts/cmd_input.py
+++ b/acts/tests/google/bt/pts/cmd_input.py
@@ -18,6 +18,7 @@ Python script for wrappers to various libraries.
 """
 
 from acts.test_utils.bt.GattEnum import GattServerResponses
+from ble_lib import BleLib
 from gattc_lib import GattClientLib
 from gatts_lib import GattServerLib
 
@@ -42,6 +43,7 @@ class CmdInput(cmd.Cmd):
         self.log = log
 
         # Initialize libraries
+        self.ble_lib = BleLib(log, mac_addr, self.pri_dut)
         self.gattc_lib = GattClientLib(log, mac_addr, self.pri_dut)
         self.gatts_lib = GattServerLib(log, mac_addr, self.pri_dut)
 
@@ -469,3 +471,116 @@ class CmdInput(cmd.Cmd):
             self.log.info(FAILURE.format(cmd, err))
 
     """End GATT Server wrappers"""
+    """Begin Ble wrappers"""
+
+    def complete_ble_adv_data_include_local_name(self, text, line, begidx,
+                                                 endidx):
+        options = ['true', 'false']
+        if not text:
+            completions = list(options)[:]
+        else:
+            completions = [s for s in options if s.startswith(text)]
+        return completions
+
+    def complete_ble_adv_data_include_tx_power_level(self, text, line, begidx,
+                                                     endidx):
+        options = ['true', 'false']
+        if not text:
+            completions = list(options)[:]
+        else:
+            completions = [s for s in options if s.startswith(text)]
+        return completions
+
+    def complete_ble_stop_advertisement(self, text, line, begidx, endidx):
+        str_adv_list = list(map(str, self.ble_lib.ADVERTISEMENT_LIST))
+        if not text:
+            completions = str_adv_list[:]
+        else:
+            completions = [s for s in str_adv_list if s.startswith(text)]
+        return completions
+
+    def do_ble_start_generic_connectable_advertisement(self, line):
+        """Start a connectable LE advertisement"""
+        cmd = "Start a connectable LE advertisement"
+        try:
+            self.ble_lib.start_generic_connectable_advertisement(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    def do_ble_start_connectable_advertisement_set(self, line):
+        """Start a connectable advertisement set"""
+        try:
+            self.ble_lib.start_connectable_advertisement_set(line)
+        except Exception as err:
+            self.log.error("Failed to start advertisement: {}".format(err))
+
+    def do_ble_stop_all_advertisement_set(self, line):
+        """Stop all advertisement sets"""
+        try:
+            self.ble_lib.stop_all_advertisement_set(line)
+        except Exception as err:
+            self.log.error("Failed to stop advertisement: {}".format(err))
+
+    def do_ble_adv_add_service_uuid_list(self, line):
+        """Add service UUID to the LE advertisement inputs:
+         [uuid1 uuid2 ... uuidN]"""
+        cmd = "Add a valid service UUID to the advertisement."
+        try:
+            self.ble_lib.adv_add_service_uuid_list(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    def do_ble_adv_data_include_local_name(self, line):
+        """Include local name in the advertisement. inputs: [true|false]"""
+        cmd = "Include local name in the advertisement."
+        try:
+            self.ble_lib.adv_data_include_local_name(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    def do_ble_adv_data_include_tx_power_level(self, line):
+        """Include tx power level in the advertisement. inputs: [true|false]"""
+        cmd = "Include local name in the advertisement."
+        try:
+            self.ble_lib.adv_data_include_tx_power_level(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    def do_ble_adv_data_add_manufacturer_data(self, line):
+        """Include manufacturer id and data to the advertisment:
+        [id data1 data2 ... dataN]"""
+        cmd = "Include manufacturer id and data to the advertisment."
+        try:
+            self.ble_lib.adv_data_add_manufacturer_data(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    def do_ble_start_generic_nonconnectable_advertisement(self, line):
+        """Start a nonconnectable LE advertisement"""
+        cmd = "Start a nonconnectable LE advertisement"
+        try:
+            self.ble_lib.start_generic_connectable_advertisement(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    def do_ble_list_active_advertisement_ids(self, line):
+        """List all active BLE advertisements"""
+        self.log.info("IDs: {}".format(self.ble_lib.advertisement_list))
+
+    def do_ble_stop_all_advertisements(self, line):
+        """Stop all LE advertisements"""
+        cmd = "Stop all LE advertisements"
+        try:
+            self.ble_lib.stop_all_advertisements(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    def do_ble_stop_advertisement(self, line):
+        """Stop an LE advertisement"""
+        cmd = "Stop a connectable LE advertisement"
+        try:
+            self.do_ble_stop_advertisement(line)
+        except Exception as err:
+            self.log.info(FAILURE.format(cmd, err))
+
+    """End Ble wrappers"""