一文速通ESP32（基于MicroPython）——含示例代码

作者

ESP32

简介

ESP32-S3 是一款集成 2.4 GHz Wi-Fi 和 Bluetooth 5 (LE) 的 MCU 芯片，支持远距离模式 (Long Range)。ESP32-S3 搭载 Xtensa® 32 位 LX7 双核处理器，主频高达 240 MHz，内置 512 KB SRAM (TCM)，具有 45 个可编程 GPIO 管脚和丰富的通信接口。ESP32-S3 支持更大容量的高速 Octal SPI flash 和片外 RAM，支持用户配置数据缓存与指令缓存。                                  ——乐鑫官网


从乐鑫的官网可以看出ESP32也分为很多种类，我这边用的是ESP32-S3型号的，型号不一样的小伙伴只需要在配置环境的时候选择对应的文件即可，其余代码部分都是一样的。


引脚分布图

可以看到上面这个引脚分布图跟我们之前看的都不一样，这个引脚分布图非常简洁，没有说哪个引脚一定是做什么用的（除了VCC和GND，RST等之类的），原因就在官方提供的datasheet中。

命名规则 

剩下一个介绍的就是命名规则了。

我手上这款是ESP32-S3-N8R2，所以含义是正常Flash温度，8MB的Flash，具有2MB的PSRAM。但是有一点我没有明白，那就是我买的某宝店铺的定价。

按理说16MB的Flash，8MB的PSRAM（N16R8）应该是要比我手上的8MB的Flash，2MB的PSRAM要贵的，但是结果却是相反的，我当初买的时候不知道，光挑贵的买了，结果好像是配置反而更低了。不懂是我有什么地方不明白的还是店家定价定错了。懂的小伙伴可以在评论区教我一下。


那么就简简单单介绍这么多，我们的主要目的还是快速上手。


环境配置

https://thonny.org/

我们使用Thonny作为IDE，进入上面的链接直接下载即可。


一般情况下，按照我下面图片的指示，直接在Thonny中把固件烧录进去即可。


固件文件在下面的链接中去寻找。

https://micropython.org/download/

但我这个型号(ESP32-S3-N8R2)一直是烧不进去的，但应该是烧进去的，但是是有问题的。

整了一下午还是没弄好，固件反反复复烧了n次，还是一个样，最终还是找到了方法。


第一个就是在选择型号的时候，虽然我们是ESP32但是我们需要选择为ESP8266，原因仅是因为烧录的地址需要为0x00处，而在Thonny中无法指定烧录地址。

第二个就是选择固件文件的时候，选择版本低的。我一开始选择是最新版本的，然后烧录之后还是无法使用，之后跟了一个又一个教程去一次次烧录，最终以一种朴素的方法解决了我的问题，那就是换个低版本的，我最终成功的版本是我框出来的，v1.19.1的那个。

接下来我们就可以开始愉快的敲代码了。


microPython的用法我们可以从官网里找到，我们接下来也会围绕着官方文档进行。

http://micropython.com.cn/en/latet/esp32/quickref.html

GPIO

from machine import Pin

其实直接import machine也是可以的，但是熟悉Python的小伙伴会知道，当我们仅用其中几个类的时候，我们使用上面的写法会更好一些。 


我们照例是来点个灯，点个灯就需要控制GPIO，那么我们需要初始化GPIO对吧，其实所有单片机编程起来那可以说是万变不离其宗，我们都知道需要做什么，唯一不同的是不同单片机在不同环境下我们操作的方法不同。


我们知道需要初始化GPIO口了，那么我们从人家官方文档里找对应的方法即可。

参数id就是指定GPIO口的，我们直接填入对应的GPIO口的编号即可。

第二个模式可以有多种可选的，常用的就开头两个。

第三个选择是否连接电阻，一般我们仅输出的话是不连的，但如果要配置外部中断的话，也就是要输入，那么再根据需要去配置上拉电阻还是下拉电阻，简单来说，配置上拉电阻，那么默认读入的是高电平；配置下拉电阻，那么默认读入的是低电平。

第四个指定初始值。


后两个很少用，感兴趣的小伙伴自行查阅文档，这里就不介绍了。


配置完之后我们还需要操作GPIO的输出，可用的有以下五种方法。

最重要的是最后一个value函数。


上面文档中的解释太复杂了，简单来说就是给value函数参数，那么就算是指定输出的电平，如果不给参数，那么就算是读入当前这个GPIO口的电平。


知道怎么操作之后我们就可以开始写代码了。


以防有小伙伴不懂怎么使用Thonny来写MicroPython，我这里简单说一下。我们的MicroPython设备（也就是我们连接的ESP32）默认是只有一个boot.py文件，这个文件的内容是每次ESP32上电之后都会执行的。

但是我们一般代码不写在这里面，我们会新建一个py文件，然后把要执行的内容写在别的文件里，调试的时候我们直接在对应文件的界面进行运行即可，当我们程序定稿之后需要变成上电自动执行之后，我们会封装好执行内容，然后写在main.py文件里。


没错，main.py文件也是要我们新建的，ESP32上电后自动执行boot.py，然后boot.py会自动执行main.py。


关于这个我们可以做一个小实验。


我们分别在两个文件中都输出一句话。

没错，我们可以直接使用print来对我们的程序进行调试。


然后我们保存修改后的文件，接着按下复位键。

可以看到确实是像我们之前说的先执行boot再执行main。


闲话说多了，接下来我们开始点灯。

from machine import Pin p5=Pin(5,Pin.OUT) while True:    #p5.off()    #p5.low()    p5.value(0)

编辑完代码之后我们直接点击运行当前脚本，然后我们就惊奇地发现，点灯成功。

根据我们的惯例（参考之前的一文速通ESP8266），点灯完之后我们就要让它闪烁了。


延时

import time

根据官方文档提供的例子我们可以发现，一共有三种延时函数。例子中最后两行设计的函数是统计ESP32运行时间的，基本没什么用。


三种延时函数分别是以秒，毫秒，微妙为单位的，我们随便用哪个都行，一般毫秒用的比较多。


接下来我们让LED闪烁。

from machine import Pinimport time p5=Pin(5,Pin.OUT) while True:    p5.value(0)    time.sleep_ms(500)    p5.value(1)    time.sleep_ms(500)

由于照片拍不出闪烁的效果，那我就不拍照了（偷懒）。


要让LED闪烁，除了上面的这种写法，我们还可以通过获取当前GPIO的电平，然后输出相反的电平即可，这样可以使代码更加简洁。

from machine import Pinimport time p5=Pin(5,Pin.OUT) while True:    p5.value(not p5.value())    time.sleep_ms(500)

平时C/C++的代码写多了，代码报错之后才回过神来，Python中的取非不是！，而是直接写个not。


定时

from machine import Timer

除了使用延时，我们还可以使用定时来完成我们的LED闪烁效果，最重要的是不会阻塞我们的主逻辑。

我们最多可以使用四个定时器，因此给Timer构造函数传入的参数（ID）应为0~3，但实际上你写多少都可以，但是会对传入的参数进行一个取余4的操作，因此还是只能用四个定时器，如果多个定时器初始化传入的ID相同，那么后面创建的定时器会把更早创建的覆盖掉。


获得定时器对象之后我们还需要调用初始化（init）函数（有一点点反直觉，因为我觉得创建时候调用的构造函数应该就算是初始化了）。


init函数的第一个参数传入定时器周期，单位为毫秒。


第二个参数指定模式，定时任务可以是只执行一次（Timer.ONE_SHOT），也可以是周期性执行（Timer.PERIODIC）。


第三个参数传入回调函数，可以是lambda。回调函数接收一个参数，这个参数是定时器对象，但这个不是我们的定时器。


还有一个函数，上面的例子中没有提及，那就是取消定时任务，我们使用deinit函数。


接下来我们使用定时来改一下LED闪烁的代码。


from machine import Pinfrom machine import Timerimport time p5=Pin(5,Pin.OUT)t0=Timer(0)t1=Timer(1) t0.init(period=500,mode=Timer.PERIODIC,callback=lambda t:p5.value(not p5.value()))t1.init(period=5000,mode=Timer.ONE_SHOT,callback=lambda t:t0.deinit()) while True:    pass

上面的代码使用了两个定时器，一个定了500ms为周期的翻转GPIO电平的定时任务，一个定了5s后执行一次的关闭第一个定时器任务的定时任务。


可以发现LED闪烁五次之后就停止闪烁了，可见我们已经成功地使用了定时。


PWM

from machine import PWM

除了让LED闪烁，我们还可以让LED出现呼吸灯的效果，那就是输出PWM。

构造函数的第一个参数指定一个GPIO口，这里不是填入一个编号了，而是需要传入一个GPIO的对象。


第二个参数填入PWM的频率，范围是1~40MHz。


后面有三个参数，三选一。分别是duty，duty_u16，duty_ns。都是设置占空比的。


duty传入0~1024，duty_u16传入0~65535，而duty_ns传入的是以纳秒为单位来设置脉冲宽度。


接下来我就直接拿代码来演示一下LED呼吸灯。

from machine import Pinfrom machine import PWMimport time p5=Pin(5,Pin.OUT)p=PWM(p5,10000000,duty=0) while True:    for i in range(0,1023):        p.duty(i)        time.sleep_ms(1)            for i in range(1023,0,-1):        p.duty(i)        time.sleep_ms(1)

另外，使用freq，duty，dutty_u16等同名函数也可以获取与设置频率，占空比等数值，可以参考GPIO的value函数。


ADC

from machine import ADC

参数比较简单，第一个是GPIO口的对象，第二个指定输入衰减，也可以看作是指定模拟信号的电压范围。


然后我们调用read函数即可读取转换值。

from machine import Pinfrom machine import ADCimport time p5=Pin(5,Pin.IN)a=ADC(p5,atten=ADC.ATTN_11DB) while True:    print(a.read())    time.sleep(1)

操作存储器

在一文速通ESP8266中我们有操作存储器的介绍，在ESP32中就不介绍了，因为完全跟Python进行文件操作一样。

import time with open("abc.txt","w") as f:    f.write("abcdefg")    f.flush() while True:    with open("boot.py","r") as f:        while True:            data=f.readline()            if not data:                break            print(data)    time.sleep(5)

UART串口通信

from machine import UART

第一个参数id，传入0~2，一个三个UART。

第二个参数传入波特率。

第三第四个指定tx和rx的引脚。


后面几个参数指定数据位，校验位，停止位什么的，我们一般用默认即可，也就是八位数据一位停止，没有校验。感兴趣的小伙伴可以自行去官方文档查阅。


构造出UART对象之后，我们最重要的是收发数据，相关的函数也很简单，write和read分别是写和读，其中read需要传入一个参数，表示我们最多读取的字节数，读出的也是字节。


使用any函数可以判断是否有接收到数据。

from machine import Pinfrom machine import UARTimport time u=UART(1,9600,rx=9,tx=10)p5=Pin(5,Pin.OUT) while True:    u.write("hello world")    if u.any():        p5.value(not p5.value())        print(u.read(1024))    time.sleep(1)

WiFi

import network

接下来就是重头戏啦，我们连接上WiFi之后就可以进入物联网的重头戏MQTT啦。

与之前ArduinoIDE写的ESP8266不一样的是，如果我们连上了WiFi，那么默认是就算是后续断开了WiFI，它也会继续尝试连接，而我们之前ESP8266里我们需要手动检测WiFi是否断开，如果断开则需要手动重连。


关于这一点我已经用下面的代码测试过了，逻辑就是一开始配置一下连接代码，在主循环里，如果WiFI保持连接，那么LED亮，没有连接WiFi则LED灭。


运行下面代码后我手动断过几次手机热点，只要手机热点再次开启，那么LED还是会亮起来。

from machine import Pinimport timeimport network p5=Pin(5,Pin.OUT) w=network.WLAN(network.STA_IF)w.active(True)w.connect("zhetu","zhetu123") while True:    if w.isconnected():        p5.off()    else:        p5.on()    time.sleep(1)


其实根据我上面的代码以及更上面官方文档中的示例，也可以知道如何连接WiFi了，或者直接把官方文档中提供的函数do_connect拿来用即可。

def do_connect():    import network    wlan = network.WLAN(network.STA_IF)    wlan.active(True)    if not wlan.isconnected():        print('connecting to network...')        wlan.connect('ssid', 'key')        while not wlan.isconnected():            pass    print('network config:', wlan.ifconfig())

不过接下来就稍微讲一下连接WiFi的过程。


首先是要构造出一个WLAN对象出来，WLAN的构造函数只需要传入一个参数，那就是指定模式，可以是network.AP_IF和network.STA_IF，分别是开热点模式和连接WiFi模式，我们一般用的都是后者，因为就算我们开了热点那也是无法联网的。


接着是激活它，用active函数，给一个布尔值，True则激活，False则关闭。


然后使用connect函数连接WiFi，传入两个参数，依次是WiFI名和WiFi密码。


最后使用isconnected函数，根据其返回值判断是否连接上了WiFi。


disconnect函数可以手动关闭WiFi连接。


其他一些获取配置或是设置配置的函数，感兴趣的小伙伴可以去官方文档查看。


MQTT

https://github.com/juwul/umqtt_aws_iot

MicroPython中没有提供给我们MQTT，但是不代表我们不能使用MicroPython来使用MQTT，上面的链接中我们可以获取到别人写的第三方库。


直接把lib中的simple.py复制出来放进ESP32中即可。

这个第三方库的优点是它非常小，仅仅是200行代码，后续如果想要学习一下，200行代码也不难看懂，并且操作非常简单。缺点是因为小巧，所以功能相对简陋，但是绝对是够用的，另外一个就是在github上，需要有点方法才可以进去，进不去的小伙伴复制下面的代码也是一样的。

import usocket as socketimport ustruct as structfrom ubinascii import hexlify class MQTTException(Exception):    pass class MQTTClient:     def __init__(self, client_id, server, port=0, user=None, password=None, keepalive=0,                 ssl=False, ssl_params={}):        if port == 0:            port = 8883 if ssl else 1883        self.client_id = client_id        self.sock = None        self.addr = socket.getaddrinfo(server, port)[0][-1]        self.ssl = ssl        self.ssl_params = ssl_params        self.pid = 0        self.cb = None        self.user = user        self.pswd = password        self.keepalive = keepalive        self.lw_topic = None        self.lw_msg = None        self.lw_qos = 0        self.lw_retain = False     def _send_str(self, s):        self.sock.write(struct.pack("!H", len(s)))        self.sock.write(s)     def _recv_len(self):        n = 0        sh = 0        while 1:            b = self.sock.read(1)[0]            n |= (b & 0x7f) << sh            if not b & 0x80:                return n            sh += 7     def set_callback(self, f):        self.cb = f     def set_last_will(self, topic, msg, retain=False, qos=0):        assert 0 <= qos <= 2        assert topic        self.lw_topic = topic        self.lw_msg = msg        self.lw_qos = qos        self.lw_retain = retain     def connect(self, clean_session=True):        self.sock = socket.socket()        self.sock.connect(self.addr)        if self.ssl:            import ussl            self.sock = ussl.wrap_socket(self.sock, **self.ssl_params)        msg = bytearray(b"\x10\0\0\x04MQTT\x04\x02\0\0")        msg[1] = 10 + 2 + len(self.client_id)        msg[9] = clean_session << 1        if self.user is not None:            msg[1] += 2 + len(self.user) + 2 + len(self.pswd)            msg[9] |= 0xC0        if self.keepalive:            assert self.keepalive < 65536            msg[10] |= self.keepalive >> 8            msg[11] |= self.keepalive & 0x00FF        if self.lw_topic:            msg[1] += 2 + len(self.lw_topic) + 2 + len(self.lw_msg)            msg[9] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3            msg[9] |= self.lw_retain << 5        self.sock.write(msg)        #print(hex(len(msg)), hexlify(msg, ":"))        self._send_str(self.client_id)        if self.lw_topic:            self._send_str(self.lw_topic)            self._send_str(self.lw_msg)        if self.user is not None:            self._send_str(self.user)            self._send_str(self.pswd)        resp = self.sock.read(4)        assert resp[0] == 0x20 and resp[1] == 0x02        if resp[3] != 0:            raise MQTTException(resp[3])        return resp[2] & 1     def disconnect(self):        self.sock.write(b"\xe0\0")        self.sock.close()     def ping(self):        self.sock.write(b"\xc0\0")     def publish(self, topic, msg, retain=False, qos=0):        pkt = bytearray(b"\x30\0\0\0")        pkt[0] |= qos << 1 | retain        sz = 2 + len(topic) + len(msg)        if qos > 0:            sz += 2        assert sz < 2097152        i = 1        while sz > 0x7f:            pkt[i] = (sz & 0x7f) | 0x80            sz >>= 7            i += 1        pkt[i] = sz        #print(hex(len(pkt)), hexlify(pkt, ":"))        self.sock.write(pkt, i + 1)        self._send_str(topic)        if qos > 0:            self.pid += 1            pid = self.pid            struct.pack_into("!H", pkt, 0, pid)            self.sock.write(pkt, 2)        self.sock.write(msg)        if qos == 1:            while 1:                op = self.wait_msg()                if op == 0x40:                    sz = self.sock.read(1)                    assert sz == b"\x02"                    rcv_pid = self.sock.read(2)                    rcv_pid = rcv_pid[0] << 8 | rcv_pid[1]                    if pid == rcv_pid:                        return        elif qos == 2:            assert 0     def subscribe(self, topic, qos=0):        assert self.cb is not None, "Subscribe callback is not set"        pkt = bytearray(b"\x82\0\0\0")        self.pid += 1        struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid)        #print(hex(len(pkt)), hexlify(pkt, ":"))        self.sock.write(pkt)        self._send_str(topic)        self.sock.write(qos.to_bytes(1, "little"))        while 1:            op = self.wait_msg()            if op == 0x90:                resp = self.sock.read(4)                #print(resp)                assert resp[1] == pkt[2] and resp[2] == pkt[3]                if resp[3] == 0x80:                    raise MQTTException(resp[3])                return     # Wait for a single incoming MQTT message and process it.    # Subscribed messages are delivered to a callback previously    # set by .set_callback() method. Other (internal) MQTT    # messages processed internally.    def wait_msg(self):        res = self.sock.read(1)        self.sock.setblocking(True)        if res is None:            return None        if res == b"":            raise OSError(-1)        if res == b"\xd0":  # PINGRESP            sz = self.sock.read(1)[0]            assert sz == 0            return None        op = res[0]        if op & 0xf0 != 0x30:            return op        sz = self._recv_len()        topic_len = self.sock.read(2)        topic_len = (topic_len[0] << 8) | topic_len[1]        topic = self.sock.read(topic_len)        sz -= topic_len + 2        if op & 6:            pid = self.sock.read(2)            pid = pid[0] << 8 | pid[1]            sz -= 2        msg = self.sock.read(sz)        self.cb(topic, msg)        if op & 6 == 2:            pkt = bytearray(b"\x40\x02\0\0")            struct.pack_into("!H", pkt, 2, pid)            self.sock.write(pkt)        elif op & 6 == 4:            assert 0     # Checks whether a pending message from server is available.    # If not, returns immediately with None. Otherwise, does    # the same processing as wait_msg.    def check_msg(self):        self.sock.setblocking(False)        return self.wait_msg()

人家项目的名字我框出来了，那么我们就直接在ESP32中新建一个“umqtt.py”然后把代码直接复制进去就可以直接使用了(文件名随意起)。


由于这个第三方库比较简单，估计了解MQTT的小伙伴，光是看人家的函数命名都可以知道该怎么操作，那么我们就先简单来个小例子，然后再稍微讲解一下。

from machine import Pinimport timeimport networkimport umqtt p5=Pin(5,Pin.OUT) w=network.WLAN(network.STA_IF)w.active(True)w.connect("zhetu","zhetu123") def mqtt_callback(topic,data):    #回调函数，直接把主题名和内容体打印出来    print(topic,data) while not w.isconnected():    #确保WiFI连接上了，后续才可以连接MQTT服务器    pass    mqtt=umqtt.MQTTClient("连接服务器用的ID","mqtt服务器IP")mqtt.set_callback(mqtt_callback)mqtt.connect()mqtt.subscribe(b"aaa") while True:    mqtt.check_msg()            #检测是否接受到订阅信息    if w.isconnected():        p5.off()    else:        p5.on()    time.sleep(1)

根据构造函数我们可以得知，我们仅需要传入两个参数即可，一个是连接MQTT服务器用的ID，另一个是MQTT服务器的IP地址。剩下的配置也可以进行设置，但是不是必须的，我们简单演示一下就没用上。

那么构造完得到一个MQTTClient的对象之后，我们要做的是设置回调函数，就是下面这个函数，我们传入一个回调函数，回调函数接收两个参数，依次是主题名和内容体。

那么我是怎么知道的呢，当然是看代码啦。


根据对MQTT的了解以及对英文名称大概的意思就可以知道。

设置完回调函数之后我们还需要订阅主题，要收到消息我们还需要先订阅是吧。


使用下面的函数，直接传入主题名称即可。

接着就可以连接MQTT服务器了。


用下面这函数，可以不传入参数。

上面的配置完之后我们离成功就剩一点点啦。


我们需要在主循环体里不断的检测是否收到数据，收到数据则会主动调用回调函数。


检测使用下面这个函数。

看得出来这还涉及到了另外一个函数，感兴趣的小伙伴自己去看代码吧。


最后还有一个功能，那就是发布信息。


我们至少需要传入两个参数，依次是主题名和内容体。

小结

至此，我们就速通了ESP32啦。


这里就小小的总结一下踩过的坑，第一个就是环境配置啦，不要选择最新的固件文件！！！


第二个就是串口的引脚选择，尽量咱就是使用人家默认的引脚，虽然自定义也可以，但是很容易出问题，比如我不信邪，试过使用与UART的ID不匹配的引脚，结果遇到了print使用不了以及RX引脚接收不到数据等的事情，所以虽然ESP32的引脚映射功能很强大，但是在UART串口中，我们还是尽量使用官方推荐的默认引脚。


最后一个是MQTT，值得一提的是，接受订阅信息的回调函数的两个参数都是字节类型的，因此如果我们需要锁定某个主题的时候，需要把和回调函数的主题名的参数比较的那个字符串转为字节。


最后的最后打个小广告。文中涉及的所有资料我都已经打包好了，关注我的公众号“折途想要敲代码”回复关键词“ESP32”即可免费下载。


当然，文章中的链接也都是可以直接点击前往下载对应的资源的。

发表评论