diff --git a/__init__.py b/__init__.py index 2a8b550..127ce13 100644 --- a/__init__.py +++ b/__init__.py @@ -6,7 +6,7 @@ import einops from comfy.model_patcher import ModelPatcher from comfy.ldm.modules.attention import optimized_attention, optimized_attention_masked import comfy.ops -from typing import Union +from typing import Optional, Union T = torch.Tensor @@ -77,16 +77,14 @@ def sdpa(q: T, k: T, v: T, mask=None, heads: int = 8) -> T: class SharedAttentionProcessor: - def __init__( - self, - args: StyleAlignedArgs, - style_image: T - ): + def __init__(self, args: StyleAlignedArgs, style_image: Optional[T]): self.args = args self.ref_img = style_image def __call__(self, q, k, v, extra_options): - current_index = "{}_{}".format(extra_options["transformer_index"], extra_options["block_index"]) + current_index = "{}_{}".format( + extra_options["transformer_index"], extra_options["block_index"] + ) if self.args.adain_queries: q = adain(q) @@ -100,40 +98,55 @@ class SharedAttentionProcessor: return q, k, v + +def get_norm_layers( + layer: nn.Module, + norm_layers_: dict[str, list[Union[nn.GroupNorm, nn.LayerNorm]]], + share_layer_norm: bool, + share_group_norm: bool, +): + print( + "get layer", + layer.__class__.__name__, + isinstance(layer, nn.LayerNorm), + isinstance(layer, nn.GroupNorm), + ) + if isinstance(layer, nn.LayerNorm) and share_layer_norm: + norm_layers_["layer"].append(layer) + if isinstance(layer, nn.GroupNorm) and share_group_norm: + norm_layers_["group"].append(layer) + else: + for child_layer in layer.children(): + get_norm_layers( + child_layer, norm_layers_, share_layer_norm, share_group_norm + ) + + +def register_norm_forward( + norm_layer: Union[nn.GroupNorm, nn.LayerNorm], +) -> Union[nn.GroupNorm, nn.LayerNorm]: + if not hasattr(norm_layer, "orig_forward"): + setattr(norm_layer, "orig_forward", norm_layer.forward) + orig_forward = norm_layer.orig_forward + + def forward_(hidden_states: T) -> T: + n = hidden_states.shape[-2] + hidden_states = concat_first(hidden_states, dim=-2) + hidden_states = orig_forward(hidden_states) + return hidden_states[..., :n, :] + + norm_layer.forward = forward_ # type: ignore + return norm_layer + + def register_shared_norm( model: ModelPatcher, share_group_norm: bool = True, share_layer_norm: bool = True, ): - def register_norm_forward( - norm_layer: Union[nn.GroupNorm, nn.LayerNorm], - ) -> Union[nn.GroupNorm, nn.LayerNorm]: - if not hasattr(norm_layer, "orig_forward"): - setattr(norm_layer, "orig_forward", norm_layer.forward) - orig_forward = norm_layer.orig_forward - - def forward_(hidden_states: T) -> T: - n = hidden_states.shape[-2] - hidden_states = concat_first(hidden_states, dim=-2) - hidden_states = orig_forward(hidden_states) - return hidden_states[..., :n, :] - - norm_layer.forward = forward_ # type: ignore - return norm_layer - - def get_norm_layers( - layer, norm_layers_: dict[str, list[Union[nn.GroupNorm, nn.LayerNorm]]] - ): - if isinstance(layer, nn.LayerNorm) and share_layer_norm: - norm_layers_["layer"].append(layer) - if isinstance(layer, nn.GroupNorm) and share_group_norm: - norm_layers_["group"].append(layer) - else: - for layer in layer.children(): - get_norm_layers(layer, norm_layers_) - norm_layers = {"group": [], "layer": []} - get_norm_layers(model, norm_layers) + get_norm_layers(model.model, norm_layers, share_layer_norm, share_group_norm) + print(f"Patching {len(norm_layers['group'])} group norms, {len(norm_layers['layer'])} layer norms.") return [register_norm_forward(layer) for layer in norm_layers["group"]] + [ register_norm_forward(layer) for layer in norm_layers["layer"] ] @@ -157,15 +170,16 @@ def _get_switch_vec(total_num_layers, level): return vec - class StyleAlignedPatch: @classmethod def INPUT_TYPES(cls): return { "required": { "model": ("MODEL",), + }, + "optional": { "style_image": ("IMAGE",), - } + }, } RETURN_TYPES = ("MODEL",) @@ -174,13 +188,12 @@ class StyleAlignedPatch: def __init__(self) -> None: self.args = StyleAlignedArgs() - # TODO patch norm layers - # self.norm_layers = register_shared_norm( - # model, self.args.share_group_norm, self.args.share_layer_norm - # ) - def patch(self, model: ModelPatcher, style_image: T): + def patch(self, model: ModelPatcher, style_image: Optional[T] = None): m = model.clone() + register_shared_norm( + model, self.args.share_group_norm, self.args.share_layer_norm + ) m.set_model_attn1_patch(SharedAttentionProcessor(self.args, style_image)) return (m,)