完成警告消息对应中文显示

biz/base_frame_processor.py

@@ -15,6 +15,7 @@ from typing import Dict, Any, Callable
 from concurrent.futures import ThreadPoolExecutor
 
 from common import constants
+from common.type_mapping import get_alert_label
 from utils.logger import get_logger
 from utils.hls_utils import get_segments_before_current, parse_segment_info
 
@@ -90,10 +91,10 @@ class BaseFrameProcessorWorker(threading.Thread):
         将 msg 中的 result_type 从数组展开为多个独立的 msg
         
         Args:
-            msg: 原始消息，result_type 为数组
+            msg: 原始消息，result_type 为 action code 字符串数组
             
         Returns:
-            msg 列表，每个 msg 的 result_type 为数组中的单个元素
+            msg 列表，每个 msg 的 result_type 为包含 action_code 和 action_name 的对象
         """
         result_types = msg.get("result_type", [])
         if not isinstance(result_types, list):
@@ -104,9 +105,12 @@ class BaseFrameProcessorWorker(threading.Thread):
             return [msg]
         
         result = []
-        for r_type in result_types:
+        for action_code in result_types:
             new_msg = msg.copy()
-            new_msg["result_type"] = r_type
+            new_msg["result_type"] = {
+                "action_code": action_code,
+                "action_name": get_alert_label(action_code)
+            }
             result.append(new_msg)
         
         return result

biz/checkpoint/checkpoint_biz.py

@@ -371,12 +371,12 @@ class KadianDetector:
 
             # 情况1：通过时间太短 -> Ignore (Too Fast)
             if duration_frames < self.frame_thresh_car_min_duration:
-                print(f"ALARM: Car {car_id} passed too fast -> Regarded as Ignore Checked!")
+                logger.info(f"ALARM: Car {car_id} passed too fast -> Regarded as Ignore Checked!")
                 self.fast_pass_alerts[car_id] = self.current_frame_idx + int(self.ignore_show_seconds * self.fps)
 
             # 情况2：时间够长，但没检查后备箱 -> Unchecked Trunk
             elif not car_info['is_checked']:
-                print(f"ALARM: Car {car_id} left without checking trunk!")
+                logger.info(f"ALARM: Car {car_id} left without checking trunk!")
                 self.unchecked_trunk_alerts[car_id] = self.current_frame_idx + int(
                     self.openTrunk_show_seconds * self.fps)
 

common/type_mapping.py

@@ -0,0 +1,96 @@
+# common/type_mapping.py
+"""
+告警类型映射配置模块
+从 config.yaml 加载告警 code 到 label 的映射关系
+"""
+
+import yaml
+from typing import Dict, Optional
+from utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class TypeMapping:
+    """类型映射类"""
+    
+    def __init__(self, mapping: Dict[str, str], name: str = ""):
+        self._mapping = mapping or {}
+        self._name = name
+    
+    def get(self, code: str, default: Optional[str] = None) -> str:
+        """获取label，支持自定义默认值"""
+        if default is None:
+            default = f"{code}"
+        return self._mapping.get(code, default)
+    
+    def __getitem__(self, code: str) -> str:
+        """支持 [] 语法访问"""
+        return self.get(code)
+    
+    def __contains__(self, code: str) -> bool:
+        """支持 in 操作符"""
+        return code in self._mapping
+    
+    def all(self) -> Dict[str, str]:
+        """获取所有映射"""
+        return self._mapping.copy()
+    
+    def codes(self) -> list:
+        """获取所有code"""
+        return list(self._mapping.keys())
+    
+    def labels(self) -> list:
+        """获取所有label"""
+        return list(self._mapping.values())
+
+
+# 全局映射实例
+_alert_types: Optional[TypeMapping] = None
+
+
+def init_type_mappings(config_path: str = "config.yaml"):
+    """
+    从配置文件初始化类型映射
+    
+    Args:
+        config_path: 配置文件路径
+    """
+    global _alert_types
+    
+    try:
+        with open(config_path, "r", encoding="utf-8") as f:
+            cfg = yaml.safe_load(f)
+        
+        _alert_types = TypeMapping(
+            cfg.get("alert_types", {}),
+            name="告警类型"
+        )
+        
+        logger.info(f"[INFO] Alert type mappings initialized from {config_path}")
+        logger.info(f"  - alert_types: {len(_alert_types.codes())} items")
+        
+    except Exception as e:
+        logger.error(f"[ERROR] Failed to load type mappings from {config_path}: {e}")
+        _alert_types = TypeMapping({}, "告警类型")
+
+
+def alert_types() -> TypeMapping:
+    """获取告警类型映射"""
+    if _alert_types is None:
+        init_type_mappings()
+    return _alert_types
+
+
+def get_alert_label(code: str, default: str = None) -> str:
+    """
+    快捷获取告警类型label
+    
+    Args:
+        code: 告警类型代码
+        default: 默认值，未提供时返回 "未知告警类型(code)"
+        
+    Returns:
+        告警类型中文名称
+    """
+    return alert_types().get(code, default)

config.yaml

@@ -78,3 +78,20 @@ service_groups:
 #        - [0.5, 0.001]
 #        - [1.0, 0.8]
 #        - [0.35, 1.0]
+
+# 告警类型映射 (code -> 中文名称)
+alert_types:
+  # 卡点检测 (checkpoint)
+  "Unchecked Trunk": "未检查后备箱"
+  "Ignore": "漏检"
+  "Nobody": "无人在场"
+  "Only One": "单人单检"
+
+  # 监狱检测 (prison)
+  "prisoner": "带出犯人"
+  "violation": "路线违规"
+  
+  # 监控室检测 (supervision_room)
+  "Playing Phone": "玩手机"
+  "Smoke": "吸烟"
+  "Nobody Checking": "无人在场"