(
self,
hidden_states, time_emb, text_emb, res_stack,
cross_frame_attention=False,
tiled=False, tile_size=64, tile_stride=32,
ipadapter_kwargs_list={},
**kwargs
)
| 146 | self.proj_out = torch.nn.Linear(inner_dim, in_channels) |
| 147 | |
| 148 | def forward( |
| 149 | self, |
| 150 | hidden_states, time_emb, text_emb, res_stack, |
| 151 | cross_frame_attention=False, |
| 152 | tiled=False, tile_size=64, tile_stride=32, |
| 153 | ipadapter_kwargs_list={}, |
| 154 | **kwargs |
| 155 | ): |
| 156 | batch, _, height, width = hidden_states.shape |
| 157 | residual = hidden_states |
| 158 | |
| 159 | hidden_states = self.norm(hidden_states) |
| 160 | inner_dim = hidden_states.shape[1] |
| 161 | hidden_states = hidden_states.permute(0, 2, 3, 1).reshape(batch, height * width, inner_dim) |
| 162 | hidden_states = self.proj_in(hidden_states) |
| 163 | |
| 164 | if cross_frame_attention: |
| 165 | hidden_states = hidden_states.reshape(1, batch * height * width, inner_dim) |
| 166 | encoder_hidden_states = text_emb.mean(dim=0, keepdim=True) |
| 167 | else: |
| 168 | encoder_hidden_states = text_emb |
| 169 | if encoder_hidden_states.shape[0] != hidden_states.shape[0]: |
| 170 | encoder_hidden_states = encoder_hidden_states.repeat(hidden_states.shape[0], 1, 1) |
| 171 | |
| 172 | if tiled: |
| 173 | tile_size = min(tile_size, min(height, width)) |
| 174 | hidden_states = hidden_states.permute(0, 2, 1).reshape(batch, inner_dim, height, width) |
| 175 | def block_tile_forward(x): |
| 176 | b, c, h, w = x.shape |
| 177 | x = x.permute(0, 2, 3, 1).reshape(b, h*w, c) |
| 178 | x = block(x, encoder_hidden_states) |
| 179 | x = x.reshape(b, h, w, c).permute(0, 3, 1, 2) |
| 180 | return x |
| 181 | for block in self.transformer_blocks: |
| 182 | hidden_states = TileWorker().tiled_forward( |
| 183 | block_tile_forward, |
| 184 | hidden_states, |
| 185 | tile_size, |
| 186 | tile_stride, |
| 187 | tile_device=hidden_states.device, |
| 188 | tile_dtype=hidden_states.dtype |
| 189 | ) |
| 190 | hidden_states = hidden_states.permute(0, 2, 3, 1).reshape(batch, height * width, inner_dim) |
| 191 | else: |
| 192 | for block_id, block in enumerate(self.transformer_blocks): |
| 193 | hidden_states = block( |
| 194 | hidden_states, |
| 195 | encoder_hidden_states=encoder_hidden_states, |
| 196 | ipadapter_kwargs=ipadapter_kwargs_list.get(block_id, None) |
| 197 | ) |
| 198 | if cross_frame_attention: |
| 199 | hidden_states = hidden_states.reshape(batch, height * width, inner_dim) |
| 200 | |
| 201 | if self.need_proj_out: |
| 202 | hidden_states = self.proj_out(hidden_states) |
| 203 | hidden_states = hidden_states.reshape(batch, height, width, inner_dim).permute(0, 3, 1, 2).contiguous() |
| 204 | hidden_states = hidden_states + residual |
| 205 | else: |
nothing calls this directly
no test coverage detected