Small script to read out RFID-Tags

rfid.py

import serial
import time

# --- Config ---
SERIAL_PORT = "COM4"
BAUD_RATE = 9600
BUFFER_SIZE = 1024

def epc_to_gs1(epc_hex: str) -> str:
    """
    Converts a SGTIN-96 EPC-HEX string into a GS1-Code as (01)GTIN(21)Serial.
    Args:
        epc_hex (str): 24 Hex-String (SGTIN-96 EPC).
    Returns:
        str: GS1-Code in Format (01)GTIN(21)Serial.
    """
    # Length check
    if len(epc_hex) != 24:
        raise ValueError("Expected exactly 24 bytes.")

    epc_bin = bin(int(epc_hex, 16))[2:].zfill(96)

    # Header check
    header = epc_bin[0:8]
    if header != '00110000': # SGTIN-96 EPC
        raise ValueError("Unexpected header")

    # Filter and partition extraction
    filter_value = int(epc_bin[8:11], 2)
    partition = int(epc_bin[11:14], 2)

    # Partitions from GS1 TDT 2.2.0
    partition_table = {
        0: (40, 12, 4, 1),
        1: (37, 11, 7, 2),
        2: (34, 10, 10, 3),
        3: (30, 9, 14, 4),
        4: (27, 8, 17, 5),
        5: (24, 7, 20, 6),
        6: (20, 6, 24, 7),
    }

    if partition not in partition_table:
        raise ValueError("Unexpected value.")

    cp_bits, cp_digits, ir_bits, ir_digits = partition_table[partition]

    # Extract bit field
    cp_bin = epc_bin[14:14+cp_bits]
    ir_bin = epc_bin[14+cp_bits:14+cp_bits+ir_bits]
    serial_bin = epc_bin[14+cp_bits+ir_bits:]

    # Binary to int
    cp_int = int(cp_bin, 2)
    ir_int = int(ir_bin, 2)
    serial_int = int(serial_bin, 2)

    # Fill zeros
    cp_str = str(cp_int).zfill(cp_digits)
    ir_str = str(ir_int).zfill(ir_digits)
    serial_str = str(serial_int).zfill(12)

    # Building GTIN
    indicator_digit = ir_str[0]
    item_ref_body = ir_str[1:]
    gtin = indicator_digit + cp_str + item_ref_body

    # Checkdigit (GS1-Modulo-10)
    def calculate_check_digit(gtin):
        total = 0
        for i, c in enumerate(reversed(gtin)):
            n = int(c)
            total += n * 3 if (i % 2 == 0) else n
        return str((10 - (total % 10)) % 10)

    check_digit = calculate_check_digit(gtin)
    final_gtin = gtin + check_digit

    return f"(01){final_gtin}(21){serial_str}"

def decode_tlv(data):
    i = 0
    while i < len(data):
        if i + 1 >= len(data): break
        t_type = data[i]
        t_len = data[i+1]
        t_val = data[i+2 : i+2+t_len]
        
        if t_type == 0x50: # Single Tag Container
            decode_tlv(t_val)
        elif t_type == 0x01: # EPC
            epc_hex = t_val.hex().upper()
            print(f"EPC: {epc_hex}")
            print(epc_to_gs1(epc_hex))
        elif t_type == 0x05: # RSSI Attribut
            rssi = int.from_bytes(t_val, 'big')
            print(f" -> Signal (RSSI): -{rssi} dBm")
            print()
        
        i += 2 + t_len

def run_reader():
    buffer = b""
    try:
        ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=0.1)
        print(f"Waiting for RFID Tags at {SERIAL_PORT}... (Press Ctrl+C to exit)")
    except Exception as e:
        print(f"Error opening {SERIAL_PORT}: {e}")
        return

    try:
        while True:
            chunk = ser.read(BUFFER_SIZE)
            if chunk:
                buffer += chunk

            while len(buffer) >= 9:
                # Header check 'RF'
                start = buffer.find(b"RF")
                if start == -1:
                    buffer = b""
                    break
                if start > 0:
                    buffer = buffer[start:]
                
                if len(buffer) < 8: break # Wait for next bytes...
                
                # Param Length aus Bytes 6 & 7
                param_len = int.from_bytes(buffer[6:8], "big")
                total_frame_len = 8 + param_len + 1 # Header(8) + Params + Checksum(1)

                if len(buffer) < total_frame_len:
                    break # Still waiting

                # Extract and process
                frame = buffer[:total_frame_len]
                buffer = buffer[total_frame_len:]
                
                # Notification Frames (0x02) / Tag uploads (0x80)
                if frame[2] == 0x02 and frame[5] == 0x80:
                    decode_tlv(frame[8:-1])

    except KeyboardInterrupt:
        print("Exit.")
    finally:
        ser.close()

if __name__ == "__main__":
    run_reader()