scripts/examples: Refactor WiFi and Bluetooth examples.

Unified set of examples for all WiFi/BT modules.
This commit is contained in:
iabdalkader 2023-10-27 21:09:30 +02:00
parent a99a8fad95
commit 0101917e0e
59 changed files with 76 additions and 2060 deletions

View File

@ -1,21 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Connect Example
#
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net.
import network
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())

View File

@ -1,24 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# DNS Example
#
# This example shows how to get the IP address for websites via DNS.
import network
import usocket
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
print(usocket.getaddrinfo("www.google.com", 80)[0][4])

View File

@ -1,42 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Simple HTTP client example.
import network
import usocket
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
PORT = 80
HOST = "www.google.com"
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
# Get addr info via DNS
addr = usocket.getaddrinfo(HOST, PORT)[0][4]
print(addr)
# Create a new socket and connect to addr
client = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
client.connect(addr)
# Set timeout
client.settimeout(3.0)
# Send HTTP request and recv response
client.send("GET / HTTP/1.1\r\nHost: %s\r\n\r\n" % (HOST))
print(client.recv(1024))
# Close socket
client.close()

View File

@ -1,54 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Simple HTTPS client example.
import network
import usocket
import ussl
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
PORT = 443
HOST = "www.google.com"
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
# Get addr info via DNS
addr = usocket.getaddrinfo(HOST, PORT)[0][4]
print(addr)
# Create a new socket and connect to addr
client = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
client.connect(addr)
# Set timeout
client.settimeout(3.0)
client = ussl.wrap_socket(client, server_hostname=HOST)
# Send HTTP request and recv response
request = "GET / HTTP/1.1\r\n"
request += "HOST: %s\r\n"
request += "User-Agent: Mozilla/5.0\r\n"
request += "Connection: keep-alive\r\n\r\n"
# Add more headers if needed.
client.write(request % (HOST) + "\r\n")
response = client.read(1024)
for l in response.split(b"\r\n"):
print(l.decode())
# Close socket
client.close()

View File

@ -1,52 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Post files with HTTP/Post urequests module example
import network
import urequests
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
url = "http://website.com/upload.php/"
# Or <ip>/<host>:port
# url = 'http://website.com:80/upload.php/'
# SSL is supported.
# url = 'https://192.168.1.102:443/upload.php/'
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.12; rv:55.0) Gecko/20100101 Firefox/55.0",
# Add more headers if needed
}
# Send some files
files = {
"image1": ("example1.jpg", open("example1.jpg", "rb")),
"image2": ("example2.jpg", open("example2.jpg", "rb")),
}
# Post a request
if True:
# Send some files
r = urequests.post(
url, files=files, headers=headers
) # Can add auth=('username', 'password') if needed
else:
# Or send some JSON data
r = urequests.post(
url, json={"some": "data"}, headers=headers
) # Can add auth=('username', 'password') if needed
print(r.status_code, r.reason)
print(r.headers, r.content)

View File

@ -1,90 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MJPEG Streaming
#
# This example shows off how to do MJPEG streaming to a FIREFOX webrowser
# Chrome, Firefox and MJpegViewer App on Android have been tested.
# Connect to the IP address/port printed out from ifconfig to view the stream.
import sensor
import time
import network
import usocket
SSID = "" # Network SSID
KEY = "" # Network key
HOST = "" # Use first available interface
PORT = 8080 # Arbitrary non-privileged port
# Reset sensor
sensor.reset()
sensor.set_framesize(sensor.QQVGA)
sensor.set_pixformat(sensor.GRAYSCALE)
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
# Create server socket
s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
s.setsockopt(usocket.SOL_SOCKET, usocket.SO_REUSEADDR, True)
# Bind and listen
s.bind([HOST, PORT])
s.listen(5)
# Set server socket to blocking
s.setblocking(True)
def start_streaming(s):
print("Waiting for connections..")
client, addr = s.accept()
# set client socket timeout to 2s
client.settimeout(2.0)
print("Connected to " + addr[0] + ":" + str(addr[1]))
# Read request from client
data = client.recv(1024)
# Should parse client request here
# Send multipart header
client.sendall(
"HTTP/1.1 200 OK\r\n"
"Server: OpenMV\r\n"
"Content-Type: multipart/x-mixed-replace;boundary=openmv\r\n"
"Cache-Control: no-cache\r\n"
"Pragma: no-cache\r\n\r\n"
)
# FPS clock
clock = time.clock()
# Start streaming images
# NOTE: Disable IDE preview to increase streaming FPS.
while True:
clock.tick() # Track elapsed milliseconds between snapshots().
frame = sensor.snapshot()
cframe = frame.compressed(quality=35)
header = (
"\r\n--openmv\r\n"
"Content-Type: image/jpeg\r\n"
"Content-Length:" + str(cframe.size()) + "\r\n\r\n"
)
client.sendall(header)
client.sendall(cframe)
print(clock.fps())
while True:
try:
start_streaming(s)
except OSError as e:
print("socket error: ", e)
# sys.print_exception(e)

View File

@ -1,95 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MJPEG Streaming with FIR
#
# This example shows off how to do MJPEG streaming to a FIREFOX webrowser
# (IE and Chrome do not work). Just input your network SSID and KEY and then
# connect to the IP address/port printed out from ifconfig.
import sensor
import network
import usocket
import fir
SSID = "" # Network SSID
KEY = "" # Network key
HOST = "" # Use first available interface
PORT = 8000 # Arbitrary non-privileged port
# Reset sensor
sensor.reset()
sensor.set_framesize(sensor.QQVGA)
sensor.set_pixformat(sensor.RGB565)
# Initialize the thermal sensor
fir.init()
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
# Create server socket
s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
# Bind and listen
s.bind((HOST, PORT))
s.listen(5)
# Set server socket to blocking
s.setblocking(True)
# Set timeout to 1s
s.settimeout(1.0)
print("Waiting for connections..")
client, addr = s.accept()
print("Connected to " + addr[0] + ":" + str(addr[1]))
# Read request from client
data = client.recv(1024)
# Should parse client request here
# Send multipart header
client.send(
"HTTP/1.1 200 OK\r\n"
"Server: OpenMV\r\n"
"Content-Type: multipart/x-mixed-replace;boundary=openmv\r\n"
"Cache-Control: no-cache\r\n"
"Pragma: no-cache\r\n\r\n"
)
# Start streaming images
while True:
image = sensor.snapshot()
# Capture FIR data
# ta: Ambient temperature
# ir: Object temperatures (IR array)
# to_min: Minimum object temperature
# to_max: Maximum object temperature
ta, ir, to_min, to_max = fir.read_ir()
# Scale the image and belnd it with the framebuffer
fir.draw_ir(image, ir)
# Draw ambient, min and max temperatures.
image.draw_string(0, 0, "Ta: %0.2f" % ta, color=(0xFF, 0x00, 0x00))
image.draw_string(0, 8, "To min: %0.2f" % to_min, color=(0xFF, 0x00, 0x00))
image.draw_string(0, 16, "To max: %0.2f" % to_max, color=(0xFF, 0x00, 0x00))
cimage = image.compressed(quality=90)
client.sendall(
"\r\n--openmv\r\n"
"Content-Type: image/jpeg\r\n"
"Content-Length:" + str(cimage.size()) + "\r\n\r\n"
)
client.sendall(cimage)
client.close()

View File

@ -1,35 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MQTT Example.
# This example shows how to use the MQTT library to publish to a topic.
#
# 1) Copy the mqtt.py library to OpenMV storage.
# 2) Run this script on the OpenMV camera.
# 3) Install the mosquitto client on PC and run the following command:
# mosquitto_sub -h test.mosquitto.org -t "openmv/test" -v
#
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time
import network
from mqtt import MQTTClient
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
client = MQTTClient("openmv", "test.mosquitto.org", port=1883)
client.connect()
while True:
client.publish("openmv/test", "Hello World!")
time.sleep_ms(1000)

View File

@ -1,44 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MQTT Example.
# This example shows how to use the MQTT library to subscribe to a topic.
#
# 1) Copy the mqtt.py library to OpenMV storage.
# 2) Run this script on the OpenMV camera.
# 3) Install the mosquitto client on PC and run the following command:
# mosquitto_pub -t "openmv/test" -m "Hello World!" -h test.mosquitto.org -p 1883
#
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time
import network
from mqtt import MQTTClient
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
client = MQTTClient("openmv", "test.mosquitto.org", port=1883)
client.connect()
def callback(topic, msg):
print(topic, msg)
# must set callback first
client.set_callback(callback)
client.subscribe("openmv/test")
while True:
client.check_msg() # poll for messages.
time.sleep_ms(1000)

View File

@ -1,40 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# NTP Example
#
# This example shows how to get the current time using NTP with the WiFi shield.
import network
import usocket
import ustruct
import utime
SSID = "" # Network SSID
KEY = "" # Network key
TIMESTAMP = 2208988800 + 946684800
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
# Create new socket
client = usocket.socket(usocket.AF_INET, usocket.SOCK_DGRAM)
# Get addr info via DNS
addr = usocket.getaddrinfo("pool.ntp.org", 123)[0][4]
# Send query
client.sendto("\x1b" + 47 * "\0", addr)
data, address = client.recvfrom(1024)
# Print time
t = ustruct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (utime.localtime(t)[0:6]))

View File

@ -1,20 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Scan Example
#
# This example shows how to scan for networks with the WiFi shield.
import time
import network
wlan = network.WINC()
print("\nFirmware version:", wlan.fw_version())
while True:
scan_result = wlan.scan()
for ap in scan_result:
print("Channel:%d RSSI:%d Auth:%d BSSID:%s SSID:%s" % (ap))
print()
time.sleep_ms(1000)

View File

@ -1,39 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# NTP Example using static IP.
#
# This example shows how to get the current time using NTP with the WiFi shield.
import network
import usocket
import ustruct
import utime
SSID = "" # Network SSID
KEY = "" # Network key
TIMESTAMP = 2208988800 + 946684800
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WINC()
# ifconfig must be called before connect()
wlan.ifconfig(("192.168.1.200", "255.255.255.0", "192.168.1.1", "192.168.1.1"))
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# Create new socket
client = usocket.socket(usocket.AF_INET, usocket.SOCK_DGRAM)
# Get addr info via DNS
addr = usocket.getaddrinfo("pool.ntp.org", 123)[0][4]
# Send query
client.sendto("\x1b" + 47 * "\0", addr)
data, address = client.recvfrom(1024)
# Print time
t = ustruct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (utime.localtime(t)[0:6]))

View File

@ -0,0 +1,12 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Atmel WINC1500 Firmware dump.
import network
wlan = network.WINC(mode=network.WINC.MODE_FIRMWARE)
# For ATWINC1500-MR210PB only.
wlan.fw_dump("/winc_19_7_6.bin")

View File

@ -2,7 +2,7 @@
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved. # Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE # https://github.com/openmv/openmv/blob/master/LICENSE
# #
# WINC Firmware Update Script. # Atmel WINC1500 Firmware Update.
# #
# This script updates the ATWINC1500 WiFi module firmware. # This script updates the ATWINC1500 WiFi module firmware.
# 1) Copy the firmware image to a FAT32/exFAT SD card. # 1) Copy the firmware image to a FAT32/exFAT SD card.

View File

@ -0,0 +1,10 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Atmel WINC1500 firmware version.
import network
wlan = network.WINC()
print("\nFirmware version:", wlan.fw_version())

View File

@ -4,7 +4,7 @@
# #
# Connect Example # Connect Example
# #
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net. # This example shows how to connect to a WiFi network.
import network import network
import time import time

View File

@ -8,6 +8,7 @@
import network import network
import time import time
import socket
SSID = "" # Network SSID SSID = "" # Network SSID
KEY = "" # Network key KEY = "" # Network key
@ -23,4 +24,4 @@ while not wlan.isconnected():
# We should have a valid IP now via DHCP # We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig()) print("WiFi Connected ", wlan.ifconfig())
print(usocket.getaddrinfo("www.google.com", 80)[0][4]) print(socket.getaddrinfo("www.google.com", 80)[0][4])

View File

@ -3,6 +3,7 @@
# https://github.com/openmv/openmv/blob/master/LICENSE # https://github.com/openmv/openmv/blob/master/LICENSE
# #
# Simple HTTPS client example. # Simple HTTPS client example.
import network import network
import socket import socket
import ssl import ssl

View File

@ -7,6 +7,7 @@
# This example shows off how to do MJPEG streaming to a FIREFOX webrowser # This example shows off how to do MJPEG streaming to a FIREFOX webrowser
# Chrome, Firefox and MJpegViewer App on Android have been tested. # Chrome, Firefox and MJpegViewer App on Android have been tested.
# Connect to the IP address/port printed out from ifconfig to view the stream. # Connect to the IP address/port printed out from ifconfig to view the stream.
import sensor import sensor
import time import time
import network import network

View File

@ -11,7 +11,7 @@
import sensor import sensor
import time import time
import network import network
import usocket import socket
SSID = "OPENMV_AP" # Network SSID SSID = "OPENMV_AP" # Network SSID
KEY = "1234567890" # Network key (must be 10 chars) KEY = "1234567890" # Network key (must be 10 chars)
@ -24,21 +24,17 @@ sensor.set_framesize(sensor.QQVGA)
sensor.set_pixformat(sensor.GRAYSCALE) sensor.set_pixformat(sensor.GRAYSCALE)
# Init wlan module in AP mode. # Init wlan module in AP mode.
wlan = network.WINC(mode=network.WINC.MODE_AP) wlan = network.WLAN(network.AP_IF)
wlan.start_ap(SSID, key=KEY, security=wlan.WEP, channel=2) wlan.active(True)
# Note some WiFi modules only support WEP in AP mode.
wlan.config(ssid=SSID, key=KEY, channel=2) # security=wlan.WEP
print("AP mode started. SSID: {} IP: {}".format(SSID, wlan.ifconfig()[0])) print("AP mode started. SSID: {} IP: {}".format(SSID, wlan.ifconfig()[0]))
# You can block waiting for client to connect # You can block waiting for client to connect
# print(wlan.wait_for_sta(10000)) # print(wlan.wait_for_sta(100000))
def start_streaming(s): def start_streaming(client):
print("Waiting for connections..")
client, addr = s.accept()
# set client socket timeout to 2s
client.settimeout(2.0)
print("Connected to " + addr[0] + ":" + str(addr[1]))
# Read request from client # Read request from client
data = client.recv(1024) data = client.recv(1024)
# Should parse client request here # Should parse client request here
@ -71,23 +67,34 @@ def start_streaming(s):
print(clock.fps()) print(clock.fps())
while True: server = None
# Create server socket
s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
s.setsockopt(usocket.SOL_SOCKET, usocket.SO_REUSEADDR, True)
try:
# Bind and listen
s.bind([HOST, PORT])
s.listen(5)
# Set server socket to blocking
s.setblocking(True)
# Set server socket timeout while True:
# NOTE: Due to a WINC FW bug, the server socket must be closed and reopened if if server is None:
# the client disconnects. Use a timeout here to close and re-create the socket. # Create server socket
s.settimeout(3) server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
start_streaming(s) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
# Bind and listen
server.bind([HOST, PORT])
server.listen(5)
# Set server socket to blocking
server.setblocking(True)
try:
print("Waiting for connections..")
client, addr = server.accept()
except OSError as e: except OSError as e:
s.close() server.close()
print("socket error: ", e) server = None
print("server socket error:", e)
continue
try:
# set client socket timeout to 2s
client.settimeout(5.0)
print("Connected to " + addr[0] + ":" + str(addr[1]))
start_streaming(client)
except OSError as e:
client.close()
print("client socket error:", e)
# sys.print_exception(e) # sys.print_exception(e)

View File

@ -11,6 +11,7 @@
# mosquitto_sub -h test.mosquitto.org -t "openmv/test" -v # mosquitto_sub -h test.mosquitto.org -t "openmv/test" -v
# #
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com) # NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time import time
import network import network
from mqtt import MQTTClient from mqtt import MQTTClient

View File

@ -11,6 +11,7 @@
# mosquitto_pub -t "openmv/test" -m "Hello World!" -h test.mosquitto.org -p 1883 # mosquitto_pub -t "openmv/test" -m "Hello World!" -h test.mosquitto.org -p 1883
# #
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com) # NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time import time
import network import network
from mqtt import MQTTClient from mqtt import MQTTClient

View File

@ -4,7 +4,7 @@
# #
# NTP Example # NTP Example
# #
# This example shows how to get the current time using NTP with the WiFi shield. # This example shows how to get the current time using NTP.
import network import network
import socket import socket
@ -14,7 +14,10 @@ import time
SSID = "" # Network SSID SSID = "" # Network SSID
KEY = "" # Network key KEY = "" # Network key
TIMESTAMP = 2208988800 + 946684800 TIMESTAMP = 2208988800
if time.gmtime(0)[0] == 2000:
TIMESTAMP += 946684800
# Init wlan module and connect to network # Init wlan module and connect to network
print("Trying to connect... (This may take a while)...") print("Trying to connect... (This may take a while)...")

View File

@ -4,7 +4,7 @@
# #
# NTP Example using static IP. # NTP Example using static IP.
# #
# This example shows how to get the current time using NTP with the WiFi shield. # This example shows how to set a static IP config.
import network import network
import socket import socket

View File

@ -1,25 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Connect Example
#
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net.
import network
import time
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())

View File

@ -1,26 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# DNS Example
#
# This example shows how to get the IP address for websites via DNS.
import network
import time
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
print(usocket.getaddrinfo("www.google.com", 80)[0][4])

View File

@ -1,47 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MQTT Example.
# This example shows how to use the MQTT library to subscribe to a topic.
#
# 1) Copy the mqtt.py library to OpenMV storage.
# 2) Run this script on the OpenMV camera.
# 3) Install the mosquitto client on PC and run the following command:
# mosquitto_pub -t "openmv/test" -m "Hello World!" -h test.mosquitto.org -p 1883
#
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time
import network
from mqtt import MQTTClient
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
client = MQTTClient("openmv", "test.mosquitto.org", port=1883)
client.connect()
def callback(topic, msg):
print(topic, msg)
# must set callback first
client.set_callback(callback)
client.subscribe("openmv/test")
while True:
client.check_msg() # poll for messages.
time.sleep_ms(1000)

View File

@ -1,44 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# NTP Example
#
# This example shows how to get the current time using NTP with the WiFi shield.
import network
import socket
import struct
import time
SSID = "" # Network SSID
KEY = "" # Network key
TIMESTAMP = 2208988800 + 946684800
# Init wlan module and connect to network
print("Trying to connect... (This may take a while)...")
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Create new socket
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# Get addr info via DNS
addr = socket.getaddrinfo("pool.ntp.org", 123)[0][4]
# Send query
client.sendto("\x1b" + 47 * "\0", addr)
data, address = client.recvfrom(1024)
# Print time
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))

View File

@ -1,24 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Scan Example
#
# This example shows how to scan for networks with the WiFi shield.
import time
import network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
print("Scanning...")
while True:
scan_result = wlan.scan()
for ap in scan_result:
print(
"SSID: %s BSSID: %s Channel: %d RSSI: %d Auth: %d"
% (ap[0], ":".join(["%X" % i for i in ap[1]]), ap[2], ap[3], ap[4])
)
print()
time.sleep_ms(1000)

View File

@ -1,43 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Simple HTTP client example.
import network
import socket
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
PORT = 80
HOST = "www.google.com"
# Init wlan module and connect to network
print("Trying to connect. Note this may take a while...")
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Get addr info via DNS
addr = socket.getaddrinfo(HOST, PORT)[0][4]
print(addr)
# Create a new socket and connect to addr
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(addr)
# Set timeout
client.settimeout(3.0)
# Send HTTP request and recv response
client.send("GET / HTTP/1.1\r\nHost: %s\r\n\r\n" % (HOST))
print(client.recv(1024))
# Close socket
client.close()

View File

@ -1,44 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# NTP Example
#
# This example shows how to get the current time using NTP with the WiFi shield.
import network
import usocket
import ustruct
import utime
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
TIMESTAMP = 2208988800
# Init wlan module and connect to network
print("Trying to connect... (may take a while)...")
wlan = network.WLAN()
wlan.active(True)
wlan.connect(SSID, key=KEY, security=wlan.WPA_PSK)
# We should have a valid IP now via DHCP
print(wlan.ifconfig())
# Create new socket
client = usocket.socket(usocket.AF_INET, usocket.SOCK_DGRAM)
client.bind(("", 8080))
# client.settimeout(3.0)
# Get addr info via DNS
addr = usocket.getaddrinfo("pool.ntp.org", 123)[0][4]
# Send query
client.sendto("\x1b" + 47 * "\0", addr)
data, address = client.recvfrom(1024)
# Print time
t = ustruct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (utime.localtime(t)[0:6]))

View File

@ -1,25 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Connect Example
#
# This example shows how to connect your OpenMV Cam with a WiFi shield to the net.
import network
import time
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())

View File

@ -1,26 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# DNS Example
#
# This example shows how to get the IP address for websites via DNS.
import network
import time
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
print(usocket.getaddrinfo("www.google.com", 80)[0][4])

View File

@ -1,46 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Simple HTTP client example.
import network
import socket
import time
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
PORT = 80
HOST = "www.google.com"
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Get addr info via DNS
addr = socket.getaddrinfo(HOST, PORT)[0][4]
print(addr)
# Create a new socket and connect to addr
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(addr)
# Set timeout
client.settimeout(3.0)
# Send HTTP request and recv response
client.send("GET / HTTP/1.1\r\nHost: %s\r\n\r\n" % (HOST))
print(client.recv(1024))
# Close socket
client.close()

View File

@ -1,57 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Simple HTTPS client example.
import network
import socket
import ssl
import time
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
PORT = 443
HOST = "www.google.com"
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Get addr info via DNS
addr = socket.getaddrinfo(HOST, PORT)[0][4]
print(addr)
# Create a new socket and connect to addr
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(addr)
# Set timeout
client.settimeout(3.0)
client = ssl.wrap_socket(client, server_hostname=HOST)
# Send HTTP request and recv response
request = "GET / HTTP/1.1\r\n"
request += "HOST: %s\r\n"
request += "User-Agent: Mozilla/5.0\r\n"
request += "Connection: keep-alive\r\n\r\n"
# Add more headers if needed.
client.write(request % (HOST) + "\r\n")
response = client.read(1024)
for l in response.split(b"\r\n"):
print(l.decode())
# Close socket
client.close()

View File

@ -1,93 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MJPEG Streaming
#
# This example shows off how to do MJPEG streaming to a FIREFOX webrowser
# Chrome, Firefox and MJpegViewer App on Android have been tested.
# Connect to the IP address/port printed out from ifconfig to view the stream.
import sensor
import time
import network
import socket
SSID = "" # Network SSID
KEY = "" # Network key
HOST = "" # Use first available interface
PORT = 8080 # Arbitrary non-privileged port
# Init sensor
sensor.reset()
sensor.set_framesize(sensor.QVGA)
sensor.set_pixformat(sensor.RGB565)
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Create server socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
# Bind and listen
s.bind([HOST, PORT])
s.listen(5)
# Set server socket to blocking
s.setblocking(True)
def start_streaming(s):
print("Waiting for connections..")
client, addr = s.accept()
# set client socket timeout to 5s
client.settimeout(5.0)
print("Connected to " + addr[0] + ":" + str(addr[1]))
# Read request from client
data = client.recv(1024)
# Should parse client request here
# Send multipart header
client.sendall(
"HTTP/1.1 200 OK\r\n"
"Server: OpenMV\r\n"
"Content-Type: multipart/x-mixed-replace;boundary=openmv\r\n"
"Cache-Control: no-cache\r\n"
"Pragma: no-cache\r\n\r\n"
)
# FPS clock
clock = time.clock()
# Start streaming images
# NOTE: Disable IDE preview to increase streaming FPS.
while True:
clock.tick() # Track elapsed milliseconds between snapshots().
frame = sensor.snapshot()
cframe = frame.compressed(quality=35)
header = (
"\r\n--openmv\r\n"
"Content-Type: image/jpeg\r\n"
"Content-Length:" + str(cframe.size()) + "\r\n\r\n"
)
client.sendall(header)
client.sendall(cframe)
print(clock.fps())
while True:
try:
start_streaming(s)
except OSError as e:
print("socket error: ", e)
# sys.print_exception(e)

View File

@ -1,38 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MQTT Example.
# This example shows how to use the MQTT library to publish to a topic.
#
# 1) Copy the mqtt.py library to OpenMV storage.
# 2) Run this script on the OpenMV camera.
# 3) Install the mosquitto client on PC and run the following command:
# mosquitto_sub -h test.mosquitto.org -t "openmv/test" -v
#
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time
import network
from mqtt import MQTTClient
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
client = MQTTClient("openmv", "test.mosquitto.org", port=1883)
client.connect()
while True:
client.publish("openmv/test", "Hello World!")
time.sleep_ms(1000)

View File

@ -1,44 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# NTP Example
#
# This example shows how to get the current time using NTP with the WiFi shield.
import network
import socket
import struct
import time
SSID = "" # Network SSID
KEY = "" # Network key
TIMESTAMP = 2208988800 + 946684800
# Init wlan module and connect to network
print("Trying to connect... (This may take a while)...")
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Create new socket
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# Get addr info via DNS
addr = socket.getaddrinfo("pool.ntp.org", 123)[0][4]
# Send query
client.sendto("\x1b" + 47 * "\0", addr)
data, address = client.recvfrom(1024)
# Print time
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))

View File

@ -1,24 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Scan Example
#
# This example shows how to scan for networks with the WiFi shield.
import time
import network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
print("Scanning...")
while True:
scan_result = wlan.scan()
for ap in scan_result:
print(
"SSID: %s BSSID: %s Channel: %d RSSI: %d Auth: %d"
% (ap[0], ":".join(["%X" % i for i in ap[1]]), ap[2], ap[3], ap[4])
)
print()
time.sleep_ms(1000)

View File

@ -1,43 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# NTP Example using static IP.
#
# This example shows how to get the current time using NTP with the WiFi shield.
import network
import socket
import struct
import time
SSID = "" # Network SSID
KEY = "" # Network key
TIMESTAMP = 2208988800 + 946684800
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
# ifconfig must be called before connect()
wlan.ifconfig(("192.168.1.200", "255.255.255.0", "192.168.1.1", "192.168.1.1"))
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# Create new socket
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# Get addr info via DNS
addr = socket.getaddrinfo("pool.ntp.org", 123)[0][4]
# Send query
client.sendto("\x1b" + 47 * "\0", addr)
data, address = client.recvfrom(1024)
# Print time
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))

View File

@ -1,65 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# 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)

View File

@ -1,102 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# This example demonstrates a simple temperature sensor peripheral.
#
# The sensor's local value updates every second, and it will notify
# any connected central every 10 seconds.
import bluetooth
import random
import struct
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_INDICATE_DONE = const(20)
_FLAG_READ = const(0x0002)
_FLAG_NOTIFY = const(0x0010)
_FLAG_INDICATE = const(0x0020)
# org.bluetooth.service.environmental_sensing
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
# org.bluetooth.characteristic.temperature
_TEMP_CHAR = (
bluetooth.UUID(0x2A6E),
_FLAG_READ | _FLAG_NOTIFY | _FLAG_INDICATE,
)
_ENV_SENSE_SERVICE = (
_ENV_SENSE_UUID,
(_TEMP_CHAR,),
)
# org.bluetooth.characteristic.gap.appearance.xml
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
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((_ENV_SENSE_SERVICE,))
self._connections = set()
self._payload = advertising_payload(
name=name,
services=[_ENV_SENSE_UUID],
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
)
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)
self.led.on()
elif event == _IRQ_CENTRAL_DISCONNECT:
conn_handle, _, _ = data
self._connections.remove(conn_handle)
# Start advertising again to allow a new connection.
self._advertise()
self.led.off()
elif event == _IRQ_GATTS_INDICATE_DONE:
conn_handle, value_handle, status = data
def set_temperature(self, temp_deg_c, notify=False, indicate=False):
# Data is sint16 in degrees Celsius with a resolution of 0.01 degrees Celsius.
# Write the local value, ready for a central to read.
self._ble.gatts_write(self._handle, struct.pack("<h", int(temp_deg_c * 100)))
if notify or indicate:
for conn_handle in self._connections:
if notify:
# Notify connected centrals.
self._ble.gatts_notify(conn_handle, self._handle)
if indicate:
# Indicate connected centrals.
self._ble.gatts_indicate(conn_handle, self._handle)
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)
t = 25
i = 0
while True:
# Write every second, notify every 10 seconds.
i = (i + 1) % 10
temp.set_temperature(t, notify=i == 0, indicate=False)
# Random walk the temperature.
t += random.uniform(-0.5, 0.5)
time.sleep_ms(1000)

View File

@ -1,67 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
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())

View File

@ -1,46 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Simple HTTP client example.
import network
import socket
import time
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
PORT = 80
HOST = "www.google.com"
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Get addr info via DNS
addr = socket.getaddrinfo(HOST, PORT)[0][4]
print(addr)
# Create a new socket and connect to addr
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(addr)
# Set timeout
client.settimeout(3.0)
# Send HTTP request and recv response
client.send("GET / HTTP/1.1\r\nHost: %s\r\n\r\n" % (HOST))
print(client.recv(1024))
# Close socket
client.close()

View File

@ -1,57 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Simple HTTPS client example.
import network
import socket
import ssl
import time
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
PORT = 443
HOST = "www.google.com"
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Get addr info via DNS
addr = socket.getaddrinfo(HOST, PORT)[0][4]
print(addr)
# Create a new socket and connect to addr
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(addr)
# Set timeout
client.settimeout(3.0)
client = ssl.wrap_socket(client, server_hostname=HOST)
# Send HTTP request and recv response
request = "GET / HTTP/1.1\r\n"
request += "HOST: %s\r\n"
request += "User-Agent: Mozilla/5.0\r\n"
request += "Connection: keep-alive\r\n\r\n"
# Add more headers if needed.
client.write(request % (HOST) + "\r\n")
response = client.read(1024)
for l in response.split(b"\r\n"):
print(l.decode())
# Close socket
client.close()

View File

@ -1,93 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MJPEG Streaming
#
# This example shows off how to do MJPEG streaming to a FIREFOX webrowser
# Chrome, Firefox and MJpegViewer App on Android have been tested.
# Connect to the IP address/port printed out from ifconfig to view the stream.
import sensor
import time
import network
import socket
SSID = "" # Network SSID
KEY = "" # Network key
HOST = "" # Use first available interface
PORT = 8080 # Arbitrary non-privileged port
# Init sensor
sensor.reset()
sensor.set_framesize(sensor.QVGA)
sensor.set_pixformat(sensor.GRAYSCALE)
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
# Create server socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
# Bind and listen
s.bind([HOST, PORT])
s.listen(5)
# Set server socket to blocking
s.setblocking(True)
def start_streaming(s):
print("Waiting for connections..")
client, addr = s.accept()
# set client socket timeout to 5s
client.settimeout(5.0)
print("Connected to " + addr[0] + ":" + str(addr[1]))
# Read request from client
data = client.recv(1024)
# Should parse client request here
# Send multipart header
client.sendall(
"HTTP/1.1 200 OK\r\n"
"Server: OpenMV\r\n"
"Content-Type: multipart/x-mixed-replace;boundary=openmv\r\n"
"Cache-Control: no-cache\r\n"
"Pragma: no-cache\r\n\r\n"
)
# FPS clock
clock = time.clock()
# Start streaming images
# NOTE: Disable IDE preview to increase streaming FPS.
while True:
clock.tick() # Track elapsed milliseconds between snapshots().
frame = sensor.snapshot()
cframe = frame.compressed(quality=35)
header = (
"\r\n--openmv\r\n"
"Content-Type: image/jpeg\r\n"
"Content-Length:" + str(cframe.size()) + "\r\n\r\n"
)
client.sendall(header)
client.sendall(cframe)
print(clock.fps())
while True:
try:
start_streaming(s)
except OSError as e:
print("socket error: ", e)
# sys.print_exception(e)

View File

@ -1,38 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MQTT Example.
# This example shows how to use the MQTT library to publish to a topic.
#
# 1) Copy the mqtt.py library to OpenMV storage.
# 2) Run this script on the OpenMV camera.
# 3) Install the mosquitto client on PC and run the following command:
# mosquitto_sub -h test.mosquitto.org -t "openmv/test" -v
#
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time
import network
from mqtt import MQTTClient
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
client = MQTTClient("openmv", "test.mosquitto.org", port=1883)
client.connect()
while True:
client.publish("openmv/test", "Hello World!")
time.sleep_ms(1000)

View File

@ -1,47 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# MQTT Example.
# This example shows how to use the MQTT library to subscribe to a topic.
#
# 1) Copy the mqtt.py library to OpenMV storage.
# 2) Run this script on the OpenMV camera.
# 3) Install the mosquitto client on PC and run the following command:
# mosquitto_pub -t "openmv/test" -m "Hello World!" -h test.mosquitto.org -p 1883
#
# NOTE: If the mosquitto broker is unreachable, try another broker (For example: broker.hivemq.com)
import time
import network
from mqtt import MQTTClient
SSID = "" # Network SSID
KEY = "" # Network key
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# We should have a valid IP now via DHCP
print("WiFi Connected ", wlan.ifconfig())
client = MQTTClient("openmv", "test.mosquitto.org", port=1883)
client.connect()
def callback(topic, msg):
print(topic, msg)
# must set callback first
client.set_callback(callback)
client.subscribe("openmv/test")
while True:
client.check_msg() # poll for messages.
time.sleep_ms(1000)

View File

@ -1,24 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Scan Example
#
# This example shows how to scan for networks with the WiFi shield.
import time
import network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
print("Scanning...")
while True:
scan_result = wlan.scan()
for ap in scan_result:
print(
"SSID: %s BSSID: %s Channel: %d RSSI: %d Auth: %d"
% (ap[0], ":".join(["%X" % i for i in ap[1]]), ap[2], ap[3], ap[4])
)
print()
time.sleep_ms(1000)

View File

@ -1,43 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# NTP Example using static IP.
#
# This example shows how to get the current time using NTP with the WiFi shield.
import network
import socket
import struct
import time
SSID = "" # Network SSID
KEY = "" # Network key
TIMESTAMP = 2208988800 + 946684800
# Init wlan module and connect to network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
# ifconfig must be called before connect()
wlan.ifconfig(("192.168.1.200", "255.255.255.0", "192.168.1.1", "192.168.1.1"))
wlan.connect(SSID, KEY)
while not wlan.isconnected():
print('Trying to connect to "{:s}"...'.format(SSID))
time.sleep_ms(1000)
# Create new socket
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# Get addr info via DNS
addr = socket.getaddrinfo("pool.ntp.org", 123)[0][4]
# Send query
client.sendto("\x1b" + 47 * "\0", addr)
data, address = client.recvfrom(1024)
# Print time
t = struct.unpack(">IIIIIIIIIIII", data)[10] - TIMESTAMP
print("Year:%d Month:%d Day:%d Time: %d:%d:%d" % (time.localtime(t)[0:6]))

View File

@ -1,65 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# 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)

View File

@ -1,102 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# This example demonstrates a simple temperature sensor peripheral.
#
# The sensor's local value updates every second, and it will notify
# any connected central every 10 seconds.
import bluetooth
import random
import struct
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_INDICATE_DONE = const(20)
_FLAG_READ = const(0x0002)
_FLAG_NOTIFY = const(0x0010)
_FLAG_INDICATE = const(0x0020)
# org.bluetooth.service.environmental_sensing
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
# org.bluetooth.characteristic.temperature
_TEMP_CHAR = (
bluetooth.UUID(0x2A6E),
_FLAG_READ | _FLAG_NOTIFY | _FLAG_INDICATE,
)
_ENV_SENSE_SERVICE = (
_ENV_SENSE_UUID,
(_TEMP_CHAR,),
)
# org.bluetooth.characteristic.gap.appearance.xml
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
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((_ENV_SENSE_SERVICE,))
self._connections = set()
self._payload = advertising_payload(
name=name,
services=[_ENV_SENSE_UUID],
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
)
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)
self.led.on()
elif event == _IRQ_CENTRAL_DISCONNECT:
conn_handle, _, _ = data
self._connections.remove(conn_handle)
# Start advertising again to allow a new connection.
self._advertise()
self.led.off()
elif event == _IRQ_GATTS_INDICATE_DONE:
conn_handle, value_handle, status = data
def set_temperature(self, temp_deg_c, notify=False, indicate=False):
# Data is sint16 in degrees Celsius with a resolution of 0.01 degrees Celsius.
# Write the local value, ready for a central to read.
self._ble.gatts_write(self._handle, struct.pack("<h", int(temp_deg_c * 100)))
if notify or indicate:
for conn_handle in self._connections:
if notify:
# Notify connected centrals.
self._ble.gatts_notify(conn_handle, self._handle)
if indicate:
# Indicate connected centrals.
self._ble.gatts_indicate(conn_handle, self._handle)
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)
t = 25
i = 0
while True:
# Write every second, notify every 10 seconds.
i = (i + 1) % 10
temp.set_temperature(t, notify=i == 0, indicate=False)
# Random walk the temperature.
t += random.uniform(-0.5, 0.5)
time.sleep_ms(1000)

View File

@ -1,67 +0,0 @@
# This work is licensed under the MIT license.
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
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())

View File

@ -29,8 +29,8 @@ _LED_SERVICE = (
) )
class BLETemperature: class BLEBlinky:
def __init__(self, ble, name="Giga-H7"): def __init__(self, ble, name="mpy-blinky"):
self._ble = ble self._ble = ble
self._ble.active(True) self._ble.active(True)
self._ble.irq(self._irq) self._ble.irq(self._irq)
@ -59,7 +59,7 @@ class BLETemperature:
if __name__ == "__main__": if __name__ == "__main__":
ble = bluetooth.BLE() ble = bluetooth.BLE()
temp = BLETemperature(ble) temp = BLEBlinky(ble)
while True: while True:
time.sleep_ms(1000) time.sleep_ms(1000)

View File

@ -40,7 +40,7 @@ _ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
class BLETemperature: class BLETemperature:
def __init__(self, ble, name="Giga-H7"): def __init__(self, ble, name="mpy-temp"):
self._ble = ble self._ble = ble
self._ble.active(True) self._ble.active(True)
self._ble.irq(self._irq) self._ble.irq(self._irq)

View File

@ -49,7 +49,7 @@ async def peripheral_task():
while True: while True:
async with await aioble.advertise( async with await aioble.advertise(
_ADV_INTERVAL_MS, _ADV_INTERVAL_MS,
name="Giga-H7", name="mpy-temp",
services=[_ENV_SENSE_UUID], services=[_ENV_SENSE_UUID],
appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER, appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
) as connection: ) as connection: