【得捷电子Follow me第4期】使用micropython完成全部任务
[复制链接]
这次活动使用的开发板w5500-pico,相对来说比较小众。虽然w5500和pico都有非常多的应用,但将他们组合起来,资源就显得捉襟见肘。官方库对micropython有一定支持,但并不完善,比如基本的ping方法都没有。由于年前在讨论群里吹水说年后整个活,那现在就兑现一下吧,全部任务都使用micropython来完成,如果需要造轮子,那就造个轮子。
任务使用树莓派5作为上位机,全部代码编写等操作都在树莓派5中进行完成。搭配官方散热器后,整体运行相当给力。
入门任务:开发环境搭建,BLINK,驱动液晶显示器进行显示(没有则串口HelloWorld)
首先先到micropython官网下载最新版的uf2固件:
下载好后,插入我们的开发板,应该会直接弹出一个u盘,把刚刚下载的uf2复制进去后,u盘消失,这就完成了烧写工作。
接下来我们打开terminal,输入thonny,就会自动打开thonny。点击右下角就可以看到我们的开发板。
这两个选择任意一个,都可以成功连接。
下面开始代码。
# 入门任务:开发环境搭建,BLINK,驱动液晶显示器进行显示(没有则串口HelloWorld)
import machine
import time
led = machine.Pin(25, machine.Pin.OUT)
for i in range(5):
led.high()
time.sleep(0.5)
led.low()
time.sleep(0.5)
print("")
print("Hello EEWorld!")
print("")
print("你好,电子工程世界!")
运行后就可以看到板子上的绿灯闪了5次,闪完后shell输出相应字符。
基础任务一:完成主控板W5500初始化(静态IP配置),并能使用局域网电脑ping通,同时W5500可以ping通互联网站点;通过抓包软件(Wireshark、Sniffer等)抓取本地PC的ping报文,展示并分析
我们把连接网络的功能放进单独的w5500文件中:
from machine import Pin,SPI
import network
import time
def w5500(config): # could be "dhcp" or ('192.168.x.xxx','255.255.255.0','192.168.x.1','8.8.8.8')
#spi init
spi=SPI(0,2_000_000, mosi=Pin(19),miso=Pin(16),sck=Pin(18))
nic = network.WIZNET5K(spi,Pin(17),Pin(20)) #spi,cs,reset pin
nic.active(True)#nicwork active
nic.ifconfig(config)
while not nic.isconnected():
time.sleep(1)
# print(nic.regs())#Print register information
#Print nicwork address information
print("IP Address:",nic.ifconfig()[0])
print("Subnic Mask:",nic.ifconfig()[1])
print("Gateway:",nic.ifconfig()[2])
print("DNS:",nic.ifconfig()[3])
return nic
下面先讲讲ping的实现,ping是基于icmp协议发送的报文,虽然也是通过socket方法实现的,但要注意的是这里我们的socket使用的是SOCK_RAW和ICMP,它的index是1。这里和我们使用常规tcp方法不一样,icmp更为底层。
好在开源代码有不少使用python实现ping方法的案例,让我不需要真的完全从头造轮子。随便找了一个,修改了一下socket的使用,使用uctypes来封装结构体,并且修改了一下原始代码中ping逻辑有问题的地方,就可以完成一个仅依赖于micropython中socket方法的ping。
# μPing (MicroPing) for MicroPython
# copyright (c) 2018 Shawwwn <shawwwn1@gmail.com>
# License: MIT
# Internet Checksum Algorithm
# Author: Olav Morken
# https://github.com/olavmrk/python-ping/blob/master/ping.py
# @data: bytes
def checksum(data):
if len(data) & 0x1: # Odd number of bytes
data += b'\0'
cs = 0
for pos in range(0, len(data), 2):
b1 = data[pos]
b2 = data[pos + 1]
cs += (b1 << 8) + b2
while cs >= 0x10000:
cs = (cs & 0xffff) + (cs >> 16)
cs = ~cs & 0xffff
return cs
def ping(host, count=4, timeout=30, interval=1, quiet=False, size=64):
import time
import select
import uctypes
import socket
import struct
import random
# prepare packet
assert size >= 16, "pkt size too small"
pkt = b'Q'*size
pkt_desc = {
"type": uctypes.UINT8 | 0,
"code": uctypes.UINT8 | 1,
"checksum": uctypes.UINT16 | 2,
"id": uctypes.UINT16 | 4,
"seq": uctypes.INT16 | 6,
"timestamp": uctypes.UINT64 | 8,
} # packet header descriptor
h = uctypes.struct(uctypes.addressof(pkt), pkt_desc, uctypes.BIG_ENDIAN)
h.type = 8 # ICMP_ECHO_REQUEST
h.code = 0
h.checksum = 0
h.id = random.getrandbits(16)
h.seq = 1
# init socket
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, 1)
sock.setblocking(0)
sock.settimeout(timeout/1000)
c = 0
addr = None
while c < 10:
try:
c += 1
time.sleep_ms(100)
addr = socket.getaddrinfo(host, 1)[0][-1][0] # ip address
break
except:
pass
if not addr:
print("DNS lookup failed")
return
sock.connect((addr, 1))
not quiet and print("PING %s (%s): %u data bytes" % (host, addr, len(pkt)))
seqs = list(range(1, count+1)) # [1,2,...,count]
c = 1
t = time.time()
tstart = time.time()
n_trans = 0
n_recv = 0
finish = False
while time.time() - tstart < timeout:
if time.time() - t >= interval and c<=count:
# send packet
h.checksum = 0
h.seq = c
h.timestamp = time.ticks_us()
h.checksum = checksum(pkt)
if sock.send(pkt) == size:
n_trans += 1
t = time.time() # reset timeout
else:
seqs.remove(c)
# recv packet
socks, _, _ = select.select([sock], [], [], 0)
if socks:
resp = socks[0].recv(4096)
resp_mv = memoryview(resp)
h2 = uctypes.struct(uctypes.addressof(resp_mv[20:]), pkt_desc, uctypes.BIG_ENDIAN)
# 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 = (time.ticks_us()-h2.timestamp) / 1000
ttl = struct.unpack('!B', resp_mv[8:9])[0] # time-to-live
n_recv += 1
c += 1
not quiet and print("%u bytes from %s: icmp_seq=%u, ttl=%u, time=%f ms" % (len(resp), addr, seq, ttl, t_elasped))
seqs.remove(seq)
if len(seqs) == 0:
finish = True
if finish:
break
# close
sock.close()
ret = (n_trans, n_recv)
not quiet and print("%u packets transmitted, %u packets received" % (n_trans, n_recv))
return (n_trans, n_recv)
接着在主程序里写入以下代码运行,就可以看到打印。
这里我是用ping网关方法来判断是否成功连接上网络。因为我发现虽然在联网中已经有了connected判断,但实际上即使判断已联网,但并未连接,还需要一段时间。这应该是mpy中驱动的bug。
# 基础任务一:完成主控板W5500初始化(静态IP配置),并能使用局域网电脑ping通,同时W5500可以ping通互联网站点;通过抓包软件(Wireshark、Sniffer等)抓取本地PC的ping报文,展示并分析
from w5500 import w5500
from ping import ping
ip = "192.168.8.213"
gate = ip.split(".")
gate[-1] = "1"
gate = ".".join(gate)
nic = w5500((ip,"255.255.255.0",gate,"8.8.8.8"))
ping(gate, quiet=True)
ping("eeworld.com.cn")
运行后,可以看到以下输出:
这时候我们尝试一下用主机去ping w5500:
抓包之前,我们先用sudo apt install wireshark 来安装一下。为了确保我们使用非root用户在桌面环境下也可以抓包,当安装过程中弹出dumpcap功能时,我们要选择yes启动。如果不小心错过,也可以通过sudo dpkg-reconfigure wireshark-common来开启。
接着我们输入以下命令,配置当前用户权限,然后注销重新登陆,就可以正常使用了。
sudo chmod +x /usr/bin/dumpcap
sudo usermod -a -G wireshark pi
用wireshark抓包看一下,可以看到抓到对应的包:
基础任务二:主控板建立TCPIP或UDP服务器,局域网PC使用TCPIP或UDP客户端进行连接并发送数据,主控板接收到数据后,送液晶屏显示(没有则通过串口打印显示);通过抓包软件抓取交互报文,展示并分析。(TCP和UDP二选一,或者全都操作)
这个任务我们TCP与UDP都尝试一下。
TCP实验我们在W5500上搭建一个HTTP服务来进行,而UDP实验我们通过MQTT来实现。使用一个树莓派zero w作为MQTT broker,用另一个pico WH作为MQTT客户端,与我们的W5500 pico实现UDP通讯。
首先看HTTP部分,我们直接用socket方法发送HTTP头和内容就可以实现通信:
import socket
addr = socket.getaddrinfo('0.0.0.0', 80)[0][-1]
s = socket.socket()
s.bind(addr)
s.listen(1)
print('listening on', addr)
def callback(content):
start_index = content.find("comment=") + len("comment=")
end_index = content.find("HTTP/1.1")
if start_index != -1 and end_index != -1:
comment = content[start_index:end_index].strip()
print()
print(comment)
html = """
<!DOCTYPE html>
<html>
<head> <title>Micropython</title> </head>
<body>%s</body>
</html>
"""
for i in range(3):
cl, addr = s.accept()
# print('client connected from', addr)
content = b""
while True:
msg=cl.recv(1024)
content += msg
if content[-4:] == b"\r\n\r\n":
break
callback(content.decode())
response = html % """
<form action="/action_page.php" id="usrform">
<textarea rows="4" cols="50" name="comment"></textarea><br>
<input type="submit">
</form>
"""
cl.send('HTTP/1.0 200 OK\r\nContent-type: text/html\r\n\r\n')
cl.send(response)
cl.close()
s.close()
这里我们选择接受三次连接后结束。打开网页,输入W5500的ip,就可以看到对应页面。输入内容后,shell就可以打印出对应内容。
由于这是主机和W5500之间的通信,因此wireshark中可以抓到相应的包:
当三次访问完成后,就进入MQTT部分。
先在树莓派zero w上搭建我们的MQTT服务器。用ssh登陆上后,使用 sudo apt install mosquitto来安装服务。安装完成后,我们首先需要关闭原本的服务,因为这里我们想手动启动,并且观察broker的输出。
sudo systemctl stop mosquitto
sudo systemctl disable mosquitto
sudo systemctl status mosquitto
看到如下内容,说明已经成功关闭服务。
接着我们用nano mosquitto.conf新建一个配置文件,写入以下内容
allow_anonymous true
listener 1883
保存后,就可以使用mosquitto -c mosquitto.conf来启动broker,启动后显示以下内容:
下面开始写W5500的代码。我使用的是官方的umqttsimple库,这个库有个小问题,默认的keepalive参数会导致无法运行,我们需要手动指定为10。这里我们连接上broker,并订阅主题w5500,收到内容后讲内容打印出来。完整代码如下:
print("test MQTT")
from umqttsimple import MQTTClient
def callback(topic, msg):
print((topic, msg))
if topic == b'w5500':
print(msg)
client = MQTTClient(
client_id = "w5500",
server = '192.168.8.174',
keepalive = 10,
)
client.set_callback(callback)
client.connect()
client.subscribe("w5500")
print('Connected')
while True:
try:
client.check_msg()
except:
client.disconnect()
client.connect()
client.subscribe("w5500")
下面我们写pico WH上的代码。pico WH我们使用circuitpython来完成,尽量多做一些不同的实验,避免重复。circuitpython刷机过程和micropython一致,就不重复说明了。刷好后,在settings.toml中写入以下内容,修改ssid和密码,然后断电重启后,就可以自动联网。切记一定要断电重启,soft reboot不行。
CIRCUITPY_WIFI_SSID="ssid"
CIRCUITPY_WIFI_PASSWORD="password"
主程序实现每隔两秒钟向w5500主题发送一次消息。
import time
import wifi
import socketpool
import adafruit_minimqtt.adafruit_minimqtt as MQTT
pool = socketpool.SocketPool(wifi.radio)
mqtt_client = MQTT.MQTT(
broker='192.168.8.174',
port=1883,
socket_pool=pool,
client_id='pico_wh'
)
mqtt_client.connect()
print('Start')
while True:
mqtt_client.publish('w5500', "Hello EEWorld!!!")
time.sleep(2)
把上面代码写入code.py,就可以实现上电自动运行。如果一切顺例,我们就可以在w5500的shell中看到打印的内容:
进阶任务:从NTP服务器(注意数据交互格式的解析)同步时间,获取时间送显示屏(串口)显示
这里我们直接使用micropython自带的ntptime就可以实现。但这个包也有点小问题,网上很多教程都是使用包自带的变量NTP_DELTA来设置UTC+8时区,但实测并不能用。因此我们对时后手动增加对应的秒数,再转回struct time就可。
# 进阶任务:从NTP服务器(注意数据交互格式的解析)同步时间,获取时间送显示屏(串口)显示
import ntptime
import time
for i in range(10):
try:
ntptime.settime()
t = time.localtime(time.time() + 8*3600)
print("ntp time(BeiJing): %s-%s-%s %s:%s:%s" % (t[0],t[1],t[2],t[3],t[4],t[5]))
break
except:
print("Can not get time!")
time.sleep_ms(1000)
如果板子之前没有硬重启过,还保持着网络连接,运行后就可以看到时间已经校对:
终极任务二:使用外部存储器,组建简易FTP文件服务器,并能正常上传下载文件。
这里和之前的ping一样,使用socket方法来发送接收ftp命令,来实现ftp。这里的ftp代码是基于网上的开源例程修改的,原例程为esp8266编写,少了几个ftp命令,这里已经补上。
先在主机上使用sudo apt install ftp来安装ftp功能。完成后我们把下面代码保存为ftpserver.py文件。
#
# Small ftp server for ESP8266 ans ESP32 Micropython
#
# Based on the work of chrisgp - Christopher Popp and pfalcon - Paul Sokolovsky
#
# The server accepts passive mode only.
# It runs in foreground and quits, when it receives a quit command
# Start the server with:
#
# import ftp
#
# Copyright (c) 2016 Christopher Popp (initial ftp server framework)
# Copyright (c) 2016 Robert Hammelrath (putting the pieces together
# and a few extensions)
# Distributed under MIT License
#
import socket
import network
import uos
import gc
def send_list_data(path, dataclient, full):
try: # whether path is a directory name
for fname in sorted(uos.listdir(path), key=str.lower):
dataclient.sendall(make_description(path, fname, full))
except: # path may be a file name or pattern
pattern = path.split("/")[-1]
path = path[:-(len(pattern) + 1)]
if path == "":
path = "/"
for fname in sorted(uos.listdir(path), key=str.lower):
if fncmp(fname, pattern):
dataclient.sendall(make_description(path, fname, full))
def make_description(path, fname, full):
if full:
stat = uos.stat(get_absolute_path(path, fname))
file_permissions = ("drwxr-xr-x"
if (stat[0] & 0o170000 == 0o040000)
else "-rw-r--r--")
file_size = stat[6]
description = "{} 1 owner group {:>10} Jan 1 2000 {}\r\n".format(
file_permissions, file_size, fname)
else:
description = fname + "\r\n"
return description
def send_file_data(path, dataclient):
with open(path, "rb") as file:
chunk = file.read(512)
while len(chunk) > 0:
dataclient.sendall(chunk)
chunk = file.read(512)
def save_file_data(path, dataclient):
with open(path, "wb") as file:
chunk = dataclient.recv(512)
while len(chunk) > 0:
file.write(chunk)
chunk = dataclient.recv(512)
def get_absolute_path(cwd, payload):
# Just a few special cases "..", "." and ""
# If payload start's with /, set cwd to /
# and consider the remainder a relative path
if payload.startswith('/'):
cwd = "/"
for token in payload.split("/"):
if token == '..':
if cwd != '/':
cwd = '/'.join(cwd.split('/')[:-1])
if cwd == '':
cwd = '/'
elif token != '.' and token != '':
if cwd == '/':
cwd += token
else:
cwd = cwd + '/' + token
return cwd
# compare fname against pattern. Pattern may contain
# wildcards ? and *.
def fncmp(fname, pattern):
pi = 0
si = 0
while pi < len(pattern) and si < len(fname):
if (fname[si] == pattern[pi]) or (pattern[pi] == '?'):
si += 1
pi += 1
else:
if pattern[pi] == '*': # recurse
if (pi + 1) == len(pattern):
return True
while si < len(fname):
if fncmp(fname[si:], pattern[pi+1:]):
return True
else:
si += 1
return False
else:
return False
if pi == len(pattern.rstrip("*")) and si == len(fname):
return True
else:
return False
def ftpserver(net, port=21, timeout=None):
DATA_PORT = 13333
ftpsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ftpsocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
ftpsocket.bind(socket.getaddrinfo("0.0.0.0", port)[0][4])
datasocket.bind(socket.getaddrinfo("0.0.0.0", DATA_PORT)[0][4])
ftpsocket.listen(1)
ftpsocket.settimeout(timeout)
datasocket.listen(1)
datasocket.settimeout(None)
msg_250_OK = '250 OK\r\n'
msg_550_fail = '550 Failed\r\n'
addr = net.ifconfig()[0]
print("FTP Server started on ", addr)
try:
dataclient = None
fromname = None
do_run = True
while do_run:
cl, remote_addr = ftpsocket.accept()
cl.settimeout(300)
cwd = '/'
try:
# print("FTP connection from:", remote_addr)
cl.sendall("220 Hello, this is Micropython.\r\n")
while True:
gc.collect()
data = cl.readline().decode("utf-8").rstrip("\r\n")
if len(data) <= 0:
print("Client disappeared")
do_run = False
break
command = data.split(" ")[0].upper()
payload = data[len(command):].lstrip()
path = get_absolute_path(cwd, payload)
print("Command={}, Payload={}".format(command, payload))
if command == "USER":
cl.sendall("230 Logged in.\r\n")
elif command == "SYST":
cl.sendall("215 UNIX Type: L8\r\n")
elif command == "NOOP":
cl.sendall("200 OK\r\n")
elif command == "FEAT":
cl.sendall("211 no-features\r\n")
elif command == "PWD" or command == "XPWD":
cl.sendall('257 "{}"\r\n'.format(cwd))
elif command == "CWD" or command == "XCWD":
try:
files = uos.listdir(path)
cwd = path
cl.sendall(msg_250_OK)
except:
cl.sendall(msg_550_fail)
elif command == "CDUP":
cwd = get_absolute_path(cwd, "..")
cl.sendall(msg_250_OK)
elif command == "TYPE":
# probably should switch between binary and not
cl.sendall('200 Transfer mode set\r\n')
elif command == "SIZE":
try:
size = uos.stat(path)[6]
cl.sendall('213 {}\r\n'.format(size))
except:
cl.sendall(msg_550_fail)
elif command == "QUIT":
cl.sendall('221 Bye.\r\n')
do_run = False
break
elif command == "PASV":
cl.sendall('227 Entering Passive Mode ({},{},{}).\r\n'.
format(addr.replace('.', ','), DATA_PORT >> 8,
DATA_PORT % 256))
dataclient, data_addr = datasocket.accept()
print("FTP Data connection from:", data_addr)
DATA_PORT = 13333
active = False
elif command == "PORT":
items = payload.split(",")
if len(items) >= 6:
data_addr = '.'.join(items[:4])
# replace by command session addr
if data_addr == "127.0.1.1":
data_addr = remote_addr
DATA_PORT = int(items[4]) * 256 + int(items[5])
dataclient = socket.socket(socket.AF_INET,
socket.SOCK_STREAM)
dataclient.settimeout(10)
dataclient.connect((data_addr, DATA_PORT))
print("FTP Data connection with:", data_addr)
cl.sendall('200 OK\r\n')
active = True
else:
cl.sendall('504 Fail\r\n')
elif command == "LIST" or command == "NLST":
if not payload.startswith("-"):
place = path
else:
place = cwd
try:
cl.sendall("150 Here comes the directory listing.\r\n")
send_list_data(place, dataclient,
command == "LIST" or payload == "-l")
cl.sendall("226 Listed.\r\n")
except:
cl.sendall(msg_550_fail)
if dataclient is not None:
dataclient.close()
dataclient = None
elif command == "RETR":
try:
cl.sendall("150 Opening data connection.\r\n")
send_file_data(path, dataclient)
cl.sendall("226 Transfer complete.\r\n")
except:
cl.sendall(msg_550_fail)
if dataclient is not None:
dataclient.close()
dataclient = None
elif command == "STOR":
try:
cl.sendall("150 Ok to send data.\r\n")
save_file_data(path, dataclient)
cl.sendall("226 Transfer complete.\r\n")
except:
cl.sendall(msg_550_fail)
if dataclient is not None:
dataclient.close()
dataclient = None
elif command == "DELE":
try:
uos.remove(path)
cl.sendall(msg_250_OK)
except:
cl.sendall(msg_550_fail)
elif command == "RMD" or command == "XRMD":
try:
uos.rmdir(path)
cl.sendall(msg_250_OK)
except:
cl.sendall(msg_550_fail)
elif command == "MKD" or command == "XMKD":
try:
uos.mkdir(path)
cl.sendall(msg_250_OK)
except:
cl.sendall(msg_550_fail)
elif command == "RNFR":
fromname = path
cl.sendall("350 Rename from\r\n")
elif command == "RNTO":
if fromname is not None:
try:
uos.rename(fromname, path)
cl.sendall(msg_250_OK)
except:
cl.sendall(msg_550_fail)
else:
cl.sendall(msg_550_fail)
fromname = None
elif command == "MDTM":
try:
tm=localtime(uos.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')
elif command == "STAT":
if payload == "":
cl.sendall("211-Connected to ({})\r\n"
" Data address ({})\r\n"
"211 TYPE: Binary STRU: File MODE:"
" Stream\r\n".format(
remote_addr[0], addr))
else:
cl.sendall("213-Directory listing:\r\n")
send_list_data(path, cl, True)
cl.sendall("213 Done.\r\n")
else:
cl.sendall("502 Unsupported command.\r\n")
print("Unsupported command {} with payload {}".format(
command, payload))
except Exception as err:
print(err)
finally:
cl.close()
cl = None
except Exception as e:
print(e)
finally:
datasocket.close()
ftpsocket.close()
if dataclient is not None:
dataclient.close()
# ftpserver()
接着在主程序中我们重新连接一下网络,这里的目的不是为了联网,因为如果没有重置的话,网本来就连着。目的是为了获取到一个网络object,传入ftp方法用来获取ip。
# 终极任务二:使用外部存储器,组建简易FTP文件服务器,并能正常上传下载文件。
from ftpserver import ftpserver
from w5500 import w5500
from ping import ping
ip = "192.168.8.213"
gate = ip.split(".")
gate[-1] = "1"
gate = ".".join(gate)
nic = w5500((ip,"255.255.255.0",gate,"8.8.8.8"))
ftpserver(nic)
运行后,我们在terminal中输入ftp 192.168.8.213来连接到w5500,使用ls方法来显示文件列表,通过get和put方法来下载和上传文件,验证一下,一切正常。
源码:https://download.eeworld.com.cn/detail/eew_AG2DvH/631127
体验心得
以往的开发都是在轮子齐全的情况下进行。这次活动为了全在micopython上完成,尝试做了一些简单的移植,我觉得还是非常有意思的。这是followme的最后一期,希望这样有意思的活动还可以一直持续下去。
|