山区道路落石智能检测预警系统 —— 护路先锋一、实际应用场景描述场景川藏线怒江72拐路段凌晨巡查想象这样一个惊险的场景凌晨5点318国道怒江72拐路段海拔4600米的业拉山上寒风呼啸。养路工人老王驾驶着皮卡车例行巡查车载收音机里播放着路况通报前方K3692段有零星落石请注意避让。老王放慢车速至15km/h凭借多年的经验目视搜寻路面异物。突然前方弯道内侧山体传来一阵异响——一块篮球大小的岩石正从崖壁松动脱落老王猛踩刹车但由于能见度极低仅有车前灯照射范围且反应距离不足车辆左前轮还是碾压到了碎石堆。虽然避免了致命撞击但爆胎导致车辆失控滑行30米后撞上护栏老王轻微脑震荡道路中断6小时后方积压车辆达200余辆。行业现状与痛点痛点 传统方案 后果人工巡查 每日2-3次覆盖率30% 盲区事故频发伤亡率高群众上报 依赖路人发现滞后性强 二次事故发生率40%定时监控 视频轮巡响应慢 平均响应时间15-30分钟无智能识别 海量视频靠人眼筛查 人力成本占养护费用60%创新方案AI视觉落石实时检测受智能车辆环境感知技术启发我们设计基于深度学习的山区落石智能检测预警系统- 全天候感知红外可见光双光谱摄像头夜间可视距离100米- 智能识别改进的YOLOv8-Rock模型专门针对不规则岩石形状优化- 多级预警边缘端实时预警云端调度中心下游情报板联动- 预测性维护结合地质雷达数据提前72小时预警高风险区段- 自动处置联动无人机巡检应急清障车调度颠覆点将事后处置变为事前预警将人工巡查升级为AI全天候值守将单点监控扩展为路网协同感知。二、痛点分析1. 生命安全威胁- 死亡率高山区落石事故致死率达23%远超一般交通事故- 突发性强落石发生无任何征兆传统预警完全失效- 连锁反应一处落石可能引发多车连环相撞次生灾害严重2. 经济损失巨大- 直接损失单车维修5-50万元道路抢修10-100万元- 间接损失交通中断造成的物流延误、旅游损失难以估量- 保险赔付年均赔付金额超亿元保费持续上涨3. 养护效率低下- 人力密集每100公里需配备8-12名巡查员人工成本极高- 覆盖不足夜间、恶劣天气基本无法有效巡查- 主观性强依赖个人经验漏检率高达35%4. 技术局限性- 传统CV算法基于边缘检测、纹理分析误报率50%- 单一传感器仅靠震动传感器无法区分落石、塌方、爆破- 网络依赖山区4G信号不稳定云端方案经常断联三、核心逻辑讲解1. 系统架构┌─────────────┐ ┌─────────────┐ ┌─────────────┐│ 红外摄像头 │ │ 可见光摄像 │ │ 地质雷达 ││ (夜间可视) │ │ 头 │ │ (形变监测) │└──────┬──────┘ └──────┬──────┘ └──────┬──────┘│ │ │▼ ▼ ▼┌─────────────────────────────────────────────┐│ 边缘计算单元 ││ (Jetson Orin NX / 华为Atlas) ││ ┌─────────┐ ┌─────────┐ ┌─────────┐ ││ │YOLOv8- │ │岩石 │ │多光谱 │ ││ │Rock │→ │分类器 │→ │融合 │ ││ └─────────┘ └─────────┘ └─────────┘ │└─────────────┬───────────────────────────────┘│┌──────┴──────┐▼ ▼┌─────────────┐ ┌─────────────┐│ 现场预警 │ │ 云端平台 ││ (声光情报 │ │ (路网协同) ││ 板广播) │ │ (预测分析) │└─────────────┘ └─────────────┘2. 核心算法流程阶段1多光谱数据采集与预处理红外图像 可见光图像 → 图像配准 → 去雾增强 → 拼接融合 → 感兴趣区域提取阶段2落石目标检测YOLOv8-Rock输入图像 → C2f骨干网络 → SPPFBiFPN颈部 → Rock-Detect头部 → 落石候选框阶段3岩石属性分类落石ROI → ResNet34特征提取 → 岩石类型分类 → 尺寸估算 → 风险评估阶段4时空关联分析当前检测 历史轨迹 气象数据 → 落石轨迹预测 → 影响范围评估 → 预警等级判定3. 关键技术点- 岩石专用Anchor针对不规则岩石形状重新聚类anchor box长宽比2:1~5:1- 多光谱融合红外突出岩石轮廓可见光保留纹理细节特征级融合提升检出率- 时序一致性3D卷积LSTM捕捉落石下落轨迹区分静态石块vs动态落石- 地形感知结合DEM高程数据预判高风险落石区域坡度45°植被稀少- 边缘-云端协同边缘端实时告警云端大数据分析预测高发时段和区段四、完整代码实现项目结构mountain_rockfall_detection/├── README.md├── requirements.txt├── config.py├── main.py├── data_collection.py # 数据采集模块├── preprocessing.py # 图像预处理├── rock_detector.py # YOLOv8落石检测├── rock_classifier.py # 岩石属性分类├── multispectral_fusion.py # 多光谱融合├── spatiotemporal_analysis.py # 时空关联分析├── alert_system.py # 多级预警系统├── geological_interface.py # 地质雷达接口├── edge_deployment.py # 边缘部署工具├── cloud_api.py # 云端API接口├── utils.py # 工具函数├── models/ # 预训练模型│ ├── yolov8_rock.pt│ ├── rock_classifier.pth│ └── terrain_model.pth├── data/│ ├── raw_images/│ │ ├── visible/│ │ └── infrared/│ ├── processed/│ ├── annotations/│ └── terrain_maps/├── logs/│ ├── detection_log.json│ ├── alert_log.csv│ └── prediction_log.json├── configs/│ ├── model_config.yaml│ ├── alert_rules.yaml│ └── terrain_config.yaml└── scripts/├── deploy_edge.sh├── collect_data.py└── evaluate_model.py1. requirements.txtultralytics8.0.0 # YOLOv8torch2.0.0 # PyTorchtorchvision0.15.0 # 视觉工具opencv-python4.8.0 # 图像处理opencv-contrib-python4.8.0 # OpenCV扩展含红外处理numpy1.24.0 # 数值计算pandas2.0.0 # 数据处理scikit-learn1.3.0 # 机器学习scipy1.11.0 # 科学计算pillow10.0.0 # 图像IOpyyaml6.0 # YAML配置requests2.31.0 # HTTP请求pyserial3.5 # 串口通信numba0.57.0 # JIT加速geopandas0.13.0 # 地理数据处理rasterio1.3.0 # 栅格数据读取shapely2.0.0 # 几何运算matplotlib3.7.0 # 可视化seaborn0.12.0 # 统计可视化tqdm4.65.0 # 进度条loguru0.7.0 # 日志pycocotools2.0.6 # COCO格式处理albumentations1.3.0 # 数据增强onnx1.14.0 # ONNX格式onnxruntime1.16.0 # ONNX推理tensorrt8.6.0 # NVIDIA推理加速(可选)2. config.py山区落石检测预警系统配置包含模型参数、检测规则、预警策略、地形数据等from dataclasses import dataclass, fieldfrom typing import List, Dict, Tuple, Optional, Setfrom pathlib import Pathimport osimport yamldataclassclass ModelConfig:模型配置# YOLOv8落石检测模型yolo_model_path: str models/yolov8_rock.ptyolo_conf_threshold: float 0.35 # 检测置信度阈值落石检测需要较低阈值yolo_iou_threshold: float 0.45 # NMS IOU阈值yolo_input_size: Tuple[int, int] (800, 800) # 输入尺寸更大以捕捉远处落石# 岩石分类模型classifier_model_path: str models/rock_classifier.pthclassifier_input_size: Tuple[int, int] (256, 256)classifier_num_classes: int 5 # 小/中/大/巨石/碎石堆classifier_conf_threshold: float 0.6# 地形分析模型terrain_model_path: str models/terrain_model.pthslope_threshold: float 45.0 # 危险坡度阈值vegetation_threshold: float 0.3 # 植被覆盖率阈值# 边缘部署配置enable_tensorrt: bool Truetensorrt_engine_path: str models/yolov8_rock.enginequantization: str fp16 # fp32/fp16/int8device: str cuda:0 if os.environ.get(CUDA_VISIBLE_DEVICES) else cpu# 多光谱配置enable_multispectral: bool Trueinfrared_weight: float 0.4visible_weight: float 0.6dataclassclass DetectionRules:检测规则配置# 落石尺寸定义相对图像宽度比例small_rock_threshold: float 0.02 # 2% 小石子medium_rock_threshold: float 0.08 # 2%-8% 中石头large_rock_threshold: float 0.15 # 8%-15% 大石头huge_rock_threshold: float 0.30 # 15% 巨石# 动态落石检测motion_threshold: float 0.1 # 运动幅度阈值motion_history_length: int 5 # 运动历史帧数# 时序确认规则confirmation_frames: int 3 # 连续N帧确认confirmation_threshold: float 0.7 # 确认帧占比cooldown_seconds: float 60.0 # 预警冷却时间# 区域配置roi_left_top: Tuple[float, float] (0.0, 0.1)roi_right_bottom: Tuple[float, float] (1.0, 0.9)# 高风险区域基于DEM数据high_risk_areas: List[Dict] field(default_factorylambda: [{name: K3692弯道, lat_range: (29.5, 29.6), lon_range: (97.2, 97.3)},{name: 怒江观景台, lat_range: (29.55, 29.58), lon_range: (97.25, 97.28)}])dataclassclass TerrainConfig:地形数据配置# DEM数据路径dem_data_path: str data/terrain_maps/dem_30m.tifslope_data_path: str data/terrain_maps/slope.tifaspect_data_path: str data/terrain_maps/aspect.tifvegetation_path: str data/terrain_maps/ndvi.tif# 分析参数analysis_resolution: float 30.0 # 米/像素search_radius: float 500.0 # 分析半径米# 风险因子权重slope_weight: float 0.4vegetation_weight: float 0.3distance_to_road_weight: float 0.2historical_rockfall_weight: float 0.1dataclassclass AlertConfig:预警配置# 预警级别level_info: str info # 小落石信息提示level_warning: str warning # 中落石黄色预警level_critical: str critical # 大落石/巨石红色警报# 通知方式enable_edge_alert: bool True # 边缘端声光报警enable_vms: bool True # 可变情报板enable_broadcast: bool True # 应急广播enable_cloud_push: bool True # 云端推送enable_sms: bool True # 短信通知# 通知内容模板vms_template: str 【落石预警】前方{K}米有{size}落石请减速慢行sms_template: str 【护路先锋】路段{section}检测到{size}落石风险等级{level}# 接收人配置alert_recipients: List[str] field(default_factorylambda: [road_admintransport.gov.cn,emergencylocal.gov.cn])# 预警范围米upstream_distance: float 2000.0 # 上游预警距离downstream_distance: float 500.0 # 下游预警距离dataclassclass SystemConfig:系统配置# 运行模式mode: str realtime # realtime/batch/video/offline_analysis# 数据源配置camera_urls: List[str] field(default_factorylambda: [rtsp://192.168.1.100:554/visible,rtsp://192.168.1.101:554/infrared])camera_index: int 0 # 本地摄像头索引video_path: str offline_data_path: str # 输出配置output_log_path: str logs/detection_log.jsonoutput_video_path: str output/rockfall_detection.avienable_visualization: bool Truevisualization_scale: float 1.0# 性能配置max_fps: int 25frame_skip: int 1batch_size: int 1# 数据存储enable_storage: bool Truestorage_path: str data/events/retention_days: int 90# 全局配置实例MODEL_CONFIG ModelConfig()DETECTION_RULES DetectionRules()TERRAIN_CONFIG TerrainConfig()ALERT_CONFIG AlertConfig()SYSTEM_CONFIG SystemConfig()# 落石尺寸枚举ROCK_SIZE {0: SMALL_ROCK, # 小石子 (10cm)1: MEDIUM_ROCK, # 中石头 (10-30cm)2: LARGE_ROCK, # 大石头 (30-80cm)3: HUGE_ROCK, # 巨石 (80cm)4: ROCK_PILE # 碎石堆}SIZE_COLORS {SMALL_ROCK: (0, 255, 0), # 绿色MEDIUM_ROCK: (0, 255, 255), # 黄色LARGE_ROCK: (0, 165, 255), # 橙色HUGE_ROCK: (0, 0, 255), # 红色ROCK_PILE: (128, 0, 128), # 紫色UNKNOWN: (128, 128, 128) # 灰色}# 预警级别颜色ALERT_COLORS {info: (0, 255, 0),warning: (0, 255, 255),critical: (0, 0, 255)}# 创建必要目录for dir_path in [models,data/raw_images/visible,data/raw_images/infrared,data/processed,data/annotations,data/terrain_maps,logs,output,assets,data/events]:os.makedirs(dir_path, exist_okTrue)# 加载YAML配置如果存在def load_yaml_config(config_path: str) - dict:加载YAML配置文件if os.path.exists(config_path):with open(config_path, r, encodingutf-8) as f:return yaml.safe_load(f)return {}# 合并YAML配置到dataclassdef merge_yaml_to_config(yaml_path: str, config_obj):将YAML配置合并到配置对象yaml_config load_yaml_config(yaml_path)if yaml_config:for key, value in yaml_config.items():if hasattr(config_obj, key):setattr(config_obj, key, value)print(f已加载配置文件: {yaml_path})3. main.py#!/usr/bin/env python3山区落石智能检测预警系统主程序护路先锋 - Mountain Rockfall Detection Early Warning Systemimport cv2import timeimport threadingimport queueimport signalimport sysfrom datetime import datetimefrom typing import Optional, Dict, Any, Listfrom pathlib import Pathimport jsonimport tracebackfrom config import MODEL_CONFIG, DETECTION_RULES, TERRAIN_CONFIG, ALERT_CONFIG, SYSTEM_CONFIG, ROCK_SIZE, SIZE_COLORS, ALERT_COLORSfrom rock_detector import RockDetectorfrom rock_classifier import RockClassifierfrom multispectral_fusion import MultispectralFusionfrom spatiotemporal_analysis import SpatioTemporalAnalyzerfrom alert_system import AlertSystemfrom geological_interface import GeologicalInterfacefrom utils import setup_logging, draw_detection_result, save_detection_log, create_event_folderfrom data_collection import DataCollector# 设置日志setup_logging(SYSTEM_CONFIG.output_log_path)class MountainRockfallDetectionSystem:山区落石检测预警系统主类整合所有模块实现端到端的落石检测与预警def __init__(self):初始化系统self.logger logging.getLogger(__name__)self.logger.info( * 70)self.logger.info(️ 山区落石智能检测预警系统 - 护路先锋)self.logger.info( Mountain Rockfall Detection Early Warning System)self.logger.info( * 70)# 系统状态self.running Falseself.paused Falseself.current_frame Noneself.current_timestamp None# 初始化各模块self._init_modules()# 状态变量self.current_alert_level normalself.rockfall_events: List[Dict] []self.statistics {total_frames: 0,rockfall_detections: 0,alerts_triggered: 0,start_time: datetime.now()}# 信号处理signal.signal(signal.SIGINT, self._signal_handler)signal.signal(signal.SIGTERM, self._signal_handler)def _init_modules(self):初始化各个功能模块try:# 1. 落石检测器YOLOv8-Rockself.logger.info(1. 初始化落石检测器...)self.rock_detector RockDetector(MODEL_CONFIG)self.logger.info( ✅ 落石检测器初始化完成)# 2. 岩石分类器self.logger.info(2. 初始化岩石属性分类器...)self.rock_classifier RockClassifier(MODEL_CONFIG)self.logger.info( ✅ 岩石分类器初始化完成)# 3. 多光谱融合器self.logger.info(3. 初始化多光谱融合器...)self.fusion_engine MultispectralFusion(MODEL_CONFIG, DETECTION_RULES)self.logger.info( ✅ 多光谱融合器初始化完成)# 4. 时空关联分析器self.logger.info(4. 初始化时空关联分析器...)self.spatiotemporal_analyzer SpatioTemporalAnalyzer(DETECTION_RULES, TERRAIN_CONFIG)self.logger.info( ✅ 时空关联分析器初始化完成)# 5. 预警系统self.logger.info(5. 初始化多级预警系统...)self.alert_system AlertSystem(ALERT_CONFIG, TERRAIN_CONFIG)self.logger.info( ✅ 预警系统初始化完成)# 6. 地质雷达接口self.logger.info(6. 初始化地质雷达接口...)self.geological_interface GeologicalInterface(TERRAIN_CONFIG)self.logger.info( ✅ 地质雷达接口初始化完成)# 7. 数据采集器可选if SYSTEM_CONFIG.mode in [realtime, offline_analysis]:self.logger.info(7. 初始化数据采集器...)self.data_collector DataCollector(SYSTEM_CONFIG)self.logger.info( ✅ 数据采集器初始化完成)self.logger.info( * 70)self.logger.info(✅ 所有模块初始化完成系统就绪)self.logger.info( * 70)except Exception as e:self.logger.error(f❌ 模块初始化失败: {str(e)})self.logger.error(traceback.format_exc())raisedef _signal_handler(self, signum, frame):处理系统信号self.logger.info(f收到信号 {signum}正在安全停止系统...)self.stop()def start(self):启动系统self.running Trueself.logger.info( 系统启动中...)try:if SYSTEM_CONFIG.mode realtime:self._run_realtime_mode()elif SYSTEM_CONFIG.mode video:self._run_video_mode()elif SYSTEM_CONFIG.mode batch:self._run_batch_mode()elif SYSTEM_CONFIG.mode offline_analysis:self._run_offline_analysis()else:self.logger.error(f未知运行模式: {SYSTEM_CONFIG.mode})except Exception as e:self.logger.error(f系统运行错误: {str(e)})self.logger.error(traceback.format_exc())finally:self._cleanup()def _run_realtime_mode(self):实时检测模式self.logger.info( 启动实时检测模式...)# 获取视频源if SYSTEM_CONFIG.camera_urls:# 多光谱视频流self._init_multispectral_streams()else:# 单摄像头self.cap cv2.VideoCapture(SYSTEM_CONFIG.camera_index)if not self.cap.isOpened():raise RuntimeError(无法打开视频源)self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)self.cap.set(cv2.CAP_PROP_FPS, SYSTEM_CONFIG.max_fps)self.logger.info(视频源已连接开始处理帧...)# 主循环while self.running:try:if self.paused:time.sleep(0.1)continue# 读取帧if SYSTEM_CONFIG.camera_urls:visible_frame, infrared_frame self._read_multispectral_frames()else:ret, visible_frame self.cap.read()if not ret:self.logger.warning(无法读取视频帧尝试重连...)time.sleep(1)continueinfrared_frame Noneif visible_frame is None:continue# 处理帧self.current_timestamp datetime.now()result self._process_frame(visible_frame, infrared_frame)# 更新显示if SYSTEM_CONFIG.enable_visualization:self._update_display(visible_frame, result)# 检查退出if cv2.waitKey(1) 0xFF ord(q):self.logger.info(用户请求退出)breakexcept Exception as e:self.logger.error(f帧处理错误: {str(e)})time.sleep(0.1)def _init_multispectral_streams(self):初始化多光谱视频流self.streams []for url in SYSTEM_CONFIG.camera_urls:cap cv2.VideoCapture(url)if cap.isOpened():cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) # 减少延迟self.streams.append(cap)self.logger.info(f已连接视频流: {url})else:self.logger.warning(f无法连接视频流: {url})if not self.streams:raise RuntimeError(没有可用的视频流)def _read_multispectral_frames(self) - Tuple[Optional[np.ndarray], Optional[np.ndarray]]:读取多光谱帧visible_frame Noneinfrared_frame Nonefor i, stream in enumerate(self.streams):ret, frame stream.read()if not ret:continueif i 0:visible_frame frameelif i 1:infrared_frame framereturn visible_frame, infrared_framedef _run_video_mode(self):视频文件检测模式self.logger.info(f 启动视频检测模式: {SYSTEM_CONFIG.video_path})if not os.path.exists(SYSTEM_CONFIG.video_path):raise FileNotFoundError(f视频文件不存在: {SYSTEM_CONFIG.video_path})cap cv2.VideoCapture(SYSTEM_CONFIG.video_path)if not cap.isOpened():raise RuntimeError(f无法打开视频文件: {SYSTEM_CONFIG.video_path})# 获取视频信息total_frames int(cap.get(cv2.CAP_PROP_FRAME_COUNT))fps cap.get(cv2.CAP_PROP_FPS)self.logger.info(f视频信息: {total_frames}帧, {fps}fps)# 输出视频写入器fourcc cv2.VideoWriter_fourcc(*XVID)out cv2.VideoWriter(SYSTEM_CONFIG.output_video_path,fourcc, fps,(int(cap.get(3)), int(cap.get(4))))frame_count 0self.running Truetry:while self.running and frame_count total_frames:ret, frame cap.read()if not ret:breakframe_count 1# 跳帧处理if frame_count % SYSTEM_CONFIG.frame_skip ! 0:continueself.current_timestamp datetime.now()result self._process_frame(frame)self.statistics[total_frames] 1利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛