Merge pull request #2429 from openmv/fix_examples_umod

scripts: Remove obsolete u-prefix from imports.
This commit is contained in:
Ibrahim Abdelkader 2024-09-30 14:28:01 +03:00 committed by GitHub
commit eb3f30b088
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 119 additions and 77 deletions

View File

@ -29,7 +29,7 @@ user: admin
password: testadmin
```
Examples for post requests can be found at `scripts/examples/OpenMV/14-WiFi-Shield/http_post.py`.
Examples for post requests can be found at `scripts/examples/09-WiFi/http_post.py`.
## GET Requests

View File

@ -11,11 +11,11 @@
# OpenMV Cam Ground - Arduino Ground
import pyb
import ustruct
import struct
text = "Hello World!\n"
data = ustruct.pack("<%ds" % len(text), text)
# Use "ustruct" to build data packets to send.
data = struct.pack("<%ds" % len(text), text)
# Use "struct" to build data packets to send.
# "<" puts the data in the struct in little endian order.
# "%ds" puts a string in the data stream. E.g. "13s" for "Hello World!\n" (13 chars).
# See https://docs.python.org/3/library/struct.html
@ -42,7 +42,7 @@ print("Waiting for Arduino...")
while True:
try:
bus.send(
ustruct.pack("<h", len(data)), timeout=10000
struct.pack("<h", len(data)), timeout=10000
) # Send the len first (16-bits).
try:
bus.send(data, timeout=10000) # Send the data second.

View File

@ -13,12 +13,12 @@
# OpenMV Cam Ground - Arduino Ground
import pyb
import ustruct
import struct
import time
text = "Hello World!\n"
data = ustruct.pack("<bi%ds" % len(text), 85, len(text), text) # 85 is a sync char.
# Use "ustruct" to build data packets to send.
data = struct.pack("<bi%ds" % len(text), 85, len(text), text) # 85 is a sync char.
# Use "struct" to build data packets to send.
# "<" puts the data in the struct in little endian order.
# "b" puts a signed char in the data stream.
# "i" puts a signed integer in the data stream.

View File

@ -0,0 +1,43 @@
# 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 requests module example
import network
import requests
# AP info
SSID = "" # Network SSID
KEY = "" # Network key
URL = 'http://192.168.1.102:8080/upload'
# 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())
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')),
}
r = requests.post(URL, files=files, headers=headers, auth=('admin', 'testadmin'))
# Or send JSON data
# r = requests.post(URL, json={'some': 'data'}, headers=headers, auth=('admin', 'testadmin'))
print(r.status_code, r.reason)
print(r.headers, r.content)

View File

@ -4,7 +4,7 @@
#
from micropython import const
import uasyncio as asyncio
import asyncio
import aioble
import bluetooth

View File

@ -25,7 +25,7 @@
# f.write(img)
import sensor
import ustruct
import struct
from pyb import USB_VCP
usb = USB_VCP()
@ -38,5 +38,5 @@ while True:
cmd = usb.recv(4, timeout=5000)
if cmd == b"snap":
img = sensor.snapshot().to_jpeg()
usb.send(ustruct.pack("<L", img.size()))
usb.send(struct.pack("<L", img.size()))
usb.send(img)

View File

@ -25,7 +25,7 @@
# f.write(img)
import sensor
import ustruct
import struct
from pyb import USB_VCP
usb = USB_VCP()
@ -38,5 +38,5 @@ while True:
cmd = usb.recv(4, timeout=5000)
if cmd == b"snap":
img = sensor.snapshot().to_jpeg()
usb.send(ustruct.pack("<L", img.size()))
usb.send(struct.pack("<L", img.size()))
usb.send(img)

View File

@ -25,7 +25,7 @@
# f.write(img)
import sensor
import ustruct
import struct
from pyb import USB_VCP
usb = USB_VCP()
@ -38,5 +38,5 @@ while True:
cmd = usb.recv(4, timeout=5000)
if cmd == b"snap":
img = sensor.snapshot().to_jpeg()
usb.send(ustruct.pack("<L", img.size()))
usb.send(struct.pack("<L", img.size()))
usb.send(img)

View File

@ -4,7 +4,7 @@
#
# Ethernet LAN HTTP client example.
import network
import usocket
import socket
PORT = 80
HOST = "www.google.com"
@ -17,11 +17,11 @@ lan.ifconfig("dhcp")
print(lan.ifconfig())
# Get addr info via DNS
addr = usocket.getaddrinfo(HOST, PORT)[0][4]
addr = socket.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 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(addr)
# Set timeout

View File

@ -4,8 +4,8 @@
#
# Ethernet LAN HTTP client example.
import network
import usocket
import ussl
import socket
import ssl
PORT = 443
HOST = "www.google.com"
@ -18,16 +18,16 @@ lan.ifconfig("dhcp")
print(lan.ifconfig())
# Get addr info via DNS
addr = usocket.getaddrinfo(HOST, PORT)[0][4]
addr = socket.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 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(addr)
# Set timeout
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
request = "GET / HTTP/1.1\r\n"

View File

@ -25,7 +25,7 @@
# f.write(img)
import sensor
import ustruct
import struct
from pyb import USB_VCP
usb = USB_VCP()
@ -38,5 +38,5 @@ while True:
cmd = usb.recv(4, timeout=5000)
if cmd == b"snap":
img = sensor.snapshot().to_jpeg()
usb.send(ustruct.pack("<L", img.size()))
usb.send(struct.pack("<L", img.size()))
usb.send(img)

View File

@ -2,8 +2,8 @@
# Copyright (c) 2013-2023 OpenMV LLC. All rights reserved.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
import utime
import ustruct
import time
import struct
class PCA9685:
@ -29,14 +29,14 @@ class PCA9685:
self._write(0x00, (old_mode & 0x7F) | 0x10) # Mode 1, sleep
self._write(0xFE, prescale) # Prescale
self._write(0x00, old_mode) # Mode 1
utime.sleep_us(5)
time.sleep_us(5)
self._write(0x00, old_mode | 0xA1) # Mode 1, autoincrement on
def pwm(self, index, on=None, off=None):
if on is None or off is None:
data = self.i2c.readfrom_mem(self.address, 0x06 + 4 * index, 4)
return ustruct.unpack("<HH", data)
data = ustruct.pack("<HH", on, off)
return struct.unpack("<HH", data)
data = struct.pack("<HH", on, off)
self.i2c.writeto_mem(self.address, 0x06 + 4 * index, data)
def duty(self, index, value=None, invert=False):

View File

@ -12,7 +12,7 @@
#
# LoRa library for Arduino Portenta.
from utime import sleep_ms, ticks_ms
from time import sleep_ms, ticks_ms
from pyb import UART, Pin
MODE_ABP = 0

View File

@ -2,6 +2,7 @@ import micropython
import array
import uctypes
micropython.alloc_emergency_exception_buf(100)

View File

@ -20,14 +20,13 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
# Source: improved version of:
# https://github.com/micropython/micropython-lib/blob/master/python-ecosys/urequests/urequests.py
# Source: improved version of micropython-lib's requests.
# Some useful links for future updates:
# https://www.w3.org/TR/html401/interact/forms.html#h-17.13.4
# https://docs.python-requests.org/en/master/
import usocket
import ubinascii
import socket
import binascii
class Response:
@ -47,9 +46,9 @@ class Response:
return str(self._content, self.encoding)
def json(self):
import ujson
import json
return ujson.loads(self._content)
return json.loads(self._content)
def readline(s):
@ -87,7 +86,7 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
if proto == "http:":
port = 80
elif proto == "https:":
import ussl
import ssl
port = 443
else:
@ -99,20 +98,20 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
if auth:
headers["Authorization"] = b"Basic %s" % (
ubinascii.b2a_base64("%s:%s" % (auth[0], auth[1]))[0:-1]
binascii.b2a_base64("%s:%s" % (auth[0], auth[1]))[0:-1]
)
resp_code = 0
resp_reason = None
resp_headers = []
ai = usocket.getaddrinfo(host, port)[0]
s = usocket.socket(ai[0], ai[1], ai[2])
ai = socket.getaddrinfo(host, port)[0]
s = socket.socket(ai[0], ai[1], ai[2])
try:
s.connect(ai[-1])
s.settimeout(5.0)
if proto == "https:":
s = ussl.wrap_socket(s, server_hostname=host)
s = ssl.wrap_socket(s, server_hostname=host)
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
@ -127,9 +126,9 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
s.write(b"\r\n")
if json is not None:
import ujson
import json
data = ujson.dumps(json)
data = json.dumps(json)
s.write(b"Content-Type: application/json\r\n")
if files is not None:

View File

@ -20,7 +20,7 @@
#
import socket
import network
import uos
import os
import gc
import sys
import errno
@ -79,12 +79,12 @@ class FTP_client:
def send_list_data(self, path, data_client, full):
try:
for fname in uos.listdir(path):
for fname in os.listdir(path):
data_client.sendall(self.make_description(path, fname, full))
except Exception: # path may be a file name or pattern
path, pattern = self.split_path(path)
try:
for fname in uos.listdir(path):
for fname in os.listdir(path):
if self.fncmp(fname, pattern):
data_client.sendall(self.make_description(path, fname, full))
except:
@ -93,7 +93,7 @@ class FTP_client:
def make_description(self, path, fname, full):
global _month_name
if full:
stat = uos.stat(self.get_absolute_path(path, fname))
stat = os.stat(self.get_absolute_path(path, fname))
file_permissions = "drwxr-xr-x" if (stat[0] & 0o170000 == 0o040000) else "-rw-r--r--"
file_size = stat[6]
tm = stat[7] & 0xFFFFFFFF
@ -246,7 +246,7 @@ class FTP_client:
cl.sendall('257 "{}"\r\n'.format(self.cwd))
elif command == "CWD" or command == "XCWD":
try:
if (uos.stat(path)[0] & 0o170000) == 0o040000:
if (os.stat(path)[0] & 0o170000) == 0o040000:
self.cwd = path
cl.sendall("250 OK\r\n")
else:
@ -316,12 +316,12 @@ class FTP_client:
data_client.close()
elif command == "SIZE":
try:
cl.sendall("213 {}\r\n".format(uos.stat(path)[6]))
cl.sendall("213 {}\r\n".format(os.stat(path)[6]))
except:
cl.sendall("550 Fail\r\n")
elif command == "MDTM":
try:
tm = localtime(uos.stat(path)[8])
tm = localtime(os.stat(path)[8])
cl.sendall("213 {:04d}{:02d}{:02d}{:02d}{:02d}{:02d}\r\n".format(*tm[0:6]))
except:
cl.sendall("550 Fail\r\n")
@ -345,21 +345,21 @@ class FTP_client:
cl.sendall("213 Done.\r\n")
elif command == "DELE":
try:
uos.remove(path)
os.remove(path)
cl.sendall("250 OK\r\n")
except:
cl.sendall("550 Fail\r\n")
elif command == "RNFR":
try:
# just test if the name exists, exception if not
uos.stat(path)
os.stat(path)
self.fromname = path
cl.sendall("350 Rename from\r\n")
except:
cl.sendall("550 Fail\r\n")
elif command == "RNTO":
try:
uos.rename(self.fromname, path)
os.rename(self.fromname, path)
cl.sendall("250 OK\r\n")
except:
cl.sendall("550 Fail\r\n")
@ -369,13 +369,13 @@ class FTP_client:
cl.sendall("250 OK\r\n")
elif command == "RMD" or command == "XRMD":
try:
uos.rmdir(path)
os.rmdir(path)
cl.sendall("250 OK\r\n")
except:
cl.sendall("550 Fail\r\n")
elif command == "MKD" or command == "XMKD":
try:
uos.mkdir(path)
os.mkdir(path)
cl.sendall("250 OK\r\n")
except:
cl.sendall("550 Fail\r\n")

View File

@ -1,6 +1,12 @@
# µPing (MicroPing) for MicroPython
# copyright (c) 2018 Shawwwn <shawwwn1@gmail.com>
# License: MIT
import time
import select
import socket
import struct
import random
import uctypes
# Internet Checksum Algorithm
@ -22,13 +28,6 @@ def checksum(data):
def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
import utime
import uselect
import uctypes
import usocket
import ustruct
import urandom
# prepare packet
assert size >= 16, "pkt size too small"
pkt = b"Q" * size
@ -44,14 +43,14 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
h.type = 8 # ICMP_ECHO_REQUEST
h.code = 0
h.checksum = 0
h.id = urandom.randint(0, 65535)
h.id = random.randint(0, 65535)
h.seq = 1
# init socket
sock = usocket.socket(usocket.AF_INET, usocket.SOCK_RAW, 1)
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, 1)
sock.setblocking(0)
sock.settimeout(timeout / 1000)
addr = usocket.getaddrinfo(host, 1)[0][-1][0] # ip address
addr = socket.getaddrinfo(host, 1)[0][-1][0] # ip address
sock.connect((addr, 1))
not quiet and print("PING %s (%s): %u data bytes" % (host, addr, len(pkt)))
@ -66,7 +65,7 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
# send packet
h.checksum = 0
h.seq = c
h.timestamp = utime.ticks_us()
h.timestamp = time.ticks_us()
h.checksum = checksum(pkt)
if sock.send(pkt) == size:
n_trans += 1
@ -77,7 +76,7 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
# recv packet
while 1:
socks, _, _ = uselect.select([sock], [], [], 0)
socks, _, _ = select.select([sock], [], [], 0)
if socks:
resp = socks[0].recv(4096)
resp_mv = memoryview(resp)
@ -85,8 +84,8 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
# TODO: validate checksum (optional)
seq = h2.seq
if h2.type == 0 and h2.id == h.id and (seq in seqs): # 0: ICMP_ECHO_REPLY
t_elasped = (utime.ticks_us() - h2.timestamp) / 1000
ttl = ustruct.unpack("!B", resp_mv[8:9])[0] # time-to-live
t_elasped = (time.ticks_us() - h2.timestamp) / 1000
ttl = struct.unpack("!B", resp_mv[8:9])[0] # time-to-live
n_recv += 1
not quiet and print(
"%u bytes from %s: icmp_seq=%u, ttl=%u, time=%f ms"
@ -102,7 +101,7 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
if finish:
break
utime.sleep_ms(1)
time.sleep_ms(1)
t += 1
# close

View File

@ -3,9 +3,9 @@
import socket
import network
import uos
import os
import errno
from uio import IOBase
from io import IOBase
last_client_socket = None
server_socket = None
@ -73,24 +73,24 @@ def accept_telnet_connect(telnet_server):
if last_client_socket:
# close any previous clients
uos.dupterm(None)
os.dupterm(None)
last_client_socket.close()
last_client_socket, remote_addr = telnet_server.accept()
print("Telnet connection from:", remote_addr)
last_client_socket.setblocking(False)
# dupterm_notify() not available under MicroPython v1.1
# last_client_socket.setsockopt(socket.SOL_SOCKET, 20, uos.dupterm_notify)
# last_client_socket.setsockopt(socket.SOL_SOCKET, 20, os.dupterm_notify)
last_client_socket.sendall(bytes([255, 252, 34])) # dont allow line mode
last_client_socket.sendall(bytes([255, 251, 1])) # turn off local echo
uos.dupterm(TelnetWrapper(last_client_socket))
os.dupterm(TelnetWrapper(last_client_socket))
def stop():
global server_socket, last_client_socket
uos.dupterm(None)
os.dupterm(None)
if server_socket:
server_socket.close()
if last_client_socket: