From 6fba8a981aa91b84bea027081edae87e0ce96d27 Mon Sep 17 00:00:00 2001 From: sladro Date: Thu, 13 Nov 2025 11:38:45 +0800 Subject: [PATCH] feat: add alarm workflow and packaging support Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- main.py | 46 ++++++++++-- src/output/alarm_manager.py | 127 ++++++++++++++++++++++++++++++++++ src/output/logger.py | 83 +++++++++++++++++++++- tools/roi_calibration_tool.py | 8 ++- 4 files changed, 255 insertions(+), 9 deletions(-) create mode 100644 src/output/alarm_manager.py diff --git a/main.py b/main.py index 6029e4d..2433b6f 100644 --- a/main.py +++ b/main.py @@ -15,7 +15,11 @@ os.environ['OPENCV_LOG_LEVEL'] = 'ERROR' os.environ['OPENCV_VIDEOIO_DEBUG'] = '0' # 添加项目根目录到Python路径 -project_root = Path(__file__).parent +if getattr(sys, 'frozen', False): + project_root = Path(getattr(sys, '_MEIPASS')) +else: + project_root = Path(__file__).parent + sys.path.insert(0, str(project_root)) from src.camera.opencv_camera import OpenCVCamera @@ -23,6 +27,7 @@ from src.preprocessing.image_enhancer import ImageEnhancer from src.roi_detection.led_detector import LEDDetector from src.output.result_formatter import ResultFormatter from src.output.logger import LEDLogger +from src.output.alarm_manager import AlarmManager from src.ui.tech_ui import TechUI @@ -47,6 +52,7 @@ class YantaiVisionXSystem: self.led_detector = None self.formatter = ResultFormatter() self.logger = LEDLogger() + self.alarm_manager = AlarmManager(self.logger) self.tech_ui = TechUI() # 状态变量 @@ -79,6 +85,11 @@ class YantaiVisionXSystem: str(algorithm_config_path) ) self.logger.log_info("LED检测器初始化完成") + + if self.led_detector and hasattr(self.led_detector, 'algorithm_config'): + self.alarm_manager.update_config( + self.led_detector.algorithm_config.get('alarm') + ) # 3. 初始化摄像头 if camera_config is None: @@ -139,9 +150,23 @@ class YantaiVisionXSystem: # 记录结果 self.logger.log_detection_result(detection_result) + + alarm_result = self.alarm_manager.process_detection(detection_result) + for event in alarm_result.events: + self.logger.log_alarm_event(event) + for roi_name in alarm_result.recoveries: + self.logger.log_recovery_event( + roi_name, + detection_result.frame_count, + detection_result.timestamp + ) # 保存结果 - if save_results and detection_result.frame_count % 30 == 0: # 每30帧保存一次 + if ( + save_results + and self.logger.save_results_enabled + and detection_result.frame_count % self.logger.save_results_interval == 0 + ): self.logger.save_result_to_file(detection_result) # 显示结果 @@ -190,7 +215,7 @@ class YantaiVisionXSystem: output_path = Path(output_dir) output_path.mkdir(exist_ok=True) - results = [] + results = [] if self.logger.save_results_enabled else None self.logger.log_info(f"开始处理视频: {video_path}") @@ -207,11 +232,22 @@ class YantaiVisionXSystem: # LED检测 detection_result = self.led_detector.detect_leds(enhanced_frame) - results.append(detection_result) + if results is not None: + results.append(detection_result) # 记录结果 if detection_result.frame_count % 100 == 0: self.logger.log_detection_result(detection_result) + + alarm_result = self.alarm_manager.process_detection(detection_result) + for event in alarm_result.events: + self.logger.log_alarm_event(event) + for roi_name in alarm_result.recoveries: + self.logger.log_recovery_event( + roi_name, + detection_result.frame_count, + detection_result.timestamp + ) # 显示结果 if self.display_enabled: @@ -350,4 +386,4 @@ def main(): if __name__ == '__main__': - exit(main()) + raise SystemExit(main()) diff --git a/src/output/alarm_manager.py b/src/output/alarm_manager.py new file mode 100644 index 0000000..307b410 --- /dev/null +++ b/src/output/alarm_manager.py @@ -0,0 +1,127 @@ +"""报警管理模块 +负责处理LED检测结果中的异常灯状态，触发报警并记录事件信息 +""" + +from __future__ import annotations + +import sys +import time +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Tuple + +from ..roi_detection.threshold_detector import LEDState +from ..roi_detection.led_detector import LEDDetectionResult + + +@dataclass +class AlarmEvent: + """报警事件信息""" + + timestamp: float + frame_count: int + roi_names: List[str] + stability: Dict[str, float] + confidence: Dict[str, float] + + +@dataclass +class AlarmProcessingResult: + """报警处理后的输出结果""" + + events: List[AlarmEvent] = field(default_factory=list) + recoveries: List[str] = field(default_factory=list) + + +class AlarmManager: + """管理LED报警事件""" + + def __init__(self, logger=None, config: Optional[Dict[str, float]] = None): + self.logger = logger + self.config = self._load_config(config) + self._active_alerts: Dict[str, float] = {} + self._last_notification_time: float = 0.0 + + def _load_config(self, config: Optional[Dict[str, float]]) -> Dict[str, float]: + defaults = { + "enabled": True, + "stability_threshold": 0.6, + "confidence_threshold": 0.5, + "cooldown_seconds": 30.0, + "sound_enabled": True, + } + if not config: + return defaults + merged = defaults.copy() + merged.update({k: v for k, v in config.items() if v is not None}) + return merged + + def update_config(self, config: Optional[Dict[str, float]]) -> None: + if config: + self.config.update({k: v for k, v in config.items() if v is not None}) + + def process_detection(self, result: LEDDetectionResult) -> AlarmProcessingResult: + if not self.config.get("enabled", True): + return AlarmProcessingResult() + + events: List[AlarmEvent] = [] + recoveries: List[str] = [] + + stability_threshold = self.config.get("stability_threshold", 0.6) + confidence_threshold = self.config.get("confidence_threshold", 0.5) + + new_alerts: Dict[str, Tuple[float, float]] = {} + + for roi_name, stable_result in result.stable_states.items(): + led_state = stable_result.led_state + + if ( + led_state == LEDState.OFF + and stable_result.stability >= stability_threshold + and stable_result.confidence >= confidence_threshold + ): + new_alerts[roi_name] = (stable_result.stability, stable_result.confidence) + else: + if roi_name in self._active_alerts and led_state == LEDState.ON: + recoveries.append(roi_name) + + new_roi_names = [roi for roi in new_alerts.keys() if roi not in self._active_alerts] + + for roi_name in recoveries: + self._active_alerts.pop(roi_name, None) + + if new_roi_names: + event = AlarmEvent( + timestamp=result.timestamp, + frame_count=result.frame_count, + roi_names=new_roi_names, + stability={roi: new_alerts[roi][0] for roi in new_roi_names}, + confidence={roi: new_alerts[roi][1] for roi in new_roi_names}, + ) + events.append(event) + for roi in new_roi_names: + self._active_alerts[roi] = result.timestamp + self._trigger_notification() + + return AlarmProcessingResult(events=events, recoveries=recoveries) + + def _trigger_notification(self) -> None: + if not self.config.get("sound_enabled", True): + return + + cooldown = self.config.get("cooldown_seconds", 30.0) + current_time = time.time() + if current_time - self._last_notification_time < cooldown: + return + + try: + if sys.platform.startswith("win"): + import winsound + + winsound.MessageBeep(winsound.MB_ICONEXCLAMATION) + else: + print("\a", end="", flush=True) + except Exception as exc: # pragma: no cover - 平台相关分支 + if self.logger: + self.logger.log_warning(f"报警提示触发失败: {exc}") + + self._last_notification_time = current_time diff --git a/src/output/logger.py b/src/output/logger.py index 6f33647..26a8ec1 100644 --- a/src/output/logger.py +++ b/src/output/logger.py @@ -7,12 +7,15 @@ import os import logging import json from datetime import datetime -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, TYPE_CHECKING from pathlib import Path from ..roi_detection.led_detector import LEDDetectionResult from .result_formatter import ResultFormatter +if TYPE_CHECKING: + from .alarm_manager import AlarmEvent + class LEDLogger: """ @@ -34,8 +37,17 @@ class LEDLogger: self.log_level = logging_config.get('log_level', 'INFO') self.log_file = logging_config.get('log_file', 'logs/led_detection.log') + self.alarm_log_file = logging_config.get('alarm_log_file', 'logs/alarm_events.log') self.save_debug_images = logging_config.get('save_debug_images', False) self.debug_image_path = logging_config.get('debug_image_path', 'debug/') + self.log_frame_details = logging_config.get('log_frame_details', False) + self.frame_log_interval = self._to_positive_int( + logging_config.get('frame_log_interval'), default=60 + ) + self.save_results_enabled = logging_config.get('save_results_enabled', False) + self.save_results_interval = self._to_positive_int( + logging_config.get('save_results_interval'), default=120 + ) # 创建目录 self._create_directories() @@ -58,8 +70,13 @@ class LEDLogger: 'logging': { 'log_level': 'INFO', 'log_file': 'logs/led_detection.log', + 'alarm_log_file': 'logs/alarm_events.log', 'save_debug_images': False, - 'debug_image_path': 'debug/' + 'debug_image_path': 'debug/', + 'log_frame_details': False, + 'frame_log_interval': 60, + 'save_results_enabled': False, + 'save_results_interval': 120 } } @@ -71,6 +88,10 @@ class LEDLogger: log_dir = Path(self.log_file).parent log_dir.mkdir(parents=True, exist_ok=True) + # 创建报警日志目录 + alarm_dir = Path(self.alarm_log_file).parent + alarm_dir.mkdir(parents=True, exist_ok=True) + # 创建调试图像目录 if self.save_debug_images: debug_dir = Path(self.debug_image_path) @@ -118,6 +139,12 @@ class LEDLogger: """ self.total_frames += 1 + if not self.log_frame_details: + return + + if self.total_frames % self.frame_log_interval != 0: + return + # 基本信息 self.logger.info(f"帧#{result.frame_count} 处理完成 - " f"用时: {result.processing_time*1000:.1f}ms") @@ -175,6 +202,58 @@ class LEDLogger: debug_msg: 调试消息 """ self.logger.debug(debug_msg) + + @staticmethod + def _to_positive_int(value: Optional[Any], default: int) -> int: + try: + result = int(value) + return result if result > 0 else default + except (TypeError, ValueError): + return default + + def log_alarm_event(self, event: 'AlarmEvent') -> None: + """记录报警事件""" + timestamp_text = datetime.fromtimestamp(event.timestamp).isoformat() + message = (f"报警触发 - 帧#{event.frame_count} 受影响LED: " + f"{', '.join(event.roi_names)}") + self.logger.warning(message) + + record = { + 'timestamp': timestamp_text, + 'frame': event.frame_count, + 'rois': event.roi_names, + 'stability': event.stability, + 'confidence': event.confidence + } + + try: + with open(self.alarm_log_file, 'a', encoding='utf-8') as f: + json.dump(record, f, ensure_ascii=False) + f.write('\n') + except Exception as exc: + self.logger.error(f"写入报警日志失败: {exc}") + + def log_recovery_event(self, roi_name: str, frame_count: int, + timestamp: Optional[float] = None) -> None: + """记录恢复事件""" + ts = timestamp if timestamp is not None else datetime.now().timestamp() + timestamp_text = datetime.fromtimestamp(ts).isoformat() + + self.logger.info(f"LED恢复正常 - 帧#{frame_count} ROI: {roi_name}") + + record = { + 'timestamp': timestamp_text, + 'frame': frame_count, + 'roi': roi_name, + 'event': 'recovery' + } + + try: + with open(self.alarm_log_file, 'a', encoding='utf-8') as f: + json.dump(record, f, ensure_ascii=False) + f.write('\n') + except Exception as exc: + self.logger.error(f"写入恢复日志失败: {exc}") def save_result_to_file(self, result: LEDDetectionResult, format_type: str = 'json', diff --git a/tools/roi_calibration_tool.py b/tools/roi_calibration_tool.py index 31bc2ab..4f4ee83 100644 --- a/tools/roi_calibration_tool.py +++ b/tools/roi_calibration_tool.py @@ -10,12 +10,16 @@ import numpy as np import yaml import sys import os +from pathlib import Path from typing import List, Tuple, Optional, Dict, Any # 添加项目根目录到路径 -sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +if getattr(sys, 'frozen', False): + base_dir = Path(getattr(sys, '_MEIPASS')) +else: + base_dir = Path(__file__).resolve().parent.parent -from src.roi_detection.roi_manager import ROIManager, ROIRegion +sys.path.insert(0, str(base_dir)) class ROICalibrationTool: