【摘要】 我们购买智能家居产品后，买回来拆箱后第一件事通常就是给这个新的硬件进行配网，所谓配网，也就是让这个新的物联网设备联入我们的局域网内，让这个物联网设备可以进行网络通讯。我们在上一篇文章《MicroPython(ESP32/ESP8266) 实现web控制GPIO》中已经了解到了如何使用ESP32和ESP8266通过联网来实现在Web中控制板载的 LED 灯开关。本文将介绍…
我们购买智能家居产品后，买回来拆箱后第一件事通常就是给这个新的硬件进行配网，所谓配网，也就是让这个新的物联网设备联入我们的局域网内，让这个物联网设备可以进行网络通讯。我们在上一篇文章《MicroPython(ESP32/ESP8266) 实现web控制GPIO》 中已经了解到了如何使用ESP32和ESP8266通过联网来实现在Web中控制板载的 LED 灯开关。本文将介绍基于 MicroPython 来实现的 ESP32/ESP8266 Wifi配网。
在开始代码之前，需要先准备以下：
回想以下我们的智能家居物联网设备，以小米生态圈的设备为例，新设备开箱通电后，一般是打开米家APP，然后搜索到新买的设备，然后需要手动将wifi连接到这个设配上，然后在 APP 中填入 SSID 和 wifi密码信息，等待传输，传输完成后，就算完成配网，在 APP 的界面中就可以看到新的设备了。
配网的流程总结如上图所示。然而我们的使用当中，配网通常只发生在新设备加入或者网络环境改变的时候才需要，正常情况下设备重启，是不需要每一次都要来一次配网操作的。所以一般情况下，在一次配网之后，我们会将我们的Wifi信息保存下来，设备重启后如果有存在的配网信息，会自动直接联网。
而针对我们整个开发版的程序，我们可以在 main.py 执行在开始，就先执行网络检查，然后根据是否成功联网来判断是否需要配网操作，流程如下：
上文梳理了整个配网过程的流程。在这个流程中，最开始的步骤就是判断网络是否连接。以下将介绍如何使用 MicroPython 操作开发板的 Wifi。
我们开发板(ESP32/ESP8266)的wifi有AP和STA模式，AP就是开发版上创建一个热点，其他设备连接到AP上，而STA模式和我们普通的手机电脑使用Wifi联网类似。这里的要点就是我们需要检查STA模式下开发版是否能正常联网，如果不能，我们利用开发板的AP模式，让我们的其他设备连接开发板，把我们局域网Wifi的配置信息告知开发板，从而使开发板能正常联网。
import network
wlan_sta = network.WLAN(network.STA_IF)
wlan_sta.isconnected()
通过调用 isconnected() 函数，可以获取到开发板是否正常联网，如果正常联网，返回结果会是 True 否则为 False 。
wlan_sta.scan()
scan() 函数扫描设备附近可以搜索到的 Wifi，会返回一个列表，列表中每一条为可连接wifi的信息。
[(b'WifiSSID', b'LPw\xb7\xs8\x94', 1, -48, 3, 0),...]
以上是省略了部分信息的返回值，可以看到，每一条记录中有6个信息，它们分别代表了 SSID名称 BSSID(MAC地址) 频道 RSSI信号强度 加密模式 是否隐藏 。其中加密模式，包含了 WEP、WPA-PSK、WPA2-PSK、WPA/WPA2-PSK等。
接下来，我们就可以尝试连接Wifi。
wlan_sta.connect('ssid', 'password')
wlan_sta.isconnected()
如果连接成功，则返回 True 。如果需要断开连接，可以使用 disconnect() 函数。
wlan_sta.disconnect()
完成了 Wifi 连接和检查网络是否正常后，我们开始解决利用 AP 配网的问题。
先看代码：
import network
import socket
wlan_ap = network.WLAN(network.AP_IF)
wlan_ap.active(True)
wlan_ap.config(essid='MyESP8266',authmode=0)
server_socket = socket.socket()
server_socket.bind(('0.0.0.0', 80))
server_socket.listen(3)
def web_page(): return b"""<html> <head> <title>MYESP8266 AP Test</title> </head> <body> <h1>This is MyESP8266 AP Test Page.</h1> </body> </html>"""
while True: conn, addr = server_socket.accept() print('Connection: %s ' % str(addr)) response = web_page() conn.send('HTTP/1.1 200 OK\n') conn.send('Content-Type: text/html\n') conn.send('Connection: close\n\n') conn.sendall(response) conn.close()
从上面的代码我们可以看到，当我们创建好AP后，就打开一个 socket ，并且绑定80端口开始监听，然后开启一个循环，当接受到连接后就给客户端发送页面代码。如果对 socket 不了解的，可以参考《快速了解Python socket编程》 。
这时用手机或者电脑的 wifi 连接 SSID 名为 MYESP8266 的 Wifi 热点，因为我们authmode选择了 open 所以不需要密码。连接成功后，用浏览器打开地址 192.168.4.1 ，就可以看到我们上面的页面。
我们一般情况下，如果要进行 Web 开发，通常会使用 Flask 或者 Django 之类的框架。而针对开发版这种运算能力有限的硬件，也有对应的框架可以用。但我们这里为了能深入的了解，就通过自己完成最基本的功能来了解整个程序的运行方式。
封装HTML响应
根据上面的示例代码，我们可以了解到，如果要在客户端正常显示页面，我们需要先给客户端发一个HTTP的Header信息，然后再发送具体的页面内容。所以，为了方便日后的调用，我们对上面的代码进行修改：
import network
import socket
wlan_ap = network.WLAN(network.AP_IF)
wlan_ap.active(True)
wlan_ap.config(essid='MyESP8266',authmode=0)
server_socket = socket.socket()
server_socket.bind(('0.0.0.0', 80))
server_socket.listen(3)
def send_header(conn, status_code=200, content_length=None ): conn.sendall("HTTP/1.0 {} OK\r\n".format(status_code)) conn.sendall("Content-Type: text/html\r\n") if content_length is not None: conn.sendall("Content-Length: {}\r\n".format(content_length)) conn.sendall("\r\n")
def send_response(conn, payload, status_code=200): content_length = len(payload) send_header(conn, status_code, content_length) if content_length > 0: conn.sendall(payload) conn.close()
def config_page(): return b"""<html> <head> <title>MYESP8266 AP Test</title> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> </head> <body> <h1>Wifi 配网</h1> <form action="configure" method="post"> <div> <label>SSID</label> <input type="text" name="ssid"> </div> <div> <label>PASSWORD</label> <input type="password" name="password"> </div> <input type="submit" value="连接"> <form> </body> </html>"""
while True: conn, addr = server_socket.accept() print('Connection: %s ' % str(addr)) try: conn.settimeout(3) request = b"" try: while "\r\n\r\n" not in request: request += conn.recv(512) except OSError: pass print(request) response = config_page() send_response(conn, response)
finally: conn.close()
我们添加了三个函数，分别为 send_header() send_response() config_page() 。其中 send_header() 把我们需要发送的 Header 信息打包，config_page() 则是创建我们的 HTML 页面，最后由 send_response() 将其整合，发送给客户端。
运行代码，如果正常，用手机连接开发板的AP，打开 192.168.4.1 ，就可以看到上图的页面。
路由
上面的代码中，页面中有一个 form ，里面可以输入 SSID 和 Wifi 密码，当我们输入完成后，点击连接，将会将我们输入的内容 POST 到 /configure 路径中。处理这个问题，在 Web 框架中，会有现成的路由模块，但这里我们需要自己用代码进行处理。
我们的代码中，当客户端连接后，我们的开发板会接受来自客户端传来的信息——request ，打印这个变量看看客户端传来的内存：
# 连接 192.168.4.1
Connection: ('192.168.4.2', 44794)
b'GET / HTTP/1.1\r\nUser-Agent: Dalvik/2.1.0 (Linux; U; Android 9; MIX 2 MIUI/20.6.18)\r\nHost: 192.168.4.1\r\nConnection: Keep-Alive\r\nAccept-Encoding: gzip\r\n\r\n'
# 连接 192.168.4.1/test
b'GET /test HTTP/1.1\r\nHost: 192.168.4.1\r\nConnection: keep-alive\r\nCache-Control: max-age=0\r\nUpgrade-Insecure-Requests: 1\r\nUser-Agent: Mozilla/5.0 (Linux; Android 9; MIX 2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.82 Mobile Safari/537.36\r\nAccept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9\r\nAccept-Encoding: gzip, deflate\r\nAccept-Language: zh-CN,zh;q=0.9,en;q=0.8\r\n\r\n'
# 输入信息，点击连接按钮
b'POST /configure HTTP/1.1\r\nHost: 192.168.4.1\r\nConnection: keep-alive\r\nContent-Length: 26\r\nCache-Control: max-age=0\r\nUpgrade-Insecure-Requests: 1\r\nOrigin: http://192.168.4.1\r\nContent-Type: application/x-www-form-urlencoded\r\nUser-Agent: Mozilla/5.0 (Linux; Android 9; MIX 2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.82 Mobile Safari/537.36\r\nAccept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9\r\nReferer: http://192.168.4.1/\r\nAccept-Encoding: gzip, deflate\r\nAccept-Language: zh-CN,zh;q=0.9,en;q=0.8\r\n\r\nssid=xdbdh&password=ddjxdj'
可以看到，当我们连接不同的地址，开发板接受到的信息是不同的，我们就可以通过正则表达式来抓去不同的内容即可实现类似 Web 框架路由的功能。
try: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).decode("utf-8").rstrip("/") except Exception: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).rstrip("/") print("URL is {}".format(url))
我们将上面 print() 函数替换乘上面的代码，再尝试上面三个地址：
# 连接 192.168.4.1
URL is
# 连接 192.168.4.1/test
URL is test
# 输入信息，点击连接按钮
URL is configure
这样，我们的精简版路由功能就完成了。
POST 传参获取
解决了页面显示和路由，剩下就是如何获取 POST 的传参了。我们再看一次当我们使用 POST 时，返回过来的信息：
b'POST /configure HTTP/1.1\r\n
Host: 192.168.4.1\r\n
Connection: keep-alive\r\n
Content-Length: 26\r\n
Cache-Control: max-age=0\r\n
Upgrade-Insecure-Requests: 1\r\n
Origin: http://192.168.4.1\r\n
Content-Type: application/x-www-form-urlencoded\r\nUser-Agent: Mozilla/5.0 (Linux; Android 9; MIX 2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.82 Mobile Safari/537.36\r\nAccept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9\r\nReferer: http://192.168.4.1/\r\n
Accept-Encoding: gzip, deflate\r\nAccept-Language: zh-CN,zh;q=0.9,en;q=0.8\r\n\r\n
ssid=xdbdh&password=ddjxdj'
可以看到，信息开头是 POST 方法，然后结 ssid=....&password=... 就是我们传过来的参数，和处理路由的方法类似，我们使用正则表达式过滤一下，即可获取到我们需要的 ssid 和 Wifi 密码了。
# POST 参数解析
def get_wifi_conf(request): match = ure.search("ssid=([^&]*)&password=(.*)", request) if match is None: return False try: ssid = match.group(1).decode("utf-8").replace("%3F", "?").replace("%21", "!") password = match.group(2).decode("utf-8").replace("%3F", "?").replace("%21", "!") except Exception: ssid = match.group(1).replace("%3F", "?").replace("%21", "!") password = match.group(2).replace("%3F", "?").replace("%21", "!") if len(ssid) == 0: return False return (ssid, password)
我们再修改一下代码，添加一个新页面，用来显示 ssid 和 Wifi 密码，来确认我们的路由功能和 POST 参数正常获取。
def wifi_conf_page(ssid, passwd): return b"""<html> <head> <title>Wifi Conf Info</title> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> </head> <body> <h1>Post data:</h1> <p>SSID: %s</p> <p>PASSWD: %s</p> <a href="https://bbs.huaweicloud.com/">Return Configure Page</a> </body> </html>""" % (ssid, passwd)
修改后的代码：
# 前面相同的部分省略
while True: conn, addr = server_socket.accept() print('Connection: %s ' % str(addr)) try: conn.settimeout(3) request = b"" try: while "\r\n\r\n" not in request: request += conn.recv(512) except OSError: pass # url process try: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).decode("utf-8").rstrip("/") except Exception: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).rstrip("/") print("URL is {}".format(url)) if url == "": response = config_page() send_response(conn, response) elif url == "configure": ret = get_wifi_conf(request) response = wifi_conf_page(ret[0], ret[1]) send_response(conn, response) finally: conn.close()
执行代码，输入 ssid 和密码后，点击连接，应该能跳转到新页面并且显示刚才输入的 ssid 和密码。点击返回，能重新跳回信息输入的页面。
在《MicroPython(ESP32/ESP8266) 实现web控制GPIO》 中，我们已经介绍了如何通过 connect() 方法来连接我们已知的 Wifi。接下来，我们要做的也很简单，就是创建一个 do_connect() 方法来处理我们上面传过来的 ssid 和密码。
def do_connect(ssid, password): wlan_sta.active(True) if wlan_sta.isconnected(): return None print('Connect to %s' % ssid) wlan_sta.connect(ssid, password) for retry in range(100): connected = wlan_sta.isconnected() if connected: break time.sleep(0.1) print('.', end='') if connected: print('\nConnected : ', wlan_sta.ifconfig()) else: print('\nFailed. Not Connected to: ' + ssid) return connected
可以看到，这个函数会接受传来的 wifi 配置参数，进行连接，如果成功，会返回 True。然后我们还还需要一个执行连接的方法，这个方法用于连接成功，就自动获取连如局域网后的ip地址。
def handle_wifi_configure(ssid, password): if do_connect(ssid, password): new_ip = wlan_sta.ifconfig()[0] return new_ip else: print('connect fail') return False
这些都完成后，我们只需要把开发板 AP 联网配置部分封装好，成为一个 start_ap() 方法，即可：
# response 的方法都为创建 HTML 代码方法，这里省略
# 可以在文末完整代码中查看
def startAP(): global server_socket stop() wlan_ap.active(True) wlan_ap.config(essid='MyEsp8266',authmode=0) server_socket = socket.socket() server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind(('0.0.0.0', 80)) server_socket.listen(3) while not wlan_sta.isconnected(): conn, addr = server_socket.accept() print('Connection: %s ' % str(addr)) try: conn.settimeout(3) request = b"" try: while "\r\n\r\n" not in request: request += conn.recv(512) except OSError: pass # url process try: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).decode("utf-8").rstrip("/") except Exception: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).rstrip("/") print("URL is {}".format(url)) if url == "": response = config_page() send_response(conn, response) elif url == "configure": ret = get_wifi_conf(request) ret = handle_wifi_configure(ret[0], ret[1]) if ret is not None: response = connect_sucess(ret) send_response(conn, response) print('connect sucess') elif url == "disconnect": wlan_sta.disconnect() finally: conn.close() wlan_ap.active(False) print('ap exit')
这里我们实现的功能为让开发板创建AP，生成一个 Wifi 信息的配置页面，然后通过路由来处理输入和参数，最后执行 Wifi 联网，如果连接成功，即退出循环，关闭 AP 热点。
我们从用手机输入完成点击连接后，如果连接成功，将会自动返回成功连接的页面：
到这里，我们的 wifi 配网就已经基本完成了。
本文开始先从配网的需求、流程进行分析，然后一步步分别介绍 MicroPython Wifi的操作，AP的使用以及简单的 Web 实现，然后将上述的要点结合我们的配网需求，完成完整的设配配网代码开发。
但是，文章为了比较清晰的展示内容，因此在代码上可能会显得比较冗长和繁复，有很大的优化空间。大家可以根据自己的实际情况，对代码进行进一步的优化和调整，以下给出几个可以调整方法：
优化代码结构，模块化部分功能 将 web 部分整合成一个模块，比如带有 html 模板渲染功能的模块、路由模块灯 尝试在用户体验上优化配网的流程 其他创新的需求等…… 此外，还存还存在一个问题，就是可能因为 ESP8266 的内存和算力问题，代码运行的时候有时会出错和跳出，需要重启或者断电，但同样的代码在 ESP32 开发板上，却没有问题。可能是 MircoPython 的问题，也有可能是因为代码设计问题，这方面需要进一步研究和尝试。
物联网开发涉及到很多硬件和软件的问题，但是在实践中，经常会遇到各种奇怪的问题，这很可能打击了学习的热情，加上网上的教程和示例不多，初学者更容易遇到问题解决不了而不得不放弃。本文尽可能的详细解释代码和原理，但由于水平经验有限，难免会有所疏漏， 望读者见谅，并且欢迎大家一起来交流进步。
希望本文对你有用。如果你觉得文章对你用，记得关注收藏。你的关注和收藏是继续更新的动力哦。
import network
import socket
import ure
import time
NETWORK_PROFILES = 'wifi.dat'
wlan_ap = network.WLAN(network.AP_IF)
wlan_sta = network.WLAN(network.STA_IF)
server_socket = None
def send_header(conn, status_code=200, content_length=None ): conn.sendall("HTTP/1.0 {} OK\r\n".format(status_code)) conn.sendall("Content-Type: text/html\r\n") if content_length is not None: conn.sendall("Content-Length: {}\r\n".format(content_length)) conn.sendall("\r\n")
def send_response(conn, payload, status_code=200): content_length = len(payload) send_header(conn, status_code, content_length) if content_length > 0: conn.sendall(payload) conn.close()
def config_page(): return b"""<html> <head> <title>MYESP8266 AP Test</title> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> </head> <body> <h1>Wifi 配网</h1> <form action="configure" method="post"> <div> <label>SSID</label> <input type="text" name="ssid"> </div> <div> <label>PASSWORD</label> <input type="password" name="password"> </div> <input type="submit" value="连接"> <form> </body> </html>"""
def wifi_conf_page(ssid, passwd): return b"""<html> <head> <title>Wifi Conf Info</title> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> </head> <body> <h1>Post data:</h1> <p>SSID: %s</p> <p>PASSWD: %s</p> <a href="https://bbs.huaweicloud.com/">Return Configure Page</a> </body> </html>""" % (ssid, passwd)
def connect_sucess(new_ip): return b"""<html> <head> <title>Connect Sucess!</title> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> </head> <body> <p>Wifi Connect Sucess</p> <p>IP Address: %s</p> <a href="http://%s">Home</a> <a href="https://bbs.huaweicloud.com/disconnect">Disconnect</a> </body> </html>""" % (new_ip, new_ip)
def get_wifi_conf(request): match = ure.search("ssid=([^&]*)&password=(.*)", request) if match is None: return False try: ssid = match.group(1).decode("utf-8").replace("%3F", "?").replace("%21", "!") password = match.group(2).decode("utf-8").replace("%3F", "?").replace("%21", "!") except Exception: ssid = match.group(1).replace("%3F", "?").replace("%21", "!") password = match.group(2).replace("%3F", "?").replace("%21", "!") if len(ssid) == 0: return False return (ssid, password)
def handle_wifi_configure(ssid, password): if do_connect(ssid, password):
# try:
# profiles = read_profiles()
# except OSError:
# profiles = {}
# profiles[ssid] = password
# write_profiles(profiles)
#
# time.sleep(5)
# new_ip = wlan_sta.ifconfig()[0] return new_ip else: print('connect fail') return False
def check_wlan_connected(): if wlan_sta.isconnected(): return True else: return False def do_connect(ssid, password): wlan_sta.active(True) if wlan_sta.isconnected(): return None print('Connect to %s' % ssid) wlan_sta.connect(ssid, password) for retry in range(100): connected = wlan_sta.isconnected() if connected: break time.sleep(0.1) print('.', end='') if connected: print('\nConnected : ', wlan_sta.ifconfig()) else: print('\nFailed. Not Connected to: ' + ssid) return connected
def read_profiles(): with open(NETWORK_PROFILES) as f: lines = f.readlines() profiles = {} for line in lines: ssid, password = line.strip("\n").split(";") profiles[ssid] = password return profiles
def write_profiles(profiles): lines = [] for ssid, password in profiles.items(): lines.append("%s;%s\n" % (ssid, password)) with open(NETWORK_PROFILES, "w") as f: f.write(''.join(lines)) def stop(): global server_socket if server_socket: server_socket.close() server_socket = None
def startAP(): global server_socket stop() wlan_ap.active(True) wlan_ap.config(essid='MyEsp8266',authmode=0) server_socket = socket.socket() server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind(('0.0.0.0', 80)) server_socket.listen(3) while not wlan_sta.isconnected(): conn, addr = server_socket.accept() print('Connection: %s ' % str(addr)) try: conn.settimeout(3) request = b"" try: while "\r\n\r\n" not in request: request += conn.recv(512) except OSError: pass # url process try: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).decode("utf-8").rstrip("/") except Exception: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).rstrip("/") print("URL is {}".format(url)) if url == "": response = config_page() send_response(conn, response) elif url == "configure": ret = get_wifi_conf(request) ret = handle_wifi_configure(ret[0], ret[1]) if ret is not None: response = connect_sucess(ret) send_response(conn, response) print('connect sucess') elif url == "disconnect": wlan_sta.disconnect() finally: conn.close() wlan_ap.active(False) print('ap exit')
def home(): global server_socket stop() wlan_sta.active(True) ip_addr = wlan_sta.ifconfig()[0] print('wifi connected') server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind(('0.0.0.0', 80)) server_socket.listen(3) while check_wlan_connected(): conn, addr = server_socket.accept() try: conn.settimeout(3) request = b"" try: while "\r\n\r\n" not in request: request += conn.recv(512) except OSError: pass # url process try: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).decode("utf-8").rstrip("/") except Exception: url = ure.search("(?:GET|POST) /(.*?)(?:\\?.*?)? HTTP", request).group(1).rstrip("/") if url == "": response = connect_sucess(ip_addr) send_response(conn, response) elif url == "disconnect": wlan_sta.disconnect() finally: conn.close() wlan_sta.active(False) print('sta exit')
def main(): while True: if not check_wlan_connected(): startAP() else: home() main()
原文链接：
https://zhuanlan.zhihu.com/p/369017239
http://www.proyy.com/885253b7289c422da561fdfdc78f3bd2.html
