公路边坡落石滑坡监测摄像机：全天候AI感知与边缘计算在交通地质安全中的实践

import cv2
import numpy as np
import time

class LandslideRockfallDetector:
    def __init__(self, model_path, confidence_threshold=0.75):
        # 加载针对地质场景优化的AI模型（如YOLO系列或自定义CNN）
        self.net = self._load_model(model_path)
        self.conf_threshold = confidence_threshold
        self.alarm_cooldown = 60  # 报警冷却时间，防止同一事件重复上报
        self.last_alarm_time = 0
        
    def _load_model(self, path):
        # 模拟加载边缘AI加速模型
        print(f"Loading specialized geological model from {path}...")
        return True

    def detect_hazard(self, frame):
        """
        执行灾害推理
        :param frame: 视频帧图像
        :return: (is_hazard, hazard_type, bbox)
        """
        # 此处应调用TensorRT或ONNX Runtime进行前向推理
        # 模拟推理结果：识别到落石，置信度0.88，坐标[x, y, w, h]
        mock_result = {
            "class": "rockfall",
            "confidence": 0.88,
            "bbox": [100, 150, 50, 50]
        }
        
        if mock_result["confidence"] > self.conf_threshold:
            return True, mock_result["class"], mock_result["bbox"]
        return False, None, None

    def process_stream(self, video_source):
        cap = cv2.VideoCapture(video_source)
        
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret: break
            
            # 1. 图像预处理：去雾、低光照增强（模拟）
            # enhanced_frame = self.enhance_image(frame)
            
            # 2. AI推理检测
            is_hazard, h_type, bbox = self.detect_hazard(frame)
            
            # 3. 逻辑判断与报警触发
            if is_hazard:
                current_time = time.time()
                if current_time - self.last_alarm_time > self.alarm_cooldown:
                    self._trigger_alarm(h_type, frame, bbox)
                    self.last_alarm_time = current_time
            
            # 4. 可视化展示
            if bbox:
                x, y, w, h = bbox
                cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 0, 255), 2)
                cv2.putText(frame, f"ALERT: {h_type}", (x, y-10), 
                            cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 0, 255), 2)
            
            cv2.imshow('Geological Monitoring', frame)
            if cv2.waitKey(1) & 0xFF == ord('q'): break
            
        cap.release()

    def _trigger_alarm(self, hazard_type, frame, bbox):
        # 触发多级报警：抓拍、上传中心、联动现场声光
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{timestamp}] CRITICAL ALERT: {hazard_type.upper()} detected!")
        # 实际逻辑中会调用HTTP/MQTT接口上传至云平台
        # self._upload_to_cloud(frame, hazard_type)

# 实例化与运行
# detector = LandslideRockfallDetector(model_path="geo_resnet50.onnx")
# detector.process_stream("rtsp://camera_ip/stream")