摘要

本周利用阳朔水站的数据对1D浅水方程进行了求解，通过将1D浅水方程的物理约束纳入到神经网络，将时空坐标x，t作为输入到神经网络中，得到预测解水深和流量的数据，然后利用真实的数据进行比较，计算损失。

ABSTRACT

This week, the data of Yangshuo water station was used to solve the 1D shallow water equation, and the physical constraints of the 1D shallow water equation were incorporated into the neural network, and the spatiotemporal coordinates x,t were used as inputs to the neural network to obtain the data that predicted the water depth and flow rate of the solution, and then the real data were used to compare and calculate the loss.

具体实现

一、数据处理

数据被从Excel文件加载、预处理（例如清除缺失值、转换日期时间格式、创建假设的空间坐标、标准化或归一化等），然后被分割为训练集和测试集。

import pandas as pd
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import train_test_split

def preprocess_data(filepath):
    data = pd.read_excel(filepath)
    data.dropna(inplace=True)  # 删除缺失值
    data['DateTime'] = data['DateTime'].astype('int64') // 10**9

    # 假设空间坐标
    data['Space'] = range(len(data))

    # 特征和标签
    X = data[['Space', 'DateTime']]
    y = data[['WaterLevel', '流量']]

    # 数据标准化
    x_scaler = MinMaxScaler()
    y_scaler = StandardScaler()
    X_scaled = x_scaler.fit_transform(X)
    y_scaled = y_scaler.fit_transform(y)

    # 分割数据为训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(X_scaled, y_scaled, test_size=0.2, random_state=42)

    return X_train, X_test, y_train, y_test, x_scaler, y_scaler

二、model

import torch
import torch.nn as nn
import torch.autograd as autograd

class PINN(nn.Module):
    def __init__(self, gravity=9.81):
        super(PINN, self).__init__()
        self.gravity = gravity
        self.dense1 = nn.Linear(2, 50)  # 假设输入是位置和时间
        self.dense2 = nn.Linear(50, 50)
        self.dense3 = nn.Linear(50, 50)
        self.out = nn.Linear(50, 2)  # 输出水位h和速度u

        self.activation = nn.Tanh()

    def forward(self, inputs):
        x = self.activation(self.dense1(inputs))
        x = self.activation(self.dense2(x))
        x = self.activation(self.dense3(x))
        return self.out(x)

    def physics(self, inputs):
        inputs.requires_grad_(True)
        h_u = self(inputs)
        h = h_u[:, 0]
        u = h_u[:, 1]

        h_t = autograd.grad(h.sum(), inputs, create_graph=True)[0][:, 1]
        h_x = autograd.grad(h.sum(), inputs, create_graph=True)[0][:, 0]
        u_t = autograd.grad(u.sum(), inputs, create_graph=True)[0][:, 1]
        u_x = autograd.grad(u.sum(), inputs, create_graph=True)[0][:, 0]

        continuity = h_t + (h * u_x + u * h_x)
        momentum = (h * u).t() + autograd.grad((h * u**2 + 0.5 * self.gravity * h**2).sum(), inputs, create_graph=True)[0][:, 0]

        return continuity, momentum

    def loss(self, inputs, outputs, targets):
        prediction_loss = nn.MSELoss()(outputs, targets)
        continuity_res, momentum_res = self.physics(inputs)
        physics_loss = torch.mean(continuity_res**2 + momentum_res**2)

        total_loss = prediction_loss + physics_loss
        return total_loss

三、train

import torch
from torch.optim import Adam
from model import PINN
from data_preprocess import preprocess_data
from config import learning_rate, epochs, batch_size

def train_model(X_train, y_train):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = PINN().to(device)
    optimizer = Adam(model.parameters(), lr=learning_rate)

    model.train()
    for epoch in range(epochs):
        optimizer.zero_grad()
        inputs = torch.tensor(X_train).float().to(device)
        targets = torch.tensor(y_train).float().to(device)
        outputs = model(inputs)
        loss = model.loss(inputs, outputs, targets)
        loss.backward()
        optimizer.step()

        if epoch % 100 == 0:
            print(f"Epoch {epoch}, Loss: {loss.item()}")

    return model

if __name__ == "__main__":
    X_train, _, y_train, _, _, _ = preprocess_data('./阳朔水站流量流速数据.xlsx')
    trained_model = train_model(X_train, y_train)
    torch.save(trained_model.state_dict(), 'trained_pinn_model.pth')

训练结果：

四、test

import torch
from model import PINN
from data_preprocess import preprocess_data

def test_model(model_path, X_test, y_test):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = PINN().to(device)
    model.load_state_dict(torch.load(model_path))
    # 保留模型在训练模式以保持梯度计算，或者确保模型中的物理损失计算适当处理
    model.train()

    # 确保输入启用梯度计算
    inputs = torch.tensor(X_test, requires_grad=True).float().to(device)
    targets = torch.tensor(y_test).float().to(device)

    predictions = model(inputs)
    test_loss = model.loss(inputs, predictions, targets)  # 这里计算包括物理损失在内的总损失
    print(f"Test Loss: {test_loss.item()}")

if __name__ == "__main__":
    _, X_test, _, y_test, _, _ = preprocess_data('./阳朔水站流量流速数据.xlsx')
    model_path = 'trained_pinn_model.pth'
    test_model(model_path, X_test, y_test)

测试结果：                                                                  可视化结果：