Bluetooth Remote Control for Raspberry Pi Pico W

remote.py

# Kevin McAleer
# 2023-06-28
# Bluetooth cores specification versio 5.4 (0x0D)

import sys

import aioble
import bluetooth
import machine
import uasyncio as asyncio
from micropython import const
from pimoroni import Button

def uid():
    """ Return the unique id of the device as a string """
    return "{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}".format(
        *machine.unique_id())

MANUFACTURER_ID = const(0x02A29)
MODEL_NUMBER_ID = const(0x2A24)
SERIAL_NUMBER_ID = const(0x2A25)
HARDWARE_REVISION_ID = const(0x2A26)
BLE_VERSION_ID = const(0x2A28)

button_a = Button(12)
button_b = Button(13)
button_x = Button(14)
button_y = Button(15)

led = machine.Pin("LED", machine.Pin.OUT)

_ENV_SENSE_UUID = bluetooth.UUID(0x180A)
_GENERIC = bluetooth.UUID(0x1848)
_ENV_SENSE_TEMP_UUID = bluetooth.UUID(0x1800)
_BUTTON_UUID = bluetooth.UUID(0x2A6E)

_BLE_APPEARANCE_GENERIC_REMOTE_CONTROL = const(384)

# Advertising frequency
ADV_INTERVAL_MS = 250_000

device_info = aioble.Service(_ENV_SENSE_UUID)

connection = None

# Create characteristics for device info
aioble.Characteristic(device_info, bluetooth.UUID(MANUFACTURER_ID), read=True, initial="KevsRobotsRemote")
aioble.Characteristic(device_info, bluetooth.UUID(MODEL_NUMBER_ID), read=True, initial="1.0")
aioble.Characteristic(device_info, bluetooth.UUID(SERIAL_NUMBER_ID), read=True, initial=uid())
aioble.Characteristic(device_info, bluetooth.UUID(HARDWARE_REVISION_ID), read=True, initial=sys.version)
aioble.Characteristic(device_info, bluetooth.UUID(BLE_VERSION_ID), read=True, initial="1.0")

remote_service = aioble.Service(_GENERIC)

button_characteristic = aioble.Characteristic(
    remote_service, _BUTTON_UUID, read=True, notify=True
)

print('registering services')
aioble.register_services(remote_service, device_info)

connected = False

async def remote_task():
    """ Send the event to the connected device """

    while True:
        if not connected:
            print('not connected')
            await asyncio.sleep_ms(1000)
            continue
        if button_a.read():
            print(f'Button A pressed, connection is: {connection}')
            
            #only need to write OR notify, not both!
            # button_characteristic.write(b"a")    
            button_characteristic.notify(connection,b"a")
        elif button_b.read():
            print('Button B pressed')
            # button_characteristic.write(b"b")
            button_characteristic.notify(connection,b"b")
        elif button_x.read():
            print('Button X pressed')
            # button_characteristic.write(b"x")
            button_characteristic.notify(connection,b"x")
        elif button_y.read():
            print('Button Y pressed')
            # button_characteristic.write(b"y")
            button_characteristic.notify(connection,b"x")
        else:
            button_characteristic.notify(b"!")
        await asyncio.sleep_ms(10)
            
# Serially wait for connections. Don't advertise while a central is
# connected.    
async def peripheral_task():
    print('peripheral task started')
    global connected, connection
    while True:
        connected = False
        async with await aioble.advertise(
            ADV_INTERVAL_MS, 
            name="KevsRobots", 
            appearance=_BLE_APPEARANCE_GENERIC_REMOTE_CONTROL, 
            services=[_ENV_SENSE_TEMP_UUID]
        ) as connection:
            print("Connection from", connection.device)
            connected = True
            print(f"connected: {connected}")
            await connection.disconnected()
            print(f'disconnected')
        

async def blink_task():
    print('blink task started')
    toggle = True
    while True:
        led.value(toggle)
        toggle = not toggle
        blink = 1000
        if connected:
            blink = 1000
        else:
            blink = 250
        await asyncio.sleep_ms(blink)
        
async def main():
    tasks = [
        asyncio.create_task(peripheral_task()),
        asyncio.create_task(blink_task()),
        asyncio.create_task(remote_task()),
    ]
    await asyncio.gather(*tasks)

asyncio.run(main())

robot.py

# June 2023
# Bluetooth cores specification versio 5.4 (0x0D)
# Bluetooth Remote Control
# Kevin McAleer
# KevsRobot.com

import aioble
import bluetooth
import machine
import uasyncio as asyncio
from burgerbot import Burgerbot

# Bluetooth UUIDS can be found online at https://www.bluetooth.com/specifications/gatt/services/

_REMOTE_UUID = bluetooth.UUID(0x1848)
_ENV_SENSE_UUID = bluetooth.UUID(0x1848) 
_REMOTE_CHARACTERISTICS_UUID = bluetooth.UUID(0x2A6E)

led = machine.Pin("LED", machine.Pin.OUT)
connected = False
alive = False

bot = Burgerbot()
bot.stop()

async def find_remote():
    # Scan for 5 seconds, in active mode, with very low interval/window (to
    # maximise detection rate).
    async with aioble.scan(5000, interval_us=30000, window_us=30000, active=True) as scanner:
        async for result in scanner:

            # See if it matches our name
            if result.name() == "KevsRobots":
                print("Found KevsRobots")
                for item in result.services():
                    print (item)
                if _ENV_SENSE_UUID in result.services():
                    print("Found Robot Remote Service")
                    return result.device
            
    return None

async def blink_task():
    """ Blink the LED on and off every second """
    
    print('blink task started')
    toggle = True
    
    while True and alive:
        blink = 250
        led.value(toggle)
        toggle = not toggle
        # print(f'blink {toggle}, connected: {connected}')
        if connected:
            blink = 1000
        else:
            blink = 250
        await asyncio.sleep_ms(blink)
    print('blink task stopped')

def move_robot(command):
    if command == b'a':
#         print("a button pressed")
        bot.forward(0.1)
    elif command == b'b':
#         print("b button pressed")
        bot.backward(0.1)
    elif command == b'x':
#         print("x button pressed")
        bot.turnleft(0.1)
    elif command == b'y':
#         print("y button pressed")
        bot.turnright(0.1)

def notify_callback(char, data):
    print('Data received: ', data)
    command = data
    if command == b'a':
        # print("a button pressed")
        bot.forward(0.01)
    elif command == b'b':
        # print("b button pressed")
        bot.backward(0.01)
    elif command == b'x':
        # print("x button pressed")
        bot.turnleft(0.01)
    elif command == b'y':
        # print("y button pressed")
        bot.turnright(0.01)

async def peripheral_task():
    print('starting peripheral task')
    global connected, alive
    connected = False
    device = await find_remote()
    if not device:
        print("Robot Remote not found")
        return
    try:
        print("Connecting to", device)
        connection = await device.connect()
        
    except asyncio.TimeoutError:
        print("Timeout during connection")
        return
      
    async with connection:
        print("Connected")
        alive = True
        connected = True
#         alive = True

        robot_service = await connection.service(_REMOTE_UUID)
        control_characteristic = await robot_service.characteristic(_REMOTE_CHARACTERISTICS_UUID)
        
        while True:
            try:
                if robot_service == None:
                    print('remote disconnected')
                    alive = False
                    break
                
            except asyncio.TimeoutError:
                print("Timeout discovering services/characteristics")
                alive = False
                break
            
            if control_characteristic == None:
                print('no control')
                alive = False
                break
           
            try:
                data = await control_characteristic.read(timeout_ms=1000)

                await control_characteristic.subscribe(notify=True)
                while True:
                    command = await control_characteristic.notified()
                    move_robot(command)
                    bot.stop()
                                                            
            except Exception as e:
                print(f'something went wrong; {e}')
                connected = False
                alive = False
                break
        await connection.disconnected()
        print(f'disconnected')
        alive = False
                
async def main():
    tasks = []
    tasks = [
        asyncio.create_task(blink_task()),
        asyncio.create_task(peripheral_task()),
    ]
    await asyncio.gather(*tasks)
    
while True:
    asyncio.run(main())

Little bit improved version, but removed robot. i use you code for remote control light

robot.py

# June 2023
# Bluetooth cores specification versio 5.4 (0x0D)
# Bluetooth Remote Control
# Kevin McAleer
# KevsRobot.com

import aioble
import bluetooth
import machine
import uasyncio as asyncio
# from burgerbot import Burgerbot

# Bluetooth UUIDS can be found online at https://www.bluetooth.com/specifications/gatt/services/

_REMOTE_UUID = bluetooth.UUID(0x1848)
_ENV_SENSE_UUID = bluetooth.UUID(0x1800) 
_REMOTE_CHARACTERISTICS_UUID = bluetooth.UUID(0x2A6E)

led = machine.Pin(4, machine.Pin.OUT)
connected = False
alive = False

# bot = Burgerbot()
# bot.stop()

async def find_remote():
    # Scan for 5 seconds, in active mode, with very low interval/window (to
    # maximise detection rate).
    async with aioble.scan(5000, interval_us=30000, window_us=30000, active=True) as scanner:
        async for result in scanner:

            # See if it matches our name
            if result.name() == "KevsRobots":
                print("Found KevsRobots services:", list(result.services()))
                for item in result.services():
                    print (item)
                if _ENV_SENSE_UUID in result.services():
                    print("Found Robot Remote Service")
                    return result.device
            
    return None



async def blink_task():
    print('blink task started')
    toggle = True
    while True:
        led.value(toggle)
        toggle = not toggle
        blink = 1000
        if connected:
            blink = 1000
        else:
            blink = 250
        await asyncio.sleep_ms(blink)
    print('blink task stopped')

        
# async def blink_task():
#     """ Blink the LED on and off every second """
#
#     print('blink task started')
#     toggle = True
#
#     while True and alive:
#         blink = 250
#         led.value(toggle)
#         toggle = not toggle
#         # print(f'blink {toggle}, connected: {connected}')
#         if connected:
#             blink = 1000
#         else:
#             blink = 250
#         toggle = not toggle
#         await asyncio.sleep_ms(blink)
#     print('blink task stopped')

def move_robot(command):
    if command == b'a':
        print("a button pressed")
        # bot.forward(0.1)
    elif command == b'b':
        print("b button pressed")
        # bot.backward(0.1)
    elif command == b'c':
        print("c button pressed")
        # bot.turnleft(0.1)
    elif command == b'd':
        print("d button pressed")
        # bot.turnright(0.1)

async def peripheral_task():
    print('starting peripheral task')
    global connected, alive
    while True:
        connected = False
        device = await find_remote()
        if not device:
            print("Robot Remote not found")
            await asyncio.sleep_ms(50)
            continue
            # return
        try:
            print("Connecting to", device)
            connection = await device.connect()
        
        except asyncio.TimeoutError:
            print("Timeout during connection")
            await asyncio.sleep_ms(50)
            continue
            # return
      
        async with connection:
            print("Connected")
            alive = True
            connected = True
    #         alive = True

            robot_service = await connection.service(_REMOTE_UUID)
            control_characteristic = await robot_service.characteristic(_REMOTE_CHARACTERISTICS_UUID)
        
            while True:
                try:
                    if robot_service == None:
                        print('remote disconnected')
                        alive = False
                        break
                
                except asyncio.TimeoutError:
                    print("Timeout discovering services/characteristics")
                    alive = False
                    break
            
                if control_characteristic == None:
                    print('no control')
                    alive = False
                    break
           
                try:
                    data = await control_characteristic.read(timeout_ms=1000)

                    await control_characteristic.subscribe(notify=True)
                    while True:
                        command = await control_characteristic.notified()
                        move_robot(command)
                        # bot.stop()
                    
                except Exception as e:
                    print(f'something went wrong;', e)
                    connected = False
                    alive = False
                    break

            await connection.disconnected()
            print(f'disconnected')
            alive = False
        await asyncio.sleep_ms(50)
                
async def main():
    tasks = []
    tasks = [
        asyncio.create_task(blink_task()),
        asyncio.create_task(peripheral_task()),
    ]
    await asyncio.gather(*tasks)
    
while True:
    asyncio.run(main())
    

remote.py

# Kevin McAleer
# 2023-06-28
# Bluetooth cores specification versio 5.4 (0x0D)

import sys

import aioble
import bluetooth
import machine
import uasyncio as asyncio
from micropython import const
# from pimoroni import Button
from machine import Pin


def uid():
    """ Return the unique id of the device as a string """
    return "{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}".format(
        *machine.unique_id())

MANUFACTURER_ID = const(0x02A29)
MODEL_NUMBER_ID = const(0x2A24)
SERIAL_NUMBER_ID = const(0x2A25)
HARDWARE_REVISION_ID = const(0x2A26)
BLE_VERSION_ID = const(0x2A28)

# button_a = Button(12)
# button_b = Button(13)
# button_x = Button(14)
# button_y = Button(15)


KEY_UP   = const(0)
KEY_DOWN = const(1)

keys = [['1', '2', '3', 'A'], ['4', '5', '6', 'B'], ['7', '8', '9', 'C'], ['*', '0', '#', 'D']]

cols = [4,17,18,22]
rows = [23,21,19,5]


# set pins for rows as outputs
row_pins = [Pin(pin_name, mode=Pin.OUT) for pin_name in rows]

# set pins for cols as inputs
col_pins = [Pin(pin_name, mode=Pin.IN, pull=Pin.PULL_DOWN) for pin_name in cols]

def init():
    for row in range(0,4):
        for col in range(0,4):
            row_pins[row].value(0)


def scan(row, col):
    """ scan the keypad """

    # set the current column to high
    row_pins[row].value(1)
    key = None

    # check for keypressed events
    if col_pins[col].value() == KEY_DOWN:
        key = KEY_DOWN
    if col_pins[col].value() == KEY_UP:
        key = KEY_UP
    row_pins[row].value(0)

    # return the key state
    return key

led = machine.Pin(27, machine.Pin.OUT)

_ENV_SENSE_UUID = bluetooth.UUID(0x180A)
_GENERIC = bluetooth.UUID(0x1848)
_ENV_SENSE_TEMP_UUID = bluetooth.UUID(0x1800)
_BUTTON_UUID = bluetooth.UUID(0x2A6E)

_BLE_APPEARANCE_GENERIC_REMOTE_CONTROL = const(384)

# Advertising frequency
ADV_INTERVAL_MS = 250_000

device_info = aioble.Service(_ENV_SENSE_UUID)

connection = None

# Create characteristics for device info
aioble.Characteristic(device_info, bluetooth.UUID(MANUFACTURER_ID), read=True, initial="KevsRobotsRemote")
aioble.Characteristic(device_info, bluetooth.UUID(MODEL_NUMBER_ID), read=True, initial="1.0")
aioble.Characteristic(device_info, bluetooth.UUID(SERIAL_NUMBER_ID), read=True, initial='3132333435363738')
aioble.Characteristic(device_info, bluetooth.UUID(HARDWARE_REVISION_ID), read=True, initial=sys.version)
aioble.Characteristic(device_info, bluetooth.UUID(BLE_VERSION_ID), read=True, initial="1.0")

remote_service = aioble.Service(_GENERIC)

button_characteristic = aioble.Characteristic(
    remote_service, _BUTTON_UUID, read=True, notify=True
)

print('registering services')
aioble.register_services(remote_service, device_info)
init()

connected = False
keys = [[],[],[],[]]

async def remote_task():
    """ Send the event to the connected device """

    while True:
        for row in range(4):
            keys[row] = [KEY_UP,KEY_UP,KEY_UP,KEY_UP]
            for col in range(4):
                keys[row][col] = scan(row, col)
        if not connected:
            print('not connected')
            await asyncio.sleep_ms(1000)
            continue
        if keys[0][3] == KEY_DOWN:
            print(f'Button A pressed')
            #only need to write OR notify, not both!
            # button_characteristic.write(b"a")    
            button_characteristic.notify(connection,b"a")
        elif keys[1][3] == KEY_DOWN:
            print('Button B pressed')
            # button_characteristic.write(b"b")
            button_characteristic.notify(connection,b"b")
        elif keys[2][3] == KEY_DOWN:
            print('Button X pressed')
            # button_characteristic.write(b"x")
            button_characteristic.notify(connection,b"c")
        elif keys[3][3] == KEY_DOWN:
            print('Button Y pressed')
            # button_characteristic.write(b"y")
            button_characteristic.notify(connection,b"d")
        else:
            button_characteristic.notify(connection, b"!")
        await asyncio.sleep_ms(200)
            
# Serially wait for connections. Don't advertise while a central is
# connected.    
async def peripheral_task():
    print('peripheral task started')
    global connected, connection
    while True:
        connected = False
        try:
            async with await aioble.advertise(
                ADV_INTERVAL_MS, 
                name="KevsRobots", 
                appearance=_BLE_APPEARANCE_GENERIC_REMOTE_CONTROL, 
                services=[_ENV_SENSE_TEMP_UUID]
            ) as connection:
                print("Connection from", connection.device)
                connected = True
                print(f"connected: {connected}")
                await connection.disconnected()

        except asyncio.CancelledError:
            print(f"asyncio.CancelledError")
        
        print(f'disconnected')
        await asyncio.sleep_ms(50)
        

async def blink_task():
    print('blink task started')
    toggle = True
    while True:
        led.value(toggle)
        toggle = not toggle
        blink = 1000
        if connected:
            blink = 1000
        else:
            blink = 250
        await asyncio.sleep_ms(blink)
        
async def main():
    tasks = [
        asyncio.create_task(peripheral_task()),
        asyncio.create_task(blink_task()),
        asyncio.create_task(remote_task()),
    ]
    await asyncio.gather(*tasks)

asyncio.run(main())