基于深度学习的多任务空气质量预测模型设计与实现+高分项目+源码.zip

# coding:utf-8 import numpy as np import torch import torch.nn as nn import torch.nn.functional as F import torch.optim as optim from torch.autograd import Variable from torch.nn.utils import weight_norm # from torch_geometric.nn import GCNConv #GCN相关 # class GCN(torch.nn.Module): # def __init__(self, num_node_features, num_classes): # super(GCN, self).__init__() # self.conv1 = GCNConv(num_node_features, 16) # self.conv2 = GCNConv(16, num_classes) # def forward(self, data): # x, edge_index = data.x, data.edge_index # x = self.conv1(x, edge_index) # x = F.relu(x) # x = F.dropout(x, training=self.training) # x = self.conv2(x, edge_index) # x = F.relu(x) # x = F.dropout(x, training=self.training) # x = F.softmax(x, dim=1) # return x class SimpleRNN(nn.Module): def __init__(self, input_size, hidden_size=32, output_size=1, num_layers=1, dropout=0.25): super(SimpleRNN, self).__init__() self.hidden_size = hidden_size self.rnn = nn.RNN( input_size=input_size, hidden_size=hidden_size, nonlinearity='relu', # 'tanh' or 'relu' num_layers=num_layers, dropout=dropout, batch_first=True ) self.linear = nn.Linear(hidden_size, output_size) def forward(self, x, hidden): x = x.permute(1,0,2,3)#rnn（以及lstm）需要这样子的数据格式 多了一个 18站点那个维度 x_input = x[0] output, hidden = self.rnn(x_input, hidden) pred = self.linear(output[:, -1, :]) return pred, hidden def init_hidden(self): return torch.randn(1, 24, self.hidden_size) class SimpleGRU(nn.Module): def __init__(self, input_size, hidden_size, output_size=1, num_layers=1, dropout=0.25): super(SimpleGRU, self).__init__() self.hidden_size = hidden_size self.gru = nn.GRU( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, # dropout=dropout, batch_first=True ) self.linear = nn.Linear(hidden_size, output_size) def forward(self, x, hidden): x = x.permute(1,0,2,3) x_input = x[0] output, hidden = self.gru(x_input, hidden) pred = self.linear(output[:, -1, :]) return pred, hidden def init_hidden(self): return torch.randn(1, 24, self.hidden_size) class SimpleLSTM(nn.Module): def __init__(self, input_size, hidden_size, output_size=1, num_layers=1, dropout=0.25): super(SimpleLSTM, self).__init__() self.hidden_size = hidden_size self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, # dropout=dropout, batch_first=True ) self.linear = nn.Linear(hidden_size, output_size) def forward(self, x): x = x.permute(1,0,2,3) x_input = x[0] output, (h_n, c_n) = self.lstm(x_input) pred = self.linear(output[:, -1, :]) return pred def init_hidden(self): return torch.randn(1, 24, self.hidden_size) ###################### class TCN(nn.Module): def __init__(self, input_size, output_size, num_channels, kernel_size, dropout): super(TCN, self).__init__() self.tcn = TemporalConvNet(input_size, num_channels, kernel_size, dropout=dropout) self.linear = nn.Linear(num_channels[-1], output_size) def forward(self, x): output = self.tcn(x.transpose(1, 2)).transpose(1, 2) pred = self.linear(output[:, -1, :]) return pred class Chomp1d(nn.Module): def __init__(self, chomp_size): super(Chomp1d, self).__init__() self.chomp_size = chomp_size def forward(self, x): return x[:, :, :-self.chomp_size].contiguous() class TemporalBlock(nn.Module): def __init__(self, n_inputs, n_outputs, kernel_size, stride, dilation, padding, dropout=0.2): super(TemporalBlock, self).__init__() self.conv1 = weight_norm(nn.Conv1d(n_inputs, n_outputs, kernel_size, stride=stride, padding=padding, dilation=dilation)) self.chomp1 = Chomp1d(padding) self.relu1 = nn.ReLU() self.dropout1 = nn.Dropout(dropout) self.conv2 = weight_norm(nn.Conv1d(n_outputs, n_outputs, kernel_size, stride=stride, padding=padding, dilation=dilation)) self.chomp2 = Chomp1d(padding) self.relu2 = nn.ReLU() self.dropout2 = nn.Dropout(dropout) self.net = nn.Sequential(self.conv1, self.chomp1, self.relu1, self.dropout1, self.conv2, self.chomp2, self.relu2, self.dropout2) self.downsample = nn.Conv1d(n_inputs, n_outputs, 1) if n_inputs != n_outputs else None self.relu = nn.ReLU() self.init_weights() def init_weights(self): self.conv1.weight.data.normal_(0, 0.01) self.conv2.weight.data.normal_(0, 0.01) if self.downsample is not None: self.downsample.weight.data.normal_(0, 0.01) def forward(self, x): out = self.net(x) res = x if self.downsample is None else self.downsample(x) return self.relu(out + res) class TemporalConvNet(nn.Module): def __init__(self, num_inputs, num_channels, kernel_size=2, dropout=0.2): super(TemporalConvNet, self).__init__() layers = [] num_levels = len(num_channels) for i in range(num_levels): dilation_size = 2 ** i in_channels = num_inputs if i == 0 else num_channels[i-1] out_channels = num_channels[i] layers += [TemporalBlock(in_channels, out_channels, kernel_size, stride=1, dilation=dilation_size, padding=(kernel_size-1) * dilation_size, dropout=dropout)] self.network = nn.Sequential(*layers) def forward(self, x): return self.network(x) class STCN(nn.Module): def __init__(self, input_size, in_channels, output_size, num_channels, kernel_size, dropout): super(STCN, self).__init__() self.conv = torch.nn.Sequential( torch.nn.Conv2d(in_channels=in_channels, out_channels=64, kernel_size=(1, 1), stride=1, padding=0), torch.nn.BatchNorm2d(64), torch.nn.ReLU(), torch.nn.Conv2d(in_channels=64, out_channels=1, kernel_size=(1, 1), stride=1, padding=0), torch.nn.BatchNorm2d(1), torch.nn.ReLU() ) self.tcn = TemporalConvNet(input_size, num_channels, kernel_size, dropout=dropout) self.linear = nn.Linear(num_channels[-1], output_size) def forward(self, x): conv_out = self.conv(x).squeeze(0) output = self.tcn(conv_out.transpose(1, 2)).transpose(1, 2) pred = self.linear(output[:, -1, :]) return pred