uhf-rfid-gui-fonkan/uhf_rfid_gui.py

#!/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()
        # 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()

        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, write_timeout=1.0)
        except Exception as exc:
            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.",
            )
            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:
            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, para que
        o usuario possa simplesmente clicar em "Conectar" de novo.
        """
        self._log(reason)
        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)
        self.status_var.set("Desconectado (erro)")

    # -------------------------------------------------------------- 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.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)
                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()