
'''simple docstring'''







import argparse
import json

import torch

from diffusers import DDPMScheduler, LDMPipeline, UNetaDModel, VQModel
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: Optional[int]      ,      __SCREAMING_SNAKE_CASE					: int=1							):



						"""simple docstring"""





						if n_shave_prefix_segments >= 0:
												return ".".join(path.split('''.'''							)[n_shave_prefix_segments:]							)
						else:
												return ".".join(path.split('''.'''							)[:n_shave_prefix_segments]							)
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: List[Any]      ,      __SCREAMING_SNAKE_CASE					: int=0							):



						"""simple docstring"""





						lowercase_	:     Union[str, Any]													=       []
						for old_item in old_list:
												lowercase_	:     Optional[Any]													=       old_item.replace('''in_layers.0'''      ,      '''norm1'''							)
												lowercase_	:     Dict													=       new_item.replace('''in_layers.2'''      ,      '''conv1'''							)

												lowercase_	:     int													=       new_item.replace('''out_layers.0'''      ,      '''norm2'''							)
												lowercase_	:     Union[str, Any]													=       new_item.replace('''out_layers.3'''      ,      '''conv2'''							)

												lowercase_	:     List[Any]													=       new_item.replace('''emb_layers.1'''      ,      '''time_emb_proj'''							)
												lowercase_	:     int													=       new_item.replace('''skip_connection'''      ,      '''conv_shortcut'''							)

												lowercase_	:     Union[str, Any]													=       shave_segments(__SCREAMING_SNAKE_CASE      ,      n_shave_prefix_segments=__SCREAMING_SNAKE_CASE							)

												mapping.append({'''old''': old_item, '''new''': new_item}							)

						return mapping
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: int      ,      __SCREAMING_SNAKE_CASE					: Optional[Any]=0							):



						"""simple docstring"""





						lowercase_	:     List[Any]													=       []
						for old_item in old_list:
												lowercase_	:     Union[str, Any]													=       old_item

												lowercase_	:     Optional[Any]													=       new_item.replace('''norm.weight'''      ,      '''group_norm.weight'''							)
												lowercase_	:     List[str]													=       new_item.replace('''norm.bias'''      ,      '''group_norm.bias'''							)

												lowercase_	:     Tuple													=       new_item.replace('''proj_out.weight'''      ,      '''proj_attn.weight'''							)
												lowercase_	:     List[Any]													=       new_item.replace('''proj_out.bias'''      ,      '''proj_attn.bias'''							)

												lowercase_	:     Tuple													=       shave_segments(__SCREAMING_SNAKE_CASE      ,      n_shave_prefix_segments=__SCREAMING_SNAKE_CASE							)

												mapping.append({'''old''': old_item, '''new''': new_item}							)

						return mapping
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: Union[str, Any]      ,      __SCREAMING_SNAKE_CASE					: Optional[Any]      ,      __SCREAMING_SNAKE_CASE					: str      ,      __SCREAMING_SNAKE_CASE					: str=None      ,      __SCREAMING_SNAKE_CASE					: Dict=None      ,      __SCREAMING_SNAKE_CASE					: Dict=None							):



						"""simple docstring"""





						assert isinstance(__SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE							), "Paths should be a list of dicts containing 'old' and 'new' keys."

						# Splits the attention layers into three variables.
						if attention_paths_to_split is not None:
												for path, path_map in attention_paths_to_split.items():
																		lowercase_	:     Tuple													=       old_checkpoint[path]
																		lowercase_	:     str													=       old_tensor.shape[0] // 3

																		lowercase_	:     List[str]													=       (-1, channels) if len(old_tensor.shape							) == 3 else (-1)

																		lowercase_	:     List[Any]													=       old_tensor.shape[0] // config['''num_head_channels'''] // 3

																		lowercase_	:     int													=       old_tensor.reshape((num_heads, 3 * channels // num_heads) + old_tensor.shape[1:]							)
																		lowercase_						, lowercase_						, lowercase_	:     Optional[int]													=       old_tensor.split(channels // num_heads      ,      dim=1							)

																		lowercase_	:     Tuple													=       query.reshape(__SCREAMING_SNAKE_CASE							)
																		lowercase_	:     Any													=       key.reshape(__SCREAMING_SNAKE_CASE							)
																		lowercase_	:     int													=       value.reshape(__SCREAMING_SNAKE_CASE							)

						for path in paths:
												lowercase_	:     List[str]													=       path['''new''']

												# These have already been assigned
												if attention_paths_to_split is not None and new_path in attention_paths_to_split:
																		continue

												# Global renaming happens here
												lowercase_	:     str													=       new_path.replace('''middle_block.0'''      ,      '''mid_block.resnets.0'''							)
												lowercase_	:     List[Any]													=       new_path.replace('''middle_block.1'''      ,      '''mid_block.attentions.0'''							)
												lowercase_	:     Dict													=       new_path.replace('''middle_block.2'''      ,      '''mid_block.resnets.1'''							)

												if additional_replacements is not None:
																		for replacement in additional_replacements:
																								lowercase_	:     Dict													=       new_path.replace(replacement['''old''']      ,      replacement['''new''']							)

        # proj_attn.weight has to be converted from conv 1D to linear
												if "proj_attn.weight" in new_path:
																		lowercase_	:     Optional[Any]													=       old_checkpoint[path['''old''']][:, :, 0]
												else:
																		lowercase_	:     Optional[Any]													=       old_checkpoint[path['''old''']]
def 					snake_case_		(				__SCREAMING_SNAKE_CASE					: str      ,      __SCREAMING_SNAKE_CASE					: List[Any]							):



						"""simple docstring"""





						lowercase_	:     List[Any]													=       {}

						lowercase_	:     Optional[int]													=       checkpoint['''time_embed.0.weight''']
						lowercase_	:     Optional[Any]													=       checkpoint['''time_embed.0.bias''']
						lowercase_	:     Optional[Any]													=       checkpoint['''time_embed.2.weight''']
						lowercase_	:     int													=       checkpoint['''time_embed.2.bias''']

						lowercase_	:     Optional[Any]													=       checkpoint['''input_blocks.0.0.weight''']
						lowercase_	:     List[str]													=       checkpoint['''input_blocks.0.0.bias''']

						lowercase_	:     int													=       checkpoint['''out.0.weight''']
						lowercase_	:     Tuple													=       checkpoint['''out.0.bias''']
						lowercase_	:     Dict													=       checkpoint['''out.2.weight''']
						lowercase_	:     Any													=       checkpoint['''out.2.bias''']

						# Retrieves the keys for the input blocks only
						lowercase_	:     int													=       len({'''.'''.join(layer.split('''.'''							)[:2]							) for layer in checkpoint if '''input_blocks''' in layer}							)
						lowercase_	:     Union[str, Any]													=       {
						    layer_id: [key for key in checkpoint if F'''input_blocks.{layer_id}''' in key]
						    for layer_id in range(__SCREAMING_SNAKE_CASE							)
						}

						# Retrieves the keys for the middle blocks only
						lowercase_	:     int													=       len({'''.'''.join(layer.split('''.'''							)[:2]							) for layer in checkpoint if '''middle_block''' in layer}							)
						lowercase_	:     str													=       {
						    layer_id: [key for key in checkpoint if F'''middle_block.{layer_id}''' in key]
						    for layer_id in range(__SCREAMING_SNAKE_CASE							)
						}

						# Retrieves the keys for the output blocks only
						lowercase_	:     List[Any]													=       len({'''.'''.join(layer.split('''.'''							)[:2]							) for layer in checkpoint if '''output_blocks''' in layer}							)
						lowercase_	:     str													=       {
						    layer_id: [key for key in checkpoint if F'''output_blocks.{layer_id}''' in key]
						    for layer_id in range(__SCREAMING_SNAKE_CASE							)
						}

						for i in range(1      ,      __SCREAMING_SNAKE_CASE							):
												lowercase_	:     Any													=       (i - 1) // (config['''num_res_blocks'''] + 1)
												lowercase_	:     Any													=       (i - 1) % (config['''num_res_blocks'''] + 1)

												lowercase_	:     List[str]													=       [key for key in input_blocks[i] if F'''input_blocks.{i}.0''' in key]
												lowercase_	:     List[Any]													=       [key for key in input_blocks[i] if F'''input_blocks.{i}.1''' in key]

												if F'''input_blocks.{i}.0.op.weight''' in checkpoint:
																		lowercase_	:     List[Any]													=       checkpoint[
																		    F'''input_blocks.{i}.0.op.weight'''
																		]
																		lowercase_	:     List[str]													=       checkpoint[
																		    F'''input_blocks.{i}.0.op.bias'''
																		]
																		continue

												lowercase_	:     Union[str, Any]													=       renew_resnet_paths(__SCREAMING_SNAKE_CASE							)
												lowercase_	:     Optional[int]													=       {'''old''': F'''input_blocks.{i}.0''', '''new''': F'''down_blocks.{block_id}.resnets.{layer_in_block_id}'''}
												lowercase_	:     List[Any]													=       {'''old''': '''resnets.2.op''', '''new''': '''downsamplers.0.op'''}
												assign_to_checkpoint(
												    __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      additional_replacements=[meta_path, resnet_op]      ,      config=__SCREAMING_SNAKE_CASE							)

												if len(__SCREAMING_SNAKE_CASE							):
																		lowercase_	:     Optional[Any]													=       renew_attention_paths(__SCREAMING_SNAKE_CASE							)
																		lowercase_	:     List[str]													=       {
																		    '''old''': F'''input_blocks.{i}.1''',
																		    '''new''': F'''down_blocks.{block_id}.attentions.{layer_in_block_id}''',
																		}
																		lowercase_	:     Dict													=       {
																		    F'''input_blocks.{i}.1.qkv.bias''': {
																		        '''key''': F'''down_blocks.{block_id}.attentions.{layer_in_block_id}.key.bias''',
																		        '''query''': F'''down_blocks.{block_id}.attentions.{layer_in_block_id}.query.bias''',
																		        '''value''': F'''down_blocks.{block_id}.attentions.{layer_in_block_id}.value.bias''',
																		    },
																		    F'''input_blocks.{i}.1.qkv.weight''': {
																		        '''key''': F'''down_blocks.{block_id}.attentions.{layer_in_block_id}.key.weight''',
																		        '''query''': F'''down_blocks.{block_id}.attentions.{layer_in_block_id}.query.weight''',
																		        '''value''': F'''down_blocks.{block_id}.attentions.{layer_in_block_id}.value.weight''',
																		    },
																		}
																		assign_to_checkpoint(
																		    __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      additional_replacements=[meta_path]      ,      attention_paths_to_split=__SCREAMING_SNAKE_CASE      ,      config=__SCREAMING_SNAKE_CASE      ,      )

						lowercase_	:     int													=       middle_blocks[0]
						lowercase_	:     Union[str, Any]													=       middle_blocks[1]
						lowercase_	:     Optional[Any]													=       middle_blocks[2]

						lowercase_	:     Dict													=       renew_resnet_paths(__SCREAMING_SNAKE_CASE							)
						assign_to_checkpoint(__SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      config=__SCREAMING_SNAKE_CASE							)

						lowercase_	:     List[str]													=       renew_resnet_paths(__SCREAMING_SNAKE_CASE							)
						assign_to_checkpoint(__SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      config=__SCREAMING_SNAKE_CASE							)

						lowercase_	:     Dict													=       renew_attention_paths(__SCREAMING_SNAKE_CASE							)
						lowercase_	:     Optional[Any]													=       {
						    '''middle_block.1.qkv.bias''': {
						        '''key''': '''mid_block.attentions.0.key.bias''',
						        '''query''': '''mid_block.attentions.0.query.bias''',
						        '''value''': '''mid_block.attentions.0.value.bias''',
						    },
						    '''middle_block.1.qkv.weight''': {
						        '''key''': '''mid_block.attentions.0.key.weight''',
						        '''query''': '''mid_block.attentions.0.query.weight''',
						        '''value''': '''mid_block.attentions.0.value.weight''',
						    },
						}
						assign_to_checkpoint(
						    __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      attention_paths_to_split=__SCREAMING_SNAKE_CASE      ,      config=__SCREAMING_SNAKE_CASE							)

						for i in range(__SCREAMING_SNAKE_CASE							):
												lowercase_	:     List[Any]													=       i // (config['''num_res_blocks'''] + 1)
												lowercase_	:     int													=       i % (config['''num_res_blocks'''] + 1)
												lowercase_	:     Optional[Any]													=       [shave_segments(__SCREAMING_SNAKE_CASE      ,      2							) for name in output_blocks[i]]
												lowercase_	:     Any													=       {}

												for layer in output_block_layers:
																		lowercase_						, lowercase_	:     Any													=       layer.split('''.'''							)[0], shave_segments(__SCREAMING_SNAKE_CASE      ,      1							)
																		if layer_id in output_block_list:
																								output_block_list[layer_id].append(__SCREAMING_SNAKE_CASE							)
																		else:
																								lowercase_	:     Optional[Any]													=       [layer_name]

												if len(__SCREAMING_SNAKE_CASE							) > 1:
																		lowercase_	:     List[Any]													=       [key for key in output_blocks[i] if F'''output_blocks.{i}.0''' in key]
																		lowercase_	:     Union[str, Any]													=       [key for key in output_blocks[i] if F'''output_blocks.{i}.1''' in key]

																		lowercase_	:     Optional[Any]													=       renew_resnet_paths(__SCREAMING_SNAKE_CASE							)
																		lowercase_	:     List[Any]													=       renew_resnet_paths(__SCREAMING_SNAKE_CASE							)

																		lowercase_	:     List[Any]													=       {'''old''': F'''output_blocks.{i}.0''', '''new''': F'''up_blocks.{block_id}.resnets.{layer_in_block_id}'''}
																		assign_to_checkpoint(__SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      additional_replacements=[meta_path]      ,      config=__SCREAMING_SNAKE_CASE							)

																		if ["conv.weight", "conv.bias"] in output_block_list.values():
																								lowercase_	:     Any													=       list(output_block_list.values()							).index(['''conv.weight''', '''conv.bias''']							)
																								lowercase_	:     str													=       checkpoint[
																								    F'''output_blocks.{i}.{index}.conv.weight'''
																								]
																								lowercase_	:     Union[str, Any]													=       checkpoint[
																								    F'''output_blocks.{i}.{index}.conv.bias'''
																								]

																								# Clear attentions as they have been attributed above.
																								if len(__SCREAMING_SNAKE_CASE							) == 2:
																														lowercase_	:     Tuple													=       []

																		if len(__SCREAMING_SNAKE_CASE							):
																								lowercase_	:     Union[str, Any]													=       renew_attention_paths(__SCREAMING_SNAKE_CASE							)
																								lowercase_	:     Any													=       {
																								    '''old''': F'''output_blocks.{i}.1''',
																								    '''new''': F'''up_blocks.{block_id}.attentions.{layer_in_block_id}''',
																								}
																								lowercase_	:     List[str]													=       {
																								    F'''output_blocks.{i}.1.qkv.bias''': {
																								        '''key''': F'''up_blocks.{block_id}.attentions.{layer_in_block_id}.key.bias''',
																								        '''query''': F'''up_blocks.{block_id}.attentions.{layer_in_block_id}.query.bias''',
																								        '''value''': F'''up_blocks.{block_id}.attentions.{layer_in_block_id}.value.bias''',
																								    },
																								    F'''output_blocks.{i}.1.qkv.weight''': {
																								        '''key''': F'''up_blocks.{block_id}.attentions.{layer_in_block_id}.key.weight''',
																								        '''query''': F'''up_blocks.{block_id}.attentions.{layer_in_block_id}.query.weight''',
																								        '''value''': F'''up_blocks.{block_id}.attentions.{layer_in_block_id}.value.weight''',
																								    },
																								}
																								assign_to_checkpoint(
																								    __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      additional_replacements=[meta_path]      ,      attention_paths_to_split=to_split if any('''qkv''' in key for key in attentions							) else None      ,      config=__SCREAMING_SNAKE_CASE      ,      )
												else:
																		lowercase_	:     int													=       renew_resnet_paths(__SCREAMING_SNAKE_CASE      ,      n_shave_prefix_segments=1							)
																		for path in resnet_0_paths:
																								lowercase_	:     Optional[int]													=       '''.'''.join(['''output_blocks''', str(__SCREAMING_SNAKE_CASE							), path['''old''']]							)
																								lowercase_	:     Union[str, Any]													=       '''.'''.join(['''up_blocks''', str(__SCREAMING_SNAKE_CASE							), '''resnets''', str(__SCREAMING_SNAKE_CASE							), path['''new''']]							)

																								lowercase_	:     Optional[Any]													=       checkpoint[old_path]

						return new_checkpoint


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

							parser.add_argument(
							    "--checkpoint_path", default=None, type=str, required=True, help="Path to the checkpoint to convert."
							)

							parser.add_argument(
							    "--config_file",
							    default=None,
							    type=str,
							    required=True,
							    help="The config json file corresponding to the architecture.",
							)

							parser.add_argument("--dump_path", default=None, type=str, required=True, help="Path to the output model.")

							_lowercase :  int    							=       parser.parse_args()

							_lowercase :  Dict    							=       torch.load(args.checkpoint_path)

							with open(args.config_file) as f:
														_lowercase :  str    							=       json.loads(f.read())

							_lowercase :  Union[str, Any]    							=       convert_ldm_checkpoint(checkpoint, config)

							if "ldm" in config:
														del config["ldm"]

							_lowercase :  Any    							=       UNetaDModel(**config)
							model.load_state_dict(converted_checkpoint)

							try:
														_lowercase :  str    							=       DDPMScheduler.from_config("/".join(args.checkpoint_path.split("/")[:-1]))
														_lowercase :  List[str]    							=       VQModel.from_pretrained("/".join(args.checkpoint_path.split("/")[:-1]))

														_lowercase :  Optional[Any]    							=       LDMPipeline(unet=model, scheduler=scheduler, vae=vqvae)
														pipe.save_pretrained(args.dump_path)
							except:  # noqa: E722
														model.save_pretrained(args.dump_path)





from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block




@dataclass
class        UpperCAmelCase_			( _A     ):


  '''simple docstring'''







  a__           =		42







class        UpperCAmelCase_			( nn.Module     ):


  '''simple docstring'''







  def __init__( self			: Tuple			,							UpperCamelCase__			: List[Any]=3			,							UpperCamelCase__			: str=3			,							UpperCamelCase__			: List[Any]=("DownEncoderBlock2D",)			,							UpperCamelCase__			: Optional[Any]=(64,)			,							UpperCamelCase__			: Optional[Any]=2			,							UpperCamelCase__			: Union[str, Any]=32			,							UpperCamelCase__			: Optional[Any]="silu"			,							UpperCamelCase__			: List[str]=True			,							)			->	str:







    """simple docstring"""






    super().__init__()
    __magic_name__				     =       layers_per_block

    __magic_name__				     =       torch.nn.Convad(
        UpperCamelCase__			,							block_out_channels[0]			,							kernel_size=3			,							stride=1			,							padding=1			,							)

    __magic_name__				     =       None
    __magic_name__				     =       nn.ModuleList([]    )

    # down
    __magic_name__				     =       block_out_channels[0]
    for i, down_block_type in enumerate(UpperCamelCase__    ):
      __magic_name__				     =       output_channel
      __magic_name__				     =       block_out_channels[i]
      __magic_name__				     =       i == len(UpperCamelCase__    ) - 1

      __magic_name__				     =       get_down_block(
          UpperCamelCase__			,							num_layers=self.layers_per_block			,							in_channels=UpperCamelCase__			,							out_channels=UpperCamelCase__			,							add_downsample=not is_final_block			,							resnet_eps=1E-6			,							downsample_padding=0			,							resnet_act_fn=UpperCamelCase__			,							resnet_groups=UpperCamelCase__			,							attention_head_dim=UpperCamelCase__			,							temb_channels=UpperCamelCase__			,							)
      self.down_blocks.append(UpperCamelCase__    )

    # mid
    __magic_name__				     =       UNetMidBlockaD(
        in_channels=block_out_channels[-1]			,							resnet_eps=1E-6			,							resnet_act_fn=UpperCamelCase__			,							output_scale_factor=1			,							resnet_time_scale_shift="""default"""			,							attention_head_dim=block_out_channels[-1]			,							resnet_groups=UpperCamelCase__			,							temb_channels=UpperCamelCase__			,							)

    # out
    __magic_name__				     =       nn.GroupNorm(num_channels=block_out_channels[-1]			,							num_groups=UpperCamelCase__			,							eps=1E-6    )
    __magic_name__				     =       nn.SiLU()

    __magic_name__				     =       2 * out_channels if double_z else out_channels
    __magic_name__				     =       nn.Convad(block_out_channels[-1]			,							UpperCamelCase__			,							3			,							padding=1    )

    __magic_name__				     =       False





  def 							_lowercase							( self			: List[str]			,							UpperCamelCase__			: Optional[Any]    )			->	int:







    """simple docstring"""






    __magic_name__				     =       x
    __magic_name__				     =       self.conv_in(UpperCamelCase__    )

    if self.training and self.gradient_checkpointing:

      def create_custom_forward(UpperCamelCase__			: int    ):
        def custom_forward(*UpperCamelCase__			: str    ):
          return module(*UpperCamelCase__    )

        return custom_forward

      # down
      if is_torch_version(""">="""			,							"""1.11.0"""    ):
        for down_block in self.down_blocks:
          __magic_name__				     =       torch.utils.checkpoint.checkpoint(
              create_custom_forward(UpperCamelCase__    )			,							UpperCamelCase__			,							use_reentrant=UpperCamelCase__    )
        # middle
        __magic_name__				     =       torch.utils.checkpoint.checkpoint(
            create_custom_forward(self.mid_block    )			,							UpperCamelCase__			,							use_reentrant=UpperCamelCase__    )
      else:
        for down_block in self.down_blocks:
          __magic_name__				     =       torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCamelCase__    )			,							UpperCamelCase__    )
        # middle
        __magic_name__				     =       torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block    )			,							UpperCamelCase__    )

    else:
      # down
      for down_block in self.down_blocks:
        __magic_name__				     =       down_block(UpperCamelCase__    )

      # middle
      __magic_name__				     =       self.mid_block(UpperCamelCase__    )

    # post-process
    __magic_name__				     =       self.conv_norm_out(UpperCamelCase__    )
    __magic_name__				     =       self.conv_act(UpperCamelCase__    )
    __magic_name__				     =       self.conv_out(UpperCamelCase__    )

    return sample







class        UpperCAmelCase_			( nn.Module     ):


  '''simple docstring'''







  def __init__( self			: List[str]			,							UpperCamelCase__			: int=3			,							UpperCamelCase__			: Dict=3			,							UpperCamelCase__			: List[Any]=("UpDecoderBlock2D",)			,							UpperCamelCase__			: List[Any]=(64,)			,							UpperCamelCase__			: Dict=2			,							UpperCamelCase__			: int=32			,							UpperCamelCase__			: Optional[int]="silu"			,							UpperCamelCase__			: Tuple="group"			,							)			->	Dict:







    """simple docstring"""






    super().__init__()
    __magic_name__				     =       layers_per_block

    __magic_name__				     =       nn.Convad(
        UpperCamelCase__			,							block_out_channels[-1]			,							kernel_size=3			,							stride=1			,							padding=1			,							)

    __magic_name__				     =       None
    __magic_name__				     =       nn.ModuleList([]    )

    __magic_name__				     =       in_channels if norm_type == """spatial""" else None

    # mid
    __magic_name__				     =       UNetMidBlockaD(
        in_channels=block_out_channels[-1]			,							resnet_eps=1E-6			,							resnet_act_fn=UpperCamelCase__			,							output_scale_factor=1			,							resnet_time_scale_shift="""default""" if norm_type == """group""" else norm_type			,							attention_head_dim=block_out_channels[-1]			,							resnet_groups=UpperCamelCase__			,							temb_channels=UpperCamelCase__			,							)

    # up
    __magic_name__				     =       list(reversed(UpperCamelCase__    )    )
    __magic_name__				     =       reversed_block_out_channels[0]
    for i, up_block_type in enumerate(UpperCamelCase__    ):
      __magic_name__				     =       output_channel
      __magic_name__				     =       reversed_block_out_channels[i]

      __magic_name__				     =       i == len(UpperCamelCase__    ) - 1

      __magic_name__				     =       get_up_block(
          UpperCamelCase__			,							num_layers=self.layers_per_block + 1			,							in_channels=UpperCamelCase__			,							out_channels=UpperCamelCase__			,							prev_output_channel=UpperCamelCase__			,							add_upsample=not is_final_block			,							resnet_eps=1E-6			,							resnet_act_fn=UpperCamelCase__			,							resnet_groups=UpperCamelCase__			,							attention_head_dim=UpperCamelCase__			,							temb_channels=UpperCamelCase__			,							resnet_time_scale_shift=UpperCamelCase__			,							)
      self.up_blocks.append(UpperCamelCase__    )
      __magic_name__				     =       output_channel

    # out
    if norm_type == "spatial":
      __magic_name__				     =       SpatialNorm(block_out_channels[0]			,							UpperCamelCase__    )
    else:
      __magic_name__				     =       nn.GroupNorm(num_channels=block_out_channels[0]			,							num_groups=UpperCamelCase__			,							eps=1E-6    )
    __magic_name__				     =       nn.SiLU()
    __magic_name__				     =       nn.Convad(block_out_channels[0]			,							UpperCamelCase__			,							3			,							padding=1    )

    __magic_name__				     =       False





  def 							_lowercase							( self			: str			,							UpperCamelCase__			: Union[str, Any]			,							UpperCamelCase__			: Tuple=None    )			->	Tuple:







    """simple docstring"""






    __magic_name__				     =       z
    __magic_name__				     =       self.conv_in(UpperCamelCase__    )

    __magic_name__				     =       next(iter(self.up_blocks.parameters()    )    ).dtype
    if self.training and self.gradient_checkpointing:

      def create_custom_forward(UpperCamelCase__			: Optional[int]    ):
        def custom_forward(*UpperCamelCase__			: int    ):
          return module(*UpperCamelCase__    )

        return custom_forward

      if is_torch_version(""">="""			,							"""1.11.0"""    ):
        # middle
        __magic_name__				     =       torch.utils.checkpoint.checkpoint(
            create_custom_forward(self.mid_block    )			,							UpperCamelCase__			,							UpperCamelCase__			,							use_reentrant=UpperCamelCase__    )
        __magic_name__				     =       sample.to(UpperCamelCase__    )

        # up
        for up_block in self.up_blocks:
          __magic_name__				     =       torch.utils.checkpoint.checkpoint(
              create_custom_forward(UpperCamelCase__    )			,							UpperCamelCase__			,							UpperCamelCase__			,							use_reentrant=UpperCamelCase__    )
      else:
        # middle
        __magic_name__				     =       torch.utils.checkpoint.checkpoint(
            create_custom_forward(self.mid_block    )			,							UpperCamelCase__			,							UpperCamelCase__    )
        __magic_name__				     =       sample.to(UpperCamelCase__    )

        # up
        for up_block in self.up_blocks:
          __magic_name__				     =       torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCamelCase__    )			,							UpperCamelCase__			,							UpperCamelCase__    )
    else:
      # middle
      __magic_name__				     =       self.mid_block(UpperCamelCase__			,							UpperCamelCase__    )
      __magic_name__				     =       sample.to(UpperCamelCase__    )

      # up
      for up_block in self.up_blocks:
        __magic_name__				     =       up_block(UpperCamelCase__			,							UpperCamelCase__    )

        # post-process
    if latent_embeds is None:
      __magic_name__				     =       self.conv_norm_out(UpperCamelCase__    )
    else:
      __magic_name__				     =       self.conv_norm_out(UpperCamelCase__			,							UpperCamelCase__    )
    __magic_name__				     =       self.conv_act(UpperCamelCase__    )
    __magic_name__				     =       self.conv_out(UpperCamelCase__    )

    return sample







class        UpperCAmelCase_			( nn.Module     ):


  '''simple docstring'''







  def __init__( self			: str			,							UpperCamelCase__			: Optional[Any]			,							UpperCamelCase__			: Optional[Any]			,							UpperCamelCase__			: List[str]			,							UpperCamelCase__			: Union[str, Any]=None			,							UpperCamelCase__			: Dict="random"			,							UpperCamelCase__			: List[Any]=False			,							UpperCamelCase__			: Dict=True    )			->	Optional[Any]:







    """simple docstring"""






    super().__init__()
    __magic_name__				     =       n_e
    __magic_name__				     =       vq_embed_dim
    __magic_name__				     =       beta
    __magic_name__				     =       legacy

    __magic_name__				     =       nn.Embedding(self.n_e			,							self.vq_embed_dim    )
    self.embedding.weight.data.uniform_(-1.0 / self.n_e			,							1.0 / self.n_e    )

    __magic_name__				     =       remap
    if self.remap is not None:
      self.register_buffer("""used"""			,							torch.tensor(np.load(self.remap    )    )    )
      __magic_name__				     =       self.used.shape[0]
      __magic_name__				     =       unknown_index  # "random" or "extra" or integer
      if self.unknown_index == "extra":
        __magic_name__				     =       self.re_embed
        __magic_name__				     =       self.re_embed + 1
      print(
          F'''Remapping {self.n_e} indices to {self.re_embed} indices. '''
          F'''Using {self.unknown_index} for unknown indices.'''    )
    else:
      __magic_name__				     =       n_e

    __magic_name__				     =       sane_index_shape
  def 							_lowercase							( self			: Union[str, Any]			,							UpperCamelCase__			: Dict    )			->	Union[str, Any]:







    """simple docstring"""






    __magic_name__				     =       inds.shape
    assert len(UpperCamelCase__    ) > 1
    __magic_name__				     =       inds.reshape(ishape[0]			,							-1    )
    __magic_name__				     =       self.used.to(UpperCamelCase__    )
    __magic_name__				     =       (inds[:, :, None] == used[None, None, ...]).long()
    __magic_name__				     =       match.argmax(-1    )
    __magic_name__				     =       match.sum(2    ) < 1
    if self.unknown_index == "random":
      __magic_name__				     =       torch.randint(0			,							self.re_embed			,							size=new[unknown].shape    ).to(device=new.device    )
    else:
      __magic_name__				     =       self.unknown_index
    return new.reshape(UpperCamelCase__    )
  def 							_lowercase							( self			: Optional[Any]			,							UpperCamelCase__			: str    )			->	Tuple:







    """simple docstring"""






    __magic_name__				     =       inds.shape
    assert len(UpperCamelCase__    ) > 1
    __magic_name__				     =       inds.reshape(ishape[0]			,							-1    )
    __magic_name__				     =       self.used.to(UpperCamelCase__    )
    if self.re_embed > self.used.shape[0]:  # extra token
      __magic_name__				     =       0  # simply set to zero
    __magic_name__				     =       torch.gather(used[None, :][inds.shape[0] * [0], :]			,							1			,							UpperCamelCase__    )
    return back.reshape(UpperCamelCase__    )
  def 							_lowercase							( self			: List[str]			,							UpperCamelCase__			: List[str]    )			->	List[str]:







    """simple docstring"""






    __magic_name__				     =       z.permute(0			,							2			,							3			,							1    ).contiguous()
    __magic_name__				     =       z.view(-1			,							self.vq_embed_dim    )

    # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
    __magic_name__				     =       torch.argmin(torch.cdist(UpperCamelCase__			,							self.embedding.weight    )			,							dim=1    )

    __magic_name__				     =       self.embedding(UpperCamelCase__    ).view(z.shape    )
    __magic_name__				     =       None
    __magic_name__				     =       None

    # compute loss for embedding
    if not self.legacy:
      __magic_name__				     =       self.beta * torch.mean((z_q.detach() - z) ** 2    ) + torch.mean((z_q - z.detach()) ** 2    )
    else:
      __magic_name__				     =       torch.mean((z_q.detach() - z) ** 2    ) + self.beta * torch.mean((z_q - z.detach()) ** 2    )

    # preserve gradients
    __magic_name__				     =       z + (z_q - z).detach()

    # reshape back to match original input shape
    __magic_name__				     =       z_q.permute(0			,							3			,							1			,							2    ).contiguous()

    if self.remap is not None:
      __magic_name__				     =       min_encoding_indices.reshape(z.shape[0]			,							-1    )  # add batch axis
      __magic_name__				     =       self.remap_to_used(UpperCamelCase__    )
      __magic_name__				     =       min_encoding_indices.reshape(-1			,							1    )  # flatten

    if self.sane_index_shape:
      __magic_name__				     =       min_encoding_indices.reshape(z_q.shape[0]			,							z_q.shape[2]			,							z_q.shape[3]    )

    return z_q, loss, (perplexity, min_encodings, min_encoding_indices)





  def 							_lowercase							( self			: Union[str, Any]			,							UpperCamelCase__			: List[str]			,							UpperCamelCase__			: Union[str, Any]    )			->	int:







    """simple docstring"""






    if self.remap is not None:
      __magic_name__				     =       indices.reshape(shape[0]			,							-1    )  # add batch axis
      __magic_name__				     =       self.unmap_to_all(UpperCamelCase__    )
      __magic_name__				     =       indices.reshape(-1    )  # flatten again

    # get quantized latent vectors
    __magic_name__				     =       self.embedding(UpperCamelCase__    )

    if shape is not None:
      __magic_name__				     =       z_q.view(UpperCamelCase__    )
      # reshape back to match original input shape
      __magic_name__				     =       z_q.permute(0			,							3			,							1			,							2    ).contiguous()

    return z_q







class        UpperCAmelCase_			( _A     ):


  '''simple docstring'''







  def __init__( self			: Dict			,							UpperCamelCase__			: Optional[Any]			,							UpperCamelCase__			: Tuple=False    )			->	Optional[int]:







    """simple docstring"""






    __magic_name__				     =       parameters
    __magic_name__     ,    __magic_name__				     =       torch.chunk(UpperCamelCase__			,							2			,							dim=1    )
    __magic_name__				     =       torch.clamp(self.logvar			,							-30.0			,							20.0    )
    __magic_name__				     =       deterministic
    __magic_name__				     =       torch.exp(0.5 * self.logvar    )
    __magic_name__				     =       torch.exp(self.logvar    )
    if self.deterministic:
      __magic_name__				     =       __magic_name__				     =       torch.zeros_like(
          self.mean			,							device=self.parameters.device			,							dtype=self.parameters.dtype    )
  def 							_lowercase							( self			: Tuple			,							UpperCamelCase__			: Optional[torch.Generator] = None    )			->	torch.FloatTensor:







    """simple docstring"""






    __magic_name__				     =       randn_tensor(
        self.mean.shape			,							generator=UpperCamelCase__			,							device=self.parameters.device			,							dtype=self.parameters.dtype    )
    __magic_name__				     =       self.mean + self.std * sample
    return x
  def 							_lowercase							( self			: Dict			,							UpperCamelCase__			: Optional[int]=None    )			->	Any:







    """simple docstring"""






    if self.deterministic:
      return torch.Tensor([0.0]    )
    else:
      if other is None:
        return 0.5 * torch.sum(torch.pow(self.mean			,							2    ) + self.var - 1.0 - self.logvar			,							dim=[1, 2, 3]    )
      else:
        return 0.5 * torch.sum(
            torch.pow(self.mean - other.mean			,							2    ) / other.var
            + self.var / other.var
            - 1.0
            - self.logvar
            + other.logvar			,							dim=[1, 2, 3]			,							)
  def 							_lowercase							( self			: Optional[int]			,							UpperCamelCase__			: List[Any]			,							UpperCamelCase__			: Dict=[1, 2, 3]    )			->	Optional[int]:







    """simple docstring"""






    if self.deterministic:
      return torch.Tensor([0.0]    )
    __magic_name__				     =       np.log(2.0 * np.pi    )
    return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean			,							2    ) / self.var			,							dim=UpperCamelCase__    )





  def 							_lowercase							( self			: Optional[int]    )			->	Optional[int]:







    """simple docstring"""






    return self.mean

