(
self,
sample, timestep, encoder_hidden_states, add_time_id, add_text_embeds,
tiled=False, tile_size=64, tile_stride=8,
use_gradient_checkpointing=False,
**kwargs
)
| 86 | self.is_kolors = is_kolors |
| 87 | |
| 88 | def forward( |
| 89 | self, |
| 90 | sample, timestep, encoder_hidden_states, add_time_id, add_text_embeds, |
| 91 | tiled=False, tile_size=64, tile_stride=8, |
| 92 | use_gradient_checkpointing=False, |
| 93 | **kwargs |
| 94 | ): |
| 95 | # 1. time |
| 96 | t_emb = self.time_proj(timestep).to(sample.dtype) |
| 97 | t_emb = self.time_embedding(t_emb) |
| 98 | |
| 99 | time_embeds = self.add_time_proj(add_time_id) |
| 100 | time_embeds = time_embeds.reshape((add_text_embeds.shape[0], -1)) |
| 101 | add_embeds = torch.concat([add_text_embeds, time_embeds], dim=-1) |
| 102 | add_embeds = add_embeds.to(sample.dtype) |
| 103 | add_embeds = self.add_time_embedding(add_embeds) |
| 104 | |
| 105 | time_emb = t_emb + add_embeds |
| 106 | |
| 107 | # 2. pre-process |
| 108 | height, width = sample.shape[2], sample.shape[3] |
| 109 | hidden_states = self.conv_in(sample) |
| 110 | text_emb = encoder_hidden_states if self.text_intermediate_proj is None else self.text_intermediate_proj(encoder_hidden_states) |
| 111 | res_stack = [hidden_states] |
| 112 | |
| 113 | # 3. blocks |
| 114 | def create_custom_forward(module): |
| 115 | def custom_forward(*inputs): |
| 116 | return module(*inputs) |
| 117 | return custom_forward |
| 118 | for i, block in enumerate(self.blocks): |
| 119 | if self.training and use_gradient_checkpointing and not (isinstance(block, PushBlock) or isinstance(block, PopBlock)): |
| 120 | hidden_states, time_emb, text_emb, res_stack = torch.utils.checkpoint.checkpoint( |
| 121 | create_custom_forward(block), |
| 122 | hidden_states, time_emb, text_emb, res_stack, |
| 123 | use_reentrant=False, |
| 124 | ) |
| 125 | else: |
| 126 | hidden_states, time_emb, text_emb, res_stack = block( |
| 127 | hidden_states, time_emb, text_emb, res_stack, |
| 128 | tiled=tiled, tile_size=tile_size, tile_stride=tile_stride |
| 129 | ) |
| 130 | |
| 131 | # 4. output |
| 132 | hidden_states = self.conv_norm_out(hidden_states) |
| 133 | hidden_states = self.conv_act(hidden_states) |
| 134 | hidden_states = self.conv_out(hidden_states) |
| 135 | |
| 136 | return hidden_states |
| 137 | |
| 138 | @staticmethod |
| 139 | def state_dict_converter(): |
nothing calls this directly
no test coverage detected