公路落石滑坡灾害监测方案 基于YOLOv10模型

1import torch
2import torch.nn as nn
3
4class EnhancedFPN(nn.Module):
5    """
6    增强的特征金字塔网络，针对边坡灾害场景优化
7    """
8    def __init__(self, in_channels_list=[256, 512, 1024]):
9        super(EnhancedFPN, self).__init__()
10        
11        # 横向连接层
12        self.lateral_convs = nn.ModuleList([
13            nn.Conv2d(ch, 256, 1) for ch in in_channels_list
14        ])
15        
16        # 特征金字塔卷积层
17        self.fpn_convs = nn.ModuleList([
18            nn.Conv2d(256, 256, 3, padding=1) for _ in in_channels_list
19        ])
20        
21        # 高分辨率特征增强
22        self.high_res_enhance = nn.Sequential(
23            nn.Conv2d(256, 128, 3, padding=1),
24            nn.BatchNorm2d(128),
25            nn.ReLU(inplace=True),
26            nn.Conv2d(128, 256, 3, padding=1)
27        )
28        
29        # 注意力增强模块
30        self.attention = ChannelSpatialAttention(256)
31        
32    def forward(self, features):
33        """
34        features: list of feature maps from backbone
35        """
36        # 自上而下处理
37        lateral_features = []
38        for i, feature in enumerate(reversed(features)):
39            lateral_features.append(self.lateral_convs[i](feature))
40        
41        # 特征融合
42        for i in range(len(lateral_features) - 1):
43            upsampled = nn.functional.interpolate(
44                lateral_features[i], 
45                size=lateral_features[i+1].shape[2:], 
46                mode='nearest'
47            )
48            lateral_features[i+1] = lateral_features[i+1] + upsampled
49        
50        # FPN输出
51        fpn_outputs = []
52        for i, lateral in enumerate(lateral_features):
53            fpn_output = self.fpn_convs[i](lateral)
54            # 注意力增强
55            fpn_output = self.attention(fpn_output)
56            fpn_outputs.append(fpn_output)
57        
58        # 高分辨率特征增强
59        fpn_outputs[0] = self.high_res_enhance(fpn_outputs[0])
60        
61        return fpn_outputs

1class ChannelSpatialAttention(nn.Module):
2    """
3    通道和空间注意力融合机制
4    """
5    def __init__(self, channels, reduction=16):
6        super(ChannelSpatialAttention, self).__init__()
7        
8        # 通道注意力
9        self.avg_pool = nn.AdaptiveAvgPool2d(1)
10        self.max_pool = nn.AdaptiveMaxPool2d(1)
11        
12        self.fc = nn.Sequential(
13            nn.Linear(channels, channels // reduction, bias=False),
14            nn.ReLU(inplace=True),
15            nn.Linear(channels // reduction, channels, bias=False)
16        )
17        
18        self.sigmoid = nn.Sigmoid()
19        
20        # 空间注意力
21        self.conv = nn.Conv2d(2, 1, kernel_size=7, padding=3, bias=False)
22        
23    def forward(self, x):
24        # 通道注意力
25        avg_out = self.fc(self.avg_pool(x).view(x.size(0), -1))
26        max_out = self.fc(self.max_pool(x).view(x.size(0), -1))
27        channel_att = self.sigmoid(avg_out + max_out).view(x.size(0), x.size(1), 1, 1)
28        
29        x = x * channel_att
30        
31        # 空间注意力
32        avg_out = torch.mean(x, dim=1, keepdim=True)
33        max_out, _ = torch.max(x, dim=1, keepdim=True)
34        spatial_att = torch.cat([avg_out, max_out], dim=1)
35        spatial_att = self.sigmoid(self.conv(spatial_att))
36        
37        x = x * spatial_att
38        
39        return x

1class RockfallTrajectoryPredictor(nn.Module):
2    """
3    落石轨迹预测模块
4    """
5    def __init__(self, input_dim=256, hidden_dim=128):
6        super(RockfallTrajectoryPredictor, self).__init__()
7        
8        self.feature_extractor = nn.Sequential(
9            nn.Conv2d(input_dim, hidden_dim, 3, padding=1),
10            nn.BatchNorm2d(hidden_dim),
11            nn.ReLU(inplace=True),
12            nn.Conv2d(hidden_dim, hidden_dim, 3, padding=1),
13            nn.AdaptiveAvgPool2d(1)
14        )
15        
16        self.lstm = nn.LSTM(hidden_dim, hidden_dim, num_layers=2, batch_first=True)
17        
18        # 轨迹预测
19        self.trajectory_fc = nn.Linear(hidden_dim, 4)  # x, y, vx, vy
20        
21        # 落点预测
22        self.impact_fc = nn.Linear(hidden_dim, 2)  # impact_x, impact_y
23        
24    def forward(self, x, history_positions=None):
25        """
26        x: current frame feature
27        history_positions: historical positions [batch, seq_len, 2]
28        """
29        # 特征提取
30        feat = self.feature_extractor(x).view(x.size(0), -1)
31        
32        if history_positions is not None:
33            # LSTM处理历史轨迹
34            lstm_out, _ = self.lstm(history_positions)
35            feat = feat + lstm_out[:, -1, :]
36        
37        # 轨迹预测
38        trajectory = self.trajectory_fc(feat)
39        
40        # 落点预测
41        impact_point = self.impact_fc(feat)
42        
43        return trajectory, impact_point

1class LandslideRiskAssessor(nn.Module):
2    """
3    滑坡风险评估模块
4    """
5    def __init__(self, input_dim=256):
6        super(LandslideRiskAssessor, self).__init__()
7        
8        self.crack_detector = nn.Sequential(
9            nn.Conv2d(input_dim, 128, 3, padding=1),
10            nn.BatchNorm2d(128),
11            nn.ReLU(inplace=True),
12            nn.Conv2d(128, 64, 3, padding=1),
13            nn.AdaptiveAvgPool2d(1),
14            nn.Flatten(),
15            nn.Linear(64, 32),
16            nn.ReLU(inplace=True)
17        )
18        
19        self.displacement_analyzer = nn.Sequential(
20            nn.Conv2d(input_dim, 128, 3, padding=1),
21            nn.BatchNorm2d(128),
22            nn.ReLU(inplace=True),
23            nn.Conv2d(128, 64, 3, padding=1),
24            nn.AdaptiveAvgPool2d(1),
25            nn.Flatten(),
26            nn.Linear(64, 32),
27            nn.ReLU(inplace=True)
28        )
29        
30        self.vegetation_analyzer = nn.Sequential(
31            nn.Conv2d(input_dim, 128, 3, padding=1),
32            nn.BatchNorm2d(128),
33            nn.ReLU(inplace=True),
34            nn.Conv2d(128, 64, 3, padding=1),
35            nn.AdaptiveAvgPool2d(1),
36            nn.Flatten(),
37            nn.Linear(64, 32),
38            nn.ReLU(inplace=True)
39        )
40        
41        # 风险等级分类
42        self.risk_classifier = nn.Sequential(
43            nn.Linear(96, 64),
44            nn.ReLU(inplace=True),
45            nn.Dropout(0.3),
46            nn.Linear(64, 4)  # 低、中、高、极高风险
47        )
48        
49    def forward(self, x):
50        crack_feat = self.crack_detector(x)
51        displacement_feat = self.displacement_analyzer(x)
52        vegetation_feat = self.vegetation_analyzer(x)
53        
54        combined_feat = torch.cat([crack_feat, displacement_feat, vegetation_feat], dim=1)
55        
56        risk_level = self.risk_classifier(combined_feat)
57        
58        return risk_level

1import albumentations as A
2import cv2
3import numpy as np
4
5def landslide_specific_augmentation():
6    """
7    边坡灾害场景特有数据增强
8    """
9    return A.Compose([
10        # 光照变化模拟
11        A.RandomBrightnessContrast(brightness_limit=0.4, contrast_limit=0.4, p=0.6),
12        # 雾气效果模拟
13        A.RandomFog(fog_coef_lower=0.1, fog_coef_upper=0.4, alpha_coef=0.15, p=0.3),
14        # 雨雪效果模拟
15        A.RandomRain(brightness_coefficient=0.8, drop_width=1, blur_value=3, p=0.25),
16        # 常规增强
17        A.HorizontalFlip(p=0.5),
18        A.RandomRotate90(p=0.5),
19        A.HueSaturationValue(hue_shift_limit=15, sat_shift_limit=30, val_shift_limit=30, p=0.4),
20        # 高斯噪声
21        A.GaussNoise(var_limit=(10.0, 60.0), p=0.3),
22        # 运动模糊
23        A.MotionBlur(blur_limit=7, p=0.2),
24        # 雪花效果
25        A.RandomSnow(snow_point_lower=0.1, snow_point_upper=0.3, brightness_coeff=2.0, p=0.15),
26    ], bbox_params=A.BboxParams(format='yolo', label_fields=['class_labels']))

1def adaptive_enhancement_for_landslide(img):
2    """
3    边坡场景自适应图像增强
4    """
5    # 多尺度CLAHE增强
6    lab = cv2.cvtColor(img, cv2.COLOR_BGR2LAB)
7    l, a, b = cv2.split(lab)
8    
9    # 针对不同区域应用不同增强策略
10    clahe_strong = cv2.createCLAHE(clipLimit=3.5, tileGridSize=(8, 8))
11    clahe_medium = cv2.createCLAHE(clipLimit=2.5, tileGridSize=(12, 12))
12    clahe_weak = cv2.createCLAHE(clipLimit=1.8, tileGridSize=(16, 16))
13    
14    # 分割图像区域
15    h, w = l.shape
16    center_region = np.zeros((h, w), dtype=np.uint8)
17    cv2.rectangle(center_region, (w//4, h//4), (3*w//4, 3*h//4), 255, -1)
18    
19    edge_region = np.zeros((h, w), dtype=np.uint8)
20    cv2.rectangle(edge_region, (w//8, h//8), (7*w//8, 7*h//8), 255, -1)
21    edge_region = edge_region - center_region
22    
23    # 不同区域应用不同增强
24    l_center = clahe_strong.apply(l)
25    l_edge = clahe_medium.apply(l)
26    l_corner = clahe_weak.apply(l)
27    
28    # 融合
29    l_enhanced = np.where(center_region > 0, l_center, 
30                         np.where(edge_region > 0, l_edge, l_corner))
31    
32    lab_enhanced = cv2.merge([l_enhanced, a, b])
33    img_enhanced = cv2.cvtColor(lab_enhanced, cv2.COLOR_LAB2BGR)
34    
35    return img_enhanced

1from ultralytics import YOLO
2
3def train_landslide_detection_model():
4    """
5    边坡灾害检测模型训练配置
6    """
7    model = YOLO('yolov10n.pt')
8    
9    results = model.train(
10        data='landslide_dataset.yaml',
11        epochs=180,
12        imgsz=1280,
13        batch=8,
14        multi_scale=True,
15        augment=True,
16        # 数据增强参数
17        hsv_h=0.015,
18        hsv_s=0.7,
19        hsv_v=0.5,
20        degrees=20.0,
21        translate=0.15,
22        scale=0.6,
23        shear=3.0,
24        perspective=0.001,
25        flipud=0.0,
26        fliplr=0.5,
27        mosaic=1.0,
28        mixup=0.15,
29        copy_paste=0.1,
30        # 优化器参数
31        lr0=0.01,
32        lrf=0.01,
33        momentum=0.937,
34        weight_decay=0.0005,
35    )
36    
37    return results