r""" Default processor for performing attention-related computations.
| 431 | |
| 432 | |
| 433 | class AttnProcessor: |
| 434 | r""" |
| 435 | Default processor for performing attention-related computations. |
| 436 | """ |
| 437 | |
| 438 | def __call__( |
| 439 | self, |
| 440 | attn: Attention, |
| 441 | hidden_states, |
| 442 | encoder_hidden_states=None, |
| 443 | attention_mask=None, |
| 444 | temb=None, |
| 445 | ): |
| 446 | residual = hidden_states |
| 447 | |
| 448 | if attn.spatial_norm is not None: |
| 449 | hidden_states = attn.spatial_norm(hidden_states, temb) |
| 450 | |
| 451 | input_ndim = hidden_states.ndim |
| 452 | |
| 453 | if input_ndim == 4: |
| 454 | batch_size, channel, height, width = hidden_states.shape |
| 455 | hidden_states = hidden_states.view(batch_size, channel, height * width).transpose(1, 2) |
| 456 | |
| 457 | batch_size, sequence_length, _ = ( |
| 458 | hidden_states.shape if encoder_hidden_states is None else encoder_hidden_states.shape |
| 459 | ) |
| 460 | attention_mask = attn.prepare_attention_mask(attention_mask, sequence_length, batch_size) |
| 461 | |
| 462 | if attn.group_norm is not None: |
| 463 | hidden_states = attn.group_norm(hidden_states.transpose(1, 2)).transpose(1, 2) |
| 464 | |
| 465 | query = attn.to_q(hidden_states) |
| 466 | |
| 467 | if encoder_hidden_states is None: |
| 468 | encoder_hidden_states = hidden_states |
| 469 | elif attn.norm_cross: |
| 470 | encoder_hidden_states = attn.norm_encoder_hidden_states(encoder_hidden_states) |
| 471 | |
| 472 | key = attn.to_k(encoder_hidden_states) |
| 473 | value = attn.to_v(encoder_hidden_states) |
| 474 | |
| 475 | query = attn.head_to_batch_dim(query) |
| 476 | key = attn.head_to_batch_dim(key) |
| 477 | value = attn.head_to_batch_dim(value) |
| 478 | |
| 479 | attention_probs = attn.get_attention_scores(query, key, attention_mask) |
| 480 | hidden_states = torch.bmm(attention_probs, value) |
| 481 | hidden_states = attn.batch_to_head_dim(hidden_states) |
| 482 | |
| 483 | # linear proj |
| 484 | hidden_states = attn.to_out[0](hidden_states) |
| 485 | # dropout |
| 486 | hidden_states = attn.to_out[1](hidden_states) |
| 487 | |
| 488 | if input_ndim == 4: |
| 489 | hidden_states = hidden_states.transpose(-1, -2).reshape(batch_size, channel, height, width) |
| 490 |
no outgoing calls
no test coverage detected