(
self,
hidden_states,
encoder_hidden_states=None,
attention_mask=None,
video_length=None,
**cross_attention_kwargs,
)
| 348 | self.set_processor(processor) |
| 349 | |
| 350 | def forward( |
| 351 | self, |
| 352 | hidden_states, |
| 353 | encoder_hidden_states=None, |
| 354 | attention_mask=None, |
| 355 | video_length=None, |
| 356 | **cross_attention_kwargs, |
| 357 | ): |
| 358 | if self.attention_mode == "Temporal": |
| 359 | d = hidden_states.shape[1] # d means HxW |
| 360 | hidden_states = rearrange( |
| 361 | hidden_states, "(b f) d c -> (b d) f c", f=video_length |
| 362 | ) |
| 363 | |
| 364 | if self.pos_encoder is not None: |
| 365 | hidden_states = self.pos_encoder(hidden_states) |
| 366 | |
| 367 | encoder_hidden_states = ( |
| 368 | repeat(encoder_hidden_states, "b n c -> (b d) n c", d=d) |
| 369 | if encoder_hidden_states is not None |
| 370 | else encoder_hidden_states |
| 371 | ) |
| 372 | |
| 373 | else: |
| 374 | raise NotImplementedError |
| 375 | |
| 376 | hidden_states = self.processor( |
| 377 | self, |
| 378 | hidden_states, |
| 379 | encoder_hidden_states=encoder_hidden_states, |
| 380 | attention_mask=attention_mask, |
| 381 | **cross_attention_kwargs, |
| 382 | ) |
| 383 | |
| 384 | if self.attention_mode == "Temporal": |
| 385 | hidden_states = rearrange(hidden_states, "(b d) f c -> (b f) d c", d=d) |
| 386 | |
| 387 | return hidden_states |
nothing calls this directly
no outgoing calls
no test coverage detected