ble_nus.py

import aioble
import bluetooth

_NUS = bluetooth.UUID("6e400001-b5a3-f393-e0a9-e50e24dcca9e")
_RX = bluetooth.UUID("6e400002-b5a3-f393-e0a9-e50e24dcca9e")
_TX = bluetooth.UUID("6e400003-b5a3-f393-e0a9-e50e24dcca9e")

_ADV_INTERVAL_US = 250_000
_MTU = 256

_ADV_TYPE_FLAGS = 0x01
_ADV_TYPE_UUID128_COMPLETE = 0x07
_ADV_TYPE_NAME_COMPLETE = 0x09
_FLAGS_GENERAL_DISC_NO_BREDR = 0x06


def _ad_field(ad_type, value):
    return bytes((len(value) + 1, ad_type)) + value


def _adv_payload():
    return _ad_field(_ADV_TYPE_FLAGS, bytes((_FLAGS_GENERAL_DISC_NO_BREDR,))) + _ad_field(
        _ADV_TYPE_UUID128_COMPLETE, bytes(_NUS)
    )


def _resp_payload(name):
    return _ad_field(_ADV_TYPE_NAME_COMPLETE, name.encode("utf-8"))


def _name_from_mac():
    # Claude's desktop scanner filters on names starting with "Claude". Match the
    # reference firmware's "Claude-XXXX" form (last two BT MAC bytes, hex).
    _, mac = bluetooth.BLE().config("mac")
    return "Claude-{:02X}{:02X}".format(mac[4], mac[5])


class NUSPeripheral:
    def __init__(self, name=None):
        ble = bluetooth.BLE()
        # Widen the max ATT MTU we'll accept before the stack goes active —
        # Claude's desktop app sends full JSON lines in single writes and relies
        # on MTU negotiation to avoid truncation at the default 23-byte MTU.
        try:
            ble.config(mtu=_MTU)
        except (OSError, ValueError):
            pass
        ble.active(True)
        self.name = name or _name_from_mac()
        self._svc = aioble.Service(_NUS)
        self._rx = aioble.Characteristic(
            self._svc, _RX, write=True, write_no_response=True, capture=True
        )
        self._tx = aioble.Characteristic(self._svc, _TX, read=True, notify=True)
        aioble.register_services(self._svc)
        # MicroPython's default per-characteristic GATT buffer is ~20 bytes and
        # silently truncates longer writes even with a large negotiated MTU.
        # Bump the RX buffer so full JSON lines arrive intact.
        try:
            ble.gatts_set_buffer(self._rx._value_handle, _MTU, False)
        except (AttributeError, OSError):
            pass
        self._conn = None
        self._buf = bytearray()

    @property
    def connected(self):
        return self._conn is not None

    async def run(self, on_line, logger=print):
        adv = _adv_payload()
        resp = _resp_payload(self.name)
        while True:
            try:
                logger("ble: advertising as", self.name)
                async with await aioble.advertise(
                    _ADV_INTERVAL_US,
                    adv_data=adv,
                    resp_data=resp,
                ) as connection:
                    self._conn = connection
                    logger("ble: connected from", connection.device)
                    try:
                        await connection.exchange_mtu(_MTU)
                    except Exception as e:
                        logger("ble: mtu exchange failed", repr(e))
                    logger("ble: mtu now", getattr(connection, "mtu", "?"))
                    await self._pump(on_line, logger)
            except Exception as e:
                logger("ble: error", repr(e))
            finally:
                self._conn = None
                self._buf = bytearray()
                logger("ble: disconnected")

    async def _pump(self, on_line, logger=print):
        while True:
            _, data = await self._rx.written()
            self._buf += data
            while b"\n" in self._buf:
                line, _, rest = self._buf.partition(b"\n")
                self._buf = bytearray(rest)
                try:
                    text = line.decode("utf-8").strip()
                except UnicodeError:
                    continue
                if text:
                    await on_line(text)

    async def send(self, text, logger=print):
        if self._conn is None:
            return False
        payload = (text + "\n").encode("utf-8")
        chunk = max(20, getattr(self._conn, "mtu", 23) - 3)
        try:
            for i in range(0, len(payload), chunk):
                self._tx.notify(self._conn, payload[i : i + chunk])
        except Exception as e:
            logger("ble: tx error", repr(e))
            return False
        return True