Adding custom service UUID in pc_ble_driver_py problems

[SYJ]

When I used nrf52840 to connect the peripherals with pc-ble-driver-py, I had my own custom UUID, so I needed to add a custom UUID. I referred to this article, but the following problems occurred.

import sys
import time
import logging
from queue import Queue, Empty
from pc_ble_driver_py.observers import *
from serial import Serial
import serial.tools.list_ports

TARGET_DEV_NAME = "GACYF000006037ECB2E4"
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,
        BLEGapScanParams,
        BLEConfigCommon,
        BLEConfig,
        BLEConfigConnGatt,
        BLEUUIDBase
    )

    # 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()
    # logging.debug("api version" + nrf_sd_ble_api_ver)


class HRCollector(BLEDriverObserver, BLEAdapterObserver):
    def __init__(self, adapter):
        super(HRCollector, self).__init__()
        self.adapter = adapter
        self.conn_q = Queue()
        self.adapter.observer_register(self)
        self.adapter.driver.observer_register(self)
        self.adapter.default_mtu = 250

        self.localName = ""
        self.address_Str = TARGET_DEV_NAME
        self.advData = ""

    BASE_UUID = BLEUUIDBase(
        [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x10, 0x00, 0x80, 0x00, 0x00, 0x80, 0x5F, 0x9B, 0x34, 0xFB])
    kVCKServiceStatusCharacteristicUUID = BLEUUID(0xFFE1, BASE_UUID)
    kVCKServiceNotifyCharacteristicUUID = BLEUUID(0xFFF2, BASE_UUID)
    
    #pyth
    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_vs_uuid_add(BASE_UUID)

            self.adapter.driver.ble_enable()

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


    def connect_and_discover(self):
        scan_duration = 2
        params = BLEGapScanParams(interval_ms=200, window_ms=150, timeout_s=scan_duration)

        self.adapter.driver.ble_gap_scan_start(scan_params=params)
        print("sssssssssssssssss")

        # print(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, BLEUUID(BLEUUID.Standard.battery_level)
            )

            self.adapter.enable_notification(new_conn, BLEUUID(BLEUUID.Standard.heart_rate))
            return new_conn

        except Empty:
            print(f"No heart rate collector advertising with name {TARGET_DEV_NAME} found.")
            return None

    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
    ):

        dev_name_list = ""
        dev_adv_data = ""
        dev_name = ""
        if BLEAdvData.Types.complete_local_name in adv_data.records:
            dev_name_list = adv_data.records[BLEAdvData.Types.complete_local_name]
            dev_name = "".join(chr(e) for e in dev_name_list)

        if BLEAdvData.Types.manufacturer_specific_data in adv_data.records:
            if adv_data.records.get(BLEAdvData.Types.manufacturer_specific_data) == None:
                pass
            else:
                dev_adv_data = adv_data.records[BLEAdvData.Types.manufacturer_specific_data]
                dev_adv_data = "".join("{0:02X}".format(b) for b in dev_adv_data)

        address_string = "".join("{0:02X}".format(b) for b in peer_addr.addr)
        # ":".join("{:02X}".format(a) for a in addr.addr)
        if address_string == TARGET_DEV_NAME:
            if len(dev_name) > 0:
                self.localName = dev_name
            if len(dev_adv_data) > 0:
                if len(self.advData) > 0:
                    if self.advData == dev_adv_data:
                        self.adapter.driver.ble_gap_scan_stop()
                        print("stop scan")
                        print("localName ========{}".format(self.localName))
                        print("address ========{}".format(self.address_Str))
                        print("advData ========{}".format(self.advData))
                        self.adapter.connect(peer_addr, tag=CFG_TAG)
                    else:
                        self.advData = "{}{}".format(self.advData,dev_adv_data)
                else:
                    self.advData = dev_adv_data

        # if address_string == TARGET_DEV_NAME:
        #     print("find device")
        #



    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 = HRCollector(adapter)
    collector.open()
    conn = collector.connect_and_discover()

    if conn is not None:
        time.sleep(10)

    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

#查找端口函数
def findPort():
    ports = serial.tools.list_ports.comports()
    print(ports)

    if len(ports) <= 0:
        print ("The Serial port can't find!")
        exit(1)
    else:
        if len(ports) == 1:
            port0 = list(ports[0])
            port_serial = port0[0]
        else:
            port0 = list(ports[1])
            port_serial = port0[0]
        return port_serial
        print(f"-----{ports}")
        print(f"-----{port_serial}")



if __name__ == "__main__":
    logging.basicConfig(
        level="DEBUG",
        format="%(asctime)s Ingeek [%(thread)d/%(threadName)s] %(message)s",
    )

    TARGET_DEV_NAME = str(input("输入要连接的设备名称:"))
    print(TARGET_DEV_NAME)

    serial_port = None
    port = findPort()
    sys.argv = ["","NRF52",port]
    print(sys.argv)
    if len(sys.argv) < 2:
        print("Please specify connectivity IC identifier (NRF51, NRF52)")
        exit(1)
    init(sys.argv[1])
    if len(sys.argv) == 3:
        print(sys.argv)
        serial_port = sys.argv[2]
    else:
        descs = BLEDriver.enum_serial_ports()
        print(descs)
        choices = ["{}: {}".format(d.port, d.serial_number) for d in descs]
        choice = item_choose(choices)
        serial_port = descs[choice].port
    main(serial_port)
    quit()

He told me of the mistake

Traceback (most recent call last):
  File "/Users/syj/Downloads/pc-ble-driver-py/pc_ble_driver_py/examples/heart_rate_collector.py", line 50, in <module>
    BASE_UUID = BLEUUIDBase(
NameError: name 'BLEUUIDBase' is not defined

If I put the method that created the UUID in my code inside def open(self):, I get the following error

Serial port used: /dev/cu.usbmodemC20FBBC0D20F2
2020-05-19 18:54:58,209 Ingeek [123145309003776/LogThread] Successfully opened /dev/cu.usbmodemC20FBBC0D20F2. Baud rate: 1000000. Flow control: none. Parity: none.
2020-05-19 18:54:58,209 Ingeek [123145309003776/LogThread] evt> severity(20) message(Successfully opened /dev/cu.usbmodemC20FBBC0D20F2. Baud rate: 1000000. Flow control: none. Parity: none.)
2020-05-19 18:54:58,210 Ingeek [123145314258944/StatusThread] RpcAppStatus.resetPerformed: Target Reset performed
2020-05-19 18:54:58,210 Ingeek [123145314258944/StatusThread] evt> status code(RpcAppStatus.resetPerformed) message(Target Reset performed)
2020-05-19 18:54:59,428 Ingeek [123145314258944/StatusThread] RpcAppStatus.connectionActive: Connection active
2020-05-19 18:54:59,428 Ingeek [123145314258944/StatusThread] evt> status code(RpcAppStatus.connectionActive) message(Connection active)
Traceback (most recent call last):
  File "/Users/syj/Downloads/pc-ble-driver-py/pc_ble_driver_py/examples/heart_rate_collector.py", line 286, in <module>
    main(serial_port)
  File "/Users/syj/Downloads/pc-ble-driver-py/pc_ble_driver_py/examples/heart_rate_collector.py", line 215, in main
    collector.open()
  File "/Users/syj/Downloads/pc-ble-driver-py/pc_ble_driver_py/examples/heart_rate_collector.py", line 117, in open
    self.adapter.driver.ble_vs_uuid_add(BASE_UUID)
  File "/usr/local/lib/python3.7/site-packages/pc_ble_driver_py/ble_driver.py", line 107, in wrapper
    error_code=err_code,
pc_ble_driver_py.exceptions.NordicSemiException: Failed to ble_vs_uuid_add. Error code: 12289

self.adapter.driver.ble_vs_uuid_add(BASE_UUID) This sentence should be wrong, but I don't know why this is wrong.

[Jørgen Holmefjord]

Hi,

I recently created an example for using Nordic UART Service with the latest version of pc-ble-driver-py. Can you check if you are able to use my code from this post?

Best regards,
Jørgen

[SYJ]

New connection: 0
Disconnected: 0 BLEHci.conn_failed_to_be_established
Traceback (most recent call last):
  File "/Users/syj/Desktop/StudyDemo/PythonStudy/BleConnect.py", line 289, in <module>
    main(serial_port)
  File "/Users/syj/Desktop/StudyDemo/PythonStudy/BleConnect.py", line 236, in main
    conn = collector.connect_and_discover()
  File "/Users/syj/Desktop/StudyDemo/PythonStudy/BleConnect.py", line 136, in connect_and_discover
    self.adapter.service_discovery(new_conn)
  File "/usr/local/lib/python3.7/site-packages/pc_ble_driver_py/ble_driver.py", line 101, in wrapper
    err_code = wrapped(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/pc_ble_driver_py/ble_adapter.py", line 270, in service_discovery
    if response["status"] == BLEGattStatusCode.success:
TypeError: 'NoneType' object is not subscriptable

I'm going to use this to report this error, so it should be service_discovery method, response is of type None, so what do you do with that?

[SYJ]

One more question, how can I read the value of characteristic

[Jørgen Holmefjord]

I think that is related to this issue: https://github.com/NordicSemiconductor/pc-ble-driver-py/issues/55

Reads can be done using the ble_gattc_read function.