espressif--esp-idf/tools/ble/ble_uart_bridge/docs/PORTING.md

Porting ESP-BLE-UART Bridge to Custom Scripts

This guide explains how to reuse ESP-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 ESP-BLE-UART Bridge:

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:

PYTHONPATH=tools/ble/ble_uart_bridge python my_script.py

Basic script

The simplest script connects, sends one line, and disconnects:

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:

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:

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.

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:

await bridge.send(b"\x01\x02", with_response=True)

Use a custom BLE UART profile

The default profile uses the de-facto BLE UART-over-GATT UUIDs. For custom firmware, create a BLEUARTProfile:

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:

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:

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

scan_devices() performs an unfiltered BLE scan. If your script needs to restrict results to a specific profile, filter or connect-verify devices in your application code.

Error handling guidance

The current Core API returns False for connection or send failures and logs details through loguru.

Recommended script pattern:

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:

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.