mirror of
https://github.com/openmv/openmv.git
synced 2025-11-04 14:49:50 +08:00
Merge pull request #1968 from openmv/examples_update
scripts/examples: Update WiFi and Bluetooth examples.
This commit is contained in:
commit
6d1c4cd4fb
@ -3,6 +3,7 @@
|
|||||||
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net.
|
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net.
|
||||||
|
|
||||||
import network
|
import network
|
||||||
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
|
|||||||
@ -3,7 +3,7 @@
|
|||||||
# This example shows how to get the IP address for websites via DNS.
|
# This example shows how to get the IP address for websites via DNS.
|
||||||
|
|
||||||
import network
|
import network
|
||||||
import usocket
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
|
|||||||
@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
|
import time
|
||||||
|
|
||||||
# AP info
|
# AP info
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
|
|||||||
@ -1,7 +1,8 @@
|
|||||||
# Simple HTTPS client example.
|
# Simple HTTPS client example.
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
import ussl
|
import ssl
|
||||||
|
import time
|
||||||
|
|
||||||
# AP info
|
# AP info
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
@ -34,7 +35,7 @@ client.connect(addr)
|
|||||||
# Set timeout
|
# Set timeout
|
||||||
client.settimeout(3.0)
|
client.settimeout(3.0)
|
||||||
|
|
||||||
client = ussl.wrap_socket(client, server_hostname=HOST)
|
client = ssl.wrap_socket(client, server_hostname=HOST)
|
||||||
|
|
||||||
# Send HTTP request and recv response
|
# Send HTTP request and recv response
|
||||||
request = "GET / HTTP/1.1\r\n"
|
request = "GET / HTTP/1.1\r\n"
|
||||||
|
|||||||
@ -4,8 +4,8 @@
|
|||||||
|
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
import ustruct
|
import struct
|
||||||
import utime
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
@ -36,5 +36,5 @@ client.sendto("\x1b" + 47 * "\0", addr)
|
|||||||
data, address = client.recvfrom(1024)
|
data, address = client.recvfrom(1024)
|
||||||
|
|
||||||
# Print time
|
# Print time
|
||||||
t = ustruct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
||||||
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (utime.localtime(t)[0:6]))
|
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))
|
||||||
|
|||||||
@ -4,8 +4,8 @@
|
|||||||
|
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
import ustruct
|
import struct
|
||||||
import utime
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
@ -35,5 +35,5 @@ client.sendto("\x1b" + 47 * "\0", addr)
|
|||||||
data, address = client.recvfrom(1024)
|
data, address = client.recvfrom(1024)
|
||||||
|
|
||||||
# Print time
|
# Print time
|
||||||
t = ustruct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
||||||
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (utime.localtime(t)[0:6]))
|
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))
|
||||||
|
|||||||
@ -0,0 +1,61 @@
|
|||||||
|
# Bluetooth Blinky Example
|
||||||
|
#
|
||||||
|
# Use nRFConnect app from the App store, connect to the Nano and write 1/0 to control the LED.
|
||||||
|
|
||||||
|
import bluetooth
|
||||||
|
import time
|
||||||
|
from ble_advertising import advertising_payload
|
||||||
|
from machine import LED
|
||||||
|
from micropython import const
|
||||||
|
|
||||||
|
_IRQ_CENTRAL_CONNECT = const(1)
|
||||||
|
_IRQ_CENTRAL_DISCONNECT = const(2)
|
||||||
|
_IRQ_GATTS_WRITE = const(3)
|
||||||
|
|
||||||
|
_FLAG_READ = const(0x0002)
|
||||||
|
_FLAG_WRITE = const(0x0008)
|
||||||
|
_FLAG_NOTIFY = const(0x0010)
|
||||||
|
_FLAG_INDICATE = const(0x0020)
|
||||||
|
|
||||||
|
_SERVICE_UUID = bluetooth.UUID(0x1523)
|
||||||
|
_LED_CHAR_UUID = (bluetooth.UUID(0x1525), _FLAG_WRITE)
|
||||||
|
_LED_SERVICE = (
|
||||||
|
_SERVICE_UUID,
|
||||||
|
(_LED_CHAR_UUID,),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class BLETemperature:
|
||||||
|
def __init__(self, ble, name="Nicla-Vision"):
|
||||||
|
self._ble = ble
|
||||||
|
self._ble.active(True)
|
||||||
|
self._ble.irq(self._irq)
|
||||||
|
((self._handle,),) = self._ble.gatts_register_services((_LED_SERVICE,))
|
||||||
|
self._connections = set()
|
||||||
|
self._payload = advertising_payload(name=name, services=[_SERVICE_UUID])
|
||||||
|
self._advertise()
|
||||||
|
self.led = LED("LED_BLUE")
|
||||||
|
|
||||||
|
def _irq(self, event, data):
|
||||||
|
# Track connections so we can send notifications.
|
||||||
|
if event == _IRQ_CENTRAL_CONNECT:
|
||||||
|
conn_handle, _, _ = data
|
||||||
|
self._connections.add(conn_handle)
|
||||||
|
elif event == _IRQ_CENTRAL_DISCONNECT:
|
||||||
|
conn_handle, _, _ = data
|
||||||
|
self._connections.remove(conn_handle)
|
||||||
|
# Start advertising again to allow a new connection.
|
||||||
|
self._advertise()
|
||||||
|
elif event == _IRQ_GATTS_WRITE:
|
||||||
|
self.led.value(self._ble.gatts_read(data[-1])[0])
|
||||||
|
|
||||||
|
def _advertise(self, interval_us=500000):
|
||||||
|
self._ble.gap_advertise(interval_us, adv_data=self._payload)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
ble = bluetooth.BLE()
|
||||||
|
temp = BLETemperature(ble)
|
||||||
|
|
||||||
|
while True:
|
||||||
|
time.sleep_ms(1000)
|
||||||
@ -8,7 +8,7 @@ import random
|
|||||||
import struct
|
import struct
|
||||||
import time
|
import time
|
||||||
from ble_advertising import advertising_payload
|
from ble_advertising import advertising_payload
|
||||||
|
from machine import LED
|
||||||
from micropython import const
|
from micropython import const
|
||||||
|
|
||||||
_IRQ_CENTRAL_CONNECT = const(1)
|
_IRQ_CENTRAL_CONNECT = const(1)
|
||||||
@ -36,7 +36,7 @@ _ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
|
|||||||
|
|
||||||
|
|
||||||
class BLETemperature:
|
class BLETemperature:
|
||||||
def __init__(self, ble, name="PORTENTA_BLE"):
|
def __init__(self, ble, name="Nicla-Vision"):
|
||||||
self._ble = ble
|
self._ble = ble
|
||||||
self._ble.active(True)
|
self._ble.active(True)
|
||||||
self._ble.irq(self._irq)
|
self._ble.irq(self._irq)
|
||||||
@ -48,17 +48,20 @@ class BLETemperature:
|
|||||||
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
|
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
|
||||||
)
|
)
|
||||||
self._advertise()
|
self._advertise()
|
||||||
|
self.led = LED("LED_BLUE")
|
||||||
|
|
||||||
def _irq(self, event, data):
|
def _irq(self, event, data):
|
||||||
# Track connections so we can send notifications.
|
# Track connections so we can send notifications.
|
||||||
if event == _IRQ_CENTRAL_CONNECT:
|
if event == _IRQ_CENTRAL_CONNECT:
|
||||||
conn_handle, _, _ = data
|
conn_handle, _, _ = data
|
||||||
self._connections.add(conn_handle)
|
self._connections.add(conn_handle)
|
||||||
|
self.led.on()
|
||||||
elif event == _IRQ_CENTRAL_DISCONNECT:
|
elif event == _IRQ_CENTRAL_DISCONNECT:
|
||||||
conn_handle, _, _ = data
|
conn_handle, _, _ = data
|
||||||
self._connections.remove(conn_handle)
|
self._connections.remove(conn_handle)
|
||||||
# Start advertising again to allow a new connection.
|
# Start advertising again to allow a new connection.
|
||||||
self._advertise()
|
self._advertise()
|
||||||
|
self.led.off()
|
||||||
elif event == _IRQ_GATTS_INDICATE_DONE:
|
elif event == _IRQ_GATTS_INDICATE_DONE:
|
||||||
conn_handle, value_handle, status = data
|
conn_handle, value_handle, status = data
|
||||||
|
|
||||||
@ -79,7 +82,7 @@ class BLETemperature:
|
|||||||
self._ble.gap_advertise(interval_us, adv_data=self._payload)
|
self._ble.gap_advertise(interval_us, adv_data=self._payload)
|
||||||
|
|
||||||
|
|
||||||
def demo():
|
if __name__ == "__main__":
|
||||||
ble = bluetooth.BLE()
|
ble = bluetooth.BLE()
|
||||||
temp = BLETemperature(ble)
|
temp = BLETemperature(ble)
|
||||||
|
|
||||||
@ -93,7 +96,3 @@ def demo():
|
|||||||
# Random walk the temperature.
|
# Random walk the temperature.
|
||||||
t += random.uniform(-0.5, 0.5)
|
t += random.uniform(-0.5, 0.5)
|
||||||
time.sleep_ms(1000)
|
time.sleep_ms(1000)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
demo()
|
|
||||||
|
|||||||
@ -0,0 +1,63 @@
|
|||||||
|
from micropython import const
|
||||||
|
|
||||||
|
import uasyncio as asyncio
|
||||||
|
import aioble
|
||||||
|
import bluetooth
|
||||||
|
|
||||||
|
import random
|
||||||
|
import struct
|
||||||
|
|
||||||
|
# org.bluetooth.service.environmental_sensing
|
||||||
|
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
|
||||||
|
# org.bluetooth.characteristic.temperature
|
||||||
|
_ENV_SENSE_TEMP_UUID = bluetooth.UUID(0x2A6E)
|
||||||
|
# org.bluetooth.characteristic.gap.appearance.xml
|
||||||
|
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
|
||||||
|
|
||||||
|
# How frequently to send advertising beacons.
|
||||||
|
_ADV_INTERVAL_MS = 250_000
|
||||||
|
|
||||||
|
# Register GATT server.
|
||||||
|
temp_service = aioble.Service(_ENV_SENSE_UUID)
|
||||||
|
temp_characteristic = aioble.Characteristic(
|
||||||
|
temp_service, _ENV_SENSE_TEMP_UUID, read=True, notify=True
|
||||||
|
)
|
||||||
|
aioble.register_services(temp_service)
|
||||||
|
|
||||||
|
|
||||||
|
# Helper to encode the temperature characteristic encoding (sint16, hundredths of a degree).
|
||||||
|
def _encode_temperature(temp_deg_c):
|
||||||
|
return struct.pack("<h", int(temp_deg_c * 100))
|
||||||
|
|
||||||
|
|
||||||
|
# This would be periodically polling a hardware sensor.
|
||||||
|
async def sensor_task():
|
||||||
|
t = 24.5
|
||||||
|
while True:
|
||||||
|
temp_characteristic.write(_encode_temperature(t))
|
||||||
|
t += random.uniform(-0.5, 0.5)
|
||||||
|
await asyncio.sleep_ms(1000)
|
||||||
|
|
||||||
|
|
||||||
|
# Serially wait for connections. Don't advertise while a central is
|
||||||
|
# connected.
|
||||||
|
async def peripheral_task():
|
||||||
|
while True:
|
||||||
|
async with await aioble.advertise(
|
||||||
|
_ADV_INTERVAL_MS,
|
||||||
|
name="Nicla-Vision",
|
||||||
|
services=[_ENV_SENSE_UUID],
|
||||||
|
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
|
||||||
|
) as connection:
|
||||||
|
print("Connection from", connection.device)
|
||||||
|
await connection.disconnected()
|
||||||
|
|
||||||
|
|
||||||
|
# Run both tasks.
|
||||||
|
async def main():
|
||||||
|
t1 = asyncio.create_task(sensor_task())
|
||||||
|
t2 = asyncio.create_task(peripheral_task())
|
||||||
|
await asyncio.gather(t1, t2)
|
||||||
|
|
||||||
|
|
||||||
|
asyncio.run(main())
|
||||||
@ -3,6 +3,7 @@
|
|||||||
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net.
|
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net.
|
||||||
|
|
||||||
import network
|
import network
|
||||||
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
|
|||||||
@ -3,7 +3,7 @@
|
|||||||
# This example shows how to get the IP address for websites via DNS.
|
# This example shows how to get the IP address for websites via DNS.
|
||||||
|
|
||||||
import network
|
import network
|
||||||
import usocket
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
|
|||||||
@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
|
import time
|
||||||
|
|
||||||
# AP info
|
# AP info
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
|
|||||||
@ -1,7 +1,8 @@
|
|||||||
# Simple HTTPS client example.
|
# Simple HTTPS client example.
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
import ussl
|
import ssl
|
||||||
|
import time
|
||||||
|
|
||||||
# AP info
|
# AP info
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
@ -34,7 +35,7 @@ client.connect(addr)
|
|||||||
# Set timeout
|
# Set timeout
|
||||||
client.settimeout(3.0)
|
client.settimeout(3.0)
|
||||||
|
|
||||||
client = ussl.wrap_socket(client, server_hostname=HOST)
|
client = ssl.wrap_socket(client, server_hostname=HOST)
|
||||||
|
|
||||||
# Send HTTP request and recv response
|
# Send HTTP request and recv response
|
||||||
request = "GET / HTTP/1.1\r\n"
|
request = "GET / HTTP/1.1\r\n"
|
||||||
|
|||||||
@ -4,8 +4,8 @@
|
|||||||
|
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
import ustruct
|
import struct
|
||||||
import utime
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
@ -13,6 +13,7 @@ KEY = "" # Network key
|
|||||||
TIMESTAMP = 2208988800 + 946684800
|
TIMESTAMP = 2208988800 + 946684800
|
||||||
|
|
||||||
# Init wlan module and connect to network
|
# Init wlan module and connect to network
|
||||||
|
print("Trying to connect... (This may take a while)...")
|
||||||
wlan = network.WLAN(network.STA_IF)
|
wlan = network.WLAN(network.STA_IF)
|
||||||
wlan.active(True)
|
wlan.active(True)
|
||||||
wlan.connect(SSID, KEY)
|
wlan.connect(SSID, KEY)
|
||||||
@ -35,5 +36,5 @@ client.sendto("\x1b" + 47 * "\0", addr)
|
|||||||
data, address = client.recvfrom(1024)
|
data, address = client.recvfrom(1024)
|
||||||
|
|
||||||
# Print time
|
# Print time
|
||||||
t = ustruct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
||||||
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (utime.localtime(t)[0:6]))
|
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))
|
||||||
|
|||||||
@ -4,8 +4,8 @@
|
|||||||
|
|
||||||
import network
|
import network
|
||||||
import socket
|
import socket
|
||||||
import ustruct
|
import struct
|
||||||
import utime
|
import time
|
||||||
|
|
||||||
SSID = "" # Network SSID
|
SSID = "" # Network SSID
|
||||||
KEY = "" # Network key
|
KEY = "" # Network key
|
||||||
@ -35,5 +35,5 @@ client.sendto("\x1b" + 47 * "\0", addr)
|
|||||||
data, address = client.recvfrom(1024)
|
data, address = client.recvfrom(1024)
|
||||||
|
|
||||||
# Print time
|
# Print time
|
||||||
t = ustruct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
|
||||||
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (utime.localtime(t)[0:6]))
|
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))
|
||||||
|
|||||||
@ -0,0 +1,61 @@
|
|||||||
|
# Bluetooth Blinky Example
|
||||||
|
#
|
||||||
|
# Use nRFConnect app from the App store, connect to the Nano and write 1/0 to control the LED.
|
||||||
|
|
||||||
|
import bluetooth
|
||||||
|
import time
|
||||||
|
from ble_advertising import advertising_payload
|
||||||
|
from machine import LED
|
||||||
|
from micropython import const
|
||||||
|
|
||||||
|
_IRQ_CENTRAL_CONNECT = const(1)
|
||||||
|
_IRQ_CENTRAL_DISCONNECT = const(2)
|
||||||
|
_IRQ_GATTS_WRITE = const(3)
|
||||||
|
|
||||||
|
_FLAG_READ = const(0x0002)
|
||||||
|
_FLAG_WRITE = const(0x0008)
|
||||||
|
_FLAG_NOTIFY = const(0x0010)
|
||||||
|
_FLAG_INDICATE = const(0x0020)
|
||||||
|
|
||||||
|
_SERVICE_UUID = bluetooth.UUID(0x1523)
|
||||||
|
_LED_CHAR_UUID = (bluetooth.UUID(0x1525), _FLAG_WRITE)
|
||||||
|
_LED_SERVICE = (
|
||||||
|
_SERVICE_UUID,
|
||||||
|
(_LED_CHAR_UUID,),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class BLETemperature:
|
||||||
|
def __init__(self, ble, name="Portenta-H7"):
|
||||||
|
self._ble = ble
|
||||||
|
self._ble.active(True)
|
||||||
|
self._ble.irq(self._irq)
|
||||||
|
((self._handle,),) = self._ble.gatts_register_services((_LED_SERVICE,))
|
||||||
|
self._connections = set()
|
||||||
|
self._payload = advertising_payload(name=name, services=[_SERVICE_UUID])
|
||||||
|
self._advertise()
|
||||||
|
self.led = LED("LED_BLUE")
|
||||||
|
|
||||||
|
def _irq(self, event, data):
|
||||||
|
# Track connections so we can send notifications.
|
||||||
|
if event == _IRQ_CENTRAL_CONNECT:
|
||||||
|
conn_handle, _, _ = data
|
||||||
|
self._connections.add(conn_handle)
|
||||||
|
elif event == _IRQ_CENTRAL_DISCONNECT:
|
||||||
|
conn_handle, _, _ = data
|
||||||
|
self._connections.remove(conn_handle)
|
||||||
|
# Start advertising again to allow a new connection.
|
||||||
|
self._advertise()
|
||||||
|
elif event == _IRQ_GATTS_WRITE:
|
||||||
|
self.led.value(self._ble.gatts_read(data[-1])[0])
|
||||||
|
|
||||||
|
def _advertise(self, interval_us=500000):
|
||||||
|
self._ble.gap_advertise(interval_us, adv_data=self._payload)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
ble = bluetooth.BLE()
|
||||||
|
temp = BLETemperature(ble)
|
||||||
|
|
||||||
|
while True:
|
||||||
|
time.sleep_ms(1000)
|
||||||
@ -8,7 +8,7 @@ import random
|
|||||||
import struct
|
import struct
|
||||||
import time
|
import time
|
||||||
from ble_advertising import advertising_payload
|
from ble_advertising import advertising_payload
|
||||||
|
from machine import LED
|
||||||
from micropython import const
|
from micropython import const
|
||||||
|
|
||||||
_IRQ_CENTRAL_CONNECT = const(1)
|
_IRQ_CENTRAL_CONNECT = const(1)
|
||||||
@ -36,7 +36,7 @@ _ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
|
|||||||
|
|
||||||
|
|
||||||
class BLETemperature:
|
class BLETemperature:
|
||||||
def __init__(self, ble, name="PORTENTA_BLE"):
|
def __init__(self, ble, name="Portenta-H7"):
|
||||||
self._ble = ble
|
self._ble = ble
|
||||||
self._ble.active(True)
|
self._ble.active(True)
|
||||||
self._ble.irq(self._irq)
|
self._ble.irq(self._irq)
|
||||||
@ -48,17 +48,20 @@ class BLETemperature:
|
|||||||
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
|
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
|
||||||
)
|
)
|
||||||
self._advertise()
|
self._advertise()
|
||||||
|
self.led = LED("LED_BLUE")
|
||||||
|
|
||||||
def _irq(self, event, data):
|
def _irq(self, event, data):
|
||||||
# Track connections so we can send notifications.
|
# Track connections so we can send notifications.
|
||||||
if event == _IRQ_CENTRAL_CONNECT:
|
if event == _IRQ_CENTRAL_CONNECT:
|
||||||
conn_handle, _, _ = data
|
conn_handle, _, _ = data
|
||||||
self._connections.add(conn_handle)
|
self._connections.add(conn_handle)
|
||||||
|
self.led.on()
|
||||||
elif event == _IRQ_CENTRAL_DISCONNECT:
|
elif event == _IRQ_CENTRAL_DISCONNECT:
|
||||||
conn_handle, _, _ = data
|
conn_handle, _, _ = data
|
||||||
self._connections.remove(conn_handle)
|
self._connections.remove(conn_handle)
|
||||||
# Start advertising again to allow a new connection.
|
# Start advertising again to allow a new connection.
|
||||||
self._advertise()
|
self._advertise()
|
||||||
|
self.led.off()
|
||||||
elif event == _IRQ_GATTS_INDICATE_DONE:
|
elif event == _IRQ_GATTS_INDICATE_DONE:
|
||||||
conn_handle, value_handle, status = data
|
conn_handle, value_handle, status = data
|
||||||
|
|
||||||
@ -79,7 +82,7 @@ class BLETemperature:
|
|||||||
self._ble.gap_advertise(interval_us, adv_data=self._payload)
|
self._ble.gap_advertise(interval_us, adv_data=self._payload)
|
||||||
|
|
||||||
|
|
||||||
def demo():
|
if __name__ == "__main__":
|
||||||
ble = bluetooth.BLE()
|
ble = bluetooth.BLE()
|
||||||
temp = BLETemperature(ble)
|
temp = BLETemperature(ble)
|
||||||
|
|
||||||
@ -93,7 +96,3 @@ def demo():
|
|||||||
# Random walk the temperature.
|
# Random walk the temperature.
|
||||||
t += random.uniform(-0.5, 0.5)
|
t += random.uniform(-0.5, 0.5)
|
||||||
time.sleep_ms(1000)
|
time.sleep_ms(1000)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
demo()
|
|
||||||
|
|||||||
@ -0,0 +1,63 @@
|
|||||||
|
from micropython import const
|
||||||
|
|
||||||
|
import uasyncio as asyncio
|
||||||
|
import aioble
|
||||||
|
import bluetooth
|
||||||
|
|
||||||
|
import random
|
||||||
|
import struct
|
||||||
|
|
||||||
|
# org.bluetooth.service.environmental_sensing
|
||||||
|
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
|
||||||
|
# org.bluetooth.characteristic.temperature
|
||||||
|
_ENV_SENSE_TEMP_UUID = bluetooth.UUID(0x2A6E)
|
||||||
|
# org.bluetooth.characteristic.gap.appearance.xml
|
||||||
|
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
|
||||||
|
|
||||||
|
# How frequently to send advertising beacons.
|
||||||
|
_ADV_INTERVAL_MS = 250_000
|
||||||
|
|
||||||
|
# Register GATT server.
|
||||||
|
temp_service = aioble.Service(_ENV_SENSE_UUID)
|
||||||
|
temp_characteristic = aioble.Characteristic(
|
||||||
|
temp_service, _ENV_SENSE_TEMP_UUID, read=True, notify=True
|
||||||
|
)
|
||||||
|
aioble.register_services(temp_service)
|
||||||
|
|
||||||
|
|
||||||
|
# Helper to encode the temperature characteristic encoding (sint16, hundredths of a degree).
|
||||||
|
def _encode_temperature(temp_deg_c):
|
||||||
|
return struct.pack("<h", int(temp_deg_c * 100))
|
||||||
|
|
||||||
|
|
||||||
|
# This would be periodically polling a hardware sensor.
|
||||||
|
async def sensor_task():
|
||||||
|
t = 24.5
|
||||||
|
while True:
|
||||||
|
temp_characteristic.write(_encode_temperature(t))
|
||||||
|
t += random.uniform(-0.5, 0.5)
|
||||||
|
await asyncio.sleep_ms(1000)
|
||||||
|
|
||||||
|
|
||||||
|
# Serially wait for connections. Don't advertise while a central is
|
||||||
|
# connected.
|
||||||
|
async def peripheral_task():
|
||||||
|
while True:
|
||||||
|
async with await aioble.advertise(
|
||||||
|
_ADV_INTERVAL_MS,
|
||||||
|
name="Portenta-H7",
|
||||||
|
services=[_ENV_SENSE_UUID],
|
||||||
|
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
|
||||||
|
) as connection:
|
||||||
|
print("Connection from", connection.device)
|
||||||
|
await connection.disconnected()
|
||||||
|
|
||||||
|
|
||||||
|
# Run both tasks.
|
||||||
|
async def main():
|
||||||
|
t1 = asyncio.create_task(sensor_task())
|
||||||
|
t2 = asyncio.create_task(peripheral_task())
|
||||||
|
await asyncio.gather(t1, t2)
|
||||||
|
|
||||||
|
|
||||||
|
asyncio.run(main())
|
||||||
Loading…
Reference in New Issue
Block a user