hookvision-rk/hookvision/rfid_client.py

import os
import socket
import subprocess
import threading
import time
import logging
from typing import Callable, Optional

logger = logging.getLogger(__name__)

_MIN_BACKOFF = 1
_MAX_BACKOFF = 60

class RFIDClient:
    def __init__(self, config, on_tag_callback: Callable[[Optional[str]], None],
                 on_state_changed=None, tag_offset: int = -6, tag_length: int = 6, hex_output: bool = True,
                 valid_codes: Optional[set] = None):
        self.config = config
        self.on_tag_callback = on_tag_callback
        self.tag_offset = tag_offset
        self.tag_length = tag_length
        self.hex_output = hex_output
        self.valid_codes = valid_codes  # 允许的卡号集合，None 表示不过滤
        self.sock: Optional[socket.socket] = None
        self.running = False
        self._online: Optional[bool] = None
        self._last_tag: Optional[str] = None
        self._stable_count = 0
        self._last_time = time.time()
        self._last_receive_time = time.time()
        self._backoff = _MIN_BACKOFF
        self._last_connect_attempt = 0
        self.thread = None
        self.buffer = b""
        self.on_state_changed = on_state_changed
        self._ping_interval = 5        # 周期 ping 间隔（秒）
        self._ping_threshold = 3       # 连续 ping 失败多少次判定离线
        self._ping_failures = 0
        self._ping_running = False
        self._ping_thread = None
        self._start_ping_monitor()

    def _start_ping_monitor(self):
        """启动后台周期 ping，用于检测读卡器设备是否在线（与 TCP 数据连接无关）"""
        if self._ping_thread and self._ping_thread.is_alive():
            return
        self._ping_running = True
        self._ping_thread = threading.Thread(target=self._ping_loop, daemon=True)
        self._ping_thread.start()

    def _ping_once(self) -> bool:
        try:
            if os.name == 'nt':
                cmd = ['ping', '-n', '1', '-w', '1000', self.config.ip]
                flags = 0x08000000   # CREATE_NO_WINDOW，避免弹控制台窗口
            else:
                cmd = ['ping', '-c', '1', '-W', '1', self.config.ip]
                flags = 0
            r = subprocess.run(cmd, capture_output=True, timeout=3, creationflags=flags)
            return r.returncode == 0
        except Exception:
            return False

    def _ping_loop(self):
        while self._ping_running:
            if self._ping_once():
                self._ping_failures = 0
                self._notify_state(True)
            else:
                self._ping_failures += 1
                if self._ping_failures >= self._ping_threshold:
                    self._notify_state(False)
            time.sleep(self._ping_interval)

    def _notify_state(self, online: bool):
        """仅在上线/离线状态发生切换时通知上层，避免重试期间重复报警"""
        if online == self._online:
            return
        self._online = online
        if self.on_state_changed:
            self.on_state_changed(online)

    def connect(self) -> bool:
        """尝试建立连接，若已连接则直接返回 True，否则尝试重连（有间隔限制）"""
        if self.sock:
            return True
        # 避免频繁重连
        if time.time() - self._last_connect_attempt < self._backoff:
            return False
        self._last_connect_attempt = time.time()
        try:
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.sock.settimeout(3)   # 设置接收超时，避免阻塞
            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
            if hasattr(socket, 'TCP_KEEPIDLE'):
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 5)
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 3)
                self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)
            self.sock.connect((self.config.ip, self.config.port))
            self.running = True
            self.thread = threading.Thread(target=self._read_loop, daemon=True)
            self.thread.start()
            self._last_receive_time = time.time()
            self._backoff = _MIN_BACKOFF
            logger.info(f"RFID连接成功 {self.config.ip}:{self.config.port}")
            self._notify_state(True)
            return True
        except Exception as e:
            self._backoff = min(self._backoff * 2, _MAX_BACKOFF)
            logger.error(f"RFID连接失败，下次{self._backoff}s后重试: {e}")
            self.sock = None
            self._notify_state(False)
            return False

    def disconnect(self):
        """主动断开连接"""
        self.running = False
        self._ping_running = False
        sock = self.sock
        self.sock = None
        if sock:
            try:
                sock.close()
            except OSError:
                pass
        logger.info("RFID已断开")
        self._notify_state(False)
        self.on_tag_callback(None)   # 通知上层

    def is_connected(self) -> bool:
        """检查连接是否正常"""
        return self.sock is not None and self.running

    def _read_loop(self):
        while self.running:
            sock = self.sock
            if sock is None:
                break
            try:
                data = sock.recv(1024)
                if not data:
                    break
                self._last_receive_time = time.time()   # 收到任何数据都更新时间
                self.buffer += data
                if len(self.buffer) > 65536:
                    self.buffer = self.buffer[-32768:]
                while True:
                    start = self.buffer.find(b"CT")
                    if start == -1:
                        break
                    next_start = self.buffer.find(b"CT", start + 2)
                    if next_start == -1:
                        break
                    packet = self.buffer[start:next_start]
                    self.buffer = self.buffer[next_start:]
                    self._process_packet(packet)
            except socket.timeout:
                continue   # 超时不影响循环
            except OSError:
                break
            except Exception as e:
                logger.error(f"RFID读取错误: {e}")
                break
        # 退出循环（读卡器主动断开或网络异常），静默清理并快速重连。
        # 该读卡器会周期性主动断开连接，属正常行为，在线状态由周期 ping 判定。
        logger.info("RFID读取循环结束，将触发重连")
        self.running = False
        # 先重置重连时间戳，避免 check_idle 在清理完成前提前重连（旧线程误报）
        self._last_connect_attempt = time.time()
        self._backoff = _MIN_BACKOFF
        sock = self.sock
        if sock is not None:
            self.sock = None
            try:
                sock.close()
            except OSError:
                pass
        # 不通知上层：读卡器周期主动断开会话属正常行为，卡号仍有效，
        # 卡离开只由 check_idle 空闲超时判定

    def _process_packet(self, packet):
        hex_str = packet.hex()
        logger.debug(f"RFID包 长度 {len(packet)} 十六进制: {hex_str}")

        offset = self.tag_offset
        length = self.tag_length
        if offset < 0:
            offset = len(packet) + offset
        if offset < 0 or offset + length > len(packet):
            logger.debug("提取范围越界，跳过")
            return

        tag_bytes = packet[offset:offset + length]
        if self.hex_output:
            tag = tag_bytes.hex().upper()
        else:
            try:
                tag = tag_bytes.decode('ascii')
            except UnicodeDecodeError:
                tag = tag_bytes.hex().upper()

        if len(tag) < 4 or tag.strip('0') == '':
            logger.debug("无效卡号（空或全0），忽略")
            return

        # 过滤不在配置中的卡号（读卡器同时读到多张卡时，只保留配置的工位卡）
        if self.valid_codes and tag not in self.valid_codes:
            logger.debug(f"卡号 {tag} 不在配置列表中，过滤")
            return

        logger.debug(f"提取卡号: {tag}")
        self._process_tag(tag)

    def _process_tag(self, tag):
        now = time.time()
        if tag == self._last_tag:
            self._stable_count += 1
        else:
            self._last_tag = tag
            self._stable_count = 1
        self._last_time = now
        logger.debug(f"稳定: 标签 {tag}, 计数 {self._stable_count}/{self.config.stability_count}")
        if self._stable_count >= self.config.stability_count:
            logger.info(f"触发稳定标签: {tag}")
            self.on_tag_callback(tag)
            self._stable_count = 0

    def check_idle(self):
        """
        外部定时（每秒）调用的维护方法：
        1. 检查空闲超时（卡离开）
        2. 检查连接状态，若断开或长时间无数据则自动重连
        """
        # 空闲超时：仅在连接正常时才触发，避免断开期间误判
        if self.sock is not None and self._last_tag and (time.time() - self._last_time) > self.config.stability_time:
            logger.info("空闲超时，清空状态")
            self._last_tag = None
            self._stable_count = 0
            self.on_tag_callback(None)

        # 连接状态维护
        if self.sock is None:
            # 如果 socket 已断开，尝试重连
            if time.time() - self._last_connect_attempt > self._backoff:
                logger.debug("RFID socket为None，尝试重连")
                self.connect()
        else:
            # socket 存在，检查是否长时间无数据（可能连接已假死）
            idle_time = time.time() - self._last_receive_time
            if idle_time > self.config.stability_time * 2:
                # 超过两倍空闲时间无数据，主动断开重连
                logger.warning(f"RFID连接假死（{idle_time:.1f}s无数据），主动重连")
                self.running = False
                sock = self.sock
                self.sock = None
                if sock:
                    try:
                        sock.close()
                    except OSError:
                        pass