73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
|
import asyncio
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
from io import BytesIO
|
||
|
|
from bleak import BleakClient
|
||
|
|
from bleak import _logger as logger
|
||
|
|
from bleak.uuids import uuid16_dict
|
||
|
|
|
||
|
|
os.chdir(os.path.dirname(os.path.realpath(__file__)))
|
||
|
|
|
||
|
|
REGULAR_UUID = "D89B0002-95C5-6F8D-1D44-8B1245563C4D"
|
||
|
|
|
||
|
|
target = 'raw1.bin'
|
||
|
|
f = open(target, 'wb')
|
||
|
|
f.truncate()
|
||
|
|
|
||
|
|
global IsFirstPacket
|
||
|
|
global TimeStart
|
||
|
|
global TimeStop
|
||
|
|
global TotalBytes
|
||
|
|
|
||
|
|
IsFirstPacket = True
|
||
|
|
TotalBytes = 0
|
||
|
|
|
||
|
|
def notification_handler(sender, data):
|
||
|
|
global IsFirstPacket
|
||
|
|
global TimeStart
|
||
|
|
global TimeStop
|
||
|
|
global TotalBytes
|
||
|
|
if IsFirstPacket :
|
||
|
|
TimeStart = time.time()
|
||
|
|
TimeStop = time.time()
|
||
|
|
IsFirstPacket = False
|
||
|
|
else :
|
||
|
|
TimeStop = time.time()
|
||
|
|
TotalBytes += len(data)
|
||
|
|
f.write(data)
|
||
|
|
if (TimeStop-TimeStart) > 0:
|
||
|
|
print("Throughput: {0} kbps ({1})".format(round((TotalBytes * 8 / 1024)/(TimeStop-TimeStart), 2),len(data)))
|
||
|
|
|
||
|
|
async def ble_notify_demo(mac_addr: str):
|
||
|
|
print("Connect to {0}".format(mac_addr))
|
||
|
|
client = BleakClient(mac_addr)
|
||
|
|
await client.connect()
|
||
|
|
IsConnected = await client.is_connected()
|
||
|
|
if IsConnected == False :
|
||
|
|
print("Connect fail.")
|
||
|
|
return False
|
||
|
|
print("Connect success.")
|
||
|
|
print("Wait 2 sec for PHY update procedure.")
|
||
|
|
await asyncio.sleep(2.0)
|
||
|
|
|
||
|
|
await client.start_notify(REGULAR_UUID, notification_handler)
|
||
|
|
|
||
|
|
# delay 10 sec
|
||
|
|
await asyncio.sleep(15)
|
||
|
|
|
||
|
|
await client.stop_notify(REGULAR_UUID)
|
||
|
|
|
||
|
|
await client.disconnect()
|
||
|
|
|
||
|
|
return True
|
||
|
|
|
||
|
|
mac_addr = "DF:4C:23:1C:F5:59"
|
||
|
|
mac_addr = "EB:E6:E0:52:26:ED"
|
||
|
|
mac_addr = "D8:96:4A:0B:14:4A"
|
||
|
|
mac_addr = "DC:05:1F:1F:71:DA"
|
||
|
|
if asyncio.run(ble_notify_demo(mac_addr)) == True :
|
||
|
|
print("Recv total: {0} bytes".format(TotalBytes))
|
||
|
|
f.close()
|
||
|
|
# 資料讀取 n秒會自動存檔，並用 hex editor 開啟
|
||
|
|
os.system('010Editor.exe ' + target) # open file with hex editor
|