树莓派Pico W存为main.py后无法连接AWS IoT Core MQTT问题

针对你遇到的「代码在IDLE中运行正常，保存为main.py上电后无法建立MQTT连接」的问题，以下是具体的排查方向和修复方案：

核心问题分析 #

IDLE环境下运行时，设备已通过USB获取时间、网络连接完全稳定；而单独上电时，Pico W存在时间未同步、网络连接后未等待稳定、错误无捕获无法定位等潜在问题，这些是导致MQTT连接失败的主要原因。

具体修复步骤 #

1. 强制同步NTP时间（关键） #

Pico W无硬件RTC，上电后默认时间为初始值（如2021-01-01），而AWS IoT的SSL证书验证要求设备时间在证书有效期内，否则直接拒绝连接。在网络连接成功后添加NTP同步代码：

# 网络连接成功后添加NTP时间同步
import ntptime
try:
    ntptime.host = "pool.ntp.org"
    ntptime.settime()
    print(f"时间同步完成: {time.localtime()}")
except Exception as e:
    print(f"时间同步失败: {e}")
    time.sleep(5)
    machine.reset()

2. 优化网络连接后的等待逻辑 #

wifimgr.get_connection()仅返回连接对象，但此时可能IP未完全获取、DNS解析未生效，需等待网络稳定后再尝试MQTT连接：

wlan = wifimgr.get_connection()
if wlan is None:
    print("无法初始化网络连接。")
    while True:
        pass

# 等待IP地址分配完成
while wlan.ifconfig()[0] == '0.0.0.0':
    print("等待IP地址...")
    time.sleep(1)
# 额外等待3秒确保DNS解析生效
time.sleep(3)

3. 给MQTT连接添加异常捕获 #

原代码中mqtt_client.connect()无异常处理，连接失败后直接崩溃无法查看错误信息，添加捕获后可输出具体问题：

print(f"连接MQTT Broker中")
blink_led()
try:
    mqtt_client.connect()
    print("MQTT连接成功")
    blink_led()
except Exception as e:
    error_msg = f"MQTT连接失败: {str(e)}"
    print(error_msg)
    # 将错误写入文件，方便后续通过USB读取排查
    with open("mqtt_error.log", "w") as f:
        f.write(error_msg)
    time.sleep(5)
    machine.reset()

4. 优化PEM证书读取逻辑 #

原读取函数可能因换行符（Windows的\r\n）导致base64解析错误，修改为更鲁棒的提取方式：

def read_pem(file):
    with open(file, "r") as input:
        text = input.read().strip()
        # 移除所有换行符和回车符
        text = text.replace("\n", "").replace("\r", "")
        # 精准提取BEGIN和END之间的base64内容
        start_idx = text.find("-----BEGIN") + len("-----BEGIN")
        end_idx = text.find("-----END")
        base64_text = text[start_idx:end_idx].strip()
        return ubinascii.a2b_base64(base64_text)

5. 确保MQTT_CLIENT_ID为字符串类型 #

原代码中ubinascii.hexlify(machine.unique_id())返回bytes类型，部分版本的simple.py要求Client ID为字符串，需转码：

MQTT_CLIENT_ID = ubinascii.hexlify(machine.unique_id()).decode('utf-8')

6. 验证AWS IoT Policy权限 #

确保设备证书绑定的IoT Policy包含以下权限（替换为你的AWS区域和账号ID）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:你的区域:你的账号ID:client/${iot:ClientId}"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": "arn:aws:iot:你的区域:你的账号ID:topic/distance"
    }
  ]
}

修改后的完整代码 #

import wifimgr
import machine
import gc
import network
import ssl
import time
import ubinascii
import ntptime
from machine import Pin
from simple import MQTTClient
from random import random

try:
    import usocket as socket
except:
    import socket

led = machine.Pin('LED', machine.Pin.OUT)
wlan = wifimgr.get_connection()

if wlan is None:
    print("无法初始化网络连接。")
    while True:
        pass

# 等待IP地址分配完成
while wlan.ifconfig()[0] == '0.0.0.0':
    print("等待IP地址...")
    time.sleep(1)
# 额外等待确保DNS生效
time.sleep(3)

led.on()
def blink_led():
    led.off()
    time.sleep(0.2)
    led.on()
    time.sleep(0.2)

print("Raspberry Pi Pico W 网络连接正常")

# 同步NTP时间
try:
    ntptime.host = "pool.ntp.org"
    ntptime.settime()
    print(f"时间同步完成: {time.localtime()}")
except Exception as e:
    print(f"时间同步失败: {e}")
    time.sleep(5)
    machine.reset()

MQTT_CLIENT_ID = ubinascii.hexlify(machine.unique_id()).decode('utf-8')
MQTT_CLIENT_KEY = "private.pem.key"
MQTT_CLIENT_CERT = "certificate.pem.crt"
MQTT_BROKER = "你的AWS IoT端点"
MQTT_BROKER_CA = "AmazonRootCA1.pem"

def read_pem(file):
    with open(file, "r") as input:
        text = input.read().strip()
        text = text.replace("\n", "").replace("\r", "")
        start_idx = text.find("-----BEGIN") + len("-----BEGIN")
        end_idx = text.find("-----END")
        base64_text = text[start_idx:end_idx].strip()
        return ubinascii.a2b_base64(base64_text)

key = read_pem(MQTT_CLIENT_KEY)
cert = read_pem(MQTT_CLIENT_CERT)
ca = read_pem(MQTT_BROKER_CA)

mqtt_client = MQTTClient(
    MQTT_CLIENT_ID,
    MQTT_BROKER,
    keepalive=30,
    ssl=True,
    ssl_params={
        "key": key,
        "cert": cert,
        "server_hostname": MQTT_BROKER,
        "cert_reqs": ssl.CERT_REQUIRED,
        "cadata": ca,
    },
)

print(f"连接MQTT Broker中")
blink_led()
try:
    mqtt_client.connect()
    print("MQTT连接成功")
    blink_led()
except Exception as e:
    error_msg = f"MQTT连接失败: {str(e)}"
    print(error_msg)
    with open("mqtt_error.log", "w") as f:
        f.write(error_msg)
    time.sleep(5)
    machine.reset()

# 数据上传循环
while True:
    blink_led()
    distance_cm = int(random()*100)
    print(distance_cm)
    try:
        topic = "distance"
        payload = str(distance_cm)
        mqtt_client.publish(topic, payload)
        print(f"已发布到 {topic}: {payload}")
        blink_led()
        time.sleep(2)  # 增加上传间隔，避免过于频繁
    except Exception as e:
        print(f"发布失败: {e}")
        time.sleep(5)

内容的提问来源于stack exchange，提问作者munnangi_harsha