lronetto/uhf-rfid-gui-fonkan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
7b41ac054b
uhf-rfid-gui-fonkan/uhf_rfid_gui.py
lronetto 7b41ac054b 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 <noreply@anthropic.com>
2026-07-16 15:04:56 +00:00

558 lines
20 KiB
Python
Raw Blame History

#!/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()
Reference in New Issue View Git Blame Copy Permalink