(
self,
hidden_states: torch.FloatTensor,
num_frames: int,
encoder_hidden_states: Optional[torch.FloatTensor] = None,
)
| 434 | self._chunk_dim = 1 |
| 435 | |
| 436 | def forward( |
| 437 | self, |
| 438 | hidden_states: torch.FloatTensor, |
| 439 | num_frames: int, |
| 440 | encoder_hidden_states: Optional[torch.FloatTensor] = None, |
| 441 | ) -> torch.FloatTensor: |
| 442 | # Notice that normalization is always applied before the real computation in the following blocks. |
| 443 | # 0. Self-Attention |
| 444 | batch_size = hidden_states.shape[0] |
| 445 | |
| 446 | batch_frames, seq_length, channels = hidden_states.shape |
| 447 | batch_size = batch_frames // num_frames |
| 448 | |
| 449 | hidden_states = hidden_states[None, :].reshape(batch_size, num_frames, seq_length, channels) |
| 450 | hidden_states = hidden_states.permute(0, 2, 1, 3) |
| 451 | hidden_states = hidden_states.reshape(batch_size * seq_length, num_frames, channels) |
| 452 | |
| 453 | residual = hidden_states |
| 454 | hidden_states = self.norm_in(hidden_states) |
| 455 | |
| 456 | if self._chunk_size is not None: |
| 457 | hidden_states = _chunked_feed_forward(self.ff, hidden_states, self._chunk_dim, self._chunk_size) |
| 458 | else: |
| 459 | hidden_states = self.ff_in(hidden_states) |
| 460 | |
| 461 | if self.is_res: |
| 462 | hidden_states = hidden_states + residual |
| 463 | |
| 464 | norm_hidden_states = self.norm1(hidden_states) |
| 465 | attn_output = self.attn1(norm_hidden_states, encoder_hidden_states=None) |
| 466 | hidden_states = attn_output + hidden_states |
| 467 | |
| 468 | # 3. Cross-Attention |
| 469 | if self.attn2 is not None: |
| 470 | norm_hidden_states = self.norm2(hidden_states) |
| 471 | attn_output = self.attn2(norm_hidden_states, encoder_hidden_states=encoder_hidden_states) |
| 472 | hidden_states = attn_output + hidden_states |
| 473 | |
| 474 | # 4. Feed-forward |
| 475 | norm_hidden_states = self.norm3(hidden_states) |
| 476 | |
| 477 | if self._chunk_size is not None: |
| 478 | ff_output = _chunked_feed_forward(self.ff, norm_hidden_states, self._chunk_dim, self._chunk_size) |
| 479 | else: |
| 480 | ff_output = self.ff(norm_hidden_states) |
| 481 | |
| 482 | if self.is_res: |
| 483 | hidden_states = ff_output + hidden_states |
| 484 | else: |
| 485 | hidden_states = ff_output |
| 486 | |
| 487 | hidden_states = hidden_states[None, :].reshape(batch_size, seq_length, num_frames, channels) |
| 488 | hidden_states = hidden_states.permute(0, 2, 1, 3) |
| 489 | hidden_states = hidden_states.reshape(batch_size * num_frames, seq_length, channels) |
| 490 | |
| 491 | return hidden_states |
| 492 | |
| 493 |
nothing calls this directly
no test coverage detected