how can we use different 128 Bit UUID for different read/write in pc-ble-driver-py?

[nagendrapilladi]

Hi ,

how can we use different 128 Bit UUID for different read/write in pc-ble-driver-py?


for example i defined three different characteristic 128 bit UUID for different operations

TEST_CHAR_ONE = BLEUUIDBase([0x12, 0x23, 0x00, 0x00, 0x56, 0x78, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x54, 0x49, 0x43])
TEST_CHAR_ONE_UUID = BLEUUID(0x1456, TEST_CHAR_ONE)

TEST_CHAR_TWO = BLEUUIDBase([0xaa, 0xbb, 0x00, 0x00, 0xcc, 0xdd, 0xee, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff, 0x66, 0x98, 0x76])
TEST_CHAR_TWO_UUID = BLEUUID(0x1456, TEST_CHAR_TWO)

TEST_CHAR_THREE = BLEUUIDBase([0x99, 0x88, 0x00, 0x00, 0x11, 0x22, 0x87, 0x65, 0x43, 0x21, 0xdd, 0xee, 0xff, 0x76, 0x87, 0x87])
TEST_CHAR_THREE_UUID = BLEUUID(0x1456, TEST_CHAR_THREE)

self.adapter.driver.ble_vs_uuid_add(TEST_CHAR_ONE_UUID)
self.adapter.driver.ble_vs_uuid_add(TEST_CHAR_TWO_UUID)
self.adapter.driver.ble_vs_uuid_add(TEST_CHAR_THREE_UUID)

self.adapter.enable_notification(TEST_CHAR_ONE_UUID)
self.adapter.enable_notification(TEST_CHAR_TWO_UUID)
self.adapter.enable_notification(TEST_CHAR_THREE_UUID)


i got error CCCD not found

"pc_ble_driver_py.exceptions.NordicSemiException: CCCD not found"

please support how to read and write operations on for different 128 bit UUID .

[Vidar Berg]

Hi,

Do all three characteristics have a CCCD descriptor and have you completed service discovery when you try to enable notifications? I'm including another example here which demonstrates how you can enable notifications for a custom characteristic:

Fullscreen nus_central.py Download

#
# Copyright (c) 2016 Nordic Semiconductor ASA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without modification,
# are permitted provided that the following conditions are met:
#
#   1. Redistributions of source code must retain the above copyright notice, this
#   list of conditions and the following disclaimer.
#
#   2. Redistributions in binary form must reproduce the above copyright notice, this
#   list of conditions and the following disclaimer in the documentation and/or
#   other materials provided with the distribution.
#
#   3. Neither the name of Nordic Semiconductor ASA nor the names of other
#   contributors to this software may be used to endorse or promote products
#   derived from this software without specific prior written permission.
#
#   4. This software must only be used in or with a processor manufactured by Nordic
#   Semiconductor ASA, or in or with a processor manufactured by a third party that
#   is used in combination with a processor manufactured by Nordic Semiconductor.
#
#   5. Any software provided in binary or object form under this license must not be
#   reverse engineered, decompiled, modified and/or disassembled.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#

import sys
import time
import logging
from queue import Queue, Empty
from pc_ble_driver_py.observers import *

TARGET_DEV_NAME = "Nordic_UART"
CONNECTIONS = 1
CFG_TAG = 1


def init(conn_ic_id):
    # noinspection PyGlobalUndefined
    global config, BLEDriver, BLEAdvData, BLEEvtID, BLEAdapter, BLEEnableParams, BLEGapTimeoutSrc, BLEUUID, BLEUUIDBase, BLEConfigCommon, BLEConfig, BLEConfigConnGatt, BLEGapScanParams
    from pc_ble_driver_py import config

    config.__conn_ic_id__ = conn_ic_id
    # noinspection PyUnresolvedReferences
    from pc_ble_driver_py.ble_driver import (
        BLEDriver,
        BLEAdvData,
        BLEEvtID,
        BLEEnableParams,
        BLEGapTimeoutSrc,
        BLEUUID,
        BLEUUIDBase,
        BLEGapScanParams,
        BLEConfigCommon,
        BLEConfig,
        BLEConfigConnGatt,
    )

    # noinspection PyUnresolvedReferences
    from pc_ble_driver_py.ble_adapter import BLEAdapter

    global nrf_sd_ble_api_ver
    nrf_sd_ble_api_ver = config.sd_api_ver_get()


class NusCollector(BLEDriverObserver, BLEAdapterObserver):
    def __init__(self, adapter):
        super(NusCollector, self).__init__()
        self.first = 1
        self.adapter = adapter
        self.conn_q = Queue()
        self.adapter.observer_register(self)
        self.adapter.driver.observer_register(self)
        self.adapter.default_mtu = 247
        self.nus_base = BLEUUIDBase([
            0x6e, 0x40, 0x00, 0x00, 0xb5, 0xa3, 0xf3, 0x93, 0xe0, 0xa9, 0xe5, 0x0e, 0x24, 0xdc, 0xca, 0x9e
        ])
        self.nus_rx = BLEUUID(0x0002, self.nus_base)
        self.nus_tx = BLEUUID(0x0003, self.nus_base)

    def open(self):
        self.adapter.driver.open()
        if config.__conn_ic_id__ == "NRF51":
            self.adapter.driver.ble_enable(
                BLEEnableParams(
                    vs_uuid_count=1,
                    service_changed=0,
                    periph_conn_count=0,
                    central_conn_count=1,
                    central_sec_count=0,
                )
            )
        elif config.__conn_ic_id__ == "NRF52":
            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()
            self.adapter.driver.ble_vs_uuid_add(self.nus_base)



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

    def connect_and_discover(self):
        scan_duration = 5
        params = BLEGapScanParams(interval_ms=200, window_ms=150, timeout_s=scan_duration)
        message = "We are connected!\r\n"

        self.adapter.driver.ble_gap_scan_start(scan_params=params)

        try:
            new_conn = self.conn_q.get(timeout=scan_duration)
            self.adapter.service_discovery(new_conn)
            self.adapter.enable_notification(new_conn, self.nus_tx)
            data = [ord(n) for n in list(message)]
            self.adapter.write_cmd(new_conn, self.nus_rx, data)
            return new_conn
        except Empty:
            print("No device advertising with name {TARGET_DEV_NAME} found.")
            return None

    def on_gattc_evt_exchange_mtu_rsp(self, ble_driver, conn_handle, status, att_mtu):
        print("ATT MTU updated to {}".format(att_mtu))
    
    def on_gap_evt_data_length_update(
        self, ble_driver, conn_handle, data_length_params
    ):
        print("Max rx octets: {}".format(data_length_params.max_rx_octets))
        print("Max tx octets: {}".format(data_length_params.max_tx_octets))
        print("Max rx time: {}".format(data_length_params.max_rx_time_us))
        print("Max tx time: {}".format(data_length_params.max_tx_time_us))

    def on_gatts_evt_exchange_mtu_request(self, ble_driver, conn_handle, client_mtu):
        print("Client requesting to update ATT MTU to {} bytes".format(client_mtu))
    def on_gap_evt_connected(
        self, ble_driver, conn_handle, peer_addr, role, conn_params
    ):
        print("New connection: {}".format(conn_handle))
        self.conn_q.put(conn_handle)

    def on_gap_evt_disconnected(self, ble_driver, conn_handle, reason):
        print("Disconnected: {} {}".format(conn_handle, reason))

    def on_gap_evt_adv_report(
        self, ble_driver, conn_handle, peer_addr, rssi, adv_type, adv_data
    ):
        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]

        else:
            return

        dev_name = "".join(chr(e) for e in dev_name_list)
        address_string = "".join("{0:02X}".format(b) for b in peer_addr.addr)
        print(
            "Received advertisment report, address: 0x{}, device_name: {}".format(
                address_string, dev_name
            )
        )

        if dev_name == TARGET_DEV_NAME:
            self.adapter.connect(peer_addr, tag=CFG_TAG)

    def on_notification(self, ble_adapter, conn_handle, uuid, data):
        if len(data) > 32:
            data = "({}...)".format(data[0:10])
        print("Connection: {}, {} = {}".format(conn_handle, uuid, data))


def main(selected_serial_port):
    print("Serial port used: {}".format(selected_serial_port))
    driver = BLEDriver(
        serial_port=selected_serial_port, auto_flash=False, baud_rate=1000000, log_severity_level="info"
    )

    adapter = BLEAdapter(driver)
    collector = NusCollector(adapter)
    collector.open()
    conn = collector.connect_and_discover()

    while conn is not None:
        time.sleep(100)

    collector.close()


def item_choose(item_list):
    for i, it in enumerate(item_list):
        print("\t{} : {}".format(i, it))
    print(" ")

    while True:
        try:
            choice = int(input("Enter your choice: "))
            if (choice >= 0) and (choice < len(item_list)):
                break
        except Exception:
            pass
        print("\tTry again...")
    return choice


if __name__ == "__main__":
    logging.basicConfig(
        level="DEBUG",
        format="%(asctime)s [%(thread)d/%(threadName)s] %(message)s",
    )
    serial_port = None
    if len(sys.argv) < 2:
        print("Please specify connectivity IC identifier (NRF51, NRF52)")
        exit(1)
    init(sys.argv[1])
    if len(sys.argv) == 3:
        serial_port = sys.argv[2]
    else:
        descs = BLEDriver.enum_serial_ports()
        choices = ["{}: {}".format(d.port, d.serial_number) for d in descs]
        choice = item_choose(choices)
        serial_port = descs[choice].port
    main(serial_port)
    quit()