xref: /tools/test/connectivity/acts_tests/tests/google/fuchsia/bt/command_input.py

#!/usr/bin/env python3
#
# Copyright (C) 2019 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.
"""
Python script for wrappers to various libraries.

Class CmdInput inherts from the cmd library.

Functions that start with "do_" have a method
signature that doesn't match the actual command
line command and that is intended. This is so the
"help" command knows what to display (in this case
the documentation of the command itself).

For example:
Looking at the function "do_tool_set_target_device_name"
has the inputs self and line which is expected of this type
of method signature. When the "help" command is done on the
method name you get the function documentation as such:

(Cmd) help tool_set_target_device_name

        Description: Reset the target device name.
        Input(s):
            device_name: Required. The advertising name to connect to.
        Usage: tool_set_target_device_name new_target_device name
          Examples:
            tool_set_target_device_name le_watch

This is all to say this documentation pattern is expected.

"""

from acts_contrib.test_utils.audio_analysis_lib.check_quality import quality_analysis
from acts_contrib.test_utils.bt.bt_constants import audio_bits_per_sample_32
from acts_contrib.test_utils.bt.bt_constants import audio_sample_rate_48000
from acts_contrib.test_utils.abstract_devices.bluetooth_device import create_bluetooth_device
from acts_contrib.test_utils.bt.bt_constants import bt_attribute_values
from acts_contrib.test_utils.bt.bt_constants import sig_appearance_constants
from acts_contrib.test_utils.bt.bt_constants import sig_uuid_constants
from acts_contrib.test_utils.fuchsia.sdp_records import sdp_pts_record_list

import acts_contrib.test_utils.bt.gatt_test_database as gatt_test_database

import cmd
import pprint
import time
"""Various Global Strings"""
BASE_UUID = sig_uuid_constants['BASE_UUID']
CMD_LOG = "CMD {} result: {}"
FAILURE = "CMD {} threw exception: {}"
BASIC_ADV_NAME = "fs_test"


class CommandInput(cmd.Cmd):
    ble_adv_interval = 1000
    ble_adv_appearance = None
    ble_adv_data_include_tx_power_level = False
    ble_adv_include_name = True
    ble_adv_include_scan_response = False
    ble_adv_name = "fs_test"
    ble_adv_data_manufacturer_data = None
    ble_adv_data_service_data = None
    ble_adv_data_service_uuid_list = None
    ble_adv_data_uris = None

    bt_control_ids = []
    bt_control_names = []
    bt_control_devices = []
    bt_scan_poll_timer = 0.5
    target_device_name = ""
    le_ids = []
    unique_mac_addr_id = None

    def setup_vars(self, dut, target_device_name, log):
        self.pri_dut = dut
        # Note: test_dut is the start of a slow conversion from a Fuchsia specific
        # Tool to an abstract_device tool. Only commands that use test_dut will work
        # Otherwise this tool is primarially targeted at Fuchsia devices.
        self.test_dut = create_bluetooth_device(self.pri_dut)
        self.test_dut.initialize_bluetooth_controller()
        self.target_device_name = target_device_name
        self.log = log

    def emptyline(self):
        pass

    def do_EOF(self, line):
        "End Script"
        return True

    """ Useful Helper functions and cmd line tooling """

    def str_to_bool(self, s):
        if s.lower() == 'true':
            return True
        elif s.lower() == 'false':
            return False

    def _find_unique_id_over_le(self):
        scan_filter = {"name_substring": self.target_device_name}
        self.unique_mac_addr_id = None
        self.pri_dut.gattc_lib.bleStartBleScan(scan_filter)
        tries = 10
        for i in range(tries):
            time.sleep(self.bt_scan_poll_timer)
            scan_res = self.pri_dut.gattc_lib.bleGetDiscoveredDevices(
            )['result']
            for device in scan_res:
                name, did, connectable = device["name"], device["id"], device[
                    "connectable"]
                if (self.target_device_name in name):
                    self.unique_mac_addr_id = did
                    self.log.info(
                        "Successfully found device: name, id: {}, {}".format(
                            name, did))
                    break
            if self.unique_mac_addr_id:
                break
        self.pri_dut.gattc_lib.bleStopBleScan()

    def _find_unique_id_over_bt_control(self):
        self.unique_mac_addr_id = None
        self.bt_control_devices = []
        self.pri_dut.bts_lib.requestDiscovery(True)
        tries = 10
        for i in range(tries):
            if self.unique_mac_addr_id:
                break
            time.sleep(self.bt_scan_poll_timer)
            device_list = self.pri_dut.bts_lib.getKnownRemoteDevices(
            )['result']
            for id_dict in device_list:
                device = device_list[id_dict]
                self.bt_control_devices.append(device)
                name = None
                if device['name'] is not None:
                    name = device['name']
                did, address = device['id'], device['address']

                self.bt_control_ids.append(did)
                if name is not None:
                    self.bt_control_names.append(name)
                    if self.target_device_name in name:
                        self.unique_mac_addr_id = did
                        self.log.info(
                            "Successfully found device: name, id, address: {}, {}, {}"
                            .format(name, did, address))
                        break
        self.pri_dut.bts_lib.requestDiscovery(False)

    def do_tool_take_bt_snoop_log(self, custom_name):
        """
        Description: Takes the bt snoop log from the Fuchsia device.
        Logs will show up in your config files' logpath directory.

        Input(s):
            custom_name: Optional. Override the default pcap file name.

        Usage: tool_set_target_device_name new_target_device name
          Examples:
            tool_take_bt_snoop_log connection_error
            tool_take_bt_snoop_log
        """
        self.pri_dut.take_bt_snoop_log(custom_name)

    def do_tool_refresh_unique_id(self, line):
        """
        Description: Refresh command line tool mac unique id.
        Usage:
          Examples:
            tool_refresh_unique_id
        """
        try:
            self._find_unique_id_over_le()
        except Exception as err:
            self.log.error(
                "Failed to scan or find scan result: {}".format(err))

    def do_tool_refresh_unique_id_using_bt_control(self, line):
        """
        Description: Refresh command line tool mac unique id.
        Usage:
          Examples:
            tool_refresh_unique_id_using_bt_control
        """
        try:
            self._find_unique_id_over_bt_control()
        except Exception as err:
            self.log.error(
                "Failed to scan or find scan result: {}".format(err))

    def do_tool_set_target_device_name(self, line):
        """
        Description: Reset the target device name.
        Input(s):
            device_name: Required. The advertising name to connect to.
        Usage: tool_set_target_device_name new_target_device name
          Examples:
            tool_set_target_device_name le_watch
        """
        self.log.info("Setting target_device_name to: {}".format(line))
        self.target_device_name = line

    def do_tool_set_unique_mac_addr_id(self, line):
        """
        Description: Sets the unique mac address id (Specific to Fuchsia)
        Input(s):
            device_id: Required. The id to set the unique mac address id to
        Usage: tool_set_unique_mac_addr_id device_id
          Examples:
            tool_set_unique_mac_addr_id 7fb2cae53aad9e0d
        """
        self.unique_mac_addr_id = line

    """Begin BLE advertise wrappers"""

    def complete_ble_adv_data_include_name(self, text, line, begidx, endidx):
        roles = ["true", "false"]
        if not text:
            completions = roles
        else:
            completions = [s for s in roles if s.startswith(text)]
        return completions

    def do_ble_adv_data_include_name(self, line):
        cmd = "Include name in the advertisement."
        try:
            self.ble_adv_include_name = self.str_to_bool(line)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_ble_adv_data_set_name(self, line):
        cmd = "Set the name to be included in the advertisement."
        try:
            self.ble_adv_name = line
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def complete_ble_adv_data_set_appearance(self, text, line, begidx, endidx):
        if not text:
            completions = list(sig_appearance_constants.keys())
        else:
            completions = [
                s for s in sig_appearance_constants.keys()
                if s.startswith(text)
            ]
        return completions

    def do_ble_adv_data_set_appearance(self, line):
        cmd = "Set the appearance to known SIG values."
        try:
            self.ble_adv_appearance = line
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    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 do_ble_adv_data_include_tx_power_level(self, line):
        """Include the tx_power_level in the advertising data.
        Description: Adds tx_power_level to the advertisement data to the BLE
            advertisement.
        Input(s):
            value: Required. True or False
        Usage: ble_adv_data_include_tx_power_level bool_value
          Examples:
            ble_adv_data_include_tx_power_level true
            ble_adv_data_include_tx_power_level false
        """
        cmd = "Include tx_power_level in advertisement."
        try:
            self.ble_adv_data_include_tx_power_level = self.str_to_bool(line)
        except Exception as err:
            self.log.info(FAILURE.format(cmd, err))

    def complete_ble_adv_include_scan_response(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 do_ble_adv_include_scan_response(self, line):
        """Include scan response in advertisement. inputs: [true|false]
            Note: Currently just sets the scan response data to the
                Advertisement data.
        """
        cmd = "Include tx_power_level in advertisement."
        try:
            self.ble_adv_include_scan_response = self.str_to_bool(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
        Description: Adds manufacturer data to the BLE advertisement.
        Input(s):
            id: Required. The int representing the manufacturer id.
            data: Required. The string representing the data.
        Usage: ble_adv_data_add_manufacturer_data id data
          Examples:
            ble_adv_data_add_manufacturer_data 1 test
        """
        cmd = "Include manufacturer id and data to the advertisment."
        try:

            info = line.split()
            if self.ble_adv_data_manufacturer_data is None:
                self.ble_adv_data_manufacturer_data = []
            self.ble_adv_data_manufacturer_data.append({
                "id": int(info[0]),
                "data": info[1]
            })
        except Exception as err:
            self.log.info(FAILURE.format(cmd, err))

    def do_ble_adv_data_add_service_data(self, line):
        """Include service data to the advertisment
        Description: Adds service data to the BLE advertisement.
        Input(s):
            uuid: Required. The string representing the uuid.
            data: Required. The string representing the data.
        Usage: ble_adv_data_add_service_data uuid data
          Examples:
            ble_adv_data_add_service_data 00001801-0000-1000-8000-00805f9b34fb test
        """
        cmd = "Include manufacturer id and data to the advertisment."
        try:
            info = line.split()
            if self.ble_adv_data_service_data is None:
                self.ble_adv_data_service_data = []
            self.ble_adv_data_service_data.append({
                "uuid": info[0],
                "data": info[1]
            })
        except Exception as err:
            self.log.info(FAILURE.format(cmd, err))

    def do_ble_adv_add_service_uuid_list(self, line):
        """Include a list of service uuids to the advertisment:
        Description: Adds service uuid list to the BLE advertisement.
        Input(s):
            uuid: Required. A list of N string UUIDs to add.
        Usage: ble_adv_add_service_uuid_list uuid0 uuid1 ... uuidN
          Examples:
            ble_adv_add_service_uuid_list 00001801-0000-1000-8000-00805f9b34fb
            ble_adv_add_service_uuid_list 00001801-0000-1000-8000-00805f9b34fb 00001802-0000-1000-8000-00805f9b34fb
        """
        cmd = "Include service uuid list to the advertisment data."
        try:
            self.ble_adv_data_service_uuid_list = line
        except Exception as err:
            self.log.info(FAILURE.format(cmd, err))

    def do_ble_adv_data_set_uris(self, uris):
        """Set the URIs of the LE advertisement data:
        Description: Adds list of String UIRs
          See (RFC 3986 1.1.2 https://tools.ietf.org/html/rfc3986)
          Valid URI examples:
            ftp://ftp.is.co.za/rfc/rfc1808.txt
            http://www.ietf.org/rfc/rfc2396.txt
            ldap://[2001:db8::7]/c=GB?objectClass?one
            mailto:John.Doe@example.com
            news:comp.infosystems.www.servers.unix
            tel:+1-816-555-1212
            telnet://192.0.2.16:80/
            urn:oasis:names:specification:docbook:dtd:xml:4.1.2
        Input(s):
            uris: Required. A list of URIs to add.
        Usage: ble_adv_data_set_uris uri0 uri1 ... uriN
          Examples:
            ble_adv_data_set_uris telnet://192.0.2.16:80/
            ble_adv_data_set_uris tel:+1-816-555-1212
        """
        cmd = "Set the appearance to known SIG values."
        try:
            self.ble_adv_data_uris = uris.split()
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def start_advertisement(self, connectable):
        """ Handle setting advertising data and the advertisement
            Note: After advertisement is successful, clears values set for
                * Manufacturer data
                * Appearance information
                * Scan Response
                * Service UUIDs
                * URI list
            Args:
                connectable: Bool of whether to start a connectable
                    advertisement or not.
        """
        adv_data_name = self.ble_adv_name
        if not self.ble_adv_include_name:
            adv_data_name = None

        manufacturer_data = self.ble_adv_data_manufacturer_data

        tx_power_level = None
        if self.ble_adv_data_include_tx_power_level:
            tx_power_level = 1  # Not yet implemented so set to 1

        scan_response = self.ble_adv_include_scan_response

        adv_data = {
            "name": adv_data_name,
            "appearance": self.ble_adv_appearance,
            "service_data": self.ble_adv_data_service_data,
            "tx_power_level": tx_power_level,
            "service_uuids": self.ble_adv_data_service_uuid_list,
            "manufacturer_data": manufacturer_data,
            "uris": self.ble_adv_data_uris,
        }

        if not self.ble_adv_include_scan_response:
            scan_response = None
        else:
            scan_response = adv_data

        result = self.pri_dut.ble_lib.bleStartBleAdvertising(
            adv_data, scan_response, self.ble_adv_interval, connectable)
        self.log.info("Result of starting advertisement: {}".format(result))
        self.ble_adv_data_manufacturer_data = None
        self.ble_adv_appearance = None
        self.ble_adv_include_scan_response = False
        self.ble_adv_data_service_uuid_list = None
        self.ble_adv_data_uris = None
        self.ble_adv_data_service_data = None

    def do_ble_start_generic_connectable_advertisement(self, line):
        """
        Description: Start a connectable LE advertisement

        Usage: ble_start_generic_connectable_advertisement
        """
        cmd = "Start a connectable LE advertisement"
        try:
            connectable = True
            self.start_advertisement(connectable)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_ble_start_generic_nonconnectable_advertisement(self, line):
        """
        Description: Start a non-connectable LE advertisement

        Usage: ble_start_generic_nonconnectable_advertisement
        """
        cmd = "Start a nonconnectable LE advertisement"
        try:
            connectable = False
            self.start_advertisement(connectable)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_ble_stop_advertisement(self, line):
        """
        Description: Stop a BLE advertisement.
        Usage: ble_stop_advertisement
        """
        cmd = "Stop a connectable LE advertisement"
        try:
            self.pri_dut.ble_lib.bleStopBleAdvertising()
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End BLE advertise wrappers"""
    """Begin GATT client wrappers"""

    def complete_gattc_connect_by_id(self, text, line, begidx, endidx):
        if not text:
            completions = list(self.le_ids)[:]
        else:
            completions = [s for s in self.le_ids if s.startswith(text)]
        return completions

    def do_gattc_connect_by_id(self, line):
        """
        Description: Connect to a LE peripheral.
        Input(s):
            device_id: Required. The unique device ID from Fuchsia
                discovered devices.
        Usage:
          Examples:
            gattc_connect device_id
        """
        cmd = "Connect to a LE peripheral by input ID."
        try:

            connection_status = self.pri_dut.gattc_lib.bleConnectToPeripheral(
                line)
            self.log.info("Connection status: {}".format(
                pprint.pformat(connection_status)))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_connect(self, line):
        """
        Description: Connect to a LE peripheral.
        Optional input: device_name
        Input(s):
            device_name: Optional. The peripheral ID to connect to.
        Usage:
          Examples:
            gattc_connect
            gattc_connect eddystone_123
        """
        cmd = "Connect to a LE peripheral."
        try:
            if len(line) > 0:
                self.target_device_name = line
                self.unique_mac_addr_id = None
            if not self.unique_mac_addr_id:
                try:
                    self._find_unique_id()
                except Exception as err:
                    self.log.info("Failed to scan or find device.")
                    return
            connection_status = self.pri_dut.gattc_lib.bleConnectToPeripheral(
                self.unique_mac_addr_id)
            self.log.info("Connection status: {}".format(
                pprint.pformat(connection_status)))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_connect_disconnect_iterations(self, line):
        """
        Description: Connect then disconnect to a LE peripheral multiple times.
        Input(s):
            iterations: Required. The number of iterations to run.
        Usage:
          Examples:
            gattc_connect_disconnect_iterations 10
        """
        cmd = "Connect to a LE peripheral."
        try:
            if not self.unique_mac_addr_id:
                try:
                    self._find_unique_id()
                except Exception as err:
                    self.log.info("Failed to scan or find device.")
                    return
            for i in range(int(line)):
                self.log.info("Running iteration {}".format(i + 1))
                connection_status = self.pri_dut.gattc_lib.bleConnectToPeripheral(
                    self.unique_mac_addr_id)
                self.log.info("Connection status: {}".format(
                    pprint.pformat(connection_status)))
                time.sleep(4)
                disc_status = self.pri_dut.gattc_lib.bleDisconnectPeripheral(
                    self.unique_mac_addr_id)
                self.log.info("Disconnect status: {}".format(disc_status))
                time.sleep(3)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_disconnect(self, line):
        """
        Description: Disconnect from LE peripheral.
        Assumptions: Already connected to a peripheral.
        Usage:
          Examples:
            gattc_disconnect
        """
        cmd = "Disconenct from LE peripheral."
        try:
            disconnect_status = self.pri_dut.gattc_lib.bleDisconnectPeripheral(
                self.unique_mac_addr_id)
            self.log.info("Disconnect status: {}".format(disconnect_status))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_list_services(self, discover_chars):
        """
        Description: List services from LE peripheral.
        Assumptions: Already connected to a peripheral.
        Input(s):
            discover_chars: Optional. An optional input to discover all
                characteristics on the service.
        Usage:
          Examples:
            gattc_list_services
            gattc_list_services true
        """
        cmd = "List services from LE peripheral."
        try:

            services = self.pri_dut.gattc_lib.listServices(
                self.unique_mac_addr_id)
            self.log.info("Discovered Services: \n{}".format(
                pprint.pformat(services)))
            discover_characteristics = self.str_to_bool(discover_chars)
            if discover_chars:
                for service in services.get('result'):
                    self.pri_dut.gattc_lib.connectToService(
                        self.unique_mac_addr_id, service.get('id'))
                    chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                    self.log.info("Discovered chars:\n{}".format(
                        pprint.pformat(chars)))

        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_connect_to_service(self, line):
        """
        Description: Connect to Peripheral GATT server service.
        Assumptions: Already connected to peripheral.
        Input(s):
            service_id: Required. The service id reference on the GATT server.
        Usage:
          Examples:
            gattc_connect_to_service service_id
        """
        cmd = "GATT client connect to GATT server service."
        try:
            self.pri_dut.gattc_lib.connectToService(self.unique_mac_addr_id,
                                                    int(line))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_discover_characteristics(self, line):
        """
        Description: Discover characteristics from a connected service.
        Assumptions: Already connected to a GATT server service.
        Usage:
          Examples:
            gattc_discover_characteristics
        """
        cmd = "Discover and list characteristics from a GATT server."
        try:
            chars = self.pri_dut.gattc_lib.discoverCharacteristics()
            self.log.info("Discovered chars:\n{}".format(
                pprint.pformat(chars)))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_notify_all_chars(self, line):
        """
        Description: Enable all notifications on all Characteristics on
            a GATT server.
        Assumptions: Basic GATT connection made.
        Usage:
          Examples:
            gattc_notify_all_chars
        """
        cmd = "Read all characteristics from the GATT service."
        try:
            services = self.pri_dut.gattc_lib.listServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Reading chars in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    # quick char filter for apple-4 test... remove later
                    print("found uuid {}".format(char_uuid))
                    try:
                        self.pri_dut.gattc_lib.enableNotifyCharacteristic(
                            char_id)
                    except Exception as err:
                        print("error enabling notification")
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_all_chars(self, line):
        """
        Description: Read all Characteristic values from a GATT server across
            all services.
        Assumptions: Basic GATT connection made.
        Usage:
          Examples:
            gattc_read_all_chars
        """
        cmd = "Read all characteristics from the GATT service."
        try:
            services = self.pri_dut.gattc_lib.listServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Reading chars in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    try:
                        read_val =  \
                            self.pri_dut.gattc_lib.readCharacteristicById(
                                char_id)
                        print("  Characteristic uuid / Value: {} / {}".format(
                            char_uuid, read_val['result']))
                        str_value = ""
                        for val in read_val['result']:
                            str_value += chr(val)
                        print("    str val: {}".format(str_value))
                    except Exception as err:
                        print(err)
                        pass
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_all_desc(self, line):
        """
        Description: Read all Descriptors values from a GATT server across
            all services.
        Assumptions: Basic GATT connection made.
        Usage:
          Examples:
            gattc_read_all_chars
        """
        cmd = "Read all descriptors from the GATT service."
        try:
            services = self.pri_dut.gattc_lib.listServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Reading descs in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    descriptors = char['descriptors']
                    print("  Reading descs in char uuid: {}".format(char_uuid))
                    for desc in descriptors:
                        desc_id = desc["id"]
                        desc_uuid = desc["uuid_type"]
                    try:
                        read_val = self.pri_dut.gattc_lib.readDescriptorById(
                            desc_id)
                        print("    Descriptor uuid / Value: {} / {}".format(
                            desc_uuid, read_val['result']))
                    except Exception as err:
                        pass
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_all_desc(self, line):
        """
        Description: Write a value to all Descriptors on the GATT server.
        Assumptions: Basic GATT connection made.
        Input(s):
            offset: Required. The offset to start writing to.
            size: Required. The size of bytes to write (value will be generated).
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
        Usage:
          Examples:
            gattc_write_all_desc 0 100
            gattc_write_all_desc 10 2
        """
        cmd = "Read all descriptors from the GATT service."
        try:
            args = line.split()
            if len(args) != 2:
                self.log.info("2 Arguments required: [Offset] [Size]")
                return
            offset = int(args[0])
            size = args[1]
            write_value = []
            for i in range(int(size)):
                write_value.append(i % 256)
            services = self.pri_dut.gattc_lib.listServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Writing descs in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    descriptors = char['descriptors']
                    print("  Reading descs in char uuid: {}".format(char_uuid))
                    for desc in descriptors:
                        desc_id = desc["id"]
                        desc_uuid = desc["uuid_type"]
                    try:
                        write_val = self.pri_dut.gattc_lib.writeDescriptorById(
                            desc_id, offset, write_value)
                        print("    Descriptor uuid / Result: {} / {}".format(
                            desc_uuid, write_val['result']))
                    except Exception as err:
                        pass
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_all_long_desc(self, line):
        """
        Description: Read all long Characteristic Descriptors
        Assumptions: Basic GATT connection made.
        Input(s):
            offset: Required. The offset to start reading from.
            max_bytes: Required. The max size of bytes to return.
        Usage:
          Examples:
            gattc_read_all_long_desc 0 100
            gattc_read_all_long_desc 10 20
        """
        cmd = "Read all long descriptors from the GATT service."
        try:
            args = line.split()
            if len(args) != 2:
                self.log.info("2 Arguments required: [Offset] [Size]")
                return
            offset = int(args[0])
            max_bytes = int(args[1])
            services = self.pri_dut.ble_lib.bleListServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Reading descs in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    descriptors = char['descriptors']
                    print("  Reading descs in char uuid: {}".format(char_uuid))
                    for desc in descriptors:
                        desc_id = desc["id"]
                        desc_uuid = desc["uuid_type"]
                    try:
                        read_val = self.pri_dut.gattc_lib.readLongDescriptorById(
                            desc_id, offset, max_bytes)
                        print("    Descriptor uuid / Result: {} / {}".format(
                            desc_uuid, read_val['result']))
                    except Exception as err:
                        pass
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_all_long_char(self, line):
        """
        Description: Read all long Characteristic
        Assumptions: Basic GATT connection made.
        Input(s):
            offset: Required. The offset to start reading from.
            max_bytes: Required. The max size of bytes to return.
        Usage:
          Examples:
            gattc_read_all_long_char 0 100
            gattc_read_all_long_char 10 20
        """
        cmd = "Read all long Characteristics from the GATT service."
        try:
            args = line.split()
            if len(args) != 2:
                self.log.info("2 Arguments required: [Offset] [Size]")
                return
            offset = int(args[0])
            max_bytes = int(args[1])
            services = self.pri_dut.ble_lib.bleListServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Reading chars in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    try:
                        read_val = self.pri_dut.gattc_lib.readLongCharacteristicById(
                            char_id, offset, max_bytes)
                        print("    Char uuid / Result: {} / {}".format(
                            char_uuid, read_val['result']))
                    except Exception as err:
                        pass
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_all_chars(self, line):
        """
        Description: Write all characteristic values from a GATT server across
            all services.
        Assumptions: Basic GATT connection made.
        Input(s):
            offset: Required. The offset to start writing on.
            size: The write value size (value will be generated)
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
        Usage:
          Examples:
            gattc_write_all_chars 0 10
            gattc_write_all_chars 10 1
        """
        cmd = "Read all characteristics from the GATT service."
        try:
            args = line.split()
            if len(args) != 2:
                self.log.info("2 Arguments required: [Offset] [Size]")
                return
            offset = int(args[0])
            size = int(args[1])
            write_value = []
            for i in range(size):
                write_value.append(i % 256)
            services = self.pri_dut.gattc_lib.listServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Writing chars in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    try:
                        write_result = self.pri_dut.gattc_lib.writeCharById(
                            char_id, offset, write_value)
                        print("  Characteristic uuid write result: {} / {}".
                              format(char_uuid, write_result['result']))
                    except Exception as err:
                        print("error writing char {}".format(err))
                        pass
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_all_chars_without_response(self, line):
        """
        Description: Write all characteristic values from a GATT server across
            all services.
        Assumptions: Basic GATT connection made.
        Input(s):
            size: The write value size (value will be generated).
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
        Usage:
          Examples:
            gattc_write_all_chars_without_response 100
        """
        cmd = "Read all characteristics from the GATT service."
        try:
            args = line.split()
            if len(args) != 1:
                self.log.info("1 Arguments required: [Size]")
                return
            size = int(args[0])
            write_value = []
            for i in range(size):
                write_value.append(i % 256)
            services = self.pri_dut.gattc_lib.listServices(
                self.unique_mac_addr_id)
            for service in services['result']:
                service_id = service['id']
                service_uuid = service['uuid_type']
                self.pri_dut.gattc_lib.connectToService(
                    self.unique_mac_addr_id, service_id)
                chars = self.pri_dut.gattc_lib.discoverCharacteristics()
                print("Reading chars in service uuid: {}".format(service_uuid))

                for char in chars['result']:
                    char_id = char['id']
                    char_uuid = char['uuid_type']
                    try:
                        write_result = \
                            self.pri_dut.gattc_lib.writeCharByIdWithoutResponse(
                                char_id, write_value)
                        print("  Characteristic uuid write result: {} / {}".
                              format(char_uuid, write_result['result']))
                    except Exception as err:
                        pass
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_char_by_id(self, line):
        """
        Description: Write char by characteristic id reference.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
            offset: The offset value to use
            size: Function will generate random bytes by input size.
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
        Usage:
          Examples:
            gattc_write_char_by_id char_id 0 5
            gattc_write_char_by_id char_id 20 1
        """
        cmd = "Write to GATT server characteristic ."
        try:
            args = line.split()
            if len(args) != 3:
                self.log.info("3 Arguments required: [Id] [Offset] [Size]")
                return
            id = int(args[0], 16)
            offset = int(args[1])
            size = int(args[2])
            write_value = []
            for i in range(size):
                write_value.append(i % 256)
            self.test_dut.gatt_client_write_characteristic_by_handle(
                self.unique_mac_addr_id, id, offset, write_value)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_long_char_by_id(self, line):
        """
        Description: Write long char by characteristic id reference.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
            offset: The offset value to use
            size: Function will generate random bytes by input size.
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
            reliable_mode: Optional: Reliable writes represented as bool
        Usage:
          Examples:
            gattc_write_long_char_by_id char_id 0 5
            gattc_write_long_char_by_id char_id 20 1
            gattc_write_long_char_by_id char_id 20 1 true
            gattc_write_long_char_by_id char_id 20 1 false
        """
        cmd = "Long Write to GATT server characteristic ."
        try:
            args = line.split()
            if len(args) < 3:
                self.log.info("3 Arguments required: [Id] [Offset] [Size]")
                return
            id = int(args[0], 16)
            offset = int(args[1])
            size = int(args[2])
            reliable_mode = False
            if len(args) > 3:
                reliable_mode = self.str_to_bool(args[3])
            write_value = []
            for i in range(size):
                write_value.append(i % 256)
            self.test_dut.gatt_client_write_long_characteristic_by_handle(
                self.unique_mac_addr_id, id, offset, write_value,
                reliable_mode)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_long_desc_by_id(self, line):
        """
        Description: Write long char by descrioptor id reference.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
            offset: The offset value to use
            size: Function will generate random bytes by input size.
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
        Usage:
          Examples:
            gattc_write_long_desc_by_id char_id 0 5
            gattc_write_long_desc_by_id char_id 20 1
        """
        cmd = "Long Write to GATT server descriptor ."
        try:
            args = line.split()
            if len(args) != 3:
                self.log.info("3 Arguments required: [Id] [Offset] [Size]")
                return
            id = int(args[0], 16)
            offset = int(args[1])
            size = int(args[2])
            write_value = []
            for i in range(size):
                write_value.append(i % 256)
            self.test_dut.gatt_client_write_long_descriptor_by_handle(
                self.unique_mac_addr_id, id, offset, write_value)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_char_by_id_without_response(self, line):
        """
        Description: Write char by characteristic id reference without response.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
            size: Function will generate random bytes by input size.
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
        Usage:
          Examples:
            gattc_write_char_by_id_without_response char_id 5
        """
        cmd = "Write characteristic by id without response."
        try:
            args = line.split()
            if len(args) != 2:
                self.log.info("2 Arguments required: [Id] [Size]")
                return
            id = int(args[0], 16)
            size = args[1]
            write_value = []
            for i in range(int(size)):
                write_value.append(i % 256)
            self.test_dut.gatt_client_write_characteristic_without_response_by_handle(
                self.unique_mac_addr_id, id, write_value)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_enable_notify_char_by_id(self, line):
        """
        Description: Enable Characteristic notification on Characteristic ID.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
        Usage:
          Examples:
            gattc_enable_notify_char_by_id char_id
        """
        cmd = "Enable notifications by Characteristic id."
        try:
            id = int(line, 16)
            self.test_dut.gatt_client_enable_notifiy_characteristic_by_handle(
                self.unique_mac_addr_id, id)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_disable_notify_char_by_id(self, line):
        """
        Description: Disable Characteristic notification on Characteristic ID.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
        Usage:
          Examples:
            gattc_disable_notify_char_by_id char_id
        """
        cmd = "Disable notify Characteristic by id."
        try:
            id = int(line, 16)
            self.test_dut.gatt_client_disable_notifiy_characteristic_by_handle(
                self.unique_mac_addr_id, id)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_char_by_id(self, line):
        """
        Description: Read Characteristic by ID.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
        Usage:
          Examples:
            gattc_read_char_by_id char_id
        """
        cmd = "Read Characteristic value by ID."
        try:
            id = int(line, 16)
            read_val = self.test_dut.gatt_client_read_characteristic_by_handle(
                self.unique_mac_addr_id, id)
            self.log.info("Characteristic Value with id {}: {}".format(
                id, read_val))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_char_by_uuid(self, characteristic_uuid):
        """
        Description: Read Characteristic by UUID (read by type).
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_uuid: The characteristic id reference on the GATT
                service
        Usage:
          Examples:
            gattc_read_char_by_id char_id
        """
        cmd = "Read Characteristic value by ID."
        try:
            short_uuid_len = 4
            if len(characteristic_uuid) == short_uuid_len:
                characteristic_uuid = BASE_UUID.format(characteristic_uuid)

            read_val = self.test_dut.gatt_client_read_characteristic_by_uuid(
                self.unique_mac_addr_id, characteristic_uuid)
            self.log.info("Characteristic Value with id {}: {}".format(
                id, read_val))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_write_desc_by_id(self, line):
        """
        Description: Write Descriptor by characteristic id reference.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            descriptor_id: The Descriptor id reference on the GATT service
            offset: The offset value to use
            size: Function will generate random bytes by input size.
                IE: Input of 5 will send a byte array of [00, 01, 02, 03, 04]
        Usage:
          Examples:
            gattc_write_desc_by_id desc_id 0 5
            gattc_write_desc_by_id desc_id 20 1
        """
        cmd = "Write Descriptor by id."
        try:
            args = line.split()
            id = int(args[0], 16)
            offset = int(args[1])
            size = args[2]
            write_value = []
            for i in range(int(size)):
                write_value.append(i % 256)
            write_result = self.test_dut.gatt_client_write_descriptor_by_handle(
                self.unique_mac_addr_id, id, offset, write_value)
            self.log.info("Descriptor Write result {}: {}".format(
                id, write_result))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_desc_by_id(self, line):
        """
        Description: Read Descriptor by ID.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            descriptor_id: The Descriptor id reference on the GATT service
        Usage:
          Examples:
            gattc_read_desc_by_id desc_id
        """
        cmd = "Read Descriptor by ID."
        try:
            id = int(line, 16)
            read_val = self.test_dut.gatt_client_read_descriptor_by_handle(
                self.unique_mac_addr_id, id)
            self.log.info("Descriptor Value with id {}: {}".format(
                id, read_val))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_gattc_read_long_char_by_id(self, line):
        """
        Description: Read long Characteristic value by id.
        Assumptions: Already connected to a GATT server service.
        Input(s):
            characteristic_id: The characteristic id reference on the GATT
                service
            offset: The offset value to use.
            max_bytes: The max bytes size to return.
        Usage:
          Examples:
            gattc_read_long_char_by_id char_id 0 10
            gattc_read_long_char_by_id char_id 20 1
        """
        cmd = "Read long Characteristic value by id."
        try:
            args = line.split()
            if len(args) != 3:
                self.log.info("3 Arguments required: [Id] [Offset] [Size]")
                return
            id = int(args[0], 16)
            offset = int(args[1])
            max_bytes = int(args[2])
            read_val = self.test_dut.gatt_client_read_long_characteristic_by_handle(
                self.unique_mac_addr_id, id, offset, max_bytes)
            self.log.info("Characteristic Value with id {}: {}".format(
                id, read_val['result']))

        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End GATT client wrappers"""
    """Begin LE scan wrappers"""

    def _update_scan_results(self, scan_results):
        self.le_ids = []
        for scan in scan_results['result']:
            self.le_ids.append(scan['id'])

    def do_ble_start_scan(self, line):
        """
        Description: Perform a BLE scan.
        Default filter name: ""
        Optional input: filter_device_name
        Usage:
          Examples:
            ble_start_scan
            ble_start_scan eddystone
        """
        cmd = "Perform a BLE scan and list discovered devices."
        try:
            scan_filter = {"name_substring": ""}
            if line:
                scan_filter = {"name_substring": line}
            self.pri_dut.gattc_lib.bleStartBleScan(scan_filter)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_ble_stop_scan(self, line):
        """
        Description: Stops a BLE scan and returns discovered devices.
        Usage:
          Examples:
            ble_stop_scan
        """
        cmd = "Stops a BLE scan and returns discovered devices."
        try:
            scan_results = self.pri_dut.gattc_lib.bleStopBleScan()
            self._update_scan_results(scan_results)
            self.log.info(pprint.pformat(scan_results))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_ble_get_discovered_devices(self, line):
        """
        Description: Get discovered LE devices of an active scan.
        Usage:
          Examples:
            ble_stop_scan
        """
        cmd = "Get discovered LE devices of an active scan."
        try:
            scan_results = self.pri_dut.gattc_lib.bleGetDiscoveredDevices()
            self._update_scan_results(scan_results)
            self.log.info(pprint.pformat(scan_results))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End LE scan wrappers"""
    """Begin GATT Server wrappers"""

    def do_gatts_close(self, line):
        """
        Description: Close active GATT server.

        Usage:
          Examples:
            gatts_close
        """
        cmd = "Close active GATT server."
        try:
            result = self.pri_dut.gatts_lib.closeServer()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def complete_gatts_setup_database(self, text, line, begidx, endidx):
        if not text:
            completions = list(
                gatt_test_database.GATT_SERVER_DB_MAPPING.keys())
        else:
            completions = [
                s for s in gatt_test_database.GATT_SERVER_DB_MAPPING.keys()
                if s.startswith(text)
            ]
        return completions

    def do_gatts_setup_database(self, line):
        """
        Description: Setup a Gatt server database based on pre-defined inputs.
            Supports Tab Autocomplete.
        Input(s):
            descriptor_db_name: The descriptor db name that matches one in
                acts_contrib.test_utils.bt.gatt_test_database
        Usage:
          Examples:
            gatts_setup_database LARGE_DB_1
        """
        cmd = "Setup GATT Server Database Based of pre-defined dictionaries"
        try:
            scan_results = self.pri_dut.gatts_lib.publishServer(
                gatt_test_database.GATT_SERVER_DB_MAPPING.get(line))
            self.log.info(scan_results)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End GATT Server wrappers"""
    """Begin Bluetooth Controller wrappers"""

    def complete_btc_pair(self, text, line, begidx, endidx):
        """ Provides auto-complete for btc_pair cmd.

        See Cmd module for full description.
        """
        arg_completion = len(line.split(" ")) - 1
        pairing_security_level_options = ['ENCRYPTED', 'AUTHENTICATED', 'NONE']
        bondable_options = ['BONDABLE', 'NON_BONDABLE', 'NONE']
        transport_options = ['BREDR', 'LE']
        if arg_completion == 1:
            if not text:
                completions = pairing_security_level_options
            else:
                completions = [
                    s for s in pairing_security_level_options
                    if s.startswith(text)
                ]
            return completions
        if arg_completion == 2:
            if not text:
                completions = bondable_options
            else:
                completions = [
                    s for s in bondable_options if s.startswith(text)
                ]
            return completions
        if arg_completion == 3:
            if not text:
                completions = transport_options
            else:
                completions = [
                    s for s in transport_options if s.startswith(text)
                ]
            return completions

    def do_btc_pair(self, line):
        """
        Description: Sends an outgoing pairing request.

        Input(s):
            pairing security level: ENCRYPTED, AUTHENTICATED, or NONE
            bondable: BONDABLE, NON_BONDABLE, or NONE
            transport: BREDR or LE

        Usage:
          Examples:
            btc_pair NONE NONE BREDR
            btc_pair ENCRYPTED NONE LE
            btc_pair AUTHENTICATED NONE LE
            btc_pair NONE NON_BONDABLE BREDR
        """
        cmd = "Send an outgoing pairing request."
        pairing_security_level_mapping = {
            "ENCRYPTED": 1,
            "AUTHENTICATED": 2,
            "NONE": None,
        }

        bondable_mapping = {
            "BONDABLE": True,
            "NON_BONDABLE": False,
            "NONE": None,
        }

        transport_mapping = {
            "BREDR": 1,
            "LE": 2,
        }

        try:
            options = line.split(" ")
            result = self.test_dut.init_pair(
                self.unique_mac_addr_id,
                pairing_security_level_mapping.get(options[0]),
                bondable_mapping.get(options[1]),
                transport_mapping.get(options[2]),
            )
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def complete_btc_accept_pairing(self, text, line, begidx, endidx):
        """ Provides auto-complete for btc_set_io_capabilities cmd.

        See Cmd module for full description.
        """
        arg_completion = len(line.split(" ")) - 1
        input_options = ['NONE', 'CONFIRMATION', 'KEYBOARD']
        output_options = ['NONE', 'DISPLAY']
        if arg_completion == 1:
            if not text:
                completions = input_options
            else:
                completions = [s for s in input_options if s.startswith(text)]
            return completions
        if arg_completion == 2:
            if not text:
                completions = output_options
            else:
                completions = [s for s in output_options if s.startswith(text)]
            return completions

    def do_btc_accept_pairing(self, line):
        """
        Description: Accept all incoming pairing requests.

        Input(s):
            input: String - The input I/O capabilities to use
                Available Values:
                NONE - Input capability type None
                CONFIRMATION - Input capability type confirmation
                KEYBOARD - Input capability type Keyboard
            output: String - The output I/O Capabilities to use
                Available Values:
                NONE - Output capability type None
                DISPLAY - output capability type Display

        Usage:
          Examples:
            btc_accept_pairing
            btc_accept_pairing NONE DISPLAY
            btc_accept_pairing NONE NONE
            btc_accept_pairing KEYBOARD DISPLAY
        """
        cmd = "Accept incoming pairing requests"
        try:
            input_capabilities = "NONE"
            output_capabilities = "NONE"
            options = line.split(" ")
            if len(options) > 1:
                input_capabilities = options[0]
                output_capabilities = options[1]
            result = self.pri_dut.bts_lib.acceptPairing(
                input_capabilities, output_capabilities)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_forget_device(self, line):
        """
        Description: Forget pairing of the current device under test.
            Current device under test is the device found by
            tool_refresh_unique_id from custom user param. This function
            will also perform a clean disconnect if actively connected.

        Usage:
          Examples:
            btc_forget_device
        """
        cmd = "For pairing of the current device under test."
        try:
            self.log.info("Forgetting device id: {}".format(
                self.unique_mac_addr_id))
            result = self.pri_dut.bts_lib.forgetDevice(self.unique_mac_addr_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_set_discoverable(self, discoverable):
        """
        Description: Change Bluetooth Controller discoverablility.
        Input(s):
            discoverable: true to set discoverable
                          false to set non-discoverable
        Usage:
          Examples:
            btc_set_discoverable true
            btc_set_discoverable false
        """
        cmd = "Change Bluetooth Controller discoverablility."
        try:
            result = self.test_dut.set_discoverable(
                self.str_to_bool(discoverable))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_set_name(self, name):
        """
        Description: Change Bluetooth Controller local name.
        Input(s):
            name: The name to set the Bluetooth Controller name to.

        Usage:
          Examples:
            btc_set_name fs_test
        """
        cmd = "Change Bluetooth Controller local name."
        try:
            result = self.test_dut.set_bluetooth_local_name(name)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_request_discovery(self, discover):
        """
        Description: Change whether the Bluetooth Controller is in active.
            discovery or not.
        Input(s):
            discover: true to start discovery
                      false to end discovery
        Usage:
          Examples:
            btc_request_discovery true
            btc_request_discovery false
        """
        cmd = "Change whether the Bluetooth Controller is in active."
        try:
            result = self.pri_dut.bts_lib.requestDiscovery(
                self.str_to_bool(discover))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_get_known_remote_devices(self, line):
        """
        Description: Get a list of known devices.

        Usage:
          Examples:
            btc_get_known_remote_devices
        """
        cmd = "Get a list of known devices."
        self.bt_control_devices = []
        try:
            device_list = self.pri_dut.bts_lib.getKnownRemoteDevices(
            )['result']
            for id_dict in device_list:
                device = device_list[id_dict]
                self.bt_control_devices.append(device)
                self.log.info("Device found {}".format(device))

        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_forget_all_known_devices(self, line):
        """
        Description: Forget all known devices.

        Usage:
          Examples:
            btc_forget_all_known_devices
        """
        cmd = "Forget all known devices."
        try:
            device_list = self.pri_dut.bts_lib.getKnownRemoteDevices(
            )['result']
            for device in device_list:
                d = device_list[device]
                if d['bonded'] or d['connected']:
                    self.log.info("Unbonding deivce: {}".format(d))
                    self.log.info(
                        self.pri_dut.bts_lib.forgetDevice(d['id'])['result'])
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_connect_device(self, line):
        """
        Description: Connect to device under test.
            Device under test is specified by either user params
            or
                tool_set_target_device_name <name>
                do_tool_refresh_unique_id_using_bt_control

        Usage:
          Examples:
            btc_connect_device
        """
        cmd = "Connect to device under test."
        try:
            result = self.pri_dut.bts_lib.connectDevice(
                self.unique_mac_addr_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def complete_btc_connect_device_by_id(self, text, line, begidx, endidx):
        if not text:
            completions = list(self.bt_control_ids)[:]
        else:
            completions = [
                s for s in self.bt_control_ids if s.startswith(text)
            ]
        return completions

    def do_btc_connect_device_by_id(self, device_id):
        """
        Description: Connect to device id based on pre-defined inputs.
            Supports Tab Autocomplete.
        Input(s):
            device_id: The device id to connect to.

        Usage:
          Examples:
            btc_connect_device_by_id <device_id>
        """
        cmd = "Connect to device id based on pre-defined inputs."
        try:
            result = self.pri_dut.bts_lib.connectDevice(device_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def complete_btc_connect_device_by_name(self, text, line, begidx, endidx):
        if not text:
            completions = list(self.bt_control_names)[:]
        else:
            completions = [
                s for s in self.bt_control_names if s.startswith(text)
            ]
        return completions

    def do_btc_connect_device_by_name(self, device_name):
        """
        Description: Connect to device id based on pre-defined inputs.
            Supports Tab Autocomplete.
        Input(s):
            device_id: The device id to connect to.

        Usage:
          Examples:
            btc_connect_device_by_name <device_id>
        """
        cmd = "Connect to device name based on pre-defined inputs."
        try:
            for device in self.bt_control_devices:
                if device_name is device['name']:

                    result = self.pri_dut.bts_lib.connectDevice(device['id'])
                    self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_disconnect_device(self, line):
        """
        Description: Disconnect to device under test.
            Device under test is specified by either user params
            or
                tool_set_target_device_name <name>
                do_tool_refresh_unique_id_using_bt_control

        Usage:
          Examples:
            btc_disconnect_device
        """
        cmd = "Disconnect to device under test."
        try:
            result = self.pri_dut.bts_lib.disconnectDevice(
                self.unique_mac_addr_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_init_bluetooth_control(self, line):
        """
        Description: Initialize the Bluetooth Controller.

        Usage:
          Examples:
            btc_init_bluetooth_control
        """
        cmd = "Initialize the Bluetooth Controller."
        try:
            result = self.test_dut.initialize_bluetooth_controller()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_get_local_address(self, line):
        """
        Description: Get the local BR/EDR address of the Bluetooth Controller.

        Usage:
          Examples:
            btc_get_local_address
        """
        cmd = "Get the local BR/EDR address of the Bluetooth Controller."
        try:
            result = self.test_dut.get_local_bluetooth_address()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_input_pairing_pin(self, line):
        """
        Description: Sends a pairing pin to SL4F's Bluetooth Control's
        Pairing Delegate.

        Usage:
          Examples:
            btc_input_pairing_pin 123456
        """
        cmd = "Input pairing pin to the Fuchsia device."
        try:
            result = self.pri_dut.bts_lib.inputPairingPin(line)['result']
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_btc_get_pairing_pin(self, line):
        """
        Description: Gets the pairing pin from SL4F's Bluetooth Control's
        Pairing Delegate.

        Usage:
          Examples:
            btc_get_pairing_pin
        """
        cmd = "Get the pairing pin from the Fuchsia device."
        try:
            result = self.pri_dut.bts_lib.getPairingPin()['result']
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End Bluetooth Control wrappers"""
    """Begin Profile Server wrappers"""

    def do_sdp_pts_example(self, num_of_records):
        """
        Description: An example of how to setup a generic SDP record
            and SDP search capabilities. This example will pass a few
            SDP tests.

        Input(s):
            num_of_records: The number of records to add.

        Usage:
          Examples:
            sdp_pts_example 1
            sdp pts_example 10
        """
        cmd = "Setup SDP for PTS testing."

        attributes = [
            bt_attribute_values['ATTR_PROTOCOL_DESCRIPTOR_LIST'],
            bt_attribute_values['ATTR_SERVICE_CLASS_ID_LIST'],
            bt_attribute_values['ATTR_BLUETOOTH_PROFILE_DESCRIPTOR_LIST'],
            bt_attribute_values['ATTR_A2DP_SUPPORTED_FEATURES'],
        ]

        try:
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['AudioSource'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['A/V_RemoteControl'], 16))
            self.pri_dut.sdp_lib.addSearch(attributes,
                                           int(sig_uuid_constants['PANU'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['SerialPort'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['DialupNetworking'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['OBEXObjectPush'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['OBEXFileTransfer'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['Headset'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['HandsfreeAudioGateway'],
                                16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['Handsfree'], 16))
            self.pri_dut.sdp_lib.addSearch(
                attributes, int(sig_uuid_constants['SIM_Access'], 16))
            for i in range(int(num_of_records)):
                result = self.pri_dut.sdp_lib.addService(
                    sdp_pts_record_list[i])
                self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_sdp_cleanup(self, line):
        """
        Description: Cleanup any existing SDP records

        Usage:
          Examples:
            sdp_cleanup
        """
        cmd = "Cleanup SDP objects."
        try:
            result = self.pri_dut.sdp_lib.cleanUp()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_sdp_init(self, line):
        """
        Description: Init the profile proxy for setting up SDP records

        Usage:
          Examples:
            sdp_init
        """
        cmd = "Initialize profile proxy objects for adding SDP records"
        try:
            result = self.pri_dut.sdp_lib.init()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_sdp_connect_l2cap(self, line):
        """
        Description: Send an l2cap connection request over an input psm value.

        Note: Must be already connected to a peer.

        Input(s):
            psm: The int hex value to connect over. Available PSMs:
                SDP 0x0001  See Bluetooth Service Discovery Protocol (SDP)
                RFCOMM  0x0003  See RFCOMM with TS 07.10
                TCS-BIN 0x0005  See Bluetooth Telephony Control Specification /
                    TCS Binary
                TCS-BIN-CORDLESS    0x0007  See Bluetooth Telephony Control
                    Specification / TCS Binary
                BNEP    0x000F  See Bluetooth Network Encapsulation Protocol
                HID_Control 0x0011  See Human Interface Device
                HID_Interrupt   0x0013  See Human Interface Device
                UPnP    0x0015  See [ESDP]
                AVCTP   0x0017  See Audio/Video Control Transport Protocol
                AVDTP   0x0019  See Audio/Video Distribution Transport Protocol
                AVCTP_Browsing  0x001B  See Audio/Video Remote Control Profile
                UDI_C-Plane 0x001D  See the Unrestricted Digital Information
                    Profile [UDI]
                ATT 0x001F  See Bluetooth Core Specification​
                ​3DSP   0x0021​ ​​See 3D Synchronization Profile.
                ​LE_PSM_IPSP    ​0x0023 ​See Internet Protocol Support Profile
                    (IPSP)
                OTS 0x0025  See Object Transfer Service (OTS)
                EATT    0x0027  See Bluetooth Core Specification
            mode: String - The channel mode to connect to. Available values:
                Basic mode: BASIC
                Enhanced Retransmission mode: ERTM

        Usage:
          Examples:
            sdp_connect_l2cap 0001 BASIC
            sdp_connect_l2cap 0019 ERTM
        """
        cmd = "Connect l2cap"
        try:
            info = line.split()
            result = self.pri_dut.sdp_lib.connectL2cap(self.unique_mac_addr_id,
                                                       int(info[0], 16),
                                                       info[1])
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End Profile Server wrappers"""
    """Begin AVDTP wrappers"""

    def do_avdtp_init(self, initiator_delay):
        """
        Description: Init the A2DP component start and AVDTP service to
            initiate.

        Input(s):
            initiator_delay: [Optional] The stream initiator delay to set in
            milliseconds.

        Usage:
          Examples:
            avdtp_init 0
            avdtp_init 2000
            avdtp_init
        """
        cmd = "Initialize AVDTP proxy"
        try:
            if not initiator_delay:
                initiator_delay = None
            result = self.pri_dut.avdtp_lib.init(initiator_delay)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_kill_a2dp(self, line):
        """
        Description: Quickly kill any A2DP components.

        Usage:
          Examples:
            avdtp_kill_a2dp
        """
        cmd = "Kill A2DP service"
        try:
            result = self.pri_dut.control_daemon("bt-a2dp.cmx", "stop")
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_get_connected_peers(self, line):
        """
        Description: Get the connected peers for the AVDTP service

        Usage:
          Examples:
            avdtp_get_connected_peers
        """
        cmd = "AVDTP get connected peers"
        try:
            result = self.pri_dut.avdtp_lib.getConnectedPeers()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_set_configuration(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: set configuration

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_set_configuration <peer_id>
        """
        cmd = "Send AVDTP set configuration to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.setConfiguration(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_get_configuration(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: get configuration

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_get_configuration <peer_id>
        """
        cmd = "Send AVDTP get configuration to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.getConfiguration(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_get_capabilities(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: get capabilities

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_get_capabilities <peer_id>
        """
        cmd = "Send AVDTP get capabilities to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.getCapabilities(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_get_all_capabilities(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: get all capabilities

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_get_all_capabilities <peer_id>
        """
        cmd = "Send AVDTP get all capabilities to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.getAllCapabilities(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_reconfigure_stream(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: reconfigure stream

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_reconfigure_stream <peer_id>
        """
        cmd = "Send AVDTP reconfigure stream to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.reconfigureStream(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_suspend_stream(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: suspend stream

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_suspend_stream <peer_id>
        """
        cmd = "Send AVDTP suspend stream to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.suspendStream(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_suspend_reconfigure(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: suspend reconfigure

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_suspend_reconfigure <peer_id>
        """
        cmd = "Send AVDTP suspend reconfigure to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.suspendAndReconfigure(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_release_stream(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: release stream

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_release_stream <peer_id>
        """
        cmd = "Send AVDTP release stream to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.releaseStream(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_establish_stream(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: establish stream

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_establish_stream <peer_id>
        """
        cmd = "Send AVDTP establish stream to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.establishStream(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_start_stream(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: start stream

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_start_stream <peer_id>
        """
        cmd = "Send AVDTP start stream to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.startStream(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_abort_stream(self, peer_id):
        """
        Description: Send AVDTP command to connected peer: abort stream

        Input(s):
            peer_id: The specified peer_id.

        Usage:
          Examples:
            avdtp_abort_stream <peer_id>
        """
        cmd = "Send AVDTP abort stream to connected peer"
        try:
            result = self.pri_dut.avdtp_lib.abortStream(int(peer_id))
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_avdtp_remove_service(self, line):
        """
        Description: Removes the AVDTP service in use.

        Usage:
          Examples:
            avdtp_establish_stream <peer_id>
        """
        cmd = "Remove AVDTP service"
        try:
            result = self.pri_dut.avdtp_lib.removeService()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End AVDTP wrappers"""
    """Begin Audio wrappers"""

    def do_audio_start_output_save(self, line):
        """
        Description: Start audio output save

        Usage:
          Examples:
            audio_start_output_save
        """
        cmd = "Start audio capture"
        try:
            result = self.pri_dut.audio_lib.startOutputSave()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_audio_stop_output_save(self, line):
        """
        Description: Stop audio output save

        Usage:
          Examples:
            audio_stop_output_save
        """
        cmd = "Stop audio capture"
        try:
            result = self.pri_dut.audio_lib.stopOutputSave()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_audio_get_output_audio(self, line):
        """
        Description: Get the audio output saved to a local file

        Usage:
          Examples:
            audio_get_output_audio
        """
        cmd = "Get audio capture"
        try:
            save_path = "{}/{}".format(self.pri_dut.log_path, "audio.raw")
            result = self.pri_dut.audio_lib.getOutputAudio(save_path)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_audio_5_min_test(self, line):
        """
        Description: Capture and anlyize sine audio waves played from a Bluetooth A2DP
        Source device.

        Pre steps:
        1. Pair A2DP source device
        2. Prepare generated SOX file over preferred codec on source device.
            Quick way to generate necessary audio files:
            sudo apt-get install sox
            sox -b 16 -r 48000 -c 2 -n audio_file_2k1k_5_min.wav synth 300 sine 2000 sine 3000

        Usage:
          Examples:
            audio_5_min_test
        """
        cmd = "5 min audio capture test"
        input("Press Enter once Source device is streaming audio file")
        try:
            result = self.pri_dut.audio_lib.startOutputSave()
            self.log.info(result)
            for i in range(5):
                print("Minutes left: {}".format(10 - i))
                time.sleep(60)
            result = self.pri_dut.audio_lib.stopOutputSave()
            log_time = int(time.time())
            save_path = "{}/{}".format(self.pri_dut.log_path,
                                       "{}_audio.raw".format(log_time))
            analysis_path = "{}/{}".format(
                self.pri_dut.log_path,
                "{}_audio_analysis.txt".format(log_time))
            result = self.pri_dut.audio_lib.getOutputAudio(save_path)

            channels = 1
            try:
                quality_analysis(filename=save_path,
                                 output_file=analysis_path,
                                 bit_width=audio_bits_per_sample_32,
                                 rate=audio_sample_rate_48000,
                                 channel=channels,
                                 spectral_only=False)

            except Exception as err:
                self.log.error("Failed to analyze raw audio: {}".format(err))
                return False

            self.log.info("Analysis output here: {}".format(analysis_path))
            self.log.info("Analysis Results: {}".format(
                open(analysis_path).readlines()))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End Audio wrappers"""
    """Begin HFP wrappers"""

    def do_hfp_init(self, line):
        """
        Description: Init the HFP component initiate.

        Usage:
          Examples:
            hfp_init
        """
        cmd = "Initialize HFP proxy"
        try:
            result = self.pri_dut.hfp_lib.init()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_remove_service(self, line):
        """
        Description: Removes the HFP service in use.

        Usage:
          Examples:
            hfp_remove_service
        """
        cmd = "Remove HFP service"
        try:
            result = self.pri_dut.hfp_lib.removeService()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_list_peers(self, line):
        """
        Description: List all HFP Hands-Free peers connected to the DUT.

        Input(s):

        Usage:
          Examples:
            hfp_list_peers
        """
        cmd = "Lists connected peers"
        try:
            result = self.pri_dut.hfp_lib.listPeers()
            self.log.info(pprint.pformat(result))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_active_peer(self, line):
        """
        Description: Set the active HFP Hands-Free peer for the DUT.

        Input(s):
            peer_id: The id of the peer to be set active.

        Usage:
          Examples:
            hfp_set_active_peer <peer_id>
        """
        cmd = "Set the active peer"
        try:
            peer_id = int(line.strip())
            result = self.pri_dut.hfp_lib.setActivePeer(peer_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_list_calls(self, line):
        """
        Description: List all calls known to the sl4f component on the DUT.

        Input(s):

        Usage:
          Examples:
            hfp_list_calls
        """
        cmd = "Lists all calls"
        try:
            result = self.pri_dut.hfp_lib.listCalls()
            self.log.info(pprint.pformat(result))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_new_call(self, line):
        """
        Description: Simulate a call on the call manager

        Input(s):
            remote: The number of the remote party on the simulated call
            state: The state of the call. Must be one of "ringing", "waiting",
                   "dialing", "alerting", "active", "held".
            direction: The direction of the call. Must be one of "incoming", "outgoing".

        Usage:
          Examples:
            hfp_new_call <remote> <state> <direction>
            hfp_new_call 14085555555 active incoming
            hfp_new_call 14085555555 held outgoing
            hfp_new_call 14085555555 ringing incoming
            hfp_new_call 14085555555 waiting incoming
            hfp_new_call 14085555555 alerting outgoing
            hfp_new_call 14085555555 dialing outgoing
        """
        cmd = "Simulates a call"
        try:
            info = line.strip().split()
            if len(info) != 3:
                raise ValueError(
                    "Exactly three command line arguments required: <remote> <state> <direction>"
                )
            remote, state, direction = info[0], info[1], info[2]
            result = self.pri_dut.hfp_lib.newCall(remote, state, direction)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_incoming_call(self, line):
        """
        Description: Simulate an incoming call on the call manager

        Input(s):
            remote: The number of the remote party on the incoming call

        Usage:
          Examples:
            hfp_incoming_call <remote>
            hfp_incoming_call 14085555555
        """
        cmd = "Simulates an incoming call"
        try:
            remote = line.strip()
            result = self.pri_dut.hfp_lib.initiateIncomingCall(remote)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_waiting_call(self, line):
        """
        Description: Simulate an incoming call on the call manager when there is
        an onging active call already.

        Input(s):
            remote: The number of the remote party on the incoming call

        Usage:
          Examples:
            hfp_waiting_call <remote>
            hfp_waiting_call 14085555555
        """
        cmd = "Simulates an incoming call"
        try:
            remote = line.strip()
            result = self.pri_dut.hfp_lib.initiateIncomingWaitingCall(remote)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_outgoing_call(self, line):
        """
        Description: Simulate an outgoing call on the call manager

        Input(s):
            remote: The number of the remote party on the outgoing call

        Usage:
          Examples:
            hfp_outgoing_call <remote>
        """
        cmd = "Simulates an outgoing call"
        try:
            remote = line.strip()
            result = self.pri_dut.hfp_lib.initiateOutgoingCall(remote)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_call_active(self, line):
        """
        Description: Set the specified call to the "OngoingActive" state.

        Input(s):
            call_id: The unique id of the call.

        Usage:
          Examples:
            hfp_outgoing_call <call_id>
        """
        cmd = "Set the specified call to active"
        try:
            call_id = int(line.strip())
            result = self.pri_dut.hfp_lib.setCallActive(call_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_call_held(self, line):
        """
        Description: Set the specified call to the "OngoingHeld" state.

        Input(s):
            call_id: The unique id of the call.

        Usage:
          Examples:
            hfp_outgoing_call <call_id>
        """
        cmd = "Set the specified call to held"
        try:
            call_id = int(line.strip())
            result = self.pri_dut.hfp_lib.setCallHeld(call_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_call_terminated(self, line):
        """
        Description: Set the specified call to the "Terminated" state.

        Input(s):
            call_id: The unique id of the call.

        Usage:
          Examples:
            hfp_outgoing_call <call_id>
        """
        cmd = "Set the specified call to terminated"
        try:
            call_id = int(line.strip())
            result = self.pri_dut.hfp_lib.setCallTerminated(call_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_call_transferred_to_ag(self, line):
        """
        Description: Set the specified call to the "TransferredToAg" state.

        Input(s):
            call_id: The unique id of the call.

        Usage:
          Examples:
            hfp_outgoing_call <call_id>
        """
        cmd = "Set the specified call to TransferredToAg"
        try:
            call_id = int(line.strip())
            result = self.pri_dut.hfp_lib.setCallTransferredToAg(call_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_speaker_gain(self, line):
        """
        Description: Set the active peer's speaker gain.

        Input(s):
            value: The gain value to set. Must be between 0-15 inclusive.

        Usage:
          Examples:
            hfp_set_speaker_gain <value>
        """
        cmd = "Set the active peer's speaker gain"
        try:
            value = int(line.strip())
            result = self.pri_dut.hfp_lib.setSpeakerGain(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_microphone_gain(self, line):
        """
        Description: Set the active peer's microphone gain.

        Input(s):
            value: The gain value to set. Must be between 0-15 inclusive.

        Usage:
          Examples:
            hfp_set_microphone_gain <value>
        """
        cmd = "Set the active peer's microphone gain"
        try:
            value = int(line.strip())
            result = self.pri_dut.hfp_lib.setMicrophoneGain(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_service_available(self, line):
        """
        Description: Sets the simulated network service status reported by the call manager.

        Input(s):
            value: "true" to set the network connection to available.

        Usage:
          Examples:
            hfp_set_service_available <value>
            hfp_set_service_available true
            hfp_set_service_available false
        """
        cmd = "Sets the simulated network service status reported by the call manager"
        try:
            value = line.strip() == "true"
            result = self.pri_dut.hfp_lib.setServiceAvailable(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_roaming(self, line):
        """
        Description: Sets the simulated roaming status reported by the call manager.

        Input(s):
            value: "true" to set the network connection to roaming.

        Usage:
          Examples:
            hfp_set_roaming <value>
            hfp_set_roaming true
            hfp_set_roaming false
        """
        cmd = "Sets the simulated roaming status reported by the call manager"
        try:
            value = line.strip() == "true"
            result = self.pri_dut.hfp_lib.setRoaming(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_signal_strength(self, line):
        """
        Description: Sets the simulated signal strength reported by the call manager.

        Input(s):
            value: The signal strength value to set. Must be between 0-5 inclusive.

        Usage:
          Examples:
            hfp_set_signal_strength <value>
            hfp_set_signal_strength 0
            hfp_set_signal_strength 3
            hfp_set_signal_strength 5
        """
        cmd = "Sets the simulated signal strength reported by the call manager"
        try:
            value = int(line.strip())
            result = self.pri_dut.hfp_lib.setSignalStrength(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_subscriber_number(self, line):
        """
        Description: Sets the subscriber number reported by the call manager.

        Input(s):
            value: The subscriber number to set. Maximum length 128 characters.

        Usage:
          Examples:
            hfp_set_subscriber_number <value>
            hfp_set_subscriber_number 14085555555
        """
        cmd = "Sets the subscriber number reported by the call manager"
        try:
            value = line.strip()
            result = self.pri_dut.hfp_lib.setSubscriberNumber(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_operator(self, line):
        """
        Description: Sets the operator value reported by the call manager.

        Input(s):
            value: The operator value to set. Maximum length 16 characters.

        Usage:
          Examples:
            hfp_set_operator <value>
            hfp_set_operator GoogleFi
        """
        cmd = "Sets the operator value reported by the call manager"
        try:
            value = line.strip()
            result = self.pri_dut.hfp_lib.setOperator(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_nrec_support(self, line):
        """
        Description: Sets the noise reduction/echo cancelation support reported by the call manager.

        Input(s):
            value: The nrec support bool.

        Usage:
          Examples:
            hfp_set_nrec_support <value>
            hfp_set_nrec_support true
            hfp_set_nrec_support false
        """
        cmd = "Sets the noise reduction/echo cancelation support reported by the call manager"
        try:
            value = line.strip() == "true"
            result = self.pri_dut.hfp_lib.setNrecSupport(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_battery_level(self, line):
        """
        Description: Sets the battery level reported by the call manager.

        Input(s):
            value: The integer battery level value. Must be 0-5 inclusive.

        Usage:
          Examples:
            hfp_set_battery_level <value>
            hfp_set_battery_level 0
            hfp_set_battery_level 3
        """
        cmd = "Set the battery level reported by the call manager"
        try:
            value = int(line.strip())
            result = self.pri_dut.hfp_lib.setBatteryLevel(value)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_last_dialed(self, line):
        """
        Description: Sets the last dialed number in the call manager.

        Input(s):
            number: The number of the remote party.

        Usage:
          Examples:
            hfp_set_last_dialed <number>
            hfp_set_last_dialed 14085555555
        """
        cmd = "Sets the last dialed number in the call manager."
        try:
            number = line.strip()
            result = self.pri_dut.hfp_lib.setLastDialed(number)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_clear_last_dialed(self, line):
        """
        Description: Clears the last dialed number in the call manager.

        Usage:
          Examples:
            hfp_clear_last_dialed
        """
        cmd = "Clears the last dialed number in the call manager."
        try:
            result = self.pri_dut.hfp_lib.clearLastDialed()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_memory_location(self, line):
        """
        Description: Sets a memory location to point to a remote number.

        Input(s):
            location: The memory location at which to store the number.
            number: The number of the remote party to be stored.

        Usage:
          Examples:
            hfp_set_memory_location <location> <number>
            hfp_set_memory_location 0 14085555555
        """
        cmd = "Sets a memory location to point to a remote number."
        try:
            info = line.strip().split()
            if len(info) != 2:
                raise ValueError(
                    "Exactly two command line arguments required: <location> <number>"
                )
            location, number = info[0], info[1]
            result = self.pri_dut.hfp_lib.setMemoryLocation(location, number)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_clear_memory_location(self, line):
        """
        Description: Sets a memory location to point to a remote number.

        Input(s):
            localtion: The memory location to clear.

        Usage:
          Examples:
            hfp_clear_memory_location <location>
            hfp_clear_memory_location 0
        """
        cmd = "Sets a memory location to point to a remote number."
        try:
            location = line.strip()
            result = self.pri_dut.hfp_lib.clearMemoryLocation(location)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_dial_result(self, line):
        """
        Description: Sets the status result to be returned when the number is dialed.

        Input(s):
            number: The number of the remote party.
            status: The status to be returned when an outgoing call is initiated to the number.

        Usage:
          Examples:
            hfp_set_battery_level <value>
        """
        cmd = "Sets the status result to be returned when the number is dialed."
        try:
            info = line.strip().split()
            if len(info) != 2:
                raise ValueError(
                    "Exactly two command line arguments required: <number> <status>"
                )
            number, status = info[0], int(info[1])
            result = self.pri_dut.hfp_lib.setDialResult(number, status)
            self.log.info(pprint.pformat(result))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_get_state(self, line):
        """
        Description: Get the call manager's complete state

        Usage:
          Examples:
            hfp_get_state
        """
        cmd = "Get the call manager's state"
        try:
            result = self.pri_dut.hfp_lib.getState()
            self.log.info(pprint.pformat(result))
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_hfp_set_connection_behavior(self, line):
        """
        Description: Set the Service Level Connection (SLC) behavior when a new peer connects.

        Input(s):
            autoconnect: Enable/Disable autoconnection of SLC.

        Usage:
          Examples:
            hfp_set_connection_behavior <autoconnect>
            hfp_set_connection_behavior true
            hfp_set_connection_behavior false
        """
        cmd = "Set the Service Level Connection (SLC) behavior"
        try:
            autoconnect = line.strip().lower() == "true"
            result = self.pri_dut.hfp_lib.setConnectionBehavior(autoconnect)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End HFP wrappers"""
    """Begin RFCOMM wrappers"""

    def do_rfcomm_init(self, line):
        """
        Description: Initialize the RFCOMM component services.

        Usage:
          Examples:
            rfcomm_init
        """
        cmd = "Initialize RFCOMM proxy"
        try:
            result = self.pri_dut.rfcomm_lib.init()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_rfcomm_remove_service(self, line):
        """
        Description: Removes the RFCOMM service in use.

        Usage:
          Examples:
            rfcomm_remove_service
        """
        cmd = "Remove RFCOMM service"
        try:
            result = self.pri_dut.rfcomm_lib.removeService()
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_rfcomm_disconnect_session(self, line):
        """
        Description: Closes the RFCOMM Session.

        Usage:
          Examples:
            rfcomm_disconnect_session
            rfcomm_disconnect_session
        """
        cmd = "Disconnect the RFCOMM Session"
        try:
            result = self.pri_dut.rfcomm_lib.disconnectSession(
                self.unique_mac_addr_id)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_rfcomm_connect_rfcomm_channel(self, line):
        """
        Description: Make an outgoing RFCOMM connection.

        Usage:
          Examples:
            rfcomm_connect_rfcomm_channel <server_channel_number>
            rfcomm_connect_rfcomm_channel 2
        """
        cmd = "Make an outgoing RFCOMM connection"
        try:
            server_channel_number = int(line.strip())
            result = self.pri_dut.rfcomm_lib.connectRfcommChannel(
                self.unique_mac_addr_id, server_channel_number)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_rfcomm_disconnect_rfcomm_channel(self, line):
        """
        Description: Close the RFCOMM connection with the peer

        Usage:
          Examples:
            rfcomm_disconnect_rfcomm_channel <server_channel_number>
            rfcomm_disconnect_rfcomm_channel 2
        """
        cmd = "Close the RFCOMM channel"
        try:
            server_channel_number = int(line.strip())
            result = self.pri_dut.rfcomm_lib.disconnectRfcommChannel(
                self.unique_mac_addr_id, server_channel_number)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_rfcomm_send_remote_line_status(self, line):
        """
        Description: Send a remote line status for the RFCOMM channel.

        Usage:
          Examples:
            rfcomm_send_remote_line_status <server_channel_number>
            rfcomm_send_remote_line_status 2
        """
        cmd = "Send a remote line status update for the RFCOMM channel"
        try:
            server_channel_number = int(line.strip())
            result = self.pri_dut.rfcomm_lib.sendRemoteLineStatus(
                self.unique_mac_addr_id, server_channel_number)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    def do_rfcomm_write_rfcomm(self, line):
        """
        Description: Send data over the RFCOMM channel.

        Usage:
          Examples:
            rfcomm_write_rfcomm <server_channel_number> <data>
            rfcomm_write_rfcomm 2 foobar
        """
        cmd = "Send data using the RFCOMM channel"
        try:
            info = line.strip().split()
            if len(info) != 2:
                raise ValueError(
                    "Exactly two command line arguments required: <server_channel_number> <data>"
                )
            server_channel_number = int(info[0])
            data = info[1]
            result = self.pri_dut.rfcomm_lib.writeRfcomm(
                self.unique_mac_addr_id, server_channel_number, data)
            self.log.info(result)
        except Exception as err:
            self.log.error(FAILURE.format(cmd, err))

    """End RFCOMM wrappers"""