(self, text, token, device)
| 126 | self.text_encoder.load_state_dict(checkpoint['text_encoder']) |
| 127 | |
| 128 | def forward(self, text, token, device): |
| 129 | B = len(text) |
| 130 | pos_one_hot = [] |
| 131 | word_emb = [] |
| 132 | sent_len = [] |
| 133 | for i in range(B): |
| 134 | tokens = token[i].split(" ") |
| 135 | if len(tokens) < self.max_text_len: |
| 136 | tokens = ['sos/OTHER'] + tokens + ['eos/OTHER'] |
| 137 | batch_sent_len = len(tokens) |
| 138 | tokens = tokens + ['unk/OTHER'] * (self.max_text_len + 2 - batch_sent_len) |
| 139 | else: |
| 140 | tokens = tokens[: self.max_text_len] |
| 141 | tokens = ['sos/OTHER'] + tokens + ['eos/OTHER'] |
| 142 | batch_sent_len = len(tokens) |
| 143 | sent_len.append(batch_sent_len) |
| 144 | batch_word_emb = [] |
| 145 | batch_pos_one_hot = [] |
| 146 | for cur_token in tokens: |
| 147 | cur_word_emb, cur_pos_one_hot = self.w_vectorizer[cur_token] |
| 148 | cur_word_emb = torch.from_numpy(cur_word_emb).float() |
| 149 | cur_pos_one_hot = torch.from_numpy(cur_pos_one_hot).float() |
| 150 | batch_word_emb.append(cur_word_emb) |
| 151 | batch_pos_one_hot.append(cur_pos_one_hot) |
| 152 | |
| 153 | batch_word_emb = torch.stack(batch_word_emb, dim=0) |
| 154 | batch_pos_one_hot = torch.stack(batch_pos_one_hot, dim=0) |
| 155 | word_emb.append(batch_word_emb) |
| 156 | pos_one_hot.append(batch_pos_one_hot) |
| 157 | word_emb = torch.stack(word_emb, dim=0).to(device) |
| 158 | pos_one_hot = torch.stack(pos_one_hot, dim=0).to(device) |
| 159 | sent_len = torch.tensor(sent_len, dtype=torch.long).to(device) |
| 160 | text_embedding = self.text_encoder(word_emb, pos_one_hot, sent_len) |
| 161 | return text_embedding |
| 162 | |
| 163 | |
| 164 | class TextEncoderBiGRUCo(nn.Module): |
nothing calls this directly
no outgoing calls
no test coverage detected