Wokwi ESP.32 RFID模擬 + Python Sqlite 資料管理
Wokwi ESP32程式
# --- ```pip install --upgrade paho-mqtt
# --- ```python
import tkinter as tk
from tkinter import messagebox
import sqlite3
import paho.mqtt.client as mqtt
from datetime import datetime
import threading
import time
import random
# --- MQTT 設定 ---
MQTT_BROKER = "broker.mqttgo.io"
MQTT_TOPIC_RFID = "alex9ufo/rfidUID" # 從 RFID 模擬器接收 UID 的主題
MQTT_TOPIC_AUTO_ADD = "2k7" # 觸發自動生成 UID 並新增的主題
# 為確保客戶端ID唯一性，新增隨機數
MQTT_CLIENT_ID = "Python_RFID_DB_App" + str(random.randint(1000, 9999))
# MQTT 重連與狀態顯示相關設定
MQTT_RECONNECT_INTERVAL = 10 # 每10秒嘗試重連一次
MQTT_BLINK_INTERVAL_MS = 500 # 每500毫秒閃爍一次
# --- 資料庫設定 ---
DB_NAME = "rfid_records.db"
TABLE_NAME = "rfid_logs"
class RFIDApp:
def __init__(self, master):
self.master = master
master.title("MQTT RFID 資料庫管理")
master.geometry("800x600")
self.conn = None
self.cursor = None
self.connect_db()
# --- GUI 元件 ---
# 確保在設置 MQTT 狀態標籤之前先創建 UI 元件
self.create_widgets()
# --- MQTT 客戶端設定 ---
# 使用 paho-mqtt 的最新 API 版本 Client 初始化方式
self.mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, MQTT_CLIENT_ID)
self.mqtt_client.on_connect = self.on_connect
self.mqtt_client.on_message = self.on_message
self.mqtt_client.on_disconnect = self.on_disconnect # 新增斷線回呼
self.mqtt_client_connected = False # 初始狀態為未連線
self.mqtt_blink_state = True # 閃爍狀態，用於交替顯示
self.mqtt_blink_job = None # 用於取消閃爍的 job ID
self.mqtt_reconnect_job = None # 用於取消重連的 job ID
# 初始顯示 MQTT 連線狀態，此時標籤已創建
self.update_mqtt_status_label() # 這會立即觸發閃爍或顯示初始狀態
# 嘗試連接 MQTT Broker (非同步執行)
self.connect_mqtt_async() # 更改為非阻塞連接方法
# 啟動 MQTT 迴圈在單獨的執行緒中
self.mqtt_thread = threading.Thread(target=self.mqtt_loop_forever, daemon=True)
self.mqtt_thread.start()
# 最後顯示所有記錄，確保UI和MQTT設置完成
self.display_all_records()
def connect_db(self):
try:
self.conn = sqlite3.connect(DB_NAME)
self.cursor = self.conn.cursor()
messagebox.showinfo("資料庫", f"已連接到資料庫: {DB_NAME}")
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"連接資料庫失敗: {e}")
self.conn = None
self.cursor = None
def create_table(self):
if not self.conn:
messagebox.showerror("錯誤", "資料庫未連接。")
return
try:
self.cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
UID TEXT NOT NULL,
Date TEXT NOT NULL,
Time TEXT NOT NULL
)
""")
self.cursor.execute(f"INSERT OR IGNORE INTO sqlite_sequence (name, seq) VALUES (?, ?)", (TABLE_NAME, 1000))
self.conn.commit()
messagebox.showinfo("資料庫", f"表格 '{TABLE_NAME}' 已建立或已存在，起始 ID 設定為 1001。")
self.display_all_records()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"建立表格失敗: {e}")
# --- 輔助函式：設定輸入框內容 ---
def set_entry_fields(self, record_id="", uid="", date="", time=""):
"""統一設定 ID, UID, Date, Time 輸入框的內容。"""
self.id_entry.config(state=tk.NORMAL)
self.uid_entry.config(state=tk.NORMAL)
self.date_entry.config(state=tk.NORMAL)
self.time_entry.config(state=tk.NORMAL)
self.id_entry.delete(0, tk.END)
self.id_entry.insert(0, str(record_id))
self.uid_entry.delete(0, tk.END)
self.uid_entry.insert(0, str(uid))
self.date_entry.delete(0, tk.END)
self.date_entry.insert(0, str(date))
self.time_entry.delete(0, tk.END)
self.time_entry.insert(0, str(time))
self.date_entry.config(state='readonly') # Date 和 Time 設為只讀
self.time_entry.config(state='readonly')
# --- 輔助函式：清空輸入框 ---
def clear_entry_fields(self):
"""清空所有輸入框，並將 Date 和 Time 設為只讀。"""
self.set_entry_fields("", "", "", "")
# --- 輔助函式：根據 ID 填充欄位 ---
def fill_fields_by_id(self, event=None):
"""根據 ID 查詢資料並填充到輸入框中。"""
if not self.conn:
messagebox.showerror("錯誤", "資料庫未連接。")
return
record_id = self.id_entry.get()
if not record_id:
self.clear_entry_fields()
return
try:
self.cursor.execute(f"SELECT * FROM {TABLE_NAME} WHERE id = ?", (record_id,))
record = self.cursor.fetchone()
if record:
self.set_entry_fields(record[0], record[1], record[2], record[3])
else:
self.clear_entry_fields()
self.id_entry.insert(0, record_id) # 即使沒找到，也保留用戶輸入的ID
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"查詢資料失敗: {e}")
self.clear_entry_fields()
# --- 功能：手動新增一筆 ---
def insert_record(self, uid_data):
if not self.conn:
messagebox.showerror("錯誤", "資料庫未連接。")
return
# (1) 修正: 手動新增時，如果 UID 為空則提示
if not uid_data:
messagebox.showwarning("警告", "手動新增需要輸入 UID。")
return
try:
current_date = datetime.now().strftime("%Y-%m-%d")
current_time = datetime.now().strftime("%H:%M:%S")
self.cursor.execute(f"INSERT INTO {TABLE_NAME} (UID, Date, Time) VALUES (?, ?, ?)",
(uid_data, current_date, current_time))
self.conn.commit()
new_record_id = self.cursor.lastrowid
# (1) 修正: 手動新增時顯示 Date 和 Time 欄位到輸入框
self.set_entry_fields(new_record_id, uid_data, current_date, current_time)
messagebox.showinfo("新增資料", "資料已成功新增。")
self.display_all_records()
return True
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"新增資料失敗: {e}")
return False
# --- 輔助函式：生成隨機 UID ---
def generate_random_uid(self):
return ''.join(random.choices('0123456789ABCDEF', k=8))
# --- 功能：更正一筆 ---
def update_record(self):
if not self.conn:
messagebox.showerror("錯誤", "資料庫未連接。")
return
record_id = self.id_entry.get()
new_uid = self.uid_entry.get()
if not record_id:
messagebox.showwarning("警告", "請輸入要更正的 ID。")
return
# (2) 修正: 更正一筆時，輸入ID會將顯示ID, UID, Date, Time四個欄位再進行修正
# 如果UID欄位為空，或者UID有值但與當前顯示的不同且用戶未確認（這是為了避免直接保存未經確認的修改）
# 此處的邏輯調整為：如果 UID 為空，或不是當前顯示的 UID 且使用者未確認，則先載入資料供編輯。
current_displayed_uid = self.uid_entry.get() if self.uid_entry.winfo_exists() else ""
if not new_uid: # 如果 UID 欄位為空，表示用戶可能想先查詢並填充
self.fill_fields_by_id()
if self.uid_entry.get() and self.id_entry.get() == record_id: # 找到資料
messagebox.showinfo("提示", f"ID: {record_id} 的資料已載入。請修改 UID 後再次點擊「更正一筆」以保存。")
else: # 沒找到資料
messagebox.showwarning("找不到資料", f"找不到 ID 為 {record_id} 的記錄。")
return
# 如果 UID 欄位不為空，且與當前顯示的 UID 不同，則詢問確認
if current_displayed_uid and new_uid != current_displayed_uid:
if not messagebox.askyesno("確認更正", f"您確定要將 ID: {record_id} 的 UID 從 '{current_displayed_uid}' 更正為 '{new_uid}' 嗎？"):
return # 用戶取消，不執行更新
# 如果 ID 和 UID 都有值，且用戶已確認，則執行實際的更新
try:
self.cursor.execute(f"UPDATE {TABLE_NAME} SET UID = ? WHERE id = ?", (new_uid, record_id))
self.conn.commit()
if self.cursor.rowcount > 0:
self.cursor.execute(f"SELECT * FROM {TABLE_NAME} WHERE id = ?", (record_id,))
updated_record = self.cursor.fetchone()
if updated_record:
self.set_entry_fields(updated_record[0], updated_record[1], updated_record[2], updated_record[3])
messagebox.showinfo("更正資料", f"ID: {record_id} 的資料已成功更新。")
else:
self.clear_entry_fields()
messagebox.showinfo("更正資料", f"ID: {record_id} 的資料已更新為 UID: {new_uid}。")
self.display_all_records()
else:
messagebox.showwarning("找不到資料", f"找不到 ID 為 {record_id} 的記錄。")
self.clear_entry_fields()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"更正資料失敗: {e}")
# --- 功能：刪除一筆 ---
def delete_record(self):
if not self.conn:
messagebox.showerror("錯誤", "資料庫未連接。")
return
record_id = self.id_entry.get()
if not record_id:
messagebox.showwarning("警告", "請輸入要刪除的 ID。")
return
try:
self.cursor.execute(f"SELECT * FROM {TABLE_NAME} WHERE id = ?", (record_id,))
record_to_delete = self.cursor.fetchone()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"查詢刪除前資料失敗: {e}")
return
if not record_to_delete:
messagebox.showwarning("找不到資料", f"找不到 ID 為 {record_id} 的記錄。")
self.clear_entry_fields()
return
# (3) 修正: 刪除一筆時顯示 ID, UID, Date, Time 四個欄位並詢問
self.set_entry_fields(record_to_delete[0], record_to_delete[1], record_to_delete[2], record_to_delete[3])
confirm_delete = messagebox.askyesno(
"確認刪除",
f"您確定要刪除以下記錄嗎？\n\n"
f"ID: {record_to_delete[0]}\n"
f"UID: {record_to_delete[1]}\n"
f"日期: {record_to_delete[2]}\n"
f"時間: {record_to_delete[3]}\n\n"
f"此操作不可恢復！"
)
if confirm_delete:
try:
self.cursor.execute(f"DELETE FROM {TABLE_NAME} WHERE id = ?", (record_id,))
self.conn.commit()
if self.cursor.rowcount > 0:
messagebox.showinfo("刪除資料", f"ID: {record_id} 的資料已成功刪除。")
self.clear_entry_fields()
self.display_all_records()
else:
messagebox.showwarning("刪除失敗", f"未能刪除 ID 為 {record_id} 的記錄。")
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"刪除資料失敗: {e}")
else:
messagebox.showinfo("取消操作", "刪除操作已取消。")
# --- 功能：刪除所有資料 ---
def clear_all_records(self):
if not self.conn:
messagebox.showerror("錯誤", "資料庫未連接。")
return
# (3) 修正: 刪除所有資料時詢問確認
confirm_clear = messagebox.askyesno(
"確認刪除所有資料",
"您確定要刪除資料庫中所有記錄嗎？\n\n"
"此操作將**永久刪除所有資料**且**不可恢復**！\n\n"
"您確定要繼續嗎？"
)
if confirm_clear:
try:
self.cursor.execute(f"DELETE FROM {TABLE_NAME}")
self.cursor.execute(f"DELETE FROM sqlite_sequence WHERE name = '{TABLE_NAME}'")
self.conn.commit()
messagebox.showinfo("刪除所有資料", "所有資料已清除，ID 序列已重置。")
self.display_all_records()
self.clear_entry_fields()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"清除所有資料失敗: {e}")
else:
messagebox.showinfo("取消操作", "刪除所有資料操作已取消。")
# --- 功能：查詢一筆 ---
def query_record(self):
# (1) 修正: 查詢一筆時顯示 ID, UID, Date, Time 四個欄位到輸入框
self.fill_fields_by_id()
if self.id_entry.get() and self.uid_entry.get():
messagebox.showinfo("查詢結果", f"ID: {self.id_entry.get()} 的資料已顯示在輸入框中。")
elif self.id_entry.get() and not self.uid_entry.get():
messagebox.showwarning("找不到資料", f"找不到 ID 為 {self.id_entry.get()} 的記錄。")
# --- 功能：顯示所有資料 ---
def display_all_records(self):
if not self.conn:
self.records_text.config(state=tk.NORMAL)
self.records_text.delete(1.0, tk.END)
self.records_text.insert(tk.END, "資料庫未連接或無資料。\n")
self.records_text.config(state=tk.DISABLED)
return
try:
self.cursor.execute(f"SELECT * FROM {TABLE_NAME} ORDER BY id DESC")
records = self.cursor.fetchall()
self.records_text.config(state=tk.NORMAL)
self.records_text.delete(1.0, tk.END)
if records:
self.records_text.insert(tk.END, "ID\tUID \tDate \tTime\n")
self.records_text.insert(tk.END, "-------------------------------------------------------\n")
for record in records:
uid_display = str(record[1]).ljust(12)
date_display = str(record[2]).ljust(12)
self.records_text.insert(tk.END, f"{record[0]}\t{uid_display}\t{date_display}\t{record[3]}\n")
else:
self.records_text.insert(tk.END, "目前資料庫中沒有記錄。\n")
self.records_text.config(state=tk.DISABLED)
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"顯示所有資料失敗: {e}")
self.records_text.config(state=tk.NORMAL)
self.records_text.delete(1.0, tk.END)
self.records_text.insert(tk.END, f"載入資料失敗: {e}\n")
self.records_text.config(state=tk.DISABLED)
def create_widgets(self):
input_frame = tk.Frame(self.master, padx=10, pady=10)
input_frame.pack(fill=tk.X)
tk.Label(input_frame, text="ID:").grid(row=0, column=0, padx=5, pady=5, sticky="w")
self.id_entry = tk.Entry(input_frame, width=30)
self.id_entry.grid(row=0, column=1, padx=5, pady=5)
# 綁定 ID 欄位，實時填充或在按 Enter 時填充
self.id_entry.bind("<Return>", self.fill_fields_by_id)
self.id_entry.bind("<KeyRelease>", self.fill_fields_by_id) # 為了更即時的顯示
tk.Label(input_frame, text="UID:").grid(row=1, column=0, padx=5, pady=5, sticky="w")
self.uid_entry = tk.Entry(input_frame, width=30)
self.uid_entry.grid(row=1, column=1, padx=5, pady=5)
# 新增 Date 和 Time 欄位，用於顯示
tk.Label(input_frame, text="Date:").grid(row=2, column=0, padx=5, pady=5, sticky="w")
self.date_entry = tk.Entry(input_frame, width=30, state='readonly') # 預設為只讀
self.date_entry.grid(row=2, column=1, padx=5, pady=5)
tk.Label(input_frame, text="Time:").grid(row=3, column=0, padx=5, pady=5, sticky="w")
self.time_entry = tk.Entry(input_frame, width=30, state='readonly') # 預設為只讀
self.time_entry.grid(row=3, column=1, padx=5, pady=5)
# (4) 修正: 新增 MQTT 連線狀態標籤
self.mqtt_status_label = tk.Label(input_frame, text="MQTT 狀態: 初始化中...", fg="gray", font=("Arial", 10, "bold"))
self.mqtt_status_label.grid(row=0, column=2, padx=10, pady=5, sticky="e")
button_frame = tk.Frame(self.master, padx=10, pady=10)
button_frame.pack(fill=tk.X)
tk.Button(button_frame, text="建立資料庫", command=self.create_table).grid(row=0, column=0, padx=5, pady=5)
tk.Button(button_frame, text="新增一筆 (手動)", command=lambda: self.insert_record(self.uid_entry.get())).grid(row=0, column=1, padx=5, pady=5)
tk.Button(button_frame, text="更正一筆", command=self.update_record).grid(row=0, column=2, padx=5, pady=5)
tk.Button(button_frame, text="刪除一筆", command=self.delete_record).grid(row=0, column=3, padx=5, pady=5)
tk.Button(button_frame, text="刪除所有資料", command=self.clear_all_records).grid(row=0, column=4, padx=5, pady=5)
tk.Button(button_frame, text="查詢一筆", command=self.query_record).grid(row=0, column=5, padx=5, pady=5)
tk.Button(button_frame, text="顯示所有", command=self.display_all_records).grid(row=0, column=6, padx=5, pady=5)
tk.Button(button_frame, text="清空欄位", command=self.clear_entry_fields).grid(row=1, column=0, padx=5, pady=5) # 新增清空欄位按鈕
self.records_text = tk.Text(self.master, wrap=tk.WORD, height=20, width=80, bg="lightyellow")
self.records_text.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)
self.records_text.config(state=tk.DISABLED)
# --- MQTT 相關方法 (包含狀態顯示、閃爍與自動重連) ---
def start_mqtt_reconnect_timer(self):
"""啟動定時器，每隔一段時間嘗試重新連接 MQTT Broker。"""
if self.mqtt_reconnect_job:
self.master.after_cancel(self.mqtt_reconnect_job)
if self.master.winfo_exists(): # 避免在 UI 銷毀後調用 after
self.mqtt_reconnect_job = self.master.after(
MQTT_RECONNECT_INTERVAL * 1000, # 轉換為毫秒
self.try_reconnect_mqtt
)
def stop_mqtt_reconnect_timer(self):
"""停止 MQTT 重連定時器。"""
if self.mqtt_reconnect_job:
self.master.after_cancel(self.mqtt_reconnect_job)
self.mqtt_reconnect_job = None
def start_mqtt_blink_animation(self):
"""啟動 MQTT 狀態標籤的閃爍動畫。"""
if self.mqtt_blink_job:
self.master.after_cancel(self.mqtt_blink_job)
self.mqtt_blink_state = True # 確保從顯示文字開始閃爍
if self.master.winfo_exists(): # 避免在 UI 銷毀後調用 after
self.do_mqtt_blink()
def stop_mqtt_blink_animation(self):
"""停止 MQTT 狀態標籤的閃爍動畫。"""
if self.mqtt_blink_job:
self.master.after_cancel(self.mqtt_blink_job)
self.mqtt_blink_job = None
if self.master.winfo_exists(): # 確保 UI 元件存在
self.mqtt_status_label.config(text="MQTT 狀態: 已連線", fg="green")
def do_mqtt_blink(self):
"""執行 MQTT 狀態標籤的單次閃爍。"""
# 只有在未連線且 UI 存在時才閃爍
if not self.mqtt_client_connected and self.master.winfo_exists():
if self.mqtt_blink_state:
self.mqtt_status_label.config(text="MQTT 狀態: 未連線", fg="red")
else:
self.mqtt_status_label.config(text="", fg="red") # 顯示空白
self.mqtt_blink_state = not self.mqtt_blink_state
self.mqtt_blink_job = self.master.after(MQTT_BLINK_INTERVAL_MS, self.do_mqtt_blink)
elif self.mqtt_client_connected: # 如果在閃爍過程中連上了，則停止閃爍
self.stop_mqtt_blink_animation()
def update_mqtt_status_label(self):
"""根據 mqtt_client_connected 狀態更新狀態標籤並控制閃爍/重連。"""
if not self.master.winfo_exists(): # 如果 UI 已銷毀，則不執行更新
return
if self.mqtt_client_connected:
self.stop_mqtt_blink_animation() # 停止閃爍
self.stop_mqtt_reconnect_timer() # 停止重連計時器
self.mqtt_status_label.config(text="MQTT 狀態: 已連線", fg="green")
else:
self.start_mqtt_blink_animation() # 啟動閃爍
self.start_mqtt_reconnect_timer() # 啟動重連計時器
def try_reconnect_mqtt(self):
"""嘗試重新連接 MQTT Broker。這個函數由定時器呼叫。"""
if not self.mqtt_client_connected: # 只有在未連線時才嘗試重連
print(f"嘗試重新連接 MQTT Broker: {MQTT_BROKER}...")
try:
self.mqtt_client.reconnect() # 使用 reconnect 方法
# reconnect() 是非阻塞的，結果會在 on_connect/on_disconnect 中處理
except Exception as e:
print(f"嘗試重新連接 MQTT Broker 時發生錯誤: {e}")
# 不論成功或失敗，如果目前仍未連線，都重新安排下一次嘗試
# 這個調用必須在主執行緒中
self.master.after(0, self.start_mqtt_reconnect_timer)
def connect_mqtt_async(self):
"""首次連接 MQTT Broker (非同步)。"""
try:
self.mqtt_client.connect_async(MQTT_BROKER, 1883, 60) # 使用 connect_async
except Exception as e:
print(f"首次連接到 MQTT Broker 失敗: {e}")
self.mqtt_client_connected = False
# 立即更新狀態，啟動閃爍和重連
self.master.after(0, self.update_mqtt_status_label)
def mqtt_loop_forever(self):
try:
self.mqtt_client.loop_forever()
except Exception as e:
print(f"MQTT loop error: {e}")
# 當 loop 終止時，也應該將狀態設為未連線
self.mqtt_client_connected = False
# 立即更新狀態，啟動閃爍和重連
self.master.after(0, self.update_mqtt_status_label)
def on_connect(self, client, userdata, flags, rc):
if rc == 0:
print("已連接到 MQTT Broker!")
self.mqtt_client_connected = True
# 同時訂閱兩個主題
client.subscribe([(MQTT_TOPIC_RFID, 0), (MQTT_TOPIC_AUTO_ADD, 0)])
print(f"已訂閱主題: {MQTT_TOPIC_RFID} 和 {MQTT_TOPIC_AUTO_ADD}")
else:
print(f"MQTT 連接失敗，返回碼: {rc}")
self.mqtt_client_connected = False
# 在主線程中更新 UI
self.master.after(0, self.update_mqtt_status_label)
def on_disconnect(self, client, userdata, rc):
"""MQTT 客戶端斷線回呼。"""
print(f"MQTT 已斷開連接，返回碼: {rc}")
self.mqtt_client_connected = False
# 在主線程中更新 UI 狀態並啟動閃爍/重連
self.master.after(0, self.update_mqtt_status_label)
def on_message(self, client, userdata, msg):
topic = msg.topic
payload = msg.payload.decode("utf-8").strip()
print(f"收到訊息 - Topic: {topic}, Payload: '{payload}'")
if topic == MQTT_TOPIC_RFID:
print(f"收到 RFID UID: {payload}")
self.master.after(0, self.insert_record, payload)
elif topic == MQTT_TOPIC_AUTO_ADD:
if payload.lower() == "錯誤輸入":
random_uid = self.generate_random_uid()
print(f"收到 '{MQTT_TOPIC_AUTO_ADD}' 錯誤輸入訊息，自動生成並插入 UID: {random_uid}")
self.master.after(0, self.insert_record, random_uid)
else:
print(f"收到 '{MQTT_TOPIC_AUTO_ADD}' 非錯誤輸入訊息 '{payload}'，不執行自動新增。")
def on_closing(self):
if messagebox.askokcancel("退出", "您確定要退出應用程式嗎？"):
# 停止所有定時任務
self.stop_mqtt_blink_animation()
self.stop_mqtt_reconnect_timer()
if self.conn:
self.conn.close()
print("資料庫連接已關閉。")
# 確保在斷開前停止 loop
self.mqtt_client.loop_stop()
if self.mqtt_client.is_connected(): # 只有在連線時才嘗試斷開
self.mqtt_client.disconnect()
print("MQTT 連接已斷開。")
else:
print("MQTT loop 已停止。")
self.master.destroy()
# --- 程式主入口 ---
if __name__ == "__main__":
root = tk.Tk()
app = RFIDApp(root)
root.protocol("WM_DELETE_WINDOW", app.on_closing)
root.mainloop()
沒有留言:
張貼留言