

'''simple docstring'''
from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
)







@flax.struct.dataclass
class 							A__   (	A__  ):
  A__					=    42
  A__					=    42







class 							A__   (	nn.Module  ):
  A__					=    42
  A__					=    (16, 32, 96, 2_56)
  A__					=    jnp.floataa





  def 						A  (		self				:     Any			)     ->   Optional[Any]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =nn.Conv(
           self.block_out_channels[0]							, kernel_size=(3, 3)							, padding=((1, 1), (1, 1))							, dtype=self.dtype							, )

       _SCREAMING_SNAKE_CASE							   =[]
       for i in range(len(self.block_out_channels			) - 1			):
            _SCREAMING_SNAKE_CASE							   =self.block_out_channels[i]
            _SCREAMING_SNAKE_CASE							   =self.block_out_channels[i + 1]
            _SCREAMING_SNAKE_CASE							   =nn.Conv(
                _a							, kernel_size=(3, 3)							, padding=((1, 1), (1, 1))							, dtype=self.dtype							, )
            blocks.append(_a			)
            _SCREAMING_SNAKE_CASE							   =nn.Conv(
                _a							, kernel_size=(3, 3)							, strides=(2, 2)							, padding=((1, 1), (1, 1))							, dtype=self.dtype							, )
            blocks.append(_a			)
       _SCREAMING_SNAKE_CASE							   =blocks

       _SCREAMING_SNAKE_CASE							   =nn.Conv(
           self.conditioning_embedding_channels							, kernel_size=(3, 3)							, padding=((1, 1), (1, 1))							, kernel_init=nn.initializers.zeros_init()							, bias_init=nn.initializers.zeros_init()							, dtype=self.dtype							, )





  def __call__(		self				:     Tuple							, _a				:     List[str]			)     ->   Any:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.conv_in(_a			)
       _SCREAMING_SNAKE_CASE							   =nn.silu(_a			)

       for block in self.blocks:
            _SCREAMING_SNAKE_CASE							   =block(_a			)
            _SCREAMING_SNAKE_CASE							   =nn.silu(_a			)

       _SCREAMING_SNAKE_CASE							   =self.conv_out(_a			)

       return embedding







@flax_register_to_config
class 							A__   (	nn.Module					,       A__					,       A__  ):
  A__					=    32
  A__					=    4
  A__					=    (
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "DownBlock2D",
  )
  A__					=    False
  A__					=    (3_20, 6_40, 12_80, 12_80)
  A__					=    2
  A__					=    8
  A__					=    None
  A__					=    12_80
  A__					=    0.0
  A__					=    False
  A__					=    jnp.floataa
  A__					=    True
  A__					=    0
  A__					=    "rgb"
  A__					=    (16, 32, 96, 2_56)





  def 						A  (		self				:     List[str]							, _a				:     jax.random.KeyArray			)     ->   FrozenDict:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =(1, self.in_channels, self.sample_size, self.sample_size)
       _SCREAMING_SNAKE_CASE							   =jnp.zeros(_a							, dtype=jnp.floataa			)
       _SCREAMING_SNAKE_CASE							   =jnp.ones((1,)							, dtype=jnp.intaa			)
       _SCREAMING_SNAKE_CASE							   =jnp.zeros((1, 1, self.cross_attention_dim)							, dtype=jnp.floataa			)
       _SCREAMING_SNAKE_CASE							   =(1, 3, self.sample_size * 8, self.sample_size * 8)
       _SCREAMING_SNAKE_CASE							   =jnp.zeros(_a							, dtype=jnp.floataa			)

       _SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE							   =jax.random.split(_a			)
       _SCREAMING_SNAKE_CASE							   ={'params': params_rng, 'dropout': dropout_rng}

       return self.init(_a							, _a							, _a							, _a							, _a			)["params"]





  def 						A  (		self				:     Union[str, Any]			)     ->   Any:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.block_out_channels
       _SCREAMING_SNAKE_CASE							   =block_out_channels[0] * 4

       # If `num_attention_heads` is not defined (which is the case for most models)
       # it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
       # The reason for this behavior is to correct for incorrectly named variables that were introduced
       # when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
       # Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
       # which is why we correct for the naming here.
       _SCREAMING_SNAKE_CASE							   =self.num_attention_heads or self.attention_head_dim

       # input
       _SCREAMING_SNAKE_CASE							   =nn.Conv(
           block_out_channels[0]							, kernel_size=(3, 3)							, strides=(1, 1)							, padding=((1, 1), (1, 1))							, dtype=self.dtype							, )

       # time
       _SCREAMING_SNAKE_CASE							   =FlaxTimesteps(
           block_out_channels[0]							, flip_sin_to_cos=self.flip_sin_to_cos							, freq_shift=self.config.freq_shift			)
       _SCREAMING_SNAKE_CASE							   =FlaxTimestepEmbedding(_a							, dtype=self.dtype			)

       _SCREAMING_SNAKE_CASE							   =FlaxControlNetConditioningEmbedding(
           conditioning_embedding_channels=block_out_channels[0]							, block_out_channels=self.conditioning_embedding_out_channels							, )

       _SCREAMING_SNAKE_CASE							   =self.only_cross_attention
       if isinstance(_a							, _a			):
            _SCREAMING_SNAKE_CASE							   =(only_cross_attention,) * len(self.down_block_types			)

       if isinstance(_a							, _a			):
            _SCREAMING_SNAKE_CASE							   =(num_attention_heads,) * len(self.down_block_types			)

       # down
       _SCREAMING_SNAKE_CASE							   =[]
       _SCREAMING_SNAKE_CASE							   =[]

       _SCREAMING_SNAKE_CASE							   =block_out_channels[0]

       _SCREAMING_SNAKE_CASE							   =nn.Conv(
           _a							, kernel_size=(1, 1)							, padding='VALID'							, kernel_init=nn.initializers.zeros_init()							, bias_init=nn.initializers.zeros_init()							, dtype=self.dtype							, )
       controlnet_down_blocks.append(_a			)

       for i, down_block_type in enumerate(self.down_block_types			):
            _SCREAMING_SNAKE_CASE							   =output_channel
            _SCREAMING_SNAKE_CASE							   =block_out_channels[i]
            _SCREAMING_SNAKE_CASE							   =i == len(_a			) - 1

            if down_block_type == "CrossAttnDownBlock2D":
                 _SCREAMING_SNAKE_CASE							   =FlaxCrossAttnDownBlockaD(
                     in_channels=_a							, out_channels=_a							, dropout=self.dropout							, num_layers=self.layers_per_block							, num_attention_heads=num_attention_heads[i]							, add_downsample=not is_final_block							, use_linear_projection=self.use_linear_projection							, only_cross_attention=only_cross_attention[i]							, dtype=self.dtype							, )
            else:
                 _SCREAMING_SNAKE_CASE							   =FlaxDownBlockaD(
                     in_channels=_a							, out_channels=_a							, dropout=self.dropout							, num_layers=self.layers_per_block							, add_downsample=not is_final_block							, dtype=self.dtype							, )

            down_blocks.append(_a			)

            for _ in range(self.layers_per_block			):
                 _SCREAMING_SNAKE_CASE							   =nn.Conv(
                     _a							, kernel_size=(1, 1)							, padding='VALID'							, kernel_init=nn.initializers.zeros_init()							, bias_init=nn.initializers.zeros_init()							, dtype=self.dtype							, )
                 controlnet_down_blocks.append(_a			)

            if not is_final_block:
                 _SCREAMING_SNAKE_CASE							   =nn.Conv(
                     _a							, kernel_size=(1, 1)							, padding='VALID'							, kernel_init=nn.initializers.zeros_init()							, bias_init=nn.initializers.zeros_init()							, dtype=self.dtype							, )
                 controlnet_down_blocks.append(_a			)

       _SCREAMING_SNAKE_CASE							   =down_blocks
       _SCREAMING_SNAKE_CASE							   =controlnet_down_blocks

       # mid
       _SCREAMING_SNAKE_CASE							   =block_out_channels[-1]
       _SCREAMING_SNAKE_CASE							   =FlaxUNetMidBlockaDCrossAttn(
           in_channels=_a							, dropout=self.dropout							, num_attention_heads=num_attention_heads[-1]							, use_linear_projection=self.use_linear_projection							, dtype=self.dtype							, )

       _SCREAMING_SNAKE_CASE							   =nn.Conv(
           _a							, kernel_size=(1, 1)							, padding='VALID'							, kernel_init=nn.initializers.zeros_init()							, bias_init=nn.initializers.zeros_init()							, dtype=self.dtype							, )





  def __call__(		self				:     Optional[Any]							, _a				:     int							, _a				:     Any							, _a				:     List[str]							, _a				:     List[str]							, _a				:     float = 1.0							, _a				:     bool = True							, _a				:     bool = False							, )     ->   Union[FlaxControlNetOutput, Tuple]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.controlnet_conditioning_channel_order
       if channel_order == "bgr":
            _SCREAMING_SNAKE_CASE							   =jnp.flip(_a							, axis=1			)

       # 1. time
       if not isinstance(_a							, jnp.ndarray			):
            _SCREAMING_SNAKE_CASE							   =jnp.array([timesteps]							, dtype=jnp.intaa			)
       elif isinstance(_a							, jnp.ndarray			) and len(timesteps.shape			) == 0:
            _SCREAMING_SNAKE_CASE							   =timesteps.astype(dtype=jnp.floataa			)
            _SCREAMING_SNAKE_CASE							   =jnp.expand_dims(_a							, 0			)

       _SCREAMING_SNAKE_CASE							   =self.time_proj(_a			)
       _SCREAMING_SNAKE_CASE							   =self.time_embedding(_a			)

       # 2. pre-process
       _SCREAMING_SNAKE_CASE							   =jnp.transpose(_a							, (0, 2, 3, 1)			)
       _SCREAMING_SNAKE_CASE							   =self.conv_in(_a			)

       _SCREAMING_SNAKE_CASE							   =jnp.transpose(_a							, (0, 2, 3, 1)			)
       _SCREAMING_SNAKE_CASE							   =self.controlnet_cond_embedding(_a			)
       sample += controlnet_cond

       # 3. down
       _SCREAMING_SNAKE_CASE							   =(sample,)
       for down_block in self.down_blocks:
            if isinstance(_a							, _a			):
                 _SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE							   =down_block(_a							, _a							, _a							, deterministic=not train			)
            else:
                 _SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE							   =down_block(_a							, _a							, deterministic=not train			)
            down_block_res_samples += res_samples

       # 4. mid
       _SCREAMING_SNAKE_CASE							   =self.mid_block(_a							, _a							, _a							, deterministic=not train			)

       # 5. contronet blocks
       _SCREAMING_SNAKE_CASE							   =()
       for down_block_res_sample, controlnet_block in zip(_a							, self.controlnet_down_blocks			):
            _SCREAMING_SNAKE_CASE							   =controlnet_block(_a			)
            controlnet_down_block_res_samples += (down_block_res_sample,)

       _SCREAMING_SNAKE_CASE							   =controlnet_down_block_res_samples

       _SCREAMING_SNAKE_CASE							   =self.controlnet_mid_block(_a			)

       # 6. scaling
       _SCREAMING_SNAKE_CASE							   =[sample * conditioning_scale for sample in down_block_res_samples]
       mid_block_res_sample *= conditioning_scale

       if not return_dict:
            return (down_block_res_samples, mid_block_res_sample)

       return FlaxControlNetOutput(
           down_block_res_samples=_a							, mid_block_res_sample=_a			)



'''simple docstring'''
import argparse
import io

import requests
import torch
from omegaconf import OmegaConf

from diffusers import AutoencoderKL
from diffusers.pipelines.stable_diffusion.convert_from_ckpt import (
    assign_to_checkpoint,
    conv_attn_to_linear,
    create_vae_diffusers_config,
    renew_vae_attention_paths,
    renew_vae_resnet_paths,
)


def 	_lowerCAmelCase (						_UpperCamelCase						:					Tuple							,			_UpperCamelCase						:					List[Any]    )				->      Dict:
     """simple docstring"""




     _SCREAMING_SNAKE_CASE							   =checkpoint

     _SCREAMING_SNAKE_CASE							   ={}

     _SCREAMING_SNAKE_CASE							   =vae_state_dict['encoder.conv_in.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['encoder.conv_in.bias']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['encoder.conv_out.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['encoder.conv_out.bias']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['encoder.norm_out.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['encoder.norm_out.bias']

     _SCREAMING_SNAKE_CASE							   =vae_state_dict['decoder.conv_in.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['decoder.conv_in.bias']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['decoder.conv_out.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['decoder.conv_out.bias']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['decoder.norm_out.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['decoder.norm_out.bias']

     _SCREAMING_SNAKE_CASE							   =vae_state_dict['quant_conv.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['quant_conv.bias']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['post_quant_conv.weight']
     _SCREAMING_SNAKE_CASE							   =vae_state_dict['post_quant_conv.bias']

     # Retrieves the keys for the encoder down blocks only
     _SCREAMING_SNAKE_CASE							   =len({'.'.join(layer.split('.'    )[:3]    ) for layer in vae_state_dict if 'encoder.down' in layer}    )
     _SCREAMING_SNAKE_CASE							   ={
         layer_id: [key for key in vae_state_dict if f"down.{layer_id}" in key] for layer_id in range(_UpperCamelCase    )
     }

     # Retrieves the keys for the decoder up blocks only
     _SCREAMING_SNAKE_CASE							   =len({'.'.join(layer.split('.'    )[:3]    ) for layer in vae_state_dict if 'decoder.up' in layer}    )
     _SCREAMING_SNAKE_CASE							   ={
         layer_id: [key for key in vae_state_dict if f"up.{layer_id}" in key] for layer_id in range(_UpperCamelCase    )
     }

     for i in range(_UpperCamelCase    ):
          _SCREAMING_SNAKE_CASE							   =[key for key in down_blocks[i] if f"down.{i}" in key and f"down.{i}.downsample" not in key]

          if f"encoder.down.{i}.downsample.conv.weight" in vae_state_dict:
               _SCREAMING_SNAKE_CASE							   =vae_state_dict.pop(
                   f"encoder.down.{i}.downsample.conv.weight"    )
               _SCREAMING_SNAKE_CASE							   =vae_state_dict.pop(
                   f"encoder.down.{i}.downsample.conv.bias"    )

          _SCREAMING_SNAKE_CASE							   =renew_vae_resnet_paths(_UpperCamelCase    )
          _SCREAMING_SNAKE_CASE							   ={'old': f"down.{i}.block", 'new': f"down_blocks.{i}.resnets"}
          assign_to_checkpoint(_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase							,			additional_replacements=[meta_path]							,			config=_UpperCamelCase    )

     _SCREAMING_SNAKE_CASE							   =[key for key in vae_state_dict if 'encoder.mid.block' in key]
     _SCREAMING_SNAKE_CASE							   =2
     for i in range(1							,			num_mid_res_blocks + 1    ):
          _SCREAMING_SNAKE_CASE							   =[key for key in mid_resnets if f"encoder.mid.block_{i}" in key]

          _SCREAMING_SNAKE_CASE							   =renew_vae_resnet_paths(_UpperCamelCase    )
          _SCREAMING_SNAKE_CASE							   ={'old': f"mid.block_{i}", 'new': f"mid_block.resnets.{i - 1}"}
          assign_to_checkpoint(_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase							,			additional_replacements=[meta_path]							,			config=_UpperCamelCase    )

     _SCREAMING_SNAKE_CASE							   =[key for key in vae_state_dict if 'encoder.mid.attn' in key]
     _SCREAMING_SNAKE_CASE							   =renew_vae_attention_paths(_UpperCamelCase    )
     _SCREAMING_SNAKE_CASE							   ={'old': 'mid.attn_1', 'new': 'mid_block.attentions.0'}
     assign_to_checkpoint(_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase							,			additional_replacements=[meta_path]							,			config=_UpperCamelCase    )
     conv_attn_to_linear(_UpperCamelCase    )

     for i in range(_UpperCamelCase    ):
          _SCREAMING_SNAKE_CASE							   =num_up_blocks - 1 - i
          _SCREAMING_SNAKE_CASE							   =[
              key for key in up_blocks[block_id] if f"up.{block_id}" in key and f"up.{block_id}.upsample" not in key
          ]

          if f"decoder.up.{block_id}.upsample.conv.weight" in vae_state_dict:
               _SCREAMING_SNAKE_CASE							   =vae_state_dict[
                   f"decoder.up.{block_id}.upsample.conv.weight"
               ]
               _SCREAMING_SNAKE_CASE							   =vae_state_dict[
                   f"decoder.up.{block_id}.upsample.conv.bias"
               ]

          _SCREAMING_SNAKE_CASE							   =renew_vae_resnet_paths(_UpperCamelCase    )
          _SCREAMING_SNAKE_CASE							   ={'old': f"up.{block_id}.block", 'new': f"up_blocks.{i}.resnets"}
          assign_to_checkpoint(_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase							,			additional_replacements=[meta_path]							,			config=_UpperCamelCase    )

     _SCREAMING_SNAKE_CASE							   =[key for key in vae_state_dict if 'decoder.mid.block' in key]
     _SCREAMING_SNAKE_CASE							   =2
     for i in range(1							,			num_mid_res_blocks + 1    ):
          _SCREAMING_SNAKE_CASE							   =[key for key in mid_resnets if f"decoder.mid.block_{i}" in key]

          _SCREAMING_SNAKE_CASE							   =renew_vae_resnet_paths(_UpperCamelCase    )
          _SCREAMING_SNAKE_CASE							   ={'old': f"mid.block_{i}", 'new': f"mid_block.resnets.{i - 1}"}
          assign_to_checkpoint(_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase							,			additional_replacements=[meta_path]							,			config=_UpperCamelCase    )

     _SCREAMING_SNAKE_CASE							   =[key for key in vae_state_dict if 'decoder.mid.attn' in key]
     _SCREAMING_SNAKE_CASE							   =renew_vae_attention_paths(_UpperCamelCase    )
     _SCREAMING_SNAKE_CASE							   ={'old': 'mid.attn_1', 'new': 'mid_block.attentions.0'}
     assign_to_checkpoint(_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase							,			additional_replacements=[meta_path]							,			config=_UpperCamelCase    )
     conv_attn_to_linear(_UpperCamelCase    )
     return new_checkpoint


def 	_lowerCAmelCase (						_UpperCamelCase						:					str							,			_UpperCamelCase						:					str							,			)				->      List[Any]:
     """simple docstring"""




     _SCREAMING_SNAKE_CASE							   =requests.get(
         ' https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml'    )
     _SCREAMING_SNAKE_CASE							   =io.BytesIO(r.content    )

     _SCREAMING_SNAKE_CASE							   =OmegaConf.load(_UpperCamelCase    )
     _SCREAMING_SNAKE_CASE							   =5_12
     _SCREAMING_SNAKE_CASE							   ='cuda' if torch.cuda.is_available() else 'cpu'
     if checkpoint_path.endswith('safetensors'    ):
          from safetensors import safe_open

          _SCREAMING_SNAKE_CASE							   ={}
          with safe_open(_UpperCamelCase							,			framework='pt'							,			device='cpu'    ) as f:
               for key in f.keys():
                    _SCREAMING_SNAKE_CASE							   =f.get_tensor(_UpperCamelCase    )
     else:
          _SCREAMING_SNAKE_CASE							   =torch.load(_UpperCamelCase							,			map_location=_UpperCamelCase    )['state_dict']

     # Convert the VAE model.
     _SCREAMING_SNAKE_CASE							   =create_vae_diffusers_config(_UpperCamelCase							,			image_size=_UpperCamelCase    )
     _SCREAMING_SNAKE_CASE							   =custom_convert_ldm_vae_checkpoint(_UpperCamelCase							,			_UpperCamelCase    )

     _SCREAMING_SNAKE_CASE							   =AutoencoderKL(**_UpperCamelCase    )
     vae.load_state_dict(_UpperCamelCase    )
     vae.save_pretrained(_UpperCamelCase    )


if __name__ == "__main__":
     lowerCamelCase			:				Union[str, Any] 			=					argparse.ArgumentParser()

     parser.add_argument("--vae_pt_path", default=None, type=str, required=True, help="Path to the VAE.pt to convert.")
     parser.add_argument("--dump_path", default=None, type=str, required=True, help="Path to the VAE.pt to convert.")

     lowerCamelCase			:				List[str] 			=					parser.parse_args()

     vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)
