#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ MT6 RFID 读卡器测试程序 基于 USB HID 私有协议 """ import sys import os import time # 添加 PySimpleGUI 目录到路径 sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'PySimpleGUI')) import PySimpleGUI as sg try: import hid except ImportError: sg.popup_error('请安装 hidapi 库:\npip install hidapi') sys.exit(1) # ==================== 协议常量 ==================== REPORT_ID_SET_REQ = 0x01 REPORT_ID_GET_RES = 0x03 REPORT_SIZE = 256 # 设备信息 VENDOR_ID = 0xFFFF # 默认VID PRODUCT_ID = 0x0035 # 默认PID # ==================== 协议函数 ==================== def calc_outer_checksum(data_len_bytes: bytes, payload: bytes) -> int: """计算外层帧校验（XOR）""" result = 0 for b in data_len_bytes + payload: result ^= b return result def build_setreq_frame(data: bytes) -> bytes: """ 构建 SetReq 帧 :param data: Data 段数据 :return: 完整的 256 字节帧 """ # Frame Length: 从 0x07 到 Checksum（包含校验和，不含结束符） # = 2 (constant) + 2 (data length) + len(data) + 1 (checksum) frame_length = 5 + len(data) frame_length_bytes = frame_length.to_bytes(2, 'big') # Data Length data_length_bytes = len(data).to_bytes(2, 'big') # Checksum checksum = calc_outer_checksum(data_length_bytes, data) # 构建帧 frame = bytearray(REPORT_SIZE) frame[0] = REPORT_ID_SET_REQ frame[1:5] = b'\x00\x00\x00\x00' # Fixed frame[5:7] = frame_length_bytes # Frame Length frame[7:9] = b'\x00\x02' # Constant frame[9:11] = data_length_bytes # Data Length frame[11:11+len(data)] = data # Data frame[11+len(data)] = checksum # Checksum frame[12+len(data)] = 0x03 # End Marker return bytes(frame) def parse_getres_frame(data: bytes) -> tuple: """ 解析 GetRes 帧（带Checksum和End Marker验证） :param data: 接收到的数据 :return: (status, response_data) 或 None，校验失败返回None并打印错误 """ if len(data) < 14: # 最小帧长度需要包含Checksum和End Marker print(f"[DEBUG] 数据长度不足: {len(data)} < 14") return None # 检查 Report ID if data[0] != REPORT_ID_GET_RES: print(f"[DEBUG] Report ID错误: 0x{data[0]:02x} (应为 0x03)") return None # 解析帧结构（大端序） frame_length = int.from_bytes(data[5:7], 'big') data_length = int.from_bytes(data[9:11], 'big') # 检查数据长度是否合理 expected_total_len = 11 + data_length + 2 # header + data + checksum + end_marker if len(data) < expected_total_len: print(f"[DEBUG] 数据长度不匹配: 实际{len(data)}, 期望{expected_total_len}") return None # 提取 Data 段 response_data = data[11:11+data_length] # 提取 Checksum 和 End Marker received_checksum = data[11+data_length] received_end_marker = data[12+data_length] # 验证 End Marker（固定为 0x03） if received_end_marker != 0x03: print(f"[DEBUG] End Marker错误: 0x{received_end_marker:02x} (应为 0x03)") return None # 计算并验证 Checksum（XOR算法：Data Length两字节 + Data全部字节） calculated_checksum = calc_outer_checksum(data[9:11], response_data) if calculated_checksum != received_checksum: print(f"[DEBUG] Checksum错误: 计算=0x{calculated_checksum:02x}, 接收=0x{received_checksum:02x}") return None print(f"[DEBUG] 帧校验通过: Checksum=0x{received_checksum:02x}, EndMarker=0x{received_end_marker:02x}") if len(response_data) < 1: return None status = response_data[0] return (status, response_data[1:]) # ==================== 命令函数 ==================== def cmd_get_version() -> bytes: """读版本号命令""" return bytes([0xc0]) def cmd_factory_reset() -> bytes: """恢复出厂设置命令""" return bytes([0xcf]) def cmd_read_format() -> bytes: """读取格式命令""" return bytes([0x83]) def cmd_set_format() -> bytes: """设置格式命令""" return bytes([0x82, 0x0f, 0x00, 0x00, 0x00, 0x01, 0x01, 0x01, 0x00]) def cmd_buzzer(on: bool) -> bytes: """蜂鸣器控制命令""" return bytes([0xcd, 0x01 if on else 0x00]) def cmd_rf_power(on: bool) -> bytes: """射频电源控制命令""" return bytes([0x90, 0x01 if on else 0x00]) def cmd_set_power(power: int) -> bytes: """设置功率命令 (0-9)""" power = max(0, min(9, power)) return bytes([0xcc, power]) def cmd_set_mode(mode: int) -> bytes: """设置工作模式命令 1: 单标签巡查 2: 被动模式 3: 多标签巡查 """ return bytes([0x0f, mode]) def cmd_read_epc() -> bytes: """读取 EPC 命令""" return bytes([0xce, 0xbb, 0x00, 0x22, 0x00, 0x00, 0x22, 0x7e]) def cmd_select_card(epc_data: bytes) -> bytes: """选中卡命令""" # 固定前缀 + EPC 数据 payload = bytes([0x01, 0x00, 0x00, 0x00, 0x20, 0x10, 0x00]) + epc_data # 计算 checksum（包含 Card Op Command + Internal Length + Payload） card_op_cmd = bytes([0x00, 0x0c]) internal_len = len(payload).to_bytes(2, 'big') checksum = 0 for b in card_op_cmd + internal_len + payload: checksum = (checksum + b) & 0xFF return bytes([0xce, 0xbb]) + card_op_cmd + internal_len + payload + bytes([checksum, 0x7e]) def cmd_write_epc(old_epc_crc: bytes, new_epc: bytes) -> bytes: """写入 EPC 命令""" # EPC Len Indicator: EPC字节数 = 值 ÷ 4，所以值 = EPC字节数 * 4 epc_len_indicator = len(new_epc) * 4 if epc_len_indicator == 0: epc_len_indicator = 8 # 默认最小值 word_count = 2 + len(new_epc) // 2 payload = bytes([0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00]) # Reserved (8 bytes) payload += bytes([word_count]) # Word Count payload += old_epc_crc # Old EPC CRC (大端序) payload += bytes([epc_len_indicator, 0x00]) # EPC Len Indicator, Status payload += new_epc # New EPC Data # 计算 checksum（包含 Card Op Command + Internal Length + Payload） card_op_cmd = bytes([0x00, 0x49]) internal_len = len(payload).to_bytes(2, 'big') checksum = 0 for b in card_op_cmd + internal_len + payload: checksum = (checksum + b) & 0xFF return bytes([0xce, 0xbb]) + card_op_cmd + internal_len + payload + bytes([checksum, 0x7e]) # ==================== 设备管理类 ==================== class RFIDDevice: def __init__(self): self.device = None self.vendor_id = VENDOR_ID self.product_id = PRODUCT_ID self.current_path = None # 当前连接的设备路径 self.current_interface = -1 # 当前连接的接口号 self.current_path = None # 当前连接的设备路径 self.current_interface = -1 # 当前连接的接口号 def list_devices(self): """列出所有 HID 设备""" devices = [] try: for d in hid.enumerate(): devices.append({ 'vendor_id': d['vendor_id'], 'product_id': d['product_id'], 'manufacturer_string': d.get('manufacturer_string', ''), 'product_string': d.get('product_string', ''), 'serial_number': d.get('serial_number', ''), 'interface_number': d.get('interface_number', -1), 'path': d['path'] }) except Exception as e: print(f"枚举设备失败: {e}") return devices def connect(self, vendor_id=None, product_id=None): """通过VID/PID连接设备（会打开第一个匹配的设备）""" if vendor_id: self.vendor_id = vendor_id if product_id: self.product_id = product_id try: self.device = hid.device() self.device.open(self.vendor_id, self.product_id) self.current_path = None self.current_interface = -1 print(f"[DEBUG] 通过VID/PID连接: VID=0x{self.vendor_id:04x}, PID=0x{self.product_id:04x}") return True, f"已连接到设备 (VID: 0x{self.vendor_id:04x}, PID: 0x{self.product_id:04x}) - 注意：可能连接到任意接口" except Exception as e: print(f"[DEBUG] VID/PID连接失败: {e}") return False, f"连接失败: {e}" def connect_by_path(self, path, interface_number=-1): """通过路径连接设备（可以精确指定接口）""" try: self.device = hid.device() # path可能是bytes类型，需要处理 if isinstance(path, bytes): path_str = path else: path_str = path.encode('utf-8') if isinstance(path, str) else path self.device.open_path(path_str) self.current_path = path self.current_interface = interface_number # 获取设备信息 info = self.device.get_manufacturer_string() or "未知" product = self.device.get_product_string() or "未知" print(f"[DEBUG] 通过路径连接成功:") print(f"[DEBUG] 路径: {path}") print(f"[DEBUG] 接口号: {interface_number}") print(f"[DEBUG] 制造商: {info}") print(f"[DEBUG] 产品: {product}") path_display = path.decode('utf-8', errors='ignore') if isinstance(path, bytes) else str(path) return True, f"已连接到设备 (IF={interface_number}, 路径: ...{path_display[-30:]})" except Exception as e: print(f"[DEBUG] 路径连接失败: {e}") return False, f"连接失败: {e}" def disconnect(self): """断开连接""" if self.device: try: print(f"[DEBUG] 断开设备连接 (接口: {self.current_interface})") self.device.close() except: pass self.device = None self.current_path = None self.current_interface = -1 return "已断开连接" def is_connected(self): """检查是否已连接""" if self.device: #print(f"[DEBUG] 设备已连接 - 接口: {self.current_interface}, 路径: {self.current_path}") return True return False def get_connection_info(self): """获取当前连接信息""" return { 'interface': self.current_interface, 'path': self.current_path, 'vendor_id': self.vendor_id, 'product_id': self.product_id } def send_command(self, cmd_data: bytes): """发送命令并接收响应""" if not self.device: return None, "设备未连接" try: # 构建并发送 SetReq frame = build_setreq_frame(cmd_data) self.device.send_feature_report(frame) # 接收 GetRes time.sleep(0.1) response = self.device.get_feature_report(REPORT_ID_GET_RES, REPORT_SIZE) # 解析响应 result = parse_getres_frame(bytes(response)) if result is None: return None, "响应格式错误" status, data = result return data, None except Exception as e: return None, f"通信错误: {e}" # ==================== GUI 界面 ==================== def create_window(): """创建主窗口""" sg.theme('LightBlue2') # 设备连接区域 device_frame = [ [sg.Text('VID (十六进制):'), sg.Input('FFFF', size=(8, 1), key='-VID-'), sg.Text('PID (十六进制):'), sg.Input('0035', size=(8, 1), key='-PID-'), sg.Button('连接', key='-CONNECT-'), sg.Button('断开', key='-DISCONNECT-'), sg.Button('枚举设备', key='-ENUM-')], [sg.Text('状态: ', size=(12, 1)), sg.Text('未连接', key='-STATUS-', text_color='red')] ] # 基本命令区域 basic_frame = [ [sg.Button('读取版本号', key='-VERSION-', size=(12, 1)), sg.Button('恢复出厂', key='-FACTORY_RESET-', size=(12, 1)), sg.Button('读取格式', key='-READ_FORMAT-', size=(12, 1)), sg.Button('设置格式', key='-SET_FORMAT-', size=(12, 1))], [sg.Button('打开蜂鸣器', key='-BUZZER_ON-', size=(12, 1)), sg.Button('关闭蜂鸣器', key='-BUZZER_OFF-', size=(12, 1)), sg.Button('打开射频', key='-RF_ON-', size=(12, 1)), sg.Button('关闭射频', key='-RF_OFF-', size=(12, 1))], [sg.Text('功率 (0-9):'), sg.Slider(range=(0, 9), default_value=8, orientation='h', size=(20, 15), key='-POWER-'), sg.Button('设置功率', key='-SET_POWER-')], [sg.Text('工作模式:'), sg.Radio('单标签巡查', 'mode', key='-MODE1-', default=True), sg.Radio('被动模式', 'mode', key='-MODE2-'), sg.Radio('多标签巡查', 'mode', key='-MODE3-'), sg.Button('设置模式', key='-SET_MODE-')] ] # EPC 操作区域 epc_frame = [ [sg.Button('读取 EPC', key='-READ_EPC-', size=(15, 1)), sg.Button('选中卡', key='-SELECT_CARD-', size=(15, 1))], [sg.Text('EPC 数据 (十六进制):'), sg.Input('', size=(40, 1), key='-EPC_DATA-')], [sg.Text('新 EPC (十六进制):'), sg.Input('', size=(40, 1), key='-NEW_EPC-')], [sg.Button('写入 EPC', key='-WRITE_EPC-', size=(15, 1))] ] # 日志输出区域 log_frame = [ [sg.Multiline('', size=(80, 15), key='-LOG-', autoscroll=True, disabled=True)] ] layout = [ [sg.Frame('设备连接', device_frame)], [sg.Frame('基本命令', basic_frame)], [sg.Frame('EPC 操作', epc_frame)], [sg.Frame('日志', log_frame)], [sg.Button('清空日志', key='-CLEAR_LOG-'), sg.Button('退出', key='-EXIT-')] ] return sg.Window('MT6 RFID 读卡器测试程序', layout, finalize=True) def log_message(window, message, is_hex=False): """输出日志消息""" timestamp = time.strftime('%H:%M:%S') if is_hex: if isinstance(message, bytes): hex_str = ' '.join(f'{b:02x}' for b in message) window['-LOG-'].print(f'[{timestamp}] {hex_str}') else: window['-LOG-'].print(f'[{timestamp}] {message}') else: window['-LOG-'].print(f'[{timestamp}] {message}') def hex_to_bytes(hex_str: str) -> bytes: """十六进制字符串转字节""" hex_str = hex_str.replace(' ', '').replace(',', '') return bytes.fromhex(hex_str) def main(): """主函数""" window = create_window() rfid = RFIDDevice() device_list = [] while True: event, values = window.read() if event in (sg.WIN_CLOSED, '-EXIT-'): rfid.disconnect() break # 枚举设备 if event == '-ENUM-': device_list = rfid.list_devices() if device_list: log_message(window, f"找到 {len(device_list)} 个 HID 设备") device_strs = [] for i, d in enumerate(device_list): # 显示路径信息，帮助区分相同VID/PID的设备 path = d['path'] path_str = path.decode('utf-8', errors='ignore') if isinstance(path, bytes) else str(path) s = f"[{i}] VID:0x{d['vendor_id']:04x} PID:0x{d['product_id']:04x} " \ f"IF:{d['interface_number']} {d['manufacturer_string']} {d['product_string']} " \ f"Path:...{path_str[-20:]}" device_strs.append(s) log_message(window, s) # 详细日志显示完整路径 log_message(window, f" -> 完整路径: {path_str}") # 创建设备选择窗口 select_layout = [ [sg.Listbox(values=device_strs, size=(80, min(10, len(device_strs))), key='-SELECTED-')], [sg.Button('使用选中设备'), sg.Button('取消')] ] select_window = sg.Window('选择设备', select_layout, modal=True) while True: sel_event, sel_values = select_window.read() if sel_event in (sg.WIN_CLOSED, '取消'): select_window.close() break if sel_event == '使用选中设备': selected_idx = sel_values.get('-SELECTED-', []) if selected_idx: idx = device_strs.index(selected_idx[0]) d = device_list[idx] window['-VID-'].update(f"{d['vendor_id']:04x}") window['-PID-'].update(f"{d['product_id']:04x}") # 使用路径连接设备 path = d['path'] interface_number = d['interface_number'] log_message(window, f"正在连接设备...") log_message(window, f" VID: 0x{d['vendor_id']:04x}, PID: 0x{d['product_id']:04x}") log_message(window, f" 接口号: {interface_number}") path_str = path.decode('utf-8', errors='ignore') if isinstance(path, bytes) else str(path) log_message(window, f" 路径: {path_str}") success, msg = rfid.connect_by_path(path, interface_number) log_message(window, msg) if success: window['-STATUS-'].update('已连接', text_color='green') log_message(window, f"当前使用接口: IF{rfid.current_interface}") else: window['-STATUS-'].update('连接失败', text_color='red') select_window.close() break else: log_message(window, "未找到 HID 设备") # 连接设备 if event == '-CONNECT-': try: vid = int(values['-VID-'], 16) pid = int(values['-PID-'], 16) success, msg = rfid.connect(vid, pid) log_message(window, msg) if success: window['-STATUS-'].update('已连接', text_color='green') else: window['-STATUS-'].update('连接失败', text_color='red') except ValueError: log_message(window, "请输入有效的十六进制 VID/PID") # 断开连接 if event == '-DISCONNECT-': msg = rfid.disconnect() log_message(window, msg) window['-STATUS-'].update('未连接', text_color='red') # 读取版本号 if event == '-VERSION-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 读版本号") data, err = rfid.send_command(cmd_get_version()) if err: log_message(window, f"错误: {err}") else: log_message(window, data, is_hex=True) try: version = data.decode('ascii', errors='ignore') log_message(window, f"版本: {version}") except: pass # 恢复出厂设置 if event == '-FACTORY_RESET-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 恢复出厂设置") data, err = rfid.send_command(cmd_factory_reset()) if err: log_message(window, f"错误: {err}") else: log_message(window, data, is_hex=True) # 读取格式 if event == '-READ_FORMAT-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 读取格式") data, err = rfid.send_command(cmd_read_format()) if err: log_message(window, f"错误: {err}") else: log_message(window, data, is_hex=True) # 设置格式 if event == '-SET_FORMAT-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 设置格式") data, err = rfid.send_command(cmd_set_format()) if err: log_message(window, f"错误: {err}") else: log_message(window, data, is_hex=True) # 射频电源控制 if event == '-RF_ON-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 打开射频") data, err = rfid.send_command(cmd_rf_power(True)) if err: log_message(window, f"错误: {err}") else: log_message(window, data, is_hex=True) if event == '-RF_OFF-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 关闭射频") data, err = rfid.send_command(cmd_rf_power(False)) if err: log_message(window, f"错误: {err}") else: log_message(window, data, is_hex=True) # 蜂鸣器控制 if event == '-BUZZER_ON-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 打开蜂鸣器") data, err = rfid.send_command(cmd_buzzer(True)) if err: log_message(window, f"错误: {err}") else: log_message(window, f"响应: {data.hex() if data else '无数据'}") if event == '-BUZZER_OFF-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 关闭蜂鸣器") data, err = rfid.send_command(cmd_buzzer(False)) if err: log_message(window, f"错误: {err}") else: log_message(window, f"响应: {data.hex() if data else '无数据'}") # 设置功率 if event == '-SET_POWER-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: power = int(values['-POWER-']) log_message(window, f"发送命令: 设置功率 = {power}") data, err = rfid.send_command(cmd_set_power(power)) if err: log_message(window, f"错误: {err}") else: log_message(window, f"响应: {data.hex() if data else '无数据'}") # 设置工作模式 if event == '-SET_MODE-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: mode = 1 if values['-MODE2-']: mode = 2 elif values['-MODE3-']: mode = 3 mode_names = {1: '单标签巡查', 2: '被动模式', 3: '多标签巡查'} log_message(window, f"发送命令: 设置模式 = {mode_names.get(mode, mode)}") data, err = rfid.send_command(cmd_set_mode(mode)) if err: log_message(window, f"错误: {err}") else: log_message(window, f"响应: {data.hex() if data else '无数据'}") # 读取 EPC if event == '-READ_EPC-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: log_message(window, "发送命令: 读取 EPC") data, err = rfid.send_command(cmd_read_epc()) if err: log_message(window, f"错误: {err}") else: log_message(window, f"响应 (hex): ", is_hex=True) log_message(window, data, is_hex=True) # 解析 EPC # 注意: parse_getres_frame返回的data是从bb开始的（status已被去除） # 原始格式: 00 bb 02 22 00 0b d3 18 00 [EPC] ... # 去掉status后: bb 02 22 00 0b d3 18 00 [EPC] ... # 所以: data[0]=bb, data[1:3]=Card Op Resp, data[5]=RSSI, data[6]=EPC Len, data[8:]=EPC if data and len(data) > 10: # 检查 Magic 和 Card Op Resp if data[0:1] == b'\xbb' and data[1:3] == b'\x02\x22': # 有卡，解析 EPC rssi = data[5] if len(data) > 5 else 0 epc_len_indicator = data[6] if len(data) > 6 else 0 epc_len = epc_len_indicator // 4 if epc_len_indicator > 0 else 0 epc_data = data[8:8+epc_len] if len(data) > 8 else b'' log_message(window, f"Magic: 0x{data[0]:02x} (应为 bb)") log_message(window, f"Card Op Resp: {data[1:3].hex()} (02 22=有卡)") log_message(window, f"RSSI: 0x{rssi:02x}") log_message(window, f"EPC Len Indicator: 0x{epc_len_indicator:02x} (字节数={epc_len})") log_message(window, f"EPC: {epc_data.hex()}") window['-EPC_DATA-'].update(epc_data.hex()) elif data[0:1] == b'\xbb' and data[1:3] == b'\x01\xff': log_message(window, "无卡或读取失败 (Card Op Resp: 01 ff)") else: log_message(window, f"未知响应格式:") log_message(window, f" Magic: {data[0:1].hex()} (应为 bb)") log_message(window, f" Card Op Resp: {data[1:3].hex()}") # 选中卡 if event == '-SELECT_CARD-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: epc_hex = values['-EPC_DATA-'].strip() if not epc_hex: log_message(window, "错误: 请输入 EPC 数据") else: try: epc_data = hex_to_bytes(epc_hex) log_message(window, f"发送命令: 选中卡 EPC={epc_data.hex()}") data, err = rfid.send_command(cmd_select_card(epc_data)) if err: log_message(window, f"错误: {err}") else: log_message(window, f"响应 (hex): {data.hex() if data else '无数据'}") # 注意: parse_getres_frame返回的data是从bb开始的（status已被去除） # 原始格式: 00 bb 01 0c 00 01 00 0e 7e # 去掉status后: bb 01 0c 00 01 00 0e 7e # data[0]=bb, data[1:3]=01 0c(Card Op Resp), data[3:5]=00 01(Internal Length), data[5]=00(Payload-选中成功标志) if data and len(data) > 5: if data[5] == 0x00: log_message(window, "选中成功") else: log_message(window, f"选中失败: 0x{data[5]:02x}") except ValueError: log_message(window, "错误: EPC 数据格式错误，请使用十六进制") # 写入 EPC if event == '-WRITE_EPC-': if not rfid.is_connected(): log_message(window, "错误: 设备未连接") else: new_epc_hex = values['-NEW_EPC-'].strip() if not new_epc_hex: log_message(window, "错误: 请输入新的 EPC 数据") else: try: new_epc = hex_to_bytes(new_epc_hex) # 这里需要一个旧的 EPC CRC，暂时使用示例值 old_epc_crc = bytes([0xca, 0x9e]) # 需要根据实际 EPC 计算 log_message(window, f"发送命令: 写入 EPC={new_epc.hex()}") data, err = rfid.send_command(cmd_write_epc(old_epc_crc, new_epc)) if err: log_message(window, f"错误: {err}") else: log_message(window, f"响应 (hex): {data.hex() if data else '无数据'}") except ValueError: log_message(window, "错误: EPC 数据格式错误，请使用十六进制") # 清空日志 if event == '-CLEAR_LOG-': window['-LOG-'].update('') window.close() if __name__ == '__main__': main()