github
/
micropython
mirror of https://github.com/micropython/micropython.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

tests/multi_bluetooth: Update to work with new BLE events. 

Updates the tests to use non-bitmask events, event renames, as well as some
of the new completion events to improve reliability of the tests.
pull/6103/head
Jim Mussared 5 years ago
committed by Damien George
parent
c07ea3e4c2
commit
9902ce12eb
9 changed files with 197 additions and 145 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 72
      tests/multi_bluetooth/ble_characteristic.py
  2. 10
      tests/multi_bluetooth/ble_characteristic.py.exp
  3. 6
      tests/multi_bluetooth/ble_gap_advertise.py
  4. 55
      tests/multi_bluetooth/ble_gap_connect.py
  5. 53
      tests/multi_bluetooth/ble_gap_device_name.py
  6. 86
      tests/multi_bluetooth/ble_gatt_data_transfer.py
  7. 13
      tests/multi_bluetooth/ble_gatt_data_transfer.py.exp
  8. 45
      tests/multi_bluetooth/ble_gattc_discover_services.py
  9. 2
      tests/run-multitests.py

72
tests/multi_bluetooth/ble_characteristic.py
View File

@ -5,15 +5,17 @@ import time, machine, bluetooth
TIMEOUT_MS = 5000
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_GATTS_WRITE = const(1 << 2)
_IRQ_PERIPHERAL_CONNECT = const(1 << 6)
_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(1 << 9)
_IRQ_GATTC_READ_RESULT = const(1 << 11)
_IRQ_GATTC_WRITE_STATUS = const(1 << 12)
_IRQ_GATTC_NOTIFY = const(1 << 13)
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_GATTS_WRITE = const(3)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
_IRQ_GATTC_CHARACTERISTIC_DONE = const(12)
_IRQ_GATTC_READ_RESULT = const(15)
_IRQ_GATTC_READ_DONE = const(16)
_IRQ_GATTC_WRITE_DONE = const(17)
_IRQ_GATTC_NOTIFY = const(18)
SERVICE_UUID = bluetooth.UUID("A5A5A5A5-FFFF-9999-1111-5A5A5A5A5A5A")
CHAR_UUID = bluetooth.UUID("00000000-1111-2222-3333-444444444444")
@ -27,15 +29,13 @@ SERVICE = (
)
SERVICES = (SERVICE,)
last_event = None
last_data = None
waiting_event = None
waiting_data = None
value_handle = 0
def irq(event, data):
global last_event, last_data, value_handle
last_event = event
last_data = data
global waiting_event, waiting_data, value_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
elif event == _IRQ_CENTRAL_DISCONNECT:
@ -53,21 +53,32 @@ def irq(event, data):
value_handle = data[2]
elif event == _IRQ_GATTC_READ_RESULT:
print("_IRQ_GATTC_READ_RESULT", data[-1])
elif event == _IRQ_GATTC_WRITE_STATUS:
print("_IRQ_GATTC_WRITE_STATUS", data[-1])
elif event == _IRQ_GATTC_READ_DONE:
print("_IRQ_GATTC_READ_DONE", data[-1])
elif event == _IRQ_GATTC_WRITE_DONE:
print("_IRQ_GATTC_WRITE_DONE", data[-1])
elif event == _IRQ_GATTC_NOTIFY:
print("_IRQ_GATTC_NOTIFY", data[-1])
if waiting_event is not None:
if (isinstance(waiting_event, int) and event == waiting_event) or (
not isinstance(waiting_event, int) and waiting_event(event, data)
):
waiting_event = None
waiting_data = data
def wait_for_event(event, timeout_ms):
global waiting_event, waiting_data
waiting_event = event
waiting_data = None
t0 = time.ticks_ms()
while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
if isinstance(event, int):
if last_event == event:
break
elif event():
break
if waiting_data:
return True
machine.idle()
return False
# Acting in peripheral role.
@ -82,17 +93,18 @@ def instance0():
ble.gatts_write(char_handle, "periph0")
# Wait for central to connect to us.
wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_CENTRAL_CONNECT:
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = last_data
conn_handle, _, _ = waiting_data
# Wait for a write to the characteristic from the central.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS)
# Wait a bit, then write the characteristic and notify it.
time.sleep_ms(1000)
print("gatts_write")
ble.gatts_write(char_handle, "periph1")
print("gatts_notify")
ble.gatts_notify(conn_handle, char_handle)
# Wait for a write to the characteristic from the central.
@ -100,6 +112,7 @@ def instance0():
# Wait a bit, then notify a new value on the characteristic.
time.sleep_ms(1000)
print("gatts_notify")
ble.gatts_notify(conn_handle, char_handle, "periph2")
# Wait for the central to disconnect.
@ -115,14 +128,13 @@ def instance1():
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(0, BDADDR)
wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_CONNECT:
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = last_data
conn_handle, _, _ = waiting_data
# Discover characteristics.
ble.gattc_discover_characteristics(conn_handle, 1, 65535)
wait_for_event(lambda: value_handle, TIMEOUT_MS)
wait_for_event(lambda event, data: value_handle, TIMEOUT_MS)
# Issue read of characteristic, should get initial value.
print("gattc_read")
@ -132,7 +144,7 @@ def instance1():
# Write to the characteristic, and ask for a response.
print("gattc_write")
ble.gattc_write(conn_handle, value_handle, "central0", 1)
wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
# Wait for a notify, then read new value.
wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS)
@ -143,7 +155,7 @@ def instance1():
# Write to the characteristic, and ask for a response.
print("gattc_write")
ble.gattc_write(conn_handle, value_handle, "central1", 1)
wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
# Wait for a notify (should have new data), then read old value (should be unchanged).
wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS)

10
tests/multi_bluetooth/ble_characteristic.py.exp
View File

@ -2,7 +2,10 @@
gap_advertise
_IRQ_CENTRAL_CONNECT
_IRQ_GATTS_WRITE b'central0'
gatts_write
gatts_notify
_IRQ_GATTS_WRITE b'central1'
gatts_notify
_IRQ_CENTRAL_DISCONNECT
--- instance1 ---
gap_connect
@ -10,15 +13,18 @@ _IRQ_PERIPHERAL_CONNECT
_IRQ_GATTC_CHARACTERISTIC_RESULT UUID128('00000000-1111-2222-3333-444444444444')
gattc_read
_IRQ_GATTC_READ_RESULT b'periph0'
_IRQ_GATTC_READ_DONE 0
gattc_write
_IRQ_GATTC_WRITE_STATUS 0
_IRQ_GATTC_WRITE_DONE 0
_IRQ_GATTC_NOTIFY b'periph1'
gattc_read
_IRQ_GATTC_READ_RESULT b'periph1'
_IRQ_GATTC_READ_DONE 0
gattc_write
_IRQ_GATTC_WRITE_STATUS 0
_IRQ_GATTC_WRITE_DONE 0
_IRQ_GATTC_NOTIFY b'periph2'
gattc_read
_IRQ_GATTC_READ_RESULT b'central1'
_IRQ_GATTC_READ_DONE 0
gap_disconnect: True
_IRQ_PERIPHERAL_DISCONNECT

6
tests/multi_bluetooth/ble_gap_advertise.py
View File

@ -3,8 +3,8 @@
from micropython import const
import time, machine, bluetooth
_IRQ_SCAN_RESULT = const(1 << 4)
_IRQ_SCAN_COMPLETE = const(1 << 5)
_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)
ADV_TIME_S = 3
@ -43,7 +43,7 @@ def instance1():
else:
if adv_data != data[4]:
adv_data = b"MISMATCH"
elif ev == _IRQ_SCAN_COMPLETE:
elif ev == _IRQ_SCAN_DONE:
finished = True
ble.config(rxbuf=2000)

55
tests/multi_bluetooth/ble_gap_connect.py
View File

@ -5,19 +5,17 @@ import time, machine, bluetooth
TIMEOUT_MS = 4000
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_PERIPHERAL_CONNECT = const(1 << 6)
_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7)
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
last_event = None
last_data = None
waiting_event = None
waiting_data = None
def irq(event, data):
global last_event, last_data
last_event = event
last_data = data
global waiting_event, waiting_data
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
elif event == _IRQ_CENTRAL_DISCONNECT:
@ -27,11 +25,23 @@ def irq(event, data):
elif event == _IRQ_PERIPHERAL_DISCONNECT:
print("_IRQ_PERIPHERAL_DISCONNECT")
if waiting_event is not None:
if event == waiting_event:
waiting_event = None
waiting_data = data
def wait_for_event(event, timeout_ms):
global waiting_event, waiting_data
waiting_event = event
waiting_data = None
t0 = time.ticks_ms()
while last_event != event and time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
if waiting_data:
return True
machine.idle()
return False
# Acting in peripheral role.
@ -42,20 +52,20 @@ def instance0():
multitest.next()
try:
# Wait for central to connect, then wait for it to disconnect.
wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS)
if last_event != _IRQ_CENTRAL_DISCONNECT:
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
if not wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS):
return
# Start advertising again.
ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
# Wait for central to connect, then disconnect it.
wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_CENTRAL_CONNECT:
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
print("gap_disconnect:", ble.gap_disconnect(waiting_data[0]))
if not wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS):
return
print("gap_disconnect:", ble.gap_disconnect(last_data[0]))
wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS)
finally:
ble.active(0)
@ -67,12 +77,10 @@ def instance1():
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(0, BDADDR)
wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_CONNECT:
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
print("gap_disconnect:", ble.gap_disconnect(last_data[0]))
wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_DISCONNECT:
print("gap_disconnect:", ble.gap_disconnect(waiting_data[0]))
if not wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS):
return
# Wait for peripheral to start advertising again.
@ -81,8 +89,7 @@ def instance1():
# Connect to peripheral and then let the peripheral disconnect us.
print("gap_connect")
ble.gap_connect(0, BDADDR)
wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_CONNECT:
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)
finally:

53
tests/multi_bluetooth/ble_gap_device_name.py
View File

@ -5,24 +5,24 @@ import time, machine, bluetooth
TIMEOUT_MS = 5000
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_PERIPHERAL_CONNECT = const(1 << 6)
_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(1 << 9)
_IRQ_GATTC_READ_RESULT = const(1 << 11)
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
_IRQ_GATTC_CHARACTERISTIC_DONE = const(12)
_IRQ_GATTC_READ_RESULT = const(15)
_IRQ_GATTC_READ_DONE = const(16)
GAP_DEVICE_NAME_UUID = bluetooth.UUID(0x2A00)
last_event = None
last_data = None
waiting_event = None
waiting_data = None
value_handle = 0
def irq(event, data):
global last_event, last_data, value_handle
last_event = event
last_data = data
global waiting_event, waiting_data, value_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
elif event == _IRQ_CENTRAL_DISCONNECT:
@ -38,16 +38,23 @@ def irq(event, data):
elif event == _IRQ_GATTC_READ_RESULT:
print("_IRQ_GATTC_READ_RESULT", data[-1])
if waiting_event is not None:
if isinstance(waiting_event, int) and event == waiting_event:
waiting_event = None
waiting_data = data
def wait_for_event(event, timeout_ms):
global waiting_event, waiting_data
waiting_event = event
waiting_data = None
t0 = time.ticks_ms()
while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
if isinstance(event, int):
if last_event == event:
break
elif event():
break
if waiting_data:
return True
machine.idle()
return False
# Acting in peripheral role.
@ -72,9 +79,9 @@ def instance0():
ble.gap_advertise(20_000)
# Wait for central to connect, then wait for it to disconnect.
wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
wait_for_event(_IRQ_CENTRAL_DISCONNECT, 4 * TIMEOUT_MS)
if last_event != _IRQ_CENTRAL_DISCONNECT:
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
if not wait_for_event(_IRQ_CENTRAL_DISCONNECT, 4 * TIMEOUT_MS):
return
finally:
ble.active(0)
@ -91,10 +98,9 @@ def instance1():
# Connect to peripheral.
print("gap_connect")
ble.gap_connect(0, BDADDR)
wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_CONNECT:
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = last_data
conn_handle, _, _ = waiting_data
if iteration == 0:
print("gattc_discover_characteristics")
@ -111,8 +117,7 @@ def instance1():
# Disconnect from peripheral.
print("gap_disconnect:", ble.gap_disconnect(conn_handle))
wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_DISCONNECT:
if not wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS):
return
finally:
ble.active(0)

86
tests/multi_bluetooth/ble_gatt_data_transfer.py
View File

@ -5,16 +5,19 @@ import time, machine, bluetooth
TIMEOUT_MS = 5000
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_GATTS_WRITE = const(1 << 2)
_IRQ_PERIPHERAL_CONNECT = const(1 << 6)
_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7)
_IRQ_GATTC_SERVICE_RESULT = const(1 << 8)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(1 << 9)
_IRQ_GATTC_READ_RESULT = const(1 << 11)
_IRQ_GATTC_WRITE_STATUS = const(1 << 12)
_IRQ_GATTC_NOTIFY = const(1 << 13)
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_GATTS_WRITE = const(3)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_SERVICE_RESULT = const(9)
_IRQ_GATTC_SERVICE_DONE = const(10)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
_IRQ_GATTC_CHARACTERISTIC_DONE = const(12)
_IRQ_GATTC_READ_RESULT = const(15)
_IRQ_GATTC_READ_DONE = const(16)
_IRQ_GATTC_WRITE_DONE = const(17)
_IRQ_GATTC_NOTIFY = const(18)
SERVICE_UUID = bluetooth.UUID("00000001-1111-2222-3333-444444444444")
CHAR_CTRL_UUID = bluetooth.UUID("00000002-1111-2222-3333-444444444444")
@ -26,17 +29,15 @@ CHAR_TX = (CHAR_TX_UUID, bluetooth.FLAG_NOTIFY)
SERVICE = (SERVICE_UUID, (CHAR_CTRL, CHAR_RX, CHAR_TX))
SERVICES = (SERVICE,)
last_event = None
last_data = None
waiting_event = None
waiting_data = None
ctrl_value_handle = 0
rx_value_handle = 0
tx_value_handle = 0
def irq(event, data):
global last_event, last_data, ctrl_value_handle, rx_value_handle, tx_value_handle
last_event = event
last_data = data
global waiting_event, waiting_data, ctrl_value_handle, rx_value_handle, tx_value_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
elif event == _IRQ_CENTRAL_DISCONNECT:
@ -49,31 +50,44 @@ def irq(event, data):
print("_IRQ_PERIPHERAL_DISCONNECT")
elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
if data[-1] == CHAR_CTRL_UUID:
print("_IRQ_GATTC_SERVICE_RESULT", data[-1])
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
ctrl_value_handle = data[2]
elif data[-1] == CHAR_RX_UUID:
print("_IRQ_GATTC_SERVICE_RESULT", data[-1])
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
rx_value_handle = data[2]
elif data[-1] == CHAR_TX_UUID:
print("_IRQ_GATTC_SERVICE_RESULT", data[-1])
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
tx_value_handle = data[2]
elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
print("_IRQ_GATTC_CHARACTERISTIC_DONE")
elif event == _IRQ_GATTC_READ_RESULT:
print("_IRQ_GATTC_READ_RESULT", data[-1])
elif event == _IRQ_GATTC_WRITE_STATUS:
print("_IRQ_GATTC_WRITE_STATUS", data[-1])
elif event == _IRQ_GATTC_READ_DONE:
print("_IRQ_GATTC_READ_DONE", data[-1])
elif event == _IRQ_GATTC_WRITE_DONE:
print("_IRQ_GATTC_WRITE_DONE", data[-1])
elif event == _IRQ_GATTC_NOTIFY:
print("_IRQ_GATTC_NOTIFY", data[-1])
if waiting_event is not None:
if (isinstance(waiting_event, int) and event == waiting_event) or (
not isinstance(waiting_event, int) and waiting_event(event, data)
):
waiting_event = None
waiting_data = data
def wait_for_event(event, timeout_ms):
global waiting_event, waiting_data
waiting_event = event
waiting_data = None
t0 = time.ticks_ms()
while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
if isinstance(event, int):
if last_event == event:
break
elif event():
break
if waiting_data:
return True
machine.idle()
return False
# Acting in peripheral role.
@ -89,14 +103,13 @@ def instance0():
multitest.next()
try:
# Wait for central to connect to us.
wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_CENTRAL_CONNECT:
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = last_data
conn_handle, _, _ = waiting_data
# Wait for the central to signal that it's done with its part of the test.
wait_for_event(
lambda: last_event == _IRQ_GATTS_WRITE and last_data[1] == char_ctrl_handle,
lambda event, data: event == _IRQ_GATTS_WRITE and data[1] == char_ctrl_handle,
2 * TIMEOUT_MS,
)
@ -125,32 +138,33 @@ def instance1():
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(0, BDADDR)
wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_CONNECT:
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = last_data
conn_handle, _, _ = waiting_data
# Discover characteristics.
ble.gattc_discover_characteristics(conn_handle, 1, 65535)
wait_for_event(
lambda: ctrl_value_handle and rx_value_handle and tx_value_handle, TIMEOUT_MS
lambda event, data: ctrl_value_handle and rx_value_handle and tx_value_handle,
TIMEOUT_MS,
)
wait_for_event(_IRQ_GATTC_CHARACTERISTIC_DONE, TIMEOUT_MS)
# Write to the characteristic a few times, with and without response.
for i in range(4):
print("gattc_write")
ble.gattc_write(conn_handle, rx_value_handle, "central{}".format(i), i & 1)
if i & 1:
wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
time.sleep_ms(400)
# Write to say that we are done with our part of the test.
ble.gattc_write(conn_handle, ctrl_value_handle, "OK", 1)
wait_for_event(_IRQ_GATTC_WRITE_STATUS, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
# Wait for notification that peripheral is done with its part of the test.
wait_for_event(
lambda: last_event == _IRQ_GATTC_NOTIFY and last_data[1] == ctrl_value_handle,
lambda event, data: event == _IRQ_GATTC_NOTIFY and data[1] == ctrl_value_handle,
TIMEOUT_MS,
)

13
tests/multi_bluetooth/ble_gatt_data_transfer.py.exp
View File

@ -11,16 +11,17 @@ _IRQ_CENTRAL_DISCONNECT
--- instance1 ---
gap_connect
_IRQ_PERIPHERAL_CONNECT
_IRQ_GATTC_SERVICE_RESULT UUID128('00000002-1111-2222-3333-444444444444')
_IRQ_GATTC_SERVICE_RESULT UUID128('00000003-1111-2222-3333-444444444444')
_IRQ_GATTC_SERVICE_RESULT UUID128('00000004-1111-2222-3333-444444444444')
_IRQ_GATTC_CHARACTERISTIC_RESULT UUID128('00000002-1111-2222-3333-444444444444')
_IRQ_GATTC_CHARACTERISTIC_RESULT UUID128('00000003-1111-2222-3333-444444444444')
_IRQ_GATTC_CHARACTERISTIC_RESULT UUID128('00000004-1111-2222-3333-444444444444')
_IRQ_GATTC_CHARACTERISTIC_DONE
gattc_write
gattc_write
_IRQ_GATTC_WRITE_STATUS 0
_IRQ_GATTC_WRITE_DONE 0
gattc_write
gattc_write
_IRQ_GATTC_WRITE_STATUS 0
_IRQ_GATTC_WRITE_STATUS 0
_IRQ_GATTC_WRITE_DONE 0
_IRQ_GATTC_WRITE_DONE 0
_IRQ_GATTC_NOTIFY b'message0'
_IRQ_GATTC_NOTIFY b'message1'
_IRQ_GATTC_NOTIFY b'message2'

45
tests/multi_bluetooth/ble_gattc_discover_services.py
View File

@ -5,11 +5,12 @@ import time, machine, bluetooth
TIMEOUT_MS = 5000
_IRQ_CENTRAL_CONNECT = const(1 << 0)
_IRQ_CENTRAL_DISCONNECT = const(1 << 1)
_IRQ_PERIPHERAL_CONNECT = const(1 << 6)
_IRQ_PERIPHERAL_DISCONNECT = const(1 << 7)
_IRQ_GATTC_SERVICE_RESULT = const(1 << 8)
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_SERVICE_RESULT = const(9)
_IRQ_GATTC_SERVICE_DONE = const(10)
UUID_A = bluetooth.UUID(0x180D)
UUID_B = bluetooth.UUID("A5A5A5A5-FFFF-9999-1111-5A5A5A5A5A5A")
@ -23,15 +24,13 @@ SERVICE_B = (
)
SERVICES = (SERVICE_A, SERVICE_B)
last_event = None
last_data = None
waiting_event = None
waiting_data = None
num_service_result = 0
def irq(event, data):
global last_event, last_data, num_service_result
last_event = event
last_data = data
global waiting_event, waiting_data, num_service_result
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
elif event == _IRQ_CENTRAL_DISCONNECT:
@ -45,16 +44,25 @@ def irq(event, data):
print("_IRQ_GATTC_SERVICE_RESULT", data[3])
num_service_result += 1
if waiting_event is not None:
if (isinstance(waiting_event, int) and event == waiting_event) or (
not isinstance(waiting_event, int) and waiting_event(event, data)
):
waiting_event = None
waiting_data = data
def wait_for_event(event, timeout_ms):
global waiting_event, waiting_data
waiting_event = event
waiting_data = None
t0 = time.ticks_ms()
while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
if isinstance(event, int):
if last_event == event:
break
elif event():
break
if waiting_data:
return True
machine.idle()
return False
# Acting in peripheral role.
@ -78,14 +86,13 @@ def instance1():
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(0, BDADDR)
wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if last_event != _IRQ_PERIPHERAL_CONNECT:
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = last_data
conn_handle, _, _ = waiting_data
# Discover services.
ble.gattc_discover_services(conn_handle)
wait_for_event(lambda: num_service_result == 2, TIMEOUT_MS)
wait_for_event(lambda event, data: num_service_result == 2, TIMEOUT_MS)
# Disconnect from peripheral.
print("gap_disconnect:", ble.gap_disconnect(conn_handle))

2
tests/run-multitests.py
View File

@ -289,7 +289,7 @@ def run_test_on_instances(test_file, num_instances, instances):
continue
num_output += 1
last_read_time[idx] = time.time()
if out is not None:
if out is not None and not any(m in out for m in IGNORE_OUTPUT_MATCHES):
trace_instance_output(idx, out)
output[idx].append(out)
if err is not None:

Write Preview
Loading…
Cancel
Save