| 115 | |
| 116 | |
| 117 | class LstmGat(nn.Module): |
| 118 | def __init__(self, seq, n_fea, adj_dense, adj_sparse): |
| 119 | super(LstmGat, self).__init__() |
| 120 | self.nodes = adj_dense.shape[0] |
| 121 | self.gcn = nn.Linear(in_features=seq - n_fea + 1, out_features=seq - n_fea + 1, device=device) |
| 122 | self.encoder = nn.Conv2d(self.nodes, self.nodes, (n_fea, n_fea), device=device) |
| 123 | self.gat_l1 = models.MultiHeadsGATLayer(adj_sparse, seq - n_fea + 1, seq - n_fea + 1, 4, 0, 0.2) |
| 124 | self.gat_l2 = models.MultiHeadsGATLayer(adj_sparse, seq - n_fea + 1, seq - n_fea + 1, 4, 0, 0.2) |
| 125 | self.lstm = nn.LSTM(self.nodes, self.nodes, num_layers=2, batch_first=True) |
| 126 | self.decoder = nn.Linear(seq - n_fea + 1, 1, device=device) |
| 127 | |
| 128 | # Activation |
| 129 | self.dropout = nn.Dropout(p=0.5) |
| 130 | self.LeakyReLU = nn.LeakyReLU() |
| 131 | |
| 132 | def forward(self, occ, prc): # occ.shape = [batch, node, seq] |
| 133 | x = torch.stack([occ, prc], dim=3) |
| 134 | x = self.encoder(x) |
| 135 | x = torch.squeeze(x) |
| 136 | |
| 137 | # first layer |
| 138 | atts_mat = self.gat_l1(x) # attention matrix, dense(nodes, nodes) |
| 139 | occ_conv1 = torch.matmul(atts_mat, x) # (b, n, s) |
| 140 | occ_conv1 = self.dropout(self.LeakyReLU(self.gcn(occ_conv1))) |
| 141 | |
| 142 | # second layer |
| 143 | atts_mat2 = self.gat_l2(occ_conv1) # attention matrix, dense(nodes, nodes) |
| 144 | occ_conv2 = torch.matmul(atts_mat2, occ_conv1) # (b, n, s) |
| 145 | occ_conv2 = self.dropout(self.LeakyReLU(self.gcn(occ_conv2))) |
| 146 | |
| 147 | # lstm |
| 148 | x = occ_conv2.transpose(1, 2) |
| 149 | x, _ = self.lstm(x) |
| 150 | x = x.transpose(x, 1, 2) |
| 151 | |
| 152 | # decode |
| 153 | x = self.decoder(x) |
| 154 | x = torch.squeeze(x) |
| 155 | return x |
| 156 | |
| 157 | |
| 158 | class TPA(nn.Module): |
nothing calls this directly
no outgoing calls
no test coverage detected