*ESP32 MQTT WEB溫溼度計

我已根据您的要求对程序进行了修改：

程序在启动时即使无法连接到 MQTT 服务器，也会继续运行。
在运行过程中，如果与 MQTT 服务器的连接断开，程序会自动尝试重新连接。

以下是完整的修改后代码：

python

複製程式碼

import network

import ntptime

from machine import Pin, Timer, RTC

from dht import DHT11

import socket

import utime

from umqtt.simple import MQTTClient

import json

import select

# Wi-Fi 设置

SSID = "Bili-Net"

PASSWORD = "0932388283"

HOSTNAME = "Mars-1-ESP32"

CustomHostname = "Mars-1-ESP32-Custom" # 自定义主机名

# MQTT 设置

mqtt_server = '192.168.0.172'

mqtt_port = 11883 # 测试端口

client_id = HOSTNAME + '_client'

mqtt_topic = 'myTopic'

# NTP 服务器设置

ntptime.host = 'pool.ntp.org' # 使用更可靠的 NTP 服务器

# DHT11 设置

dht_pin = Pin(13)

dht = DHT11(dht_pin)

power_pin = Pin(12, Pin.OUT)

power_pin.on() # 供电给 DHT11

# LED 设置 (GPIO 2)

led_pin = Pin(2, Pin.OUT)

# RTC 初始化

rtc = RTC()

# 初始化全局变量

mqtt_client = None # MQTT 客户端对象

mqtt_connected = False # MQTT 连接状态

# Wi-Fi 连接

def connect_to_wifi():

station = network.WLAN(network.STA_IF)

station.active(True)

station.config(dhcp_hostname=HOSTNAME)

if not station.isconnected():

print('Connecting to WiFi...')

station.connect(SSID, PASSWORD)

while not station.isconnected():

pass

print('Connection successful')

print('IP:', station.ifconfig()[0])

return station.ifconfig()[0]

# 获取当前日期和时间

def get_current_time():

t = rtc.datetime() # 使用 RTC 模块获取当前时间

current_time = "{}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(t[0], t[1], t[2], t[4], t[5], t[6])

return current_time

# 网页内容生成

def web_page(temp, hum, error):

current_time = get_current_time() # 使用新方法获取时间

html = """<html>

<head>

<title>ESP32 DHT11 Sensor</title>

</head>

<body>

<h2>Host: """ + HOSTNAME + """</h2>

<h2>IP Address: """ + ip_address + """</h2>

<p>Date/Time: """ + current_time + """</p>

<p>Temperature: """ + (str(temp) if not error else 'NA') + """ °C</p>

<p>Humidity: """ + (str(hum) if not error else 'NA') + """ %</p>

</body>

</html>"""

return html

# 温湿度读取

def read_dht():

try:

dht.measure()

temp = dht.temperature()

hum = dht.humidity()

return temp, hum, False

except:

return None, None, True

# LED 闪烁功能（调整为亮 1 秒，熄灭 3 秒）

def blink(timer):

global led_state, led_counter

if led_state:

led_counter += 1

if led_counter >= 1: # 亮 1 秒

led_pin.off()

led_state = False

led_counter = 0

else:

led_counter += 1

if led_counter >= 3: # 熄灭 3 秒

led_pin.on()

led_state = True

led_counter = 0

# 初始化 LED 状态变量

led_state = True # LED 初始为亮

led_counter = 0 # 计数器

# 时间校对与时区设置

def sync_time():

try:

ntptime.settime() # 从 NTP 服务器获取时间

set_timezone() # 设置时区

print("时间校对完成:", get_current_time())

except Exception as e:

print("时间校对失败，错误:", str(e))

def set_timezone():

offset = 8 * 3600 # UTC+8 小时

current_time = utime.time() # 获取当前 UTC 秒数

local_time_tuple = utime.localtime(current_time + offset) # 转为本地时间元组

rtc.datetime((local_time_tuple[0], local_time_tuple[1], local_time_tuple[2], 0,

local_time_tuple[3], local_time_tuple[4], local_time_tuple[5], 0))

# 初始化 MQTT 客户端并连接到服务器

def connect_mqtt():

global mqtt_client, mqtt_connected

mqtt_client = MQTTClient(client_id, mqtt_server, port=mqtt_port)

try:

mqtt_client.connect()

mqtt_connected = True

print('Successfully connected to MQTT server at {} on port {}'.format(mqtt_server, mqtt_port))

except Exception as e:

mqtt_connected = False

mqtt_client = None

print('Could not connect to MQTT server {}:{}. Error: {}'.format(mqtt_server, mqtt_port, str(e)))

# 数据发布到 MQTT 服务器

def publish_data(timer):

global mqtt_client, mqtt_connected

temp, hum, error = read_dht()

if not error:

current_time = get_current_time()

data = {

"timestamp": current_time,

"hostname": CustomHostname, # 使用自定义的主机名

"ip": ip_address,

"Temperature": temp,

"Humidity": hum

}

payload = json.dumps(data)

if not mqtt_connected:

print('Attempting to reconnect to MQTT server...')

connect_mqtt()

if mqtt_connected:

try:

mqtt_client.publish(mqtt_topic, payload)

print('Published data to MQTT:', payload)

except Exception as e:

print('Failed to publish data:', str(e))

mqtt_connected = False

mqtt_client = None

else:

print('MQTT not connected. Data not published.')

else:

print('Failed to read sensor data')

# 启动 Wi-Fi 并启动服务器

def main():

global ip_address

ip_address = connect_to_wifi() # 获取 IP 地址

sync_time() # 校对时间

# 初始化 LED 状态

led_pin.on() # 初始时 LED 亮

# 启动 LED 闪烁定时器

led_timer = Timer(0)

led_timer.init(period=1000, mode=Timer.PERIODIC, callback=blink) # 每 1 秒调用一次 blink 函数

# 尝试连接到 MQTT 服务器

connect_mqtt()

if mqtt_connected:

print("MQTT 连接成功。")

else:

print("MQTT 连接失败，程序将继续运行，并尝试自动重连。")

# 设置 MQTT 发布定时器，每 10 秒发布一次

mqtt_timer = Timer(1)

mqtt_timer.init(period=10000, mode=Timer.PERIODIC, callback=publish_data)

# 启动简单的 HTTP 服务器

addr = socket.getaddrinfo('0.0.0.0', 80)[0][-1]

s = socket.socket()

s.bind(addr)

s.listen(1)

s.setblocking(False) # 设置非阻塞模式

print('Listening on', addr)

while True:

# 处理客户端连接

try:

cl, addr = s.accept()

print('Client connected from', addr)

cl.settimeout(5.0)

cl_file = cl.makefile('rwb', 0)

while True:

line = cl_file.readline()

if not line or line == b'\r\n':

break

temp, hum, error = read_dht()

response = web_page(temp, hum, error)

cl.send('HTTP/1.1 200 OK\n')

cl.send('Content-Type: text/html\n')

cl.send('Connection: close\n\n')

cl.sendall(response)

cl.close()

except OSError:

pass # 没有客户端连接，继续执行

utime.sleep_ms(100) # 避免过度占用 CPU 资源

main()

修改说明：

全局变量：
- 增加了全局变量 mqtt_client 和 mqtt_connected，用于在各个函数之间共享 MQTT 客户端对象和连接状态。
程序在启动时即使无法连接到 MQTT 服务器，也会继续运行：
- 修改了 connect_mqtt 函数，不再返回 mqtt_client，而是更新全局变量 mqtt_client 和 mqtt_connected。
- 在 main 函数中，即使 MQTT 连接失败，程序也不会终止，而是继续运行，并在后续尝试自动重连。

python
複製程式碼
def connect_mqtt():

global mqtt_client, mqtt_connected

mqtt_client = MQTTClient(client_id, mqtt_server, port=mqtt_port)

try:

mqtt_client.connect()

mqtt_connected = True

print('Successfully connected to MQTT server at {} on port {}'.format(mqtt_server, mqtt_port))

except Exception as e:

mqtt_connected = False

mqtt_client = None

print('Could not connect to MQTT server {}:{}. Error: {}'.format(mqtt_server, mqtt_port, str(e)))

- 在 main 函数中：

python
複製程式碼
connect_mqtt()

if mqtt_connected:

print("MQTT 连接成功。")

else:

print("MQTT 连接失败，程序将继续运行，并尝试自动重连。")

自动重连机制：
- 在 publish_data 函数中，如果发现 mqtt_connected 为 False，会尝试重新连接。
- 发布数据前，检查 mqtt_connected，如果未连接，则调用 connect_mqtt() 尝试重连。
- 如果重新连接成功，则继续发布数据；否则，打印提示信息，跳过此次数据发布。

python
複製程式碼
def publish_data(timer):

global mqtt_client, mqtt_connected

temp, hum, error = read_dht()

if not error:

current_time = get_current_time()

data = {

"timestamp": current_time,

"hostname": CustomHostname,

"ip": ip_address,

"Temperature": temp,

"Humidity": hum

}

payload = json.dumps(data)

if not mqtt_connected:

print('Attempting to reconnect to MQTT server...')

connect_mqtt()

if mqtt_connected:

try:

mqtt_client.publish(mqtt_topic, payload)

print('Published data to MQTT:', payload)

except Exception as e:

print('Failed to publish data:', str(e))

mqtt_connected = False

mqtt_client = None

else:

print('MQTT not connected. Data not published.')

else:

print('Failed to read sensor data')

调整 LED 闪烁模式：
- 之前的 LED 闪烁功能已按照您的要求调整为亮 1 秒，熄灭 3 秒。
删除了 data 字典中的 "status" 键值：
- 在 publish_data 函数的 data 字典中，已移除 "status" 键。
其他修改：
- 将 ip_address 定义为全局变量，以便在 web_page 函数中使用。

python
複製程式碼
def web_page(temp, hum, error):

current_time = get_current_time()

html = """...<h2>IP Address: """ + ip_address + """</h2>..."""

return html

注意事项：

资源管理：在实现自动重连时，要确保每次连接失败后，正确释放资源，防止重复创建客户端对象导致内存泄漏。
异常处理：所有异常都被捕获并处理，防止程序崩溃。
全局变量的使用：使用全局变量时，要在函数内声明 global，以确保对全局变量的修改在整个程序中生效。

测试建议：

网络异常测试：可以通过断开网络或关闭 MQTT 服务器来模拟连接失败的情况，观察程序是否能够正确尝试重连。
功能测试：确保在 MQTT 连接正常时，数据能够成功发布，并在网页上正确显示传感器数据。

希望这份修改后的程序能够满足您的需求。

Page updated

Google Sites

Report abuse