513 lines
20 KiB
Python
513 lines
20 KiB
Python
import socket
|
||
import threading
|
||
import time
|
||
import binascii
|
||
from typing import Optional, Callable, Dict, Any, Set
|
||
from collections import Counter
|
||
from .crc16 import crc16
|
||
from .command_hex import command_hex
|
||
|
||
|
||
class rfid_service:
|
||
"""
|
||
RFID读写器服务
|
||
"""
|
||
def __init__(self, host='192.168.1.190', port=6000):
|
||
"""
|
||
初始化RFID控制器
|
||
|
||
参数:
|
||
host: RFID读写器IP地址
|
||
port: RFID读写器端口号
|
||
"""
|
||
self.host = host
|
||
self.port = port
|
||
self._thread_signal = False
|
||
self._receive_thread = None
|
||
self._callback = None
|
||
self._pause_receive = False
|
||
|
||
self._data_buffer = [] # 收到推送的数据
|
||
self._last_collect_time = None
|
||
self._data_lock = threading.Lock() # 用于保护数据缓冲区的锁
|
||
|
||
# 需要过滤掉的数据(字符串)
|
||
self._filter_value = None
|
||
|
||
self.check_time_seconds = 60.0 # 采集数据时间（秒）
|
||
# 超时设置
|
||
self._connect_timeout = 5.0 # 连接超时时间（秒）
|
||
self.client_socket = None
|
||
self.connected = False
|
||
#链接失败次数
|
||
self._error_count = 0
|
||
# #30个字(读卡器工作模式中设定)+2字节CRC+4字节 len+0x00+oxEE+0x00
|
||
self._buffer_length=66
|
||
#设置读取命令设置时连接最大次数
|
||
self._max_command_count=3
|
||
|
||
|
||
def CreatConnect(self)->bool:
|
||
try:
|
||
if self.client_socket:
|
||
self.client_socket.close()
|
||
self.connected = False
|
||
self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
self.client_socket.settimeout(self._connect_timeout)
|
||
print(f"准备连接RFID读写器")
|
||
self.client_socket.connect((self.host, self.port))
|
||
self.connected = True
|
||
print(f"成功连接到RFID读写器: {self.host}:{self.port}")
|
||
self._error_count = 0 # 重置错误计数
|
||
return True
|
||
except Exception as e:
|
||
print(f"连接RFID读写器失败: {e}")
|
||
self.connected = False
|
||
self._error_count += 1
|
||
return False
|
||
|
||
def connect(self) -> bool:
|
||
"""
|
||
连接到RFID读写器
|
||
|
||
返回:
|
||
bool: 连接成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
sock.settimeout(self._connect_timeout)
|
||
sock.connect((self.host, self.port))
|
||
print(f"成功连接到RFID读写器: {self.host}:{self.port}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"连接RFID读写器失败: {e}")
|
||
return False
|
||
|
||
def _send_command(self, command_data: bytes) -> Optional[bytes]:
|
||
"""
|
||
发送命令到RFID读写器并等待响应
|
||
|
||
参数:
|
||
command_data: 要发送的命令字节数据
|
||
|
||
返回:
|
||
bytes: 读写器响应数据，如果失败返回None
|
||
"""
|
||
for i in range(self._max_command_count):
|
||
try:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
sock.settimeout(self._connect_timeout) # 设置连接超时
|
||
sock.connect((self.host, self.port))
|
||
sock.send(command_data)
|
||
response = sock.recv(1024)
|
||
print(f"收到RFID响应: {binascii.hexlify(response).decode()}")
|
||
return response
|
||
except socket.timeout:
|
||
|
||
if i==self._max_command_count-1:
|
||
print(f"RFID命令响应超时,已达最大重试次数: {self._max_command_count}")
|
||
return None
|
||
else:
|
||
print(f"RFID命令响应超时 (超时设置: {self._connect_timeout}秒,)重连中.")
|
||
except Exception as e:
|
||
print(f"RFID命令发送错误: {e}")
|
||
if i==self._max_command_count-1:
|
||
print(f"RFID命令发送错误,已达最大重试次数: {self._max_command_count}")
|
||
return None
|
||
|
||
#region ======= 配置命令封装 =======
|
||
|
||
def set_working_mode(self, address: int, mode_params: Dict[str, Any]) -> bool:
|
||
"""
|
||
设置工作模式
|
||
|
||
参数:
|
||
address: 读写器地址
|
||
mode_params: 工作模式参数字典，包含以下键:
|
||
- read_mode: 读取模式 (0x00-0x03),0应答模式,1为主动模式,2触发模式(低电平有效),3触发模式(高电平有效)
|
||
- mode_state: Bit1=0时韦根输出，Bit1=1时RS232输出,Bit2=0时开蜂鸣器提示，Bit2=1时关蜂鸣器提示.
|
||
- mem_inven: 0x00：保留区；0x01：EPC存储器；0x02：TID存储器；0x03：用户存储器；0x04：多张查询；0x05：单张查询。0x06：多张TID询查
|
||
- first_adr: 第一个地址 (0x00-0xFF)
|
||
- word_num: 字数量Word_Num不能超过3
|
||
- tag_time: 主动模式下标签输出间隔时间(0~255)*1s，对同一张标签在间隔时间内只输出一次
|
||
|
||
返回:
|
||
bool: 设置成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
# 构建命令
|
||
command = command_hex.build_set_mode_command(
|
||
address=address,
|
||
# wg_mode=mode_params.get('wg_mode', 0),
|
||
# wg_data_interval=mode_params.get('wg_data_interval', 0),
|
||
# wg_pulse_width=mode_params.get('wg_pulse_width', 0),
|
||
# wg_pulse_interval=mode_params.get('wg_pulse_interval', 0),
|
||
read_mode=mode_params.get('read_mode', 1),
|
||
mode_state=mode_params.get('mode_state', 2), #开蜂鸣器提示1为不开
|
||
mem_inven=mode_params.get('mem_inven', 3),
|
||
first_adr=mode_params.get('first_adr', 0),
|
||
word_num=mode_params.get('word_num', 30),
|
||
tag_filt_time=mode_params.get('tag_filt_time', 0)
|
||
)
|
||
|
||
# 发送命令
|
||
response = self._send_command(command)
|
||
if response:
|
||
# 解析响应
|
||
try:
|
||
return command_hex.parse_set_mode_hex(response)
|
||
except Exception as e:
|
||
print(f"解析设置响应错误: {e}")
|
||
return False
|
||
return False
|
||
except Exception as e:
|
||
print(f"设置工作模式错误: {e}")
|
||
return False
|
||
|
||
def read_working_mode(self, address: int) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
读取工作模式
|
||
|
||
参数:
|
||
address: 读写器地址
|
||
|
||
返回:
|
||
dict: 工作模式参数，如果失败返回None
|
||
"""
|
||
try:
|
||
# 构建命令
|
||
command = command_hex.build_read_mode_command(address=address)
|
||
|
||
# 发送命令
|
||
response = self._send_command(command)
|
||
if response:
|
||
# 解析响应
|
||
try:
|
||
return command_hex.parse_read_mode_hex(response)
|
||
except Exception as e:
|
||
print(f"解析读取响应错误: {e}")
|
||
return None
|
||
return None
|
||
except Exception as e:
|
||
print(f"读取工作模式错误: {e}")
|
||
return None
|
||
|
||
def set_power(self, address: int, power_value: int) -> bool:
|
||
"""
|
||
设置读写器功率
|
||
|
||
参数:
|
||
address: 读写器地址
|
||
power_value: 功率值 (0-26 dBm)
|
||
|
||
返回:
|
||
bool: 设置成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
# 构建命令
|
||
command =command_hex.build_set_power_command(address=address, power_value=power_value)
|
||
|
||
# 发送命令
|
||
response = self._send_command(command)
|
||
if response:
|
||
# 解析响应
|
||
try:
|
||
return command_hex.parse_set_power_hex(response)
|
||
except Exception as e:
|
||
print(f"解析功率设置响应错误: {e}")
|
||
return False
|
||
return False
|
||
except Exception as e:
|
||
print(f"设置功率错误: {e}")
|
||
return False
|
||
|
||
def set_trigger_delay(self, address: int, delay_time: int) -> bool:
|
||
"""
|
||
设置触发延时
|
||
|
||
参数:
|
||
address: 读写器地址
|
||
delay_time: 延时时间 (0-254秒，255表示读取参数)
|
||
|
||
返回:
|
||
bool: 设置成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
# 构建命令
|
||
command = command_hex.build_trigger_delay_command(address=address, delay_time=delay_time)
|
||
|
||
# 发送命令
|
||
response = self._send_command(command)
|
||
if response:
|
||
# 解析响应
|
||
try:
|
||
if delay_time == 255:
|
||
# 读取模式
|
||
current_delay = command_hex.parse_trigger_delay_hex(response)
|
||
print(f"当前触发延时: {current_delay}秒")
|
||
return True
|
||
else:
|
||
# 设置模式
|
||
return True # 如果发送成功且没有异常，认为设置成功
|
||
except Exception as e:
|
||
print(f"解析触发延时响应错误: {e}")
|
||
return False
|
||
return False
|
||
except Exception as e:
|
||
print(f"设置触发延时错误: {e}")
|
||
return False
|
||
|
||
def set_read_interval(self, address: int, interval_time: int) -> bool:
|
||
"""
|
||
设置主动模式读卡间隔
|
||
|
||
参数:
|
||
address: 读写器地址
|
||
interval_time: 读卡间隔 (0-254毫秒，255表示读取参数)
|
||
|
||
返回:
|
||
bool: 设置成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
# 构建命令
|
||
command = command_hex.build_set_readinterval_command(address=address, delay_time=interval_time)
|
||
|
||
# 发送命令
|
||
response = self._send_command(command)
|
||
if response:
|
||
# 解析响应
|
||
try:
|
||
return command_hex.parse_set_readinterval_hex(response)
|
||
except Exception as e:
|
||
print(f"解析读卡间隔响应错误: {e}")
|
||
return False
|
||
return False
|
||
except Exception as e:
|
||
print(f"设置读卡间隔错误: {e}")
|
||
return False
|
||
|
||
def get_read_interval(self, address: int) -> int:
|
||
"""
|
||
设置主动模式读卡间隔
|
||
|
||
参数:
|
||
address: 读写器地址
|
||
|
||
返回:
|
||
int: 当前读卡间隔（0-254毫秒），如果失败返回-1
|
||
"""
|
||
try:
|
||
# 构建命令
|
||
command = command_hex.build_set_readinterval_command(address=address, delay_time=255)
|
||
|
||
# 发送命令
|
||
response = self._send_command(command)
|
||
if response:
|
||
# 解析响应
|
||
try:
|
||
loc_suc= command_hex.parse_set_readinterval_hex(response)
|
||
if loc_suc:
|
||
return response[4]
|
||
else:
|
||
return -1
|
||
except Exception as e:
|
||
print(f"解析读卡间隔响应错误: {e}")
|
||
return False
|
||
return False
|
||
except Exception as e:
|
||
print(f"设置读卡间隔错误: {e}")
|
||
return False
|
||
|
||
def read_reader_info(self, address: int) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
读取读写器信息
|
||
|
||
参数:
|
||
address: 读写器地址
|
||
|
||
返回:
|
||
dict: 读写器信息，如果失败返回None
|
||
"""
|
||
try:
|
||
# 构建读取读写器信息的命令
|
||
command = command_hex.build_reader_info_command(address=address)
|
||
|
||
# 发送命令
|
||
response = self._send_command(command)
|
||
if response:
|
||
# 解析响应
|
||
try:
|
||
return command_hex.parse_reader_info_hex(response)
|
||
except Exception as e:
|
||
print(f"解析读写器信息响应错误: {e}")
|
||
return None
|
||
return None
|
||
except Exception as e:
|
||
print(f"读取读写器信息错误: {e}")
|
||
return None
|
||
|
||
#endregion
|
||
|
||
|
||
def start_receiver(self, callback: Optional[Callable[[str], None]] = None)->bool:
|
||
"""
|
||
开始接收RFID推送的数据
|
||
|
||
参数:
|
||
callback: 接收数据的回调函数，参数为收集到的RFID字符串数据
|
||
"""
|
||
if self._thread_signal:
|
||
print("RFID数据接收已经在运行")
|
||
return True
|
||
|
||
self._callback = callback
|
||
if self.CreatConnect():
|
||
self._thread_signal = True
|
||
|
||
# 清空数据缓冲区
|
||
with self._data_lock:
|
||
self._data_buffer.clear()
|
||
self._receive_thread = threading.Thread(target=self._run, daemon=True,name="RFID_Receive")
|
||
self._receive_thread.start()
|
||
print("RFID数据接收已启动")
|
||
return True
|
||
else:
|
||
print("RFID数据接收启动失败")
|
||
return False
|
||
|
||
def _run(self):
|
||
"""
|
||
接收线程的主循环，用于接收RFID推送的数据
|
||
"""
|
||
while self._thread_signal:
|
||
|
||
if self._pause_receive:
|
||
time.sleep(1)
|
||
continue
|
||
# 重置数据收集时间
|
||
self._last_collect_time = time.time()
|
||
while time.time() - self._last_collect_time < self.check_time_seconds:
|
||
# 持续接收数据
|
||
try:
|
||
# 使用缓冲区累积接收数据，正确处理固定长度数据包
|
||
received_data = b""
|
||
remaining_bytes = self._buffer_length
|
||
|
||
while remaining_bytes > 0:
|
||
chunk = self.client_socket.recv(remaining_bytes)
|
||
# 检查连接是否关闭
|
||
if not chunk:
|
||
print(f"[连接关闭] 接收过程中连接已关闭，已接收 {len(received_data)} 字节，预期 {self._buffer_length} 字节")
|
||
if not self.CreatConnect():
|
||
print(f"[连接断开] 重连失败，已累计失败 {self._error_count} 次，等待1秒后重试")
|
||
time.sleep(1)
|
||
break
|
||
|
||
received_data += chunk
|
||
remaining_bytes -= len(chunk)
|
||
print(f"[数据接收] 已接收 {len(received_data)}/{self._buffer_length} 字节")
|
||
|
||
# 只有接收到完整的数据才算成功
|
||
if remaining_bytes == 0:
|
||
print(f"[数据接收] 成功接收完整数据包 ({self._buffer_length} 字节)")
|
||
data = received_data # 保存完整的数据包
|
||
self._error_count=0
|
||
else:
|
||
continue # 没有接收到任何数据，继续下一次循环
|
||
except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError) as e:
|
||
# 明确区分连接断开情况
|
||
print(f"[连接断开] 网络连接已断开，错误类型: {type(e).__name__}，错误详情: {e}")
|
||
# 尝试重新连接，但避免频繁重连
|
||
if not self.CreatConnect():
|
||
print(f"[连接断开] 重连失败，已累计失败 {self._error_count} 次，等待1秒后重试")
|
||
time.sleep(1)
|
||
continue # 继续下一次循环
|
||
except socket.timeout:
|
||
# 明确区分超时情况
|
||
print(f"[超时] 接收数据超时，当前时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
# 超时不重置连接，继续等待
|
||
continue # 超时继续下一次循环
|
||
except Exception as e:
|
||
# 记录其他类型的异常
|
||
print(f"[其他错误] RFID数据接收错误，错误类型: {type(e).__name__}，错误详情: {e}")
|
||
# 判断是否可能是连接问题
|
||
if isinstance(e, (OSError, IOError)) and '远程主机强迫关闭了一个现有的连接' in str(e):
|
||
print("[其他错误] 错误原因疑似连接断开，将尝试重新连接")
|
||
self.CreatConnect()
|
||
time.sleep(1)
|
||
continue # 其他错误继续下一次循环
|
||
|
||
if data:
|
||
loc_str = command_hex.parse_user_data_hex(data)
|
||
print(f"收到RFID推送数据: {binascii.hexlify(data).decode()}")
|
||
if loc_str:
|
||
# 将数据添加到缓冲区
|
||
with self._data_lock:
|
||
if self._filter_value == loc_str:
|
||
continue
|
||
self._data_buffer.append(loc_str)
|
||
|
||
self._pause_receive = True
|
||
self._process_collected_data()
|
||
|
||
|
||
def _process_collected_data(self):
|
||
"""
|
||
处理收集到的数据，过滤掉指定值，并调用回调函数
|
||
"""
|
||
with self._data_lock:
|
||
# 如果缓冲区有数据
|
||
if self._data_buffer and self._callback:
|
||
try:
|
||
|
||
# 统计数据缓冲区中各字符串出现的次数
|
||
# 这里self._data_buffer是字符串列表，Counter会统计每个完整字符串出现的次数
|
||
counter = Counter(self._data_buffer)
|
||
|
||
# 处理空缓冲区情况，避免IndexError
|
||
if counter:
|
||
# 获取出现次数最多的字符串及其出现次数
|
||
# most_common(1)返回[(most_common_string, count)]
|
||
most_common_string, count = counter.most_common(1)[0]
|
||
print(f"出现次数最多的字符串是: '{most_common_string}', 出现次数: {count}")
|
||
|
||
# 使用出现次数最多的字符串作为结果传递给回调
|
||
self._callback(most_common_string)
|
||
else:
|
||
# 空缓冲区情况
|
||
print("数据缓冲区为空")
|
||
self._callback(None)
|
||
|
||
print(f"收集了{len(self._data_buffer)}条RFID数据，过滤后数据为{most_common_string}")
|
||
except Exception as e:
|
||
print(f"回调函数执行错误: {e}")
|
||
finally:
|
||
# 清空缓冲区
|
||
self._data_buffer.clear()
|
||
else:
|
||
print("未接收到数据")
|
||
# self._callback(None)
|
||
|
||
def __del__(self):
|
||
"""
|
||
析构函数，确保在对象销毁时关闭连接
|
||
"""
|
||
if self.client_socket:
|
||
self.client_socket.close()
|
||
self.client_socket = None
|
||
|
||
def stop_receiver(self):
|
||
"""
|
||
停止接收RFID推送的数据
|
||
"""
|
||
self._thread_signal = False
|
||
if self._receive_thread:
|
||
self._receive_thread.join(timeout=2.0)
|
||
print("RFID数据接收已停止")
|
||
# 关闭连接
|
||
if self.client_socket:
|
||
self.client_socket.close()
|
||
self.client_socket = None
|
||
|
||
|
||
|