from machine import Pin, SoftI2C, UART
import network
import time
from umqtt.simple import MQTTClient
from ssd1306 import SSD1306_I2C
# ================ 配置参数 ================
# WiFi 配置(替换为实际 WiFi)
WIFI_SSID = "Wokwi-GUEST"
WIFI_PASSWORD = ""
# 巴法云 MQTT 配置(控制台获取)
MQTT_CLIENT_ID = "8854b385911048edaa12808fb1f199e6" # 客户端 ID
MQTT_BROKER = "bemfa.com" # 服务器
MQTT_PORT = 9501 # 端口
MQTT_TOPIC_CTRL = "myled" # 控制主题(串口发 1/0)
MQTT_TOPIC_STATE = "mysta" # 状态主题(上报 on/off)
MQTT_USER = "+8618840167598" # 巴法云 UID
MQTT_PASSWORD = "8854b385911048edaa12808fb1f199e6" # 巴法云密码
# ================= 硬件配置(根据实际接线)=================
LED_PIN = 2 # LED连接的GPIO引脚
OLED_SDA = 23 # OLED SDA引脚
OLED_SCL = 18 # OLED SCL引脚
UART_TX = 1 # 串口发送引脚(默认UART0)
UART_RX = 3 # 串口接收引脚(默认UART0)
# ================= 调试辅助函数 =================
def log(msg):
"""打印调试信息到串口"""
msg = f"[LOG] {msg}"
uart.write(msg.encode() + b"\r\n")
print(msg) # 同时输出到仿真控制台
# ================= 硬件初始化 =================
led = Pin(LED_PIN, Pin.OUT)
led.off() # 初始关闭LED
i2c = SoftI2C(sda=Pin(OLED_SDA), scl=Pin(OLED_SCL), freq=400000)
oled = SSD1306_I2C(128, 64, i2c)
oled.fill(0)
oled.text("System Starting...", 0, 20)
oled.show()
uart = UART(1, baudrate=115200, tx=Pin(UART_TX), rx=Pin(UART_RX))
uart.init(bits=8, parity=None, stop=1) # 显式初始化串口
# ================= 网络与MQTT =================
def connect_wifi():
log("WiFi connecting...")
oled.fill(0)
oled.text("WiFi Connecting", 0, 20)
oled.show()
# Wokwi仿真无需实际连接
log("WiFi connected (simulated)")
oled.fill(0)
oled.text("WiFi Connected", 0, 20)
oled.show()
return True
def connect_mqtt():
log(f"MQTT connecting: ID={MQTT_CLIENT_ID}, User={MQTT_USER}")
oled.fill(0)
oled.text("MQTT Connecting", 0, 20)
oled.show()
try:
client = MQTTClient(
client_id=MQTT_CLIENT_ID,
server=MQTT_BROKER,
port=MQTT_PORT,
user=MQTT_USER,
password=MQTT_PASSWORD,
keepalive=60
)
# 标准连接方式,移除内部方法调用
client.connect()
log("MQTT connected successfully")
# 订阅控制主题
if client:
log(f"Subscribing to topic: {MQTT_TOPIC_CTRL}")
client.subscribe(MQTT_TOPIC_CTRL)
log(f"Subscribed to {MQTT_TOPIC_CTRL}")
oled.fill(0)
oled.text("MQTT Connected", 0, 20)
oled.show()
return client
except Exception as e:
log(f"MQTT connect failed: {str(e)}")
oled.fill(0)
oled.text("MQTT Error", 0, 20)
oled.text(str(e)[:16], 0, 40)
oled.show()
return None
# ================= 状态同步 =================
def publish_state(mqtt_client, state):
"""发布状态到MQTT,带详细调试"""
status = "ON" if state else "OFF"
log(f"Publishing to {MQTT_TOPIC_STATE}: {status}")
if not mqtt_client:
log("MQTT client not initialized!")
return False
try:
mqtt_client.publish(MQTT_TOPIC_STATE, status)
log("Publish successful")
return True
except Exception as e:
log(f"Publish failed: {str(e)}")
return False
def update_system(state, mqtt_client):
"""更新LED、OLED和MQTT状态"""
# 控制LED
led.value(state)
# 更新OLED
status = "ON" if state else "OFF"
oled.fill(0)
oled.text(f"LED: {status}", 0, 20)
# 发布MQTT状态
if mqtt_client:
pub_result = publish_state(mqtt_client, state)
oled.text("MQTT: " + ("Sent" if pub_result else "Fail"), 0, 40)
oled.show()
time.sleep(1)
# ================= 主程序 =================
def main():
log("\r\n=== LED Control System ===")
log(f"MQTT Config: ID={MQTT_CLIENT_ID[:8]}..., User={MQTT_USER[:8]}...")
log(f"Topics: Ctrl={MQTT_TOPIC_CTRL}, State={MQTT_TOPIC_STATE}")
uart.write(b"\r\n=== 控制命令 ===\r\n")
uart.write(b"1 = 打开LED\r\n")
uart.write(b"0 = 关闭LED\r\n")
uart.write(b"h = 发送心跳\r\n")
uart.write(b"s = 显示状态\r\n")
uart.write(b"----------------\r\n")
# 连接网络和MQTT
mqtt_client = connect_mqtt() if connect_wifi() else None
# 主循环
last_state = False
heartbeat_timer = time.ticks_ms()
while True:
# 处理串口输入
if uart.any():
cmd = uart.readline().strip().lower()
log(f"Received command: {cmd}")
if cmd == b"1":
log("Turning LED ON")
update_system(True, mqtt_client)
last_state = True
elif cmd == b"0":
log("Turning LED OFF")
update_system(False, mqtt_client)
last_state = False
elif cmd == b"h":
log("Sending heartbeat")
publish_state(mqtt_client, last_state)
elif cmd == b"s":
log(f"Current state: LED={last_state}, MQTT={bool(mqtt_client)}")
uart.write(b"\r\n")
# 每30秒发送心跳包
if time.ticks_ms() - heartbeat_timer > 30000 and mqtt_client:
log("Periodic heartbeat")
publish_state(mqtt_client, last_state)
heartbeat_timer = time.ticks_ms()
# 检查MQTT消息(非阻塞)
if mqtt_client:
try:
msg = mqtt_client.check_msg()
if msg:
topic, payload = msg
log(f"MQTT message received: {topic.decode()} = {payload.decode()}")
# 处理控制命令
if topic.decode() == MQTT_TOPIC_CTRL:
if payload == b'1':
update_system(True, mqtt_client)
last_state = True
elif payload == b'0':
update_system(False, mqtt_client)
last_state = False
except Exception as e:
log(f"MQTT check error: {str(e)}")
time.sleep(0.1)
# ================= 启动 =================
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
log("Program terminated by user")
finally:
if 'mqtt_client' in locals() and mqtt_client:
mqtt_client.disconnect()
led.off()
uart.deinit()
oled.fill(0)
oled.text("System Stopped", 0, 20)
oled.show()