""" 检测器注册中心 管理所有已注册的检测器 """ from typing import Dict, Type, Optional, List, Any import logging import threading import json import os class DetectorRegistry: """检测器注册中心""" _instance = None _initialized = False def __new__(cls): """单例模式""" if cls._instance is None: cls._instance = super(DetectorRegistry, cls).__new__(cls) return cls._instance def __init__(self): """初始化注册中心""" if not self._initialized: self.detectors: Dict[str, Type] = {} self.detector_instances: Dict[str, object] = {} self.logger = logging.getLogger(__name__) # 参数缓存管理系统 self.params_cache: Dict[str, Dict[str, Any]] = {} # 格式: {detector_name: params_dict} self.cache_lock = threading.Lock() # 保护缓存的并发访问 self._initialized = True def register(self, detector_class: Type, overwrite: bool = False): """ 注册检测器类 Args: detector_class: 检测器类，必须继承BaseDetector overwrite: 是否覆盖已存在的检测器 """ from .base_detector import BaseDetector if not issubclass(detector_class, BaseDetector): raise ValueError(f"检测器类必须继承BaseDetector: {detector_class}") # 创建临时实例获取名称（不传递参数，让检测器使用自己的默认参数） temp_instance = detector_class() detector_name = temp_instance.name if detector_name in self.detectors and not overwrite: raise ValueError(f"检测器已存在: {detector_name}，使用overwrite=True来覆盖") self.detectors[detector_name] = detector_class self.logger.info(f"注册检测器: {detector_name}") def create_detector(self, detector_name: str, *args, **kwargs) -> Optional[object]: """ 创建检测器实例 Args: detector_name: 检测器名称 *args, **kwargs: 传递给检测器构造函数的参数 Returns: 检测器实例，如果不存在则返回None """ if detector_name not in self.detectors: self.logger.error(f"未找到检测器: {detector_name}") return None try: detector_class = self.detectors[detector_name] instance = detector_class(*args, **kwargs) return instance except Exception as e: self.logger.error(f"创建检测器实例失败 {detector_name}: {e}") return None def get_detector(self, detector_name: str, create_if_not_exists: bool = True) -> Optional[object]: """ 获取检测器实例（单例模式） Args: detector_name: 检测器名称 create_if_not_exists: 如果实例不存在是否创建 Returns: 检测器实例，如果不存在则返回None """ if detector_name in self.detector_instances: return self.detector_instances[detector_name] if create_if_not_exists: instance = self.create_detector(detector_name) if instance: self.detector_instances[detector_name] = instance return instance return None def list_detectors(self) -> Dict[str, Type]: """ 获取所有已注册的检测器 Returns: 检测器名称到类的映射字典 """ return self.detectors.copy() def get_detector_info(self, detector_name: str) -> Optional[Dict]: """ 获取检测器信息 Args: detector_name: 检测器名称 Returns: 检测器信息字典，如果不存在则返回None """ detector = self.get_detector(detector_name, create_if_not_exists=False) if detector: return detector.get_info() # 如果实例不存在，尝试创建临时实例获取信息 if detector_name in self.detectors: try: temp_detector = self.create_detector(detector_name) if temp_detector: return temp_detector.get_info() except Exception as e: self.logger.error(f"获取检测器信息失败 {detector_name}: {e}") return None def unregister(self, detector_name: str): """ 注销检测器 Args: detector_name: 检测器名称 """ if detector_name in self.detectors: del self.detectors[detector_name] if detector_name in self.detector_instances: # 清理资源 instance = self.detector_instances[detector_name] if hasattr(instance, 'cleanup'): instance.cleanup() del self.detector_instances[detector_name] self.logger.info(f"注销检测器: {detector_name}") def get_registered_detectors(self): """ 获取所有已注册的检测器 Returns: Dict[str, Type]: 检测器名称到检测器类的映射 """ return self.detectors.copy() def get_detector_functions(self, detector_name: str) -> Dict[str, str]: """ 获取指定检测器的可用函数列表 Args: detector_name: 检测器名称 Returns: Dict[str, str]: 函数名称到描述的映射 """ if detector_name not in self.detectors: return {} detector_instance = self.get_detector(detector_name) if detector_instance and hasattr(detector_instance, 'get_available_functions'): return detector_instance.get_available_functions() return {'detect': '主检测函数'} def call_detector_function(self, detector_name: str, function_name: str, image, params: Dict[str, Any] = None) -> Dict[str, Any]: """ 调用检测器的指定函数 Args: detector_name: 检测器名称 function_name: 函数名称 image: 输入图像 params: 外部传入的参数（可选），会与缓存参数合并 Returns: Dict: 检测结果 """ detector_instance = self.get_detector(detector_name) if not detector_instance: raise ValueError(f"检测器不存在: {detector_name}") # 获取缓存的参数 cached_params = self.get_detector_params(detector_name) # 合并外部参数（外部参数优先级更高） if params: final_params = cached_params.copy() final_params.update(params) else: final_params = cached_params if hasattr(detector_instance, 'call_function'): return detector_instance.call_function(function_name, image, final_params) elif function_name == 'detect': return detector_instance.detect(image, final_params) else: raise ValueError(f"检测器 {detector_name} 不支持函数: {function_name}") def list_detector_names(self): """ 获取所有已注册检测器的名称列表 Returns: List[str]: 检测器名称列表 """ return list(self.detectors.keys()) def clear(self): """清空所有注册的检测器""" # 清理所有实例 for instance in self.detector_instances.values(): if hasattr(instance, 'cleanup'): instance.cleanup() self.detectors.clear() self.detector_instances.clear() # 清空参数缓存 with self.cache_lock: self.params_cache.clear() self.logger.info("清空所有检测器注册") def load_detector_params(self, detector_name: str, config_file_path: str = None) -> Dict[str, Any]: """ 加载检测器参数到缓存 Args: detector_name: 检测器名称 config_file_path: 配置文件路径，如果为None则使用默认路径 Returns: Dict[str, Any]: 合并后的参数字典 """ detector_instance = self.get_detector(detector_name) if not detector_instance: self.logger.error(f"检测器不存在: {detector_name}") return {} # 获取默认参数 default_params = detector_instance.get_default_parameters() # 如果未指定配置文件路径，尝试生成默认路径 if config_file_path is None: if hasattr(detector_instance, 'get_config_filename'): config_filename = detector_instance.get_config_filename() else: config_filename = f"{detector_name}_params.json" config_file_path = os.path.join("config", config_filename) # 加载JSON配置参数并合并 merged_params = default_params.copy() if os.path.exists(config_file_path): try: with open(config_file_path, 'r', encoding='utf-8') as f: json_params = json.load(f) merged_params.update(json_params) self.logger.info(f"已加载检测器配置: {config_file_path}") except Exception as e: self.logger.warning(f"加载检测器配置失败: {e}，使用默认参数") else: self.logger.info(f"配置文件不存在，使用默认参数: {config_file_path}") # 缓存参数 with self.cache_lock: self.params_cache[detector_name] = merged_params return merged_params def get_detector_params(self, detector_name: str, reload: bool = False) -> Dict[str, Any]: """ 获取检测器参数（从缓存或重新加载） Args: detector_name: 检测器名称 reload: 是否强制重新加载参数 Returns: Dict[str, Any]: 检测器参数字典 """ with self.cache_lock: if not reload and detector_name in self.params_cache: return self.params_cache[detector_name].copy() # 缓存中没有或需要重新加载，则从文件加载 return self.load_detector_params(detector_name) def update_detector_params(self, detector_name: str, params: Dict[str, Any]) -> bool: """ 更新检测器参数缓存 Args: detector_name: 检测器名称 params: 新的参数字典 Returns: bool: 更新是否成功 """ if detector_name not in self.detectors: self.logger.error(f"检测器不存在: {detector_name}") return False # 调试信息：显示即将更新的参数 if detector_name == "极耳完整流程检测器" and "极耳裁切模型参数" in params: new_model_path = params["极耳裁切模型参数"]["模型路径"] self.logger.info(f"🔄 即将更新缓存，新的极耳裁切模型路径: {new_model_path}") # 显示旧的缓存值 if detector_name in self.params_cache and "极耳裁切模型参数" in self.params_cache[detector_name]: old_model_path = self.params_cache[detector_name]["极耳裁切模型参数"]["模型路径"] self.logger.info(f"📦 缓存中当前模型路径: {old_model_path}") with self.cache_lock: self.params_cache[detector_name] = params.copy() # 验证缓存更新 if detector_name == "极耳完整流程检测器" and "极耳裁切模型参数" in self.params_cache[detector_name]: cached_model_path = self.params_cache[detector_name]["极耳裁切模型参数"]["模型路径"] self.logger.info(f"✅ 缓存已更新，极耳裁切模型路径: {cached_model_path}") self.logger.info(f"已更新检测器参数缓存: {detector_name}") return True def save_detector_params(self, detector_name: str, config_file_path: str = None, params: Dict[str, Any] = None) -> bool: """ 保存检测器参数到文件 Args: detector_name: 检测器名称 config_file_path: 配置文件路径，如果为None则使用默认路径 params: 要保存的参数字典，如果为None则使用缓存中的参数 Returns: bool: 保存是否成功 """ # 确定要保存的参数 if params is None: with self.cache_lock: if detector_name not in self.params_cache: self.logger.error(f"参数缓存中没有检测器 {detector_name} 的参数") return False params = self.params_cache[detector_name].copy() # 确定保存路径 if config_file_path is None: detector_instance = self.get_detector(detector_name) if hasattr(detector_instance, 'get_config_filename'): config_filename = detector_instance.get_config_filename() else: config_filename = f"{detector_name}_params.json" config_file_path = os.path.join("config", config_filename) # 确保目录存在 os.makedirs(os.path.dirname(config_file_path), exist_ok=True) # 调试信息：显示即将保存的参数 if detector_name == "极耳完整流程检测器" and "极耳裁切模型参数" in params: model_path = params["极耳裁切模型参数"]["模型路径"] self.logger.info(f"🔧 即将保存参数，极耳裁切模型路径: {model_path}") # 保存到文件 try: # 读取当前文件内容进行对比 old_content = {} if os.path.exists(config_file_path): try: with open(config_file_path, 'r', encoding='utf-8') as f: old_content = json.load(f) if detector_name == "极耳完整流程检测器" and "极耳裁切模型参数" in old_content: old_model_path = old_content["极耳裁切模型参数"]["模型路径"] self.logger.info(f"📖 文件中当前模型路径: {old_model_path}") except Exception as e: self.logger.warning(f"读取旧配置文件失败: {e}") # 写入新内容 with open(config_file_path, 'w', encoding='utf-8') as f: json.dump(params, f, indent=2, ensure_ascii=False) # 验证写入结果 with open(config_file_path, 'r', encoding='utf-8') as f: verify_content = json.load(f) if detector_name == "极耳完整流程检测器" and "极耳裁切模型参数" in verify_content: verify_model_path = verify_content["极耳裁切模型参数"]["模型路径"] self.logger.info(f"✅ 验证保存结果，极耳裁切模型路径: {verify_model_path}") # 检查是否真的改变了 if old_content and "极耳裁切模型参数" in old_content: old_model_path = old_content["极耳裁切模型参数"]["模型路径"] if verify_model_path != old_model_path: self.logger.info(f"🎯 路径已更新: {old_model_path} → {verify_model_path}") else: self.logger.warning(f"⚠️ 路径未改变: {verify_model_path}") self.logger.info(f"已保存检测器参数: {config_file_path}") return True except Exception as e: self.logger.error(f"保存检测器参数失败: {e}") return False def get_params_for_tree_viewer(self, detector_name: str) -> Dict[str, Any]: """ 获取用于TreeViewer的参数（返回缓存的引用，允许直接修改） Args: detector_name: 检测器名称 Returns: Dict[str, Any]: 参数字典的引用 """ with self.cache_lock: if detector_name not in self.params_cache: # 如果缓存中没有，先加载 self.load_detector_params(detector_name) # 返回缓存的引用，允许TreeViewer直接修改 return self.params_cache.get(detector_name, {}) # 全局注册中心实例 detector_registry = DetectorRegistry()