mirror of
https://github.com/openmv/openmv.git
synced 2025-11-04 14:49:50 +08:00
parent
be45d673a9
commit
8824e93da9
@ -11,11 +11,11 @@
|
|||||||
# OpenMV Cam Ground - Arduino Ground
|
# OpenMV Cam Ground - Arduino Ground
|
||||||
|
|
||||||
import pyb
|
import pyb
|
||||||
import ustruct
|
import struct
|
||||||
|
|
||||||
text = "Hello World!\n"
|
text = "Hello World!\n"
|
||||||
data = ustruct.pack("<%ds" % len(text), text)
|
data = struct.pack("<%ds" % len(text), text)
|
||||||
# Use "ustruct" to build data packets to send.
|
# Use "struct" to build data packets to send.
|
||||||
# "<" puts the data in the struct in little endian order.
|
# "<" 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).
|
# "%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
|
# See https://docs.python.org/3/library/struct.html
|
||||||
@ -42,7 +42,7 @@ print("Waiting for Arduino...")
|
|||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
bus.send(
|
bus.send(
|
||||||
ustruct.pack("<h", len(data)), timeout=10000
|
struct.pack("<h", len(data)), timeout=10000
|
||||||
) # Send the len first (16-bits).
|
) # Send the len first (16-bits).
|
||||||
try:
|
try:
|
||||||
bus.send(data, timeout=10000) # Send the data second.
|
bus.send(data, timeout=10000) # Send the data second.
|
||||||
|
|||||||
@ -13,12 +13,12 @@
|
|||||||
# OpenMV Cam Ground - Arduino Ground
|
# OpenMV Cam Ground - Arduino Ground
|
||||||
|
|
||||||
import pyb
|
import pyb
|
||||||
import ustruct
|
import struct
|
||||||
import time
|
import time
|
||||||
|
|
||||||
text = "Hello World!\n"
|
text = "Hello World!\n"
|
||||||
data = ustruct.pack("<bi%ds" % len(text), 85, len(text), text) # 85 is a sync char.
|
data = struct.pack("<bi%ds" % len(text), 85, len(text), text) # 85 is a sync char.
|
||||||
# Use "ustruct" to build data packets to send.
|
# Use "struct" to build data packets to send.
|
||||||
# "<" puts the data in the struct in little endian order.
|
# "<" puts the data in the struct in little endian order.
|
||||||
# "b" puts a signed char in the data stream.
|
# "b" puts a signed char in the data stream.
|
||||||
# "i" puts a signed integer in the data stream.
|
# "i" puts a signed integer in the data stream.
|
||||||
|
|||||||
@ -4,7 +4,7 @@
|
|||||||
#
|
#
|
||||||
from micropython import const
|
from micropython import const
|
||||||
|
|
||||||
import uasyncio as asyncio
|
import asyncio
|
||||||
import aioble
|
import aioble
|
||||||
import bluetooth
|
import bluetooth
|
||||||
|
|
||||||
|
|||||||
@ -25,7 +25,7 @@
|
|||||||
# f.write(img)
|
# f.write(img)
|
||||||
|
|
||||||
import sensor
|
import sensor
|
||||||
import ustruct
|
import struct
|
||||||
from pyb import USB_VCP
|
from pyb import USB_VCP
|
||||||
|
|
||||||
usb = USB_VCP()
|
usb = USB_VCP()
|
||||||
@ -38,5 +38,5 @@ while True:
|
|||||||
cmd = usb.recv(4, timeout=5000)
|
cmd = usb.recv(4, timeout=5000)
|
||||||
if cmd == b"snap":
|
if cmd == b"snap":
|
||||||
img = sensor.snapshot().to_jpeg()
|
img = sensor.snapshot().to_jpeg()
|
||||||
usb.send(ustruct.pack("<L", img.size()))
|
usb.send(struct.pack("<L", img.size()))
|
||||||
usb.send(img)
|
usb.send(img)
|
||||||
|
|||||||
@ -25,7 +25,7 @@
|
|||||||
# f.write(img)
|
# f.write(img)
|
||||||
|
|
||||||
import sensor
|
import sensor
|
||||||
import ustruct
|
import struct
|
||||||
from pyb import USB_VCP
|
from pyb import USB_VCP
|
||||||
|
|
||||||
usb = USB_VCP()
|
usb = USB_VCP()
|
||||||
@ -38,5 +38,5 @@ while True:
|
|||||||
cmd = usb.recv(4, timeout=5000)
|
cmd = usb.recv(4, timeout=5000)
|
||||||
if cmd == b"snap":
|
if cmd == b"snap":
|
||||||
img = sensor.snapshot().to_jpeg()
|
img = sensor.snapshot().to_jpeg()
|
||||||
usb.send(ustruct.pack("<L", img.size()))
|
usb.send(struct.pack("<L", img.size()))
|
||||||
usb.send(img)
|
usb.send(img)
|
||||||
|
|||||||
@ -25,7 +25,7 @@
|
|||||||
# f.write(img)
|
# f.write(img)
|
||||||
|
|
||||||
import sensor
|
import sensor
|
||||||
import ustruct
|
import struct
|
||||||
from pyb import USB_VCP
|
from pyb import USB_VCP
|
||||||
|
|
||||||
usb = USB_VCP()
|
usb = USB_VCP()
|
||||||
@ -38,5 +38,5 @@ while True:
|
|||||||
cmd = usb.recv(4, timeout=5000)
|
cmd = usb.recv(4, timeout=5000)
|
||||||
if cmd == b"snap":
|
if cmd == b"snap":
|
||||||
img = sensor.snapshot().to_jpeg()
|
img = sensor.snapshot().to_jpeg()
|
||||||
usb.send(ustruct.pack("<L", img.size()))
|
usb.send(struct.pack("<L", img.size()))
|
||||||
usb.send(img)
|
usb.send(img)
|
||||||
|
|||||||
@ -4,7 +4,7 @@
|
|||||||
#
|
#
|
||||||
# Ethernet LAN HTTP client example.
|
# Ethernet LAN HTTP client example.
|
||||||
import network
|
import network
|
||||||
import usocket
|
import socket
|
||||||
|
|
||||||
PORT = 80
|
PORT = 80
|
||||||
HOST = "www.google.com"
|
HOST = "www.google.com"
|
||||||
@ -17,11 +17,11 @@ lan.ifconfig("dhcp")
|
|||||||
print(lan.ifconfig())
|
print(lan.ifconfig())
|
||||||
|
|
||||||
# Get addr info via DNS
|
# Get addr info via DNS
|
||||||
addr = usocket.getaddrinfo(HOST, PORT)[0][4]
|
addr = socket.getaddrinfo(HOST, PORT)[0][4]
|
||||||
print(addr)
|
print(addr)
|
||||||
|
|
||||||
# Create a new socket and connect to 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)
|
client.connect(addr)
|
||||||
|
|
||||||
# Set timeout
|
# Set timeout
|
||||||
|
|||||||
@ -4,8 +4,8 @@
|
|||||||
#
|
#
|
||||||
# Ethernet LAN HTTP client example.
|
# Ethernet LAN HTTP client example.
|
||||||
import network
|
import network
|
||||||
import usocket
|
import socket
|
||||||
import ussl
|
import ssl
|
||||||
|
|
||||||
PORT = 443
|
PORT = 443
|
||||||
HOST = "www.google.com"
|
HOST = "www.google.com"
|
||||||
@ -18,16 +18,16 @@ lan.ifconfig("dhcp")
|
|||||||
print(lan.ifconfig())
|
print(lan.ifconfig())
|
||||||
|
|
||||||
# Get addr info via DNS
|
# Get addr info via DNS
|
||||||
addr = usocket.getaddrinfo(HOST, PORT)[0][4]
|
addr = socket.getaddrinfo(HOST, PORT)[0][4]
|
||||||
print(addr)
|
print(addr)
|
||||||
|
|
||||||
# Create a new socket and connect to 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)
|
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"
|
||||||
|
|||||||
@ -25,7 +25,7 @@
|
|||||||
# f.write(img)
|
# f.write(img)
|
||||||
|
|
||||||
import sensor
|
import sensor
|
||||||
import ustruct
|
import struct
|
||||||
from pyb import USB_VCP
|
from pyb import USB_VCP
|
||||||
|
|
||||||
usb = USB_VCP()
|
usb = USB_VCP()
|
||||||
@ -38,5 +38,5 @@ while True:
|
|||||||
cmd = usb.recv(4, timeout=5000)
|
cmd = usb.recv(4, timeout=5000)
|
||||||
if cmd == b"snap":
|
if cmd == b"snap":
|
||||||
img = sensor.snapshot().to_jpeg()
|
img = sensor.snapshot().to_jpeg()
|
||||||
usb.send(ustruct.pack("<L", img.size()))
|
usb.send(struct.pack("<L", img.size()))
|
||||||
usb.send(img)
|
usb.send(img)
|
||||||
|
|||||||
@ -2,8 +2,8 @@
|
|||||||
# 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
|
||||||
#
|
#
|
||||||
import utime
|
import time
|
||||||
import ustruct
|
import struct
|
||||||
|
|
||||||
|
|
||||||
class PCA9685:
|
class PCA9685:
|
||||||
@ -29,14 +29,14 @@ class PCA9685:
|
|||||||
self._write(0x00, (old_mode & 0x7F) | 0x10) # Mode 1, sleep
|
self._write(0x00, (old_mode & 0x7F) | 0x10) # Mode 1, sleep
|
||||||
self._write(0xFE, prescale) # Prescale
|
self._write(0xFE, prescale) # Prescale
|
||||||
self._write(0x00, old_mode) # Mode 1
|
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
|
self._write(0x00, old_mode | 0xA1) # Mode 1, autoincrement on
|
||||||
|
|
||||||
def pwm(self, index, on=None, off=None):
|
def pwm(self, index, on=None, off=None):
|
||||||
if on is None or off is None:
|
if on is None or off is None:
|
||||||
data = self.i2c.readfrom_mem(self.address, 0x06 + 4 * index, 4)
|
data = self.i2c.readfrom_mem(self.address, 0x06 + 4 * index, 4)
|
||||||
return ustruct.unpack("<HH", data)
|
return struct.unpack("<HH", data)
|
||||||
data = ustruct.pack("<HH", on, off)
|
data = struct.pack("<HH", on, off)
|
||||||
self.i2c.writeto_mem(self.address, 0x06 + 4 * index, data)
|
self.i2c.writeto_mem(self.address, 0x06 + 4 * index, data)
|
||||||
|
|
||||||
def duty(self, index, value=None, invert=False):
|
def duty(self, index, value=None, invert=False):
|
||||||
|
|||||||
@ -12,7 +12,7 @@
|
|||||||
#
|
#
|
||||||
# LoRa library for Arduino Portenta.
|
# LoRa library for Arduino Portenta.
|
||||||
|
|
||||||
from utime import sleep_ms, ticks_ms
|
from time import sleep_ms, ticks_ms
|
||||||
from pyb import UART, Pin
|
from pyb import UART, Pin
|
||||||
|
|
||||||
MODE_ABP = 0
|
MODE_ABP = 0
|
||||||
|
|||||||
@ -2,6 +2,7 @@ import micropython
|
|||||||
import array
|
import array
|
||||||
import uctypes
|
import uctypes
|
||||||
|
|
||||||
|
|
||||||
micropython.alloc_emergency_exception_buf(100)
|
micropython.alloc_emergency_exception_buf(100)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@ -20,14 +20,13 @@
|
|||||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||||
# THE SOFTWARE.
|
# THE SOFTWARE.
|
||||||
|
|
||||||
# Source: improved version of:
|
# Source: improved version of micropython-lib's requests.
|
||||||
# https://github.com/micropython/micropython-lib/blob/master/python-ecosys/urequests/urequests.py
|
|
||||||
# Some useful links for future updates:
|
# Some useful links for future updates:
|
||||||
# https://www.w3.org/TR/html401/interact/forms.html#h-17.13.4
|
# https://www.w3.org/TR/html401/interact/forms.html#h-17.13.4
|
||||||
# https://docs.python-requests.org/en/master/
|
# https://docs.python-requests.org/en/master/
|
||||||
|
|
||||||
import usocket
|
import socket
|
||||||
import ubinascii
|
import binascii
|
||||||
|
|
||||||
|
|
||||||
class Response:
|
class Response:
|
||||||
@ -47,9 +46,9 @@ class Response:
|
|||||||
return str(self._content, self.encoding)
|
return str(self._content, self.encoding)
|
||||||
|
|
||||||
def json(self):
|
def json(self):
|
||||||
import ujson
|
import json
|
||||||
|
|
||||||
return ujson.loads(self._content)
|
return json.loads(self._content)
|
||||||
|
|
||||||
|
|
||||||
def readline(s):
|
def readline(s):
|
||||||
@ -87,7 +86,7 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
|
|||||||
if proto == "http:":
|
if proto == "http:":
|
||||||
port = 80
|
port = 80
|
||||||
elif proto == "https:":
|
elif proto == "https:":
|
||||||
import ussl
|
import ssl
|
||||||
|
|
||||||
port = 443
|
port = 443
|
||||||
else:
|
else:
|
||||||
@ -99,20 +98,20 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
|
|||||||
|
|
||||||
if auth:
|
if auth:
|
||||||
headers["Authorization"] = b"Basic %s" % (
|
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_code = 0
|
||||||
resp_reason = None
|
resp_reason = None
|
||||||
resp_headers = []
|
resp_headers = []
|
||||||
|
|
||||||
ai = usocket.getaddrinfo(host, port)[0]
|
ai = socket.getaddrinfo(host, port)[0]
|
||||||
s = usocket.socket(ai[0], ai[1], ai[2])
|
s = socket.socket(ai[0], ai[1], ai[2])
|
||||||
try:
|
try:
|
||||||
s.connect(ai[-1])
|
s.connect(ai[-1])
|
||||||
s.settimeout(5.0)
|
s.settimeout(5.0)
|
||||||
if proto == "https:":
|
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))
|
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")
|
s.write(b"\r\n")
|
||||||
|
|
||||||
if json is not None:
|
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")
|
s.write(b"Content-Type: application/json\r\n")
|
||||||
|
|
||||||
if files is not None:
|
if files is not None:
|
||||||
@ -20,7 +20,7 @@
|
|||||||
#
|
#
|
||||||
import socket
|
import socket
|
||||||
import network
|
import network
|
||||||
import uos
|
import os
|
||||||
import gc
|
import gc
|
||||||
import sys
|
import sys
|
||||||
import errno
|
import errno
|
||||||
@ -79,12 +79,12 @@ class FTP_client:
|
|||||||
|
|
||||||
def send_list_data(self, path, data_client, full):
|
def send_list_data(self, path, data_client, full):
|
||||||
try:
|
try:
|
||||||
for fname in uos.listdir(path):
|
for fname in os.listdir(path):
|
||||||
data_client.sendall(self.make_description(path, fname, full))
|
data_client.sendall(self.make_description(path, fname, full))
|
||||||
except Exception: # path may be a file name or pattern
|
except Exception: # path may be a file name or pattern
|
||||||
path, pattern = self.split_path(path)
|
path, pattern = self.split_path(path)
|
||||||
try:
|
try:
|
||||||
for fname in uos.listdir(path):
|
for fname in os.listdir(path):
|
||||||
if self.fncmp(fname, pattern):
|
if self.fncmp(fname, pattern):
|
||||||
data_client.sendall(self.make_description(path, fname, full))
|
data_client.sendall(self.make_description(path, fname, full))
|
||||||
except:
|
except:
|
||||||
@ -93,7 +93,7 @@ class FTP_client:
|
|||||||
def make_description(self, path, fname, full):
|
def make_description(self, path, fname, full):
|
||||||
global _month_name
|
global _month_name
|
||||||
if full:
|
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_permissions = "drwxr-xr-x" if (stat[0] & 0o170000 == 0o040000) else "-rw-r--r--"
|
||||||
file_size = stat[6]
|
file_size = stat[6]
|
||||||
tm = stat[7] & 0xFFFFFFFF
|
tm = stat[7] & 0xFFFFFFFF
|
||||||
@ -246,7 +246,7 @@ class FTP_client:
|
|||||||
cl.sendall('257 "{}"\r\n'.format(self.cwd))
|
cl.sendall('257 "{}"\r\n'.format(self.cwd))
|
||||||
elif command == "CWD" or command == "XCWD":
|
elif command == "CWD" or command == "XCWD":
|
||||||
try:
|
try:
|
||||||
if (uos.stat(path)[0] & 0o170000) == 0o040000:
|
if (os.stat(path)[0] & 0o170000) == 0o040000:
|
||||||
self.cwd = path
|
self.cwd = path
|
||||||
cl.sendall("250 OK\r\n")
|
cl.sendall("250 OK\r\n")
|
||||||
else:
|
else:
|
||||||
@ -316,12 +316,12 @@ class FTP_client:
|
|||||||
data_client.close()
|
data_client.close()
|
||||||
elif command == "SIZE":
|
elif command == "SIZE":
|
||||||
try:
|
try:
|
||||||
cl.sendall("213 {}\r\n".format(uos.stat(path)[6]))
|
cl.sendall("213 {}\r\n".format(os.stat(path)[6]))
|
||||||
except:
|
except:
|
||||||
cl.sendall("550 Fail\r\n")
|
cl.sendall("550 Fail\r\n")
|
||||||
elif command == "MDTM":
|
elif command == "MDTM":
|
||||||
try:
|
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]))
|
cl.sendall("213 {:04d}{:02d}{:02d}{:02d}{:02d}{:02d}\r\n".format(*tm[0:6]))
|
||||||
except:
|
except:
|
||||||
cl.sendall("550 Fail\r\n")
|
cl.sendall("550 Fail\r\n")
|
||||||
@ -345,21 +345,21 @@ class FTP_client:
|
|||||||
cl.sendall("213 Done.\r\n")
|
cl.sendall("213 Done.\r\n")
|
||||||
elif command == "DELE":
|
elif command == "DELE":
|
||||||
try:
|
try:
|
||||||
uos.remove(path)
|
os.remove(path)
|
||||||
cl.sendall("250 OK\r\n")
|
cl.sendall("250 OK\r\n")
|
||||||
except:
|
except:
|
||||||
cl.sendall("550 Fail\r\n")
|
cl.sendall("550 Fail\r\n")
|
||||||
elif command == "RNFR":
|
elif command == "RNFR":
|
||||||
try:
|
try:
|
||||||
# just test if the name exists, exception if not
|
# just test if the name exists, exception if not
|
||||||
uos.stat(path)
|
os.stat(path)
|
||||||
self.fromname = path
|
self.fromname = path
|
||||||
cl.sendall("350 Rename from\r\n")
|
cl.sendall("350 Rename from\r\n")
|
||||||
except:
|
except:
|
||||||
cl.sendall("550 Fail\r\n")
|
cl.sendall("550 Fail\r\n")
|
||||||
elif command == "RNTO":
|
elif command == "RNTO":
|
||||||
try:
|
try:
|
||||||
uos.rename(self.fromname, path)
|
os.rename(self.fromname, path)
|
||||||
cl.sendall("250 OK\r\n")
|
cl.sendall("250 OK\r\n")
|
||||||
except:
|
except:
|
||||||
cl.sendall("550 Fail\r\n")
|
cl.sendall("550 Fail\r\n")
|
||||||
@ -369,13 +369,13 @@ class FTP_client:
|
|||||||
cl.sendall("250 OK\r\n")
|
cl.sendall("250 OK\r\n")
|
||||||
elif command == "RMD" or command == "XRMD":
|
elif command == "RMD" or command == "XRMD":
|
||||||
try:
|
try:
|
||||||
uos.rmdir(path)
|
os.rmdir(path)
|
||||||
cl.sendall("250 OK\r\n")
|
cl.sendall("250 OK\r\n")
|
||||||
except:
|
except:
|
||||||
cl.sendall("550 Fail\r\n")
|
cl.sendall("550 Fail\r\n")
|
||||||
elif command == "MKD" or command == "XMKD":
|
elif command == "MKD" or command == "XMKD":
|
||||||
try:
|
try:
|
||||||
uos.mkdir(path)
|
os.mkdir(path)
|
||||||
cl.sendall("250 OK\r\n")
|
cl.sendall("250 OK\r\n")
|
||||||
except:
|
except:
|
||||||
cl.sendall("550 Fail\r\n")
|
cl.sendall("550 Fail\r\n")
|
||||||
|
|||||||
@ -1,6 +1,12 @@
|
|||||||
# µPing (MicroPing) for MicroPython
|
# µPing (MicroPing) for MicroPython
|
||||||
# copyright (c) 2018 Shawwwn <shawwwn1@gmail.com>
|
# copyright (c) 2018 Shawwwn <shawwwn1@gmail.com>
|
||||||
# License: MIT
|
# License: MIT
|
||||||
|
import time
|
||||||
|
import select
|
||||||
|
import socket
|
||||||
|
import struct
|
||||||
|
import random
|
||||||
|
import uctypes
|
||||||
|
|
||||||
|
|
||||||
# Internet Checksum Algorithm
|
# Internet Checksum Algorithm
|
||||||
@ -22,13 +28,6 @@ def checksum(data):
|
|||||||
|
|
||||||
|
|
||||||
def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
|
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
|
# prepare packet
|
||||||
assert size >= 16, "pkt size too small"
|
assert size >= 16, "pkt size too small"
|
||||||
pkt = b"Q" * size
|
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.type = 8 # ICMP_ECHO_REQUEST
|
||||||
h.code = 0
|
h.code = 0
|
||||||
h.checksum = 0
|
h.checksum = 0
|
||||||
h.id = urandom.randint(0, 65535)
|
h.id = random.randint(0, 65535)
|
||||||
h.seq = 1
|
h.seq = 1
|
||||||
|
|
||||||
# init socket
|
# 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.setblocking(0)
|
||||||
sock.settimeout(timeout / 1000)
|
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))
|
sock.connect((addr, 1))
|
||||||
not quiet and print("PING %s (%s): %u data bytes" % (host, addr, len(pkt)))
|
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
|
# send packet
|
||||||
h.checksum = 0
|
h.checksum = 0
|
||||||
h.seq = c
|
h.seq = c
|
||||||
h.timestamp = utime.ticks_us()
|
h.timestamp = time.ticks_us()
|
||||||
h.checksum = checksum(pkt)
|
h.checksum = checksum(pkt)
|
||||||
if sock.send(pkt) == size:
|
if sock.send(pkt) == size:
|
||||||
n_trans += 1
|
n_trans += 1
|
||||||
@ -77,7 +76,7 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
|
|||||||
|
|
||||||
# recv packet
|
# recv packet
|
||||||
while 1:
|
while 1:
|
||||||
socks, _, _ = uselect.select([sock], [], [], 0)
|
socks, _, _ = select.select([sock], [], [], 0)
|
||||||
if socks:
|
if socks:
|
||||||
resp = socks[0].recv(4096)
|
resp = socks[0].recv(4096)
|
||||||
resp_mv = memoryview(resp)
|
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)
|
# TODO: validate checksum (optional)
|
||||||
seq = h2.seq
|
seq = h2.seq
|
||||||
if h2.type == 0 and h2.id == h.id and (seq in seqs): # 0: ICMP_ECHO_REPLY
|
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
|
t_elasped = (time.ticks_us() - h2.timestamp) / 1000
|
||||||
ttl = ustruct.unpack("!B", resp_mv[8:9])[0] # time-to-live
|
ttl = struct.unpack("!B", resp_mv[8:9])[0] # time-to-live
|
||||||
n_recv += 1
|
n_recv += 1
|
||||||
not quiet and print(
|
not quiet and print(
|
||||||
"%u bytes from %s: icmp_seq=%u, ttl=%u, time=%f ms"
|
"%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:
|
if finish:
|
||||||
break
|
break
|
||||||
|
|
||||||
utime.sleep_ms(1)
|
time.sleep_ms(1)
|
||||||
t += 1
|
t += 1
|
||||||
|
|
||||||
# close
|
# close
|
||||||
|
|||||||
@ -3,9 +3,9 @@
|
|||||||
|
|
||||||
import socket
|
import socket
|
||||||
import network
|
import network
|
||||||
import uos
|
import os
|
||||||
import errno
|
import errno
|
||||||
from uio import IOBase
|
from io import IOBase
|
||||||
|
|
||||||
last_client_socket = None
|
last_client_socket = None
|
||||||
server_socket = None
|
server_socket = None
|
||||||
@ -73,24 +73,24 @@ def accept_telnet_connect(telnet_server):
|
|||||||
|
|
||||||
if last_client_socket:
|
if last_client_socket:
|
||||||
# close any previous clients
|
# close any previous clients
|
||||||
uos.dupterm(None)
|
os.dupterm(None)
|
||||||
last_client_socket.close()
|
last_client_socket.close()
|
||||||
|
|
||||||
last_client_socket, remote_addr = telnet_server.accept()
|
last_client_socket, remote_addr = telnet_server.accept()
|
||||||
print("Telnet connection from:", remote_addr)
|
print("Telnet connection from:", remote_addr)
|
||||||
last_client_socket.setblocking(False)
|
last_client_socket.setblocking(False)
|
||||||
# dupterm_notify() not available under MicroPython v1.1
|
# 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, 252, 34])) # dont allow line mode
|
||||||
last_client_socket.sendall(bytes([255, 251, 1])) # turn off local echo
|
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():
|
def stop():
|
||||||
global server_socket, last_client_socket
|
global server_socket, last_client_socket
|
||||||
uos.dupterm(None)
|
os.dupterm(None)
|
||||||
if server_socket:
|
if server_socket:
|
||||||
server_socket.close()
|
server_socket.close()
|
||||||
if last_client_socket:
|
if last_client_socket:
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user