Python 控制 Wokwi ESP32 LED 跑馬燈動畫
在 ESP32 上利用 FreeRTOS 的雙核特性，我們可以將 Core 0 專門用於處理 MQTT 通訊與 WiFi 維護，而 Core 1 則專門負責 LED 跑馬燈動畫。
Core 0 (通訊核心)：
負責
client.loop()。這非常重要，因為 WiFi 協議疊（Protocol Stack）預設在 Core 0 運行。這樣可以減少核心間的上下文切換開關（Context Switch）。使用
vTaskDelay而不是delay()，讓出 CPU 時間給系統背景進程（如 WiFi 管理）。
Core 1 (應用核心)：
負責 LED 的跑馬燈計算與輸出。即便動畫中有
vTaskDelay，也不會卡死 MQTT 的接收。使用
volatile int currentMode確保當 Core 0 修改數值時，Core 1 能立刻讀取到最新的指令。
#include <WiFi.h>
#include <PubSubClient.h>
// --- 設定區 ---
const char* ssid = "Wokwi-GUEST";
const char* password = "";
const char* mqtt_server = "mqtt-dashboard.com";
const char* mqtt_topic = "alex9ufo/LEDcommand";
// LED 接腳
int ledPins[] = {19, 18, 5, 4, 2, 27, 26, 25, 33, 32};
int numLeds = 10;
// 共享變數 (使用 volatile 確保跨核心讀取正確)
volatile int currentMode = 0;
WiFiClient espClient;
PubSubClient client(espClient);
// 任務句柄
TaskHandle_t LEDTask;
// --- MQTT 接收回調 (執行於 Core 0) ---
void callback(char* topic, byte* payload, unsigned int length) {
char message[length + 1];
memcpy(message, payload, length);
message[length] = '\0';
int newMode = atoi(message);
currentMode = newMode; // 更新共享變數
Serial.printf("Core %d 收到指令: %d\n", xPortGetCoreID(), currentMode);
}
// --- Core 0 任務：處理 MQTT 通訊 ---
void MQTTTaskCode(void * pvParameters) {
Serial.printf("MQTT 任務運行於 Core %d\n", xPortGetCoreID());
for (;;) {
if (!client.connected()) {
String clientId = "ESP32-Client-" + String(random(0xffff), HEX);
if (client.connect(clientId.c_str())) {
client.subscribe(mqtt_topic);
}
}
client.loop();
vTaskDelay(10 / portTICK_PERIOD_MS); // 稍微休息，釋放資源給系統
}
}
// --- Core 1 任務：處理 LED 動畫 (原本的 Loop 內容) ---
void LEDTaskCode(void * pvParameters) {
Serial.printf("LED 任務運行於 Core %d\n", xPortGetCoreID());
static int step = 0;
static bool direction = true;
static unsigned long lastToggle = 0;
static bool flashState = false;
for (;;) {
// 0.5s 閃爍狀態計算
if (millis() - lastToggle >= 500) {
lastToggle = millis();
flashState = !flashState;
}
// 根據模式執行動畫
switch (currentMode) {
case 1: // 上到下
for (int i=0; i<numLeds; i++) {
for(int j=0; j<numLeds; j++) digitalWrite(ledPins[j], i==j);
vTaskDelay(100 / portTICK_PERIOD_MS);
if(currentMode != 1) break;
}
break;
case 2: // 下到上
for (int i=numLeds-1; i>=0; i--) {
for(int j=0; j<numLeds; j++) digitalWrite(ledPins[j], i==j);
vTaskDelay(100 / portTICK_PERIOD_MS);
if(currentMode != 2) break;
}
break;
case 3: // 彈跳
for(int j=0; j<numLeds; j++) digitalWrite(ledPins[j], step==j);
if (direction) step++; else step--;
if (step >= numLeds - 1 || step <= 0) direction = !direction;
vTaskDelay(80 / portTICK_PERIOD_MS);
break;
case 4: // 奇數閃爍
for (int i=0; i<numLeds; i++) digitalWrite(ledPins[i], (flashState && (i%2==0)));
vTaskDelay(50 / portTICK_PERIOD_MS);
break;
case 5: // 偶數閃爍
for (int i=0; i<numLeds; i++) digitalWrite(ledPins[i], (flashState && (i%2!=0)));
vTaskDelay(50 / portTICK_PERIOD_MS);
break;
case 6: // 奇偶交替
for (int i=0; i<numLeds; i++) {
if (flashState) digitalWrite(ledPins[i], i%2==0);
else digitalWrite(ledPins[i], i%2!=0);
}
vTaskDelay(50 / portTICK_PERIOD_MS);
break;
case 0: // 全滅
for (int i=0; i<numLeds; i++) digitalWrite(ledPins[i], LOW);
vTaskDelay(200 / portTICK_PERIOD_MS);
break;
}
}
}
void setup() {
Serial.begin(115200);
for (int i = 0; i < numLeds; i++) pinMode(ledPins[i], OUTPUT);
// WiFi 連線
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) { delay(500); Serial.print("."); }
Serial.println("\nWiFi Connected");
client.setServer(mqtt_server, 1883);
client.setCallback(callback);
// 分配任務到不同的核心
// xTaskCreatePinnedToCore(函式, 任務名稱, 堆疊大小, 參數, 優先度, 句柄, 核心ID)
xTaskCreatePinnedToCore(
MQTTTaskCode, "MQTT_Task", 10000, NULL, 1, NULL, 0 // Core 0 負責通訊
);
xTaskCreatePinnedToCore(
LEDTaskCode, "LED_Task", 10000, NULL, 1, &LEDTask, 1 // Core 1 負責動畫
);
}
void loop() {
// 雙核心架構下，Arduino 的 loop 可以留空或處理次要事務
vTaskDelete(NULL);
}
import tkinter as tk
import paho.mqtt.client as mqtt
# --- 設定區 ---
MQTT_SERVER = "mqtt-dashboard.com"
MQTT_TOPIC = "alex9ufo/LEDcommand"
class LEDController:
def __init__(self, root):
self.root = root
self.root.title("ESP32 LED 遠端控制")
self.root.geometry("300x450")
# 連線狀態
self.status_label = tk.Label(root, text="MQTT 狀態: 正在連線...", fg="orange", font=("Arial", 10))
self.status_label.pack(pady=10)
# 按鈕標題
tk.Label(root, text="選擇 LED 模式", font=("Arial", 12, "bold")).pack(pady=5)
# 模式按鈕定義
modes = [
("1. 從上到下", "1"),
("2. 從下到上", "2"),
("3. 上下來回", "3"),
("4. 1,3,5,7,9 亮", "4"),
("5. 2,4,6,8,10 亮", "5"),
("6. 奇偶交替閃爍", "6"),
("全部熄滅", "0")
]
for text, cmd in modes:
btn = tk.Button(root, text=text, width=20, height=2,
command=lambda c=cmd: self.send_command(c))
btn.pack(pady=5)
# MQTT 初始化
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
try:
self.client.connect(MQTT_SERVER, 1883, 60)
self.client.loop_start()
except:
self.status_label.config(text="MQTT 狀態: 連線失敗", fg="red")
def on_connect(self, client, userdata, flags, rc):
if rc == 0:
self.status_label.config(text="MQTT 狀態: 已連線", fg="green")
else:
self.status_label.config(text=f"MQTT 狀態: 錯誤({rc})", fg="red")
def send_command(self, cmd):
self.client.publish(MQTT_TOPIC, cmd)
print(f"發送指令: {cmd}")
if __name__ == "__main__":
root = tk.Tk()
app = LEDController(root)
root.mainloop()
沒有留言:
張貼留言