(
self,
x,
e,
seq_lens,
freqs,
context,
context_lens,
ref_motion, # [B, T, C]
ref_motion_lens, # [B]
attend_to_text_mask=None, # [B]: 1 for attend to text, 0 for attend to ref motion
)
| 380 | |
| 381 | |
| 382 | def forward( |
| 383 | self, |
| 384 | x, |
| 385 | e, |
| 386 | seq_lens, |
| 387 | freqs, |
| 388 | context, |
| 389 | context_lens, |
| 390 | ref_motion, # [B, T, C] |
| 391 | ref_motion_lens, # [B] |
| 392 | attend_to_text_mask=None, # [B]: 1 for attend to text, 0 for attend to ref motion |
| 393 | ): |
| 394 | assert e.dtype == torch.float32 |
| 395 | with amp.autocast(dtype=torch.float32, device_type="cuda"): |
| 396 | e = (self.modulation.to(dtype=e.dtype, device=e.device) + e).chunk(6, dim=1) |
| 397 | assert e[0].dtype == torch.float32 |
| 398 | |
| 399 | # self-attention |
| 400 | y = self.self_attn( |
| 401 | self.norm1(x).float() * (1 + e[1]) + e[0], seq_lens, |
| 402 | freqs) |
| 403 | |
| 404 | with amp.autocast(dtype=torch.float32, device_type="cuda"): |
| 405 | x = x + y * e[2] |
| 406 | |
| 407 | # Cross-attention based on attend_to_text_mask |
| 408 | text_mask = attend_to_text_mask == 1 |
| 409 | ref_motion_mask = ~text_mask |
| 410 | |
| 411 | # Dummy way to get output shape |
| 412 | b, s, d = ref_motion.shape |
| 413 | # combined_out = self.cross_attn(self.norm3(x), context, None) |
| 414 | combined_out = torch.zeros([b, s, self.dim], dtype=context.dtype, device=context.device) |
| 415 | |
| 416 | # Process text and reference motion separately |
| 417 | attend_to_text_mask = attend_to_text_mask.to(dtype=bool) |
| 418 | have_text = attend_to_text_mask.sum() > 0 |
| 419 | have_ref_motion = (~attend_to_text_mask).sum() > 0 |
| 420 | if have_text: |
| 421 | text_out = self.cross_attn(self.norm3(x[text_mask]), context[text_mask], None) |
| 422 | # print('text_out shape', text_out.shape) |
| 423 | combined_out[text_mask] = text_out |
| 424 | |
| 425 | if have_ref_motion: |
| 426 | ref_motion_out = self.ref_motion_cross_attn(self.norm4(x[ref_motion_mask]), ref_motion[ref_motion_mask], ref_motion_lens[ref_motion_mask]) |
| 427 | # print('ref_motion_out shape', ref_motion_out.shape) |
| 428 | combined_out[ref_motion_mask] = ref_motion_out if x[ref_motion_mask].size(0) > 0 else 0 |
| 429 | |
| 430 | x += combined_out |
| 431 | |
| 432 | # ffn function |
| 433 | y = self.ffn(self.norm2(x).float() * (1 + e[4]) + e[3]) |
| 434 | with amp.autocast(dtype=torch.float32, device_type="cuda"): |
| 435 | x = x + y * e[5] |
| 436 | return x |
| 437 | |
| 438 | |
| 439 | class WanAttentionBlock_SAonly(nn.Module): |
nothing calls this directly
no outgoing calls
no test coverage detected