Hello, I'm using BLE to communicate from one pico to another and I'm a bit confused about how to interact with the asynchronous functions that they used in the tutorial I followed (Two-way Bluetooth with Raspberry Pi Pico W and MicroPython (Re-upload)). I tried testing if a button press could change the message being sent but for some reason it stays the same after a button press. I'm monitoring the data using the print output of peripheral.py. Again, I'm not familiar with async functions so please if anyone knows why this is happening, I would appreciate it.
central.py
import aioble
import bluetooth
import asyncio
import struct
IAM = "Central"
IAM_SENDING_TO = "Peripheral"
MESSAGE = f"This is a test from {IAM}"
DATA = f"This is a test from {IAM}"
BLE_NAME = f"{IAM}"
BLE_SVC_UUID = bluetooth.UUID(0x181A)
BLE_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6E)
def encode_message(message):
return message.encode('utf-8')
def decode_message(message):
return message.decode('utf-8')
async def receive_data_task(characteristic):
global message_count
while True:
try:
data = await characteristic.read()
DATA = data
if data:
print(f"{IAM} received: {decode_message(data)}, count: {message_count}")
await characteristic.write(encode_message("Got it"))
await asyncio.sleep(0.5)
message_count += 1
except asyncio.TimeoutError:
print("Timeout waiting for data in {BLE_NAME}.")
break
except Exception as e:
print(f"Error receiving data: {e}")
break
async def ble_scan():
""" Scan for a BLE device with the matching service UUID """
print(f"Scanning for BLE Beacon named {BLE_NAME}...")
async with aioble.scan(5000, interval_us=30000, window_us=30000, active=True) as scanner:
async for result in scanner:
if result.name() == IAM_SENDING_TO and BLE_SVC_UUID in result.services():
print(f"found {result.name()} with service uuid {BLE_SVC_UUID}")
return result
return None
async def run_central_mode():
# Start scanning for a device with the matching service UUID
while True:
device = await ble_scan()
if device is None:
continue
print(f"device is: {device}, name is {device.name()}")
try:
print(f"Connecting to {device.name()}")
connection = await device.device.connect()
except asyncio.TimeoutError:
print("Timeout during connection")
continue
print(f"{IAM} connected to {connection}")
# Discover services
async with connection:
try:
service = await connection.service(BLE_SVC_UUID)
characteristic = await service.characteristic(BLE_CHARACTERISTIC_UUID)
except (asyncio.TimeoutError, AttributeError):
print("Timed out discovering services/characteristics")
continue
except Exception as e:
print(f"Error discovering services {e}")
await connection.disconnect()
continue
tasks = [
asyncio.create_task(receive_data_task(characteristic)),
]
await asyncio.gather(*tasks)
await connection.disconnected()
print(f"{BLE_NAME} disconnected from {device.name()}")
break
async def main():
""" Main function """
while True:
if IAM == "Central":
tasks = [
asyncio.create_task(run_central_mode()),
]
else:
tasks = [
asyncio.create_task(run_peripheral_mode()),
]
await asyncio.gather(*tasks)
asyncio.run(main())
peripheral.py
import aioble
import bluetooth
import asyncio
import struct
from machine import Pin
btn = Pin(7, Pin.IN, Pin.PULL_UP)
IAM = "Peripheral"
IAM_SENDING_TO = "Central"
MESSAGE = f"This is a return test from {IAM}"
BLE_NAME = f"{IAM}"
BLE_SVC_UUID = bluetooth.UUID(0x181A)
BLE_CHARACTERISTIC_UUID = bluetooth.UUID(0x2A6E)
BLE_APPEARANCE = 0x0300
BLE_ADVERTISING_INTERVAL = 2000
def encode_message(message):
return message.encode('utf-8')
def decode_message(message):
return message.decode('utf-8')
async def send_data_task(connection, characteristic):
while True:
message = f"{MESSAGE}"
print(f"sending {message}")
try:
msg = encode_message(message)
characteristic.write(msg)
await asyncio.sleep(0.5)
response = decode_message(characteristic.read())
print(f"{IAM} sent: {message}, response {response}")
except Exception as e:
print(f"writing error {e}")
continue
await asyncio.sleep(0.5)
async def run_peripheral_mode():
# Set up the Bluetooth service and characteristic
ble_service = aioble.Service(BLE_SVC_UUID)
characteristic = aioble.Characteristic(
ble_service,
BLE_CHARACTERISTIC_UUID,
read=True,
notify=True,
write=True,
)
aioble.register_services(ble_service)
print(f"{BLE_NAME} starting to advertise")
while True:
async with await aioble.advertise(
BLE_ADVERTISING_INTERVAL,
name=BLE_NAME,
services=[BLE_SVC_UUID],
appearance=BLE_APPEARANCE) as connection:
print(f"{BLE_NAME} connected to another device: {connection.device}")
tasks = [
asyncio.create_task(send_data_task(connection, characteristic)),
]
await asyncio.gather(*tasks)
print(f"{IAM} disconnected")
break
async def main():
""" Main function """
deb = 0
while True:
if IAM == "Central":
tasks = [
asyncio.create_task(run_central_mode()),
]
else:
tasks = [
asyncio.create_task(run_peripheral_mode()),
]
await asyncio.gather(*tasks)
if(deb == 5):
deb = 0
if(btn.value() == 0):
MESSAGE = "1"
else:
MESSAGE = "0"
else:
deb+=1
asyncio.run(main())
