Learn how to establish a bidirectional Bluetooth Low Energy (BLE) communication between a Raspberry Pi and a Raspberry Pi Pico. The Raspberry Pi Pico will be set as a BLE peripheral with two characteristics. The Raspberry Pi will read and write to those characteristics, establishing a bidirectional communication.
For the Raspberry Pi Python program, we’ll use the bleak library, and for the Pico MicroPython program, we’ll use the aioble library.
This tutorial was based on the code and works by Edgardo Peregrino.
In a previous tutorial, we learned how to communicate between the Raspberry Pi and the Pico W with BLE. Refer to this article to learn more about it.
We also learned about BLE, the differences between BLE and Bluetooth Classic, and how it works in this case. In this tutorial, you will learn how to do bidirectional BLE communication between the Pi and the Pico W.
If BLE is a new subject for you, we recommend reading that tutorial first, as well as this BLE getting started guide for the Pico:
Prerequisites
Before we begin, make sure you have the setup as in this previous tutorial. Here’s the setup briefly explained:
- First, make sure you have a Raspberry Pi that supports Bluetooth, like an RPi 4 or RPi 5.
- You need a Pico W or Pico 2W.
- Set up a virtual environment on the Pi, install the Bleak library in the virtual environment, and pipgio as explained in the following section of the previous tutorial “Preparing the Raspberry Pi“.
- Install MicroPython firmware on the Pico W or Pico 2W. Make sure you use the correct and updated version of MicroPython.
Parts Required
These are the parts required for this tutorial:
- Raspberry Pi (we’re using a Raspberry Pi 5)
- Raspberry Pi Pico W or Raspberry Pi Pico 2 W
- 1x LED
- 1x 220 Ohm resistor (or similar value)
- Breadboard
- Jumper wires
You can use the preceding links or go directly to MakerAdvisor.com/tools to find all the parts for your projects at the best price!
Wiring Diagram
The wiring diagram is the same as the previous tutorial. In the case of the Raspberry Pi, we need an LED connected to GPIO 17. You can follow the next schematic diagram.
The LED connected to GPIO 17 will give us visual feedback to test if the code is working properly.
The Pico will be mounted on the breadboard just for practical reasons, and that will be it for the wiring. If you prefer, you don’t need to mount the Pico on the breadboard.
Bidirectional Communication: How Does It Work?
So, how does BLE bidirectional communication work in our example?
The Raspberry Pi Pico sets up a GATT service with two characteristics.
- rx_characteristic – characteristic where the Pi writes data. The Pico reads this characteristic’s value (receives data from the Pi).
- tx_characteristic – characteristic where the Pico writes data. The Pico can read data from that characteristic. So, the Pico sends data to the Pi.
Bidirectional Communication Code Example
Now that you understand how the bidirectional communication works in BLE, let’s set up a basic example.
Raspberry Pi Pico W – MicroPython Code
Let’s first start with the MicroPython code for the Raspberry Pi Pico. Copy the following code to Thonny IDE.
You need to install the aioble package.
1) Connect the board to your computer and connect it to Thonny IDE.
2) On Thonny IDE, go to Tools > Manage Packages…
3) Search for aioble and click on the aioble option.
4) Finally, click the Install button.
5) Wait a few seconds while it installs.
# Rui Santos & Sara Santos - Random Nerd Tutorials
# Complete project details at https://RandomNerdTutorials.com/bidirecional-ble-communication-raspberry-pi-and-pico/
import asyncio
import aioble
import bluetooth
# Bluetooth configuration
_SERVICE_UUID = bluetooth.UUID(0x1848)
# Characteristic where the Raspberry Pi writes data TO the Pico (Inbox)
_PI_TO_PICO_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6E)
# Characteristic where the Pico writes data TO the Raspberry Pi (Outbox)
_PICO_TO_PI_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6F)
BLE_NAME = "Pico W Peripheral"
# Register GATT server
ble_service = aioble.Service(_SERVICE_UUID)
# Inbox: Raspberry Pi writes data to the Pico
rx_characteristic = aioble.Characteristic(
ble_service,
_PI_TO_PICO_CHARACTERISTIC_UUID,
read=True,
write=True,
capture=True
)
# Outbox: Pico writes data to the Raspberry Pi
tx_characteristic = aioble.Characteristic(
ble_service,
_PICO_TO_PI_CHARACTERISTIC_UUID,
read=True,
write=True,
notify=True,
capture=True
)
aioble.register_services(ble_service)
# Helper functions
def _encode_message(message):
"""Encode a message to bytes."""
return message.encode('utf-8')
def _decode_message(data):
"""Decode a message from bytes."""
return data.decode('utf-8')
# Task to handle sending data
async def send_task():
message_count = 1
while True:
message = f" From {BLE_NAME}!: {message_count}"
message_count += 1
try:
tx_characteristic.write(_encode_message(message), send_update=True)
print(f"Sent: {message}")
except Exception as e:
print(f"Error while sending data: {e}")
await asyncio.sleep_ms(1000)
# Task to handle receiving data
async def receive_task():
last_data = None
while True:
try:
data = rx_characteristic.read()
if data and data != last_data:
decoded = _decode_message(data)
print(f"Received: {decoded}")
last_data = data
await asyncio.sleep_ms(500)
except Exception as e:
print(f"Error receiving data: {e}")
await asyncio.sleep_ms(500)
# Serially wait for connections
async def peripheral_task():
# Show MAC address once at start
ble = bluetooth.BLE()
_, mac_address = ble.config('mac')
formatted_mac = ':'.join('{:02X}'.format(b) for b in mac_address)
print(f"Bluetooth MAC Address: {formatted_mac}")
while True:
try:
async with await aioble.advertise(
2000, name=BLE_NAME, services=[_SERVICE_UUID], appearance=768
) as connection:
print("Connection from", connection.device)
await connection.disconnected()
except Exception as e:
print("Error in peripheral_task:", e)
finally:
print(f"{BLE_NAME} disconnected")
await asyncio.sleep_ms(100)
# Run all tasks
async def main():
t1 = asyncio.create_task(send_task())
t2 = asyncio.create_task(receive_task())
t3 = asyncio.create_task(peripheral_task())
await asyncio.gather(t1, t2, t3)
# Run the program
asyncio.run(main())
We use the aioble library to set up the Raspberry Pi Pico as a BLE peripheral.
For a getting started guide on how to use the aioble library with the Pico, read this guide: Raspberry Pi Pico W: Bluetooth Low Energy (BLE) with MicroPython.
How Does the Code Work?
Let’s take a quick look at how the code works, or skip to the next section.
First, we import the libraries.
import asyncio
import aioble
import bluetoothDefine UUIDs for the GATT service we’ll set up, for the characteristic where we’ll write data, and for the characteristic where we’ll read the data.
# Bluetooth configuration
_SERVICE_UUID = bluetooth.UUID(0x1848)
# Characteristic where the Raspberry Pi writes data TO the Pico (Inbox)
_PI_TO_PICO_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6E)
# Characteristic where the Pico writes data TO the Raspberry Pi (Outbox)
_PICO_TO_PI_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6F)Set up the BLE name for the device. We’re setting it to Pico W Peripheral.
BLE_NAME = "Pico W Peripheral"Set the service for the GATT server using the specified UUID. This service serves as a “container” for the characteristics.
ble_service = aioble.Service(_SERVICE_UUID)The following section creates a characteristic that the Raspberry Pi can write to (write=True). The Pico can read the data written by the Pi (read=True), and capture=True makes the Pico automatically store the latest value so it can be easily read later.
# Inbox: Raspberry Pi writes data to the Pico
rx_characteristic = aioble.Characteristic(
ble_service,
_PI_TO_PICO_UUID,
read=True,
write=True,
capture=True
)The next snippet creates a characteristic that the Pico uses to send data to the Raspberry Pi. The Pico can write messages here, and notify=True allows it to automatically notify the Pi whenever new data is written.
# Outbox: Pico writes data to the Raspberry Pi
tx_characteristic = aioble.Characteristic(
ble_service,
_PICO_TO_PI_UUID,
read=True,
write=True,
notify=True,
capture=True
)This line registers the service and its characteristics, making them visible and accessible to other devices like the Raspberry Pi.
aioble.register_services(ble_service)The _encode_message() and _decode_message() functions encode or decode the message as UTF-8.
# Helper functions
def _encode_message(message):
"""Encode a message to bytes."""
return message.encode('utf-8')
def _decode_message(data):
"""Decode a message from bytes."""
return data.decode('utf-8')Sending Data
We create an asynchronous task to send data to the Pi (writing to the tx_characteristic). We’re sending a message followed by a counter every second.
# Task to handle sending data
async def send_task():
message_count = 1
while True:
message = f" From {BLE_NAME}! Count: {message_count}"
message_count += 1
try:
tx_characteristic.write(_encode_message(message), send_update=True)
print(f"Sent: {message}")
except Exception as e:
print(f"Error while sending data: {e}")
await asyncio.sleep_ms(1000)This is the line of code that writes to the characteristic.
tx_characteristic.write(_encode_message(message), send_update=True)To learn more about asynchronous programming with the Raspberry Pi Pico, check out this guide: Raspberry Pi Pico Asynchronous Programming – Run Multiple Tasks (MicroPython).
Receiving Data
We create another task to receive data (reading the value of the rx_characteristic). We read from the characteristic every 500 milliseconds and only print to the Shell if there is a new characteristic value.
# Task to handle receiving data
async def receive_task():
last_data = None
while True:
try:
data = rx_characteristic.read()
if data and data != last_data:
decoded = _decode_message(data)
print(f"Received: {decoded}")
last_data = data
await asyncio.sleep_ms(500)
except Exception as e:
print(f"Error receiving data: {e}")
await asyncio.sleep_ms(500)This is the line that reads the characteristic value and saves it to the data variable.
data = rx_characteristic.read()Set the Pico as a BLE Peripheral
The peripheral_task() sets the Raspberry Pi Pico as a BLE peripheral. It first prints the Bluetooth MAC address, and then starts advertising the Pico as a peripheral with the service UUID we defined earlier.
# Serially wait for connections
async def peripheral_task():
# Show MAC address once at start
ble = bluetooth.BLE()
_, mac_address = ble.config('mac')
formatted_mac = ':'.join('{:02X}'.format(b) for b in mac_address)
print(f"Bluetooth MAC Address: {formatted_mac}")
while True:
try:
async with await aioble.advertise(
2000, name=BLE_NAME, services=[_SERVICE_UUID], appearance=768
) as connection:
print("Connection from", connection.device)
await connection.disconnected()
except Exception as e:
print("Error in peripheral_task:", e)
finally:
print(f"{BLE_NAME} disconnected")
await asyncio.sleep_ms(100)Creating and Running the Tasks
Finally, we create an asynchronous main() function, where we’ll write the base for our code. We create three asynchronous tasks: one for writing to the characteristic (t1 – sendtask()), another to read data from the characteristic (t2 – receive_task()), and finally another one to set the Pico as a BLE peripheral (peripheral_task()).
# Run all tasks
async def main():
t1 = asyncio.create_task(send_task())
t2 = asyncio.create_task(receive_task())
t3 = asyncio.create_task(peripheral_task())Then, we gather and run the tasks asynchronously by calling the main() function.
await asyncio.gather(t1, t2, t3)
# Run the program
asyncio.run(main())Running the Code on the Raspberry Pi Pico
Run the code on the Pico by pressing the green play button in Thonny, or you can press F5 on the keyboard.
Once it’s done, it’ll display the Bluetooth MAC address of the Pico. Save it because you’ll need it for the Raspberry Pi Python code.
It will also display the current message being written to the tx_characteristic (the one that the RPi will read).
Raspberry Pi – Python Code
On the Raspberry Pi, create a new file with the code below. You can call it pi_send_receive.py. The file should be located inside the same folder where you created the virtual environment.
There are many ways to create and run files on the Pi. I like to use Remote-SSH on VS Code. This extension allows you to establish an SSH connection with your Pi, create files, write code, and execute it directly on your Raspberry Pi board from your computer using the VS Code interface. Learn more here: Programming Raspberry Pi Remotely using VS Code (Remote-SSH).
# Rui Santos & Sara Santos - Random Nerd Tutorials
# Complete project details at https://RandomNerdTutorials.com/bidirecional-ble-communication-raspberry-pi-and-pico/
import asyncio
from bleak import BleakClient, uuids
from gpiozero import LED
connected = False
led = LED(17)
# REPLACE WITH YOUR PICO MAC ADDRESS
pico_address = "FF:FF:FF:FF:FF:FF"
SERVICE_UUID = uuids.normalize_uuid_16(0x1848)
WRITE_CHARACTERISTIC_UUID = uuids.normalize_uuid_16(0x2A6E)
READ_CHARACTERISTIC_UUID = uuids.normalize_uuid_16(0x2A6F)
message_count = 0
async def send_data_task(client):
global message_count
while True:
message_count += 1
message = f"From central: {message_count}".encode("utf-8")
try:
if not client.is_connected:
break
await client.write_gatt_char(WRITE_CHARACTERISTIC_UUID, message)
print(f"Central sent: {message.decode('utf-8')}")
except Exception as e:
print(f"Send error: {e}")
break
await asyncio.sleep(5)
async def receive_data_task(client):
last_message = None
while True:
try:
if not client.is_connected:
break
response = await client.read_gatt_char(READ_CHARACTERISTIC_UUID)
if response:
message = response.decode("utf-8")
if message != last_message:
print(f"Central received: {message}")
last_message = message
except Exception as e:
print(f"Receive error: {e}")
break
await asyncio.sleep(2)
async def blink_task():
global connected
print("blink task started")
while True:
led.toggle()
blink = 1000 if connected else 250
await asyncio.sleep(blink / 1000)
async def connect_and_communicate(address):
global connected
print(f"Connecting to {address}...")
blink_task_obj = asyncio.create_task(blink_task())
while True:
client = None
try:
client = BleakClient(address)
await client.connect()
connected = True
print("Connected to Pico")
tasks = [
asyncio.create_task(send_data_task(client)),
asyncio.create_task(receive_data_task(client))
]
done, pending = await asyncio.wait(tasks, timeout=30, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
t.cancel()
except asyncio.CancelledError:
raise
except Exception as e:
print(f"Connection failed or lost: {e}")
finally:
connected = False
if client:
try:
if client.is_connected:
await asyncio.wait_for(client.disconnect(), timeout=2.0)
except:
pass
print("Disconnected - trying to reconnect in 3 seconds...")
await asyncio.sleep(3)
async def main():
try:
await connect_and_communicate(pico_address)
except asyncio.CancelledError:
print("\nCancelling tasks...")
except KeyboardInterrupt:
print("\nProgram stopped by user")
except Exception as e:
print(f"Unexpected error: {e}")
finally:
connected = False
led.off()
print("Cleanup completed.")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
led.off()
Before running the code, you need to modify the following line with the Bluetooth MAC address of the Pico. You should have gotten it in the previous example.
# Replace with the MAC address of your Pico
pico_address = "2C:CF:67:B6:D7:4C"How Does the Code Work?
The Raspberry Pi connects to the Pico BLE client. It writes to the characteristic that the Pico will read and reads from the characteristic that the Pico writes to.
Let’s now take a closer look at the code to see how it works.
First, import the required libraries. We’re including bleak for the Bluetooth functions.
import asyncio
from bleak import BleakClient, uuids
from gpiozero import LEDWe have a boolean variable to keep track of whether we’re connected to another Bluetooth peripheral.
connected = FalseWe set GPIO 17 as an LED object that we’ll control later on to give us a visual feedback of whether the board is connected or not.
led = LED(17)Modify the following line with the Bluetooth MAC address of your Raspberry Pi Pico board.
# Replace with your Pico W's MAC address
pico_address = "2C:CF:67:B6:D7:4C" # Update this with the MAC address of the peripheralWe define the service and characteristics’ UUIDs. These are the service and characteristics of the Pico that we’ll search for to read the message sent from the Pico and to write a message to the Pico. So, these must be the same in the Pico.
# UUIDs
SERVICE_UUID = uuids.normalize_uuid_16(0x1848)
WRITE_CHARACTERISTIC_UUID = uuids.normalize_uuid_16(0x2A6E) #_PI_TO_PICO_CHARACTERISTIC_UUID on the Pico
READ_CHARACTERISTIC_UUID = uuids.normalize_uuid_16(0x2A6F) # _PICO_TO_PI_CHARACTERISTIC_UUIDWe’ll write a message with a counter to the Pico characteristic. So, we create a variable to hold the counter.
message_count = 0Writing on the Characteristic (sending data)
We create an asynchronous function called send_data_task() that writes on the Pico rx_characteristic every five seconds.
async def send_data_task(client):
global message_count
while True:
message_count += 1
message = f"From central: {message_count}".encode("utf-8")
try:
if not client.is_connected:
break
await client.write_gatt_char(WRITE_CHARACTERISTIC_UUID, message)
print(f"Central sent: {message.decode('utf-8')}")
except Exception as e:
print(f"Send error: {e}")
break
await asyncio.sleep(5)The line of code that actually writes on the characteristic is the following.
await client.write_gatt_char(WRITE_CHARACTERISTIC_UUID, message)It uses the write_gatt_char() function on the client object, and we pass as arguments the UUID of the characteristic and the data we want to write.
Reading the Characteristic (receiving data)
Then, we have another asynchronous function called receive_data_task() that waits to read the characteristic value of the peripheral device. It reads the characteristic every two seconds.
async def receive_data_task(client):
last_message = None
while True:
try:
if not client.is_connected:
break
response = await client.read_gatt_char(READ_CHARACTERISTIC_UUID)
if response:
message = response.decode("utf-8")
if message != last_message:
print(f"Central received: {message}")
last_message = message
except Exception as e:
print(f"Receive error: {e}")
break
await asyncio.sleep(2)The line of code that actually reads the characteristic is the following.
response = await client.read_gatt_char(READ_CHARACTERISTIC_UUID)It uses the read_gatt_char() function on the client object, and we pass as arguments the UUID of the characteristic we want to read.
Blink Task
We have another task called blink_task() that will run at the same time. This task will blink the LED connected to GPIO 17 on the Raspberry Pi. It will run every second if we are connected to the peripheral device. Otherwise, it will blink every 250ms.
async def blink_task():
global connected
print("blink task started")
while True:
led.toggle()
blink = 1000 if connected else 250
await asyncio.sleep(blink / 1000)Connect to BLE Device
Finally, the connect_and_communicate() function tries to connect to the peripheral device with the MAC address we defined earlier.
async def connect_and_communicate(address):
global connected
print(f"Connecting to {address}...")
blink_task_obj = asyncio.create_task(blink_task())
while True:
client = None
try:
client = BleakClient(address)
await client.connect()
connected = True
print("Connected to Pico")
tasks = [
asyncio.create_task(send_data_task(client)),
asyncio.create_task(receive_data_task(client))
]
done, pending = await asyncio.wait(tasks, timeout=30, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
t.cancel()
except asyncio.CancelledError:
raise
except Exception as e:
print(f"Connection failed or lost: {e}")
finally:
connected = False
if client:
try:
if client.is_connected:
await asyncio.wait_for(client.disconnect(), timeout=2.0)
except:
pass
print("Disconnected - trying to reconnect in 3 seconds...")
await asyncio.sleep(3)We first create the blink_task() so that it immediately starts giving us visual feedback. It is running in the background while the rest of the program executes.
blink_task_obj = asyncio.create_task(blink_task())Then, we create an infinite loop, where we try to create a BLE client with the Pico MAC address and connect to it. Once we’re connected, we set the connected variable to True.
while True:
client = None
try:
client = BleakClient(address)
await client.connect()
connected = True
print("Connected to Pico")We start the asynchronous tasks to send and receive data.
tasks = [
asyncio.create_task(send_data_task(client)),
asyncio.create_task(receive_data_task(client))
]The program waits for the two tasks (send_data_task() and receive_data_task()) to run. It continues waiting until either one of the tasks finishes or 30 seconds have passed.
done, pending = await asyncio.wait(tasks, timeout=30, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
t.cancel()When asyncio.wait() returns, the completed task is placed in done, while any task still running is placed in pending so it can be cancelled before the program attempts to reconnect.
Basically, the program waits while the send and receive tasks communicate with the Pico. If one task finishes or 30 seconds pass, it stops the remaining task and prepares to reconnect.
If an error occurs, the program handles it, cleans up the connection, waits briefly, and then tries again.
except asyncio.CancelledError:
raise
except Exception as e:
print(f"Connection failed or lost: {e}")
finally:
connected = False
if client:
try:
if client.is_connected:
await asyncio.wait_for(client.disconnect(), timeout=2.0)
except:
pass
print("Disconnected - trying to reconnect in 3 seconds...")
await asyncio.sleep(3)Running the Tasks
Create a main() function to call the connect_and_communicate() function that will run all tasks.
async def main():
try:
await connect_and_communicate(pico_address)
except asyncio.CancelledError:
print("\nCancelling tasks...")
except KeyboardInterrupt:
print("\nProgram stopped by user")
except Exception as e:
print(f"Unexpected error: {e}")
finally:
connected = False
led.off()
print("Cleanup completed.")Finally, run the program:
# Run the program
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
led.off()Running the Python Code
After creating the pi_send_receive.py file with the code we’ve shared previously and inserting your Pico MAC address, you can run it on your Raspberry Pi board. We’re using the Remote-SSH extension on VS Code, but you can create a new file using the nano command.
After saving the file, you can run it. But first, make sure you’re in the virtual environment we created previously. If you’ve exited the virtual environment, you can activate it again like so:
source blue/bin/activate
Finally, you can run your Python program like this:
python3 pi_led_receive.pyDemonstration
The code will start running, and after a few seconds, it will connect to the Raspberry Pi Pico (make sure the RPi Pico is running the code in the Thonny IDE).
The Raspberry Pi will start “sending” and “receiving” data to the Pico.
At the same time, the LED connected to GPIO 17 will start blinking every second.
In the Thonny IDE window, you can see that the Pico is “receiving” and “sending” data to and from the Pi.
Now, you successfully achieved bidirectional communication between the Raspberry Pi and the Pico boards.
Wrapping Up
In this tutorial, you learned how to establish a bidirectional BLE communication between a Raspberry Pi Pico and a Raspberry Pi. In our example, the Pico sets up a GATT service with two characteristics. By writing and reading on those two characteristics, the two boards can exchange data with each other.
We’re simply sending a counter message, but it can be easily modified to send sensor readings, data about the GPIOs’ state, and more.
We hope you’ve found this guide useful. Other BLE related guides that you may like:
- BLE Communication between Raspberry Pi and Raspberry Pi Pico W
- Raspberry Pi Pico W: Bluetooth Low Energy (BLE) with MicroPython – Getting Started Guide
- MicroPython: ESP32 – Getting Started with Bluetooth Low Energy (BLE)
- Getting Started with ESP32 Bluetooth Low Energy (BLE) on Arduino IDE
We hope you’ve found this tutorial useful.
Special thanks to our reader Edgardo Peregrino, who created the base for this tutorial as well as the example codes. Edgardo Peregrino is the author of “Programming Raspberry Pi in 30 Days” and “Cloud Powered Robotics with Raspberry Pi” and writer of several articles, especially in Servo Magazine. You can check his work here on its GitHub page or YouTube channel (LinuxRobotGeek).
Learn more about the Raspberry Pi Pico and the Raspberry Pi:
- All our Raspberry Pi Tutorials
- Learn Raspberry Pi Pico with MicroPython eBook
- All our Raspberry Pi Pico Tutorials
Thanks for reading.
