pc_ble_driver_py issue with Multi notification per message

[sguitton01]

Hi All,

I have the same issue than in this thread : pc-ble-driver-py missing high-rate notifications 

To explain fastly, I am using the pc_ble_driver_py v0.15.0, with a nrf52840 Dongle to try catching voice sent by a device which should send aroud 5 notification of 20 bytes per connection slots.

On my software I only gets 1 notification per connection slots though if I try on nrf Connect Desktop I can see all the notification are being received (with the same dongle).

Do you know what is missing on my software code (I tried to add the last answer of the linked thread but it did not work)

Should I update my pc_ble_driver version (is it solved on new versions?)?

Thank you,

Sylvain

[Jørgen Holmefjord]

Hi,

Could you post your application code? Since nRF Connect for Desktop is also based on pc-ble-driver(-js), it should be possible to get it working with the Python version as well. There could be some wrong configs, etc. Not easy to see what is missing in your code without having a look at your code.

I can't see anything in the changelogs that indicate that such a problem should be fixed in later versions, but you could try to update the pc-ble-driver-py version if you want.

Best regards,
Jørgen

[sguitton01]

Fullscreen 1667.Test_3701.py Download

import sys

# Test to use local customized version of pc-ble-driver-py...
# Don't know how to do so for now, I just:
#   pip install pc-ble-driver-py
#   then directly motified the files in C:\Python38\Lib\site-packages\pc_ble_driver_py ...
#sys.path.insert(0, "./pc-ble-driver-py")
#import pc_ble_driver_py
#print("TEST : {}".format(pc_ble_driver_py.__file__))

import time
from datetime import datetime
import logging
import re
import json
import os

import queue
from traceback import print_tb
from pc_ble_driver_py.observers import (BLEDriverObserver, BLEAdapterObserver, gen_conn_params_str)

import threading
import tkinter as tk
from tkinter.scrolledtext import ScrolledText
from tkinter.ttk import Treeview

import wave, struct
import winsound

import numpy as np
import scipy
import scipy.io.wavfile
import scipy.fftpack

VERSION = "V0.0.7"

logger = logging.getLogger(__name__)
# https://docs.python.org/3/library/logging.html#logging-levels
LOGGING_LEVEL_TEST_RESULT_SUCCESS = 25
LOGGING_LEVEL_TEST_RESULT_FAILURE = 26
logging.addLevelName(LOGGING_LEVEL_TEST_RESULT_SUCCESS, "TEST_RESULT_SUCCESS")
logging.addLevelName(LOGGING_LEVEL_TEST_RESULT_FAILURE, "TEST_RESULT_FAILURE")

# Load configuration
CONFIG = {}
try:
    with open(sys.argv[1] if len(sys.argv) == 2 else 'config.json') as config_file:
        CONFIG = json.load(config_file)
except:
    pass

# Preparing test results logging
test_results_file = CONFIG.get("test_results_file", "")
if test_results_file == "automatic":
    TEST_RESULTS_FOLDER = './test_results'
    if not os.path.isdir(TEST_RESULTS_FOLDER):
        os.makedirs(TEST_RESULTS_FOLDER)
    test_results_file = datetime.now().strftime(TEST_RESULTS_FOLDER + "/%Y%m%d_%H%M%S.csv")
if len(test_results_file) > 0:
    logger.debug("Saving test results to {}".format(test_results_file))
    test_results_file = open(test_results_file, "a+")
else:
    test_results_file = None

TARGET_DEV_NAME = "xxxxxxx"
TARGET_FW_VERSION = "2.0"
CFG_TAG = 1

TARGET_KEYS = {
    "GUIDE"        : {"page":0x0C, "code": 0x8D},
    "POWER"        : {"page":0x0C, "code": 0x30},
    "1"            : {"page":0x07, "code": 0x1E},
    "2"            : {"page":0x07, "code": 0x1F},
    "3"            : {"page":0x07, "code": 0x20},
    "4"            : {"page":0x07, "code": 0x21},
    "5"            : {"page":0x07, "code": 0x22},
    "6"            : {"page":0x07, "code": 0x23},
    "7"            : {"page":0x07, "code": 0x24},
    "8"            : {"page":0x07, "code": 0x25},
    "9"            : {"page":0x07, "code": 0x26},
    "MOVIE"        : {"page":0x07, "code": 0x3F},
    "0"            : {"page":0x07, "code": 0x27},
    "TV"           : {"page":0x0C, "code": 0x89},
    "UP"           : {"page":0x0C, "code": 0x42},
    "LEFT"         : {"page":0x0C, "code": 0x44},
    "OK"           : {"page":0x0C, "code": 0x41},
    "RIGHT"        : {"page":0x0C, "code": 0x45},
    "DOWN"         : {"page":0x0C, "code": 0x43},
    "BACK"         : {"page":0x0C, "code":0x224},
    "MIC"          : {"page":0x0C, "code":0x221},
    "HOME"         : {"page":0x0C, "code":0x223},
    "NETFLIX"      : {"page":0x07, "code": 0x43},
    "VOL_P"        : {"page":0x0C, "code": 0xE9},
    "PLAY_PAUSE"   : {"page":0x0C, "code": 0xCD},
    "CH_P"         : {"page":0x0C, "code": 0x9C},
    "VOL_M"        : {"page":0x0C, "code": 0xEA},
    "MUTE"         : {"page":0x0C, "code": 0xE2},
    "CH_M"         : {"page":0x0C, "code": 0x9D},
}

TARGET_KEYS_COORDINATES = {
    "GUIDE"        : {"x": 22, "y": 72, "w":40, "h":40},
    "POWER"        : {"x":125, "y": 72, "w":40, "h":40},
    "1"            : {"x": 22, "y":116, "w":40, "h":40},
    "2"            : {"x": 72, "y":116, "w":40, "h":40},
    "3"            : {"x":125, "y":116, "w":40, "h":40},
    "4"            : {"x": 22, "y":161, "w":40, "h":40},
    "5"            : {"x": 72, "y":161, "w":40, "h":40},
    "6"            : {"x":125, "y":161, "w":40, "h":40},
    "7"            : {"x": 22, "y":207, "w":40, "h":40},
    "8"            : {"x": 72, "y":207, "w":40, "h":40},
    "9"            : {"x":125, "y":207, "w":40, "h":40},
    "MOVIE"        : {"x": 22, "y":252, "w":40, "h":40},
    "0"            : {"x": 72, "y":252, "w":40, "h":40},
    "TV"           : {"x":125, "y":252, "w":40, "h":40},
    "UP"           : {"x": 72, "y":295, "w":40, "h":40},
    "LEFT"         : {"x": 22, "y":347, "w":40, "h":40},
    "OK"           : {"x": 72, "y":347, "w":40, "h":40},
    "RIGHT"        : {"x":125, "y":347, "w":40, "h":40},
    "DOWN"         : {"x": 72, "y":390, "w":40, "h":40},
    "BACK"         : {"x": 22, "y":441, "w":40, "h":40},
    "MIC"          : {"x": 72, "y":441, "w":40, "h":40},
    "HOME"         : {"x":125, "y":441, "w":40, "h":40},
    "NETFLIX"      : {"x": 53, "y":492, "w":77, "h":24},
    "VOL_P"        : {"x": 22, "y":527, "w":40, "h":40},
    "PLAY_PAUSE"   : {"x": 72, "y":527, "w":40, "h":40},
    "CH_P"         : {"x":125, "y":527, "w":40, "h":40},
    "VOL_M"        : {"x": 22, "y":572, "w":40, "h":40},
    "MUTE"         : {"x": 72, "y":572, "w":40, "h":40},
    "CH_M"         : {"x":125, "y":572, "w":40, "h":40},

    "AUDIO"        : {"x": 72, "y": 72, "w":40, "h":40},
}

TARGET_TEST_LIST = [*TARGET_KEYS.keys(), "AUDIO"]

# noinspection PyGlobalUndefined
global config, BLEDriver, BLEAdvData, BLEEvtID, BLEAdapter, BLEEnableParams, BLEGapTimeoutSrc, BLEUUID, BLEConfigCommon, BLEConfig, BLEConfigConnGatt, BLEGapScanParams
from pc_ble_driver_py import (config, exceptions as BLEException)

config.__conn_ic_id__ = "NRF52"
# noinspection PyUnresolvedReferences
from pc_ble_driver_py.ble_driver import (
    BLEDriver,
    BLEAdvData,
    BLEEvtID,
    BLEEnableParams,
    BLEGapTimeoutSrc,
    BLEUUIDBase,
    BLEUUID,
    BLEGapScanParams,
    BLEConfigCommon,
    BLEConfig,
    BLEConfigConnGatt,
    BLEGattStatusCode,
    BLEGapRoles,
    BLEGapConnParams,
    BLEConfigConnGap,
)

# 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()

HID_BASE_UUID = BLEUUID(0x2A4D)
BATTERY_UUID = BLEUUID(0x2A19)


ATVV_BASE_UUID = BLEUUIDBase([0xAB, 0x5E, 0x00, 0x01, 0x5A, 0x21, 0x4F, 0x05, 0xBC, 0x7D, 0xAF, 0x01, 0xF6, 0x17, 0xB6, 0x64])
ATVV_TX_CHAR = BLEUUID(0x0002, ATVV_BASE_UUID)
ATVV_RX_CHAR = BLEUUID(0x0003, ATVV_BASE_UUID)
ATVV_CTL_CHAR = BLEUUID(0x0004, ATVV_BASE_UUID)

ATVV_FREQUENCY = 8000

def wav_frequency_peak(filename):
    # Get signal data
    Fs, signal = scipy.io.wavfile.read(filename)
    N = signal.shape[0]
    Ts = 1.0 / Fs

    # Compute signal FFT
    FFT = abs(scipy.fft.fft(signal))
    FFT_side = FFT[range(N // 2)]

    # Compute corresponding FFT frequency scale
    freqs = scipy.fftpack.fftfreq(signal.size, Ts)
    freqs_side = freqs[range(N // 2)]

    # Keep only frequency > 50Hz
    freqs_side = freqs_side[freqs_side > 50]
    FFT_side = FFT_side[-len(freqs_side):]

    # Find peak
    peak_idx = np.where(FFT_side == max(FFT_side))
    peak_frequency = max(freqs_side[peak_idx])

    return peak_frequency

from serial.tools.list_ports import comports
def serial_ports(vendor_id: int = None, product_id: int = None):
    return [port.device for port in comports()
            if ((vendor_id is None or vendor_id == port.vid) and
                (product_id is None or product_id == port.pid))]

class ADPCMDecoder:
    STEPSIZE_LUT = [
        7,      8,      9,      10,     11,     12,     13,     14,     16,     17,     19,     21,     23,     25,     28,     31,
        34,     37,     41,     45,     50,     55,     60,     66,     73,     80,     88,     97,     107,    118,    130,    143,
        157,    173,    190,    209,    230,    253,    279,    307,    337,    371,    408,    449,    494,    544,    598,    658,
        724,    796,    876,    963,    1060,   1166,   1282,   1411,   1552,   1707,   1878,   2066,   2272,   2499,   2749,   3024,
        3327,   3660,   4026,   4428,   4871,   5358,   5894,   6484,   7132,   7845,   8630,   9493,   10442,  11487,  12635,  13899,
        15289,  16818,  18500,  20350,  22385,  24623,  27086,  29794,  32767,
    ]

    INDEX_LUT = [
        -1, -1, -1, -1, 2, 4, 6, 8,
        -1, -1, -1, -1, 2, 4, 6, 8,
    ]

    @staticmethod
    def clamp(sample):
        sample = sample if sample < 32767 else 32767
        sample = sample if sample > -32767 else -32767
        return sample

    def __init__(self):
        self.audio_frame = []
        self.audio_pcm_frame_num = 0

    @staticmethod
    def decode_frame(adpcm):
        siDec = adpcm[2]
        if siDec & 0x80:
            siDec = (siDec - 256)

        pvDec = ((adpcm[0] << 8) | adpcm[1])
        if pvDec & 0x8000:
            pvDec = (pvDec - 65536)

        def decode_nibble(nibble):
            nonlocal siDec
            nonlocal pvDec

            step = ADPCMDecoder.STEPSIZE_LUT[siDec]
            cumDiff = (step >> 3)

            siDec += ADPCMDecoder.INDEX_LUT[nibble]

            if siDec < 0:
                siDec = 0
            elif siDec >= len(ADPCMDecoder.STEPSIZE_LUT):
                siDec = (len(ADPCMDecoder.STEPSIZE_LUT) - 1)

            if (nibble & 4) != 0:
                cumDiff += step

            if (nibble & 2) != 0:
                cumDiff += (step >> 1)

            if (nibble & 1) != 0:
                cumDiff += (step >> 2)

            if (nibble & 8) != 0:
                if pvDec < (cumDiff - 32767):
                    pvDec = -32767
                else:
                    pvDec -= cumDiff
            else:
                if pvDec > (32767 - cumDiff):
                    pvDec = 32767
                else:
                    pvDec += cumDiff

            return ADPCMDecoder.clamp(pvDec)

        pcm = []
        i = 3
        while i < len(adpcm):
            pcm.append(decode_nibble(adpcm[i] >> 4))
            pcm.append(decode_nibble(adpcm[i] & 0x0F))
            i = i + 1

        return pcm

    def feed(self, data):
        pcm = None
        '''
        if len(self.audio_frame) >= 128:
            self.audio_frame = []
        self.audio_frame.extend(data)
        print("len data : {}, len audio_frame : {}".format(len(data), len(self.audio_frame)))
        if len(data) <= 20:
            if len(self.audio_frame) == 128:
                pcm = ADPCMDecoder.decode_frame(self.audio_frame[3:])
                self.audio_pcm_frame_num += 1
                print(self.audio_pcm_frame_num)
        '''
        self.audio_frame.extend(data)
        if len(data) < 20:
            if len(self.audio_frame) == 134:
                pcm = ADPCMDecoder.decode_frame(self.audio_frame)
        if len(self.audio_frame) >= 134:
            self.audio_frame = []

        return pcm


class BLE(BLEDriverObserver, BLEAdapterObserver):
    def __init__(self, gui, port = None):
        super(BLE, self).__init__()
        self.gui = gui
        self.gui.set_ble(self)

        if port == None:
            ports = BLEDriver.enum_serial_ports() # This is bugged
            if len(ports) <= 0:
                ports = serial_ports(0x1915, 0xC00A)
                if len(ports) > 0:
                    port = ports[0]
            else:
                port = ports[0].port

        if port == None:
            raise IOError("BLE dongle serial port not found !")

        self.gui.set_port(port)

        self.conn_q = queue.Queue(maxsize=1) # Used to sync connection event

        self.adpcm_decoder = None
        self.scanned_devices = {}
        self.pressed_keys = {
            "KBRD" : [],
            "CSMR" : [],
        }

        self.adapter = BLEAdapter(BLEDriver(serial_port=port, auto_flash=False, baud_rate=1000000, log_severity_level="info"))

        self.adapter.observer_register(self)
        self.adapter.driver.observer_register(self)

        self.adapter.driver.open()
        gatt_cfg = BLEConfigConnGatt()
        gatt_cfg.att_mtu = 250
        gatt_cfg.tag = CFG_TAG
        self.adapter.driver.ble_cfg_set(BLEConfig.conn_gatt, gatt_cfg)

        conn_cfg = BLEConfigConnGap()
        conn_cfg.conn_count = 1
        conn_cfg.event_length = 320
        self.adapter.driver.ble_cfg_set(BLEConfig.conn_gap, conn_cfg)

        self.adapter.driver.ble_enable()

        self.adapter.driver.ble_vs_uuid_add(ATVV_BASE_UUID)

    def start_scan(self):
        self.scanned_devices = {}
        self.adapter.driver.ble_gap_scan_start(scan_params=BLEGapScanParams(interval_ms=200, window_ms=150, timeout_s=0))

    def stop_scan(self):
        self.adapter.driver.ble_gap_scan_stop()

    def disconnect(self, conn):
        self.adapter.disconnect(conn)

    def connect(self, mac, on_connected = None):
        def connection_thread(dev, on_connected):
            conn = None
            self.adapter.connect(dev["addr"], scan_params=BLEGapScanParams(interval_ms=200, window_ms=150, timeout_s=5), conn_params=BLEGapConnParams(7.5, 7.5, 1500, 0), tag=CFG_TAG)
            try:
                conn = self.conn_q.get(timeout=5)
            except queue.Empty:
                pass

            if on_connected is not None:
                on_connected(dev, conn)

        thread = threading.Thread(target=connection_thread, args=(self.scanned_devices[mac], on_connected))
        thread.isDaemon = True
        thread.start()

    def configure_device(self, conn, on_configured):
        self.pressed_keys["KBRD"] = []
        self.pressed_keys["CSMR"] = []

        def configuration_thread(conn, on_configured):
            try:
                self.adapter.service_discovery(conn)
                for s in self.adapter.db_conns[conn].services:
                    logger.debug(str(s))
                    for c in s.chars :
                        logger.debug(str(c))
            except BLEException.NordicSemiException as err:
                if on_configured is not None:
                    on_configured(conn, "Couldn't discover services : {}".format(err))
            try:
                self.adapter.authenticate(conn, _role=BLEGapRoles.central)
            except BLEException.NordicSemiException as err:
                if on_configured is not None:
                    on_configured(conn, "Couldn't pair : {}".format(err))

            try:
                self.adapter.enable_notification(conn, HID_BASE_UUID, attr_handle=0x2C)
                self.adapter.enable_notification(conn, HID_BASE_UUID, attr_handle=0x30)

                self.adapter.enable_notification(conn, ATVV_RX_CHAR)
                self.adapter.enable_notification(conn, ATVV_CTL_CHAR)
                self.adapter.write_cmd(conn, ATVV_TX_CHAR, [0x0A, 0x01, 0x00, 0x00, 0x03, 0x01])  # Request capacity
            except BLEException.NordicSemiException as err:
                if on_configured is not None:
                    on_configured(conn, "Couldn't enable notifications : {}".format(err))

            if on_configured is not None:
                on_configured(conn)

        thread = threading.Thread(target=configuration_thread, args=(conn, on_configured))
        thread.isDaemon = True
        thread.start()

    def get_device_fw_version(self, conn):
        status_code, value = self.adapter.read_req(conn, BLEUUID(0x2A26))
        if value is None:
            logger.error("Couldn't read device firmware version : {}".format(status_code))
        else:
            value = "".join(chr(e) for e in value)
        return value

    def on_gap_evt_connected(self, ble_driver, conn_handle, peer_addr, role, conn_params):
        logger.debug("New connection: {}".format(conn_handle))
        self.conn_q.put(conn_handle)

    def on_gap_evt_disconnected(self, ble_driver, conn_handle, reason):
        try:
            self.conn_q.get_nowait()
        except:
            pass
        self.gui.on_disconnected(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:
            dev_name_list = []

        name = "".join(chr(e) for e in dev_name_list)
        mac = "".join("{0:02X}:".format(b) for b in peer_addr.addr)[:-1]

        self.scanned_devices[mac] = {"name": name, "mac" : mac, "addr": peer_addr, "rssi": rssi}
        self.gui.on_scanned(self.scanned_devices)

    def start_voice(self, conn, from_gui=False):
        try:
            logger.info("Starting voice")
            self.adpcm_decoder = ADPCMDecoder()
            self.adapter.write_cmd(conn, ATVV_TX_CHAR, [0x0C, 0x01], attr_handle=0x1B)
            self.cnt = 0
        except:
            return False
        return True

    def stop_voice(self, conn):
        try:
            logger.info("Stopping voice")
            self.adapter.write_cmd(conn, ATVV_TX_CHAR, [0x0D], attr_handle=0x1B)
        except:
            return False
        return True

    def on_notification_handle(self, ble_adapter, conn_handle, uuid, attr_handle, data):
        if uuid == HID_BASE_UUID:
            keys = []
            page_name = "KBRD"
            page_code = 0x07
            if attr_handle == 0x30:
                data = [(data[0] | ((data[1] & 0xF) << 8)), (((data[1] & 0xF0) >> 4) | (data[2] << 4))]
                page_name = "CSMR"
                page_code = 0x0C

            for k in data:
                if k:
                    for label, value in TARGET_KEYS.items():
                        if value["page"] == page_code and value["code"] == k:
                            keys.append(label)
            for k in self.pressed_keys[page_name]:
                if not k in keys:
                    self.gui.on_key_released(conn_handle, k)
                    self.pressed_keys[page_name].remove(k)
            for k in keys:
                if not k in self.pressed_keys[page_name]:
                    self.gui.on_key_pressed(conn_handle, k)
                    self.pressed_keys[page_name].append(k)
        elif uuid == ATVV_RX_CHAR:
            self.cnt += 1
            logger.debug(str(self.cnt))
            if data is not []: 
                self.voice_received(conn_handle, data)
        elif uuid == ATVV_CTL_CHAR:
            logger.debug("CTL : {}".format(' '.join('{:02X}'.format(b) for b in data)))
            if data == [0]:
             self.gui.on_key_released(conn_handle, "MIC", False)
        elif uuid == BATTERY_UUID:
            logger.debug("Battery level : {}".format(data[0]))
        else:
            logger.warning("Unknow notification : {} {}".format(uuid, data))

    def voice_received(self, conn, data):
        if self.adpcm_decoder:
            pcm = self.adpcm_decoder.feed(data)
            if pcm is not None:
                self.gui.on_audio_received(conn, pcm)

    def on_indication(self, ble_adapter, conn_handle, uuid, char_handle, data):
        logger.debug("Indication: {}, {} ({}) = {}".format(conn_handle, uuid, char_handle, data))

    def on_conn_param_update_request(self, ble_adapter, conn_handle, conn_params):
        logger.debug("Conn param update request: {}\n Params({})".format(conn_handle, gen_conn_params_str(conn_params)))

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


class GUI():
    class LoggingQueueHandler(logging.Handler):
        def __init__(self, st):
            super().__init__()

            self.log_queue = queue.Queue()

            self.st = st

            self.window = st
            while self.window.master is not None:
                self.window = self.window.master

            self.window.after(50, self.poll_log_queue)

        def emit(self, record):
            self.log_queue.put(record)

        def poll_log_queue(self):
            while True:
                try:
                    record = self.log_queue.get(block=False)
                except queue.Empty:
                    break
                else:
                    self.display_log(record)
            self.window.after(50, self.poll_log_queue)

        def display_log(self, record):
            msg = self.format(record)
            scroll = self.st.bbox("end-1c") is not None
            self.st.configure(state='normal')
            self.st.insert(tk.END, msg + '\n', record.levelname)
            self.st.update_idletasks()
            self.st.configure(state='disabled')
            if scroll or CONFIG.get("force_logs_autoscroll", False):
                self.st.yview(tk.END)
                self.st.update_idletasks()

    def __init__(self):
        self.ble = None

        self.tested_devices = []
        self.current_tested_device = None

        self.audio_data = None
        self.test_list = None
        self.clicked_keys = {}

        self.window = tk.Tk()
        self.window.title("BLE Test {} ({})".format(TARGET_DEV_NAME, VERSION))
        self.window.grid_rowconfigure(0, weight=1)
        self.window.grid_columnconfigure(0, weight=1)

        self.state_panel = tk.Frame(self.window)
        self.state_panel.grid(row=0, column=0, sticky=(tk.N + tk.E + tk.W + tk.S))

        self.state_panel.grid_rowconfigure(0, weight=1)
        self.state_panel.grid_columnconfigure(0, weight=1)
        self.state_panel.rowconfigure(0, weight=1)
        self.state_panel.columnconfigure(0, weight=1)

        # Setup GUI logging
        st = ScrolledText(self.state_panel, state='disabled')
        st.configure(font='TkFixedFont')
        st.tag_config('INFO', foreground='black')
        st.tag_config('DEBUG', foreground='gray')
        st.tag_config('WARNING', foreground='orange')
        st.tag_config('ERROR', foreground='red')
        st.tag_config('CRITICAL', foreground='red', underline=1)
        st.tag_config('TEST_RESULT_SUCCESS', background='green', font=("TkFixedFont", 16))
        st.tag_config('TEST_RESULT_FAILURE', background='red', font=("TkFixedFont", 16))
        st.grid(row=0, column=0, sticky=(tk.N + tk.E + tk.W + tk.S))

        self.logging_handler = GUI.LoggingQueueHandler(st)
        self.logging_handler.setFormatter(logging.Formatter('%(asctime)s: %(message)s'))
        logger.addHandler(self.logging_handler)
        logging.getLogger("pc_ble_driver_py.observers").addHandler(self.logging_handler)
        logging.getLogger("pc_ble_driver_py.ble_adapter").addHandler(self.logging_handler)

        logger.debug(CONFIG)

        self.status_label = tk.Label(self.state_panel, text="")
        self.set_status("Initializing...")

        # DUT panel
        self.dut_panel = tk.Frame(self.window)

        self.devices_treeview = Treeview(self.window)
        self.devices_treeview["columns"] = ("name", "rssi")
        self.devices_treeview.column("#0", width=120, minwidth=120, stretch=tk.NO)
        self.devices_treeview.column("name", width=150, minwidth=150)
        self.devices_treeview.column("rssi", width=45, minwidth=45, stretch=tk.NO)
        self.devices_treeview.heading("#0", text="MAC")
        self.devices_treeview.heading("name", text="Name")
        self.devices_treeview.heading("rssi", text="RSSI")
        self.devices_treeview.bind("<<TreeviewSelect>>", self.on_device_selected)
        self.devices_treeview.grid(row=0, column=1, sticky=(tk.N + tk.S))

        self.fw_version_label = tk.Label(self.dut_panel, text="FW Version : {}".format("UNKNOW"))
        self.fw_version_label.grid(row=0, column=0)

        self.remote_img = tk.PhotoImage(file="xxxxx.png") # We must keep a reference for some reason
        self.remote_canvas = tk.Canvas(self.dut_panel, width=self.remote_img.width(), height=self.remote_img.height())
        self.remote_canvas.create_image((0, 0), image=self.remote_img, anchor=tk.NW)
        self.remote_canvas.grid(row=1, column=0)

        #self.dut_panel.grid(row=0, column=1)

    def set_status(self, status):
        logger.info(status)
        self.status_label["text"] = status
        self.status_label.grid(row=1, column=0)

    def on_device_selected(self, event):
        try:
            item = self.devices_treeview.selection()
        except IndexError as err:
            logger.error("{}".format(err))
            return

        mac = self.devices_treeview.item(item, "text")

        self.devices_treeview.delete(*self.devices_treeview.get_children())
        self.devices_treeview.grid_forget()

        self.set_status("Stopping scan...")
        self.ble.stop_scan()

        self.set_status("Connecting to {}...".format(mac))
        self.ble.connect(mac, self.on_connected)

    def mainloop(self):
        return self.window.mainloop()

    def set_ble(self, ble):
        self.ble = ble

    def set_port(self, port):
        logger.debug("Using port {}".format(port))
        self.window.title("{} ({})".format(self.window.title(), port))

    def on_configured(self, conn, error = None):
        if error is None:
            self.set_status("Device configured")
            version = self.ble.get_device_fw_version(conn)
            if version is None:
                logger.error("Couldn't read device firmware version")
                self.ble.disconnect(conn)
            else:
                logger.info("FW Version : {}".format(version))
                self.fw_version_label["text"] = "FW Version : {}".format(version)
                if version != TARGET_FW_VERSION:
                    logger.error("Invalid firmware version : {}".format(version))
                    self.ble.disconnect(conn)

            self.test_list = TARGET_TEST_LIST.copy()
            self.clicked_keys = {}
            for k in TARGET_KEYS_COORDINATES.keys():
                self.clicked_keys[k] = 0
            self.set_status("Testing...")
            self.dut_panel.grid(row=0, column=1)
        else:
            logger.error("Configuration failed : {}".format(error))
            self.ble.disconnect(conn)

    def on_connected(self, dev, conn):
        if conn is not None:
            self.set_status("Connected to {}".format(dev["mac"]))

            self.current_tested_device = dev

            self.set_status("Configuring device...")
            self.ble.configure_device(conn, self.on_configured)
        else:
            logger.error("Couldn't connect to {}".format(dev["mac"]))

            self.set_status("Scanning...")
            self.ble.start_scan()

    def on_disconnected(self, reason):
        self.set_status("Disconnected ({})".format(reason))

        if len(self.test_list) > 0:
            self.on_test_end(False, self.current_tested_device, None)

        self.current_tested_device = None

        #winsound.PlaySound("1KHz.wav", winsound.SND_FILENAME | winsound.SND_PURGE)
        winsound.PlaySound(None, winsound.SND_PURGE)

        self.dut_panel.grid_forget()
        self.remote_canvas.delete(tk.ALL)
        self.remote_canvas.create_image((0, 0), image=self.remote_img, anchor=tk.NW)

        self.devices_treeview.grid(row=0, column=1, sticky=(tk.N + tk.S))

        self.set_status("Scanning...")
        self.ble.start_scan()

    SCAN_REFRESH_PERIOD_SEC = 0.5
    next_scan_refresh = 0
    def on_scanned(self, devices):
        if GUI.next_scan_refresh < time.time():
            GUI.next_scan_refresh = (time.time() + GUI.SCAN_REFRESH_PERIOD_SEC)
            self.devices_treeview.delete(*self.devices_treeview.get_children())
            for _, dev in sorted(devices.items(), key=lambda item: -item[1]["rssi"]):
                if dev["name"] == TARGET_DEV_NAME:
                    if dev["rssi"] >= CONFIG.get("scan_rssi_threshold", float("-inf")):
                        if not (CONFIG.get("hide_successfully_tested", False) and (dev["mac"] in self.tested_devices)):
                            self.devices_treeview.insert("", tk.END, text=dev["mac"], values=(dev["name"], dev["rssi"]))
            self.devices_treeview.grid(row=0, column=1, sticky=(tk.N + tk.S))

            if CONFIG.get("autoconnect_nearest", False):
                children = self.devices_treeview.get_children()
                if len(children) > 0:
                    self.devices_treeview.selection_set(children[0])

    def voice_end(self, conn):
        self.ble.stop_voice(conn)

    def on_key_pressed(self, conn, key):
        logger.debug("Key {} pressed".format(key))

        coord = TARGET_KEYS_COORDINATES[key]
        self.remote_canvas.create_rectangle(coord["x"], coord["y"], coord["x"] + coord["w"], coord["y"] + coord["h"], fill="yellow")

        if key == "MIC":

            #threading.Timer(7, self.voice_end, args=[conn]).start()

            self.audio_data = []

            #winsound.PlaySound("1KHz.wav", winsound.SND_FILENAME | winsound.SND_ASYNC | winsound.SND_NODEFAULT | winsound.SND_LOOP)

            self.ble.start_voice(conn)

    def on_key_released(self, conn, key, mic_state = True):
        
        if key == "MIC" and not mic_state:
            #self.ble.stop_voice(conn)

            #winsound.PlaySound("1KHz.wav", winsound.SND_FILENAME | winsound.SND_PURGE)
            #winsound.PlaySound(None, winsound.SND_PURGE)

            def save_voice_record(self, conn):
                if (self.audio_data is None) or (len(self.audio_data) <= 0):
                    logger.warning("No audio data to save")
                    return

                audio_file = wave.open("test.wav", "w")
                audio_file.setnchannels(1)
                audio_file.setsampwidth(2)
                audio_file.setframerate(ATVV_FREQUENCY)
                for s in self.audio_data:
                    audio_file.writeframes(struct.pack('h', ADPCMDecoder.clamp(s)))
                audio_file.close()
                self.audio_data = None

                #winsound.PlaySound("test.wav", winsound.SND_FILENAME | winsound.SND_ASYNC | winsound.SND_NODEFAULT)
                freq_peak = wav_frequency_peak("test.wav")
                color = "red"
                if (freq_peak < 950) or (freq_peak > 1050):
                    logger.warning("Wrong frequency peak : {}".format(freq_peak))
                else:
                    logger.debug("Frequency peak : {}".format(freq_peak))
                    color = "green"
                    if "AUDIO" in self.test_list:
                        self.test_list.remove("AUDIO")
                        if len(self.test_list) <= 0:
                            self.on_test_end(True, self.current_tested_device, conn)

                coord = TARGET_KEYS_COORDINATES["AUDIO"]
                self.remote_canvas.create_rectangle(coord["x"], coord["y"], coord["x"] + coord["w"], coord["y"] + coord["h"], fill=color)

            if self.audio_data is not None:
                threading.Timer(0.1, save_voice_record, [self, conn]).start()
        else:
            logger.debug("Key {} released".format(key))

            COLORS = ["green", "blue", "grey", "cyan", "brown", "purple", "#404040"]
            color = COLORS[self.clicked_keys[key] % len(COLORS)]
            self.clicked_keys[key] += 1

            coord = TARGET_KEYS_COORDINATES[key]
            self.remote_canvas.create_rectangle(coord["x"], coord["y"], coord["x"] + coord["w"], coord["y"] + coord["h"], fill=color)
            
            if key in self.test_list:
                self.test_list.remove(key)
                if len(self.test_list) <= 0:
                    self.on_test_end(True, self.current_tested_device, conn)

    def on_audio_received(self, conn, pcm):
        if self.audio_data is not None:
            self.audio_data.extend(pcm)

    def on_test_end(self, success, dev=None, conn=None):
        if success:
            self.set_status("TEST SUCCESS")
            logger.log(LOGGING_LEVEL_TEST_RESULT_SUCCESS, "TEST SUCCESS")
            if dev is not None:
                self.tested_devices.append(dev["mac"])
        else:
            self.set_status("TEST FAILURE")
            logger.log(LOGGING_LEVEL_TEST_RESULT_FAILURE, "TEST FAILURE")

        if dev is not None:
            if test_results_file is not None:
                test_results_file.write("{};{};{}\n".format(dev["mac"], success, datetime.now().strftime("%Y/%m/%d %H:%M:%S")))
                test_results_file.flush()

        if conn is not None:
            self.ble.disconnect(conn)

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

    logging.getLogger("pc_ble_driver_py.observers").setLevel("INFO")
    logging.getLogger("pc_ble_driver_py.ble_adapter").setLevel("INFO")

    log_file = CONFIG.get("log_file", "")
    if log_file == "automatic":
        LOGS_FOLDER = './logs'
        if not os.path.isdir(LOGS_FOLDER):
            os.makedirs(LOGS_FOLDER)
        log_file = datetime.now().strftime(LOGS_FOLDER + "/%Y%m%d_%H%M%S.log")
    if len(log_file) > 0:
        logger.debug("Saving logs to {}".format(log_file))
        fh = logging.FileHandler(log_file)
        fh.setFormatter(logging.Formatter('%(asctime)s: %(message)s'))
        fh.setLevel(logging.DEBUG)
        logger.addHandler(fh)

        logging.getLogger("pc_ble_driver_py.observers").addHandler(fh)
        logging.getLogger("pc_ble_driver_py.ble_adapter").addHandler(fh)

    gui = GUI()
    ble = BLE(gui, CONFIG.get("com_port", None))
    gui.set_status("Scanning...")
    ble.start_scan()
    gui.mainloop()

    ble.stop()

Hi,

Thank you for your answer,

You will find the code attached,

Best regards,

Sylvain

[Jørgen Holmefjord]

Hi,

Can you try to increase the GATTS hvn_tx_queue_size? It looks like this is set to 1 by default.

Best regards,
Jørgen

[sguitton01]

Hi,

I just tried it does not seem to improve the issue, I still have only one data every 30ms instead of seven frames in that time on nrf Connect.

Best regards,

Sylvain

[sguitton01]

I use Python 3.8, should I use a newer version of python?

[Jørgen Holmefjord]

Hi,

Please check if the suggestions in this thread solves your issues.

Best regards,
Jørgen