Hi,
I have a BLE OTA application and I try to send data with the nRF52840 dongle and a Python application using the pc_ble_driver for Python. I have tested my MCU application with the nRF Connect for Desktop app and I´m able to transmit up to 216 bytes, but my Python application is reporting "NRF_ERROR_DATA_SIZE" when I try to send more than 20 bytes.
import sys
import time
import math
import logging
import argparse
import progressbar
from queue import Queue, Empty
from pc_ble_driver_py.observers import *
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__ = "NRF52"
from pc_ble_driver_py.ble_driver import (
BLEDriver,
BLEAdvData,
BLEEvtID,
BLEEnableParams,
BLEGapTimeoutSrc,
BLEUUID,
BLEUUIDBase,
BLEGapScanParams,
BLEConfigCommon,
BLEConfig,
BLEConfigConnGatt,
)
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 OTAConnector(BLEDriverObserver, BLEAdapterObserver):
from enum import Enum
class States(Enum):
DEVICE_INFORMATION = 0x01
OTA_INIT = 0x02
OTA_DONE = 0x03
def __init__(self, adapter):
super(OTAConnector, 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 = 300
self.ota_base = BLEUUIDBase([0x6e, 0x40, 0x00, 0x00, 0xb5, 0xa3, 0xf3, 0x93, 0xe0, 0xa9, 0xe5, 0x0e, 0x24, 0xdc, 0xca, 0x9e])
self.ota_ctrl = BLEUUID(0x0002, self.ota_base)
self.ota_data = BLEUUID(0x0003, self.ota_base)
self.ota_info = BLEUUID(0x0004, self.ota_base)
self.conn = None
def open(self):
self.adapter.driver.open()
gatt_cfg = BLEConfigConnGatt()
gatt_cfg.att_mtu = self.adapter.default_mtu
gatt_cfg.tag = args.config
self.adapter.driver.ble_cfg_set(BLEConfig.conn_gatt, gatt_cfg)
self.adapter.driver.ble_enable()
self.adapter.driver.ble_vs_uuid_add(self.ota_base)
def close(self):
self.adapter.driver.close()
def connect_and_discover(self, ScanDuration = 5):
self.adapter.driver.ble_gap_scan_start(scan_params = BLEGapScanParams(interval_ms = 200, window_ms = 150, timeout_s = ScanDuration))
try:
self.conn = self.conn_q.get(timeout = ScanDuration)
self.adapter.service_discovery(self.conn)
self.adapter.enable_notification(self.conn, self.ota_ctrl)
self.adapter.enable_notification(self.conn, self.ota_info)
return True
except Empty:
print("No device advertising with name {} found.".format(args.device))
return False
def run_ota(self, file):
if(self.conn == None):
return False
try:
print("Opening {}...".format(file))
with open(file, "rb") as File:
# Get the device information
self.adapter.write_cmd(self.conn, self.ota_ctrl, [OTAConnector.States.DEVICE_INFORMATION.value])
# Get the total length of the firmware file
CurrentPos = File.tell()
FileSize = File.seek(0, 2)
File.seek(CurrentPos, 0)
print("File size: {} bytes".format(FileSize))
print("Packages: {}".format(math.ceil(FileSize / args.size)))
# Set the packet size
self.adapter.write_cmd(self.conn, self.ota_data, args.size.to_bytes(2, "little"))
self.adapter.write_cmd(self.conn, self.ota_ctrl, [OTAConnector.States.OTA_INIT.value])
time.sleep(2)
# Process the OTA update
Bar = progressbar.ProgressBar(max_value = FileSize)
Transmit = 0
while(FileSize > 0):
Bytes = File.read(args.size)
self.adapter.write_cmd(self.conn, self.ota_data, Bytes)
FileSize -= args.size
Transmit += args.size
Bar.update(Transmit)
# Finish the OTA update
self.adapter.write_cmd(self.conn, self.ota_ctrl, [OTAConnector.States.OTA_DONE.value])
return True
except Exception as e:
print("{}".format(e))
return False
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 == args.device):
self.adapter.connect(peer_addr, tag = args.config)
def on_notification(self, ble_adapter, conn_handle, uuid, data):
data = "({}...)".format(data[0:10])
print("Connection: {}, {} = {}".format(conn_handle, uuid, data))
if(__name__ == "__main__"):
driver = BLEDriver(serial_port = args.port, auto_flash = False, baud_rate = args.baud, log_severity_level = "info")
adapter = BLEAdapter(driver)
collector = OTAConnector(adapter)
collector.open()
if(collector.connect_and_discover() == True):
print("Device found. Start OTA...")
Result = collector.run_ota(args.input)
print("Update successful: {}".format(Result))
collector.close()
quit()
How can I increase the transmission length?
