mesh-network-project/relay/bluetooth_handler.py

import bluetooth
import struct
from micropython import const


# i just made a UUID up and changed the last number to change what protocol you're using
SERVICE_UUID = bluetooth.UUID("E1898FF7-5063-4441-a6eb-526073B00001")
TX_UUID = bluetooth.UUID("E1898FF7-5063-4441-a6eb-526073B00002")
RX_UUID = bluetooth.UUID("E1898FF7-5063-4441-a6eb-526073B00003")

TX_CHAR = (TX_UUID, bluetooth.FLAG_NOTIFY)
RX_CHAR = (RX_UUID, bluetooth.FLAG_WRITE)

SERVICE = (SERVICE_UUID, (TX_CHAR, RX_CHAR))

IRQ_CONNECT = const(1)
IRQ_DISCONNECT = const(2)
IRQ_GATTS_WRITE = const(3)

# Advertising payloads are repeated packets of the following form:
#   1 byte data length (N + 1)
#   1 byte type (see constants below)
#   N bytes type-specific data

_ADV_TYPE_FLAGS = const(0x01)
_ADV_TYPE_NAME = const(0x09)
_ADV_TYPE_UUID16_COMPLETE = const(0x3)
_ADV_TYPE_UUID32_COMPLETE = const(0x5)
_ADV_TYPE_UUID128_COMPLETE = const(0x7)
_ADV_TYPE_UUID16_MORE = const(0x2)
_ADV_TYPE_UUID32_MORE = const(0x4)
_ADV_TYPE_UUID128_MORE = const(0x6)
_ADV_TYPE_APPEARANCE = const(0x19)

_ADV_MAX_PAYLOAD = const(31)


class BluetoothHandler:
    def __init__(self):
        print("Initializing Bluetooth...")
        self.ble = bluetooth.BLE()
        print("Activating...")
        self.ble.active(True)
        print("Setting IRQ callback...")
        self.ble.irq(self.irq)
        
        print("Getting MAC address...")
        self.mac_address = self._get_mac_address()
        print(f"MAC address: {self.mac_address}")
        self.ble.config(gap_name="NODE-" + self.mac_address)
        
        ((self.tx_handle, self.rx_handle),) = self.ble.gatts_register_services((SERVICE,))
        
        self.connections = set()
        
        self.advertise()
        
    def deserialize_msg(self, s: bytes):
        # returns packet type (int) and deserialized data
        return s[0], eval(s[1:].decode())
        
    def _get_mac_address(self):
        mac = self.ble.config("mac")[1]
        return ':'.join('{:02X}'.format(b) for b in mac)
    
    def advertising_payload(limited_disc=False, br_edr=False, name=None, services=None, appearance=0):
        payload = bytearray()

        def _append(adv_type, value):
            nonlocal payload
            payload += struct.pack("BB", len(value) + 1, adv_type) + value

        _append(
            _ADV_TYPE_FLAGS,
            struct.pack("B", (0x01 if limited_disc else 0x02) + (0x18 if br_edr else 0x04)),
        )

        if name:
            _append(_ADV_TYPE_NAME, name)

        if services:
            for uuid in services:
                b = bytes(uuid)
                if len(b) == 2:
                    _append(_ADV_TYPE_UUID16_COMPLETE, b)
                elif len(b) == 4:
                    _append(_ADV_TYPE_UUID32_COMPLETE, b)
                elif len(b) == 16:
                    _append(_ADV_TYPE_UUID128_COMPLETE, b)

        # See org.bluetooth.characteristic.gap.appearance.xml
        if appearance:
            _append(_ADV_TYPE_APPEARANCE, struct.pack("<h", appearance))

        if len(payload) > _ADV_MAX_PAYLOAD:
            raise ValueError("advertising payload too large")

        return payload
        
    def advertise(self):
        print("Advertising Bluetooth...")
        self.ble.gap_advertise(100_000, self.advertising_payload(name="MeshNode", services=[SERVICE_UUID]))
        
    def irq(self, event, data):
        print(f"BLUETOOTH IRQ | EVENT: {event}, DATA: {data}")
        if event == IRQ_CONNECT:
            conn_handle, _, _ = data
            self.connections.add(conn_handle)
        elif event == IRQ_DISCONNECT:
            conn_handle, _, _ = data
            self.connections.remove(conn_handle)
            self.advertise()
        elif event == IRQ_GATTS_WRITE:
            conn_handle, value_handle = data

            msg = self.ble.gatts_read(value_handle)
            packet_type, msg = self.deserialize_msg(msg)
            
            print(f"Received: \"{msg}\"")