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: Improve reliability of event waiting. 

Use the same `wait_for_event` in all tests that doesn't hold a reference to
the event data tuple and handles repeat events.

Also fix a few misc reliability issues around timeouts and sequencing.

Signed-off-by: Jim Mussared <jim.mussared@gmail.com>
pull/6596/head
Jim Mussared 4 years ago
committed by Damien George
parent
309fb822e6
commit
ccfd535af4
10 changed files with 155 additions and 221 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. 91
      tests/multi_bluetooth/ble_characteristic.py
  2. 1
      tests/multi_bluetooth/ble_characteristic.py.exp
  3. 5
      tests/multi_bluetooth/ble_gap_advertise.py
  4. 70
      tests/multi_bluetooth/ble_gap_connect.py
  5. 53
      tests/multi_bluetooth/ble_gap_device_name.py
  6. 1
      tests/multi_bluetooth/ble_gap_device_name.py.exp
  7. 62
      tests/multi_bluetooth/ble_gatt_data_transfer.py
  8. 37
      tests/multi_bluetooth/ble_gattc_discover_services.py
  9. 2
      tests/multi_bluetooth/ble_gattc_discover_services.py.exp
  10. 54
      tests/multi_bluetooth/ble_mtu.py

91
tests/multi_bluetooth/ble_characteristic.py
View File

@ -31,28 +31,31 @@ SERVICE = (
)
SERVICES = (SERVICE,)
waiting_event = None
waiting_data = None
value_handle = 0
waiting_events = {}
def irq(event, data):
global waiting_event, waiting_data, value_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_CENTRAL_DISCONNECT:
print("_IRQ_CENTRAL_DISCONNECT")
elif event == _IRQ_GATTS_WRITE:
print("_IRQ_GATTS_WRITE", ble.gatts_read(data[-1]))
elif event == _IRQ_PERIPHERAL_CONNECT:
print("_IRQ_PERIPHERAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_PERIPHERAL_DISCONNECT:
print("_IRQ_PERIPHERAL_DISCONNECT")
elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
# conn_handle, def_handle, value_handle, properties, uuid = data
if data[-1] == CHAR_UUID:
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
value_handle = data[2]
waiting_events[event] = data[2]
else:
return
elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
print("_IRQ_GATTC_CHARACTERISTIC_DONE")
elif event == _IRQ_GATTC_READ_RESULT:
print("_IRQ_GATTC_READ_RESULT", bytes(data[-1]))
elif event == _IRQ_GATTC_READ_DONE:
@ -66,25 +69,19 @@ def irq(event, data):
elif event == _IRQ_GATTS_INDICATE_DONE:
print("_IRQ_GATTS_INDICATE_DONE", 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
if event not in waiting_events:
waiting_events[event] = None
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 waiting_data:
return True
if event in waiting_events:
result = waiting_events[event]
del waiting_events[event]
return result
machine.idle()
return False
raise ValueError("Timeout waiting for {}".format(event))
# Acting in peripheral role.
@ -99,39 +96,35 @@ def instance0():
ble.gatts_write(char_handle, "periph0")
# Wait for central to connect to us.
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = waiting_data
conn_handle = wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
# Wait for a write to the characteristic from the central.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS)
# A
# Wait a bit, then write the characteristic and notify it.
time.sleep_ms(1000)
# Wait for a write to the characteristic from the central,
# then reply with a notification.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS)
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.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS)
# B
# Wait a bit, then notify a new value on the characteristic.
time.sleep_ms(1000)
# Wait for a write to the characteristic from the central,
# then reply with value-included notification.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS)
print("gatts_notify")
ble.gatts_notify(conn_handle, char_handle, "periph2")
# Wait for a write to the characteristic from the central.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS)
# C
# Wait a bit, then notify a new value on the characteristic.
time.sleep_ms(1000)
# Wait for a write to the characteristic from the central,
# then reply with an indication.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS)
print("gatts_write")
ble.gatts_write(char_handle, "periph3")
print("gatts_indicate")
ble.gatts_indicate(conn_handle, char_handle)
# Wait for the indicate ack.
wait_for_event(_IRQ_GATTS_INDICATE_DONE, TIMEOUT_MS)
# Wait for the central to disconnect.
@ -147,49 +140,45 @@ def instance1():
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(*BDADDR)
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = waiting_data
conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
# Discover characteristics.
ble.gattc_discover_characteristics(conn_handle, 1, 65535)
wait_for_event(lambda event, data: value_handle, TIMEOUT_MS)
value_handle = wait_for_event(_IRQ_GATTC_CHARACTERISTIC_RESULT, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_CHARACTERISTIC_DONE, TIMEOUT_MS)
# Issue read of characteristic, should get initial value.
print("gattc_read")
ble.gattc_read(conn_handle, value_handle)
wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS)
# Write to the characteristic, and ask for a response.
# Write to the characteristic, which will trigger a notification.
print("gattc_write")
ble.gattc_write(conn_handle, value_handle, "central0", 1)
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
# Wait for a notify, then read new value.
# A
wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS)
print("gattc_read")
print("gattc_read") # Read the new value set immediately before notification.
ble.gattc_read(conn_handle, value_handle)
wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS)
# Write to the characteristic, and ask for a response.
# Write to the characteristic, which will trigger a value-included notification.
print("gattc_write")
ble.gattc_write(conn_handle, value_handle, "central1", 1)
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
# Wait for a notify (should have new data), then read old value (should be unchanged).
# B
wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS)
print("gattc_read")
print("gattc_read") # Read value should be unchanged.
ble.gattc_read(conn_handle, value_handle)
wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS)
# Write to the characteristic, and ask for a response.
# Write to the characteristic, which will trigger an indication.
print("gattc_write")
ble.gattc_write(conn_handle, value_handle, "central2", 1)
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
# Wait for a indicate (should have new data), then read new value.
# C
wait_for_event(_IRQ_GATTC_INDICATE, TIMEOUT_MS)
print("gattc_read")
print("gattc_read") # Read the new value set immediately before indication.
ble.gattc_read(conn_handle, value_handle)
wait_for_event(_IRQ_GATTC_READ_RESULT, TIMEOUT_MS)

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

@ -15,6 +15,7 @@ _IRQ_CENTRAL_DISCONNECT
gap_connect
_IRQ_PERIPHERAL_CONNECT
_IRQ_GATTC_CHARACTERISTIC_RESULT UUID('00000000-1111-2222-3333-444444444444')
_IRQ_GATTC_CHARACTERISTIC_DONE
gattc_read
_IRQ_GATTC_READ_RESULT b'periph0'
_IRQ_GATTC_READ_DONE 0

5
tests/multi_bluetooth/ble_gap_advertise.py
View File

@ -46,7 +46,10 @@ def instance1():
elif ev == _IRQ_SCAN_DONE:
finished = True
ble.config(rxbuf=2000)
try:
ble.config(rxbuf=2000)
except:
pass
ble.irq(irq)
ble.gap_scan(2 * ADV_TIME_S * 1000, 10000, 10000)
while not finished:

70
tests/multi_bluetooth/ble_gap_connect.py
View File

@ -10,101 +10,77 @@ _IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
central_connected = False
central_disconnected = False
peripheral_connected = False
peripheral_disconnected = False
conn_handle = None
waiting_events = {}
def irq(event, data):
global central_connected, central_disconnected, peripheral_connected, peripheral_disconnected, conn_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
central_connected = True
conn_handle = data[0]
waiting_events[event] = data[0]
elif event == _IRQ_CENTRAL_DISCONNECT:
print("_IRQ_CENTRAL_DISCONNECT")
central_disconnected = True
elif event == _IRQ_PERIPHERAL_CONNECT:
print("_IRQ_PERIPHERAL_CONNECT")
peripheral_connected = True
conn_handle = data[0]
waiting_events[event] = data[0]
elif event == _IRQ_PERIPHERAL_DISCONNECT:
print("_IRQ_PERIPHERAL_DISCONNECT")
peripheral_disconnected = True
remote_addr = data[0]
if event not in waiting_events:
waiting_events[event] = None
def wait_for(fn, timeout_ms):
def wait_for_event(event, timeout_ms):
t0 = time.ticks_ms()
while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
if fn():
return True
if event in waiting_events:
result = waiting_events[event]
del waiting_events[event]
return result
machine.idle()
return False
raise ValueError("Timeout waiting for {}".format(event))
# Acting in peripheral role.
def instance0():
global central_connected, central_disconnected
multitest.globals(BDADDR=ble.config("mac"))
print("gap_advertise")
ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
multitest.next()
try:
# Wait for central to connect, then wait for it to disconnect.
if not wait_for(lambda: central_connected, TIMEOUT_MS):
return
if not wait_for(lambda: central_disconnected, TIMEOUT_MS):
return
central_connected = False
central_disconnected = False
wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS)
# Start advertising again.
print("gap_advertise")
ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
# Wait for central to connect, then disconnect it.
if not wait_for(lambda: central_connected, TIMEOUT_MS):
return
conn_handle = wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
print("gap_disconnect:", ble.gap_disconnect(conn_handle))
if not wait_for(lambda: central_disconnected, TIMEOUT_MS):
return
wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS)
finally:
ble.active(0)
# Acting in central role.
def instance1():
global peripheral_connected, peripheral_disconnected
multitest.next()
try:
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(*BDADDR)
if not wait_for(lambda: peripheral_connected, TIMEOUT_MS):
return
conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
print("gap_disconnect:", ble.gap_disconnect(conn_handle))
if not wait_for(lambda: peripheral_disconnected, TIMEOUT_MS):
return
peripheral_connected = False
peripheral_disconnected = False
# Wait for peripheral to start advertising again.
time.sleep_ms(100)
wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)
# Connect to peripheral and then let the peripheral disconnect us.
# Extra scan timeout allows for the peripheral to receive the disconnect
# event and start advertising again.
print("gap_connect")
ble.gap_connect(*BDADDR)
if not wait_for(lambda: peripheral_connected, TIMEOUT_MS):
return
if not wait_for(lambda: peripheral_disconnected, TIMEOUT_MS):
return
ble.gap_connect(BDADDR[0], BDADDR[1], 5000)
wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)
finally:
ble.active(0)

53
tests/multi_bluetooth/ble_gap_device_name.py
View File

@ -16,45 +16,44 @@ _IRQ_GATTC_READ_DONE = const(16)
GAP_DEVICE_NAME_UUID = bluetooth.UUID(0x2A00)
waiting_event = None
waiting_data = None
value_handle = 0
waiting_events = {}
def irq(event, data):
global waiting_event, waiting_data, value_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_CENTRAL_DISCONNECT:
print("_IRQ_CENTRAL_DISCONNECT")
elif event == _IRQ_PERIPHERAL_CONNECT:
print("_IRQ_PERIPHERAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_PERIPHERAL_DISCONNECT:
print("_IRQ_PERIPHERAL_DISCONNECT")
elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
if data[-1] == GAP_DEVICE_NAME_UUID:
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
value_handle = data[2]
waiting_events[event] = data[2]
else:
return
elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
print("_IRQ_GATTC_CHARACTERISTIC_DONE")
elif event == _IRQ_GATTC_READ_RESULT:
print("_IRQ_GATTC_READ_RESULT", bytes(data[-1]))
if waiting_event is not None:
if isinstance(waiting_event, int) and event == waiting_event:
waiting_event = None
waiting_data = data
if event not in waiting_events:
waiting_events[event] = None
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 waiting_data:
return True
if event in waiting_events:
result = waiting_events[event]
del waiting_events[event]
return result
machine.idle()
return False
raise ValueError("Timeout waiting for {}".format(event))
# Acting in peripheral role.
@ -79,10 +78,8 @@ def instance0():
ble.gap_advertise(20_000)
# Wait for central to connect, then wait for it to disconnect.
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
if not wait_for_event(_IRQ_CENTRAL_DISCONNECT, 4 * TIMEOUT_MS):
return
wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
wait_for_event(_IRQ_CENTRAL_DISCONNECT, 4 * TIMEOUT_MS)
finally:
ble.active(0)
@ -91,6 +88,7 @@ def instance0():
def instance1():
multitest.next()
try:
value_handle = None
for iteration in range(2):
# Wait for peripheral to start advertising.
time.sleep_ms(500)
@ -98,17 +96,15 @@ def instance1():
# Connect to peripheral.
print("gap_connect")
ble.gap_connect(*BDADDR)
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = waiting_data
conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if iteration == 0:
# Only do characteristic discovery on the first iteration,
# assume value_handle is unchanged on the second.
print("gattc_discover_characteristics")
ble.gattc_discover_characteristics(conn_handle, 1, 65535)
wait_for_event(lambda: value_handle, TIMEOUT_MS)
# Wait for all characteristic results to come in (there should be an event for it).
time.sleep_ms(500)
value_handle = wait_for_event(_IRQ_GATTC_CHARACTERISTIC_RESULT, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_CHARACTERISTIC_DONE, TIMEOUT_MS)
# Read the peripheral's GAP device name.
print("gattc_read")
@ -117,8 +113,7 @@ def instance1():
# Disconnect from peripheral.
print("gap_disconnect:", ble.gap_disconnect(conn_handle))
if not wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS):
return
wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)
finally:
ble.active(0)

1
tests/multi_bluetooth/ble_gap_device_name.py.exp
View File

@ -12,6 +12,7 @@ gap_connect
_IRQ_PERIPHERAL_CONNECT
gattc_discover_characteristics
_IRQ_GATTC_CHARACTERISTIC_RESULT UUID(0x2a00)
_IRQ_GATTC_CHARACTERISTIC_DONE
gattc_read
_IRQ_GATTC_READ_RESULT b'GAP_NAME0'
gap_disconnect: True

62
tests/multi_bluetooth/ble_gatt_data_transfer.py
View File

@ -29,35 +29,33 @@ CHAR_TX = (CHAR_TX_UUID, bluetooth.FLAG_NOTIFY)
SERVICE = (SERVICE_UUID, (CHAR_CTRL, CHAR_RX, CHAR_TX))
SERVICES = (SERVICE,)
waiting_event = None
waiting_data = None
ctrl_value_handle = 0
rx_value_handle = 0
tx_value_handle = 0
waiting_events = {}
def irq(event, data):
global waiting_event, waiting_data, ctrl_value_handle, rx_value_handle, tx_value_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_CENTRAL_DISCONNECT:
print("_IRQ_CENTRAL_DISCONNECT")
elif event == _IRQ_GATTS_WRITE:
print("_IRQ_GATTS_WRITE")
waiting_events[(event, data[1])] = None
elif event == _IRQ_PERIPHERAL_CONNECT:
print("_IRQ_PERIPHERAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_PERIPHERAL_DISCONNECT:
print("_IRQ_PERIPHERAL_DISCONNECT")
elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
if data[-1] == CHAR_CTRL_UUID:
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
ctrl_value_handle = data[2]
waiting_events[(event, CHAR_CTRL_UUID)] = data[2]
elif data[-1] == CHAR_RX_UUID:
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
rx_value_handle = data[2]
waiting_events[(event, CHAR_RX_UUID)] = data[2]
elif data[-1] == CHAR_TX_UUID:
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
tx_value_handle = data[2]
waiting_events[(event, CHAR_TX_UUID)] = data[2]
elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
print("_IRQ_GATTC_CHARACTERISTIC_DONE")
elif event == _IRQ_GATTC_READ_RESULT:
@ -68,26 +66,21 @@ def irq(event, data):
print("_IRQ_GATTC_WRITE_DONE", data[-1])
elif event == _IRQ_GATTC_NOTIFY:
print("_IRQ_GATTC_NOTIFY", bytes(data[-1]))
waiting_events[(event, data[1])] = None
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
if event not in waiting_events:
waiting_events[event] = None
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 waiting_data:
return True
if event in waiting_events:
result = waiting_events[event]
del waiting_events[event]
return result
machine.idle()
return False
raise ValueError("Timeout waiting for {}".format(event))
# Acting in peripheral role.
@ -103,15 +96,10 @@ def instance0():
multitest.next()
try:
# Wait for central to connect to us.
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = waiting_data
conn_handle = wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
# Wait for the central to signal that it's done with its part of the test.
wait_for_event(
lambda event, data: event == _IRQ_GATTS_WRITE and data[1] == char_ctrl_handle,
2 * TIMEOUT_MS,
)
wait_for_event((_IRQ_GATTS_WRITE, char_ctrl_handle), 2 * TIMEOUT_MS)
# Read all accumulated data from the central.
print("gatts_read:", ble.gatts_read(char_rx_handle))
@ -138,17 +126,14 @@ def instance1():
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(*BDADDR)
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = waiting_data
conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
# Discover characteristics.
ble.gattc_discover_characteristics(conn_handle, 1, 65535)
wait_for_event(
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)
ctrl_value_handle = waiting_events[(_IRQ_GATTC_CHARACTERISTIC_RESULT, CHAR_CTRL_UUID)]
rx_value_handle = waiting_events[(_IRQ_GATTC_CHARACTERISTIC_RESULT, CHAR_RX_UUID)]
tx_value_handle = waiting_events[(_IRQ_GATTC_CHARACTERISTIC_RESULT, CHAR_TX_UUID)]
# Write to the characteristic a few times, with and without response.
for i in range(4):
@ -163,10 +148,7 @@ def instance1():
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 event, data: event == _IRQ_GATTC_NOTIFY and data[1] == ctrl_value_handle,
TIMEOUT_MS,
)
wait_for_event((_IRQ_GATTC_NOTIFY, ctrl_value_handle), TIMEOUT_MS)
# Disconnect from peripheral.
print("gap_disconnect:", ble.gap_disconnect(conn_handle))

37
tests/multi_bluetooth/ble_gattc_discover_services.py
View File

@ -24,45 +24,42 @@ SERVICE_B = (
)
SERVICES = (SERVICE_A, SERVICE_B)
waiting_event = None
waiting_data = None
waiting_events = {}
num_service_result = 0
def irq(event, data):
global waiting_event, waiting_data, num_service_result
global num_service_result
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_CENTRAL_DISCONNECT:
print("_IRQ_CENTRAL_DISCONNECT")
elif event == _IRQ_PERIPHERAL_CONNECT:
print("_IRQ_PERIPHERAL_CONNECT")
waiting_events[event] = data[0]
elif event == _IRQ_PERIPHERAL_DISCONNECT:
print("_IRQ_PERIPHERAL_DISCONNECT")
elif event == _IRQ_GATTC_SERVICE_RESULT:
if data[3] == UUID_A or data[3] == UUID_B:
print("_IRQ_GATTC_SERVICE_RESULT", data[3])
num_service_result += 1
elif event == _IRQ_GATTC_SERVICE_DONE:
print("_IRQ_GATTC_SERVICE_DONE")
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
if event not in waiting_events:
waiting_events[event] = None
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 waiting_data:
return True
if event in waiting_events:
result = waiting_events[event]
del waiting_events[event]
return result
machine.idle()
return False
raise ValueError("Timeout waiting for {}".format(event))
# Acting in peripheral role.
@ -86,13 +83,13 @@ def instance1():
# Connect to peripheral and then disconnect.
print("gap_connect")
ble.gap_connect(*BDADDR)
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle, _, _ = waiting_data
conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
# Discover services.
ble.gattc_discover_services(conn_handle)
wait_for_event(lambda event, data: num_service_result == 2, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_SERVICE_DONE, TIMEOUT_MS)
print("discovered:", num_service_result)
# Disconnect from peripheral.
print("gap_disconnect:", ble.gap_disconnect(conn_handle))

2
tests/multi_bluetooth/ble_gattc_discover_services.py.exp
View File

@ -7,5 +7,7 @@ gap_connect
_IRQ_PERIPHERAL_CONNECT
_IRQ_GATTC_SERVICE_RESULT UUID(0x180d)
_IRQ_GATTC_SERVICE_RESULT UUID('a5a5a5a5-ffff-9999-1111-5a5a5a5a5a5a')
_IRQ_GATTC_SERVICE_DONE
discovered: 2
gap_disconnect: True
_IRQ_PERIPHERAL_DISCONNECT

54
tests/multi_bluetooth/ble_mtu.py
View File

@ -51,7 +51,6 @@ waiting_events = {}
def irq(event, data):
global value_handle
if event == _IRQ_CENTRAL_CONNECT:
print("_IRQ_CENTRAL_CONNECT")
waiting_events[event] = data[0]
@ -68,6 +67,8 @@ def irq(event, data):
if data[-1] == CHAR_UUID:
print("_IRQ_GATTC_CHARACTERISTIC_RESULT", data[-1])
waiting_events[event] = data[2]
else:
return
elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
print("_IRQ_GATTC_CHARACTERISTIC_DONE")
elif event == _IRQ_GATTC_WRITE_DONE:
@ -86,9 +87,11 @@ def wait_for_event(event, timeout_ms):
t0 = time.ticks_ms()
while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
if event in waiting_events:
return True
result = waiting_events[event]
del waiting_events[event]
return result
machine.idle()
return False
raise ValueError("Timeout waiting for {}".format(event))
# Acting in peripheral role.
@ -101,8 +104,6 @@ def instance0():
multitest.next()
try:
for i in range(7):
waiting_events.clear()
if i == 1:
ble.config(mtu=200)
elif i == 2:
@ -116,31 +117,26 @@ def instance0():
ble.config(mtu=256)
# Wait for central to connect to us.
if not wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS):
return
conn_handle = waiting_events[_IRQ_CENTRAL_CONNECT]
conn_handle = wait_for_event(_IRQ_CENTRAL_CONNECT, TIMEOUT_MS)
if i >= 4:
print("gattc_exchange_mtu")
ble.gattc_exchange_mtu(conn_handle)
if not wait_for_event(_IRQ_MTU_EXCHANGED, TIMEOUT_MS):
return
mtu = waiting_events[_IRQ_MTU_EXCHANGED]
mtu = wait_for_event(_IRQ_MTU_EXCHANGED, TIMEOUT_MS)
print("gatts_notify")
ble.gatts_notify(conn_handle, char_handle, str(i) * 64)
if not wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS):
return
# Extra timeout while client does service discovery.
wait_for_event(_IRQ_GATTS_WRITE, TIMEOUT_MS * 2)
print("gatts_read")
data = ble.gatts_read(char_handle)
print("characteristic len:", len(data), chr(data[0]))
# Wait for the central to disconnect.
if not wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS):
return
wait_for_event(_IRQ_CENTRAL_DISCONNECT, TIMEOUT_MS)
print("gap_advertise")
ble.gap_advertise(20_000, b"\x02\x01\x06\x04\xffMPY")
@ -154,8 +150,6 @@ def instance1():
multitest.next()
try:
for i in range(7):
waiting_events.clear()
if i < 4:
ble.config(mtu=300)
elif i == 5:
@ -166,39 +160,33 @@ def instance1():
ble.config(mtu=256)
# Connect to peripheral and then disconnect.
# Extra scan timeout allows for the peripheral to receive the previous disconnect
# event and start advertising again.
print("gap_connect")
ble.gap_connect(*BDADDR)
if not wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS):
return
conn_handle = waiting_events[_IRQ_PERIPHERAL_CONNECT]
ble.gap_connect(BDADDR[0], BDADDR[1], 5000)
conn_handle = wait_for_event(_IRQ_PERIPHERAL_CONNECT, TIMEOUT_MS)
if i < 4:
print("gattc_exchange_mtu")
ble.gattc_exchange_mtu(conn_handle)
if not wait_for_event(_IRQ_MTU_EXCHANGED, TIMEOUT_MS):
return
mtu = waiting_events[_IRQ_MTU_EXCHANGED]
mtu = wait_for_event(_IRQ_MTU_EXCHANGED, TIMEOUT_MS)
if not wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS):
return
wait_for_event(_IRQ_GATTC_NOTIFY, TIMEOUT_MS)
print("gattc_discover_characteristics")
ble.gattc_discover_characteristics(conn_handle, 1, 65535)
if not wait_for_event(_IRQ_GATTC_CHARACTERISTIC_DONE, TIMEOUT_MS):
return
value_handle = waiting_events[_IRQ_GATTC_CHARACTERISTIC_RESULT]
value_handle = wait_for_event(_IRQ_GATTC_CHARACTERISTIC_RESULT, TIMEOUT_MS)
wait_for_event(_IRQ_GATTC_CHARACTERISTIC_DONE, TIMEOUT_MS)
# Write 20 more than the MTU to test truncation.
print("gattc_write")
ble.gattc_write(conn_handle, value_handle, chr(ord("a") + i) * (mtu + 20), 1)
if not wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS):
return
wait_for_event(_IRQ_GATTC_WRITE_DONE, TIMEOUT_MS)
# Disconnect from peripheral.
print("gap_disconnect:", ble.gap_disconnect(conn_handle))
if not wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS):
return
wait_for_event(_IRQ_PERIPHERAL_DISCONNECT, TIMEOUT_MS)
finally:
ble.active(0)

Write Preview
Loading…
Cancel
Save