Двусторонняя связь по BLE между Raspberry Pi и Raspberry Pi Pico W

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

import asyncio
import aioble
import bluetooth

_SERVICE_UUID = bluetooth.UUID(0x1848)
_PI_TO_PICO_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6E)
_PICO_TO_PI_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6F)
BLE_NAME = "Pico W Peripheral"

ble_service = aioble.Service(_SERVICE_UUID)

tx_characteristic.write(_encode_message(message), send_update=True)

data = rx_characteristic.read()

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

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

import asyncio
from bleak import BleakClient, uuids
from gpiozero import LED

SERVICE_UUID = uuids.normalize_uuid_16(0x1848)
WRITE_CHARACTERISTIC_UUID = uuids.normalize_uuid_16(0x2A6E)
READ_CHARACTERISTIC_UUID = uuids.normalize_uuid_16(0x2A6F)

await client.write_gatt_char(WRITE_CHARACTERISTIC_UUID, message)

response = await client.read_gatt_char(READ_CHARACTERISTIC_UUID)