ESP32物联网 自动浇水系统（五）

📅 2026-06-18 13:34:38
ESP32物联网自动浇水系统五
\n

图形化积木方式编程学生入门测试还够用，做项目还是有一定局限性。 今天使用AI来写microPython代码并将固件代码刷入esp32 ，全自动化。

还是使用TraeCN+deepseek-v4-flash输入：

/plan 根据此文档 mixly-esp32-guide.md ， 使用mpython代码重写并编译刷入com4口的esp32 ，  支持wifi,mqtt断网重连，异常处理完善， 代码逻辑清晰易懂

图片MicroPython 固件：ESP32 接入自动浇水系统

本文档指导如何在 ESP32 上使用 MicroPython 固件，实现自动浇水系统的设备端功能。

1. 环境搭建

1.1 安装 esptool（固件刷写工具）

pip install esptool

1.2 安装 ampy（文件上传工具）

pip install adafruit-ampy

1.3 获取 MicroPython 固件

  • 官方下载: https://micropython.org/download/ESP32_GENERIC/
  • 本文基于 MicroPython v1.28.0 开发，建议使用同版本或更新版本

1.4 刷写 MicroPython 固件到 ESP32

# 擦除 flashesptool.py --port COM4 erase_flash# 刷入固件esptool.py --port COM4 --baud 460800 write_flash -z 0x1000 ESP32_GENERIC-20260406-v1.28.0.bin

刷写时需要按住 ESP32 的 BOOT 键再上电，或按住 BOOT 键点一下 EN 键进入下载模式。

1.5 上传 main.py 到 ESP32

# 将 main.py 上传到 ESP32（自动运行）ampy --port COM4 put main.py# 软重启使程序生效（或用 ESP32 上的 RST 键）# ampy put 完成后 ESP32 会自动软重启

如果上传报错 failed to access COM4，先发送 Ctrl+C 中断正在运行的程序，再执行 ampy 命令。

2. 硬件接线

引脚分配

ESP32 引脚 连接目标 说明
GPIO 34 土壤湿度传感器 AO ADC 输入（读取湿度）
GPIO 26 L9110S B-1A 电机 PWM 控制
GPIO 27 L9110S B-1B 电机方向控制
GPIO 2 板载 LED 状态指示（闪烁=正常，长亮=异常）
3.3V 传感器 VCC —
5V L9110S VCC 电机驱动供电（电流 > 200mA 时建议独立供电）
GND 所有组件 GND 共地

L9110S 控制逻辑

动作 B-1A (GPIO 26) B-1B (GPIO 27)
正转（PWM 调速） PWM.duty(值) LOW
停止 PWM.duty(0) LOW

常用 PWM 值：200（低速，≈20%）→ 400（中速，≈39%）→ 600（中高速，≈59%）→ 800（高速，≈78%）→ 1023（全速，100%）

MicroPython 的 PWM duty 范围为 0-1023（10 位），与 Arduino 的 0-255（8 位）不同。 代码中以 duty(800) 实现约 78% 转速，等效于 Arduino 的 analogWrite(26, 200)。

3. 配置说明

编辑 main.py 顶部的配置区即可适配自己的环境：

# ----- 设备标识 -----DEVICE_ID = "esp32/001"# 设备编号，多台设备时修改，如 esp32/002# ----- WiFi 配置 -----WIFI_SSID = "你的WiFi名称"# 2.4G WiFiWIFI_PWD  = "你的WiFi密码"# ----- MQTT 配置 -----MQTT_BROKER = "127.0.0.1"# MQTT 服务器地址MQTT_PORT   = 1883# MQTT 端口MQTT_USER   = "esp32"# MQTT 用户名MQTT_PWD    = "esp32@2026"# MQTT 密码# ----- 业务参数 -----MOISTURE_THRESHOLD = 3000# 自动浇水阈值 (ADC > 此值触发)WATERING_DURATION  = 10# 每次浇水持续时间 (秒)

切换设备时只需修改 DEVICE_ID 即可，MQTT 主题会自动基于该值拼接。

4. MQTT 主题

4.1 主题生成规则

所有主题基于 DEVICE_ID 自动拼接：

TOPIC_STATUS   = DEVICE_ID + "/status"# esp32/001/statusTOPIC_MOISTURE = DEVICE_ID + "/soil_moisture"# esp32/001/soil_moistureTOPIC_CMD      = DEVICE_ID + "/watering/cmd"# esp32/001/watering/cmdTOPIC_ACK      = DEVICE_ID + "/watering/ack"# esp32/001/watering/ack

4.2 完整主题表

主题 方向 说明
esp32/001/status ESP32 → 服务器 上报 "online"（心跳）
esp32/001/soil_moisture ESP32 → 服务器 上报湿度原始值 (0-4095)
esp32/001/watering/cmd 服务器 → ESP32 接收浇水指令
esp32/001/watering/ack ESP32 → 服务器 确认浇水完成，回传 "1"

5. 程序结构

5.1 核心函数

函数 说明
ensure_wifi(timeout=15) 确保 WiFi 在线，断开自动重连，返回 True/False
ensure_mqtt() 确保 MQTT 在线，断开自动重连，返回 True/False
safe_publish(topic, msg) 安全发布 MQTT 消息（客户端为 None 时自动忽略）
mqtt_callback(topic, msg) MQTT 消息回调，收到 "1" 时启动浇水
start_watering() 开始浇水：L9110S 正转（PWM 800），持续 WATERING_DURATION 秒
stop_watering() 停止浇水，发送 ACK 确认
main() 主循环入口

5.2 工作流程

启动  ├─ 打印设备信息  ├─ ensure_wifi()    → 连接 WiFi（失败不阻塞）  ├─ ensure_mqtt()    → 连接 MQTT（失败不阻塞）  └─ 进入主循环（while True）       ├─ [每 10 秒] ensure_mqtt() → 检查并保持连接       ├─ [每循环]   mqtt_client.check_msg() → 处理 MQTT 消息       ├─ [每 30 秒] safe_publish(TOPIC_STATUS, "online") → 心跳       ├─ [每 60 秒] 读取 ADC → 上报湿度 → 干燥则自动浇水       ├─ [每 1 秒]  LED 闪烁（正常）/ 长亮（异常）       └─ 休眠 0.2 秒

5.3 WiFi/MQTT 自动重连

固件内置双层的断线自动恢复机制：

  1. 网络层 - ensure_wifi()：WiFi 断开时自动重置状态并重连，最多等待 15 秒
  2. 应用层 - ensure_mqtt()：每次调用先确保 WiFi 在线，再检查 MQTT 连接有效性，断开则重建连接

主循环每 10 秒调用一次 ensure_mqtt()，确保设备始终在线。

5.4 LED 状态指示

状态 LED 行为 含义
WiFi + MQTT 已连接 每 1 秒闪烁 正常运行
WiFi 或 MQTT 断开 长亮 连接异常

6. 数据上报

6.1 在线状态（每 30 秒）

# 主循环中每 30 秒执行safe_publish(TOPIC_STATUS, "online")

设备离线由服务端心跳检测（120 秒超时）自动标记，ESP32 无需上报 offline。

6.2 土壤湿度（每 60 秒）

# 主循环中每 60 秒执行moisture = adc.read_u16() >> 4# 12 位 ADC (0-4095)print("土壤湿度:", moisture)safe_publish(TOPIC_MOISTURE, str(moisture))# 干燥自动浇水if moisture > MOISTURE_THRESHOLD andnot watering_active:    start_watering()

7. 浇水控制

7.1 远程指令浇水

服务端向 esp32/001/watering/cmd 发送 "1"，ESP32 收到后执行：

defmqtt_callback(topic, msg):    topic_str = topic.decode()    msg_str   = msg.decode()if topic_str == TOPIC_CMD and msg_str == "1":        start_watering()  # PWM 200 正转，持续 10 秒

7.2 自动浇水

主循环每 60 秒读取湿度，当 ADC 值 > MOISTURE_THRESHOLD（默认 3000）且未在浇水时自动触发。

7.3 浇水确认

浇水结束后自动向 esp32/001/watering/ack 发送 "1" 确认完成。

8. 完整代码

完整代码位于 tools/Esp32Mpy/main.py，核心结构：

  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
  •  
"""ESP32 自动浇水系统 - MicroPython 固件=====================================功能: WiFi连接, MQTT通信, 土壤湿度采集, L9110S电机控制, 自动浇水硬件接线:  GPIO 34 - 土壤湿度传感器 AO  GPIO 26 - L9110S B-1A (PWM)  GPIO 27 - L9110S B-1B"""import networkimport utimeimport machinefrom umqtt.simple import MQTTClient# ============================================================# 配置区# ============================================================DEVICE_ID = "esp32/001"WIFI_SSID = "P70"WIFI_PWD  = "123456789"MQTT_BROKER = "127.0.0.1"MQTT_PORT   = 1883MQTT_USER   = "esp32"MQTT_PWD    = "esp32@2026"MOISTURE_THRESHOLD = 3000WATERING_DURATION  = 10# ============================================================# 主题常量# ============================================================TOPIC_STATUS   = DEVICE_ID + "/status"TOPIC_MOISTURE = DEVICE_ID + "/soil_moisture"TOPIC_CMD      = DEVICE_ID + "/watering/cmd"TOPIC_ACK      = DEVICE_ID + "/watering/ack"# ============================================================# 硬件初始化# ============================================================adc      = machine.ADC(machine.Pin(34))pwm_b1a  = machine.PWM(machine.Pin(26), freq=1000, duty=0)  # duty=0 确保启动时电机不转pin_b1b  = machine.Pin(27, machine.Pin.OUT, value=0)         # value=0 确保方向引脚 LOWpin_led  = machine.Pin(2, machine.Pin.OUT, value=0)utime.sleep(0.1)                                              # 等待引脚状态稳定# ============================================================# 全局状态# ============================================================wlan = network.WLAN(network.STA_IF)mqtt_client = Nonewatering_active = Falsewatering_stop_time = 0# ============================================================# WiFi 连接/重连（单函数）# ============================================================def ensure_wifi(timeout=15):    """确保 WiFi 在线，离线时自动重连。返回 True=已连接"""    if wlan.isconnected():        return True    # 重置 WiFi 状态，避免 Internal State Error    try:        wlan.active(False)    except:        pass    utime.sleep(0.5)    wlan.active(True)    utime.sleep(0.5)    if wlan.isconnected():        return True    print("WiFi 连接中:", WIFI_SSID)    wlan.connect(WIFI_SSID, WIFI_PWD)    for i in range(timeout):        if wlan.isconnected():            print("\nWiFi 已连接, IP:", wlan.ifconfig()[0])            return True        print(".", end="")        utime.sleep(1)    print("\nWiFi 连接失败")    return False# ============================================================# MQTT 连接/重连（单函数）# ============================================================def ensure_mqtt():    """确保 MQTT 在线，离线时自动重连。返回 True=已连接"""    global mqtt_client    if not ensure_wifi():        return False    # 客户端存在则检查连接是否有效    if mqtt_client is not None:        try:            mqtt_client.publish(TOPIC_STATUS, "online")            return True        except:            pass        # 连接断开，清理旧客户端        try:            mqtt_client.disconnect()        except:            pass        mqtt_client = None    # 首次连接或重建连接    print("MQTT 连接中...")    try:        client = MQTTClient(            client_id=DEVICE_ID,            server=MQTT_BROKER,            port=MQTT_PORT,            user=MQTT_USER,            password=MQTT_PWD,            keepalive=60        )        client.set_callback(mqtt_callback)        client.connect()        client.subscribe(TOPIC_CMD)        print("MQTT 已连接, 已订阅:", TOPIC_CMD)        client.publish(TOPIC_STATUS, "online")        mqtt_client = client        return True    except Exception as e:        print("MQTT 连接失败:", e)        return False# ============================================================# 安全发布 - 避免重复 None 检查# ============================================================def safe_publish(topic, msg):    """安全发布 MQTT 消息（mqtt_client 为 None 时忽略）"""    if mqtt_client is not None:        try:            mqtt_client.publish(topic, msg)        except:            pass# ============================================================# MQTT 消息回调# ============================================================def mqtt_callback(topic, msg):    """收到浇水指令时启动水泵"""    global watering_active, watering_stop_time    topic_str = topic.decode()    msg_str   = msg.decode()    print("收到指令: [%s] -> %s" % (topic_str, msg_str))    if topic_str == TOPIC_CMD and msg_str == "1":        start_watering()# ============================================================# 水泵控制# ============================================================def start_watering():    """开始浇水：L9110S 正转 (PWM=800, 约78%转速)，持续指定秒数"""    global watering_active, watering_stop_time    print(">>> 开始浇水 (%d 秒)" % WATERING_DURATION)    pwm_b1a.duty(800)          # Duty ≈ 800 (78% of 1023)    pin_b1b.value(0)    watering_active = True    watering_stop_time = utime.time() + WATERING_DURATIONdef stop_watering():    """停止浇水：L9110S 停止，发送浇水确认"""    global watering_active    print("<<< 停止浇水")    pwm_b1a.duty(0)    pin_b1b.value(0)    watering_active = False    safe_publish(TOPIC_ACK, "1")# ============================================================# 主循环# ============================================================def main():    print("\n========================================")    print(" ESP32 自动浇水系统 (MicroPython)")    print("========================================")    print("设备 ID:", DEVICE_ID)    print("MQTT:", MQTT_BROKER, ":", MQTT_PORT)    print()    # 首次连接（失败也不阻塞，主循环中持续重试）    ensure_wifi()    ensure_mqtt()    last_status_time   = utime.time()    last_moisture_time = utime.time()    last_conn_check    = utime.time()    last_led_toggle    = utime.time()    led_on             = False    is_healthy         = False    while True:        try:            now = utime.time()            # 每 10 秒检查连接状态            if now - last_conn_check >= 10:                is_healthy = ensure_mqtt()                last_conn_check = now            # 检查 MQTT 消息            if mqtt_client is not None:                try:                    mqtt_client.check_msg()                except:                    pass            # 检查浇水是否结束            if watering_active and now >= watering_stop_time:                stop_watering()            # 每 30 秒心跳            if now - last_status_time >= 30:                safe_publish(TOPIC_STATUS, "online")                last_status_time = now            # 每 60 秒读取并上报土壤湿度            if now - last_moisture_time >= 60:                moisture = adc.read_u16() >> 4                print("土壤湿度:", moisture)                safe_publish(TOPIC_MOISTURE, str(moisture))                if moisture > MOISTURE_THRESHOLD and not watering_active:                    start_watering()                last_moisture_time = now            # ---- LED 状态指示 ----            # 正常（WiFi+MQTT 已连接）→ 每 1 秒闪烁            # 异常（WiFi 或 MQTT 断开）→ 长亮            if is_healthy:                if now - last_led_toggle >= 1:                    led_on = not led_on                    pin_led.value(1 if led_on else 0)                    last_led_toggle = now            else:                pin_led.value(1)  # 异常时长亮                last_led_toggle = now                led_on = False            utime.sleep(0.2)        except OSError as e:            print("网络异常:", e)            utime.sleep(5)        except Exception as e:            print("运行异常:", e)            utime.sleep(5)# ============================================================# 程序入口# ============================================================main()

9. 串口调试

使用串口监视器查看输出

硬件连接时通过 USB 查看 ESP32 的 print() 输出：

# Python 脚本方式（推荐）python -c "import serial, times = serial.Serial('COM4', 115200, timeout=2)time.sleep(0.5)s.write(b'\x03')   # Ctrl+C 中断程序time.sleep(0.5)s.write(b'\x04')   # Ctrl+D 软重启time.sleep(3)print(s.read(5000).decode('utf-8', 'replace'))s.close()"

正常启动输出示例：

======================================== ESP32 自动浇水系统 (MicroPython)========================================设备 ID: esp32/001MQTT: 127.0.0.1 : 1883WiFi 连接中: P70.....WiFi 已连接, IP: 192.168.1.100MQTT 连接中...MQTT 已连接, 已订阅: esp32/001/watering/cmd土壤湿度: 2450

常用操作

操作 发送 说明
中断程序 Ctrl+C (\x03) 回到 MicroPython REPL
软重启 Ctrl+D (\x04) 重启并重新运行 main.py
进入 raw REPL Ctrl+A (\x01) ampy 传输文件所需模式

 

🌱🌱🌱未完待续

 

 

💬 讨论区

暂无评论，快来发表第一条评论吧！

📮 需求咨询