From 8bcb586b536cadd8846eef11a91b6dc0f57bd72c Mon Sep 17 00:00:00 2001 From: zqc <835569504@qq.com> Date: Thu, 5 Mar 2026 16:05:58 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=E8=AD=A6=E5=91=8A=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E5=AF=B9=E5=BA=94=E4=B8=AD=E6=96=87=E6=98=BE=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- biz/base_frame_processor.py | 12 ++-- biz/checkpoint/checkpoint_biz.py | 4 +- common/type_mapping.py | 96 ++++++++++++++++++++++++++++++++ config.yaml | 17 ++++++ 4 files changed, 123 insertions(+), 6 deletions(-) create mode 100644 common/type_mapping.py diff --git a/biz/base_frame_processor.py b/biz/base_frame_processor.py index 5e68a6f..4a43084 100644 --- a/biz/base_frame_processor.py +++ b/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 diff --git a/biz/checkpoint/checkpoint_biz.py b/biz/checkpoint/checkpoint_biz.py index d024c03..1dc18a5 100644 --- a/biz/checkpoint/checkpoint_biz.py +++ b/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) diff --git a/common/type_mapping.py b/common/type_mapping.py new file mode 100644 index 0000000..491822c --- /dev/null +++ b/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) diff --git a/config.yaml b/config.yaml index c87bed1..d565ed0 100644 --- a/config.yaml +++ b/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": "无人在场"