117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
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 _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
|
|
|
|
if value_handle == self.tx_handle:
|
|
msg = self.ble.gatts_read(self.tx_handle)
|
|
|
|
if msg == b"ping":
|
|
for conn in self.connections:
|
|
self.ble.gatts_notify(conn, self.rx_handle, b"pong") |