r""" `FluxMultiControlNetModel` wrapper class for Multi-FluxControlNetModel This module is a wrapper for multiple instances of the `FluxControlNetModel`. The `forward()` API is designed to be compatible with `FluxControlNetModel`. Args: controlnets (`List[FluxControlNetMode
| 414 | |
| 415 | |
| 416 | class FluxMultiControlNetModel(ModelMixin): |
| 417 | r""" |
| 418 | `FluxMultiControlNetModel` wrapper class for Multi-FluxControlNetModel |
| 419 | |
| 420 | This module is a wrapper for multiple instances of the `FluxControlNetModel`. The `forward()` API is designed to be |
| 421 | compatible with `FluxControlNetModel`. |
| 422 | |
| 423 | Args: |
| 424 | controlnets (`List[FluxControlNetModel]`): |
| 425 | Provides additional conditioning to the unet during the denoising process. You must set multiple |
| 426 | `FluxControlNetModel` as a list. |
| 427 | """ |
| 428 | |
| 429 | def __init__(self, controlnets, union=False): |
| 430 | super().__init__() |
| 431 | self.nets = nn.ModuleList(controlnets) |
| 432 | self.union = union |
| 433 | |
| 434 | def forward( |
| 435 | self, |
| 436 | hidden_states: torch.FloatTensor, |
| 437 | controlnet_cond: List[torch.tensor], |
| 438 | controlnet_mode: List[torch.tensor], |
| 439 | conditioning_scale: List[float], |
| 440 | encoder_hidden_states: torch.Tensor = None, |
| 441 | pooled_projections: torch.Tensor = None, |
| 442 | timestep: torch.LongTensor = None, |
| 443 | img_ids: torch.Tensor = None, |
| 444 | txt_ids: torch.Tensor = None, |
| 445 | guidance: torch.Tensor = None, |
| 446 | joint_attention_kwargs: Optional[Dict[str, Any]] = None, |
| 447 | return_dict: bool = True, |
| 448 | ) -> Union[FluxControlNetOutput, Tuple]: |
| 449 | # ControlNet-Union with multiple conditions |
| 450 | # only load one ControlNet for saving memories |
| 451 | if len(self.nets) == 1: |
| 452 | controlnet = self.nets[0] |
| 453 | for i, (image, mode, scale) in enumerate(zip(controlnet_cond, controlnet_mode, conditioning_scale)): |
| 454 | |
| 455 | block_samples, single_block_samples = controlnet( |
| 456 | hidden_states=hidden_states, |
| 457 | controlnet_cond=image, |
| 458 | controlnet_mode=mode[:, None], |
| 459 | conditioning_scale=scale, |
| 460 | timestep=timestep, |
| 461 | guidance=guidance, |
| 462 | pooled_projections=pooled_projections, |
| 463 | encoder_hidden_states=encoder_hidden_states, |
| 464 | txt_ids=txt_ids, |
| 465 | img_ids=img_ids, |
| 466 | joint_attention_kwargs=joint_attention_kwargs, |
| 467 | return_dict=return_dict, |
| 468 | ) |
| 469 | |
| 470 | # merge samples |
| 471 | if i == 0: |
| 472 | control_block_samples = block_samples |
| 473 | control_single_block_samples = single_block_samples |
nothing calls this directly
no outgoing calls
no test coverage detected