From 7b41ac054bbbd9b32a7ea64d369fd49a9a9ede6b Mon Sep 17 00:00:00 2001 From: lronetto Date: Thu, 16 Jul 2026 15:04:56 +0000 Subject: [PATCH] Add UHF RFID reader GUI app for Fonkan/JRD-100/M100 modules Tkinter desktop app to connect over serial, configure baudrate/output power, start/stop tag inventory, and display detected EPC tags in a table with RSSI and read count. Co-Authored-By: Claude Sonnet 5 --- .gitignore | 3 + uhf_rfid_gui.py | 557 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 560 insertions(+) create mode 100644 .gitignore create mode 100644 uhf_rfid_gui.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..00f2d38 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.venv/ diff --git a/uhf_rfid_gui.py b/uhf_rfid_gui.py new file mode 100644 index 0000000..74e6339 --- /dev/null +++ b/uhf_rfid_gui.py @@ -0,0 +1,557 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +uhf_rfid_gui.py +================ + +Aplicativo com interface grafica (Tkinter) para Windows, para uso com +modulos leitores UHF RFID da familia "Fonkan / JRD-100 / M100" +(ISO18000-6C / EPC Global Class 1 Gen 2, interface serial TTL232), +como o vendido em: +https://pt.aliexpress.com/item/1005006138510871.html + +Funcionalidades: + - Selecionar porta serial (com botao de atualizar lista) + - Selecionar baudrate + - Definir a potencia de saida (dBm) do leitor + - Conectar / desconectar + - Iniciar / parar varredura (inventario) de tags + - Mostrar as tags detectadas em uma tabela (EPC, RSSI, contagem, ultima leitura) + - Console de log com os quadros brutos (uteis para depuracao, pois o + fabricante nao disponibiliza datasheet oficial publico - o protocolo + abaixo foi reconstruido a partir de documentacao de terceiros/comunidade) + +------------------------------------------------------------------------ +IMPORTANTE SOBRE O PROTOCOLO +------------------------------------------------------------------------ +A Fonkan nao publica um datasheet/PDF oficial completo para este modulo. +O protocolo usado aqui foi reconstruido a partir de: + - Blog "JRD-100 - Arduino UHF-RFID Module Development" (CimpleO Group) + - Discussoes no forum oficial do Arduino sobre o JRD-100 + - Documentacao de modulos da mesma familia de chipset (comuns em varios + modulos "clone" vendidos como YRM100/M100/JRD-100/R200), que usam o + mesmo formato de quadro: + + Quadro (frame): + [0] Header = 0xBB + [1] Type = 0x00 (comando enviado ao leitor) + 0x01 (resposta do leitor) + 0x02 (notificacao / tag lida) + [2] Command = codigo do comando + [3-4] PL Len = tamanho do payload, 2 bytes big-endian + [5..] PL = payload (parametros / dados) + [.] Checksum = soma (Type+Command+PLLen+PL) & 0xFF + [.] End = 0x7E + + Comandos utilizados neste app: + 0x03 Consulta de informacoes do modulo (hw/sw/fabricante) + 0xB6 Definir potencia de saida (RF Output Power), 2 bytes, + valor em centi-dBm (dBm * 100), big-endian + 0xB7 Consultar potencia de saida atual + 0x22 Inventario/poll unico (single polling) + 0x27 Inventario continuo (multiple polling) - payload: + [modo=0x22][contagem 2 bytes big-endian] + 0x28 Parar inventario continuo + + Quadro de notificacao de tag (Type=0x02, Command=0x22): + PL = [RSSI 1 byte][PC 2 bytes][EPC N bytes][CRC 2 bytes] + +Como esses valores nao vieram de um datasheet oficial, pode ser +necessario pequenos ajustes conforme a revisao exata do firmware do seu +modulo. Use o console de log (quadros brutos em hexadecimal) para +conferir o que o leitor esta respondendo e ajustar se necessario. + +------------------------------------------------------------------------ +INSTALACAO (Windows) +------------------------------------------------------------------------ +1. Instale o Python 3 (inclui Tkinter por padrao no instalador oficial). +2. Instale a biblioteca pyserial: + pip install pyserial +3. Execute: + python uhf_rfid_gui.py +""" + +import struct +import threading +import queue +import time +from datetime import datetime + +import tkinter as tk +from tkinter import ttk, messagebox, scrolledtext + +try: + import serial + import serial.tools.list_ports +except ImportError: + serial = None + + +# -------------------------------------------------------------------------- +# Constantes do protocolo +# -------------------------------------------------------------------------- +FRAME_HEADER = 0xBB +FRAME_END = 0x7E + +TYPE_COMMAND = 0x00 +TYPE_RESPONSE = 0x01 +TYPE_NOTIFICATION = 0x02 + +CMD_GET_INFO = 0x03 +INFO_HW_VERSION = 0x00 +INFO_SW_VERSION = 0x01 +INFO_MANUFACTURER = 0x02 + +CMD_SET_POWER = 0xB6 +CMD_GET_POWER = 0xB7 + +CMD_SINGLE_POLL = 0x22 +CMD_MULTI_POLL = 0x27 +CMD_STOP_MULTI_POLL = 0x28 + +BAUDRATES = ["9600", "19200", "38400", "57600", "115200"] +DEFAULT_BAUDRATE = "115200" + + +# -------------------------------------------------------------------------- +# Construcao / interpretacao de quadros +# -------------------------------------------------------------------------- +def build_frame(command, params=b""): + """Monta um quadro de comando (Type = 0x00) pronto para envio.""" + pl_len = len(params) + body = bytes([TYPE_COMMAND, command, (pl_len >> 8) & 0xFF, pl_len & 0xFF]) + params + checksum = sum(body) & 0xFF + return bytes([FRAME_HEADER]) + body + bytes([checksum, FRAME_END]) + + +def frame_set_power(power_dbm): + """Comando para definir a potencia de saida (em dBm, aceita decimais).""" + centi_dbm = int(round(power_dbm * 100)) + params = struct.pack(">H", centi_dbm & 0xFFFF) + return build_frame(CMD_SET_POWER, params) + + +def frame_get_power(): + return build_frame(CMD_GET_POWER) + + +def frame_get_info(info_id): + return build_frame(CMD_GET_INFO, bytes([info_id])) + + +def frame_single_poll(): + return build_frame(CMD_SINGLE_POLL) + + +def frame_start_multi_poll(count=4095): + params = bytes([0x22]) + struct.pack(">H", count & 0xFFFF) + return build_frame(CMD_MULTI_POLL, params) + + +def frame_stop_multi_poll(): + return build_frame(CMD_STOP_MULTI_POLL) + + +class FrameParser: + """ + Le bytes recebidos da porta serial e extrai quadros completos + (Header ... End), tolerando fragmentacao / lixo entre quadros. + """ + + def __init__(self): + self._buf = bytearray() + + def feed(self, data: bytes): + """Adiciona bytes recebidos e retorna lista de quadros completos.""" + self._buf.extend(data) + frames = [] + while True: + # procura o header + idx = self._buf.find(bytes([FRAME_HEADER])) + if idx == -1: + self._buf.clear() + break + if idx > 0: + del self._buf[:idx] + + if len(self._buf) < 7: + break # aguarda mais dados (tamanho minimo de um quadro sem payload) + + pl_len = (self._buf[3] << 8) | self._buf[4] + total_len = 7 + pl_len # header+type+cmd+len(2)+pl+checksum+end + + if len(self._buf) < total_len: + break # quadro incompleto, aguarda mais dados + + candidate = bytes(self._buf[:total_len]) + + if candidate[-1] == FRAME_END: + frames.append(candidate) + del self._buf[:total_len] + else: + # nao terminou em 0x7E onde esperado: descarta o header + # atual e tenta resincronizar a partir do proximo byte + del self._buf[:1] + + return frames + + +def parse_frame(frame: bytes): + """ + Interpreta um quadro ja validado (comeca com 0xBB, termina com 0x7E). + Retorna um dicionario com os campos decodificados. + """ + type_ = frame[1] + command = frame[2] + pl_len = (frame[3] << 8) | frame[4] + pl = frame[5:5 + pl_len] + checksum = frame[5 + pl_len] + result = { + "type": type_, + "command": command, + "pl": pl, + "checksum": checksum, + "raw": frame, + } + + if command == CMD_SINGLE_POLL and pl_len >= 5: + rssi_raw = pl[0] + pc = pl[1:3] + epc = pl[3:pl_len - 2] + crc = pl[pl_len - 2:pl_len] + result["kind"] = "tag" + result["rssi_raw"] = rssi_raw + # RSSI costuma ser reportado como valor com offset; aqui mostramos + # tanto o valor bruto quanto uma estimativa comum (raw - 256) para + # leitores que reportam em complemento de 2. + result["rssi_dbm"] = rssi_raw - 256 if rssi_raw > 127 else rssi_raw + result["pc"] = pc.hex().upper() + result["epc"] = epc.hex().upper() + result["crc"] = crc.hex().upper() + elif command == CMD_GET_INFO: + result["kind"] = "info" + result["info_text"] = pl.hex().upper() + elif command in (CMD_SET_POWER, CMD_GET_POWER): + result["kind"] = "power" + if len(pl) >= 2: + result["power_dbm"] = struct.unpack(">H", pl[:2])[0] / 100.0 + else: + result["kind"] = "other" + + return result + + +# -------------------------------------------------------------------------- +# Interface grafica +# -------------------------------------------------------------------------- +class RfidApp(tk.Tk): + def __init__(self): + super().__init__() + self.title("Leitor UHF RFID - Fonkan / JRD-100 / M100") + self.geometry("780x560") + self.minsize(680, 480) + + self.ser = None + self.reader_thread = None + self.stop_event = threading.Event() + self.event_queue = queue.Queue() + self.parser = FrameParser() + + self.tags = {} # epc -> {"count": int, "rssi": int, "last_seen": str} + + self._build_ui() + self._refresh_ports() + self._poll_queue() + + self.protocol("WM_DELETE_WINDOW", self._on_close) + + if serial is None: + messagebox.showerror( + "Dependencia ausente", + "A biblioteca 'pyserial' nao foi encontrada.\n\n" + "Instale com:\n pip install pyserial\n\n" + "e execute o aplicativo novamente.", + ) + + # ---------------------------------------------------------- UI layout + def _build_ui(self): + pad = {"padx": 6, "pady": 4} + + conn_frame = ttk.LabelFrame(self, text="Conexao") + conn_frame.pack(fill="x", **pad) + + ttk.Label(conn_frame, text="Porta:").grid(row=0, column=0, sticky="w", **pad) + self.port_var = tk.StringVar() + self.port_combo = ttk.Combobox(conn_frame, textvariable=self.port_var, width=18, state="readonly") + self.port_combo.grid(row=0, column=1, **pad) + + ttk.Button(conn_frame, text="Atualizar", command=self._refresh_ports).grid(row=0, column=2, **pad) + + ttk.Label(conn_frame, text="Baudrate:").grid(row=0, column=3, sticky="w", **pad) + self.baud_var = tk.StringVar(value=DEFAULT_BAUDRATE) + self.baud_combo = ttk.Combobox( + conn_frame, textvariable=self.baud_var, values=BAUDRATES, width=10, state="readonly" + ) + self.baud_combo.grid(row=0, column=4, **pad) + + ttk.Label(conn_frame, text="Potencia (dBm):").grid(row=0, column=5, sticky="w", **pad) + self.power_var = tk.DoubleVar(value=20.0) + self.power_spin = ttk.Spinbox( + conn_frame, from_=15, to=33, increment=1, textvariable=self.power_var, width=6 + ) + self.power_spin.grid(row=0, column=6, **pad) + + self.connect_btn = ttk.Button(conn_frame, text="Conectar", command=self._toggle_connection) + self.connect_btn.grid(row=0, column=7, **pad) + + self.set_power_btn = ttk.Button( + conn_frame, text="Aplicar potencia", command=self._set_power, state="disabled" + ) + self.set_power_btn.grid(row=0, column=8, **pad) + + # ------------------------------------------------------ Scan controls + scan_frame = ttk.LabelFrame(self, text="Varredura") + scan_frame.pack(fill="x", **pad) + + self.start_btn = ttk.Button(scan_frame, text="Iniciar varredura", command=self._start_scan, state="disabled") + self.start_btn.grid(row=0, column=0, **pad) + + self.stop_btn = ttk.Button(scan_frame, text="Parar varredura", command=self._stop_scan, state="disabled") + self.stop_btn.grid(row=0, column=1, **pad) + + self.single_btn = ttk.Button(scan_frame, text="Leitura unica", command=self._single_poll, state="disabled") + self.single_btn.grid(row=0, column=2, **pad) + + ttk.Button(scan_frame, text="Limpar tabela", command=self._clear_tags).grid(row=0, column=3, **pad) + + self.status_var = tk.StringVar(value="Desconectado") + ttk.Label(scan_frame, textvariable=self.status_var, foreground="#a33").grid( + row=0, column=4, sticky="e", padx=12 + ) + scan_frame.grid_columnconfigure(4, weight=1) + + # ------------------------------------------------------ Tags table + tags_frame = ttk.LabelFrame(self, text="Tags detectadas") + tags_frame.pack(fill="both", expand=True, **pad) + + columns = ("epc", "rssi", "count", "last_seen") + self.tree = ttk.Treeview(tags_frame, columns=columns, show="headings", height=10) + self.tree.heading("epc", text="EPC") + self.tree.heading("rssi", text="RSSI") + self.tree.heading("count", text="Leituras") + self.tree.heading("last_seen", text="Ultima leitura") + self.tree.column("epc", width=280) + self.tree.column("rssi", width=80, anchor="center") + self.tree.column("count", width=90, anchor="center") + self.tree.column("last_seen", width=160, anchor="center") + self.tree.pack(fill="both", expand=True, side="left", padx=(0, 4)) + + scrollbar = ttk.Scrollbar(tags_frame, orient="vertical", command=self.tree.yview) + self.tree.configure(yscroll=scrollbar.set) + scrollbar.pack(side="right", fill="y") + + # ------------------------------------------------------ Log console + log_frame = ttk.LabelFrame(self, text="Log / quadros brutos (depuracao)") + log_frame.pack(fill="both", expand=False, **pad) + + self.log_text = scrolledtext.ScrolledText(log_frame, height=8, state="disabled", font=("Consolas", 9)) + self.log_text.pack(fill="both", expand=True, padx=4, pady=4) + + # ------------------------------------------------------------- helpers + def _log(self, msg): + ts = datetime.now().strftime("%H:%M:%S") + self.log_text.configure(state="normal") + self.log_text.insert("end", f"[{ts}] {msg}\n") + self.log_text.see("end") + self.log_text.configure(state="disabled") + + def _refresh_ports(self): + ports = [] + if serial is not None: + ports = [p.device for p in serial.tools.list_ports.comports()] + self.port_combo["values"] = ports + if ports and not self.port_var.get(): + self.port_var.set(ports[0]) + self._log(f"Portas encontradas: {', '.join(ports) if ports else '(nenhuma)'}") + + def _set_controls_connected(self, connected): + state_conn = "disabled" if connected else "readonly" + self.port_combo.configure(state=state_conn) + self.baud_combo.configure(state=state_conn) + self.connect_btn.configure(text="Desconectar" if connected else "Conectar") + self.set_power_btn.configure(state="normal" if connected else "disabled") + self.start_btn.configure(state="normal" if connected else "disabled") + self.single_btn.configure(state="normal" if connected else "disabled") + self.stop_btn.configure(state="disabled") + self.status_var.set("Conectado" if connected else "Desconectado") + + # ------------------------------------------------------------ connect + def _toggle_connection(self): + if self.ser is None: + self._connect() + else: + self._disconnect() + + def _connect(self): + if serial is None: + messagebox.showerror("Erro", "pyserial nao instalado. Rode: pip install pyserial") + return + + port = self.port_var.get() + if not port: + messagebox.showwarning("Atencao", "Selecione uma porta serial.") + return + + try: + baud = int(self.baud_var.get()) + except ValueError: + messagebox.showwarning("Atencao", "Baudrate invalido.") + return + + try: + self.ser = serial.Serial(port, baudrate=baud, timeout=0.2) + except Exception as exc: + messagebox.showerror("Erro ao conectar", str(exc)) + self.ser = None + return + + self.stop_event.clear() + self.parser = FrameParser() + self.reader_thread = threading.Thread(target=self._reader_loop, daemon=True) + self.reader_thread.start() + + self._set_controls_connected(True) + self._log(f"Conectado em {port} @ {baud} bps") + + # consulta informacoes basicas do modulo ao conectar + self._send(frame_get_info(INFO_HW_VERSION)) + self._send(frame_get_info(INFO_SW_VERSION)) + self._send(frame_get_power()) + + def _disconnect(self): + self._stop_scan(silent=True) + self.stop_event.set() + if self.reader_thread is not None: + self.reader_thread.join(timeout=1.0) + if self.ser is not None: + try: + self.ser.close() + except Exception: + pass + self.ser = None + self._set_controls_connected(False) + self._log("Desconectado") + + def _on_close(self): + try: + self._disconnect() + finally: + self.destroy() + + # -------------------------------------------------------------- serial + def _send(self, frame: bytes): + if self.ser is None: + return + try: + self.ser.write(frame) + except Exception as exc: + self._log(f"Erro ao enviar: {exc}") + return + self._log(f"TX: {frame.hex(' ').upper()}") + + def _reader_loop(self): + while not self.stop_event.is_set(): + try: + if self.ser is None: + break + waiting = self.ser.in_waiting + data = self.ser.read(waiting if waiting else 1) + except Exception as exc: + self.event_queue.put(("log", f"Erro de leitura serial: {exc}")) + break + + if data: + frames = self.parser.feed(data) + for frame in frames: + self.event_queue.put(("frame", frame)) + + # ------------------------------------------------------------- actions + def _set_power(self): + try: + power = float(self.power_var.get()) + except (tk.TclError, ValueError): + messagebox.showwarning("Atencao", "Valor de potencia invalido.") + return + self._send(frame_set_power(power)) + self._log(f"Solicitada potencia de saida: {power:.1f} dBm") + + def _single_poll(self): + self._send(frame_single_poll()) + + def _start_scan(self): + self._send(frame_start_multi_poll(count=4095)) + self.start_btn.configure(state="disabled") + self.stop_btn.configure(state="normal") + self.status_var.set("Varredura em andamento...") + self._log("Varredura continua iniciada") + + def _stop_scan(self, silent=False): + if self.ser is not None: + self._send(frame_stop_multi_poll()) + self.start_btn.configure(state="normal" if self.ser is not None else "disabled") + self.stop_btn.configure(state="disabled") + if self.ser is not None: + self.status_var.set("Conectado") + if not silent: + self._log("Varredura interrompida") + + def _clear_tags(self): + self.tags.clear() + for item in self.tree.get_children(): + self.tree.delete(item) + + # ---------------------------------------------------------- fila/eventos + def _poll_queue(self): + try: + while True: + kind, payload = self.event_queue.get_nowait() + if kind == "log": + self._log(payload) + elif kind == "frame": + self._handle_frame(payload) + except queue.Empty: + pass + finally: + self.after(100, self._poll_queue) + + def _handle_frame(self, frame: bytes): + self._log(f"RX: {frame.hex(' ').upper()}") + try: + info = parse_frame(frame) + except Exception as exc: + self._log(f"Falha ao interpretar quadro: {exc}") + return + + if info["kind"] == "tag": + self._register_tag(info) + elif info["kind"] == "power" and "power_dbm" in info: + self._log(f"Potencia reportada pelo modulo: {info['power_dbm']:.2f} dBm") + elif info["kind"] == "info": + self._log(f"Info do modulo (cmd 0x03): {info['info_text']}") + + def _register_tag(self, info): + epc = info["epc"] + now = datetime.now().strftime("%H:%M:%S") + if epc in self.tags: + self.tags[epc]["count"] += 1 + self.tags[epc]["rssi"] = info["rssi_dbm"] + self.tags[epc]["last_seen"] = now + self.tree.item(self.tags[epc]["item"], values=(epc, info["rssi_dbm"], self.tags[epc]["count"], now)) + else: + item = self.tree.insert("", "end", values=(epc, info["rssi_dbm"], 1, now)) + self.tags[epc] = {"count": 1, "rssi": info["rssi_dbm"], "last_seen": now, "item": item} + + +if __name__ == "__main__": + app = RfidApp() + app.mainloop()