Merge branch 'feat/support_ble_uart_service_v5.5' into 'release/v5.5'

Feat/support ble uart service (5.5)

See merge request espressif/esp-idf!47999
This commit is contained in:
Island
2026-04-28 13:26:31 +08:00
27 changed files with 4937 additions and 0 deletions
@@ -74,6 +74,7 @@ Explore examples and tutorials in the ESP-IDF examples directory:
- **Bluedroid**: :example:`bluetooth/bluedroid`
- **NimBLE**: :example:`bluetooth/nimble`
- **BLE UART Service** (turnkey serial-over-BLE peripheral on top of NimBLE or Bluedroid, exposing the BLE UART Service GATT layout): :example:`bluetooth/ble_uart_service`
Step-by-step tutorials for developing with the Bluedroid stack:
@@ -0,0 +1,7 @@
# The following lines of boilerplate have to be in your project's
# CMakeLists in this exact order for cmake to work correctly.
cmake_minimum_required(VERSION 3.22)
include($ENV{IDF_PATH}/tools/cmake/project.cmake)
idf_build_set_property(MINIMAL_BUILD ON)
project(ble_uart_service)
@@ -0,0 +1,645 @@
# BLE UART Porting & API Guide
A complete guide to integrating `ble_uart` into any ESP-IDF project.
**Two or three files plus 5 steps of glue code** are enough to bring an
encrypted BLE serial peripheral up in a fresh project — the same
`ble_uart.h` API works on top of either NimBLE or Bluedroid; pick the
host with a Kconfig knob.
This guide uses **NimBLE** as the running example because it is the
default on every ESP32 family target. The Bluedroid path is identical
from the application's point of view; the only differences are the
sdkconfig knobs called out in §4.3 and a few stack-specific notes
flagged inline.
---
## 1. What `ble_uart` Provides
| Capability | Description |
| --- | --- |
| Standard Nordic UART Service GATT (RX/TX) | Interoperates with every generic BLE-serial tool (nRF Connect, Web Bluetooth, custom scripts) |
| LE Secure Connections + Bonding pairing | Single switch; when enabled, a fresh 6-digit passkey is printed to UART |
| Auto-reconnect | After a bonded central disconnects, advertising restarts immediately and the LTK is reused — no passkey prompt |
| Raw byte pass-through | RX is delivered via a callback; TX is exposed as `ble_uart_tx` |
| Auto-fragmentation | TX is sliced according to the negotiated ATT MTU |
| Fully wrapped | The user's `app_main` only calls two functions: `install` + `open` |
`ble_uart` is agnostic of any application-layer protocol (no JSON, no
line framing). It only delivers bytes — **what you do with those bytes
is entirely up to you**.
---
## 2. Prerequisites
| Requirement | Notes |
| --- | --- |
| ESP-IDF v5.0+ | v5.x or v6.x recommended |
| BT controller | Must support BLE (ESP32 / C2 / C3 / C5 / C6 / C61 / H2 / S3 / …) |
| Host stack | Exactly one of `CONFIG_BT_NIMBLE_ENABLED=y` (default, smaller) or `CONFIG_BT_BLUEDROID_ENABLED=y` in sdkconfig (covered in detail below) |
| Flash size | At least 2 MB (the default partition table is plenty) |
---
## 3. File Inventory
Files to copy into the target project — pick the backend you want and
copy that pair plus the public header:
```
your_project/main/
├── ble_uart.h ← copy this (stack-agnostic public API, ~260 lines)
├── ble_uart_nimble.c ← if you'll set CONFIG_BT_NIMBLE_ENABLED=y (~670 lines)
└── ble_uart_bluedroid.c ← if you'll set CONFIG_BT_BLUEDROID_ENABLED=y (~900 lines)
```
You can also copy *both* `ble_uart_nimble.c` and `ble_uart_bluedroid.c`
unchanged — each `.c` file gates its body on the matching Kconfig
symbol, so the inactive one compiles to nothing. This is what the
example itself does, and it lets you flip stacks without changing the
source list.
Optional: `Kconfig.projbuild` defines `BLE_UART_DEVICE_NAME_PREFIX`
and `BLE_UART_RX_SCRATCH_SIZE`. Copy it too if you want either to be
tunable from `menuconfig`; otherwise hard-code the name in your
source and rely on the 1024-byte fallback for RX scratch.
---
## 4. Step-by-Step Integration
Assume you already have an ESP-IDF project (`my_project/`).
### 4.1 Copy the files
```bash
cd my_project/main
# Stack-agnostic public header — always.
cp /path/to/ble_uart_service/main/ble_uart.h .
# Pick one (or copy both — the inactive one compiles to nothing).
cp /path/to/ble_uart_service/main/ble_uart_nimble.c .
cp /path/to/ble_uart_service/main/ble_uart_bluedroid.c .
```
### 4.2 Edit `main/CMakeLists.txt`
```cmake
# List both backends; each .c file is gated on its matching Kconfig
# symbol, so only the active one contributes code.
idf_component_register(SRCS "main.c"
"ble_uart_nimble.c"
"ble_uart_bluedroid.c"
INCLUDE_DIRS "."
REQUIRES bt nvs_flash)
```
### 4.3 Edit `sdkconfig.defaults` (the 7 critical lines)
**NimBLE backend (default, smaller footprint):**
```ini
# Enable NimBLE
CONFIG_BT_ENABLED=y
CONFIG_BTDM_CTRL_MODE_BLE_ONLY=y # only needed on classic ESP32; C3/S3/C6/... will warn "unknown" — safe to ignore
CONFIG_BT_BLUEDROID_ENABLED=n
CONFIG_BT_NIMBLE_ENABLED=y
# Encryption + persistent bonds
CONFIG_BT_NIMBLE_SM_SC=y # LE Secure Connections
CONFIG_BT_NIMBLE_NVS_PERSIST=y # persist LTKs in NVS — passkey-free reconnects
```
`CONFIG_BT_NIMBLE_ATT_PREFERRED_MTU` is optional; the default (256) is
fine. Bumping it to 512 lets TX push larger chunks per notification, but
the central must support it.
**Bluedroid backend (drop-in alternative):**
```ini
CONFIG_BT_ENABLED=y
CONFIG_BT_NIMBLE_ENABLED=n
CONFIG_BT_BLUEDROID_ENABLED=y
# LE Secure Connections + bonding (Bluedroid persists LTKs by default)
CONFIG_BT_BLE_SMP_ENABLE=y
# Optional: bigger MTU
CONFIG_BT_GATT_MAX_MTU_SIZE=512
# BLE-only feature set (saves flash on classic-BT-capable parts)
CONFIG_BT_BLE_42_FEATURES_SUPPORTED=y
CONFIG_BT_BLE_42_ADV_EN=y
```
### 4.4 Write `app_main` (template)
Minimal working template:
```c
#include "esp_log.h"
#include "esp_mac.h"
#include "nvs_flash.h"
#include "ble_uart.h"
static const char *TAG = "app";
/* What to do with received bytes — up to you */
static void ble_uart_on_rx(const uint8_t *data, size_t len)
{
ESP_LOGI(TAG, "rx %u bytes", (unsigned)len);
/* echo it back as a demo */
ble_uart_tx(data, len);
}
void app_main(void)
{
/* 1. NVS: NimBLE uses it for PHY calibration and bond storage */
esp_err_t err = nvs_flash_init();
if (err == ESP_ERR_NVS_NO_FREE_PAGES || err == ESP_ERR_NVS_NEW_VERSION_FOUND) {
ESP_ERROR_CHECK(nvs_flash_erase());
err = nvs_flash_init();
}
ESP_ERROR_CHECK(err);
/* 2. Bring up BLE UART */
ESP_ERROR_CHECK(ble_uart_install(&(ble_uart_config_t){
.encrypted = true,
.device_name = "MyDevice",
.ble_uart_on_rx = ble_uart_on_rx,
}));
/* 3. Take off */
ESP_ERROR_CHECK(ble_uart_open());
}
```
### 4.5 Build & flash
```bash
idf.py set-target esp32s3 # or whichever target you use
idf.py build flash monitor
```
Once flashed, the UART monitor should show (NimBLE backend):
```
I (xxx) ble_uart: registered service 6e400001-... handle=14
I (xxx) ble_uart: registered chr 6e400002-... def=15 val=16
I (xxx) ble_uart: registered chr 6e400003-... def=17 val=18
I (xxx) ble_uart: addr=...
I (xxx) ble_uart: BLE host task started
I (xxx) ble_uart: advertising as 'MyDevice'
```
…or with the Bluedroid backend:
```
I (xxx) ble_uart: gatts reg status=0 app_id=85 gatts_if=3
I (xxx) ble_uart: registered service svc_handle=40 rx=42 tx=44 cccd=45
I (xxx) ble_uart: advertising started
```
nRF Connect on a phone discovers `MyDevice`; connect, enter the
passkey, subscribe to TX, write to RX, and you will see the echo come
back.
---
## 5. API Reference
### 5.1 Configuration struct
```c
typedef struct {
bool encrypted; /* Master switch for SC + Bonding + MITM */
const char *device_name; /* GAP device name; NULL uses the NimBLE default */
ble_uart_rx_cb_t ble_uart_on_rx;/* RX byte callback */
} ble_uart_config_t;
```
| Field | Type | Required | Default / meaning |
| --- | --- | --- | --- |
| `encrypted` | `bool` | yes | `true` = SC + Bonding + MITM + DisplayOnly + encrypted GATT chars; `false` = fully plaintext (sniffable, lab use only) |
| `device_name` | `const char *` | recommended | Any string. Mind the 31-byte primary advertising packet limit: flags(3) + tx_pwr(3) + name(2 + length) + 128-bit UUID(18) → keep the name ≤ 8 bytes |
| `ble_uart_on_rx` | callback | optional | `NULL` discards every received byte |
### 5.2 RX callback signature
```c
typedef void (*ble_uart_rx_cb_t)(const uint8_t *data, size_t len);
static void my_handler(const uint8_t *data, size_t len)
{
/* `data` is reused after the callback returns; memcpy into your own
* buffer if you need to keep it. */
}
```
**Caveats**:
- The callback runs in the **NimBLE host task** context — **do not
block**; offload heavy work to your own task.
- A single callback may carry only **part** of an upper-layer frame
(the central slices on ATT MTU). Framing logic (line / TLV /
length-prefixed) is your responsibility.
- The data carries **no `ctx` argument**. If your callback needs state,
use a file-scope `static` or a global.
### 5.3 Lifecycle functions
```c
int ble_uart_install(const ble_uart_config_t *cfg);
int ble_uart_open(void);
int ble_uart_close(void);
int ble_uart_uninstall(void);
```
| Function | What it does (NimBLE) | What it does (Bluedroid) | When to call | Blocking? |
| --- | --- | --- | --- | --- |
| `install` | `nimble_port_init` + `ble_hs_cfg` + SM + SIG services + UART GATT | `controller_init/enable` + `bluedroid_init/enable` + SM + `esp_ble_gatts_create_attr_tab` (waits ≤500 ms for the attr-table event) | After `nvs_flash_init()`, before `open` | No, ~50 ms (NimBLE) / ~150 ms (Bluedroid) |
| `open` | Bond store + spawn host task + start advertising once synced | Configure adv data + scan rsp + start advertising | After `install` | No, host runs in the background |
| `close` | Stop adv → graceful disconnect (LL_TERMINATE_IND, waits ≤500 ms for the disconnect event) → `nimble_port_stop()` | Stop adv → graceful disconnect (`esp_ble_gap_disconnect`, waits ≤500 ms) | After `open`, before `uninstall` | Yes, up to ~500 ms while waiting for the peer disconnect |
| `uninstall` | Calls `close` if still open, then `nimble_port_deinit()` and resets module state | Calls `close` if still open, then `bluedroid_disable+deinit` + `controller_disable+deinit` | After `close` (or directly — `uninstall` cascades into `close` on its own) | Yes, follows the same wait window as `close` |
Call order:
```
nvs_flash_init
└── ble_uart_install
└── ble_uart_open ← BLE is live
└── ble_uart_close
└── ble_uart_uninstall ← clean state, can install again
```
Each call returns `BLE_HS_EALREADY` if the corresponding state is
already true (e.g. `open` called twice, or `close` called when the
radio is already down). It is therefore safe to call `close` /
`uninstall` defensively at shutdown without checking the current state
yourself.
**Do NOT call `close` / `uninstall` from inside `ble_uart_on_rx`** —
that callback runs on the NimBLE host task, and `close` blocks on
`nimble_port_stop()` which expects the host task to exit. Self-stop
deadlocks. Forward the request to a normal FreeRTOS task instead.
### 5.4 TX interface
```c
int ble_uart_tx(const uint8_t *data, size_t len);
```
For formatted output, format into your own buffer with `snprintf` first
and pass it to `ble_uart_tx`:
```c
char line[64];
int n = snprintf(line, sizeof(line), "temp=%d.%d\n", t / 10, t % 10);
ble_uart_tx((const uint8_t *)line, (size_t)n);
```
**Return values**:
| Return | Meaning |
| --- | --- |
| `0` | Success (notification handed to the stack) |
| `BLE_HS_ENOTCONN` | No central connected; **this is normal — typically ignore** |
| `BLE_HS_EINVAL` | `data == NULL` or `len == 0` |
| `BLE_HS_ENOMEM` | Stack mbuf pool exhausted |
| other | Internal stack error — see `ble_hs.h` |
**Calling context**: any FreeRTOS task at any priority. **Not callable
from an ISR** — push the data to a queue from the ISR and let a task
call `ble_uart_tx`.
**Auto-fragmentation**: regardless of buffer size, the implementation
splits the payload into successive notifications of `(MTU - 3)` bytes.
The central receives them in transmission order.
### 5.5 Status queries
```c
bool ble_uart_is_connected(void);
bool ble_uart_is_subscribed(void);
```
- `is_connected()`: a central is connected (it may not be paired yet).
- `is_subscribed()`: the central has subscribed to TX notifications
(note: bonded reconnects often skip CCCD writes).
You usually **don't need** to query these up-front — `ble_uart_tx`
returns `ENOTCONN` to tell you.
### 5.6 Service UUID constant
```c
extern const ble_uart_uuid128_t ble_uart_service_uuid;
```
Always `6e400001-b5a3-f393-e0a9-e50e24dcca9e` (the NUS standard). It is
already inserted into the scan response, so the **application normally
does not touch it**. You only need it if you take over advertising
yourself (see 6.3).
---
## 6. Advanced Usage
### 6.1 Different RX framing strategies
**A. Split on `\n` (suits ASCII protocols / JSON)**
```c
static uint8_t s_buf[1024];
static size_t s_len;
static void on_rx(const uint8_t *d, size_t n)
{
for (size_t i = 0; i < n; i++) {
if (d[i] == '\n') { handle_line(s_buf, s_len); s_len = 0; }
else if (s_len < sizeof s_buf) s_buf[s_len++] = d[i];
}
}
```
**B. Length-prefixed binary frames**
```c
static void on_rx(const uint8_t *d, size_t n)
{
static uint16_t need = 0;
static uint8_t frame[256];
static size_t got = 0;
for (size_t i = 0; i < n; i++) {
if (need == 0) { need = d[i]; got = 0; continue; }
frame[got++] = d[i];
if (got == need) { handle_frame(frame, got); need = 0; }
}
}
```
**C. Forward straight to UART**
```c
static void on_rx(const uint8_t *d, size_t n)
{
uart_write_bytes(UART_NUM_1, (const char *)d, n);
}
```
### 6.2 Disabling encryption (lab scenarios)
```c
ble_uart_install(&(ble_uart_config_t){
.encrypted = false, /* ← turn it off */
.device_name = "OpenDev",
.ble_uart_on_rx = ...,
});
```
Effect:
- GATT characteristics drop the `_ENC | _AUTHEN` flags.
- Any central can read/write — no pairing required.
- No passkey prompt.
- Data is sniffable by any nearby nRF dongle.
**Do not ship this in production firmware.**
### 6.3 Coexisting with other GATT services
> The snippet below is for the **NimBLE backend**. With Bluedroid, register
> additional profiles via `esp_ble_gatts_app_register()` before calling
> `ble_uart_open()` — the gating rule is the same: extra services must
> be in place before advertising starts.
`ble_uart` registers its own service; you can call `ble_gatts_add_svcs()`
**multiple times** and NimBLE will build all of them into the GATT
table. **Caveat**: this must happen before `ble_uart_open()`, otherwise
the host task is already running and the GATT table is locked.
```c
ble_uart_install(&cfg);
/* Register your extra services before open() */
ble_svc_dis_init(); /* Device Information Service */
my_battery_service_init(); /* your own battery service */
ble_uart_open();
```
> If your service must appear in the **advertising packet**, you have
> to bypass `ble_uart`'s internal advertising logic — override
> `ble_hs_cfg.sync_cb` with your own implementation after
> `ble_uart_install`, then call `ble_uart_open()`. Note that
> `ble_uart`'s internal `start_advertising` will not run, so you must
> call `ble_gap_adv_start` yourself. In that case, just fork
> `ble_uart_nimble.c` (or the matching `ble_uart_bluedroid.c`).
### 6.4 Configuring the device-name prefix via Kconfig
Copy `Kconfig.projbuild` into `main/`, then:
```c
char name[24];
snprintf(name, sizeof(name), "%s-%02X%02X",
CONFIG_BLE_UART_DEVICE_NAME_PREFIX, mac[4], mac[5]);
ble_uart_install(&(ble_uart_config_t){
.encrypted = true,
.device_name = name,
.ble_uart_on_rx = on_rx,
});
```
Edit the default through `menuconfig → BLE UART Example → BLE device
name prefix`.
### 6.5 Pushing data proactively
You can call TX from any task:
```c
/* A periodic sensor-reporting task */
static void sensor_task(void *arg)
{
char line[64];
while (1) {
int t = read_temperature();
int n = snprintf(line, sizeof(line), "temp=%d.%d\n", t / 10, t % 10);
ble_uart_tx((const uint8_t *)line, (size_t)n);
vTaskDelay(pdMS_TO_TICKS(1000));
}
}
/* Spawn it from app_main */
xTaskCreate(sensor_task, "sensor", 3072, NULL, 5, NULL);
```
When nobody is subscribed, `ble_uart_tx` returns `BLE_HS_ENOTCONN` —
**just ignore it**.
---
## 7. Calling Context & Thread Safety
| Function | Calling context | Thread-safe |
| --- | --- | --- |
| `ble_uart_install` | Any task; once per uninstall cycle | One-shot until `uninstall` |
| `ble_uart_open` | Any task; after `install` | One-shot until `close` |
| `ble_uart_close` | Any task **except the BLE host task** (NimBLE host task / Bluedroid BTC task) | Idempotent; second call returns `EALREADY` |
| `ble_uart_uninstall` | Any task **except the BLE host task** | Idempotent; cascades into `close` if needed |
| `ble_uart_tx` | Any FreeRTOS task | Yes — multi-task concurrent |
| `ble_uart_is_connected` / `is_subscribed` | Any context | Yes (bool read; best-effort snapshot) |
| `ble_uart_on_rx` callback | BLE host task (NimBLE host task / Bluedroid BTC task) | Your code must not block, **must not call `close` / `uninstall`** |
| **Calling any `ble_uart` API from an ISR** | not allowed | Neither host stack supports it |
---
## 8. Memory / Performance
| Item | Footprint |
| --- | --- |
| Code segment (`ble_uart_nimble.c.o`) | ~14 KB (with `-Os`) |
| Code segment (`ble_uart_bluedroid.c.o`) | ~22 KB (with `-Os`; larger because long-write reassembly is open-coded) |
| Static RAM (globals + RX buffer) | ~1.1 KB (the bulk is `CONFIG_BLE_UART_RX_SCRATCH_SIZE`, default 1024 B) |
| Host task stack (NimBLE host / Bluedroid BTC) | 4 KB (default) |
| Controller task stack | ~3 KB (default) |
| Bond store (NVS) | ~80 bytes per bonded peer |
| ATT MTU | Negotiated; whatever you set in sdkconfig (247 / 256 / 512) |
Measured throughput (ESP32-S3, iPhone 14 Pro central, MTU 247):
- TX (notify): ~25 KB/s
- RX (write): ~20 KB/s
---
## 9. FAQ
| Symptom | Cause / fix |
| --- | --- |
| `nimble_port_init rc=...` | NVS not initialised, or BT controller not enabled |
| Compile error: `host/ble_hs.h` not found | `REQUIRES bt` is missing from CMakeLists |
| Device not discoverable | Device name exceeds the advertising packet limit (drop the tx_pwr field or shorten the name) |
| Pairing fails | Central uses "Just Works" but we require MITM (`encrypted=true`). Use a central that supports passkey entry |
| `enc_change status=13 encrypted=1 bonded=1` | `13 = BLE_HS_ETIMEOUT`. Bonded-reconnect race; **the link is actually encrypted — safe to ignore** |
| Notifications missing after a reconnect | Bonded centrals often skip the CCCD write; our TX path doesn't gate on subscription state, so notifications still go out — make sure the central side has its callback registered |
| Second connection rejected | `MAX_CONNECTIONS = 1` by default. For multi-connection support, bump the sdkconfig value and turn `s_conn_handle` (NimBLE backend) / `s_conn_id` (Bluedroid backend) into an array |
| Flash fills up | Bond entries accumulate. Periodically run `idf.py erase-flash`, or call `ble_store_clear()` in code |
---
## 10. Differences from This Example
If you **build directly on top of this example**:
| You already have | No further work needed |
| --- | --- |
| `main.c` echo template | Replace with your own `on_rx` body |
| `sdkconfig.defaults` | Reuse as-is |
| `Kconfig.projbuild` | Reuse as-is |
| `CMakeLists.txt` (root + main) | Reuse as-is |
If you **start from an empty project**:
| What you need to do | Source |
| --- | --- |
| Copy `ble_uart.h` + at least one of `ble_uart_nimble.c` / `ble_uart_bluedroid.c` into `main/` | This example |
| Copy the key lines of `sdkconfig.defaults` | §4.3 of this guide |
| Add SRC + REQUIRES to `main/CMakeLists.txt` | §4.2 of this guide |
| Write `install` + `open` in `app_main` | §4.4 of this guide |
---
## 11. API Cheat Sheet (print and pin to the wall)
```c
#include "ble_uart.h"
/* === Types === */
typedef void (*ble_uart_rx_cb_t)(const uint8_t *data, size_t len);
typedef struct {
bool encrypted;
const char *device_name;
ble_uart_rx_cb_t ble_uart_on_rx;
} ble_uart_config_t;
/* === Lifecycle === */
int ble_uart_install(const ble_uart_config_t *cfg); /* host + GATT */
int ble_uart_open(void); /* start advertising (NimBLE: spawn host task) */
int ble_uart_close(void); /* stop adv / disconnect / quiesce host */
int ble_uart_uninstall(void); /* tear down host + reset state */
/* === Send (callable from any task) === */
int ble_uart_tx(const uint8_t *data, size_t len);
/* === Receive === */
/* Via the cfg.ble_uart_on_rx callback, signature:
* void cb(const uint8_t *data, size_t len); */
/* === Status === */
bool ble_uart_is_connected(void);
bool ble_uart_is_subscribed(void);
/* === Service UUID (for advertising; usually no need to touch) === */
extern const ble_uart_uuid128_t ble_uart_service_uuid;
```
---
## 12. Minimal Project Template (ready to flash)
A complete flashable project takes 7 files (the inactive backend `.c`
compiles to nothing, so it costs you nothing to ship both):
```
my_ble_uart_project/
├── CMakeLists.txt
├── sdkconfig.defaults
└── main/
├── CMakeLists.txt
├── ble_uart.h ← copied from this example
├── ble_uart_nimble.c ← copied from this example
├── ble_uart_bluedroid.c ← copied from this example (optional)
└── main.c
```
**Root `CMakeLists.txt`**:
```cmake
cmake_minimum_required(VERSION 3.16)
include($ENV{IDF_PATH}/tools/cmake/project.cmake)
project(my_ble_uart)
```
**`main/CMakeLists.txt`**:
```cmake
idf_component_register(SRCS "main.c"
"ble_uart_nimble.c"
"ble_uart_bluedroid.c"
INCLUDE_DIRS "."
REQUIRES bt nvs_flash)
```
**`sdkconfig.defaults`** (7 lines, NimBLE backend):
```ini
CONFIG_BT_ENABLED=y
CONFIG_BTDM_CTRL_MODE_BLE_ONLY=y
CONFIG_BT_BLUEDROID_ENABLED=n
CONFIG_BT_NIMBLE_ENABLED=y
CONFIG_BT_NIMBLE_SM_SC=y
CONFIG_BT_NIMBLE_NVS_PERSIST=y
CONFIG_BT_NIMBLE_ATT_PREFERRED_MTU=512
```
**`main/main.c`** — copy the §4.4 template verbatim.
Flash:
```bash
idf.py set-target esp32s3
idf.py build flash monitor
```
Done.
@@ -0,0 +1,238 @@
# BLE UART Service Example — NimBLE / Bluedroid
| Supported Targets | ESP32 | ESP32-C2 | ESP32-C3 | ESP32-C5 | ESP32-C6 | ESP32-C61 | ESP32-H2 | ESP32-S3 |
| ----------------- | ----- | -------- | -------- | -------- | -------- | --------- | -------- | -------- |
A turnkey serial-over-BLE peripheral that implements the de-facto
**Nordic UART Service** GATT layout (RX write, TX notify), so any
standard BLE-serial central (nRF Connect, Web Bluetooth examples, your
own iOS / Android / Linux / Python scripts) can talk to it unchanged.
The example ships with **two interchangeable backends** — NimBLE and
Bluedroid — both implementing the same stack-agnostic
`ble_uart.h` API. Pick one with `idf.py menuconfig → Component config →
Bluetooth → Host`; the build system links the matching backend
automatically. Default is NimBLE (smaller footprint).
The whole BLE stack — NVS-backed bond store, NimBLE host, security
manager, advertising, pairing, GAP event handling — is wrapped behind
**two function calls** in `app_main`:
```c
ble_uart_install(&cfg); // NimBLE host + NUS GATT service
ble_uart_open(); // start advertising + auto-encrypt
```
…and two matching tear-down calls if your app ever needs to power
BLE off at runtime:
```c
ble_uart_close(); // stop advertising / disconnect / halt host
ble_uart_uninstall(); // free the NimBLE port + reset state
```
When a central connects, the firmware automatically initiates LE Secure
Connections + Bonding pairing, displays a fresh 6-digit passkey to the
UART monitor, persists the LTK in NVS, and starts delivering received
bytes to the application's `on_rx` callback. The application sends bytes
back with `ble_uart_tx()`.
## GATT layout
| | UUID | Properties | Default flags |
| -------- | -------------------------------------- | ------------------------- | ------------------------- |
| Service | `6e400001-b5a3-f393-e0a9-e50e24dcca9e` | — | — |
| RX (in) | `6e400002-b5a3-f393-e0a9-e50e24dcca9e` | Write, WriteNR | encrypted, authenticated |
| TX (out) | `6e400003-b5a3-f393-e0a9-e50e24dcca9e` | Notify (auto-CCCD) | encrypted, authenticated |
The `_ENC | _AUTHEN` flags are turned on only when `cfg.encrypted = true`
(the default in this example).
## Files
| File | Lines | Role |
| --- | ---: | --- |
| `main/main.c` | ~70 | NVS init, MAC-derived device name, install + open, RX echo handler. Identical for both backends. |
| `main/ble_uart.h` | ~260 | Stack-agnostic public API: 3-field config + 4 lifecycle functions + TX/status + UUID + `BLE_UART_E*` return codes. No NimBLE / Bluedroid types leak through. |
| `main/ble_uart_nimble.c` | ~670 | NimBLE backend: host bring-up, NUS GATT service via `ble_gatts_add_svcs`, advertising, pairing, install/open/close/uninstall. Active when `CONFIG_BT_NIMBLE_ENABLED=y`. |
| `main/ble_uart_bluedroid.c` | ~900 | Bluedroid backend: controller + host enable, NUS GATT service via `esp_ble_gatts_create_attr_tab` (service-table API), advertising, pairing, full PREP/EXEC long-write reassembly, install/open/close/uninstall. Active when `CONFIG_BT_BLUEDROID_ENABLED=y`. |
| `main/Kconfig.projbuild` | ~40 | Device-name prefix + RX scratch buffer size knobs. |
| `sdkconfig.defaults` | — | Default: NimBLE backend, MTU 512, SC + bonding + persistent NVS. |
| `sdkconfig.ci.bluedroid` | — | Overlay: switch to Bluedroid backend (used via `-D SDKCONFIG_DEFAULTS=...`, see "Choosing the host stack" below). |
## Public API
```c
typedef void (*ble_uart_rx_cb_t)(const uint8_t *data, size_t len);
typedef struct {
bool encrypted; /* SC + Bonding + MITM in one knob */
const char *device_name;
ble_uart_rx_cb_t ble_uart_on_rx;
} ble_uart_config_t;
/* Lifecycle */
int ble_uart_install(const ble_uart_config_t *cfg); /* NimBLE host + GATT */
int ble_uart_open(void); /* host task + advertising */
int ble_uart_close(void); /* stop adv / disconnect / halt host */
int ble_uart_uninstall(void); /* free NimBLE port + reset state */
/* Data path */
int ble_uart_tx(const uint8_t *data, size_t len);
/* Status (best-effort snapshot) */
bool ble_uart_is_connected(void);
bool ble_uart_is_subscribed(void);
extern const ble_uart_uuid128_t ble_uart_service_uuid;
```
## Choosing the host stack
The same `ble_uart.h` API is implemented twice — once on top of NimBLE
(`ble_uart_nimble.c`) and once on top of Bluedroid
(`ble_uart_bluedroid.c`). `main/CMakeLists.txt` registers both files;
each guards its body with `#if CONFIG_BT_NIMBLE_ENABLED` / `#if
CONFIG_BT_BLUEDROID_ENABLED`, so exactly one becomes live at compile
time.
Two ways to switch:
```bash
# A. Flip the Kconfig knob interactively
idf.py menuconfig
# Component config -> Bluetooth -> Host -> NimBLE / Bluedroid
# B. Apply the Bluedroid overlay non-interactively (great for CI)
idf.py -B build_bd \
-D SDKCONFIG_DEFAULTS="sdkconfig.defaults;sdkconfig.ci.bluedroid" \
reconfigure
idf.py -B build_bd build flash monitor
```
When neither is enabled the build fails up-front with a clear error.
### Differences callers should know about
| | NimBLE backend | Bluedroid backend |
| --- | --- | --- |
| Long-write (PREP+EXEC) reassembly | done by NimBLE itself | done in `ble_uart_bluedroid.c`, capped at `CONFIG_BLE_UART_RX_SCRATCH_SIZE` |
| TX congestion behaviour | `mbuf` pool is generous; rarely returns ENOMEM | `esp_ble_gatts_send_indicate` may return ENOMEM under load → caller should back off |
| Passkey origin | generated locally with `esp_random()` | generated by the Bluedroid SM, surfaced via `PASSKEY_NOTIF_EVT` |
| Bond persistence | needs `CONFIG_BT_NIMBLE_NVS_PERSIST=y` | persisted by default |
| `install` blocking time | ~50 ms | ~150 ms (waits for `CREAT_ATTR_TAB_EVT`) |
| `uninstall` thoroughness | `nimble_port_deinit()` releases everything | `bluedroid_disable+deinit` + `controller_disable+deinit` releases everything |
## How to use
### Configure
```bash
idf.py set-target esp32c3 # or esp32, esp32s3, esp32c6, esp32h2 ...
idf.py menuconfig # optional
# Component config -> BLE UART Example
# - BLE device name prefix (default: BleUart)
```
The two security knobs are set in `sdkconfig.defaults`:
```ini
CONFIG_BT_NIMBLE_SM_SC=y # LE Secure Connections
CONFIG_BT_NIMBLE_NVS_PERSIST=y # Bond keys persist across reboots
```
Disable `cfg.encrypted` in `main.c` (set it to `false`) for plaintext
operation in the lab — the GATT characteristics drop their `_ENC`
flags accordingly. Production firmware should keep encryption on.
### Build & flash
```bash
idf.py build flash monitor
```
Expected boot log (NimBLE backend — the per-characteristic register
lines are NimBLE-specific; Bluedroid prints the four NUS handles in a
single line, see below):
```
I (xxx) ble_uart: registered service 6e400001-... handle=14
I (xxx) ble_uart: registered chr 6e400002-... def=15 val=16
I (xxx) ble_uart: registered chr 6e400003-... def=17 val=18
I (xxx) ble_uart: addr=80:7d:3a:11:22:33
I (xxx) ble_uart: BLE host task started
I (xxx) ble_uart: advertising as 'BleUart-XXXX'
```
Expected boot log (Bluedroid backend):
```
I (xxx) ble_uart: gatts reg status=0 app_id=85 gatts_if=3
I (xxx) ble_uart: registered service svc_handle=40 rx=42 tx=44 cccd=45
I (xxx) ble_uart: advertising started
```
## Pairing & demo
1. On a phone, install **nRF Connect for Mobile**.
2. Scan, tap **Connect** on `BleUart-XXXX`. The phone prompts for a
6-digit code.
3. The device prints a fresh code in a banner on UART:
```
W (xxx) ble_uart: +-----------------------------+
W (xxx) ble_uart: | BLE PAIRING PASSKEY: |
W (xxx) ble_uart: | 427183 |
W (xxx) ble_uart: +-----------------------------+
```
4. Type that code on the phone; pairing completes. The link is now
AES-CCM-encrypted and the LTK is stored to NVS.
5. Open the *Nordic UART Service*, subscribe to TX (the down-arrow
icon), then write any bytes to RX (the up-arrow icon). The device
logs them to UART and **echoes them right back** through TX.
6. Disconnect and reconnect: no passkey prompt — the bond resumes
automatically.
To wipe the bond and force a fresh passkey, run `idf.py erase-flash`
and re-flash.
## Adapting to your application
Replace the `ble_uart_on_rx` body in `main.c` with your own protocol
parser (line / TLV / length-prefixed framing — `ble_uart` delivers raw
bytes with no framing assumptions). Send replies with `ble_uart_tx()`.
## Reusing `ble_uart` in your own project
Copy `main/ble_uart.h` plus the backend(s) you want — `main/ble_uart_nimble.c`
and/or `main/ble_uart_bluedroid.c` — into your project, add `bt nvs_flash`
to your component's `REQUIRES`, then in your `app_main`:
```c
nvs_flash_init();
ble_uart_install(&(ble_uart_config_t){
.encrypted = true,
.device_name = "MyDevice",
.ble_uart_on_rx = my_handler,
});
ble_uart_open();
```
That's it — encrypted serial-over-BLE in 4 lines.
## Troubleshooting
- **Phone shows "pairing failed"** — the central asked for "Just Works"
and our SM rejected it because MITM is required when
`cfg.encrypted = true`. Pick a phone / app that supports passkey entry.
- **No passkey appears in UART** — verify `CONFIG_BT_NIMBLE_SM_SC=y` in
your sdkconfig and that you didn't toggle `cfg.encrypted` to `false`.
- **`enc_change status=13 encrypted=1 ...`** — `13` is `BLE_HS_ETIMEOUT`,
triggered by a benign race between our `ble_gap_security_initiate()`
and the central's own auto-encryption on a bonded reconnect. Status
is non-zero but the link is fully encrypted; safe to ignore.
- **Notifications missing after a reconnect** — `ble_uart_tx` deliberately
doesn't gate on the CCCD-subscribe state because bonded reconnects
often skip the CCCD write. Bytes are still pushed; the central
delivers them based on its remembered subscription.
@@ -0,0 +1,10 @@
# Both backends are listed; each .c file body is wrapped in
# #if CONFIG_BT_NIMBLE_ENABLED / CONFIG_BT_BLUEDROID_ENABLED so only
# the matching backend produces code. This is the standard IDF
# pattern for conditional sources, because sdkconfig isn't loaded
# during the early CMake component-requirement scan.
idf_component_register(SRCS "main.c"
"ble_uart_nimble.c"
"ble_uart_bluedroid.c"
INCLUDE_DIRS "."
REQUIRES bt nvs_flash)
@@ -0,0 +1,26 @@
menu "BLE UART Example"
config BLE_UART_DEVICE_NAME_PREFIX
string "BLE device name prefix"
default "BleUart"
help
The firmware advertises as `<prefix>-XXXX` where XXXX is
the last two bytes of the BT MAC in hex.
config BLE_UART_RX_SCRATCH_SIZE
int "RX scratch buffer size (bytes)"
range 64 16384
default 1024
help
Upper bound on a single RX payload delivered to
ble_uart_on_rx(). Covers both plain writes (MTU - 3 bytes)
and reassembled long writes (PREP + EXEC). Oversized writes
are rejected with ATT error 0x0D.
The buffer lives in BSS, so this value directly translates
into RAM cost. Bump it if your protocol sends larger frames
in one shot; with the NimBLE backend, raising past ~10 KB
may also require increasing
CONFIG_BT_NIMBLE_MSYS_1_BLOCK_COUNT.
endmenu
@@ -0,0 +1,154 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*
* BLE UART — turnkey serial-over-BLE peripheral.
*
* Implements the de-facto Nordic UART Service (NUS) GATT layout
* (RX write, TX notify) on top of either NimBLE or Bluedroid; the
* backend is picked at compile time via CONFIG_BT_NIMBLE_ENABLED /
* CONFIG_BT_BLUEDROID_ENABLED.
*
* Lifecycle:
*
* ble_uart_install(&cfg); // host + GATT service
* ble_uart_open(); // start advertising + auto-encrypt
* ...
* ble_uart_close(); // stop adv / disconnect / halt host
* ble_uart_uninstall(); // free port + reset state
*
* Run-forever apps only need install + open. close / uninstall is
* for apps that need to power BLE off at runtime.
*
* GATT layout (UUIDs fixed by the NUS spec):
*
* Service: 6e400001-b5a3-f393-e0a9-e50e24dcca9e
* RX : 6e400002-b5a3-f393-e0a9-e50e24dcca9e write
* TX : 6e400003-b5a3-f393-e0a9-e50e24dcca9e notify
*
* See PORTING.md for the integration guide.
*/
#pragma once
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
/* ----- Return codes --------------------------------------------------- */
/** All ble_uart_* APIs return one of these stack-neutral codes. */
#define BLE_UART_OK 0 /* Success */
#define BLE_UART_EINVAL -1 /* Bad argument or unsupported op */
#define BLE_UART_ENOTCONN -2 /* No central currently connected */
#define BLE_UART_ENOMEM -3 /* Out of mbufs / send queue full */
#define BLE_UART_EALREADY -4 /* Lifecycle already in this state */
#define BLE_UART_EFAIL -5 /* Backend internal failure (see logs) */
/* ----- 128-bit UUID helper -------------------------------------------- */
/** Stack-agnostic 128-bit UUID in little-endian (wire) order. */
typedef struct {
uint8_t bytes[16];
} ble_uart_uuid128_t;
/* ----- Configuration -------------------------------------------------- */
/** RX byte callback. Invoked from the BLE host task whenever bytes
* arrive on the RX characteristic. The buffer is owned by the stack
* and reused after return — copy what you need to keep.
*
* Don't block here; offload heavy work to your own task.
*
* Long-write (PREP/EXEC) reassembly is handled transparently — you
* always see one contiguous payload, capped by
* CONFIG_BLE_UART_RX_SCRATCH_SIZE (default 1024). Oversized writes
* are rejected with ATT error 0x0d. */
typedef void (*ble_uart_rx_cb_t)(const uint8_t *data, size_t len);
/** Configuration handed to ble_uart_install(). */
typedef struct {
/** True = LE Secure Connections + Bonding + MITM, DisplayOnly IO,
* encrypted RX/TX chars, bond persisted in NVS (NimBLE: requires
* CONFIG_BT_NIMBLE_NVS_PERSIST=y; Bluedroid: default).
* False = plaintext (lab debugging only — sniffable). */
bool encrypted;
/** GAP device name. NULL keeps the host stack default. Mind the
* 31-byte primary advertising limit (≤ 8 bytes recommended). */
const char *device_name;
/** Byte handler for RX writes. NULL discards incoming data. */
ble_uart_rx_cb_t ble_uart_on_rx;
} ble_uart_config_t;
/* ----- Lifecycle ------------------------------------------------------ */
/** Bring up host stack + Security Manager + SIG services + NUS GATT
* service. Caller must have already called nvs_flash_init().
* cfg->device_name is copied; doesn't need to outlive the call.
* Single-shot until ble_uart_uninstall(); a second call returns
* BLE_UART_EALREADY. */
int ble_uart_install(const ble_uart_config_t *cfg);
/** Start advertising. NimBLE: spawns the host task and primes the bond
* store; advertising begins once the controller signals ready.
* Bluedroid: triggers adv-data + scan-response config; advertising
* begins once the stack acknowledges both.
*
* Returns immediately; the BLE UART then runs autonomously
* (connect, pairing, passkey display, RX delivery all via internal
* callbacks). Single-shot. */
int ble_uart_open(void);
/** Counterpart to ble_uart_open(). Stops advertising, gracefully
* disconnects (waits up to 500 ms for LL_TERMINATE_IND ack), and
* quiesces the host. install state is preserved — call open() again
* to resume.
*
* Don't call from the BLE host task (i.e. from ble_uart_on_rx). */
int ble_uart_close(void);
/** Counterpart to ble_uart_install(). Force-closes if still open,
* then tears down the host stack + controller. After this returns,
* install() can run from scratch.
*
* Don't call from the BLE host task. */
int ble_uart_uninstall(void);
/* ----- TX ------------------------------------------------------------- */
/** Send raw bytes to the connected central as one or more TX
* notifications, fragmented to fit the live ATT MTU. Safe from any
* FreeRTOS task; not safe from ISR.
*
* Returns BLE_UART_ENOTCONN when no peer is connected (this is
* normal — typically just ignore). */
int ble_uart_tx(const uint8_t *data, size_t len);
/* ----- Status (best-effort, optional) -------------------------------- */
/** True when a central is connected (link may not yet be encrypted).
* Best-effort snapshot; production callers should rely on the return
* code of ble_uart_tx() instead. */
bool ble_uart_is_connected(void);
/** True when the central has subscribed to TX notifications.
* ble_uart_tx() does NOT gate on this (bonded reconnects often skip
* the CCCD write); exposed for diagnostics only. */
bool ble_uart_is_subscribed(void);
/* ----- Service UUID -------------------------------------------------- */
/** The NUS service UUID, exposed for custom advertising payloads.
* The two characteristic UUIDs are private to the backend. */
extern const ble_uart_uuid128_t ble_uart_service_uuid;
#ifdef __cplusplus
}
#endif
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,652 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*
* BLE UART — NimBLE backend. Implements the lifecycle declared in
* ble_uart.h on top of the NimBLE host. Active when
* CONFIG_BT_NIMBLE_ENABLED=y; otherwise ble_uart_bluedroid.c is used.
*/
#include "sdkconfig.h"
#if CONFIG_BT_NIMBLE_ENABLED
#include "ble_uart.h"
#include <assert.h>
#include <inttypes.h>
#include <string.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "esp_log.h"
#include "esp_random.h"
#include "nimble/ble.h"
#include "host/ble_att.h"
#include "host/ble_gap.h"
#include "host/ble_gatt.h"
#include "host/ble_hs.h"
#include "host/ble_hs_mbuf.h"
#include "host/ble_sm.h"
#include "host/ble_uuid.h"
#include "host/util/util.h"
#include "nimble/nimble_port.h"
#include "nimble/nimble_port_freertos.h"
#include "services/gap/ble_svc_gap.h"
#include "services/gatt/ble_svc_gatt.h"
/* tx path needs notifications; the disabled-path in
* ble_gatts_notify_custom() leaks the caller's mbuf. */
#if !MYNEWT_VAL(BLE_GATT_NOTIFY)
#error "ble_uart NimBLE backend requires MYNEWT_VAL(BLE_GATT_NOTIFY)=1"
#endif
static const char *TAG = "ble_uart";
/* Map NimBLE rc → public BLE_UART_E* code; unknown rcs → EFAIL. */
static int xlate_rc(int nimble_rc)
{
switch (nimble_rc) {
case 0: return BLE_UART_OK;
case BLE_HS_EINVAL: return BLE_UART_EINVAL;
case BLE_HS_ENOTCONN: return BLE_UART_ENOTCONN;
case BLE_HS_ENOMEM: return BLE_UART_ENOMEM;
case BLE_HS_EALREADY: return BLE_UART_EALREADY;
default: return BLE_UART_EFAIL;
}
}
/* Provided by NimBLE's `store/config` lib. */
extern void ble_store_config_init(void);
/* ===== UUIDs =========================================================== */
/* NUS UUIDs in little-endian byte order. */
#define NUS_SVC_BYTES 0x9e, 0xca, 0xdc, 0x24, 0x0e, 0xe5, 0xa9, 0xe0, \
0x93, 0xf3, 0xa3, 0xb5, 0x01, 0x00, 0x40, 0x6e
#define NUS_RX_BYTES 0x9e, 0xca, 0xdc, 0x24, 0x0e, 0xe5, 0xa9, 0xe0, \
0x93, 0xf3, 0xa3, 0xb5, 0x02, 0x00, 0x40, 0x6e
#define NUS_TX_BYTES 0x9e, 0xca, 0xdc, 0x24, 0x0e, 0xe5, 0xa9, 0xe0, \
0x93, 0xf3, 0xa3, 0xb5, 0x03, 0x00, 0x40, 0x6e
const ble_uart_uuid128_t ble_uart_service_uuid = { .bytes = { NUS_SVC_BYTES } };
static const ble_uuid128_t s_svc_uuid = BLE_UUID128_INIT(NUS_SVC_BYTES);
static const ble_uuid128_t s_chr_rx_uuid = BLE_UUID128_INIT(NUS_RX_BYTES);
static const ble_uuid128_t s_chr_tx_uuid = BLE_UUID128_INIT(NUS_TX_BYTES);
/* ===== State =========================================================== */
/* RX scratch capacity. Tunable via menuconfig; fall back to 1024 if
* Kconfig.projbuild isn't carried along when reusing this file. */
#ifndef CONFIG_BLE_UART_RX_SCRATCH_SIZE
#define CONFIG_BLE_UART_RX_SCRATCH_SIZE 1024
#endif
#define RX_SCRATCH CONFIG_BLE_UART_RX_SCRATCH_SIZE
/* Cached device name. Avoids ble_svc_gap_device_name() which returns
* NULL when CONFIG_BT_NIMBLE_GAP_SERVICE=n (would NULL-deref). 32B
* covers the BLE 31-byte adv-payload limit + NUL. */
#define DEV_NAME_MAX 32
static ble_uart_config_t s_cfg;
static char s_dev_name[DEV_NAME_MAX];
static uint16_t s_tx_val_handle;
/* Volatile: written from NimBLE host task, polled from caller task. */
static volatile uint16_t s_conn_handle = BLE_HS_CONN_HANDLE_NONE;
static bool s_subscribed;
static bool s_installed;
static bool s_opened;
static bool s_shutting_down; /* gates auto-readvertise during close */
static uint8_t s_own_addr_type;
static int gap_event(struct ble_gap_event *event, void *arg);
static int start_advertising(void);
/* ===== GATT (NUS) ====================================================== */
static int chr_access(uint16_t conn_handle, uint16_t attr_handle,
struct ble_gatt_access_ctxt *ctxt, void *arg)
{
switch (ctxt->op) {
case BLE_GATT_ACCESS_OP_WRITE_CHR: {
/* File-scope (BSS) — host task is single-threaded so no reentry. */
static uint8_t s_rx_buf[RX_SCRATCH];
uint16_t total = OS_MBUF_PKTLEN(ctxt->om);
if (total > sizeof(s_rx_buf)) {
ESP_LOGW(TAG, "rx oversize: %u > %u, rejecting",
(unsigned)total, (unsigned)sizeof(s_rx_buf));
return BLE_ATT_ERR_INVALID_ATTR_VALUE_LEN;
}
uint16_t copied = 0;
int rc = ble_hs_mbuf_to_flat(ctxt->om, s_rx_buf, total, &copied);
if (rc != 0) {
return BLE_ATT_ERR_UNLIKELY;
}
if (s_cfg.ble_uart_on_rx != NULL && copied > 0) {
s_cfg.ble_uart_on_rx(s_rx_buf, copied);
}
return 0;
}
case BLE_GATT_ACCESS_OP_READ_CHR:
return BLE_ATT_ERR_READ_NOT_PERMITTED;
default:
return BLE_ATT_ERR_UNLIKELY;
}
}
/* Encryption-required flag masks. NimBLE derives CCCD permissions from
* NOTIFY_INDICATE_* (not from READ/WRITE_*), so notify-only chars need
* the NOTIFY_INDICATE mask, not just the RW mask — otherwise an
* unpaired central could subscribe and receive notifications over the
* unencrypted link (see ble_gatts.c:ble_gatts_chr_clt_cfg_flags_from_chr_flags). */
#define CHR_FLAG_RW_ENC (BLE_GATT_CHR_F_READ_ENC | BLE_GATT_CHR_F_READ_AUTHEN | \
BLE_GATT_CHR_F_WRITE_ENC | BLE_GATT_CHR_F_WRITE_AUTHEN)
#define CHR_FLAG_NOTIFY_ENC (BLE_GATT_CHR_F_NOTIFY_INDICATE_ENC | \
BLE_GATT_CHR_F_NOTIFY_INDICATE_AUTHEN)
static struct ble_gatt_chr_def s_chr_defs[3];
static struct ble_gatt_svc_def s_svc_defs[2];
static void build_gatt_table(bool encrypted)
{
/* `ble_gatt_chr_flags` is uint32_t — match width here so the
* 0x10000-and-above NOTIFY_INDICATE flags don't get truncated. */
ble_gatt_chr_flags rw_enc = encrypted ? CHR_FLAG_RW_ENC : 0;
ble_gatt_chr_flags notify_enc = encrypted ? CHR_FLAG_NOTIFY_ENC : 0;
s_chr_defs[0] = (struct ble_gatt_chr_def){
.uuid = &s_chr_rx_uuid.u,
.access_cb = chr_access,
.flags = BLE_GATT_CHR_F_WRITE | BLE_GATT_CHR_F_WRITE_NO_RSP | rw_enc,
};
s_chr_defs[1] = (struct ble_gatt_chr_def){
.uuid = &s_chr_tx_uuid.u,
.access_cb = chr_access,
.flags = BLE_GATT_CHR_F_NOTIFY | notify_enc,
.val_handle = &s_tx_val_handle,
};
s_chr_defs[2] = (struct ble_gatt_chr_def){0};
s_svc_defs[0] = (struct ble_gatt_svc_def){
.type = BLE_GATT_SVC_TYPE_PRIMARY,
.uuid = &s_svc_uuid.u,
.characteristics = s_chr_defs,
};
s_svc_defs[1] = (struct ble_gatt_svc_def){0};
}
static void register_cb(struct ble_gatt_register_ctxt *ctxt, void *arg)
{
char buf[BLE_UUID_STR_LEN];
switch (ctxt->op) {
case BLE_GATT_REGISTER_OP_SVC:
ESP_LOGI(TAG, "registered service %s handle=%d",
ble_uuid_to_str(ctxt->svc.svc_def->uuid, buf), ctxt->svc.handle);
break;
case BLE_GATT_REGISTER_OP_CHR:
ESP_LOGI(TAG, "registered chr %s def=%d val=%d",
ble_uuid_to_str(ctxt->chr.chr_def->uuid, buf),
ctxt->chr.def_handle, ctxt->chr.val_handle);
break;
default:
break;
}
}
/* ===== TX ============================================================== */
int ble_uart_tx(const uint8_t *data, size_t len)
{
/* Snapshot conn_handle once: a peer-A→peer-B disconnect+connect
* race during a multi-chunk send could otherwise leak later chunks
* to peer B (notify_custom doesn't gate on the CCCD subscription).
* Stale handle → BLE_HS_ENOTCONN, we bail cleanly. */
uint16_t conn_handle = s_conn_handle;
if (conn_handle == BLE_HS_CONN_HANDLE_NONE) {
return BLE_UART_ENOTCONN;
}
if (data == NULL || len == 0) {
return BLE_UART_EINVAL;
}
uint16_t mtu = ble_att_mtu(conn_handle);
size_t chunk = (mtu > 3) ? (size_t)(mtu - 3) : 20;
size_t sent = 0;
while (sent < len) {
size_t n = len - sent;
if (n > chunk) {
n = chunk;
}
struct os_mbuf *om = ble_hs_mbuf_from_flat(data + sent, n);
if (om == NULL) {
return BLE_UART_ENOMEM;
}
int rc = ble_gatts_notify_custom(conn_handle, s_tx_val_handle, om);
if (rc != 0) {
ESP_LOGW(TAG, "notify failed: rc=%d", rc);
/* Callee frees om on every failure path EXCEPT the
* BLE_GATT_NOTIFY-disabled early-return (BLE_HS_ENOTSUP).
* Freeing on any other rc would be a double free. */
if (rc == BLE_HS_ENOTSUP) {
os_mbuf_free_chain(om);
}
return xlate_rc(rc);
}
sent += n;
}
return BLE_UART_OK;
}
/* Best-effort snapshots; see header for threading caveat. */
bool ble_uart_is_connected(void) { return s_conn_handle != BLE_HS_CONN_HANDLE_NONE; }
bool ble_uart_is_subscribed(void) { return s_subscribed; }
/* ===== Advertising ==================================================== */
static int start_advertising(void)
{
/* 31-byte primary adv can't hold flags + tx_pwr + name + 128-bit
* UUID together, so split: primary = flags+tx_pwr+name,
* scan rsp = NUS UUID. */
const char *name = s_dev_name;
size_t name_len = strlen(name);
struct ble_hs_adv_fields adv = {
.flags = BLE_HS_ADV_F_DISC_GEN | BLE_HS_ADV_F_BREDR_UNSUP,
.tx_pwr_lvl_is_present = 1,
.tx_pwr_lvl = BLE_HS_ADV_TX_PWR_LVL_AUTO,
/* If no name was set, advertise without one (NimBLE accepts
* NULL+0); the NUS UUID in scan rsp still identifies us. */
.name = name_len > 0 ? (uint8_t *)name : NULL,
.name_len = name_len,
.name_is_complete = name_len > 0 ? 1 : 0,
};
int rc = ble_gap_adv_set_fields(&adv);
if (rc != 0) {
ESP_LOGE(TAG, "adv_set_fields rc=%d (name too long?)", rc);
return rc;
}
struct ble_hs_adv_fields rsp = {
.uuids128 = &s_svc_uuid,
.num_uuids128 = 1,
.uuids128_is_complete = 1,
};
rc = ble_gap_adv_rsp_set_fields(&rsp);
if (rc != 0) {
ESP_LOGE(TAG, "adv_rsp_set_fields rc=%d", rc);
return rc;
}
struct ble_gap_adv_params params = {
.conn_mode = BLE_GAP_CONN_MODE_UND,
.disc_mode = BLE_GAP_DISC_MODE_GEN,
};
rc = ble_gap_adv_start(s_own_addr_type, NULL, BLE_HS_FOREVER,
&params, gap_event, NULL);
if (rc != 0) {
ESP_LOGE(TAG, "adv_start rc=%d", rc);
return rc;
}
ESP_LOGI(TAG, "advertising as '%s'", name_len > 0 ? name : "<no name>");
return 0;
}
/* ===== GAP event handler ============================================== */
static void show_passkey(uint32_t passkey)
{
ESP_LOGW(TAG, "");
ESP_LOGW(TAG, " +-----------------------------+");
ESP_LOGW(TAG, " | BLE PAIRING PASSKEY: |");
ESP_LOGW(TAG, " | %06" PRIu32 " |", passkey);
ESP_LOGW(TAG, " +-----------------------------+");
ESP_LOGW(TAG, "");
}
static int gap_event(struct ble_gap_event *event, void *arg)
{
struct ble_gap_conn_desc desc;
switch (event->type) {
case BLE_GAP_EVENT_CONNECT:
ESP_LOGI(TAG, "connect %s status=%d handle=%d",
event->connect.status == 0 ? "ok" : "failed",
event->connect.status,
event->connect.conn_handle);
if (event->connect.status == 0) {
s_conn_handle = event->connect.conn_handle;
s_subscribed = false;
/* Start pairing immediately (rather than lazily on the
* first encrypted attribute access). */
if (s_cfg.encrypted) {
ble_gap_security_initiate(event->connect.conn_handle);
}
} else if (!s_shutting_down) {
start_advertising();
}
return 0;
case BLE_GAP_EVENT_DISCONNECT:
ESP_LOGI(TAG, "disconnect reason=%d", event->disconnect.reason);
s_conn_handle = BLE_HS_CONN_HANDLE_NONE;
s_subscribed = false;
if (!s_shutting_down) {
start_advertising();
}
return 0;
case BLE_GAP_EVENT_CONN_UPDATE:
ESP_LOGI(TAG, "conn_update status=%d", event->conn_update.status);
return 0;
case BLE_GAP_EVENT_ADV_COMPLETE:
ESP_LOGI(TAG, "adv_complete reason=%d", event->adv_complete.reason);
if (!s_shutting_down) {
start_advertising();
}
return 0;
case BLE_GAP_EVENT_ENC_CHANGE:
if (ble_gap_conn_find(event->enc_change.conn_handle, &desc) == 0) {
ESP_LOGI(TAG, "enc_change status=%d encrypted=%d authenticated=%d bonded=%d",
event->enc_change.status,
desc.sec_state.encrypted,
desc.sec_state.authenticated,
desc.sec_state.bonded);
}
return 0;
case BLE_GAP_EVENT_REPEAT_PAIRING:
/* Drop old keys + retry rather than reject. */
if (ble_gap_conn_find(event->repeat_pairing.conn_handle, &desc) == 0) {
ble_store_util_delete_peer(&desc.peer_id_addr);
}
return BLE_GAP_REPEAT_PAIRING_RETRY;
case BLE_GAP_EVENT_PASSKEY_ACTION:
if (event->passkey.params.action == BLE_SM_IOACT_DISP) {
/* Rejection sampling avoids the modulo bias of
* `esp_random() % 1000000` (2^32 % 1e6 != 0). */
const uint32_t passkey_max = 1000000U;
const uint32_t reject_above = UINT32_MAX -
(UINT32_MAX % passkey_max);
uint32_t r;
do {
r = esp_random();
} while (r >= reject_above);
struct ble_sm_io pkey = {
.action = BLE_SM_IOACT_DISP,
.passkey = r % passkey_max,
};
show_passkey(pkey.passkey);
int rc = ble_sm_inject_io(event->passkey.conn_handle, &pkey);
if (rc != 0) {
ESP_LOGW(TAG, "ble_sm_inject_io rc=%d", rc);
}
} else {
ESP_LOGW(TAG, "passkey action %d not handled (DisplayOnly only)",
event->passkey.params.action);
}
return 0;
case BLE_GAP_EVENT_MTU:
ESP_LOGI(TAG, "mtu=%d (conn=%d)",
event->mtu.value, event->mtu.conn_handle);
return 0;
case BLE_GAP_EVENT_SUBSCRIBE:
ESP_LOGI(TAG, "subscribe attr=%d cur_notify=%d",
event->subscribe.attr_handle, event->subscribe.cur_notify);
if (event->subscribe.attr_handle == s_tx_val_handle) {
s_subscribed = (event->subscribe.cur_notify != 0);
}
return 0;
default:
return 0;
}
}
/* ===== Host plumbing =================================================== */
static void on_reset(int reason)
{
ESP_LOGE(TAG, "Resetting NimBLE state; reason=%d", reason);
}
static void on_sync(void)
{
int rc = ble_hs_util_ensure_addr(0);
assert(rc == 0);
rc = ble_hs_id_infer_auto(0, &s_own_addr_type);
if (rc != 0) {
ESP_LOGE(TAG, "infer addr type rc=%d", rc);
return;
}
uint8_t addr[6] = {0};
ble_hs_id_copy_addr(s_own_addr_type, addr, NULL);
ESP_LOGI(TAG, "addr=%02x:%02x:%02x:%02x:%02x:%02x",
addr[5], addr[4], addr[3], addr[2], addr[1], addr[0]);
start_advertising();
}
static void nimble_host_task(void *param)
{
ESP_LOGI(TAG, "BLE host task started");
nimble_port_run();
nimble_port_freertos_deinit();
}
/* ===== Public lifecycle ================================================ */
int ble_uart_install(const ble_uart_config_t *cfg)
{
if (s_installed) {
ESP_LOGW(TAG, "ble_uart_install called twice; ignoring");
return BLE_UART_EALREADY;
}
if (cfg != NULL) {
s_cfg = *cfg;
} else {
memset(&s_cfg, 0, sizeof(s_cfg));
}
esp_err_t err = nimble_port_init();
if (err != ESP_OK) {
ESP_LOGE(TAG, "nimble_port_init rc=%d", err);
return BLE_UART_EFAIL;
}
/* From here every failure must `goto fail` so nimble_port_deinit()
* runs — leaving the port allocated breaks the next install(). */
ble_hs_cfg.reset_cb = on_reset;
ble_hs_cfg.sync_cb = on_sync;
ble_hs_cfg.store_status_cb = ble_store_util_status_rr;
ble_hs_cfg.gatts_register_cb = register_cb;
/* Encrypted = LE Secure Connections + Bonding + MITM, DisplayOnly.
* Plaintext = SM disabled. */
if (s_cfg.encrypted) {
ble_hs_cfg.sm_io_cap = BLE_HS_IO_DISPLAY_ONLY;
ble_hs_cfg.sm_sc = 1;
ble_hs_cfg.sm_bonding = 1;
ble_hs_cfg.sm_mitm = 1;
ble_hs_cfg.sm_our_key_dist = BLE_SM_PAIR_KEY_DIST_ENC | BLE_SM_PAIR_KEY_DIST_ID;
ble_hs_cfg.sm_their_key_dist = BLE_SM_PAIR_KEY_DIST_ENC | BLE_SM_PAIR_KEY_DIST_ID;
} else {
ble_hs_cfg.sm_io_cap = BLE_HS_IO_NO_INPUT_OUTPUT;
ble_hs_cfg.sm_sc = 0;
ble_hs_cfg.sm_bonding = 0;
ble_hs_cfg.sm_mitm = 0;
}
ble_svc_gap_init();
ble_svc_gatt_init();
/* Cache the device name into our own buffer (caller's pointer may
* not outlive this call; also avoids the GAP-service stub path
* which returns NULL from ble_svc_gap_device_name()). */
int rc = 0;
if (s_cfg.device_name != NULL) {
strncpy(s_dev_name, s_cfg.device_name, sizeof(s_dev_name) - 1);
s_dev_name[sizeof(s_dev_name) - 1] = '\0';
s_cfg.device_name = NULL;
/* Best-effort: also set in the GAP service for peer reads.
* Returns -1 on the stub path — fine, we already cached locally. */
rc = ble_svc_gap_device_name_set(s_dev_name);
if (rc != 0) {
ESP_LOGI(TAG, "ble_svc_gap_device_name_set rc=%d (GAP service stubbed?)",
rc);
rc = 0;
}
} else {
s_dev_name[0] = '\0';
}
build_gatt_table(s_cfg.encrypted);
rc = ble_gatts_count_cfg(s_svc_defs);
if (rc != 0) {
ESP_LOGE(TAG, "ble_gatts_count_cfg rc=%d", rc);
goto fail;
}
rc = ble_gatts_add_svcs(s_svc_defs);
if (rc != 0) {
ESP_LOGE(TAG, "ble_gatts_add_svcs rc=%d", rc);
goto fail;
}
s_installed = true;
return BLE_UART_OK;
fail:
nimble_port_deinit();
memset(&s_cfg, 0, sizeof(s_cfg));
return xlate_rc(rc);
}
int ble_uart_open(void)
{
if (!s_installed) {
ESP_LOGE(TAG, "ble_uart_open before ble_uart_install");
return BLE_UART_EINVAL;
}
if (s_opened) {
ESP_LOGW(TAG, "ble_uart_open called twice; ignoring");
return BLE_UART_EALREADY;
}
/* NVS bond store (requires CONFIG_BT_NIMBLE_NVS_PERSIST=y). */
ble_store_config_init();
/* Spawn host task; on_sync starts advertising once controller is ready. */
nimble_port_freertos_init(nimble_host_task);
s_opened = true;
return BLE_UART_OK;
}
int ble_uart_close(void)
{
if (!s_opened) {
return BLE_UART_EALREADY;
}
/* Latch first so GAP events stop re-arming advertising. */
s_shutting_down = true;
int rc = ble_gap_adv_stop();
if (rc != 0 && rc != BLE_HS_EALREADY) {
ESP_LOGW(TAG, "adv_stop rc=%d", rc);
}
/* Graceful disconnect: wait up to 500 ms for the disconnect event
* so the peer sees a proper LL_TERMINATE_IND, not a controller-yank. */
if (s_conn_handle != BLE_HS_CONN_HANDLE_NONE) {
rc = ble_gap_terminate(s_conn_handle, BLE_ERR_REM_USER_CONN_TERM);
if (rc != 0 && rc != BLE_HS_EALREADY) {
ESP_LOGW(TAG, "ble_gap_terminate rc=%d", rc);
}
for (int i = 0; i < 50 && s_conn_handle != BLE_HS_CONN_HANDLE_NONE; i++) {
vTaskDelay(pdMS_TO_TICKS(10));
}
if (s_conn_handle != BLE_HS_CONN_HANDLE_NONE) {
ESP_LOGW(TAG, "disconnect timed out; tearing down anyway");
}
}
/* nimble_host_task self-cleans (port_freertos_deinit + delete) when
* port_run returns, so no explicit join. */
rc = nimble_port_stop();
if (rc != 0) {
ESP_LOGE(TAG, "nimble_port_stop rc=%d", rc);
s_shutting_down = false;
return BLE_UART_EFAIL;
}
s_conn_handle = BLE_HS_CONN_HANDLE_NONE;
s_subscribed = false;
s_opened = false;
s_shutting_down = false;
return BLE_UART_OK;
}
int ble_uart_uninstall(void)
{
if (!s_installed) {
return BLE_UART_EALREADY;
}
/* Best-effort cleanup. Do NOT early-return on a per-step failure:
* leaving s_installed=true with partially torn-down NimBLE state
* makes the module unrecoverable (can't re-install, can't retry
* uninstall cleanly). Mirror the Bluedroid backend: record the
* first error, keep tearing down, and always wipe our state. */
int first_rc = BLE_UART_OK;
if (s_opened) {
int rc = ble_uart_close();
if (rc != BLE_UART_OK && rc != BLE_UART_EALREADY) {
ESP_LOGE(TAG, "ble_uart_close rc=%d", rc);
if (first_rc == BLE_UART_OK) {
first_rc = rc;
}
}
}
/* Best-effort: even if port_deinit fails, wipe our state anyway —
* otherwise s_installed stays true and the module is unrecoverable
* (can't re-install, can't retry uninstall cleanly). */
esp_err_t err = nimble_port_deinit();
if (err != ESP_OK) {
ESP_LOGE(TAG, "nimble_port_deinit rc=%d", err);
if (first_rc == BLE_UART_OK) {
first_rc = BLE_UART_EFAIL;
}
}
memset(&s_cfg, 0, sizeof(s_cfg));
s_dev_name[0] = '\0';
s_tx_val_handle = 0;
s_conn_handle = BLE_HS_CONN_HANDLE_NONE;
s_subscribed = false;
s_own_addr_type = 0;
s_shutting_down = false;
s_installed = false;
s_opened = false;
return first_rc;
}
#endif /* CONFIG_BT_NIMBLE_ENABLED */
@@ -0,0 +1,63 @@
/*
* SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Unlicense OR CC0-1.0
*
* BLE UART Service example. Backend (NimBLE / Bluedroid) is picked
* by the host-stack Kconfig at compile time. Whatever the central
* writes to the RX characteristic is echoed back over TX.
*/
#include <stdio.h>
#include "esp_log.h"
#include "esp_mac.h"
#include "nvs_flash.h"
#include "sdkconfig.h"
#include "ble_uart.h"
static const char *TAG = "app";
static void ble_uart_on_rx(const uint8_t *data, size_t len)
{
ESP_LOGI(TAG, "rx len: %u bytes", (unsigned)len);
if (data == NULL || len == 0) {
return;
}
ESP_LOG_BUFFER_HEX(TAG, data, len);
ble_uart_tx(data, len); /* echo back */
}
void app_main(void)
{
/* NVS is required by the BT controller (PHY calibration) and the
* bond store, so it must be live before ble_uart_install(). */
esp_err_t err = nvs_flash_init();
if (err == ESP_ERR_NVS_NO_FREE_PAGES || err == ESP_ERR_NVS_NEW_VERSION_FOUND) {
ESP_ERROR_CHECK(nvs_flash_erase());
err = nvs_flash_init();
}
ESP_ERROR_CHECK(err);
/* Device name = "<prefix>-XXXX" with XXXX = last two MAC bytes.
* If esp_read_mac() fails, mac stays zero and the suffix degrades
* to "0000" — log so the operator notices uniqueness was lost. */
uint8_t mac[6] = {0};
esp_err_t mac_err = esp_read_mac(mac, ESP_MAC_BT);
if (mac_err != ESP_OK) {
ESP_LOGW(TAG, "esp_read_mac(BT) failed (%s); device name suffix will be 0000",
esp_err_to_name(mac_err));
}
char name[24];
snprintf(name, sizeof(name), "%s-%02X%02X",
CONFIG_BLE_UART_DEVICE_NAME_PREFIX, mac[4], mac[5]);
ESP_ERROR_CHECK(ble_uart_install(&(ble_uart_config_t){
.encrypted = true,
.device_name = name,
.ble_uart_on_rx = ble_uart_on_rx,
}));
ESP_ERROR_CHECK(ble_uart_open());
}
@@ -0,0 +1,42 @@
# Overlay applied on top of sdkconfig.defaults to switch the example
# from the default NimBLE backend to Bluedroid. Use it like:
#
# idf.py -B build_bd \
# -D SDKCONFIG_DEFAULTS="sdkconfig.defaults;sdkconfig.ci.bluedroid" \
# reconfigure
# idf.py -B build_bd build flash monitor
#
# When this overlay wins, main/CMakeLists.txt links ble_uart_bluedroid.c
# instead of ble_uart_nimble.c. The public ble_uart.h API is identical
# either way.
CONFIG_BT_ENABLED=y
CONFIG_BT_NIMBLE_ENABLED=n
CONFIG_BT_BLUEDROID_ENABLED=y
# LE Secure Connections + bonding (matches the NimBLE side).
# CONFIG_BT_SMP_ENABLE is derived from this and BT_CLASSIC_ENABLED, so
# we don't set it explicitly.
CONFIG_BT_BLE_SMP_ENABLE=y
# Note: Bluedroid has no compile-time MTU Kconfig (the NimBLE
# CONFIG_BT_NIMBLE_ATT_PREFERRED_MTU has no Bluedroid counterpart).
# To negotiate a larger ATT MTU at runtime, the application calls
# esp_ble_gatt_set_local_mtu(<bytes>) before peers connect.
# ble_uart_tx auto-fragments to whatever MTU is live, so the default
# 23 also works — just at lower throughput.
# Service-table API is needed for esp_ble_gatts_create_attr_tab().
CONFIG_BT_GATTS_ENABLE=y
# CONFIG_BT_GATTC_ENABLE is not set
# CONFIG_BT_BLE_50_FEATURES_SUPPORTED is not set
CONFIG_BT_BLE_42_FEATURES_SUPPORTED=y
# CONFIG_BT_BLE_42_DTM_TEST_EN is not set
CONFIG_BT_BLE_42_ADV_EN=y
# CONFIG_BT_BLE_42_SCAN_EN is not set
@@ -0,0 +1,22 @@
# Bluetooth controller in BLE-only mode + NimBLE host stack.
# CONFIG_BTDM_CTRL_MODE_* are ESP32-classic-only knobs; on C2/C3/C5/C6/
# C61/H2/H4/S3 they may emit an "unknown symbol" warning at configure
# time but are otherwise harmless, so this single sdkconfig.defaults
# stays valid for every supported target.
CONFIG_BT_ENABLED=y
CONFIG_BTDM_CTRL_MODE_BLE_ONLY=y
CONFIG_BTDM_CTRL_MODE_BR_EDR_ONLY=n
CONFIG_BTDM_CTRL_MODE_BTDM=n
CONFIG_BT_BLUEDROID_ENABLED=n
CONFIG_BT_NIMBLE_ENABLED=y
# Negotiate the largest ATT MTU we can; ble_uart_tx auto-fragments to
# the live MTU so smaller-MTU centrals still work.
CONFIG_BT_NIMBLE_ATT_PREFERRED_MTU=512
# LE Secure Connections + persistent bond store. The example defaults to
# encrypted operation (ble_uart_config_t::encrypted=true) and stores LTK
# in NVS so a previously-paired peer reconnects without re-prompting for
# the passkey.
CONFIG_BT_NIMBLE_SM_SC=y
CONFIG_BT_NIMBLE_NVS_PERSIST=y
+273
View File
@@ -0,0 +1,273 @@
<!-- SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD -->
<!-- SPDX-License-Identifier: Apache-2.0 -->
# BLE UART Bridge
BLE UART Bridge is a host-side utility for talking to ESP-IDF applications that expose a BLE UART-style GATT service. It provides a reusable Python transport layer, an interactive console for manual testing, and a daemon mode for simple local IPC request/response workflows.
## Table of contents
- [Quick Start](#quick-start) - install dependencies and run the first commands
- [CLI overview](#cli-overview) - command list and common Console/Daemon workflows
- [Typical Console workflow](#typical-console-workflow)
- [Typical Daemon workflow](#typical-daemon-workflow)
- [Custom scripts and porting](#custom-scripts-and-porting)
- [What is included](#what-is-included) - directory layout and component roles
- [Core](#core)
- [Console](#console)
- [Daemon](#daemon)
- [Choosing Core, Console, or Daemon](#choosing-core-console-or-daemon)
- [Profile compatibility](#profile-compatibility)
- [Dependencies](#dependencies)
- [Limitations](#limitations)
- [Further reading](#further-reading)
## Quick Start
You can reuse the ESP-IDF Python environment, or use your own Python virtual environment. If you reuse the ESP-IDF environment, export it first and then install the additional BLE UART Bridge dependencies:
```bash
cd $IDF_PATH
. ./export.sh
cd tools/ble/ble_uart_bridge
python -m pip install -r requirements.txt
```
On Windows, run `export.bat` or `export.ps1` from the ESP-IDF root directory before installing `requirements.txt`. If you use your own Python virtual environment instead, activate it before running:
```bash
cd tools/ble/ble_uart_bridge
python -m pip install -r requirements.txt
```
List nearby BLE UART devices:
```bash
python main.py list-devices
```
Use the printed device identifier as `DEVICE_ID` in later commands. On macOS, this identifier is a CoreBluetooth UUID and is different from the device MAC address.
Check whether the device can be connected:
```bash
python main.py connection-check DEVICE_ID
```
Open an interactive BLE UART Console:
```bash
python main.py console DEVICE_ID
```
For Console options such as line endings, hex mode, and write-with-response, see [Quick-Start-BLE-UART-Console.md](docs/Quick-Start-BLE-UART-Console.md). If you need firmware to test against, use the [BLE UART Service example](../../../examples/bluetooth/ble_uart_service) as an Echo Server: it advertises the default Nordic UART Service profile and echoes RX writes back through TX notifications.
Run the BLE UART Daemon:
```bash
python main.py daemon DEVICE_ID
```
In another terminal, check daemon status and send a request:
```bash
python main.py daemon-status
python main.py daemon-send --op echo "hello"
python main.py daemon-notify --op set_led --json '{"state": true}'
```
For Daemon details, the HTTP API, and the JSONL RPC protocol, see [Quick-Start-BLE-UART-Daemon.md](docs/Quick-Start-BLE-UART-Daemon.md).
## CLI overview
Run:
```bash
python main.py --help
```
Available commands:
```bash
python main.py list-devices
python main.py connection-check DEVICE_ID
python main.py console DEVICE_ID
python main.py daemon DEVICE_ID
python main.py daemon-status
python main.py daemon-send DATA
python main.py daemon-notify DATA
```
### Typical Console workflow
Use Console when you want to manually test a BLE UART device from a terminal UI. For a known-compatible target, build and flash the [BLE UART Service example](../../../examples/bluetooth/ble_uart_service), which acts as an Echo Server for Console smoke tests:
```bash
python main.py list-devices
python main.py connection-check DEVICE_ID
python main.py console DEVICE_ID
```
Common Console variants:
```bash
# Use CRLF for AT-style commands
python main.py console DEVICE_ID --terminator crlf
# Send and display raw bytes in hex
python main.py console DEVICE_ID --encoding hex
# Use BLE write-with-response
python main.py console DEVICE_ID --with-response
```
For the full Console guide, see [Quick-Start-BLE-UART-Console.md](docs/Quick-Start-BLE-UART-Console.md).
### Typical Daemon workflow
Use Daemon when a local script, editor integration, or automation tool needs request/response access to a BLE UART device.
Terminal 1 starts the daemon and owns the BLE connection:
```bash
python main.py daemon DEVICE_ID
```
Terminal 2 checks status and sends requests through the daemon:
```bash
python main.py daemon-status
python main.py daemon-send --op echo "hello"
python main.py daemon-send --op set_led --json '{"state": true}'
python main.py daemon-notify --op set_led --json '{"state": true}'
```
For the HTTP API and JSONL RPC wire protocol, see [Quick-Start-BLE-UART-Daemon.md](docs/Quick-Start-BLE-UART-Daemon.md).
### Custom scripts and porting
Use the Core API directly when you want your own Python script to own the BLE connection, implement custom framing, or integrate BLE UART into a larger automation flow.
For examples using `BLEUARTBridge`, RX handlers, byte payloads, custom `BLEUARTProfile`, and custom request/response logic, see [PORTING.md](docs/PORTING.md).
## What is included
```text
tools/ble/ble_uart_bridge/
├── main.py
├── requirements.txt
├── README.md
├── docs/
│ ├── Quick-Start-BLE-UART-Console.md
│ ├── Quick-Start-BLE-UART-Daemon.md
│ ├── Profile-Compatibility.md
│ └── PORTING.md
└── src/
├── core/
├── console/
└── daemon/
```
### Core
The Core component is the reusable BLE transport layer.
Use it when you want to write your own Python script or tool on top of BLE UART without reimplementing scanning, connection management, notification subscription, and chunked GATT writes.
Main responsibilities:
- Scan for BLE UART devices.
- Check whether a target device can be connected.
- Connect and disconnect with a BLE UART GATT profile.
- Subscribe to device-to-host notifications.
- Send host-to-device data as `str`, `bytes`, or `bytearray`.
- Support a default NUS profile and user-defined BLE UART profiles.
Important APIs:
- `BLEUARTBridge`
- `BLEUARTProfile`
- `run_list_devices()`
- `run_connection_check()`
Additional models are available from `src.core.models`, including `DeviceInfo` and `ConnectionState`.
### Console
The Console component is an interactive terminal UI for quick BLE UART testing.
Use it when you want to manually type data into a BLE UART device and observe received data without writing code.
Main responsibilities:
- Open an interactive Textual-based UI.
- Display TX, RX, and INFO logs separately.
- Send text lines with configurable line terminators.
- Send and display raw bytes in hex mode.
- Optionally use BLE write-with-response.
- Detect disconnects and show a notice in the UI.
### Daemon
The Daemon component exposes BLE UART as a local HTTP service.
Use it when another local tool, script, editor integration, or automation process needs request/response or fire-and-forget IPC with a BLE UART device.
Main responsibilities:
- Keep one BLE UART connection open in a background server process.
- Expose local HTTP endpoints for status, request/response, and fire-and-forget calls.
- Encode requests as newline-delimited JSON messages over BLE UART.
- Correlate device responses by request ID.
- Provide a small JSONL RPC-style envelope as an example protocol.
The daemon protocol is intentionally small. It is not a full RPC framework. It demonstrates a portable pattern that users can copy into firmware or extend in their own application protocol.
By default, the daemon binds to `127.0.0.1`. Keep it on a loopback address unless you add your own network access control, because the daemon exposes unauthenticated HTTP endpoints that can send data to the BLE device.
## Choosing Core, Console, or Daemon
| Component | Best for | Interface |
| --- | --- | --- |
| Core | Custom Python tools and scripts | Python API |
| Console | Manual BLE UART smoke tests | Interactive TUI |
| Daemon | Local IPC and automation | HTTP + CLI client |
Use Core when your business logic lives in Python. Use Console when you only need to manually test a BLE UART endpoint. Use Daemon when multiple local processes need to share one BLE connection through a simple request/response or notification boundary.
## Profile compatibility
The default profile is compatible with the Nordic UART Service (NUS):
- Service UUID: `6E400001-B5A3-F393-E0A9-E50E24DCCA9E`
- RX characteristic UUID, host to device: `6E400002-B5A3-F393-E0A9-E50E24DCCA9E`
- TX characteristic UUID, device to host: `6E400003-B5A3-F393-E0A9-E50E24DCCA9E`
For ESP-IDF BLE SPP examples and custom profile mapping details, see [Profile-Compatibility.md](docs/Profile-Compatibility.md).
## Dependencies
The tool depends on:
- `bleak` for BLE host access
- `textual` and `rich` for the console UI
- `fastapi` and `uvicorn` for daemon mode
- `typer` for the CLI
- `loguru` for logging
## Limitations
- Custom GATT UUIDs require constructing `BLEUARTProfile` in Python code.
- Daemon mode is single-flight for request/response calls: it processes one `/request` at a time. `/notify` sends without waiting for a device response.
- Daemon mode does not automatically reconnect after a BLE disconnect; restart the daemon after the device starts advertising again.
- Daemon `/request` and `/notify` limit `op` to 64 characters and JSON-encoded `data` to 4096 bytes.
- The JSONL RPC protocol is a demonstration envelope, not a complete RPC framework.
- Unmatched device messages are logged as unsolicited messages and are not exposed as a streaming API.
- The daemon request framing is newline-delimited JSON; device firmware must send a newline after every JSON response.
## Further reading
- [Quick-Start-BLE-UART-Console.md](docs/Quick-Start-BLE-UART-Console.md)
- [Quick-Start-BLE-UART-Daemon.md](docs/Quick-Start-BLE-UART-Daemon.md)
- [Profile-Compatibility.md](docs/Profile-Compatibility.md)
- [PORTING.md](docs/PORTING.md)
+286
View File
@@ -0,0 +1,286 @@
<!-- SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD -->
<!-- SPDX-License-Identifier: Apache-2.0 -->
# Porting BLE UART Bridge to Custom Scripts
This guide explains how to reuse BLE UART Bridge in your own Python scripts.
Use the Core API when the Console and Daemon are not the right abstraction for your application. For example, use Core directly when you want to implement custom framing, a test harness, a device provisioning flow, or a domain-specific automation script.
## Choose the right integration level
| Need | Recommended integration |
| --- | --- |
| Manual testing | Use `python main.py console DEVICE_ID` |
| Local process talks to a BLE device through HTTP | Use Daemon mode |
| Custom Python logic owns the BLE connection | Use `BLEUARTBridge` directly |
| Custom service UUIDs or characteristics | Use `BLEUARTProfile` with `BLEUARTBridge` |
## Install dependencies
You can reuse the ESP-IDF Python environment, or use your own Python virtual environment. If you reuse the ESP-IDF environment, export it first and then install the extra dependencies required by BLE UART Bridge:
```bash
cd $IDF_PATH
. ./export.sh
cd tools/ble/ble_uart_bridge
python -m pip install -r requirements.txt
```
On Windows, run `export.bat` or `export.ps1` from the ESP-IDF root directory before installing `requirements.txt`. If you use your own Python virtual environment instead, activate it before installing `requirements.txt`.
When importing from a script outside this directory, make sure `tools/ble/ble_uart_bridge` is on `PYTHONPATH`, or run the script from this directory.
Example:
```bash
PYTHONPATH=tools/ble/ble_uart_bridge python my_script.py
```
## Basic script
The simplest script connects, sends one line, and disconnects:
```python
import asyncio
from src.core import BLEUARTBridge
async def main() -> None:
bridge = BLEUARTBridge("AA:BB:CC:DD:EE:FF")
try:
if not await bridge.connect():
raise RuntimeError("failed to connect")
await bridge.send("hello\n")
finally:
await bridge.disconnect()
asyncio.run(main())
```
## Receive data with handlers
Register one or more RX handlers before connecting:
```python
import asyncio
from src.core import BLEUARTBridge
def print_rx(data: bytearray) -> None:
print("RX:", data.decode(errors="replace"))
async def main() -> None:
bridge = BLEUARTBridge("AA:BB:CC:DD:EE:FF")
bridge.add_rx_handler(print_rx)
try:
if not await bridge.connect():
raise RuntimeError("failed to connect")
await bridge.send("help\n")
await asyncio.sleep(2)
finally:
await bridge.disconnect()
asyncio.run(main())
```
Handlers are synchronous callables. If your application needs async processing, push received data into an `asyncio.Queue`:
```python
import asyncio
from src.core import BLEUARTBridge
async def main() -> None:
bridge = BLEUARTBridge("AA:BB:CC:DD:EE:FF")
rx_queue: asyncio.Queue[bytes] = asyncio.Queue()
def enqueue_rx(data: bytearray) -> None:
rx_queue.put_nowait(bytes(data))
bridge.add_rx_handler(enqueue_rx)
try:
if not await bridge.connect():
raise RuntimeError("failed to connect")
await bridge.send("status\n")
data = await asyncio.wait_for(rx_queue.get(), timeout=5.0)
print("RX:", data)
finally:
await bridge.disconnect()
asyncio.run(main())
```
## Send bytes instead of text
`BLEUARTBridge.send()` accepts `str`, `bytes`, and `bytearray`.
```python
await bridge.send(b"\x01\x02\x03\x0a")
await bridge.send(bytearray([0x01, 0x02, 0x03, 0x0A]))
```
Use `with_response=True` when the target characteristic or debugging workflow should use BLE write-with-response:
```python
await bridge.send(b"\x01\x02", with_response=True)
```
## Use a custom BLE UART profile
The default profile uses Nordic UART Service UUIDs. For custom firmware, create a `BLEUARTProfile`:
```python
from src.core import BLEUARTBridge
from src.core import BLEUARTProfile
profile = BLEUARTProfile(
service_uuid="00000000-0000-0000-0000-000000000001",
rx_char_uuid="00000000-0000-0000-0000-000000000002",
tx_char_uuid="00000000-0000-0000-0000-000000000003",
)
bridge = BLEUARTBridge("AA:BB:CC:DD:EE:FF", profile=profile)
```
The naming follows BLE UART convention:
- RX characteristic: host writes to device.
- TX characteristic: device notifies host.
## Implement your own request/response protocol
If your script needs request/response semantics, use a queue or future map and correlate responses at the application layer.
The daemon uses a lightweight JSONL envelope. You can reuse the same pattern:
```python
import asyncio
import json
from uuid import uuid4
from src.core import BLEUARTBridge
async def main() -> None:
bridge = BLEUARTBridge("AA:BB:CC:DD:EE:FF")
rx_buffer = bytearray()
pending: dict[str, asyncio.Future[object]] = {}
def handle_rx(data: bytearray) -> None:
rx_buffer.extend(data)
while b"\n" in rx_buffer:
index = rx_buffer.index(b"\n")
line = bytes(rx_buffer[:index])
del rx_buffer[: index + 1]
message = json.loads(line.decode())
request_id = message.get("id")
future = pending.get(request_id)
if future is None or future.done():
continue
if message.get("ok") is False:
future.set_exception(RuntimeError(str(message.get("error"))))
else:
future.set_result(message.get("data"))
bridge.add_rx_handler(handle_rx)
try:
if not await bridge.connect():
raise RuntimeError("failed to connect")
request_id = uuid4().hex
loop = asyncio.get_running_loop()
pending[request_id] = loop.create_future()
request = {"v": 1, "id": request_id, "op": "echo", "data": "hello"}
await bridge.send(json.dumps(request) + "\n", with_response=True)
response = await asyncio.wait_for(pending[request_id], timeout=10.0)
print(response)
finally:
await bridge.disconnect()
asyncio.run(main())
```
For production scripts, add validation around incoming JSON and clean up `pending` entries on timeout.
## Scan for devices from Python
Use `scan_devices()` if your script needs to discover devices first:
```python
import asyncio
from src.core.scanner import scan_devices
async def main() -> None:
devices = await scan_devices(timeout=5.0)
for device in devices:
print(device.device_id, device.name, device.rssi)
asyncio.run(main())
```
For custom service UUID discovery:
```python
devices = await scan_devices(
timeout=5.0,
service_uuid="00000000-0000-0000-0000-000000000001",
)
```
## Error handling guidance
The current Core API returns `False` for connection or send failures and logs details through `loguru`.
Recommended script pattern:
```python
if not await bridge.connect():
raise RuntimeError("failed to connect to BLE UART device")
if not await bridge.send("hello\n"):
raise RuntimeError("failed to send BLE UART data")
```
Always disconnect in `finally`:
```python
try:
...
finally:
await bridge.disconnect()
```
## Porting checklist
- [ ] Decide whether your use case needs Console, Daemon, or Core.
- [ ] Confirm the BLE service and characteristic UUIDs.
- [ ] Decide whether your payload is text, binary, JSON, or another framing format.
- [ ] Register RX handlers before calling `connect()`.
- [ ] Add a newline delimiter if your protocol is JSONL or line-oriented.
- [ ] Use `with_response=True` only when needed.
- [ ] Clean up pending request state on timeout.
- [ ] Call `disconnect()` in `finally`.
@@ -0,0 +1,86 @@
<!-- SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD -->
<!-- SPDX-License-Identifier: Apache-2.0 -->
# BLE UART Profile Compatibility
BLE UART Bridge works with BLE GATT profiles that provide a UART-like data path:
- one characteristic that the host writes to
- one characteristic that the device uses to notify data back to the host
The default profile is compatible with the Nordic UART Service (NUS), but NUS is not the only possible BLE UART-style profile.
## Default NUS-compatible profile
The built-in default profile uses these UUIDs:
| Role | UUID |
| --- | --- |
| Service | `6E400001-B5A3-F393-E0A9-E50E24DCCA9E` |
| RX, host to device | `6E400002-B5A3-F393-E0A9-E50E24DCCA9E` |
| TX, device to host | `6E400003-B5A3-F393-E0A9-E50E24DCCA9E` |
Use the default profile when the device advertises a NUS-compatible service.
## ESP-IDF BLE SPP examples
ESP-IDF includes BLE SPP examples that implement Espressif BLE UART-like vendor-specific GATT profiles:
- `examples/bluetooth/nimble/ble_spp/spp_server`
- `examples/bluetooth/nimble/ble_spp/spp_client`
- `examples/bluetooth/bluedroid/ble/ble_spp_server`
- `examples/bluetooth/bluedroid/ble/ble_spp_client`
BLE SPP over BLE is not a Bluetooth SIG standard profile. It is a vendor-specific GATT design that emulates a serial link, similar in purpose to NUS.
ESP-IDF BLE SPP examples may define more characteristics than BLE UART Bridge needs, such as data, command, and status characteristics. To use BLE UART Bridge with such a profile, map only the UART-like data path into `BLEUARTProfile`.
## Mapping an ESP-IDF BLE SPP profile
Map the profile fields as follows:
| `BLEUARTProfile` field | Map to |
| --- | --- |
| `service_uuid` | BLE SPP service UUID |
| `rx_char_uuid` | Characteristic that the host writes to, such as the SPP data receive characteristic |
| `tx_char_uuid` | Characteristic that the device notifies from, such as the SPP data notify characteristic |
Example:
```python
from src.core import BLEUARTBridge
from src.core import BLEUARTProfile
profile = BLEUARTProfile(
service_uuid="00000000-0000-0000-0000-00000000ABF0",
rx_char_uuid="00000000-0000-0000-0000-00000000ABF1",
tx_char_uuid="00000000-0000-0000-0000-00000000ABF2",
)
bridge = BLEUARTBridge("AA:BB:CC:DD:EE:FF", profile=profile)
```
Replace the UUIDs with the actual UUIDs used by the device firmware.
## What BLE UART Bridge does not map
BLE UART Bridge is intentionally focused on the data path. It does not automatically map extra control-plane characteristics that a profile may expose, such as:
- command characteristics
- status characteristics
- custom configuration characteristics
- profile-specific flow-control semantics
If an application needs those characteristics, implement that logic in a custom script on top of `bleak`, or extend BLE UART Bridge for that specific profile.
## Classic Bluetooth SPP is different
Classic Bluetooth SPP examples, such as `examples/bluetooth/bluedroid/classic_bt/bt_spp_*`, are not BLE GATT profiles.
They use Classic Bluetooth SPP rather than BLE GATT characteristics, so they are not compatible with BLE UART Bridge.
## Related docs
- [README.md](../README.md)
- [PORTING.md](PORTING.md)
@@ -0,0 +1,202 @@
<!-- SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD -->
<!-- SPDX-License-Identifier: Apache-2.0 -->
# Quick Start: BLE UART Console
This guide shows how to use the BLE UART Console for quick manual testing.
The Console is useful when you want to type data into a BLE UART device and inspect the bytes or text sent back by the device.
## Prerequisites
1. A host machine with Bluetooth access.
2. Python environment prepared. You can reuse the ESP-IDF Python environment, or use your own Python virtual environment. If you reuse the ESP-IDF environment, export it first and then install the BLE UART Bridge dependencies:
```bash
cd $IDF_PATH
. ./export.sh
cd tools/ble/ble_uart_bridge
python -m pip install -r requirements.txt
```
On Windows, run `export.bat` or `export.ps1` from the ESP-IDF root directory before installing `requirements.txt`. If you use your own Python virtual environment instead, activate it before installing `requirements.txt`.
3. A BLE device advertising the BLE UART service. By default the tool scans for Nordic UART Service UUIDs. For a known-compatible test target, build and flash the [BLE UART Service example](../../../../examples/bluetooth/ble_uart_service), which acts as an Echo Server by echoing RX writes back through TX notifications.
## Find a device
```bash
cd tools/ble/ble_uart_bridge
python main.py list-devices
```
Example output may include a device address and name:
```text
Found: AA:BB:CC:DD:EE:FF, with name esp-ble-uart, rssi=-42
```
Use the printed device identifier as `DEVICE_ID`. On macOS, this identifier is a CoreBluetooth UUID and is different from the device MAC address.
## Check the connection
```bash
python main.py connection-check AA:BB:CC:DD:EE:FF
```
This command connects to the device, discovers the BLE UART service and characteristics, then disconnects.
## Start the Console
```bash
python main.py console AA:BB:CC:DD:EE:FF
```
The console connects before opening the UI. If connection fails, the UI is not started.
Inside the UI:
- Type a line and press Enter to send it.
- Received data is shown with an `[RX]` prefix.
- Transmitted data is shown with a `[TX]` prefix.
- Connection information is shown with an `[INFO]` prefix.
- Press `Ctrl+C` or `Ctrl+D` to quit.
- Press `Ctrl+L` to clear the log.
## Text mode
Text mode is the default. It UTF-8 encodes input and appends a line terminator.
```bash
python main.py console AA:BB:CC:DD:EE:FF
```
By default, each submitted line is sent with `\n`.
### Choose a line terminator
Use `--terminator` for protocols that expect different line endings:
```bash
python main.py console AA:BB:CC:DD:EE:FF --terminator lf
python main.py console AA:BB:CC:DD:EE:FF --terminator crlf
python main.py console AA:BB:CC:DD:EE:FF --terminator none
```
Supported values:
| Value | Bytes appended |
| --- | --- |
| `lf` | `\n` |
| `crlf` | `\r\n` |
| `none` | nothing |
Use `crlf` for many AT-style command interpreters. Use `none` if the device expects the exact bytes you type.
## Hex mode
Hex mode sends raw bytes parsed from hexadecimal input and displays received bytes as hexadecimal.
```bash
python main.py console AA:BB:CC:DD:EE:FF --encoding hex
```
Inside the UI, enter bytes as hex:
```text
01 02 03 0a
```
The console sends:
```text
0x01 0x02 0x03 0x0a
```
Notes:
- Hex input is parsed with Python `bytes.fromhex()`.
- Spaces are allowed.
- In hex mode, `--terminator` is ignored because the input already represents exact bytes.
## Write-with-response
By default the console writes without response. Use `--with-response` if the target characteristic or debugging workflow should use BLE write-with-response:
```bash
python main.py console AA:BB:CC:DD:EE:FF --with-response
```
This affects BLE GATT write behavior only. It does not create an application-level request/response protocol. For application-level request/response, use Daemon mode instead.
## Common examples
### ESP-IDF BLE UART Echo Server
Use the [BLE UART Service example](../../../../examples/bluetooth/ble_uart_service) when you want a ready-made ESP-IDF Echo Server for testing BLE UART Bridge Console. After building, flashing, and pairing with the example, open Console and type any text; the example should echo the same data back as `[RX]` output.
```bash
# List nearby BLE UART devices and use the printed device ID as DEVICE_ID
python main.py list-devices
python main.py console AA:BB:CC:DD:EE:FF
```
### ESP-IDF console-style command
```bash
python main.py console AA:BB:CC:DD:EE:FF --terminator lf
```
Then type:
```text
help
```
### AT-style command
```bash
python main.py console AA:BB:CC:DD:EE:FF --terminator crlf
```
Then type:
```text
AT
```
### Binary smoke test
```bash
python main.py console AA:BB:CC:DD:EE:FF --encoding hex --with-response
```
Then type:
```text
aa 55 01 00
```
## Troubleshooting
### No devices found
- Confirm the host Bluetooth adapter is available.
- Confirm the device is advertising the BLE UART service UUID.
- Move the device closer to the host.
### Connection fails
- Make sure no other host is already connected to the BLE device.
- Restart advertising on the device.
- Run `connection-check` before opening the console.
### Text looks broken
- The console decodes RX bytes as UTF-8 in text mode.
- Use `--encoding hex` if the device sends binary data.
### Device does not react to input
- Check the required line ending. Try `--terminator crlf` or `--terminator none`.
- Check whether the device requires write-with-response. Try `--with-response`.
@@ -0,0 +1,419 @@
<!-- SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD -->
<!-- SPDX-License-Identifier: Apache-2.0 -->
# Quick Start: BLE UART Daemon
This guide shows how to use BLE UART Daemon mode and the lightweight JSONL RPC protocol used between the host and the BLE device.
Daemon mode is useful when another local process needs to communicate with a BLE UART device without owning the BLE connection itself.
## Prerequisites
1. A host machine with Bluetooth access.
2. Python environment prepared. You can reuse the ESP-IDF Python environment, or use your own Python virtual environment. If you reuse the ESP-IDF environment, export it first and then install the BLE UART Bridge dependencies:
```bash
cd $IDF_PATH
. ./export.sh
cd tools/ble/ble_uart_bridge
python -m pip install -r requirements.txt
```
On Windows, run `export.bat` or `export.ps1` from the ESP-IDF root directory before installing `requirements.txt`. If you use your own Python virtual environment instead, activate it before installing `requirements.txt`.
3. A BLE UART device that understands the daemon JSONL request/response protocol, or a device implementation you can adapt.
## Start the daemon
First, scan for a device:
```bash
cd tools/ble/ble_uart_bridge
python main.py list-devices
```
Then start the daemon:
```bash
python main.py daemon AA:BB:CC:DD:EE:FF
```
By default, the daemon listens on `127.0.0.1:8888`.
To choose another host or port:
```bash
python main.py daemon AA:BB:CC:DD:EE:FF --host 127.0.0.1 --port 8899
```
The daemon keeps one BLE connection open until it is stopped.
Security note: the daemon HTTP API does not implement authentication or authorization. Keep `--host` on `127.0.0.1` for local-only access unless you place the daemon behind your own access control.
## Check daemon status
In another terminal:
```bash
python main.py daemon-status
```
Example response:
```json
{
"device_id": "AA:BB:CC:DD:EE:FF",
"connection_state": "CONNECTED",
"is_connected": true,
"pending_requests": 0,
"single_flight": true,
"max_request_data_bytes": 4096,
"protocol": "esp-jsonl-rpc-lite-v1"
}
```
If the daemon uses a non-default address:
```bash
python main.py daemon-status --host 127.0.0.1 --port 8899
```
## Send a request from the CLI
Send a raw string payload with the default operation name `raw`:
```bash
python main.py daemon-send "hello"
```
Send a request with an explicit operation name:
```bash
python main.py daemon-send --op echo "hello"
```
Send a JSON payload:
```bash
python main.py daemon-send --op set_led --json '{"state": true}'
```
Set the request timeout:
```bash
python main.py daemon-send --op echo --timeout 5.0 "hello"
```
Use a non-default daemon address:
```bash
python main.py daemon-send --host 127.0.0.1 --port 8899 --op echo "hello"
```
Do not send requests to a daemon bound to a shared network interface unless that network path is trusted or protected by your own access control.
The CLI prints only the response payload. If the device returns a JSON object, the CLI prints it as JSON.
## Send a notification from the CLI
Use `daemon-notify` for fire-and-forget operations where the caller only needs the daemon to write to the BLE device and does not need a protocol response:
```bash
python main.py daemon-notify --op set_led --json '{"state": true}'
```
Send a raw string notification with the default operation name `raw`:
```bash
python main.py daemon-notify "hello"
```
Use a non-default daemon address:
```bash
python main.py daemon-notify --host 127.0.0.1 --port 8899 --op set_led --json '{"state": true}'
```
`daemon-notify` returns after the local BLE write completes. It does not wait for the device to send a JSONL response.
## HTTP API
Daemon mode exposes a local HTTP API.
### `GET /status`
Returns daemon and BLE connection state:
```bash
curl http://127.0.0.1:8888/status
```
Response fields:
| Field | Meaning |
| --- | --- |
| `device_id` | BLE device ID used by the daemon |
| `connection_state` | Bridge connection state |
| `is_connected` | Whether the BLE client is currently connected |
| `pending_requests` | Number of pending request futures |
| `single_flight` | Whether the daemon serializes requests |
| `max_request_data_bytes` | Maximum JSON-encoded `data` size accepted by `/request` and `/notify` |
| `protocol` | Wire protocol name and version |
### `POST /request`
Sends one request to the BLE device and waits for the response:
```bash
curl -X POST http://127.0.0.1:8888/request \
-H 'Content-Type: application/json' \
-d '{"op":"echo","data":"hello","timeout":10}'
```
Request body:
```json
{
"op": "echo",
"data": "hello",
"timeout": 10.0
}
```
Fields:
| Field | Required | Meaning |
| --- | --- | --- |
| `op` | No | Operation name. Defaults to `raw`. |
| `data` | Yes | Request payload. Can be a string, number, boolean, array, object, or null. |
| `timeout` | No | Response timeout in seconds. Defaults to `10.0`. |
Limits:
- `op` must be 1 to 64 characters.
- JSON-encoded `data` must not exceed 4096 bytes.
Successful response:
```json
{
"ok": true,
"data": "hello"
}
```
HTTP error behavior:
| HTTP status | Meaning |
| --- | --- |
| `413` | Request data exceeds the daemon payload limit |
| `500` | Failed to send data to the BLE device |
| `502` | Device returned a protocol error or invalid response |
| `504` | Timed out waiting for the device response |
### `POST /notify`
Sends one notification to the BLE device and returns without waiting for a protocol response:
```bash
curl -X POST http://127.0.0.1:8888/notify \
-H 'Content-Type: application/json' \
-d '{"op":"set_led","data":{"state":true}}'
```
Request body:
```json
{
"op": "set_led",
"data": {
"state": true
}
}
```
Fields:
| Field | Required | Meaning |
| --- | --- | --- |
| `op` | No | Operation name. Defaults to `raw`. |
| `data` | Yes | Notification payload. Can be a string, number, boolean, array, object, or null. |
Limits:
- `op` must be 1 to 64 characters.
- JSON-encoded `data` must not exceed 4096 bytes.
Successful response:
```json
{
"ok": true
}
```
HTTP error behavior:
| HTTP status | Meaning |
| --- | --- |
| `413` | Request data exceeds the daemon payload limit |
| `500` | Failed to send data to the BLE device |
## BLE JSONL RPC protocol
The daemon communicates with the BLE device using newline-delimited JSON. Every message is one JSON object followed by `\n`.
The protocol is named:
```text
esp-jsonl-rpc-lite-v1
```
It is intentionally small:
- Human-readable during debugging.
- Easy to generate and parse on ESP-IDF firmware with `cJSON`.
- No schema registry or capability negotiation.
- No built-in routing framework.
- One request at a time in the current daemon implementation.
### Host to device request
The daemon sends this JSONL message to the BLE device:
```json
{"v":1,"id":"6f8f...","op":"echo","data":"hello"}
```
Actual wire bytes include a final newline:
```text
{"v":1,"id":"6f8f...","op":"echo","data":"hello"}\n
```
Fields:
| Field | Meaning |
| --- | --- |
| `v` | Protocol version. Current value is `1`. |
| `id` | Request ID generated by the daemon. The device must echo this in `/request` responses. `/notify` uses an empty string because no response is expected. |
| `op` | Operation name selected by the client. |
| `data` | Request payload. |
### Device to host success response
```json
{"v":1,"id":"6f8f...","ok":true,"data":"hello"}
```
Fields:
| Field | Meaning |
| --- | --- |
| `v` | Protocol version. Recommended value is `1`. |
| `id` | The request ID from the host message. |
| `ok` | `true` for success. |
| `data` | Response payload. |
The daemon requires `data` to be present when `ok` is `true`.
### Device to host error response
```json
{"v":1,"id":"6f8f...","ok":false,"error":"unsupported op"}
```
Fields:
| Field | Meaning |
| --- | --- |
| `id` | The request ID from the host message. |
| `ok` | `false` for error. |
| `error` | Human-readable error message. |
The daemon requires `error` to be a non-empty string when `ok` is `false`.
### Response validation rules
For the preferred `ok/data/error` format, the daemon validates these rules:
- `id` must be a string and must match a pending request.
- If present, `v` must be `1`.
- `ok` must be a boolean.
- `ok: true` requires a `data` field.
- `ok: false` requires a non-empty string `error` field.
Messages without a matching pending `id` are treated as unsolicited messages and are logged only.
### Legacy response compatibility
The daemon also accepts older response shapes:
```json
{"id":"6f8f...","response":"hello"}
{"id":"6f8f...","error":"failed"}
```
New device firmware should prefer the `ok/data/error` format.
## Minimal firmware-side behavior
On the BLE device, implement this loop conceptually:
1. Accumulate bytes received on the BLE UART RX characteristic.
2. Split input on `\n`.
3. Parse each line as JSON.
4. Read `id`, `op`, and `data`.
5. Execute the requested operation.
6. If `id` is non-empty, send a JSON response with the same `id` and a final `\n`.
7. If `id` is empty, treat the message as fire-and-forget and normally do not send a response.
For example, an `echo` operation can return the same data:
```json
{"v":1,"id":"6f8f...","ok":true,"data":"hello"}
```
For notifications sent through `/notify`, the daemon uses an empty `id`:
```json
{"v":1,"id":"","op":"set_led","data":{"state":true}}
```
Firmware can execute the operation without responding. If it does respond with `id: ""`, the daemon will log the message as unsolicited because no pending request is waiting for that ID.
## Single-flight behavior
The daemon currently processes one `/request` at a time. This is exposed as:
```json
"single_flight": true
```
This keeps the firmware-side example simple because the device only needs to handle one active request at a time. The request `id` is still included so the protocol can be extended later if concurrent requests are needed.
## Disconnect behavior
The daemon does not automatically reconnect after the BLE link is disconnected. If the device disconnects, stop and restart the daemon after the device starts advertising again.
## Troubleshooting
### `daemon-send` times out
- Confirm the device sends a newline after the JSON response.
- Confirm the device response contains the same `id` as the request.
- Confirm the firmware handles the requested `op`.
- Increase `--timeout` if the operation is slow.
- Restart the daemon if the BLE link was disconnected.
### Daemon returns HTTP 502
- The device returned an error response, or the response was missing required fields.
- Check daemon logs for the exact error.
### Device receives data but daemon never resolves the request
- Check that the response is valid JSON.
- Check that the response is an object, not a JSON array or string.
- Check that the response is newline terminated.
- Check that the response `id` matches the request `id` exactly.
+80
View File
@@ -0,0 +1,80 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from src import run_connection_check
from src import run_console
from src import run_daemon
from src import run_daemon_notify
from src import run_daemon_send
from src import run_daemon_status
from src import run_list_devices
from src.console import ConsoleEncoding
from src.console import ConsoleTerminator
from typer import Option
from typer import Typer
# CLI app singleton
app = Typer()
# Core service
@app.command()
def list_devices() -> None:
run_list_devices()
@app.command()
def connection_check(device_id: str) -> None:
run_connection_check(device_id)
# Console
@app.command()
def console(
device_id: str,
terminator: ConsoleTerminator = Option(ConsoleTerminator.lf, help='Line terminator for text mode'),
encoding: ConsoleEncoding = Option(ConsoleEncoding.text, help='Console encoding'),
with_response: bool = Option(False, help='Use BLE write-with-response for TX data'),
) -> None:
run_console(device_id, terminator=terminator, encoding=encoding, with_response=with_response)
# Daemon
@app.command()
def daemon(device_id: str, host: str = '127.0.0.1', port: int = 8888) -> None:
run_daemon(device_id, host, port)
@app.command()
def daemon_status(host: str = '127.0.0.1', port: int = 8888) -> None:
run_daemon_status(host=host, port=port)
@app.command()
def daemon_send(
data: str,
op: str = Option('raw', help='Operation name in the JSONL request envelope'),
json_payload: bool = Option(False, '--json', help='Parse DATA as JSON instead of sending it as a string'),
timeout: float = 10.0,
host: str = '127.0.0.1',
port: int = 8888,
) -> None:
run_daemon_send(data=data, op=op, json_payload=json_payload, timeout=timeout, host=host, port=port)
@app.command()
def daemon_notify(
data: str,
op: str = Option('raw', help='Operation name in the JSONL request envelope'),
json_payload: bool = Option(False, '--json', help='Parse DATA as JSON instead of sending it as a string'),
host: str = '127.0.0.1',
port: int = 8888,
) -> None:
run_daemon_notify(data=data, op=op, json_payload=json_payload, host=host, port=port)
def main() -> None:
app()
if __name__ == '__main__':
main()
+19
View File
@@ -0,0 +1,19 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from .console import run_console
from .core import run_connection_check
from .core import run_list_devices
from .daemon import run_daemon
from .daemon import run_daemon_notify
from .daemon import run_daemon_send
from .daemon import run_daemon_status
__all__ = [
'run_connection_check',
'run_list_devices',
'run_daemon',
'run_daemon_send',
'run_daemon_notify',
'run_daemon_status',
'run_console',
]
@@ -0,0 +1,7 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from .api import run_console
from .console import ConsoleEncoding
from .console import ConsoleTerminator
__all__ = ['run_console', 'ConsoleEncoding', 'ConsoleTerminator']
@@ -0,0 +1,33 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import asyncio
from typing import Union
from .console import BLEUARTConsole
from .console import ConsoleEncoding
from .console import ConsoleTerminator
def run_console(
device_id: str,
terminator: Union[ConsoleTerminator, str] = ConsoleTerminator.lf,
encoding: Union[ConsoleEncoding, str] = ConsoleEncoding.text,
with_response: bool = False,
) -> None:
# Initialize BLE UART Console
console = BLEUARTConsole(
device_id,
terminator=terminator,
encoding=encoding,
with_response=with_response,
)
try:
asyncio.run(console.run_console())
except KeyboardInterrupt:
pass
@@ -0,0 +1,266 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import asyncio
import codecs
from enum import Enum
from typing import Union
from loguru import logger
from rich.highlighter import Highlighter
from rich.text import Text
from textual import on
from textual.app import App
from textual.app import ComposeResult
from textual.binding import Binding
from textual.binding import BindingType
from textual.message import Message
from textual.widgets import Footer
from textual.widgets import Header
from textual.widgets import Input
from textual.widgets import Log
from ..core import BLEUARTBridge
CONSOLE_TEXT_ENCODING = 'utf-8'
class ConsoleTerminator(str, Enum):
lf = 'lf'
crlf = 'crlf'
none = 'none'
class ConsoleEncoding(str, Enum):
text = 'text'
hex = 'hex'
TERMINATORS: dict[ConsoleTerminator, bytes] = {
ConsoleTerminator.lf: b'\n',
ConsoleTerminator.crlf: b'\r\n',
ConsoleTerminator.none: b'',
}
CONSOLE_CSS = """
Screen {
layout: vertical;
}
#log {
height: 1fr;
}
"""
CONSOLE_BINDINGS: list[BindingType] = [
Binding('ctrl+c', 'quit', 'Quit', priority=True),
Binding('ctrl+d', 'quit', show=False),
Binding('ctrl+l', 'clear_log', 'Clear'),
]
class BLEUARTLogHighlighter(Highlighter):
def highlight(self, text: Text) -> None:
if text.plain.startswith('[RX] '):
text.stylize('green', 0, 5)
elif text.plain.startswith('[TX] '):
text.stylize('yellow', 0, 5)
elif text.plain.startswith('[INFO] '):
text.stylize('cyan', 0, 7)
class BLEUARTLog(Log):
can_focus = False
FOCUS_ON_CLICK = False
def __init__(self) -> None:
super().__init__(id='log', highlight=True, auto_scroll=True)
self.highlighter = BLEUARTLogHighlighter()
class BLEUARTDataReceived(Message):
def __init__(self, data: bytes) -> None:
super().__init__()
self.data = data
class BLEUARTConsole(App):
CSS = CONSOLE_CSS
BINDINGS = CONSOLE_BINDINGS
ENABLE_COMMAND_PALETTE = False
def __init__(
self,
device_id: str,
terminator: Union[ConsoleTerminator, str] = ConsoleTerminator.lf,
encoding: Union[ConsoleEncoding, str] = ConsoleEncoding.text,
with_response: bool = False,
) -> None:
super().__init__()
self._device_id = device_id
self._bridge = BLEUARTBridge(device_id)
self._terminator = self._parse_terminator(terminator)
self._encoding = self._parse_encoding(encoding)
self._with_response = with_response
self._rx_decoder = codecs.getincrementaldecoder(CONSOLE_TEXT_ENCODING)(errors='replace')
self._rx_line_open = False
self._rx_pending: list[str] = []
self._ui_ready = False
@staticmethod
def _parse_terminator(terminator: Union[ConsoleTerminator, str]) -> bytes:
try:
return TERMINATORS[ConsoleTerminator(terminator)]
except ValueError:
choices = ', '.join(item.value for item in ConsoleTerminator)
raise ValueError(f'Unsupported terminator: {terminator}. Expected one of: {choices}') from None
@staticmethod
def _parse_encoding(encoding: Union[ConsoleEncoding, str]) -> ConsoleEncoding:
try:
return ConsoleEncoding(encoding)
except ValueError:
choices = ', '.join(item.value for item in ConsoleEncoding)
raise ValueError(f'Unsupported encoding: {encoding}. Expected one of: {choices}')
async def run_console(self) -> None:
# Connect before starting Textual so failed connections do not flash an empty UI.
self._bridge.add_rx_handler(self._on_ble_notif)
try:
# Should try connection to catch KeyInterrupt during connection establishment
if not await self._bridge.connect():
logger.error(f'Failed to open BLE UART Console for {self._device_id}')
return
# Run UI event loop
await self.run_async()
finally:
# Disconnect from device
logger.info(f'Closing BLE UART Console for {self._device_id}...')
await self._bridge.disconnect()
# Textual lifecycle hook: build the widget tree before the app is mounted.
def compose(self) -> ComposeResult:
yield Header()
yield BLEUARTLog()
yield Input(placeholder='Send BLE UART data...', id='input')
yield Footer()
# Device notification callback
def _on_ble_notif(self, data: bytearray) -> None:
self.post_message(BLEUARTDataReceived(bytes(data)))
@on(BLEUARTDataReceived)
def on_ble_uart_data_received(self, message: BLEUARTDataReceived) -> None:
if self._encoding == 'hex':
self._write_rx(f'{message.data.hex(" ")}\n')
return
text = self._rx_decoder.decode(message.data)
if not text:
return
self._write_rx(text)
# Textual lifecycle hook: widgets are ready, so BLE can be connected and UI updated.
async def on_mount(self) -> None:
self._ui_ready = True
self.title = f'BLE UART — {self._device_id}'
self.query_one('#input', Input).focus()
self._write_info(f'Connected to {self._device_id}')
self._drain_rx_pending()
self.run_worker(self._monitor_connection)
async def on_unmount(self) -> None:
self._ui_ready = False
# Textual event handler: called when the Input widget is submitted with Enter.
@on(Input.Submitted)
async def on_input_submitted(self, event: Input.Submitted) -> None:
line = event.value
event.input.clear()
if not line:
return
try:
if not self._bridge.is_connected:
self._write_tx(f'failed, not connected: {line}')
return
payload = self._encode_tx(line)
if await self._bridge.send(payload, with_response=self._with_response):
self._write_tx(line)
else:
self._write_tx(f'failed: {line}')
except ValueError as e:
self._write_tx(f'failed: {e}')
except Exception as e:
logger.exception(e)
self._write_tx(f'failed: {line}')
def _encode_tx(self, line: str) -> bytes:
if self._encoding == 'hex':
try:
return bytes.fromhex(line)
except ValueError as e:
raise ValueError('invalid hex input') from e
try:
payload = line.encode(CONSOLE_TEXT_ENCODING)
except UnicodeEncodeError as e:
raise ValueError(f'failed to encode input: {e}') from e
return payload + self._terminator
async def _monitor_connection(self) -> None:
while self._bridge.is_connected:
await asyncio.sleep(0.5)
# DOM may be uninstalled already
if self._ui_ready:
self._write_info('Disconnected. Press Ctrl+C to quit.')
def _write_rx(self, text: str) -> None:
if not self._ui_ready:
self._rx_pending.append(text)
return
log = self.query_one('#log', Log)
for chunk in text.splitlines(keepends=True):
if not self._rx_line_open:
log.write('[RX] ')
self._rx_line_open = True
log.write(chunk)
if chunk.endswith(('\r', '\n')):
self._rx_line_open = False
def _drain_rx_pending(self) -> None:
for text in self._rx_pending:
self._write_rx(text)
self._rx_pending.clear()
def _write_tx(self, text: str) -> None:
self._write_log('[TX] ', text)
def _write_info(self, text: str) -> None:
self._write_log('[INFO] ', text)
def _write_log(self, prefix: str, text: str) -> None:
log = self.query_one('#log', Log)
if self._rx_line_open:
log.write('\n')
self._rx_line_open = False
log.write(f'{prefix}{text}\n')
# Textual action: invoked by the ``clear_log`` key binding.
def action_clear_log(self) -> None:
self.query_one('#log', Log).clear()
self._rx_line_open = False
self._rx_pending.clear()
@@ -0,0 +1,8 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from .api import run_daemon
from .api import run_daemon_notify
from .api import run_daemon_send
from .api import run_daemon_status
__all__ = ['run_daemon', 'run_daemon_status', 'run_daemon_send', 'run_daemon_notify']
+120
View File
@@ -0,0 +1,120 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import json
from typing import Any
from typing import Optional
from urllib.error import HTTPError
from urllib.error import URLError
from urllib.request import Request
from urllib.request import urlopen
import uvicorn
from .server import app as daemon_app
def _daemon_url(host: str, port: int, path: str) -> str:
return f'http://{host}:{port}{path}'
def _request_json(
method: str,
url: str,
payload: Optional[dict[str, Any]] = None,
timeout: float = 10.0,
) -> dict[str, Any]:
data = json.dumps(payload).encode() if payload is not None else None
headers = {'Content-Type': 'application/json'} if payload is not None else {}
request = Request(url, data=data, headers=headers, method=method)
try:
with urlopen(request, timeout=timeout) as response:
body = response.read().decode()
except HTTPError as e:
detail = e.read().decode(errors='replace')
raise RuntimeError(f'Daemon request failed with HTTP {e.code}: {detail}') from e
except TimeoutError as e:
raise RuntimeError(f'Timed out waiting for BLE UART Daemon: {url}') from e
except URLError as e:
raise RuntimeError(f'Failed to connect to BLE UART Daemon: {e.reason}') from e
if not body:
return {}
try:
result = json.loads(body)
except json.JSONDecodeError as e:
raise RuntimeError(f'Invalid daemon response: {body!r}') from e
if not isinstance(result, dict):
raise RuntimeError(f'Invalid daemon response: {result!r}')
return result
def run_daemon(device_id: str, host: str, port: int) -> None:
daemon_app.state.device_id = device_id
uvicorn.run(daemon_app, host=host, port=port)
def run_daemon_status(host: str = '127.0.0.1', port: int = 8888) -> None:
try:
status = _request_json('GET', _daemon_url(host, port, '/status'))
except RuntimeError as e:
print(e)
raise SystemExit(1) from e
print(json.dumps(status, indent=2))
def run_daemon_send(
data: str,
op: str = 'raw',
json_payload: bool = False,
timeout: float = 10.0,
host: str = '127.0.0.1',
port: int = 8888,
) -> None:
try:
payload_data: Any = json.loads(data) if json_payload else data
except json.JSONDecodeError as e:
print(f'Invalid JSON payload: {e}')
raise SystemExit(1) from e
try:
response = _request_json(
'POST',
_daemon_url(host, port, '/request'),
payload={'op': op, 'data': payload_data, 'timeout': timeout},
timeout=timeout + 1.0,
)
except RuntimeError as e:
print(e)
raise SystemExit(1) from e
result = response.get('data', response.get('response', ''))
print(result if isinstance(result, str) else json.dumps(result))
def run_daemon_notify(
data: str,
op: str = 'raw',
json_payload: bool = False,
host: str = '127.0.0.1',
port: int = 8888,
) -> None:
try:
payload_data: Any = json.loads(data) if json_payload else data
except json.JSONDecodeError as e:
print(f'Invalid JSON payload: {e}')
raise SystemExit(1) from e
try:
_request_json(
'POST',
_daemon_url(host, port, '/notify'),
payload={'op': op, 'data': payload_data},
)
except RuntimeError as e:
print(e)
raise SystemExit(1) from e
@@ -0,0 +1,103 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import asyncio
import json
from json import JSONDecodeError
from typing import Any
from loguru import logger
from ..core.constants import DEFAULT_READ_BUFFER_LIMIT
PROTOCOL_VERSION = 1
def encode_jsonl_request(request_id: str, op: str, data: Any) -> str:
return json.dumps({'v': PROTOCOL_VERSION, 'id': request_id, 'op': op, 'data': data}) + '\n'
def drain_jsonl_messages(
buffer: bytearray, data: bytes, buffer_limit: int = DEFAULT_READ_BUFFER_LIMIT
) -> list[dict[str, Any]]:
buffer.extend(data)
messages: list[dict[str, Any]] = []
while True:
try:
newline_index = buffer.index(b'\n')
except ValueError:
break
line = bytes(buffer[:newline_index])
del buffer[: newline_index + 1]
if not line:
continue
try:
message = json.loads(line.decode())
except (JSONDecodeError, UnicodeDecodeError):
logger.warning(f'Invalid JSONL message from device, dropping data: {line!r}')
continue
if not isinstance(message, dict):
logger.warning(f'JSONL message is not an object, dropping data: {message!r}')
continue
messages.append(message)
if len(buffer) > buffer_limit:
logger.warning(f'JSONL receive buffer exceeded {buffer_limit} bytes, dropping buffered data')
buffer.clear()
return messages
def resolve_pending_response(pending_requests: dict[str, asyncio.Future[Any]], message: dict[str, Any]) -> bool:
request_id = message.get('id')
if not isinstance(request_id, str):
return False
future = pending_requests.get(request_id)
if future is None:
return False
pending_requests.pop(request_id)
if future.done():
return True
version = message.get('v')
if version is not None and version != PROTOCOL_VERSION:
future.set_exception(RuntimeError(f'Invalid protocol response: unsupported version {version!r}'))
return True
if 'ok' in message:
ok = message['ok']
if not isinstance(ok, bool):
future.set_exception(RuntimeError('Invalid protocol response: ok must be boolean'))
return True
if ok:
if 'data' not in message:
future.set_exception(RuntimeError('Invalid protocol response: missing data for successful response'))
return True
future.set_result(message['data'])
return True
error = message.get('error')
if not isinstance(error, str) or not error:
future.set_exception(RuntimeError('Invalid protocol response: error must be non-empty string'))
return True
future.set_exception(RuntimeError(error))
return True
if 'error' in message:
future.set_exception(RuntimeError(str(message['error'])))
return True
if 'response' in message:
future.set_result(message['response'])
return True
future.set_exception(RuntimeError('Invalid protocol response: missing ok/data, response, or error'))
return True
@@ -0,0 +1,21 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from typing import Any
from pydantic import BaseModel
from pydantic import Field
MAX_OP_LENGTH = 64
MAX_REQUEST_DATA_BYTES = 4096
class BLEUARTRequestPayload(BaseModel):
op: str = Field('raw', min_length=1, max_length=MAX_OP_LENGTH, description='Operation name to send to BLE device')
data: Any = Field(..., description='Request payload to send to BLE device')
timeout: float = Field(10.0, gt=0, description='Response timeout in seconds')
class BLEUARTNotifyPayload(BaseModel):
op: str = Field('raw', min_length=1, max_length=MAX_OP_LENGTH, description='Operation name to send to BLE device')
data: Any = Field(..., description='Notification payload to send to BLE device')
@@ -0,0 +1,132 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
# Keep annotations compatible with Python 3.9.
# ruff: noqa: UP007
import asyncio
import json
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Optional
from uuid import uuid4
from fastapi import FastAPI
from fastapi import HTTPException
from loguru import logger
from ..core import BLEUARTBridge
from .jsonl import PROTOCOL_VERSION
from .jsonl import drain_jsonl_messages
from .jsonl import encode_jsonl_request
from .jsonl import resolve_pending_response
from .models import MAX_REQUEST_DATA_BYTES
from .models import BLEUARTNotifyPayload
from .models import BLEUARTRequestPayload
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# Server initialization
app.state.bridge = BLEUARTBridge(app.state.device_id)
app.state.request_lock = asyncio.Lock()
# Set BLE UART Bridge RX callback
loop = asyncio.get_running_loop()
app.state.rx_buffer = bytearray()
app.state.pending_requests = {}
def _handle_rx_data(data: bytes) -> None:
for message in drain_jsonl_messages(app.state.rx_buffer, data):
if resolve_pending_response(app.state.pending_requests, message):
continue
logger.debug(f'Received unsolicited BLE UART message: {message!r}')
def _rx_handler(data: bytearray) -> None:
try:
loop.call_soon_threadsafe(_handle_rx_data, bytes(data))
except RuntimeError:
logger.warning(f'Event loop is unavailable, dropping data: {data.decode(errors="replace")}')
app.state.bridge.add_rx_handler(_rx_handler)
# Try to connect to the device
if not await app.state.bridge.connect():
logger.error('Failed to start BLE UART Daemon!')
raise RuntimeError('Failed to start BLE UART Daemon!')
yield
# Disconnect from the device
await app.state.bridge.disconnect()
app = FastAPI(title='BLE UART Daemon', lifespan=lifespan)
def _request_data_size(data: object) -> int:
return len(json.dumps(data).encode())
@app.get('/status')
async def status() -> dict:
bridge: Optional[BLEUARTBridge] = getattr(app.state, 'bridge', None)
pending_requests: Optional[dict] = getattr(app.state, 'pending_requests', None)
return {
'device_id': getattr(app.state, 'device_id', None),
'connection_state': bridge.connection_state.value if bridge else 'DISCONNECTED',
'is_connected': bridge.is_connected if bridge else False,
'pending_requests': len(pending_requests or {}),
'single_flight': True,
'max_request_data_bytes': MAX_REQUEST_DATA_BYTES,
'protocol': f'esp-jsonl-rpc-lite-v{PROTOCOL_VERSION}',
}
@app.post('/request')
async def request(payload: BLEUARTRequestPayload) -> dict:
if _request_data_size(payload.data) > MAX_REQUEST_DATA_BYTES:
raise HTTPException(status_code=413, detail=f'Request data exceeds {MAX_REQUEST_DATA_BYTES} bytes')
# Request with coroutine lock
async with app.state.request_lock:
request_id = uuid4().hex
response_future = asyncio.get_running_loop().create_future()
app.state.pending_requests[request_id] = response_future
try:
# Send command to BLE device
success = await app.state.bridge.send(
encode_jsonl_request(request_id, payload.op, payload.data),
with_response=True,
)
if not success:
raise HTTPException(status_code=500, detail='Failed to send data to device')
# Wait for BLE device to respond
try:
response = await asyncio.wait_for(response_future, timeout=payload.timeout)
except asyncio.TimeoutError:
raise HTTPException(status_code=504, detail='Response timeout')
except RuntimeError as e:
raise HTTPException(status_code=502, detail=str(e))
finally:
app.state.pending_requests.pop(request_id, None)
return {'ok': True, 'data': response}
@app.post('/notify')
async def notify(payload: BLEUARTNotifyPayload) -> dict:
if _request_data_size(payload.data) > MAX_REQUEST_DATA_BYTES:
raise HTTPException(status_code=413, detail=f'Request data exceeds {MAX_REQUEST_DATA_BYTES} bytes')
success = await app.state.bridge.send(
encode_jsonl_request('', payload.op, payload.data),
with_response=False,
)
if not success:
raise HTTPException(status_code=500, detail='Failed to send data to device')
return {'ok': True}