DB_NAME = "rfid114.db"
MQTT_BROKER = "broker.hivemq.com"
MQTT_PORT = 1883
TOPIC_RFID_UID = "alex9ufo/rfid/UID"
TOPIC_LED_CONTROL = "alex9ufo/rfid/led"
TOPIC_LED_STATUS = "alex9ufo/rfid/ledStatus"
這個程式是一個綜合性的 RFID 讀取、MQTT 訊息發送/接收、SQLite 資料庫記錄和 Tkinter GUI 介面的應用程式，同時整合了 Telegram 機器人控制功能。
import tkinter as tk
from tkinter import messagebox
import paho.mqtt.client as mqtt
import threading
import time
import asyncio
import sqlite3
from datetime import datetime
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
import sys
# --- 1. 全域設定與常數 ---
DB_NAME = "rfid114.db"
MQTT_BROKER = "broker.hivemq.com"
MQTT_PORT = 1883
TOPIC_RFID_UID = "alex9ufo/rfid/UID"
TOPIC_LED_CONTROL = "alex9ufo/rfid/led"
TOPIC_LED_STATUS = "alex9ufo/rfid/ledStatus"
# Telegram 設定 (已使用您提供的 Token 和 Chat ID)
TELEGRAM_BOT_TOKEN = "8021272010986:AAGymymK9_d1HcT1GJWl3mtqHmilxB64_5Zw"
TARGET_CHAT_ID = 791652148469
# 應用程式模式
MODE_ADD = "新增模式"
MODE_COMPARE = "比對模式"
# --- 2. 資料庫操作類別 (修正為 5 欄位: id, date, time, LEDorRFID, memo) ---
class DatabaseManager:
def __init__(self, db_name):
self.db_name = db_name
self.conn = None
self.cursor = None
def connect(self):
self.conn = sqlite3.connect(self.db_name)
self.cursor = self.conn.cursor()
def close(self):
if self.conn:
self.conn.close()
# a. 建立資料表
def create_table(self):
self.connect()
try:
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS rfid_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT NOT NULL,
time TEXT NOT NULL,
LEDorRFID TEXT,
memo TEXT
)
""")
self.conn.commit()
return True, "資料表 rfid_logs 建立成功或已存在 (5 欄位)。"
except Exception as e:
return False, f"建立資料表失敗: {e}"
finally:
self.close()
# f. 刪除資料表
def delete_table(self):
self.connect()
try:
self.cursor.execute("DROP TABLE IF EXISTS rfid_logs")
self.conn.commit()
return True, "資料表 rfid_logs 刪除成功。"
except Exception as e:
return False, f"刪除資料表失敗: {e}"
finally:
self.close()
# b. 新增一筆資料
def add_log(self, led_or_rfid, memo):
self.connect()
try:
now = datetime.now()
date_str = now.strftime("%Y-%m-%d")
time_str = now.strftime("%H:%M:%S")
self.cursor.execute(
"INSERT INTO rfid_logs (date, time, LEDorRFID, memo) VALUES (?, ?, ?, ?)",
(date_str, time_str, led_or_rfid, memo)
)
self.conn.commit()
last_id = self.cursor.lastrowid
return True, f"新增記錄成功，ID: {last_id}"
except Exception as e:
return False, f"新增記錄失敗: {e}"
finally:
self.close()
# c. 刪除一筆資料
def delete_log_by_id(self, log_id):
self.connect()
try:
self.cursor.execute("DELETE FROM rfid_logs WHERE id=?", (log_id,))
rows_affected = self.cursor.rowcount
self.conn.commit()
if rows_affected > 0:
return True, f"ID {log_id} 的記錄刪除成功。"
else:
return False, f"未找到 ID {log_id} 的記錄。"
except Exception as e:
return False, f"刪除記錄失敗: {e}"
finally:
self.close()
# d. 查詢一筆資料
def query_log_by_id(self, log_id):
self.connect()
try:
self.cursor.execute("SELECT id, date, time, LEDorRFID, memo FROM rfid_logs WHERE id=?", (log_id,))
result = self.cursor.fetchone()
return True, result
except Exception as e:
return False, f"查詢記錄失敗: {e}"
finally:
self.close()
# e. 顯示所有資料最新50筆
def get_latest_logs(self):
self.connect()
try:
self.cursor.execute("SELECT id, date, time, LEDorRFID, memo FROM rfid_logs ORDER BY id DESC LIMIT 50")
results = self.cursor.fetchall()
return True, results
except Exception as e:
return False, f"查詢最新記錄失敗: {e}"
finally:
self.close()
# g. 獲取所有儲存的 UID (僅用於比對模式)
def get_all_rfid_uids(self):
self.connect()
try:
# 查詢所有 memo 欄位符合 '[新增] 卡號號碼:%' 的記錄的 LEDorRFID 欄位值
self.cursor.execute("SELECT DISTINCT LEDorRFID FROM rfid_logs WHERE memo LIKE '[新增] 卡號號碼:%'")
# 獲取結果並轉換為 UID 列表 (確保不是 None)
results = [row[0] for row in self.cursor.fetchall() if row[0] is not None]
return True, results
except Exception as e:
return False, f"獲取所有 UID 失敗: {e}"
finally:
self.close()
# --- 3. Telegram Bot 處理類別 ---
class TelegramBotHandler:
def __init__(self, app, token):
self.app = app
self.token = token
self.tg_loop = None
if token == "YOUR_TELEGRAM_BOT_TOKEN":
print("警告: Telegram Bot Token 未設定，Bot 功能將被跳過。")
self.application = None
return
self.application = Application.builder().token(token).build()
self.application.add_handler(CommandHandler("on", self.handle_command))
self.application.add_handler(CommandHandler("off", self.handle_command))
self.application.add_handler(CommandHandler("flash", self.handle_command))
self.application.add_handler(CommandHandler("timer", self.handle_command))
self.application.add_handler(CommandHandler("status", self.handle_status))
self.application.add_handler(CommandHandler("help", self.handle_help))
self.application.add_handler(CommandHandler("mode", self.handle_mode_switch))
self.bot_thread = threading.Thread(target=self._run_bot, daemon=True)
def start_bot(self):
if self.application:
self.bot_thread.start()
print("Telegram Bot 已啟動")
def _run_bot(self):
if self.application:
try:
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self.tg_loop = loop
# 運行輪詢，如果出現 Conflict 錯誤，請檢查是否有其他實例運行
self.application.run_polling(poll_interval=0.5, timeout=10)
except Exception as e:
print(f"Telegram Bot 運行錯誤: {e}")
async def handle_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message.chat_id != TARGET_CHAT_ID:
await update.message.reply_text("未授權的用戶。")
return
command = update.message.text[1:]
self.app.master.after(0, self.app.handle_telegram_command, command)
async def handle_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message.chat_id != TARGET_CHAT_ID:
await update.message.reply_text("未授權的用戶。")
return
self.app.master.after(0, self.app.show_logs_in_telegram)
async def handle_mode_switch(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message.chat_id != TARGET_CHAT_ID:
await update.message.reply_text("未授權的用戶。")
return
current_mode = self.app.mode_var.get()
new_mode = MODE_COMPARE if current_mode == MODE_ADD else MODE_ADD
self.app.master.after(0, self.app.set_mode, new_mode)
await update.message.reply_text(f"模式已切換為: {new_mode}")
async def handle_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message.chat_id != TARGET_CHAT_ID:
await update.message.reply_text("未授權的用戶。")
return
await update.message.reply_text(self._get_help_message())
def _get_help_message(self):
"""4. 啟動時發送 Help 訊息給 Telegram"""
return (
"🎉 程式已啟動！\n\n"
"可用命令:\n"
"/on - 開啟 LED (綠色)\n"
"/off - 關閉 LED (紅色)\n"
"/flash - LED 閃爍 (紅綠交替)\n"
"/timer - LED 開啟 10 秒後自動關閉\n"
"/mode - 切換模式 (新增/比對)\n"
"/status - 顯示當前模式及最新記錄"
)
def send_message(self, text):
"""通用發送訊息函數，用於 RFID 結果、LED 狀態及啟動訊息"""
if self.application and self.tg_loop and self.application.running and TARGET_CHAT_ID != 0:
asyncio.run_coroutine_threadsafe(
self.application.bot.send_message(chat_id=TARGET_CHAT_ID, text=text),
self.tg_loop
)
# --- 4. Tkinter 主應用程式類別 ---
class RfidControlApp:
def __init__(self, master):
self.master = master
master.title("RFID & MQTT 控制台")
self.db = DatabaseManager(DB_NAME)
self.status_var = tk.StringVar(value="等待 MQTT 連線...")
self.mode_var = tk.StringVar(value=MODE_ADD)
self.flash_id = None
self.timer_id = None
self.timer_end_time = None
self._setup_gui()
# *** 修正點 1/2: 確保使用最新的 VERSION2 客戶端 ***
self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
self.client.on_connect = self._on_connect
self.client.on_message = self._on_message
self.client.user_data_set(self)
self.telegram_bot = TelegramBotHandler(self, TELEGRAM_BOT_TOKEN)
if self.telegram_bot.application:
self.telegram_bot.start_bot()
# 延遲發送 Help 訊息，確保 Bot 啟動完成
self.master.after(3000, self._send_initial_help)
self.mqtt_thread = threading.Thread(target=self._mqtt_loop_start, daemon=True)
self.mqtt_thread.start()
master.protocol("WM_DELETE_WINDOW", self.on_closing)
def _setup_gui(self):
# 頂部狀態及模式
mode_frame = tk.Frame(self.master)
mode_frame.pack(pady=10)
# 放大模式標籤字體
tk.Label(mode_frame, text="當前模式:", font=("Arial", 16)).pack(side=tk.LEFT, padx=10)
tk.Label(mode_frame, textvariable=self.mode_var, font=("Arial", 16, "bold")).pack(side=tk.LEFT, padx=10)
# 放大切換模式按鈕
tk.Button(mode_frame, text="切換模式", command=self._toggle_mode, font=("Arial", 12), padx=10, pady=5).pack(side=tk.LEFT, padx=20)
# 圓形 LED 指示燈
self.canvas = tk.Canvas(self.master, width=120, height=120)
self.canvas.pack(pady=15)
self.led_indicator = self.canvas.create_oval(10, 10, 110, 110, fill="gray")
# 狀態顯示 (卡號/LED 動作) - 放大字體
tk.Label(self.master, textvariable=self.status_var, font=("Arial", 16, "bold")).pack(pady=15)
# 資料庫操作按鈕
db_frame = tk.Frame(self.master)
db_frame.pack(pady=10)
# 放大按鈕字體
button_font = ("Arial", 12)
button_padding = {'padx': 8, 'pady': 5}
tk.Button(db_frame, text="a.建表", command=self._handle_db_create, font=button_font, **button_padding).pack(side=tk.LEFT, padx=5)
tk.Button(db_frame, text="e.最新50筆", command=self._handle_db_show, font=button_font, **button_padding).pack(side=tk.LEFT, padx=5)
tk.Button(db_frame, text="f.刪表", command=self._handle_db_delete, font=button_font, **button_padding).pack(side=tk.LEFT, padx=5)
# 查詢/刪除 ID 輸入框及按鈕
query_frame = tk.Frame(self.master)
query_frame.pack(pady=5)
self.id_entry = tk.Entry(query_frame, width=15, font=("Arial", 12))
self.id_entry.pack(side=tk.LEFT, padx=5)
tk.Button(query_frame, text="c.刪除ID", command=self._handle_db_delete_id, font=button_font, **button_padding).pack(side=tk.LEFT, padx=5)
tk.Button(query_frame, text="d.查詢ID", command=self._handle_db_query_id, font=button_font, **button_padding).pack(side=tk.LEFT, padx=5)
# 資料顯示區域標題
tk.Label(self.master, text="--- 最新記錄 ---", font=("Arial", 12, "underline")).pack(pady=10)
# 資料顯示 Text Widget - 增加寬度，使用等寬字體
self.log_display = tk.Text(self.master, height=10, width=80, font=("Courier New", 10))
self.log_display.pack(pady=5, padx=10)
self._update_log_display("等待資料庫查詢...")
def _update_log_display(self, text):
"""用於在主執行緒中更新資料顯示區域 (Text Widget) 的內容"""
self.log_display.delete(1.0, tk.END)
self.log_display.insert(tk.END, text)
def _show_mqtt_error_gui(self, error_message):
"""用於在主執行緒中顯示 MQTT 連線錯誤訊息"""
# 注意: 這裡會顯示錯誤，但真正的連線錯誤是函式簽名不匹配
messagebox.showerror("MQTT 錯誤", f"無法連接到 MQTT Broker: {error_message}")
def on_closing(self):
"""應用程式關閉時的清理工作"""
print("正在關閉應用程式...")
self._cancel_timers()
if hasattr(self, 'telegram_bot') and self.telegram_bot.application:
try:
# 嘗試停止 Telegram 應用程式
self.telegram_bot.application.stop()
print("Telegram Bot 已停止")
except Exception:
pass
if hasattr(self, 'client') and self.client:
self.client.loop_stop()
self.master.destroy()
sys.exit()
# --- 資料庫操作實現 (1.a, c, d, e, f) ---
def _handle_db_create(self):
success, msg = self.db.create_table()
messagebox.showinfo("資料庫", msg)
self._update_log_display(msg)
def _handle_db_delete(self):
if messagebox.askyesno("確認刪除", "確定要刪除整個資料表嗎？資料將無法恢復！"):
success, msg = self.db.delete_table()
messagebox.showinfo("資料庫", msg)
self._update_log_display(msg)
def _handle_db_delete_id(self):
log_id = self.id_entry.get()
if not log_id.isdigit():
messagebox.showerror("錯誤", "請輸入有效的 ID 數字。")
return
success, msg = self.db.delete_log_by_id(int(log_id))
messagebox.showinfo("資料庫", msg)
self._handle_db_show()
def _handle_db_query_id(self):
log_id = self.id_entry.get()
if not log_id.isdigit():
messagebox.showerror("錯誤", "請輸入有效的 ID 數字。")
return
success, result = self.db.query_log_by_id(int(log_id))
if success and result:
msg = (f"查詢結果 (ID {result[0]}):\n"
f"日期: {result[1]}\n"
f"時間: {result[2]}\n"
f"LED/RFID: {result[3]}\n"
f"備註: {result[4]}")
elif success:
msg = f"未找到 ID {log_id} 的記錄。"
else:
msg = f"查詢失敗: {result}"
messagebox.showinfo("查詢結果", msg)
self._update_log_display(msg)
def _handle_db_show(self):
"""從資料庫讀取最新記錄並顯示在 Text Widget (調整格式以顯示 5 欄位)"""
success, results = self.db.get_latest_logs()
if success:
# 調整標題欄位寬度以適應 5 欄位
# 欄位: ID | 日期 | 時間 | LED/RFID | 備註
# 注意: 您的截圖中只有 ID, 日期, 時間, 備註，這裡保持 5 欄位結構
header = f"{'ID':<4} | {'日期':<10} | {'時間':<8} | {'LED/RFID':<15} | 備註\n"
separator = "-"*75 + "\n"
output = header + separator
# 調整資料欄位寬度
for row in results:
# 限制 memo 長度
memo_display = row[4][:30]
output += (
f"{str(row[0]):<4} | " # ID
f"{row[1]:<10} | " # 日期
f"{row[2]:<8} | " # 時間
f"{row[3]:<15} | " # LEDorRFID
f"{memo_display}\n" # 備註
)
self._update_log_display(output)
else:
self._update_log_display(f"顯示資料失敗: {results}")
def show_logs_in_telegram(self):
"""將模式和最新記錄發送到 Telegram (/status 命令)"""
mode_info = f"當前模式: {self.mode_var.get()}\n"
success, results = self.db.get_latest_logs()
if success and results:
# 僅顯示最新的 5 筆記錄，顯示 LEDorRFID 和 memo
log_list = [f"ID {row[0]} ({row[3]}): {row[4][:30]}" for row in results[:5]]
log_info = "--- 最新 5 筆記錄 ---\n" + "\n".join(log_list)
elif success:
log_info = "資料庫中沒有記錄。"
else:
log_info = f"資料庫查詢失敗: {results}"
self.telegram_bot.send_message(f"{mode_info}\n{log_info}")
def _toggle_mode(self):
current_mode = self.mode_var.get()
new_mode = MODE_COMPARE if current_mode == MODE_ADD else MODE_ADD
self.set_mode(new_mode)
def set_mode(self, new_mode):
self.mode_var.set(new_mode)
self.status_var.set(f"模式切換為: {new_mode}")
# --- MQTT 相關 ---
def _mqtt_loop_start(self):
"""在單獨的執行緒中啟動 MQTT 客戶端迴圈"""
try:
self.client.connect(MQTT_BROKER, MQTT_PORT, 60)
self.client.loop_forever()
except Exception as error:
error_message = str(error)
print(f"MQTT 連線錯誤: {error_message}")
self.master.after(0, self._show_mqtt_error_gui, error_message)
# *** 修正點 2/2: 將 _on_connect 簽名更新為 V2 格式 (6 參數) ***
def _on_connect(self, client, userdata, flags, reasonCode, properties):
if reasonCode == 0: # reasonCode 是 V2 中取代 rc 的連線結果
client.subscribe(TOPIC_RFID_UID)
client.subscribe(TOPIC_LED_STATUS)
self.master.after(0, self.status_var.set, f"MQTT 連線成功 | {self.mode_var.get()}")
self.master.after(0, self._handle_db_show)
else:
# 使用 reasonCode 檢查連線失敗原因
self.master.after(0, self.status_var.set, f"MQTT 連線失敗, 碼 {reasonCode}")
def _on_message(self, client, userdata, msg):
payload = msg.payload.decode()
if msg.topic == TOPIC_LED_STATUS:
self._handle_led_status_update(payload)
return
if msg.topic == TOPIC_RFID_UID:
self._handle_rfid_uid(payload)
def _handle_rfid_uid(self, uid):
"""7) 處理 RFID 卡號 & 8) 比對模式邏輯"""
mode = self.mode_var.get()
if mode == MODE_ADD:
# 7) 新增模式: 寫入資料庫
led_or_rfid_value = uid
memo_value = f"[新增] 卡號號碼: {uid}"
self.master.after(0, self.db.add_log, led_or_rfid_value, memo_value)
self.master.after(0, self.status_var.set, memo_value)
self.master.after(0, self._handle_db_show)
self.telegram_bot.send_message(memo_value)
elif mode == MODE_COMPARE:
# 8) 比對模式: 與資料庫的 LEDorRFID 欄位中所有 UID 碼比對
success, authorized_uids = self.db.get_all_rfid_uids()
if success:
if uid in authorized_uids:
memo = "[比對] 卡號正確"
# 比對成功時，Telegram 訊息包含 UID 碼
telegram_msg = f"{memo} ({uid})"
else:
memo = "[比對] 卡號錯誤"
# 比對失敗時，Telegram 訊息不包含 UID 碼
telegram_msg = memo
else:
memo = f"[比對] 錯誤: 無法讀取資料庫 ({authorized_uids})"
telegram_msg = memo
# GUI/Telegram 操作
self.master.after(0, self.status_var.set, f"{memo} (UID: {uid})")
self.telegram_bot.send_message(telegram_msg)
def _handle_led_status_update(self, status):
"""6) 處理來自 ESP32 的 LED 狀態回傳 (修正欄位值)"""
# 6) 欄位修正
status_upper = status.upper()
led_or_rfid_value = status_upper
memo_value = f"LED Status: {status_upper}"
telegram_msg = f"裝置確認 LED 狀態: {status_upper}"
# 寫入資料庫, 刷新 GUI, 送給 Telegram
self.master.after(0, self.db.add_log, led_or_rfid_value, memo_value)
self.master.after(0, self._handle_db_show)
self.telegram_bot.send_message(telegram_msg)
# --- Telegram 命令處理 ---
def _send_initial_help(self):
self.telegram_bot.send_message(self.telegram_bot._get_help_message())
def handle_telegram_command(self, command):
command = command.lower()
if command in ["on", "off", "flash", "timer"]:
self._set_led_indicator(command)
self.client.publish(TOPIC_LED_CONTROL, command, qos=0)
self.status_var.set(f"Telegram 命令: {command.upper()}")
self.telegram_bot.send_message(f"已發送命令 /{command} 到裝置。等待裝置確認...")
else:
self.telegram_bot.send_message(f"無效命令: /{command}")
# --- LED GUI 邏輯 ---
def _cancel_timers(self):
if self.flash_id:
self.master.after_cancel(self.flash_id)
self.flash_id = None
if self.timer_id:
self.master.after_cancel(self.timer_id)
self.timer_id = None
self.timer_end_time = None
def _set_led_indicator(self, state):
self._cancel_timers()
if state == "on":
self.canvas.itemconfig(self.led_indicator, fill="green")
elif state == "off":
self.canvas.itemconfig(self.led_indicator, fill="red")
elif state == "flash":
self._flash_toggle(True)
elif state == "timer":
self.timer_end_time = time.time() + 10
self.canvas.itemconfig(self.led_indicator, fill="green")
self._timer_countdown()
else:
self.canvas.itemconfig(self.led_indicator, fill="gray")
def _flash_toggle(self, is_green):
color = "green" if is_green else "red"
self.canvas.itemconfig(self.led_indicator, fill=color)
self.flash_id = self.master.after(500, self._flash_toggle, not is_green)
def _timer_countdown(self):
if self.timer_end_time is None:
return
if time.time() >= self.timer_end_time:
self.canvas.itemconfig(self.led_indicator, fill="red")
self.timer_id = None
return
self.timer_id = self.master.after(100, self._timer_countdown)
# --- 主程式 ---
if __name__ == '__main__':
if sys.platform == "win32":
try:
# 解決 Windows 上的 asyncio RuntimeError: Event loop is closed
if threading.current_thread() is threading.main_thread():
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
except AttributeError:
pass
root = tk.Tk()
app = RfidControlApp(root)
root.mainloop()
程式碼結構與內容說明
1. 全域設定與常數（Section 1）
| 常數 | 內容 | 說明 |
DB_NAME | "rfid114.db" | SQLite 資料庫檔案名稱。 |
MQTT_BROKER | "broker.hivemq.com" | 您設定的 MQTT 伺服器位址。 |
MQTT_PORT | 1883 | MQTT 標準端口。 |
TOPIC_RFID_UID | "alex9ufo/rfid/UID" | 接收 RFID 卡號的 MQTT 主題。 |
TOPIC_LED_CONTROL | "alex9ufo/rfid/led" | 用於發送 LED 控制命令的 MQTT 主題。 |
TOPIC_LED_STATUS | "alex9ufo/rfid/ledStatus" | 接收 LED 狀態回傳的 MQTT 主題。 |
TELEGRAM_BOT_TOKEN | (您的 Token) | Telegram 機器人的授權 Token。 |
TARGET_CHAT_ID | (您的 Chat ID) | 機器人只回應這個 ID 的使用者。 |
MODE_ADD | "新增模式" | 應用程式模式：收到卡號時，將其新增到資料庫。 |
MODE_COMPARE | "比對模式" | 應用程式模式：收到卡號時，與資料庫比對是否授權。 |
2. DatabaseManager 類別 (資料庫操作)
這個類別負責所有與 rfid114.db 檔案的互動，使用了 SQLite3 模組。
| 方法 | 功能描述 |
connect() / close() | 建立和關閉與資料庫的連線。 |
create_table() | (a) 建立 rfid_logs 資料表（包含 id, date, time, LEDorRFID, memo 五個欄位）。 |
delete_table() | (f) 刪除整個 rfid_logs 資料表。 |
add_log(led_or_rfid, memo) | (b) 新增一筆記錄（用於記錄 LED 狀態或新刷入的 RFID 卡號）。 |
delete_log_by_id(log_id) | (c) 依據 ID 刪除一筆記錄。 |
query_log_by_id(log_id) | (d) 依據 ID 查詢一筆記錄。 |
get_latest_logs() | (e) 查詢並回傳最新的 50 筆記錄。 |
get_all_rfid_uids() | (g) 關鍵比對來源：查詢所有 memo 中有 [新增] 卡號號碼 標記的記錄，並回傳它們的 LEDorRFID 值（即所有授權 UID 列表），供比對模式使用。 |
3. TelegramBotHandler 類別 (Telegram 整合)
這個類別負責初始化 Telegram Bot 並在獨立執行緒中運行，處理來自 Telegram 的命令，並負責向使用者發送通知。
| 命令/方法 | 功能描述 |
/on, /off, /flash, /timer | 接收命令後，轉發給主應用程式，由主程式透過 MQTT 控制 LED。 |
/mode | 切換主應用程式的運行模式（新增/比對）。 |
/status | 讓主應用程式查詢當前模式和最新的資料庫記錄，並將結果回傳給 Telegram。 |
send_message(text) | 應用程式內部呼叫此方法，向 TARGET_CHAT_ID 發送文字訊息（用於 RFID 刷卡結果、LED 狀態等）。 |
4. RfidControlApp 類別 (主應用程式與 GUI)
這是整個系統的核心，負責 Tkinter 介面、MQTT 連線管理和所有業務邏輯。
核心功能：
初始化 (
__init__):設定 Tkinter 介面。
初始化
DatabaseManager和TelegramBotHandler。啟動 MQTT 連線到
broker.hivemq.com(_mqtt_loop_start)。
GUI 介面 (
_setup_gui):顯示當前模式 (
mode_var)。LED 圓形指示燈 (
canvas)，會根據 LED 命令或狀態變更顏色。狀態顯示列 (
status_var)，顯示即時操作結果。資料庫操作按鈕（建表、刪表、查詢、刪除 ID）。
記錄顯示區 (
log_display)，顯示最新的資料庫記錄。
MQTT 處理 (
_on_connect,_on_message):連線成功後訂閱
TOPIC_RFID_UID和TOPIC_LED_STATUS。收到訊息後，依據主題分別呼叫對應的處理函數。
關鍵業務邏輯 (_handle_rfid_uid)
此函數處理從 MQTT 主題 alex9ufo/rfid/UID 收到的卡號 (uid)，並根據當前模式執行操作：
| 模式 | 螢幕顯示 (status_var) | Telegram 訊息 (telegram_msg) | 資料庫操作 |
新增模式 (MODE_ADD) | [新增] 卡號號碼: XXXXXXXX | [新增] 卡號號碼: XXXXXXXX | 新增記錄，LEDorRFID 欄位為 uid。 |
比對模式 (MODE_COMPARE) | [比對] 卡號正確 (UID: XXXXXXXX) | [比對] 卡號正確 (XXXXXXXX) | 比對成功： 無資料庫寫入。 |
比對模式 (MODE_COMPARE) | [比對] 卡號錯誤 (UID: XXXXXXXX) | [比對] 卡號錯誤 | 比對失敗： 無資料庫寫入。 |
LED 狀態處理 (_handle_led_status_update)
此函數處理從 alex9ufo/rfid/ledStatus 收到的 LED 狀態回傳：
GUI 顯示: 無直接改變，但
status_var會更新。Telegram 訊息: 發送
裝置確認 LED 狀態: [STATUS]。資料庫操作: 新增一筆記錄，
LEDorRFID欄位為狀態（例如ON），memo為LED Status: [STATUS]。
沒有留言:
張貼留言