Python Exec w/ BLE

main.py

import aioble
import bluetooth
import asyncio
import time

# Custom UUID for our service
SERVICE_UUID = bluetooth.UUID("A5A5A5A5-FFFF-9999-1111-5A5A5A5A5A5A")
CHAR_UUID = bluetooth.UUID("00000000-1111-2222-3333-444444444444")


# Functions available to executed scripts
def ble_print(msg):
    print(f"Script output: {msg}")


def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n - 1)


def timed_factorial(n):
    start = time.ticks_ms()
    result = factorial(n)
    duration = time.ticks_diff(time.ticks_ms(), start)
    ble_print(f"factorial({n}) = {result}")
    ble_print(f"Time taken: {duration}ms")
    return result


# Create restricted globals for script execution
RESTRICTED_GLOBALS = {
    "ble_print": ble_print,
    "timed_factorial": timed_factorial,
    "time": time,
}


async def main():
    # Set up BLE service
    service = aioble.Service(SERVICE_UUID)
    char = aioble.BufferedCharacteristic(
        service,
        CHAR_UUID,
        write=True,
        max_len=512,
        append=True,
    )
    aioble.register_services(service)

    print("Starting BLE script executor service...")

    while True:
        async with await aioble.advertise(
            250_000, name="MicroPython-Exec", services=[SERVICE_UUID]
        ) as connection:
            print("Client connected")

            try:
                mtu = await connection.exchange_mtu(515)
                print(f"Negotiated MTU: {mtu}")
            except:
                print("MTU exchange failed, using default")

            while connection.is_connected():
                try:
                    await char.written()
                    script = char.read().decode()
                    print(f"Received script: {script}")

                    try:
                        exec(script, RESTRICTED_GLOBALS)
                    except Exception as e:
                        print(f"Script execution error: {e}")

                except Exception as e:
                    print(f"Error: {e}")

            print("Client disconnected")


# Start the service
asyncio.run(main())