【智慧防洪】汉江流域汛期洪水预防：基于智慧水利的技术解决方案

// 数据传输配置示例
const networkConfig = {
  // 重点区域：5G 高速传输
  priorityZones: {
    network: '5G',
    bandwidth: '100 Mbps',
    latency: '< 10ms',
    devices: ['AI 摄像头', '视频流']
  },
  // 一般区域：4G/NB-IoT
  normalZones: {
    network: '4G/NB-IoT',
    bandwidth: '1-10 Mbps',
    latency: '< 100ms',
    devices: ['水位计', '雨量计', '流速仪']
  },
  // 偏远地区：北斗卫星通信
  remoteZones: {
    network: 'BeiDou Satellite',
    bandwidth: '9.6 kbps',
    latency: '< 1s',
    devices: ['应急监测站']
  },
  // 数据协议
  protocol: {
    transport: 'MQTT over TLS',
    dataFormat: 'JSON',
    compression: 'gzip',
    heartbeat: '60s'
  }
};

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
import numpy as np
import pandas as pd
class FloodPredictionModel:
    """汉江流域洪水预测模型"""
    def __init__(self, input_features=10, sequence_length=72):
        """
        参数:
            input_features: 输入特征数（水位、雨量、流速等）
            sequence_length: 时间序列长度（小时）
        """
        self.model = self._build_model(input_features, sequence_length)
    def _build_model(self, input_features, sequence_length):
        """构建 LSTM 预测模型"""
        model = Sequential([
            # 第一层 LSTM
            LSTM(128, return_sequences=True, 
                 input_shape=(sequence_length, input_features)),
            Dropout(0.2),
            # 第二层 LSTM
            LSTM(64, return_sequences=False),
            Dropout(0.2),
            # 全连接层
            Dense(32, activation='relu'),
            Dense(16, activation='relu'),
            # 输出层（预测未来 24 小时水位）
            Dense(24, activation='linear')
        ])
        model.compile(
            optimizer='adam',
            loss='mse',
            metrics=['mae']
        )
        return model
    def train(self, X_train, y_train, epochs=100, batch_size=32):
        """训练模型"""
        history = self.model.fit(
            X_train, y_train,
            epochs=epochs,
            batch_size=batch_size,
            validation_split=0.2,
            callbacks=[
                tf.keras.callbacks.EarlyStopping(patience=10),
                tf.keras.callbacks.ReduceLROnPlateau(factor=0.5)
            ]
        )
        return history
    def predict(self, X_test):
        """预测未来 24 小时水位"""
        predictions = self.model.predict(X_test)
        return predictions
    def evaluate_accuracy(self, y_true, y_pred):
        """评估预测准确率"""
        from sklearn.metrics import mean_absolute_error, r2_score
        mae = mean_absolute_error(y_true, y_pred)
        r2 = r2_score(y_true, y_pred)
        print(f"平均绝对误差 (MAE): {mae:.2f} cm")
        print(f"决定系数 (R²): {r2:.4f}")
        return mae, r2
# 使用示例
if __name__ == '__main__':
    # 加载历史数据
    data = pd.read_csv('hanjiang_historical_data.csv')
    # 特征工程
    features = ['water_level', 'rainfall', 'flow_rate', 'humidity', 
                'temperature', 'upstream_flow', 'soil_moisture']
    # 准备训练数据
    sequence_length = 72  # 过去 72 小时数据
    prediction_horizon = 24  # 预测未来 24 小时
    # ... 数据预处理代码 ...
    # 创建并训练模型
    model = FloodPredictionModel(input_features=len(features), 
                                 sequence_length=sequence_length)
    model.train(X_train, y_train, epochs=100)
    # 预测
    predictions = model.predict(X_test)
    # 评估
    mae, r2 = model.evaluate_accuracy(y_test, predictions)

from scipy.optimize import minimize
import numpy as np
class ReservoirOptimization:
    """丹江口水库多目标优化调度"""
    def __init__(self):
        # 水库参数
        self.max_storage = 290.5e8  # 最大库容 (m³)
        self.min_storage = 170.0e8  # 死库容 (m³)
        self.flood_limit_level = 160.0  # 防洪限制水位 (m)
        self.normal_level = 170.0  # 正常蓄水位 (m)
    def objective_function(self, discharge_schedule):
        """
        目标函数：最小化综合损失
        损失包括：
        1. 防洪风险损失
        2. 供水短缺损失
        3. 发电效益损失
        """
        flood_risk = self.calculate_flood_risk(discharge_schedule)
        water_shortage = self.calculate_water_shortage(discharge_schedule)
        power_loss = self.calculate_power_loss(discharge_schedule)
        # 权重系数
        w1, w2, w3 = 0.5, 0.3, 0.2
        total_loss = w1 * flood_risk + w2 * water_shortage + w3 * power_loss
        return total_loss
    def calculate_flood_risk(self, discharge):
        """计算防洪风险"""
        # 简化模型：超出防洪限制水位的概率
        storage = self.simulate_storage(discharge)
        risk = max(0, storage - self.flood_limit_level) ** 2
        return risk
    def calculate_water_shortage(self, discharge):
        """计算供水短缺"""
        # 南水北调供水量需求
        demand = 95e8  # 年均调水量 (m³)
        supply = np.sum(discharge)
        shortage = max(0, demand - supply)
        return shortage
    def calculate_power_loss(self, discharge):
        """计算发电效益损失"""
        # 发电量与水头、流量相关
        head = self.calculate_head(discharge)
        power = 9.81 * head * discharge * 0.85  # 效率 85%
        optimal_power = self.get_optimal_power()
        loss = max(0, optimal_power - np.sum(power))
        return loss
    def optimize(self, inflow_forecast, initial_storage):
        """
        优化调度方案
        参数:
            inflow_forecast: 未来 7 天入库流量预测 (m³/s)
            initial_storage: 初始库容 (m³)
        返回:
            最优泄流方案 (m³/s)
        """
        # 决策变量：每天的平均泄流量
        n_days = len(inflow_forecast)
        x0 = np.ones(n_days) * 1000  # 初始猜测
        # 约束条件
        constraints = [
            {'type': 'ineq', 'fun': lambda x: x - 500},   # 最小泄流 500 m³/s
            {'type': 'ineq', 'fun': lambda x: 5000 - x},  # 最大泄流 5000 m³/s
        ]
        # 边界
        bounds = [(500, 5000)] * n_days
        # 优化
        result = minimize(
            self.objective_function,
            x0,
            method='SLSQP',
            bounds=bounds,
            constraints=constraints
        )
        return result.x
# 使用示例
optimizer = ReservoirOptimization()
inflow = [2000, 2500, 3000, 3500, 4000, 3500, 3000]  # 7 天预测
optimal_discharge = optimizer.optimize(inflow, 200e8)
print("最优泄流方案 (m³/s):", optimal_discharge)

import cv2
import torch
from torchvision import models, transforms
class EmbankmentInspectionAI:
    """堤防智能巡检 AI 系统"""
    def __init__(self):
        # 加载预训练模型
        self.model = self._load_model()
        self.transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                               std=[0.229, 0.224, 0.225])
        ])
        # 险情类别
        self.defect_classes = [
            'normal',           # 正常
            'crack',            # 裂缝
            'seepage',          # 渗水
            'piping',           # 管涌
            'erosion',          # 冲刷
            'settlement'        # 沉降
        ]
    def _load_model(self):
        """加载缺陷检测模型"""
        model = models.resnet50(pretrained=True)
        num_features = model.fc.in_features
        model.fc = torch.nn.Linear(num_features, len(self.defect_classes))
        model.load_state_dict(torch.load('embankment_defect_model.pth'))
        model.eval()
        return model
    def detect_defects(self, image_path):
        """
        检测堤防缺陷
        参数:
            image_path: 图像路径
        返回:
            缺陷类型、置信度、位置
        """
        # 读取图像
        image = cv2.imread(image_path)
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        # 预处理
        input_tensor = self.transform(image_rgb).unsqueeze(0)
        # 推理
        with torch.no_grad():
            outputs = self.model(input_tensor)
            probabilities = torch.softmax(outputs, dim=1)[0]
        # 获取结果
        predicted_class = torch.argmax(probabilities).item()
        confidence = probabilities[predicted_class].item()
        defect_type = self.defect_classes[predicted_class]
        return {
            'defect_type': defect_type,
            'confidence': confidence,
            'image_path': image_path,
            'timestamp': pd.Timestamp.now()
        }
    def process_video_stream(self, video_source):
        """处理视频流，实时检测"""
        cap = cv2.VideoCapture(video_source)
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            # 每隔 10 帧检测一次
            if cap.get(cv2.CAP_PROP_POS_FRAMES) % 10 == 0:
                result = self.detect_defects_from_frame(frame)
                if result['defect_type'] != 'normal':
                    # 发现险情，立即告警
                    self.send_alert(result)
            # 显示结果
            cv2.imshow('Embankment Inspection', frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
        cap.release()
        cv2.destroyAllWindows()
    def send_alert(self, result):
        """发送告警信息"""
        alert_message = f"""
        🚨 堤防险情告警
        类型: {result['defect_type']}
        置信度: {result['confidence']:.2%}
        时间: {result['timestamp']}
        位置: GPS 坐标待补充
        请立即派人现场核查！
        """
        # 发送到告警平台
        print(alert_message)
        # TODO: 集成短信/电话/APP 推送
# 使用示例
inspector = EmbankmentInspectionAI()
# 单张图片检测
result = inspector.detect_defects('embankment_photo.jpg')
print(f"检测结果: {result}")
# 视频流实时检测
# inspector.process_video_stream('rtsp://camera_ip/stream')

import dronekit
from pymavlink import mavutil
class DroneInspection:
    """无人机自动巡查系统"""
    def __init__(self, connection_string='udp:127.0.0.1:14550'):
        self.vehicle = dronekit.connect(connection_string, wait_ready=True)
    def plan_inspection_route(self, start_point, end_point, waypoints):
        """
        规划巡查路线
        参数:
            start_point: 起点 GPS (lat, lon)
            end_point: 终点 GPS (lat, lon)
            waypoints: 途经点列表 [(lat, lon), ...]
        """
        mission_items = []
        # 起飞
        mission_items.append(self._create_takeoff_command(50))  # 50 米高度
        # 添加途经点
        for i, (lat, lon) in enumerate(waypoints):
            mission_items.append(
                self._create_waypoint_command(lat, lon, 50, i+1)
            )
        # 返航
        mission_items.append(self._create_rtl_command())
        return mission_items
    def execute_inspection(self, mission_items):
        """执行巡查任务"""
        print("开始无人机巡查...")
        # 上传任务
        cmds = self.vehicle.commands
        cmds.clear()
        for item in mission_items:
            cmds.add(item)
        cmds.upload()
        # 执行任务
        self.vehicle.armed = True
        self.vehicle.mode = dronekit.VehicleMode("AUTO")
        # 监控任务进度
        while self.vehicle.mode.name == "AUTO":
            print(f"当前位置: {self.vehicle.location.global_frame}")
            print(f"剩余航点: {len(cmds)}")
            time.sleep(5)
        print("巡查任务完成")
    def capture_and_upload_images(self):
        """拍摄照片并上传"""
        # 触发相机拍照
        self.vehicle.send_mavlink(
            self.vehicle.message_factory.command_long_encode(
                0, 0,
                mavutil.mavlink.MAV_CMD_DO_DIGICAM_CONTROL,
                0, 1, 0, 0, 0, 0, 0, 0
            )
        )
        # 等待照片传输
        time.sleep(2)
        # 上传到云平台
        self.upload_to_cloud()
    def upload_to_cloud(self):
        """上传数据到云端"""
        import os
        import requests
        from datetime import datetime
        # 云平台配置
        cloud_api_url = self.config.get('cloud_api_url')
        cloud_token = self.config.get('cloud_token')
        if not cloud_api_url or not cloud_token:
            print("⚠️ 云平台配置不完整")
            return
        headers = {'Authorization': f'Bearer {cloud_token}'}
        # 上传照片
        for photo_path in self.captured_photos:
            if os.path.exists(photo_path):
                try:
                    with open(photo_path, 'rb') as f:
                        files = {'file': f}
                        data = {
                            'type': 'inspection_photo',
                            'timestamp': datetime.now().isoformat(),
                            'location': str(self.current_position)
                        }
                        response = requests.post(
                            f"{cloud_api_url}/upload",
                            headers=headers,
                            files=files,
                            data=data,
                            timeout=30
                        )
                        response.raise_for_status()
                        print(f"✅ 照片已上传: {photo_path}")
                except requests.RequestException as e:
                    print(f"❌ 照片上传失败: {photo_path}, 错误: {e}")
        # 上传遥测数据
        for telemetry in self.telemetry_data:
            try:
                response = requests.post(
                    f"{cloud_api_url}/telemetry",
                    headers=headers,
                    json=telemetry,
                    timeout=10
                )
                response.raise_for_status()
                print(f"✅ 遥测数据已上传: {telemetry}")
            except requests.RequestException as e:
                print(f"❌ 遥测数据上传失败: {e}")
        # 上传视频（如有）
        for video_path in self.captured_videos:
            if os.path.exists(video_path):
                try:
                    # 视频分片上传（适用于大文件）
                    self._upload_large_file(
                        video_path,
                        f"{cloud_api_url}/upload/video",
                        headers
                    )
                    print(f"✅ 视频已上传: {video_path}")
                except requests.RequestException as e:
                    print(f"❌ 视频上传失败: {video_path}, 错误: {e}")
    def _upload_large_file(self, file_path: str, upload_url: str, headers: dict, chunk_size: int = 5 * 1024 * 1024):
        """分片上传大文件"""
        file_size = os.path.getsize(file_path)
        file_id = f"{datetime.now().timestamp()}_{os.path.basename(file_path)}"
        with open(file_path, 'rb') as f:
            chunk_num = 0
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                files = {'chunk': chunk}
                data = {
                    'file_id': file_id,
                    'chunk_num': chunk_num,
                    'total_chunks': (file_size + chunk_size - 1) // chunk_size
                }
                response = requests.post(
                    upload_url,
                    headers=headers,
                    files=files,
                    data=data,
                    timeout=60
                )
                response.raise_for_status()
                chunk_num += 1
# 使用示例
drone = DroneInspection()
# 汉江某段堤防巡查路线
start = (32.0500, 112.1500)  # 起点
end = (32.1000, 112.2000)    # 终点
waypoints = [
    (32.0600, 112.1600),
    (32.0700, 112.1700),
    (32.0800, 112.1800),
    (32.0900, 112.1900)
]
mission = drone.plan_inspection_route(start, end, waypoints)
drone.execute_inspection(mission)

import pyswmm
import pandas as pd
class UrbanFloodSimulation:
    """城市内涝模拟系统"""
    def __init__(self, inp_file='wuhan_drainage.inp'):
        """
        参数:
            inp_file: SWMM 输入文件（排水管网模型）
        """
        self.inp_file = inp_file
    def simulate_flooding(self, rainfall_data, duration_hours=24):
        """
        模拟内涝情况
        参数:
            rainfall_data: 降雨过程数据 (mm/h)
            duration_hours: 模拟时长 (小时)
        返回:
            积水点列表、积水深度、持续时间
        """
        # 创建 SWMM 模拟对象
        sim = pyswmm.Simulation(self.inp_file)
        # 设置降雨情景
        sim.rainfall = rainfall_data
        # 运行模拟
        flooding_results = []
        for step in sim:
            # 获取各节点状态
            for node_id in sim.nodes:
                node = sim.nodes[node_id]
                # 检查是否积水
                if node.depth > 0:
                    flooding_results.append({
                        'node_id': node_id,
                        'timestamp': sim.current_time,
                        'water_depth': node.depth,  # 积水深度 (m)
                        'inflow': node.inflow,      # 入流量 (m³/s)
                        'overflow': node.overflow   # 溢流量 (m³/s)
                    })
        sim.close()
        # 分析结果
        df_results = pd.DataFrame(flooding_results)
        # 统计严重积水点
        severe_flooding = df_results[df_results['water_depth'] > 0.3]
        return {
            'all_flooding': df_results,
            'severe_flooding': severe_flooding,
            'total_overflow_volume': df_results['overflow'].sum(),
            'max_water_depth': df_results['water_depth'].max(),
            'flooding_duration': df_results['timestamp'].max() - df_results['timestamp'].min()
        }
    def generate_early_warning(self, forecast_rainfall):
        """
        生成内涝预警
        参数:
            forecast_rainfall: 未来 6 小时降雨预报
        返回:
            预警等级、受影响区域、建议措施
        """
        results = self.simulate_flooding(forecast_rainfall, duration_hours=6)
        max_depth = results['max_water_depth']
        # 预警等级判定
        if max_depth > 0.5:
            warning_level = '红色预警'
            action = '立即启动应急预案，交通管制，人员疏散'
        elif max_depth > 0.3:
            warning_level = '橙色预警'
            action = '加强巡查，准备排涝设备'
        elif max_depth > 0.15:
            warning_level = '黄色预警'
            action = '密切关注，做好应急准备'
        else:
            warning_level = '蓝色预警'
            action = '正常 monitoring'
        return {
            'warning_level': warning_level,
            'max_water_depth': max_depth,
            'affected_areas': len(results['severe_flooding']['node_id'].unique()),
            'recommended_action': action,
            'forecast_time': pd.Timestamp.now() + pd.Timedelta(hours=6)
        }
# 使用示例
simulator = UrbanFloodSimulation('wuhan_drainage.inp')
# 模拟暴雨情景
rainfall = [0, 0, 10, 30, 50, 80, 60, 40, 20, 10, 0, 0]  # 12 小时降雨 (mm/h)
results = simulator.simulate_flooding(rainfall)
print(f"最大积水深度: {results['max_water_depth']:.2f} m")
print(f"严重积水点数: {len(results['severe_flooding'])}")
# 生成预警
warning = simulator.generate_early_warning(rainfall[:6])
print(f"预警等级: {warning['warning_level']}")
print(f"建议措施: {warning['recommended_action']}")

def validate_sensor_data(value, min_val=0, max_val=50):
    """数据有效性校验"""
    if value < min_val or value > max_val:
        return None  # 标记为无效
    return value
def data_imputation(data_series, method='interpolation'):
    """缺失值填补"""
    if method == 'interpolation':
        # 线性插值
        return data_series.interpolate(method='linear')
    elif method == 'forward_fill':
        # 前向填充
        return data_series.ffill()
    elif method == 'model_prediction':
        # 用 AI 模型预测
        return predict_missing_values(data_series)

# 方案 1: 增加特征工程
features = [
    'water_level', 'rainfall', 'flow_rate',
    'soil_moisture', 'groundwater_level',  # 新增
    'upstream_reservoir_release',           # 上游水库泄流
    'tide_level',                           # 潮汐影响（下游）
    'land_use_type'                         # 土地利用类型
]
# 方案 2: 模型融合
from sklearn.ensemble import VotingRegressor
ensemble_model = VotingRegressor([
    ('lstm', lstm_model),
    ('xgboost', xgb_model),
    ('prophet', prophet_model)
])
# 方案 3: 在线学习（持续更新模型）
model.partial_fit(new_data)  # 增量训练

# 方案 1: 消息队列削峰填谷
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='kafka:9092')
def send_sensor_data(data):
    producer.send('sensor-data-topic', value=json.dumps(data).encode())
# 方案 2: 时序数据库优化
# 使用 TDengine 替代 MySQL
# TDengine 写入性能：100 万条/秒
# 方案 3: 边缘计算
# 在 RTU 端进行数据预处理，只上传变化数据
if abs(current_value - last_value) < threshold:
    skip_upload()  # 跳过上传

1. **网络隔离**
- 生产网与办公网物理隔离
- 部署工业防火墙
2. **数据加密**
- 传输层：TLS 1.3
- 存储层：AES-256 加密
3. **身份认证**
- 双因素认证（密码 + 短信验证码）
- 基于角色的访问控制（RBAC）
4. **安全审计**
- 记录所有操作日志
- 异常行为检测（如深夜登录）
5. **应急响应**
- 定期安全演练
- 备份恢复机制

# GitHub 地址 (待更新)
# 包含内容：
# - LSTM 洪水预测模型
# - 水库调度优化算法
# - 堤防 AI 识别模型
# - 城市内涝 SWMM 模拟