Comunicación BLE bidireccional entre Raspberry Pi y Pico W

¿Alguna vez quisiste que dos placas conversaran entre sí sin cables, sin router y gastando muy poca energía? Ese es justamente el terreno donde brilla Bluetooth Low Energy (BLE). En este proyecto vas a conectar una Raspberry Pi 5 con una Raspberry Pi Pico W para que se envíen mensajes en las dos direcciones al mismo tiempo: la Pi le escribe a la Pico y la Pico le responde a la Pi, todo por BLE.

Qué vas a aprender

Al terminar este tutorial vas a tener una comunicación bidireccional funcionando entre las dos placas y, más importante, vas a entender por qué funciona: cómo se arma un servicio GATT con dos características, cuál sirve para recibir y cuál para enviar, y cómo correr varias tareas en paralelo con programación asíncrona (asyncio) tanto en MicroPython como en Python. Con esa base vas a poder reemplazar el simple contador de ejemplo por lecturas de sensores, estados de GPIO o cualquier dato que quieras mover entre ambos equipos.

Este proyecto está inspirado en el trabajo del lector Edgardo Peregrino y en el tutorial de Random Nerd Tutorials; acá lo adaptamos al español de Chile y lo extendemos con ideas propias más abajo.

Cómo funciona la comunicación de doble vía

Antes de cablear nada, conviene entender el modelo. En BLE hay dos roles: un periférico que anuncia (advertising) un servicio, y un central que se conecta a él. En nuestro caso:

• La Raspberry Pi Pico W es el periférico. Expone un servicio GATT con dos características.
• La Raspberry Pi 5 es el central. Se conecta al servicio y lee/escribe esas características.

Las dos características son la clave del truco:

• rx_characteristic (bandeja de entrada de la Pico): la Raspberry Pi escribe aquí. La Pico lee este valor, o sea, recibe datos de la Pi.
• tx_characteristic (bandeja de salida de la Pico): la Pico escribe aquí. La Pi lee este valor, o sea, recibe datos de la Pico.

Con una característica para cada sentido, las dos placas hablan y escuchan a la vez: eso es la comunicación bidireccional. A diferencia del Bluetooth clásico, BLE mantiene la radio apagada casi todo el tiempo y solo despierta para intercambiar paquetes cortos, por eso consume tan poco: es ideal para proyectos a batería.

Materiales y cableado

El montaje es mínimo. En la Raspberry Pi conectamos un LED al GPIO 17 con su resistencia en serie; ese LED nos da retroalimentación visual (parpadea distinto según si hay conexión o no). La Pico W la montamos sobre la protoboard solo por comodidad.

Por qué la resistencia de 220 Ω: el LED necesita limitar su corriente para no quemarse. Con los 3,3 V de los pines de la Raspberry Pi, una resistencia de 220 Ω deja pasar unos 10-12 mA, suficiente para un brillo cómodo y seguro. Si solo tienes de 330 Ω o 470 Ω también sirven; el LED se verá un poco más tenue.

Paso 1: Programa la Raspberry Pi Pico W (MicroPython)

La Pico corre MicroPython y usa la librería aioble (la capa asíncrona sobre el módulo bluetooth). Instálala desde Thonny: abre Tools > Manage Packages…, busca aioble y presiona Install.

Copia este código a Thonny y guárdalo en la Pico:

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

La idea del programa es correr tres tareas asíncronas en paralelo:

• send_task() escribe un mensaje con un contador en la tx_characteristic cada segundo (con send_update=True para notificar al central).
• receive_task() lee la rx_characteristic cada 500 ms y solo imprime cuando el valor cambió.
• peripheral_task() primero muestra la dirección MAC Bluetooth de la Pico y luego se queda anunciando el servicio y esperando conexiones.

Cuando lo ejecutes (botón verde o F5), la consola de Thonny imprimirá la MAC de la Pico. Anótala, porque la vas a necesitar en el código de la Raspberry Pi.

Paso 2: Prepara la Raspberry Pi (Python + bleak)

En la Raspberry Pi el rol central lo maneja la librería bleak. Necesitas una Pi con Bluetooth (una Raspberry Pi 4 o 5 sirven), un entorno virtual de Python con bleak instalado y gpiozero para controlar el LED.

Crea un archivo, por ejemplo pi_send_receive.py, dentro de la misma carpeta del entorno virtual, y pega este código:

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

Antes de correrlo, reemplaza la línea del pico_address con la dirección MAC que anotaste en el paso anterior:

pico_address = "2C:CF:67:B6:D7:4C"

Fíjate en cómo trabaja: send_data_task() escribe en la característica de la Pico con client.write_gatt_char() cada 5 segundos, receive_data_task() la lee con client.read_gatt_char() cada 2 segundos, y blink_task() hace parpadear el LED del GPIO 17 (cada 1 s si está conectado, cada 250 ms si no). El bloque connect_and_communicate() envuelve todo en un bucle que reconecta solo si la conexión se cae.

Paso 3: Ejecuta y prueba la comunicación

Activa el entorno virtual y lanza el programa en la Raspberry Pi:

source blue/bin/activate
python3 pi_send_receive.py

Asegúrate de que la Pico ya esté corriendo su código en Thonny. Tras unos segundos, la Pi se conecta y verás en la consola cómo ambas placas envían y reciben el contador, mientras el LED del GPIO 17 empieza a parpadear cada segundo.

¡Listo! Con esto ya tienes una comunicación bidireccional real entre la Raspberry Pi y la Pico W.

Variantes y mejoras

El ejemplo manda un contador, pero la gracia es reemplazarlo por datos útiles. Algunas ideas concretas para llevar el proyecto más lejos:

• Envía lecturas de un sensor DHT22 por BLE: conecta un sensor de temperatura y humedad DHT22 a la Pico W y, en send_task(), cambia el texto del contador por la lectura real (temp y humedad). Así la Pico se transforma en un nodo inalámbrico de medición ambiental que reporta a la Raspberry Pi.
• Dashboard en Home Assistant: en la Raspberry Pi, en vez de solo imprimir el mensaje recibido, publícalo por MQTT hacia una instancia de Home Assistant. Con eso tendrás la temperatura de la Pico en un panel bonito, con gráficos e historial, sin cablear nada entre las piezas.
• Registra los datos en microSD o CSV: si quieres guardar en lugar de solo mostrar, escribe cada lectura a un archivo .csv en la Raspberry Pi (o a un módulo microSD en la Pico) para después graficar la serie de tiempo.
• Controla GPIOs de forma remota: usa la rx_characteristic para mandarle comandos a la Pico ("enciende relé", "mueve servo") y cierra el lazo: no solo mides, también actúas a distancia.

Personalización para Chile

En Chile puedes conseguir todos los componentes de este proyecto en MechatronicStore. Estos son los que se usan en el montaje:

• Raspberry Pi 5. la placa central con Bluetooth integrado (una Raspberry Pi 4 también funciona).
• Raspberry Pi Pico W. el periférico BLE (la Pico 2 W es equivalente y también sirve).
• LED 5 mm (rojo). retroalimentación visual en el GPIO 17.
• Resistencia 220 Ω. limita la corriente del LED.
• Protoboard 830 puntos. para montar la Pico y el LED sin soldar.
• Jumpers macho hembra. para las conexiones entre placas y protoboard.

Nota: los precios y el stock exacto conviene verificarlos en el catálogo de MechatronicStore al momento de comprar.

Recursos

• Tutorial original (inglés): Bidirecional BLE Communication between Raspberry Pi and Raspberry Pi Pico W
• Autoría del ejemplo: código base creado por el lector Edgardo Peregrino (canal LinuxRobotGeek).
• Librería del periférico (MicroPython): aioble
• Librería del central (Python): bleak

Versión chilena con componentes en stock local en MechatronicStore.