Python3: GATT で String データを送受信

通信相手のアドレスと Characteristic UUID　を与えて、String を送受信する方法です。

プログラム

send_get_data.py

#! /usr/bin/python
#	send_get_data.py
#
#					Oct/02/2024
# ------------------------------------------------------------------
import asyncio
from bleak import BleakClient

DEVICE_ADDRESS = "2C:BC:BB:65:E8:5A"  # デバイスのMACアドレスを入力
CHARACTERISTIC_UUID_READ = "7DEF8317-7301-4EE6-8849-46FACE74CA2A"  # 通知を受信するキャラクタリスティックのUUID
CHARACTERISTIC_UUID_WRITE = "7DEF8317-7302-4EE6-8849-46FACE74CA2A"  # 書き込みを行うキャラクタリスティックのUUID

# ------------------------------------------------------------------
# 通知を処理するコールバック関数
def notification_handler(sender, data):
	print(f"{data.decode()}")
#	print(f"Notification from {sender}: {data.decode()}")

# ------------------------------------------------------------------
# 通知を受信する関数
async def receive_notifications(client, char_uuid):
	await client.start_notify(char_uuid, notification_handler)
	print(f"Started notifications on {char_uuid}")

	# 通知を受け取り続ける（別のタスクで中断されるまで）
	while True:
		await asyncio.sleep(1)

# ------------------------------------------------------------------
# データを書き込む関数
async def write_data(client, char_uuid):
#	commands = ["MR\r","SV\r","SU\r","ST\r","SC\r"]
	commands = ["MB\r","SV\r","SU\r","ST\r","SC\r"]
	for i in range(5):  # 5回データを書き込む
		data = bytearray(commands[i].encode('utf-8'))  # 書き込むデータ
		await client.write_gatt_char(char_uuid, data)
		print(f"Written data to {char_uuid}: {data}")
		await asyncio.sleep(2)  # 2秒間隔で書き込む

# ------------------------------------------------------------------
# メイン関数
async def main():
	async with BleakClient(DEVICE_ADDRESS) as client:
		print(f"Connected: {client.is_connected}")

		# 並行して通知受信とデータ書き込みを行う
		notify_task = asyncio.create_task(receive_notifications(client, CHARACTERISTIC_UUID_READ))
		write_task = asyncio.create_task(write_data(client, CHARACTERISTIC_UUID_WRITE))

		# 5秒後にデータの書き込みが終了するので、その後も通知を受信する
		await write_task

		# 通知受信をしばらく続ける（ここでは5秒間）
		await asyncio.sleep(5)

		# 通知の停止
		await client.stop_notify(CHARACTERISTIC_UUID_READ)
		print(f"Stopped notifications on {CHARACTERISTIC_UUID_READ}")

		# 通知タスクの終了
		notify_task.cancel()
#
# ------------------------------------------------------------------
# 非同期タスクの実行
asyncio.run(main())
# ------------------------------------------------------------------

実行結果

$ ./send_get_data.py 
Connected: True
Started notifications on 7DEF8317-7301-4EE6-8849-46FACE74CA2A
Written data to 7DEF8317-7302-4EE6-8849-46FACE74CA2A: bytearray(b'MB\r')
Written data to 7DEF8317-7302-4EE6-8849-46FACE74CA2A: bytearray(b'SV\r')
Oct/06/2024 PM 14:09
Written data to 7DEF8317-7302-4EE6-8849-46FACE74CA2A: bytearray(b'SU\r')
UCHIDA
Written data to 7DEF8317-7302-4EE6-8849-46FACE74CA2A: bytearray(b'ST\r')
ABCDEFGH
Written data to 7DEF8317-7302-4EE6-8849-46FACE74CA2A: bytearray(b'SC\r')
Stopped notifications on 7DEF8317-7301-4EE6-8849-46FACE74CA2A