ESP32 (ESP-IDF in VS Code) MFRC522 + MQTT + PYTHON TKinter +SQLite
ESP32 VS Code程式
; PlatformIO Project Configuration File
PYTHON TKinter程式
import tkinter as tk
from tkinter import messagebox, scrolledtext
import sqlite3
import paho.mqtt.client as mqtt
import datetime
import threading
import re
import os
import json
import time
# 嘗試導入 pygame 和 numpy (跨平台聲音解決方案)
try:
import numpy as np
import pygame
# 初始化 Pygame Mixer (使用 44.1kHz, 16-bit, 單聲道標準)
sample_rate = 44100
pygame.mixer.init(frequency=sample_rate, size=-16, channels=1, buffer=512)
PYGAME_AVAILABLE = True
except ImportError:
PYGAME_AVAILABLE = False
print("注意: pygame 或 numpy 模組無法使用。聲音功能將被禁用。")
print("請執行 'pip install pygame numpy' 安裝。")
except pygame.error as e:
# 某些情況下 mixer 初始化會失敗 (例如沒有音訊設備)
PYGAME_AVAILABLE = False
print(f"注意: Pygame Mixer 初始化失敗 ({e})。聲音功能將被禁用。")
# --- 預設值與設定 ---
class DefaultSettings:
WIFI_SSID = "alex9ufo"
WIFI_PASSWORD = "alex1234"
MQTT_BROKER = "broker.mqtt-dashboard.com"
LED_CONTROL_TOPIC = "alex9ufo/VSCode/LedControl"
LED_STATUS_TOPIC = "alex9ufo/VSCode/LedStatus"
RFID_UID_TOPIC = "alex9ufo/VSCode/RFIDUid"
CONFIG_TOPIC = "alex9ufo/VSCode/Config"
TIMER_DURATION_SEC = 20
DB_NAME = "VSCode_RFID.db"
# --- 資料庫操作類別 (SQLite) ---
class DBManager:
# (此類別內容與之前版本保持一致，處理資料庫連線、查詢、新增、刪除等操作)
def __init__(self, db_name):
self.db_name = db_name
def connect(self):
try:
conn = sqlite3.connect(self.db_name, check_same_thread=False)
return conn
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"連接資料庫失敗: {e}")
return None
def create_table(self):
conn = self.connect()
if not conn: return
try:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT,
time TEXT,
event TEXT,
note TEXT
)
""")
conn.commit()
messagebox.showinfo("資料庫", "資料表建立成功。")
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"建立資料表失敗: {e}")
finally:
conn.close()
def add_event(self, event, note):
conn = self.connect()
if not conn: return
try:
now = datetime.datetime.now()
date_str = now.strftime("%Y-%m-%d")
time_str = now.strftime("%H:%M:%S")
cursor = conn.cursor()
cursor.execute("INSERT INTO events (date, time, event, note) VALUES (?, ?, ?, ?)",
(date_str, time_str, event, note))
conn.commit()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"新增事件失敗: {e}")
finally:
conn.close()
def fetch_all(self):
conn = self.connect()
if not conn: return []
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM events ORDER BY id DESC")
return cursor.fetchall()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"查詢所有資料失敗: {e}")
return []
finally:
conn.close()
def check_uid_exists(self, uid):
conn = self.connect()
if not conn: return False
try:
cursor = conn.cursor()
cursor.execute("SELECT 1 FROM events WHERE event = ?", (uid,))
return cursor.fetchone() is not None
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"查詢 UID 失敗: {e}")
return False
finally:
conn.close()
def fetch_all_uids(self):
conn = self.connect()
if not conn: return set()
try:
cursor = conn.cursor()
cursor.execute("SELECT DISTINCT event FROM events WHERE LENGTH(event) = 8 AND event GLOB '[0-9A-F]*'")
return {row[0] for row in cursor.fetchall()}
except sqlite3.Error:
return set()
finally:
conn.close()
def delete_record(self, key_type, value):
conn = self.connect()
if not conn: return False
try:
cursor = conn.cursor()
if key_type == 'ID':
cursor.execute("DELETE FROM events WHERE id=?", (value,))
elif key_type == 'UID':
cursor.execute("DELETE FROM events WHERE event LIKE ?", (value,))
conn.commit()
return cursor.rowcount > 0
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"刪除失敗: {e}")
return False
finally:
conn.close()
def query_record(self, key_type, value):
conn = self.connect()
if not conn: return []
try:
cursor = conn.cursor()
if key_type == 'ID':
cursor.execute("SELECT * FROM events WHERE id=?", (value,))
elif key_type == 'UID':
cursor.execute("SELECT * FROM events WHERE event LIKE ?", (value,))
return cursor.fetchall()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"查詢失敗: {e}")
return []
finally:
conn.close()
def delete_db(self):
if os.path.exists(self.db_name):
try:
os.remove(self.db_name)
messagebox.showinfo("資料庫", f"資料庫檔案 {self.db_name} 已刪除。")
return True
except Exception as e:
messagebox.showerror("資料庫錯誤", f"刪除檔案失敗，請確保程式已關閉與資料庫的所有連線: {e}")
return False
return False
# --- Tkinter GUI 應用程式 ---
class RFIDGUI(tk.Tk):
def __init__(self):
super().__init__()
self.title("ESP32 RFID/LED 控制中心")
self.geometry("850x650")
self.db_manager = DBManager(DefaultSettings.DB_NAME)
# 狀態變數初始化 (Topic, SSID, Mode...)
self.wifi_ssid = tk.StringVar(value=DefaultSettings.WIFI_SSID)
self.wifi_password = tk.StringVar(value=DefaultSettings.WIFI_PASSWORD)
self.mqtt_broker = tk.StringVar(value=DefaultSettings.MQTT_BROKER)
self.led_control_topic = tk.StringVar(value=DefaultSettings.LED_CONTROL_TOPIC)
self.led_status_topic = tk.StringVar(value=DefaultSettings.LED_STATUS_TOPIC)
self.rfid_uid_topic = tk.StringVar(value=DefaultSettings.RFID_UID_TOPIC)
self.timer_duration = tk.StringVar(value=str(DefaultSettings.TIMER_DURATION_SEC))
self.config_topic = DefaultSettings.CONFIG_TOPIC
self.mode = tk.StringVar(value="新增")
self.led_status = tk.StringVar(value="OFF")
self.rfid_message = tk.StringVar(value="無卡號")
# 【修正點 1】初始化 Pygame 聲音物件 (分別用於新增和比對模式)
self.add_mode_freq = 850
self.compare_mode_freq = 450
# 聲音持續時間縮短為 0.5 秒
self.beep_sound_add = self._create_beep_sound(self.add_mode_freq, 0.5) if PYGAME_AVAILABLE else None
self.beep_sound_compare = self._create_beep_sound(self.compare_mode_freq, 0.5) if PYGAME_AVAILABLE else None
self.mqtt_client = self._setup_mqtt(self.mqtt_broker.get())
self._create_widgets()
self.db_manager.create_table()
self.update_log()
threading.Thread(target=self._start_mqtt_loop, daemon=True).start()
self.protocol("WM_DELETE_WINDOW", self._on_closing)
def _on_closing(self):
if PYGAME_AVAILABLE:
try:
pygame.mixer.quit()
except Exception:
pass
self.destroy()
# --- 聲音播放函式 (使用 Pygame Mixer) ---
def _create_beep_sound(self, frequency, duration_s, sample_rate=44100):
"""生成並返回 Pygame.mixer.Sound 物件 (用於不同頻率)"""
if not PYGAME_AVAILABLE:
return None
t = np.linspace(0, duration_s, int(sample_rate * duration_s), False)
note = np.sin(frequency * t * 2 * np.pi)
audio = note * (2**15 - 1)
audio = audio.astype(np.int16)
return pygame.mixer.Sound(audio.tobytes())
# --- MQTT 連線與處理 (與之前版本保持一致) ---
def _setup_mqtt(self, broker_addr):
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = self._on_connect
client.on_message = self._on_message
try:
client.connect(broker_addr, 1883, 60)
except Exception:
return None
return client
def _start_mqtt_loop(self):
if self.mqtt_client:
try:
self.mqtt_client.loop_forever()
except Exception as e:
print(f"MQTT 迴圈錯誤: {e}")
def _on_connect(self, client, userdata, flags, reason_code, properties=None):
if reason_code == 0:
print("MQTT 連線成功")
client.subscribe(self.led_status_topic.get())
client.subscribe(self.rfid_uid_topic.get())
else:
print(f"MQTT 連線失敗, 代碼: {reason_code}")
def _on_message(self, client, userdata, msg):
payload = msg.payload.decode()
topic = msg.topic
if topic == self.led_status_topic.get():
self.update_led_status(payload)
self.db_manager.add_event(f"LedStatus:{payload}", "LED 狀態更新")
elif topic == self.rfid_uid_topic.get():
self.handle_rfid_uid(payload)
def _apply_settings_and_reconnect(self):
# ... (設定邏輯與之前版本保持一致)
try:
timer_sec = int(self.timer_duration.get())
if timer_sec <= 0:
raise ValueError
except ValueError:
messagebox.showerror("設定錯誤", "Timer 秒數必須是有效的正整數。")
return
if self.mqtt_client and self.mqtt_client.is_connected():
self.mqtt_client.disconnect()
new_broker = self.mqtt_broker.get()
self.mqtt_client = self._setup_mqtt(new_broker)
threading.Thread(target=self._start_mqtt_loop, daemon=True).start()
if not self.mqtt_client:
messagebox.showerror("設定更新", "Tkinter 端 MQTT 連線失敗，無法發送配置到 ESP32。")
return
config_data = {
"ssid": self.wifi_ssid.get(), "pass": self.wifi_password.get(),
"broker": self.mqtt_broker.get(), "ctrlTopic": self.led_control_topic.get(),
"statTopic": self.led_status_topic.get(), "uidTopic": self.rfid_uid_topic.get(),
"timerSec": timer_sec
}
json_payload = json.dumps(config_data)
try:
self.mqtt_client.publish(self.config_topic, json_payload, retain=True)
time.sleep(0.5)
self.mqtt_client.publish(self.config_topic, "", retain=True)
messagebox.showinfo("設定更新成功", "設定已更新。")
except Exception as e:
messagebox.showerror("MQTT 發送錯誤", f"無法發送配置到 ESP32: {e}")
# --- 閃爍並恢復函式 ---
def _flash_and_restore(self, initial_status):
"""
在單獨的執行緒中執行：先發送 FLASH，然後等待短暫延遲 (確保閃爍 2 次)，再恢復狀態。
"""
flash_topic = self.led_control_topic.get()
self.mqtt_client.publish(flash_topic, "FLASH")
print(f"-> MQTT: {flash_topic} 發送 FLASH 訊號")
# 等待 2.2 秒，確保完成 2 次閃爍 (0.5s ON + 0.5s OFF) * 2 + buffer
time.sleep(2.2)
if initial_status and initial_status not in ["FLASH", "FLASHING"]:
self.mqtt_client.publish(flash_topic, initial_status)
print(f"-> MQTT: {flash_topic} 恢復狀態到 {initial_status}")
else:
self.mqtt_client.publish(flash_topic, "OFF")
print(f"-> MQTT: {flash_topic} 恢復狀態到 OFF (初始狀態未知)")
# --- RFID 處理邏輯 (修正點 2: 根據模式播放不同聲音) ---
def handle_rfid_uid(self, uid_code):
"""處理從 ESP32 傳來的 RFID UID 碼"""
mode = self.mode.get()
# 1. 播放對應模式的聲音
if PYGAME_AVAILABLE:
if mode == "新增" and self.beep_sound_add:
threading.Thread(target=self.beep_sound_add.play, daemon=True).start()
elif mode == "比對" and self.beep_sound_compare:
threading.Thread(target=self.beep_sound_compare.play, daemon=True).start()
self.rfid_display.config(text=f"最後 UID: {uid_code}")
current_led_status = self.led_status.get()
if mode == "新增":
if self.db_manager.check_uid_exists(uid_code):
self.rfid_message.set("卡片已存在")
else:
self.db_manager.add_event(uid_code, "UID 新增成功")
self.rfid_message.set("UID 新增成功")
self.after(100, self.update_log)
elif mode == "比對":
allowed_uids = self.db_manager.fetch_all_uids()
if uid_code in allowed_uids:
self.rfid_message.set("卡片正確")
self.db_manager.add_event(uid_code, "UID 比對成功 (存取允許)")
# 驗證成功：發送 FLASH 2次後，恢復到之前的狀態
threading.Thread(target=self._flash_and_restore,
args=(current_led_status,), daemon=True).start()
else:
self.rfid_message.set("卡片錯誤")
self._send_control("FLASH")
self.db_manager.add_event(uid_code, "UID 比對失敗 (存取拒絕)")
self.after(100, self.update_log)
def _send_control(self, command):
"""發送 LED 控制指令 (ON, OFF, FLASH, TIMER)"""
try:
topic = self.led_control_topic.get()
self.mqtt_client.publish(topic, command)
except Exception as e:
messagebox.showerror("MQTT 發送錯誤", f"無法發送指令到 {topic}: {e}")
# --- UI 繪圖和更新 (與之前版本保持一致) ---
def update_led_status(self, status):
self.led_status.set(status)
color = "white"
is_flashing = False
if status == "ON" or status == "TIMER ON":
color = "green"
elif status == "OFF" or status == "TIMER OFF":
color = "red"
elif status == "FLASH":
is_flashing = True
if is_flashing:
current_color = self.led_indicator.itemcget(self.led_circle, "fill")
if current_color == 'red':
color = 'green'
else:
color = 'red'
try:
self._flash_job = self.after(500, lambda: self.update_led_status("FLASH"))
except Exception:
pass
else:
try:
if hasattr(self, '_flash_job'):
self.after_cancel(self._flash_job)
except Exception:
pass
self.led_indicator.itemconfig(self.led_circle, fill=color)
def update_log(self):
self.log_text.delete(1.0, tk.END)
header = "ID | 日期 | 時間 | 事件 (UID/Topic) | 備註\n"
self.log_text.insert(tk.END, header + "="*80 + "\n")
data = self.db_manager.fetch_all()
for row in data:
self.log_text.insert(tk.END, f"{row[0]:<2} | {row[1]:<10} | {row[2]:<8} | {row[3]:<20} | {row[4]}\n")
# --- 資料庫操作方法 (保持結果畫面) ---
def _db_action(self, action):
key_type = self.db_key_type.get()
value = self.query_entry.get().strip()
if not value:
messagebox.showwarning("輸入錯誤", f"請輸入要{action}的 {key_type} 值。")
return
if action == 'delete':
deleted = self.db_manager.delete_record(key_type, value)
if deleted:
messagebox.showinfo("資料庫", f"已成功刪除 {key_type}: {value} 的紀錄。")
self.update_log()
else:
messagebox.showwarning("資料庫", f"未找到 {key_type}: {value} 的紀錄。")
elif action == 'query':
results = self.db_manager.query_record(key_type, value)
self.log_text.delete(1.0, tk.END)
self.log_text.insert(tk.END, f"查詢結果 ({key_type}:{value})\n" + "="*80 + "\n")
if results:
for row in results:
self.log_text.insert(tk.END, f"{row[0]:<2} | {row[1]:<10} | {row[2]:<8} | {row[3]:<20} | {row[4]}\n")
else:
self.log_text.insert(tk.END, "未找到符合的紀錄。\n")
def _delete_all_db(self):
if messagebox.askyesno("確認", "確定要刪除所有紀錄並刪除資料庫檔案嗎？此操作不可恢復。"):
if self.db_manager.delete_db():
self.db_manager.create_table()
self.update_log()
# --- UI 元素建立 (保持不變) ---
def _create_widgets(self):
# Frame 1: 設定與控制
frame_settings = tk.LabelFrame(self, text="連線與控制 (設定可修改)", padx=10, pady=10)
frame_settings.pack(padx=10, pady=10, fill="x")
# ------------------------ 可修改的設定輸入欄位 ------------------------
grid_row = 0
tk.Label(frame_settings, text="Wi-Fi SSID:").grid(row=grid_row, column=0, sticky="w")
tk.Entry(frame_settings, textvariable=self.wifi_ssid, width=20).grid(row=grid_row, column=1, sticky="w", padx=5)
tk.Label(frame_settings, text="Password:").grid(row=grid_row, column=2, sticky="w")
tk.Entry(frame_settings, textvariable=self.wifi_password, show="*", width=20).grid(row=grid_row, column=3, sticky="w", padx=5)
grid_row += 1
tk.Label(frame_settings, text="MQTT Broker:").grid(row=grid_row, column=0, sticky="w")
tk.Entry(frame_settings, textvariable=self.mqtt_broker, width=50).grid(row=grid_row, column=1, columnspan=3, sticky="w", padx=5)
grid_row += 1
tk.Label(frame_settings, text="控制 Topic:").grid(row=grid_row, column=0, sticky="w")
tk.Entry(frame_settings, textvariable=self.led_control_topic, width=50).grid(row=grid_row, column=1, columnspan=3, sticky="w", padx=5)
grid_row += 1
tk.Label(frame_settings, text="狀態 Topic:").grid(row=grid_row, column=0, sticky="w")
tk.Entry(frame_settings, textvariable=self.led_status_topic, width=50).grid(row=grid_row, column=1, columnspan=3, sticky="w", padx=5)
grid_row += 1
tk.Label(frame_settings, text="UID Topic:").grid(row=grid_row, column=0, sticky="w")
tk.Entry(frame_settings, textvariable=self.rfid_uid_topic, width=50).grid(row=grid_row, column=1, columnspan=3, sticky="w", padx=5)
grid_row += 1
tk.Label(frame_settings, text="配置 Topic:").grid(row=grid_row, column=0, sticky="w")
tk.Label(frame_settings, text=self.config_topic, fg="blue").grid(row=grid_row, column=1, columnspan=3, sticky="w", padx=5)
grid_row += 1
tk.Label(frame_settings, text="Timer 秒數:").grid(row=grid_row, column=0, sticky="w")
tk.Entry(frame_settings, textvariable=self.timer_duration, width=10).grid(row=grid_row, column=1, sticky="w", padx=5)
tk.Label(frame_settings, text="Sec (20 內定)").grid(row=grid_row, column=2, sticky="w")
grid_row += 1
# ------------------------ 控制與狀態區 ------------------------
control_row = 0
control_col_start = 4
tk.Button(frame_settings, text="應用設定並重新連線", command=self._apply_settings_and_reconnect).grid(row=control_row, column=control_col_start, columnspan=4, pady=5)
control_row += 1
tk.Button(frame_settings, text="ON", command=lambda: self._send_control("ON"), width=8).grid(row=control_row, column=control_col_start, padx=5, sticky="w")
tk.Button(frame_settings, text="OFF", command=lambda: self._send_control("OFF"), width=8).grid(row=control_row, column=control_col_start + 1, padx=5, sticky="w")
tk.Button(frame_settings, text="FLASH", command=lambda: self._send_control("FLASH"), width=8).grid(row=control_row + 1, column=control_col_start, padx=5, sticky="w")
tk.Button(frame_settings, text="TIMER", command=lambda: self._send_control("TIMER"), width=8).grid(row=control_row + 1, column=control_col_start + 1, padx=5, sticky="w")
tk.Label(frame_settings, text="LED 狀態:").grid(row=control_row + 2, column=control_col_start, sticky="w")
self.led_indicator = tk.Canvas(frame_settings, width=20, height=20)
self.led_indicator.grid(row=control_row + 2, column=control_col_start + 1, sticky="w")
self.led_circle = self.led_indicator.create_oval(5, 5, 20, 20, fill="red")
# Frame 2: 模式與 RFID 顯示
frame_rfid = tk.LabelFrame(self, text="RFID 模式與狀態", padx=10, pady=10)
frame_rfid.pack(padx=10, pady=10, fill="x")
tk.Label(frame_rfid, text="模式:").grid(row=0, column=0, sticky="w")
tk.Radiobutton(frame_rfid, text="新增模式", variable=self.mode, value="新增").grid(row=0, column=1)
tk.Radiobutton(frame_rfid, text="比對模式", variable=self.mode, value="比對").grid(row=0, column=2)
self.rfid_display = tk.Label(frame_rfid, text="最後 UID: 無", font=("Arial", 12, "bold"))
self.rfid_display.grid(row=0, column=3, padx=20, sticky="w")
tk.Label(frame_rfid, textvariable=self.rfid_message, fg="darkblue", font=("Arial", 12, "bold")).grid(row=1, column=0, columnspan=3, pady=5, sticky="w")
# Frame 3: 資料庫操作
frame_db = tk.LabelFrame(self, text="資料庫操作", padx=10, pady=10)
frame_db.pack(padx=10, pady=10, fill="x")
tk.Button(frame_db, text="顯示所有資料", command=self.update_log).grid(row=0, column=0, padx=5)
tk.Button(frame_db, text="建立資料庫", command=self.db_manager.create_table).grid(row=0, column=1, padx=5)
tk.Button(frame_db, text="刪除所有資料(含DB)", command=self._delete_all_db).grid(row=0, column=2, padx=5)
self.db_key_type = tk.StringVar(value="ID")
tk.Radiobutton(frame_db, text="依 ID", variable=self.db_key_type, value="ID").grid(row=1, column=0)
tk.Radiobutton(frame_db, text="依 UID", variable=self.db_key_type, value="UID").grid(row=1, column=1)
self.query_entry = tk.Entry(frame_db, width=15)
self.query_entry.grid(row=1, column=2, padx=5)
tk.Button(frame_db, text="刪除", command=lambda: self._db_action('delete')).grid(row=1, column=3, padx=5)
tk.Button(frame_db, text="查詢", command=lambda: self._db_action('query')).grid(row=1, column=4, padx=5)
# Frame 4: 輸出紀錄
frame_log = tk.LabelFrame(self, text="資料庫紀錄顯示", padx=10, pady=10)
frame_log.pack(padx=10, pady=10, fill="both", expand=True)
self.log_text = scrolledtext.ScrolledText(frame_log, wrap=tk.WORD, height=15)
self.log_text.pack(fill="both", expand=True)
if __name__ == "__main__":
app = RFIDGUI()
app.mainloop()
if PYGAME_AVAILABLE:
try:
pygame.mixer.quit()
except Exception:
pass
沒有留言:
張貼留言