Need "Long Read" feature to read BLE keyboard's HID Report Map

[hyx0329]

• Platform: ESP32-S3
• MPY version: 1.25.0

I want my ESP32-S3 process input from arbitrary BLE keyboard(ZMK for example), and thus I need to read and process the HID Report Map, which can be a long value that cannot fit in a single packet(i.e. bigger than MTU). Micropython provides no API for a long read with offset which is essential when fragmentation comes.

From my understanding, a long read is basically a read operation with an offset.

• Nimble supports this through ble_gattc_read_long(),
• BTstack has a similar API gatt_client_read_long_value_of_characteristic_using_value_handle_with_offset() but it seems tend to read all chunks within one API call.
• Zephyr supports this through bt_gatt_read().

===
Here's my original attempt.

I tried to read the HID Report Map and parse it on ESP32-S3. But I can only get 22 bytes w/o mtu exchange and 64 bytes w/ mtu exchange. The central's MTU size is default to 256. The negotiated MTU is 65. My HID descriptor size is about 80 bytes.

ESP32-S3 works as central, and it connects to a ZMK-based bluetooth keyboard. ZMK uses zephyr 3.5.0.

I've seen a few similar issues/discussions(#7206 #11865 #12398 ), but non of the proposed solutions works for me.

Here's my testing code.

from micropython import const
import asyncio
import aioble
from aioble.core import GattError
import bluetooth

_UUID_SERVICE_GENERIC_ACCESS = bluetooth.UUID(0x1800)

_UUID_SERVICE_DEVICEINFO = bluetooth.UUID(0x180a)
_UUID_SERVICE_BATTERY = bluetooth.UUID(0x180f)
_UUID_SERVICE_HID = bluetooth.UUID(0x1812)

_UUID_CHAR_BATTERY_LEVEL = bluetooth.UUID(0x2a19)
_UUID_CHAR_HID_INFO = bluetooth.UUID(0x2a4a)
_UUID_CHAR_HID_REPORT_MAP = bluetooth.UUID(0x2a4b)
_UUID_CHAR_HID_REPORT = bluetooth.UUID(0x2a4d)
_UUID_CHAR_HID_CONTROL_POINT = bluetooth.UUID(0x2a4c)

## Controls whether the device should be suspended (0) or not (1).
# _UUID_HID_CONTROL_POINT = bluetooth.UUID(0x2a4c)

_IRQ_PASSKEY_ACTION = const(31)
_PASSKEY_ACTION_INPUT = const(2)
_PASSKEY_ACTION_DISP = const(3)
_PASSKEY_ACTION_NUMCMP = const(4)

# search GATT Status Codes
_ATTERR_INSUF_AUTHENTICATION = const(0x0105)
_ATTERR_INSUF_ENCRYPTION = const(0x010f)

keyboard_name = "YourKB"


async def find_hid_gadgets():
	async with aioble.scan(5000, interval_us=30000, window_us=30000, active=True) as scanner:
		async for result in scanner:
			if result.name() == keyboard_name:
				print("{}, {}, {}".format(result.name(), result.rssi, result))
				return result.device
	return None


async def main():
	while True:
		device = await find_hid_gadgets()
		if not device:
			asyncio.sleep(0.05)
			continue
		break

	try:
		print("Connecting to", device)
		connection = await device.connect()
	except asyncio.TimeoutError:
		print("Timeout during connection")
		return

	hid_service = None
	battery_service = None

	async with connection:
		try:

			# with this line, the size of data read increased to 64 bytes
			# otherwise 22 bytes
			await connection.exchange_mtu()

			# Make sure loop runs to completion.
			async for service in connection.services():
				print("Found service: ", service.uuid)
				if service.uuid == _UUID_SERVICE_HID:
					hid_service = service
				elif service.uuid == _UUID_SERVICE_BATTERY:
					battery_service = service

			if battery_service:
				battery_level_characteristic = None
				print("In battery service:")
				async for cha in battery_service.characteristics():
					print(cha)
					if cha.uuid == _UUID_CHAR_BATTERY_LEVEL:
						battery_level_characteristic = cha

				if battery_level_characteristic:
					bat_level = await battery_level_characteristic.read()
					print("Bat level", bat_level[0], "%")

			if hid_service:
				# TODO: check read permission, otherwise it may get GattError 261
				hid_descriptor_c = None
				hid_reports = list()
				print("In HID services:")
				async for cha in hid_service.characteristics():
					print(cha)
					if cha.uuid == _UUID_CHAR_HID_REPORT_MAP:
						hid_descriptor_c = cha
					elif cha.uuid == _UUID_CHAR_HID_REPORT:
						hid_reports.append(cha)

				if hid_descriptor_c:
					status = 0
					try:
						print(await hid_descriptor_c.read())
						for cha in hid_reports:
							print(await cha.read())
					except GattError as e:
						if e._status == _ATTERR_INSUF_AUTHENTICATION:
							print("Insufficient authentication! Authenticated link required.")
						elif e._status == _ATTERR_INSUF_ENCRYPTION:
							print("Probably paired but the link is not yet encrypted.")
						else:
							print("Unhandled GattError ", e._status)
						status = e._status

					# custom implementation required if passkey is used
					if status:
						status = 0
						print("Try again with MITM resistance")
						await connection.pair(mitm=True)
						try:
							print(await hid_descriptor_c.read())
							for cha in hid_reports:
								print(await cha.read())
						except GattError as e:
							status = e._status

					if status:
						print("Failed to read HID characteristics!")

		except asyncio.TimeoutError:
			print("Timeout discovering services/characteristics")
			return

asyncio.run(main())

[hyx0329]

Now I can confirm micropython will cut characteristic values longer than negotiated MTU minus 1. Yet there's no exposed procedure to do segmented long read(e.g. ble_gattc_read_long in nimble).