aioble: ping pong

aioble_ping_pong.py

import sys
from collections import deque

from micropython import const

import asyncio
from lib import aioble
import bluetooth

import struct
import os

# Randomly generated UUIDs.
_DATA_SERVICE_UUID = bluetooth.UUID("6e400001-b5a3-f393-e0a9-e50e24dcca9e")
_IO_RX_CHARACTERISTIC_UUID = bluetooth.UUID("6e400002-b5a3-f393-e0a9-e50e24dcca9e")
_IO_TX_CHARACTERISTIC_UUID = bluetooth.UUID("6e400003-b5a3-f393-e0a9-e50e24dcca9e")
_NAME = "pingpong"

# How frequently to send advertising beacons.
_ADV_INTERVAL_MS = 250_000

ROLE_SERVER = True
ROLE_CLIENT = False

# Register GATT server.
data_service = aioble.Service(_DATA_SERVICE_UUID)
rx_characteristic = aioble.BufferedCharacteristic(
    data_service, _IO_RX_CHARACTERISTIC_UUID, read=True, write=True, write_no_response=True, capture=True, append=True
)

tx_characteristic = aioble.BufferedCharacteristic(
    data_service, _IO_TX_CHARACTERISTIC_UUID, notify=True, max_len=128
)

aioble.register_services(data_service)

async def server_task(connection):
    print("Server: Starting server_task...")
    _io_tx = tx_characteristic
    _io_rx = rx_characteristic

    try:
        with connection.timeout(None):
            count = 0
            while True:
                count += 1
                print("Server: Waiting for data...")
                try:
                    _, data = await _io_rx.written(timeout_ms=1000)
                    print(f"Server: Got {data}")
                except asyncio.TimeoutError:
                    print("Server: NO DATA!?")
                    pass
                print("Server: Writing PING")
                _io_tx.notify(connection, f"PING {count}".encode())
                    
    except aioble.DeviceDisconnectedError:
        return

async def client_task(connection):
    print("Client: Starting client_task...")
    service = await connection.service(_DATA_SERVICE_UUID)

    _io_tx = await service.characteristic(_IO_TX_CHARACTERISTIC_UUID)
    _io_rx = await service.characteristic(_IO_RX_CHARACTERISTIC_UUID)
    #_io_rx._notify_queue = deque((), 7)
    await _io_tx.subscribe(notify=True)

    try:
        with connection.timeout(None):
            while True:
                print("Client: Waiting for data...")
                data = await _io_tx.notified()
                print(f"Client: Got {data}")
                print("Client: Writing PONG")
                if b" " in data:
                    count = data.split(b" ")[1]
                await _io_rx.write(f"PONG {count}".encode(), True)
                await asyncio.sleep(1)
                    
    except aioble.DeviceDisconnectedError:
        return


async def server_listen_task():
    while True:
        print("Server: Waiting for connection...")
        connection = await aioble.advertise(
            _ADV_INTERVAL_MS,
            name=_NAME,
            services=[_DATA_SERVICE_UUID],
        )
        print("Server: Connection from", connection.device)
        await server_task(connection)
        await connection.disconnected()


async def client_connect_task():
    async with aioble.scan(1000, 30000, 30000, active=True) as scanner:
        async for result in scanner:
            if result.name() == _NAME and _DATA_SERVICE_UUID in result.services():
                device = result.device
                break
        else:
            print("Client: Server not found")
            return

        try:
            print("Client: Connecting to", device)
            connection = await device.connect()
        except asyncio.TimeoutError:
            print("Client: Timeout during connection")
            return
        
        await client_task(connection)

        await connection.disconnected()


# Run both tasks.
def main():
    print("Client: Attempting to connect to server...")
    await client_connect_task()
    print("Server: Listening...")
    await server_listen_task()

asyncio.run(main())