


"""simple docstring"""







import argparse

import huggingface_hub
import k_diffusion as K
import torch

from diffusers import UNetaDConditionModel


__lowercase          = """pcuenq/k-upscaler"""


def 	lowercase     (							A_    ,  A_    ,  *, A_    ,  A_			)-> List[Any]:
  '''simple docstring'''







  a		:						int 				=     {
      # norm1
      F'''{diffusers_resnet_prefix}.norm1.linear.weight''': checkpoint[F'''{resnet_prefix}.main.0.mapper.weight'''],
      F'''{diffusers_resnet_prefix}.norm1.linear.bias''': checkpoint[F'''{resnet_prefix}.main.0.mapper.bias'''],
      # conv1
      F'''{diffusers_resnet_prefix}.conv1.weight''': checkpoint[F'''{resnet_prefix}.main.2.weight'''],
      F'''{diffusers_resnet_prefix}.conv1.bias''': checkpoint[F'''{resnet_prefix}.main.2.bias'''],
      # norm2
      F'''{diffusers_resnet_prefix}.norm2.linear.weight''': checkpoint[F'''{resnet_prefix}.main.4.mapper.weight'''],
      F'''{diffusers_resnet_prefix}.norm2.linear.bias''': checkpoint[F'''{resnet_prefix}.main.4.mapper.bias'''],
      # conv2
      F'''{diffusers_resnet_prefix}.conv2.weight''': checkpoint[F'''{resnet_prefix}.main.6.weight'''],
      F'''{diffusers_resnet_prefix}.conv2.bias''': checkpoint[F'''{resnet_prefix}.main.6.bias'''],
  }

  if resnet.conv_shortcut is not None:
    rv.update(
        {
            F'''{diffusers_resnet_prefix}.conv_shortcut.weight''': checkpoint[F'''{resnet_prefix}.skip.weight'''],
        }			)

  return rv


def 	lowercase     (							A_    ,  *, A_    ,  A_			)-> List[Any]:
  '''simple docstring'''







  a		:						Any 				=     checkpoint[F'''{attention_prefix}.qkv_proj.weight'''].chunk(3    ,  dim=0			)
  a		:						Tuple 				=     checkpoint[F'''{attention_prefix}.qkv_proj.bias'''].chunk(3    ,  dim=0			)
  a		:						Optional[int] 				=     {
      # norm
      F'''{diffusers_attention_prefix}.norm1.linear.weight''': checkpoint[F'''{attention_prefix}.norm_in.mapper.weight'''],
      F'''{diffusers_attention_prefix}.norm1.linear.bias''': checkpoint[F'''{attention_prefix}.norm_in.mapper.bias'''],
      # to_q
      F'''{diffusers_attention_prefix}.attn1.to_q.weight''': weight_q.squeeze(-1			).squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn1.to_q.bias''': bias_q,
      # to_k
      F'''{diffusers_attention_prefix}.attn1.to_k.weight''': weight_k.squeeze(-1			).squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn1.to_k.bias''': bias_k,
      # to_v
      F'''{diffusers_attention_prefix}.attn1.to_v.weight''': weight_v.squeeze(-1			).squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn1.to_v.bias''': bias_v,
      # to_out
      F'''{diffusers_attention_prefix}.attn1.to_out.0.weight''': checkpoint[F'''{attention_prefix}.out_proj.weight''']
      .squeeze(-1			)
      .squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn1.to_out.0.bias''': checkpoint[F'''{attention_prefix}.out_proj.bias'''],
  }

  return rv


def 	lowercase     (							A_    ,  *, A_    ,  A_    ,  A_			)-> str:
  '''simple docstring'''







  a		:						Optional[int] 				=     checkpoint[F'''{attention_prefix}.kv_proj.weight'''].chunk(2    ,  dim=0			)
  a		:						Union[str, Any] 				=     checkpoint[F'''{attention_prefix}.kv_proj.bias'''].chunk(2    ,  dim=0			)

  a		:						Optional[int] 				=     {
      # norm2 (ada groupnorm)
      F'''{diffusers_attention_prefix}.norm{diffusers_attention_index}.linear.weight''': checkpoint[
          F'''{attention_prefix}.norm_dec.mapper.weight'''
      ],
      F'''{diffusers_attention_prefix}.norm{diffusers_attention_index}.linear.bias''': checkpoint[
          F'''{attention_prefix}.norm_dec.mapper.bias'''
      ],
      # layernorm on encoder_hidden_state
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.norm_cross.weight''': checkpoint[
          F'''{attention_prefix}.norm_enc.weight'''
      ],
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.norm_cross.bias''': checkpoint[
          F'''{attention_prefix}.norm_enc.bias'''
      ],
      # to_q
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_q.weight''': checkpoint[
          F'''{attention_prefix}.q_proj.weight'''
      ]
      .squeeze(-1			)
      .squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_q.bias''': checkpoint[
          F'''{attention_prefix}.q_proj.bias'''
      ],
      # to_k
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_k.weight''': weight_k.squeeze(-1			).squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_k.bias''': bias_k,
      # to_v
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_v.weight''': weight_v.squeeze(-1			).squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_v.bias''': bias_v,
      # to_out
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_out.0.weight''': checkpoint[
          F'''{attention_prefix}.out_proj.weight'''
      ]
      .squeeze(-1			)
      .squeeze(-1			),
      F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_out.0.bias''': checkpoint[
          F'''{attention_prefix}.out_proj.bias'''
      ],
  }

  return rv


def 	lowercase     (							A_    ,  A_    ,  A_    ,  A_			)-> List[Any]:
  '''simple docstring'''







  a		:						Optional[Any] 				=     """inner_model.u_net.u_blocks""" if block_type == """up""" else """inner_model.u_net.d_blocks"""
  a		:						str 				=     F'''{block_prefix}.{block_idx}'''

  a		:						List[str] 				=     {}

  if not hasattr(_lowercase    ,  "attentions"			):
    a		:						Dict 				=     1  # resnet only
  elif not block.attentions[0].add_self_attention:
    a		:						Dict 				=     2  # resnet -> cross-attention
  else:
    a		:						Any 				=     3  # resnet -> self-attention -> cross-attention)

  for resnet_idx, resnet in enumerate(block.resnets			):
    # diffusers_resnet_prefix = f"{diffusers_up_block_prefix}.resnets.{resnet_idx}"
    a		:						Union[str, Any] 				=     F'''{block_type}_blocks.{block_idx}.resnets.{resnet_idx}'''
    a		:						List[Any] 				=     n * resnet_idx if block_type == """up""" else n * resnet_idx + 1
    a		:						Union[str, Any] 				=     F'''{block_prefix}.{idx}''' if block_type == """up""" else F'''{block_prefix}.{idx}'''

    diffusers_checkpoint.update(
        resnet_to_diffusers_checkpoint(
            _lowercase    ,  _lowercase    ,  diffusers_resnet_prefix=_lowercase    ,  resnet_prefix=_lowercase			)			)

  if hasattr(_lowercase    ,  "attentions"			):
    for attention_idx, attention in enumerate(block.attentions			):
      a		:						Optional[int] 				=     F'''{block_type}_blocks.{block_idx}.attentions.{attention_idx}'''
      a		:						Optional[Any] 				=     n * attention_idx + 1 if block_type == """up""" else n * attention_idx + 2
      a		:						str 				=     F'''{block_prefix}.{idx}'''
      a		:						Optional[int] 				=     F'''{block_prefix}.{idx }'''
      a		:						int 				=     1 if not attention.add_self_attention else 2
      a		:						int 				=     (
          n * attention_idx + cross_attention_index
          if block_type == """up"""
          else n * attention_idx + cross_attention_index + 1
      )
      a		:						Optional[Any] 				=     F'''{block_prefix}.{idx }'''

      diffusers_checkpoint.update(
          cross_attn_to_diffusers_checkpoint(
              _lowercase    ,  diffusers_attention_prefix=_lowercase    ,  diffusers_attention_index=2    ,  attention_prefix=_lowercase    ,  )			)

      if attention.add_self_attention is True:
        diffusers_checkpoint.update(
            self_attn_to_diffusers_checkpoint(
                _lowercase    ,  diffusers_attention_prefix=_lowercase    ,  attention_prefix=_lowercase    ,  )			)

  return diffusers_checkpoint


def 	lowercase     (							A_    ,  A_			)-> Any:
  '''simple docstring'''







  a		:						Optional[Any] 				=     {}

  # pre-processing
  diffusers_checkpoint.update(
      {
          "conv_in.weight": checkpoint["inner_model.proj_in.weight"],
          "conv_in.bias": checkpoint["inner_model.proj_in.bias"],
      }			)

  # timestep and class embedding
  diffusers_checkpoint.update(
      {
          "time_proj.weight": checkpoint["inner_model.timestep_embed.weight"].squeeze(-1			),
          "time_embedding.linear_1.weight": checkpoint["inner_model.mapping.0.weight"],
          "time_embedding.linear_1.bias": checkpoint["inner_model.mapping.0.bias"],
          "time_embedding.linear_2.weight": checkpoint["inner_model.mapping.2.weight"],
          "time_embedding.linear_2.bias": checkpoint["inner_model.mapping.2.bias"],
          "time_embedding.cond_proj.weight": checkpoint["inner_model.mapping_cond.weight"],
      }			)

  # down_blocks
  for down_block_idx, down_block in enumerate(model.down_blocks			):
    diffusers_checkpoint.update(block_to_diffusers_checkpoint(_lowercase    ,  _lowercase    ,  _lowercase    ,  "down"			)			)

  # up_blocks
  for up_block_idx, up_block in enumerate(model.up_blocks			):
    diffusers_checkpoint.update(block_to_diffusers_checkpoint(_lowercase    ,  _lowercase    ,  _lowercase    ,  "up"			)			)

  # post-processing
  diffusers_checkpoint.update(
      {
          "conv_out.weight": checkpoint["inner_model.proj_out.weight"],
          "conv_out.bias": checkpoint["inner_model.proj_out.bias"],
      }			)

  return diffusers_checkpoint


def 	lowercase     (							A_			)-> Any:
  '''simple docstring'''







  a		:						Union[str, Any] 				=     original_config["""input_channels"""] + original_config["""unet_cond_dim"""]
  a		:						str 				=     original_config["""input_channels"""] + (1 if original_config["""has_variance"""] else 0)

  a		:						List[Any] 				=     original_config["""channels"""]

  assert (
      len(set(original_config["depths"]			)			) == 1
  ), "UNet2DConditionModel currently do not support blocks with different number of layers"
  a		:						Union[str, Any] 				=     original_config["""depths"""][0]

  a		:						Optional[Any] 				=     original_config["""mapping_cond_dim"""]
  a		:						Optional[int] 				=     original_config["""cross_cond_dim"""]

  a		:						str 				=     []
  a		:						Optional[int] 				=     []
  for s, c in zip(original_config["self_attn_depths"]    ,  original_config["cross_attn_depths"]			):
    if s:
      a		:						Union[str, Any] 				=     """self"""
      a		:						str 				=     """cross""" if c else None
    elif c:
      a		:						Any 				=     """cross"""
      a		:						Tuple 				=     None
    else:
      a		:						List[str] 				=     None
      a		:						Union[str, Any] 				=     None
    attna_types.append(_lowercase			)
    attna_types.append(_lowercase			)

  a		:						List[str] 				=     UNetaDConditionModel(
      in_channels=_lowercase    ,  out_channels=_lowercase    ,  down_block_types=("KDownBlock2D", "KCrossAttnDownBlock2D", "KCrossAttnDownBlock2D", "KCrossAttnDownBlock2D")    ,  mid_block_type=_lowercase    ,  up_block_types=("KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KUpBlock2D")    ,  block_out_channels=_lowercase    ,  layers_per_block=_lowercase    ,  act_fn="gelu"    ,  norm_num_groups=_lowercase    ,  cross_attention_dim=_lowercase    ,  attention_head_dim=64    ,  time_cond_proj_dim=_lowercase    ,  resnet_time_scale_shift="scale_shift"    ,  time_embedding_type="fourier"    ,  timestep_post_act="gelu"    ,  conv_in_kernel=1    ,  conv_out_kernel=1    ,  )

  return unet


def 	lowercase     (							A_			)-> List[Any]:
  '''simple docstring'''







  a		:						List[Any] 				=     torch.device("cuda" if torch.cuda.is_available() else "cpu"			)

  a		:						List[str] 				=     huggingface_hub.hf_hub_download(_lowercase    ,  "config_laion_text_cond_latent_upscaler_2.json"			)
  a		:						Tuple 				=     huggingface_hub.hf_hub_download(
      _lowercase    ,  "laion_text_cond_latent_upscaler_2_1_00470000_slim.pth"			)
  print(F'''loading original model configuration from {orig_config_path}'''			)
  print(F'''loading original model checkpoint from {orig_weights_path}'''			)

  print("converting to diffusers unet"			)
  a		:						int 				=     K.config.load_config(open(_lowercase			)			)["""model"""]
  a		:						int 				=     unet_model_from_original_config(_lowercase			)

  a		:						Union[str, Any] 				=     torch.load(_lowercase    ,  map_location=_lowercase			)["""model_ema"""]
  a		:						List[Any] 				=     unet_to_diffusers_checkpoint(_lowercase    ,  _lowercase			)

  model.load_state_dict(_lowercase    ,  strict=_lowercase			)
  model.save_pretrained(args.dump_path			)
  print(F'''saving converted unet model in {args.dump_path}'''			)


if __name__ == "__main__":
 __lowercase          = argparse.ArgumentParser()

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

 main(args)











'''simple docstring'''




# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

from ..utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


a  :     Optional[Any]       =       {
    """agents""": ["""Agent""", """AzureOpenAiAgent""", """HfAgent""", """LocalAgent""", """OpenAiAgent"""],
    """base""": ["""PipelineTool""", """RemoteTool""", """Tool""", """launch_gradio_demo""", """load_tool"""],
}

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							a  :     Dict       =       ["""DocumentQuestionAnsweringTool"""]
							a  :     List[str]       =       ["""ImageCaptioningTool"""]
							a  :     Tuple       =       ["""ImageQuestionAnsweringTool"""]
							a  :     Tuple       =       ["""ImageSegmentationTool"""]
							a  :     List[str]       =       ["""SpeechToTextTool"""]
							a  :     Dict       =       ["""TextClassificationTool"""]
							a  :     Union[str, Any]       =       ["""TextQuestionAnsweringTool"""]
							a  :     List[str]       =       ["""TextSummarizationTool"""]
							a  :     Optional[Any]       =       ["""TextToSpeechTool"""]
							a  :     str       =       ["""TranslationTool"""]

if TYPE_CHECKING:
							from .agents import Agent, AzureOpenAiAgent, HfAgent, LocalAgent, OpenAiAgent
							from .base import PipelineTool, RemoteTool, Tool, launch_gradio_demo, load_tool

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .document_question_answering import DocumentQuestionAnsweringTool
														from .image_captioning import ImageCaptioningTool
														from .image_question_answering import ImageQuestionAnsweringTool
														from .image_segmentation import ImageSegmentationTool
														from .speech_to_text import SpeechToTextTool
														from .text_classification import TextClassificationTool
														from .text_question_answering import TextQuestionAnsweringTool
														from .text_summarization import TextSummarizationTool
														from .text_to_speech import TextToSpeechTool
														from .translation import TranslationTool
else:
							import sys

							a  :     Dict       =       _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)

