Connecting to device using pc-ble-driver-py causes disconnect after 60 seconds

[Ryan Lindeman (CAE)]

In order to test my custom BLE protocol on my device I am attempting to use the pc-ble-driver-py Python module (either 0.11.4 or latest version) to connect to my BLE peripheral running with a modified version of the app-template from SDK v12.3.0 on nRF52832.

After scanning for, connecting to, and discovering MTU and services I have my Python script do a busy wait.

On both the device and in my Python script I observe a disconnect after ~63 seconds.

If I connect to my device using nRFConnect PC tool, I don't observe this problem and am able to stay connected indefinitely to my device.

One theory is that my BLE peripheral is failing the BLE connection parameter update sequence but I am not setting the "disconnect on failure" flag when I initialize this (see code below). Is there some other way that I might be triggering a disconnect somehow?

#define FIRST_CONN_PARAMS_UPDATE_DELAY   APP_TIMER_TICKS(5000, APP_TIMER_PRESCALER) /**< Time from initiating event (connect or start of notification) to first time sd_ble_gap_conn_param_update is called (5 seconds). */
#define NEXT_CONN_PARAMS_UPDATE_DELAY    APP_TIMER_TICKS(30000, APP_TIMER_PRESCALER)/**< Time between each call to sd_ble_gap_conn_param_update after the first call (30 seconds). */
#define MAX_CONN_PARAMS_UPDATE_COUNT     3                                          /**< Number of attempts before giving up the connection parameter negotiation. */

/**@brief Function for initializing the Connection Parameters module.
 */
static void BLE_ConnectionParamsInit(void)
{
    uint32_t               err_code;
    ble_conn_params_init_t cp_init;

    memset(&cp_init, 0, sizeof(cp_init));

    cp_init.p_conn_params                  = NULL;
    cp_init.first_conn_params_update_delay = FIRST_CONN_PARAMS_UPDATE_DELAY;
    cp_init.next_conn_params_update_delay  = NEXT_CONN_PARAMS_UPDATE_DELAY;
    cp_init.max_conn_params_update_count   = MAX_CONN_PARAMS_UPDATE_COUNT;
    cp_init.start_on_notify_cccd_handle    = BLE_GATT_HANDLE_INVALID;
    cp_init.disconnect_on_fail             = false;
    cp_init.evt_handler                    = BLE_OnConnectionParamsEvent;
    cp_init.error_handler                  = BLE_OnConnectionParamsError;

    err_code = ble_conn_params_init(&cp_init);
    APP_ERROR_CHECK(err_code);
}

/**@brief Function for handling a Connection Parameters error.
 *
 * @param[in] nrf_error  Error code containing information about what went wrong.
 */
static void BLE_OnConnectionParamsError(uint32_t nrf_error)
{
    APP_ERROR_HANDLER(nrf_error);
}

/**@brief Function for handling the Connection Parameters Module.
 *
 * @details This function will be called for all events in the Connection Parameters Module which
 *          are passed to the application.
 *          @note All this function does is to disconnect. This could have been done by simply
 *                setting the disconnect_on_fail config parameter, but instead we use the event
 *                handler mechanism to demonstrate its use.
 *
 * @param[in] p_evt  Event received from the Connection Parameters Module.
 */
static void BLE_OnConnectionParamsEvent(ble_conn_params_evt_t * p_evt)
{
    uint32_t err_code;

    if (p_evt->evt_type == BLE_CONN_PARAMS_EVT_FAILED)
    {
        DBG_PrintLn(DBG_APPLICATION, "Connection Parameter interval unacceptable");
        err_code = sd_ble_gap_disconnect(sConnectionHandle, BLE_HCI_CONN_INTERVAL_UNACCEPTABLE);
        APP_ERROR_CHECK(err_code);
    }
}

[Ryan Lindeman (CAE)]

I am already setting cp_init.disconnect_on_fail to false and it is still disconnecting.

[Ryan Lindeman (CAE)]

I adjusted my Python central code to respond to the incoming "gap_evt_conn_param_update_request" event by attempting to call the ble_adapter.conn_param_update method but unfortunately it fails. I strongly suspect that the ble_adapter.py conn_param_update method is broken as there is no "status" key provided by the "gap_evt_conn_param_update" event.

As a workaround I am calling the ble_driver.ble_gap_conn_param_update method directly instead and holding off service discovery until I receive the initial gap_evt_conn_param_update_request. This seems to solve my disconnect problem for now.

Fullscreen 1538.controller.py Download

import re
import time
try:
    import Queue
except ImportError:
    import queue as Queue
import logging
logger = logging.getLogger(__name__)

from pc_ble_driver_py import config


config.__conn_ic_id__ = "NRF52"
nrf_sd_ble_api_ver = config.sd_api_ver_get()


# These imports must come after config import and assigning of __conn_ic_id__
from pc_ble_driver_py.ble_driver import (
    BLEDriver,
    BLEAdvData,
    BLEEvtID,
    BLEEnableParams,
    BLEGapTimeoutSrc,
    BLEGattStatusCode,
    BLEUUID,
    BLEUUIDBase)
if nrf_sd_ble_api_ver >= 5:
    from pc_ble_driver_py.ble_driver import (
        BLEConfig,
        BLEConfigConnGatt)
    CFG_TAG = 1
from pc_ble_driver_py.ble_adapter import BLEAdapter
from pc_ble_driver_py.exceptions import NordicSemiException
from pc_ble_driver_py.observers  import BLEAdapterObserver, BLEDriverObserver, get_addr_str


CONNECTIONS = 1


def get_input_as_list(data):
    if isinstance(data, str):
        return map(ord, data)
    else:
        return list(data)


class Central(BLEDriverObserver):
    def __init__(self, serial_port):
        super(Central, self).__init__()
        self.adapter = BLEAdapter(BLEDriver(serial_port=serial_port,
                                            auto_flash=True))
        self.connect_filter = None
        self.conn_q = Queue.Queue()
        self.disc_q = Queue.Queue()
        self.param_q = Queue.Queue()
        self.scan_q = Queue.Queue()
        self.debug = False
        self.peripherals = dict()
        self.dev_name_to_address = dict()
        self.scan_data = dict()
        self.scan_filter = None
        self.adapter.driver.observer_register(self)


    def open(self, vs_uuid_count=1):
        self.adapter.driver.open()
        if nrf_sd_ble_api_ver >= 5:
            self.adapter.default_mtu = 250
            gatt_cfg = BLEConfigConnGatt()
            gatt_cfg.att_mtu = self.adapter.default_mtu
            gatt_cfg.tag = CFG_TAG
            self.adapter.driver.ble_cfg_set(BLEConfig.conn_gatt, gatt_cfg)
            self.adapter.driver.ble_enable()
        elif nrf_sd_ble_api_ver >= 3:
            ble_enable_params = BLEEnableParams(vs_uuid_count      = vs_uuid_count,
                                                service_changed    = False,
                                                periph_conn_count  = 0,
                                                central_conn_count = CONNECTIONS,
                                                central_sec_count  = CONNECTIONS)
            ble_enable_params.att_mtu = 50
            self.adapter.driver.ble_enable(ble_enable_params)


    def close(self):
        self.adapter.driver.close()


    def add_custom_uuid(self, custom_uuid):
        # Add big-endian reversed version of custom_uuid
        uuid_base = BLEUUIDBase(custom_uuid[::-1])
        self.adapter.driver.ble_vs_uuid_add(uuid_base)
        return uuid_base


    def connect(self, dev_name_or_address, timeout=20):
        if self.scan_filter is None:
            short_scan = True
            self.scan_start(dev_name_or_address)
        else:
            short_scan = False
        self.connect_filter = dev_name_or_address
        try:
            peer_addr = self.scan_q.get(timeout=timeout)
        except Queue.Empty:
            peer_addr = None
        if short_scan:
            self.scan_stop()
        if peer_addr:
            self.connect_filter = None
            conn_handle = self.__connect_and_verify(peer_addr, conn_timeout=timeout)
            if conn_handle is not None:
                start_time = time.time()
                logger.info('Exchange MTU')
                # Perform MTU exchange if applicable
                if nrf_sd_ble_api_ver >= 5:
                    att_mtu = self.adapter.att_mtu_exchange(conn_handle, mtu=self.adapter.default_mtu)
                elif nrf_sd_ble_api_ver >= 3:
                    att_mtu = self.adapter.att_mtu_exchange(conn_handle)
                else:
                    att_mtu = 23
                # Wait for connection parameter exchange
                logger.info('Waiting for connection parameter exchange')
                try:
                    conn_params = self.param_q.get(timeout=timeout)
                except Queue.Empty:
                    raise RuntimeError('Connection parameter exchange failed')
                else:
                    self.adapter.driver.ble_gap_conn_param_update(conn_handle, conn_params)
                # Start service discovery
                logger.info('Service discovery started')
                self.adapter.service_discovery(conn_handle)
                self.peripherals[conn_handle] = Peripheral(self, conn_handle, att_mtu, peer_addr, start_time)
                logger.info('Connection complete for "{}"'.format(dev_name_or_address))
                return self.peripherals[conn_handle]
        else:
            logger.error('Peripheral name or address "{}" not found'.format(dev_name_or_address))
        return None


    def get_scan_data(self, dev_name_or_address, timeout=10):
        if dev_name_or_address in self.dev_name_to_address:
            address = self.dev_name_to_address[dev_name_or_address]
        else:
            address = dev_name_or_address
        # FIXME: Replace this if with a busy wait loop up to timeout?
        if address not in self.scan_data:
            return None
        try:
            return self.scan_data[address].get(timeout=timeout)
        except Queue.Empty:
            pass
        return None


    def scan_start(self, scan_filter=None):
        if scan_filter:
            logger.info('Scanning for names or addresses that match {!r}'.format(scan_filter))
        else:
            logger.info('Scanning for all peripherals')
        if scan_filter != self.scan_filter:
            self.dev_name_to_address.clear()
        self.scan_filter = scan_filter
        self.adapter.driver.ble_gap_scan_start()


    def scan_stop(self):
        try:
            self.adapter.driver.ble_gap_scan_stop()
        except NordicSemiException:
            pass


    def on_gap_evt_connected(self, ble_driver, conn_handle, peer_addr, role, conn_params):
        super(Central, self).on_gap_evt_connected(ble_driver, conn_handle, peer_addr, role, conn_params)
        self.conn_q.put(conn_handle)


    def on_gap_evt_disconnected(self, ble_driver, conn_handle, reason):
        super(Central, self).on_gap_evt_disconnected(ble_driver, conn_handle, reason)
        if conn_handle in self.peripherals:
            self.peripherals[conn_handle].remote_close(reason)
        else:
            logger.info('Connection closed: {}'.format(reason))
        self.disc_q.put(conn_handle)


    def on_gap_evt_timeout(self, ble_driver, conn_handle, src):
        super(Central, self).on_gap_evt_timeout(ble_driver, conn_handle, src)
        if src == BLEGapTimeoutSrc.scan:
            if self.debug:
                logger.debug('Restarting scan')
            ble_driver.ble_gap_scan_start()


    def on_gap_evt_adv_report(self, ble_driver, conn_handle, peer_addr, rssi, adv_type, adv_data):
        super(Central, self).on_gap_evt_adv_report(ble_driver, conn_handle, peer_addr, rssi, adv_type, adv_data)
        dev_name_list = None
        address  = get_addr_str(peer_addr)
        if BLEAdvData.Types.complete_local_name in adv_data.records:
            dev_name_list = adv_data.records[BLEAdvData.Types.complete_local_name]
        elif BLEAdvData.Types.short_local_name in adv_data.records:
            dev_name_list = adv_data.records[BLEAdvData.Types.short_local_name]
        elif BLEAdvData.Types.manufacturer_specific_data in adv_data.records:
            if address in self.scan_data:
                data = adv_data.records[BLEAdvData.Types.manufacturer_specific_data]
                if self.debug:
                    logger.debug("SCN: {} {}".format(address, "".join("{:02x}".format(c) for c in data)))
                self.scan_data[address].put(bytearray(data))
            return
        else:
            return

        dev_name = "".join(chr(e) for e in dev_name_list)
        if self.scan_filter is None or re.search(self.scan_filter, dev_name) or re.search(self.scan_filter, address):
            # Allow collecting scan data from this address
            if address not in self.scan_data:
                # Add device name to address lookup
                if len(dev_name) > 1:
                    self.dev_name_to_address[dev_name] = address
                self.scan_data[address] = Queue.Queue()
            if self.debug:
                logger.debug('Adv: address: 0x{}, rssi: {}, device_name: {}'.format(address,
                                                                             rssi,
                                                                             dev_name))
        # Are we waiting to connect to this peripheral? Then return peer_addr
        if self.connect_filter and (re.search(self.connect_filter, dev_name) or
                                    re.search(self.connect_filter, address)):
            self.scan_q.put(peer_addr)


    def on_gap_evt_conn_param_update_request(self, ble_driver, conn_handle, conn_params):
        super(Central, self).on_gap_evt_conn_param_update_request(ble_driver, conn_handle, conn_params)        
        self.param_q.put(conn_params)


    def __connect_and_verify(self, peer_addr, conn_timeout=20, disc_timeout=1):
        address = get_addr_str(peer_addr)
        retries = 1
        while retries <= 5:
            # Attempt connection
            logger.info('Connecting to "{}" (attempt {})'.format(address, retries))
            if nrf_sd_ble_api_ver >= 5:
                self.adapter.connect(peer_addr, tag=CFG_TAG)
            else:
                self.adapter.connect(peer_addr)
            try:
                conn_handle = self.conn_q.get(timeout=conn_timeout)
            except Queue.Empty:
                logger.error('Unable to connect to "{}" in {}s'.format(address, conn_timeout))
                retries += 1
                continue
            
            # Verify connection (monitor for disconnects)
            disc_handle = None
            while conn_handle != disc_handle:
                try:
                    disc_handle = self.disc_q.get(timeout=disc_timeout)
                except Queue.Empty:
                    # Connection is stable if no disconnect events occurred within timeout
                    break
            if conn_handle == disc_handle:
                retries += 1
                # Wait 1 second between attempts
                time.sleep(1)
                continue
            else:
                # Connection is stable
                break
        else:
            raise RuntimeError('Unable to establish stable connection to "{}"'.format(address))
        return conn_handle


class Peripheral(BLEAdapterObserver):
    def __init__(self, central, conn_handle, att_mtu, peer_addr, start_time):
        self.central = central
        self.conn_handle = conn_handle
        self.att_mtu = att_mtu
        self.peer_addr = peer_addr
        self.disconnect_reason = None
        self.notifications = dict()
        self.start_time = start_time
        self.central.adapter.observer_register(self)


    def is_connected(self):
        return self.conn_handle is not None


    def disconnect(self):
        if self.conn_handle is not None:
            logger.info('Requesting disconnect...')
            self.central.adapter.disconnect(self.conn_handle)


    def remote_close(self, disconnect_reason):
        elapsed_time = time.time() - self.start_time
        logger.info('Connection closed after {} seconds with reason {}'.format(elapsed_time, disconnect_reason))
        self.central.adapter.observer_unregister(self)
        self.disconnect_reason = None
        self.conn_handle = None


    def enable_notification(self, uuid, base=None):
        self.notifications[uuid] = Queue.Queue()
        self.central.adapter.enable_notification(self.conn_handle, BLEUUID(uuid, base=base))


    def disable_notification(self, uuid, base=None):
        self.central.adapter.disable_notification(self.conn_handle, BLEUUID(uuid, base=base))
        del self.notifications[uuid.value]


    def get_notification(self, uuid, timeout=10):
        if uuid not in self.notifications:
            raise RuntimeError("Notification {:02X} is not enabled".format(uuid))
        try:
            return self.notifications[uuid].get(timeout=timeout)
        except Queue.Empty:
            pass
        return None


    def read(self, uuid, base=None):
        # Use read_req to get first few bytes (offset=0)
        status, data = self._read_req(uuid, base=base)
        if not status:
            return None
        # If data filled maximum size allowed there might be more data
        elif len(data) == (self.att_mtu - 1):
            # Read more while status is OK and length is not 0
            message = []
            while status and len(data):
                if data:
                    message.extend(data)
                status, data = self._read_req(uuid, len(message), base=base)
        # Otherwise, just return the short data received
        else:
            return bytearray(data)
        return bytearray(message)


    def write(self, uuid, data, base=None, no_response=False):
        # Use write command if data is within att_mtu - 3 bytes of overhead
        if len(data) <= (self.att_mtu - 3):
            if no_response:
                self._write_req(uuid, data, base=base)
            else:
                self._write_cmd(uuid, data, base=base)
        else:
            # Due to overhead, delayed writes are 5 bytes less than att_mtu
            step = self.att_mtu - 5
            for i in range(0, len(data), step):
                self._write_prep(uuid, data[i:i+step], i, base=base)
            self._write_exec()


    def print_db(self):
        for s in self.central.adapter.db_conns[self.conn_handle].services:
            print(s)
            for ch in s.chars:
                print(ch)
                for d in ch.descs:
                    print(d)
            print("================================================================================")


    def on_indication(self, ble_adapter, conn_handle, uuid, data):
        if self.central.debug:
            logger.debug('Indication: {}, {} = {}'.format(conn_handle, uuid, "".join("{:02x}".format(c) for c in data)))


    def on_notification(self, ble_adapter, conn_handle, uuid, data):
        if self.central.debug:
            logger.debug('Notification: {}, {} = {}'.format(conn_handle, uuid, "".join("{:02x}".format(c) for c in data)))
        if uuid.value not in self.notifications:
            self.notifications[uuid.value] = Queue.Queue()
        self.notifications[uuid.value].put(bytearray(data))


    def on_conn_param_update_request(self, ble_adapter, conn_handle, conn_params):
        pass


    def _read_req(self, uuid, offset=0, base=None):
        status, data = self.central.adapter.read_req(self.conn_handle, BLEUUID(uuid, base=base), offset=offset)
        return status == BLEGattStatusCode.success, data


    def _write_cmd(self, uuid, data, base=None):
        self.central.adapter.write_cmd(self.conn_handle, BLEUUID(uuid, base=base), get_input_as_list(data))


    def _write_req(self, uuid, data, base=None):
        self.central.adapter.write_req(self.conn_handle, BLEUUID(uuid, base=base), get_input_as_list(data))


    def _write_prep(self, uuid, data, offset, base=None):
        self.central.adapter.write_prep(self.conn_handle, BLEUUID(uuid, base=base), get_input_as_list(data), offset)


    def _write_exec(self):
        self.central.adapter.write_exec(self.conn_handle)