一文速通ESP32（基于MicroPython）——含示例代码_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要贵的，但是结果却是相反的，我当初买的时候不知道，光挑贵的买了，结果好像是配置反而更低了。不懂是我有什么地方不明白的还是店家定价定错了。懂的小伙伴可以在评论区教我一下。

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

环境配置

Thonny, Python IDE for beginnershttps://thonny.org/我们使用Thonny作为IDE，进入上面的链接直接下载即可。

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

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

MicroPython - Python for microcontrollershttps://micropython.org/download/

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

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

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

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

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

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

Quick reference for the ESP32 — MicroPython latest documentationhttp://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 Pin
import 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 Pin
import 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 Pin
from machine import Timer
import 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 Pin
from machine import PWM
import 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 Pin
from machine import ADC
import 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 Pin
from machine import UART
import 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 Pin
import time
import 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

GitHub - juwul/umqtt_aws_iot: Publish UMQTT messages with Micropython to AWS IoTPublish UMQTT messages with Micropython to AWS IoT - juwul/umqtt_aws_iothttps://github.com/juwul/umqtt_aws_iot

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

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

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

import usocket as socket
import ustruct as struct
from 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 Pin
import time
import network
import 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”即可免费下载。

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