BLE: Unihiker M10 talks to K10

BLE/
└── SimpleMessage
    ├── K10
    │   └── main.py
    └── M10
        └── main.py

# connect via SSH (default password dfrobot)
$ ssh [email protected]

# verify python packages (optional)
$ pip3 freeze

# install new python packages
$ pip3 install asyncio bleak

# exit ssh connection (after successful installation)
$ exit

from micropython import const
from utime import sleep_ms
from ubluetooth import BLE, UUID, FLAG_NOTIFY, FLAG_WRITE
from unihiker_k10 import screen


DELAY_MS = const(250)
MAX_MSG = const(10)

UART_UUID = UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
TX = (UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E"), FLAG_NOTIFY)
RX = (UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E"), FLAG_WRITE)
UART_SERVICE = (UART_UUID, (TX, RX))

INFO_COLOR = const(0xFFFE00)
MSG_COLOR = const(0xFFFFFF)


class BLEPeripheral:

    IRQ_CENTRAL_CONNECT = const(1)
    IRQ_CENTRAL_DISCONNECT = const(2)
    IRQ_GATTS_WRITE = const(3)
    MAX_CONNECTIONS = const(3)
    MAX_TEXT_LEN = const(20)

    def __init__(self, device_name):
        self._ble = BLE()
        self._ble.active(True)

        mac = self._ble.config('mac')[1]
        self.mac_str = ':'.join('{:02X}'.format(b) for b in mac)

        self.message_updated = True
        self.message = None

        self._ble.irq(self._irq)
        self._connections = set()
        self._payload = self._make_payload(name=device_name)

        ((self.tx_handle, self.rx_handle),) = self._ble.gatts_register_services((UART_SERVICE,))
        self._advertise()

    def _irq(self, event, data) -> None:
        if event == self.IRQ_CENTRAL_CONNECT:
            conn_handle, _, _ = data
            self._connections.add(conn_handle)
            print('[INFO] central connected')

            self._advertise()
        elif event == self.IRQ_CENTRAL_DISCONNECT:
            conn_handle, _, _ = data
            self._connections.discard(conn_handle)
            print('[INFO] central disconnected')

            self._advertise()
        elif event == self.IRQ_GATTS_WRITE:
            conn_handle, attr_handle = data
            print('[INFO] central writes')

            msg = self._ble.gatts_read(attr_handle)
            try:
                text = msg.decode('utf-8')[:self.MAX_TEXT_LEN]
                self.message = text
                self.message_updated = True
            except UnicodeDecodeError:
                print('[ERROR] invalid unicode received')

    @staticmethod
    def _make_payload(name) -> bytearray:
        name_bytes = name.encode('utf-8')

        payload = bytearray()
        payload += bytes((2, 0x01, 0x06))
        payload += bytes((len(name_bytes) + 1, 0x09)) + name_bytes

        return payload

    def _advertise(self) -> None:
        if len(self._connections) < self.MAX_CONNECTIONS:
            print('[INFO] send ble advertise')
            self._ble.gap_advertise(100, self._payload)
        else:
            self._ble.gap_advertise(None)


if __name__ == '__main__':
    # variables
    ble_name = 'Unihiker K10'
    messages = []

    # initialize BLE
    ble = BLEPeripheral(device_name=ble_name)

    # initialize display
    screen.init(dir=2)

    # display loop
    while True:
        sleep_ms(DELAY_MS)

        if not ble.message_updated:
            continue

        screen.clear()

        if ble.message:
            messages.append(ble.message)
            messages = messages[-MAX_MSG:]

            ble.message = None

        screen.draw_text(text=ble_name, x=10, y=10, color=INFO_COLOR)
        screen.draw_text(text=ble.mac_str, x=10, y=30, color=INFO_COLOR)
        screen.draw_line(x0=10, y0=55, x1=230, y1=55, color=MSG_COLOR)

        for i, txt_msg in enumerate(messages):
            pos_y = 75 + i * 20
            screen.draw_text(text=f'> {txt_msg}', x=10, y=pos_y, color=MSG_COLOR)

        screen.show_draw()
        ble.message_updated = False

import asyncio
from threading import Lock, Thread
from tkinter import Tk, Event, END, Label, StringVar, Entry
from bleak import BleakClient


K10_ADDRESS: str = "D8:3B:DA:72:2C:FE"  # K10 address
RX_UUID: str = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"  # BLE characteristic UUID for RX

SCREEN_WIDTH: int = 240
SCREEN_HEIGHT: int = 320
MAX_CHARS: int = 20


def send_ble_message(text: str) -> None:
    """
    Send a Bluetooth Low Energy (BLE) message to a specific device. This function
    establishes a connection to the BLE device based on its predefined address,
    checks the connection status, and sends the specified text message using its
    GATT characteristic.

    :param text: The string message to send to the BLE device.
    :type text: str
    """
    global sending_lock

    if not sending_lock.acquire(blocking=False):
        print("[WARNING] Ignoring message, another message is still being sent.")
        return

    async def send():
        try:
            async with BleakClient(K10_ADDRESS) as client:
                if client.is_connected:
                    await asyncio.sleep(0.5)
                    await client.write_gatt_char(RX_UUID, text.encode())
        except Exception as err:
            print(f'[ERROR] {err}')
        finally:
            sending_lock.release()

    asyncio.run(send())

def on_enter(event: Event = None) -> None:
    """
    Handles the event of pressing the Enter key, retrieves the text from an entry
    widget, deletes the entry content, and sends the text in a background thread
    to a BLE device.

    :param event: The event triggered by pressing the Enter key.
    :type event: Event
    """
    _ = event

    text = entry.get()

    if len(text) > MAX_CHARS:
        text = text[:MAX_CHARS]

    if text:
        Thread(target=send_ble_message, args=(text,), daemon=True).start()
    else:
        print("[WARNING] Empty text, ignoring.")

    entry.delete(0, END)


def limit_input(*args) -> None:
    """
    Limits the input length of a variable to a specified maximum number of characters.

    :param args: Arbitrary positional arguments passed to the function.
    """
    _ = args

    value = input_var.get()
    if len(value) > MAX_CHARS:
        input_var.set(value[:MAX_CHARS])


if __name__ == "__main__":
    sending_lock = Lock()

    root = Tk()
    root.geometry(f'{SCREEN_WIDTH}x{SCREEN_HEIGHT}+0+0')
    root.configure(bg="black")

    label = Label(root, text="Text to send:", fg="white", bg="black", font=("Arial", 16))
    label.place(x=10, y=60)

    input_var = StringVar()
    input_var.trace_add("write", limit_input)

    entry = Entry(root, textvariable=input_var, font=("Arial", 16), width=20, justify="left")
    entry.place(x=10, y=100, width=220, height=40)
    entry.bind("<Return>", on_enter)
    entry.focus()

    root.mainloop()