lronetto/uhf-rfid-gui-fonkan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
22fca0246c
uhf-rfid-gui-fonkan/uhf_rfid_gui.py
lronetto 22fca0246c Add auto-reconnect and diagnostic hints for serial Access Denied errors 
The previous fix (port access lock) only addressed contention between
our own read/write threads. A recurring "ClearCommError failed -
Acesso negado" after the lock fix indicates the OS is invalidating the
port handle itself (USB power management suspending the port, the
RFID module browning out under high RF output power, or another app
holding the port) rather than an internal race.

Add:
- A "Reconectar automaticamente" checkbox (on by default) that retries
  opening the port with backoff (up to 5 attempts) after any fatal
  serial error, resuming the scan automatically if one was active.
- A generation counter so stale scheduled reconnects are discarded if
  the user manually connects/disconnects in the meantime.
- An actionable hint logged on fatal errors explaining the likely
  causes and how to fix each one.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-16 15:38:29 +00:00

712 lines
27 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"
MAX_RECONNECT_ATTEMPTS = 5
RECONNECT_DELAY_MS = 1500
# --------------------------------------------------------------------------
# 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()
# No Windows, ler e escrever na porta serial ao mesmo tempo a partir
# de threads diferentes sem sincronizacao costuma gerar
# PermissionError/WinError 5 ("Acesso negado") em GetOverlappedResult
# ou WriteFile, dependendo do driver USB-serial. Este lock serializa
# todo acesso a self.ser (leitura e escrita) para evitar a corrida.
self._port_lock = threading.RLock()
# Estado usado pela reconexao automatica: guarda os ultimos
# parametros usados com sucesso, quantas tentativas silenciosas ja
# foram feitas, se uma varredura estava ativa (para retoma-la), e um
# "numero de geracao" que invalida tentativas de reconexao agendadas
# quando o usuario toma uma acao manual (conectar/desconectar) antes
# delas dispararem.
self._last_port = None
self._last_baud = None
self._reconnect_attempts = 0
self._pending_resume_scan = False
self._conn_generation = 0
self.scanning = False
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)
self.auto_reconnect_var = tk.BooleanVar(value=True)
ttk.Checkbutton(
conn_frame,
text="Reconectar automaticamente em caso de erro na porta",
variable=self.auto_reconnect_var,
).grid(row=1, column=0, columnspan=6, sticky="w", padx=6, pady=(0, 4))
# ------------------------------------------------------ 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._conn_generation += 1 # invalida qualquer reconexao automatica antiga pendente
self._reconnect_attempts = 0
self._connect()
else:
self._disconnect()
def _connect(self, silent=False, expected_gen=None):
"""
Abre a porta serial. Quando silent=True, esta sendo chamado pela
rotina de reconexao automatica: usa os ultimos parametros que
funcionaram, nao exibe caixas de dialogo de erro (so registra no
log) e agenda uma nova tentativa em caso de falha.
expected_gen e o "numero de geracao" capturado no momento em que a
tentativa (silenciosa) foi agendada; se o usuario tomou uma acao
manual (conectar/desconectar) enquanto ela esperava, a geracao atual
ja mudou e esta tentativa e descartada sem tocar na porta.
"""
if silent and expected_gen is not None and expected_gen != self._conn_generation:
self._log("Reconexao automatica cancelada (nova acao do usuario).")
return
gen = self._conn_generation
if serial is None:
if not silent:
messagebox.showerror("Erro", "pyserial nao instalado. Rode: pip install pyserial")
return
port = self._last_port if silent else self.port_var.get()
baud_source = self._last_baud if silent else self.baud_var.get()
if not port:
if not silent:
messagebox.showwarning("Atencao", "Selecione uma porta serial.")
return
try:
baud = int(baud_source)
except (TypeError, ValueError):
if not silent:
messagebox.showwarning("Atencao", "Baudrate invalido.")
return
try:
ser = serial.Serial(port, baudrate=baud, timeout=0.2, write_timeout=1.0)
except Exception as exc:
if silent:
self._log(f"Falha ao reconectar: {exc}")
self._schedule_reconnect(gen)
else:
messagebox.showerror(
"Erro ao conectar",
f"{exc}\n\n"
"Se o erro for 'Acesso negado'/PermissionError, verifique se a porta\n"
"nao esta aberta em outro programa (Monitor Serial da Arduino IDE,\n"
"outra instancia deste app, PuTTY, etc.) e tente novamente.",
)
return
# se o usuario desconectou/reconectou manualmente enquanto esta
# tentativa (silenciosa) estava em andamento, descarta o resultado
if gen != self._conn_generation:
try:
ser.close()
except Exception:
pass
return
self.ser = ser
self._last_port = port
self._last_baud = baud
self._reconnect_attempts = 0
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())
if self._pending_resume_scan:
self._pending_resume_scan = False
self._log("Retomando varredura apos reconexao automatica...")
self.after(300, self._start_scan)
def _disconnect(self):
self._conn_generation += 1 # cancela qualquer reconexao automatica pendente
self._reconnect_attempts = 0
self._pending_resume_scan = False
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:
with self._port_lock:
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()
def _fatal_disconnect(self, reason):
"""
Encerra a conexao apos um erro irrecuperavel (ex.: PermissionError
do Windows) e devolve a interface a um estado consistente. Se
"Reconectar automaticamente" estiver marcado, agenda novas
tentativas; senao, o usuario so precisa clicar em "Conectar" de novo.
"""
self._log(reason)
self._log(
"Dica: 'Acesso negado' (WinError 5) na porta serial normalmente vem de "
"(1) gerenciamento de energia USB suspendendo a porta - em Gerenciador de "
"Dispositivos, na porta COM, aba Gerenciamento de Energia, desmarque 'Permitir "
"que o computador desligue este dispositivo para economizar energia'; "
"(2) o modulo RFID sofrendo queda de energia/reset ao transmitir em potencia "
"alta - tente reduzir a potencia ou usar um cabo/hub USB com alimentacao "
"externa; ou (3) outro programa (Monitor Serial, outra instancia deste app) "
"usando a mesma porta ao mesmo tempo."
)
self._pending_resume_scan = self.scanning
self.scanning = False
self.stop_event.set()
if self.ser is not None:
with self._port_lock:
try:
self.ser.close()
except Exception:
pass
self.ser = None
self._set_controls_connected(False)
gen = self._conn_generation
if self.auto_reconnect_var.get():
self.status_var.set("Erro na porta - reconectando...")
self._schedule_reconnect(gen)
else:
self.status_var.set("Desconectado (erro)")
def _schedule_reconnect(self, gen):
if gen != self._conn_generation or not self.auto_reconnect_var.get():
return
if self._reconnect_attempts >= MAX_RECONNECT_ATTEMPTS:
self._log(
f"Numero maximo de tentativas de reconexao automatica ({MAX_RECONNECT_ATTEMPTS}) "
"atingido. Verifique a causa (veja a dica acima) e clique em Conectar manualmente."
)
self._reconnect_attempts = 0
self.status_var.set("Desconectado (erro)")
return
self._reconnect_attempts += 1
self._log(
f"Tentando reconectar automaticamente "
f"({self._reconnect_attempts}/{MAX_RECONNECT_ATTEMPTS}) em "
f"{RECONNECT_DELAY_MS / 1000:.1f}s..."
)
self.after(RECONNECT_DELAY_MS, lambda: self._connect(silent=True, expected_gen=gen))
# -------------------------------------------------------------- serial
def _send(self, frame: bytes):
if self.ser is None:
return
try:
with self._port_lock:
self.ser.write(frame)
except Exception as exc:
self._fatal_disconnect(f"Erro ao enviar: {exc}")
return
self._log(f"TX: {frame.hex(' ').upper()}")
def _reader_loop(self):
# Le em pequenos pedaços e mantem o lock so durante a chamada
# in_waiting+read (nao-bloqueante, pois so le o que ja esta
# disponivel), para nao competir por muito tempo com escritas
# feitas pela thread principal.
while not self.stop_event.is_set():
try:
with self._port_lock:
if self.ser is None:
break
waiting = self.ser.in_waiting
data = self.ser.read(waiting) if waiting else b""
except Exception as exc:
self.event_queue.put(("fatal", f"Erro de leitura serial: {exc}"))
break
if data:
frames = self.parser.feed(data)
for frame in frames:
self.event_queue.put(("frame", frame))
else:
time.sleep(0.02)
# ------------------------------------------------------------- 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.scanning = True
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):
self.scanning = 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)
elif kind == "fatal":
self._fatal_disconnect(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