Local-Voice/nfc_manager.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
NFC管理模块
实现NFC读取和角色切换功能
"""

import json
import os
import re
import subprocess
import threading
import time
from pathlib import Path
from typing import Callable, Dict, Optional


class NFCManager:
    """NFC管理器，负责NFC读取和角色切换"""
    
    def __init__(self, characters_dir: str = "characters"):
        self.characters_dir = Path(characters_dir)
        self.uid_to_character = {}
        self.running = False
        self.read_thread = None
        self.current_uid = None
        self.character_switch_callback: Optional[Callable] = None
        
        # 加载角色NFC映射
        self._load_character_mappings()
        
    def _load_character_mappings(self):
        """加载角色NFC映射"""
        self.uid_to_character = {}
        
        if not self.characters_dir.exists():
            print(f"⚠️ 角色目录不存在: {self.characters_dir}")
            return
            
        for config_file in self.characters_dir.glob("*.json"):
            try:
                with open(config_file, 'r', encoding='utf-8') as f:
                    config = json.load(f)
                    
                if 'nfc_uid' in config:
                    uid = config['nfc_uid'].upper().strip()
                    character_name = config_file.stem
                    self.uid_to_character[uid] = character_name
                    print(f"✅ 加载角色映射: {character_name} -> {uid}")
                    
            except Exception as e:
                print(f"❌ 加载角色配置失败 {config_file}: {e}")
                
        if not self.uid_to_character:
            print("⚠️ 未找到任何带NFC UID的角色配置")
            
    def _read_uid(self) -> Optional[str]:
        """读取NFC UID"""
        os.environ['LIBNFC_DRIVER'] = 'pn532_i2c'
        os.environ['LIBNFC_DEVICE'] = 'pn532_i2c:/dev/i2c-1'
        
        try:
            result = subprocess.run(['nfc-list'], capture_output=True, text=True, timeout=10)
            if result.returncode == 0:
                match = re.search(r'UID\s*\(NFCID1\):\s*([0-9A-Fa-f\s]+)', result.stdout)
                if match:
                    return match.group(1).strip().replace(' ', '').upper()
        except subprocess.TimeoutExpired:
            print("⚠️ NFC读取超时")
        except Exception as e:
            print(f"❌ NFC读取失败: {e}")
            
        return None
        
    def _read_loop(self):
        """NFC读取循环"""
        print("🔄 启动NFC读取循环...")
        
        while self.running:
            try:
                uid = self._read_uid()
                
                if uid and uid != self.current_uid:
                    print(f"🎯 检测到NFC卡片: {uid}")
                    
                    if uid in self.uid_to_character:
                        character_name = self.uid_to_character[uid]
                        print(f"🎭 切换角色: {character_name}")
                        
                        # 更新当前UID
                        self.current_uid = uid
                        
                        # 调用角色切换回调
                        if self.character_switch_callback:
                            self.character_switch_callback(character_name)
                    else:
                        print(f"⚠️ 未找到UID对应角色: {uid}")
                        
                        
            except Exception as e:
                print(f"❌ NFC读取循环出错: {e}")
                
            # 等待5秒
            time.sleep(1)
            
    def set_character_switch_callback(self, callback: Callable):
        """设置角色切换回调函数"""
        self.character_switch_callback = callback
        
    def start(self):
        """启动NFC管理器"""
        if self.running:
            print("⚠️ NFC管理器已在运行")
            return
            
        self.running = True
        self.read_thread = threading.Thread(target=self._read_loop, daemon=True)
        self.read_thread.start()
        print("✅ NFC管理器已启动")
        
    def stop(self):
        """停止NFC管理器"""
        if not self.running:
            return
            
        self.running = False
        if self.read_thread:
            self.read_thread.join(timeout=5)
            
        print("🛑 NFC管理器已停止")
        
    def get_current_character(self) -> Optional[str]:
        """获取当前角色"""
        if self.current_uid and self.current_uid in self.uid_to_character:
            return self.uid_to_character[self.current_uid]
        return None
        
    def reload_mappings(self):
        """重新加载角色映射"""
        print("🔄 重新加载角色NFC映射...")
        self._load_character_mappings()
        
    def __del__(self):
        """析构函数"""
        self.stop()

# 单例模式
_nfc_manager_instance = None

def get_nfc_manager(characters_dir: str = "characters") -> NFCManager:
    """获取NFC管理器单例"""
    global _nfc_manager_instance
    if _nfc_manager_instance is None:
        _nfc_manager_instance = NFCManager(characters_dir)
    return _nfc_manager_instance

if __name__ == "__main__":
    # 测试NFC管理器
    def on_character_switch(character_name):
        print(f"🎭 角色切换回调: {character_name}")
        
    nfc_manager = get_nfc_manager()
    nfc_manager.set_character_switch_callback(on_character_switch)
    
    try:
        nfc_manager.start()
        print("🔄 NFC管理器运行中，按Ctrl+C退出...")
        
        while True:
            time.sleep(1)
            
    except KeyboardInterrupt:
        print("\n👋 用户中断")
    finally:
        nfc_manager.stop()