feat: add alarm workflow and packaging support

Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>

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())

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

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', 

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: