Embedded Files
訂閱端，成功接收
adamliao@adam-de-iMac ~ % mosquitto_sub -h 192.168.0.172 -p 11883 -t 'myTopic'
Hello MQTTnew
{"temperature": 28, "humidity": 67, "date": "2024-09-28", "hostname": "Mars-1-ESP32", "time": "00:13:50"}
{"temperature": 28, "humidity": 68, "date": "2024-09-28", "hostname": "Mars-1-ESP32", "time": "00:14:00"}
{"temperature": 28, "humidity": 67, "date": "2024-09-28", "hostname": "Mars-1-ESP32", "time": "00:14:10"}
{"temperature": 28, "humidity": 68, "date": "2024-09-28", "hostname": "Mars-1-ESP32", "time": "00:14:20"}
ESP32 成功發佈
Connection successful
IP: 192.168.0.198
���間校對完成: 2024-09-28 00:36:28
Successfully connected to MQTT server at 192.168.0.172 on port 11883
Listening on ('0.0.0.0', 80)
Published data to MQTT: {"temperature": 28, "humidity": 64, "date": "2024-09-28", "hostname": "Mars-1-ESP32", "time": "00:36:38"}
Client connected from ('192.168.0.137', 52401)
Connection successful
IP: 192.168.0.198
Client connected from ('192.168.0.137', 52402)
Connection successful
I
import network
import ntptime
from machine import Pin, Timer, RTC
from dht import DHT11
import socket
import utime
from umqtt.simple import MQTTClient # 匯入 MQTT 客戶端庫
import json
import select
# Wi-Fi 設定
SSID = "Bili-Net"
PASSWORD = "0932388283"
HOSTNAME = "Mars-1-ESP32"
# 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()
# 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):
ip = connect_to_wifi()
current_time = get_current_time() # 使用新方法獲取時間
html = """<html>
<head>
<title>ESP32 DHT11 Sensor</title>
<meta http-equiv="refresh" content="5">
<meta charset="UTF-8">
</head>
<body>
<h2>Host: """ + HOSTNAME + """</h2>
<h2>IP Address: """ + ip + """</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 閃爍功能
def blink(timer):
led_pin.value(not led_pin.value()) # 反轉 LED 狀態
# 時間校對與時區設定
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 publish_data(mqtt_client):
temp, hum, error = read_dht()
if not error:
current_time_str = get_current_time()
data = {
'hostname': HOSTNAME,
'date': current_time_str.split(' ')[0],
'time': current_time_str.split(' ')[1],
'temperature': temp,
'humidity': hum
}
payload = json.dumps(data)
try:
mqtt_client.publish(mqtt_topic, payload)
print('Published data to MQTT:', payload)
except Exception as e:
print('Failed to publish data:', str(e))
else:
print('Failed to read sensor data')
# 初始化 MQTT 客戶端並連接到伺服器
def connect_mqtt():
mqtt_client = MQTTClient(client_id, mqtt_server, port=mqtt_port)
try:
mqtt_client.connect()
print('Successfully connected to MQTT server at {} on port {}'.format(mqtt_server, mqtt_port))
return mqtt_client
except Exception as e:
print('Could not connect to MQTT server {}:{}. Error: {}'.format(mqtt_server, mqtt_port, str(e)))
return None
# 啟動 Wi-Fi 並啟動伺服器
def main():
connect_to_wifi() # 啟動 Wi-Fi
sync_time() # 校對時間
timer = Timer(0)
timer.init(period=1000, mode=Timer.PERIODIC, callback=blink) # 啟動 LED 閃爍
# 連接到 MQTT 伺服器
mqtt_client = connect_mqtt()
if mqtt_client is None:
print("MQTT 連接失敗，請檢查伺服器設定。")
return
# 設置 MQTT 發佈定時器，每 10 秒發佈一次
publish_timer = Timer(1)
publish_timer.init(period=10000, mode=Timer.PERIODIC, callback=lambda t: publish_data(mqtt_client))
# 啟動簡單的 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()
Page updated
Google Sites
Report abuse