






"""simple docstring"""





import re
import time
from typing import Optional

import IPython.display as disp

from ..trainer_callback import TrainerCallback
from ..trainer_utils import IntervalStrategy, has_length



def     __lowercase					(    a :				List[Any] )					->					Optional[Any]:
      __snake_case      :    str  			=int(a )
      __snake_case			,				__snake_case			,				__snake_case      :    Tuple  			=t // 3_600, (t // 60) % 60, t % 60
      return f'''{h}:{m:02d}:{s:02d}''' if h != 0 else f'''{m:02d}:{s:02d}'''



def     __lowercase					(    a :				Tuple       ,		a :				Optional[int]       ,		a :				str       ,		a :				Any       ,		a :				Optional[Any]=300 )					->					int:
      # docstyle-ignore
      return f'''
    <div>
      {prefix}
      <progress value=\'{value}\' max=\'{total}\' style=\'width:{width}px; height:20px; vertical-align: middle;\'></progress>
      {label}
    </div>
    '''



def     __lowercase					(    a :				Tuple )					->					int:
      __snake_case      :    Any  			='''<table border="1" class="dataframe">\n'''
      html_code += """  <thead>\n <tr style="text-align: left;">\n"""
      for i in items[0]:
            html_code += f'''      <th>{i}</th>\n'''
      html_code += "    </tr>\n  </thead>\n  <tbody>\n"
      for line in items[1:]:
            html_code += "    <tr>\n"
            for elt in line:
                  __snake_case      :    Union[str, Any]  			=f'''{elt:.6f}''' if isinstance(a       ,		a ) else str(a )
                  html_code += f'''      <td>{elt}</td>\n'''
            html_code += "    </tr>\n"
      html_code += "  </tbody>\n</table><p>"
      return html_code


class      _lowercase :
      _a   :     Optional[Any]								=   5
      _a   :     List[str]								=   0.2


      def __init__(    self				:	Any							,   a				:	int							,   a				:	Optional[str] = None							,   a				:	bool = True							,   a				:	Optional["NotebookTrainingTracker"] = None							,   a				:	int = 3_0_0							,   ):





            """simple docstring"""







            __snake_case      :    Tuple  			=total
            __snake_case      :    Optional[Any]  			='''''' if prefix is None else prefix
            __snake_case      :    List[str]  			=leave
            __snake_case      :    Optional[int]  			=parent
            __snake_case      :    str  			=width
            __snake_case      :    Dict  			=None
            __snake_case      :    Dict  			=None
            __snake_case      :    Optional[int]  			=None


      def _UpperCamelCase						(    self				:	List[str]							,   a				:	int							,   a				:	bool = False							,   a				:	str = None						):





            """simple docstring"""







            __snake_case      :    Any  			=value
            if comment is not None:
                  __snake_case      :    Union[str, Any]  			=comment
            if self.last_value is None:
                  __snake_case      :    List[Any]  			=time.time()
                  __snake_case      :    Any  			=value
                  __snake_case      :    List[str]  			=None
                  __snake_case      :    Dict  			=self.warmup
                  __snake_case      :    Dict  			=1
                  self.update_bar(a						)
            elif value <= self.last_value and not force_update:
                  return
            elif force_update or self.first_calls > 0 or value >= min(self.last_value + self.wait_for							,   self.total						):
                  if self.first_calls > 0:
                        self.first_calls -= 1
                  __snake_case      :    Any  			=time.time()
                  __snake_case      :    int  			=current_time - self.start_time
                  # We could have value = self.start_value if the update is called twixe with the same start value.
                  if value > self.start_value:
                        __snake_case      :    Dict  			=self.elapsed_time / (value - self.start_value)
                  else:
                        __snake_case      :    List[Any]  			=None
                  if value >= self.total:
                        __snake_case      :    Optional[Any]  			=self.total
                        __snake_case      :    Optional[int]  			=None
                        if not self.leave:
                              self.close()
                  elif self.average_time_per_item is not None:
                        __snake_case      :    str  			=self.average_time_per_item * (self.total - value)
                  self.update_bar(a						)
                  __snake_case      :    str  			=value
                  __snake_case      :    Optional[Any]  			=current_time
                  if self.average_time_per_item is None:
                        __snake_case      :    Tuple  			=1
                  else:
                        __snake_case      :    Dict  			=max(int(self.update_every / self.average_time_per_item						)							,   1						)


      def _UpperCamelCase						(    self				:	Union[str, Any]							,   a				:	Any							,   a				:	Optional[Any]=None						):





            """simple docstring"""







            __snake_case      :    str  			=''' ''' * (len(str(self.total						)						) - len(str(a						)						)) + str(a						)
            if self.elapsed_time is None:
                  __snake_case      :    Any  			=f'''[{spaced_value}/{self.total} : < :'''
            elif self.predicted_remaining is None:
                  __snake_case      :    List[str]  			=f'''[{spaced_value}/{self.total} {format_time(self.elapsed_time						)}'''
            else:
                  __snake_case      :    int  			=(
                      f'''[{spaced_value}/{self.total} {format_time(self.elapsed_time						)} <'''
                      f''' {format_time(self.predicted_remaining						)}'''
                  )
                  self.label += f''', {1/self.average_time_per_item:.2f} it/s'''
            self.label += "]" if self.comment is None or len(self.comment						) == 0 else f''', {self.comment}]'''
            self.display()


      def _UpperCamelCase						(    self				:	int						):





            """simple docstring"""







            __snake_case      :    Tuple  			=html_progress_bar(self.value							,   self.total							,   self.prefix							,   self.label							,   self.width						)
            if self.parent is not None:
                  # If this is a child bar, the parent will take care of the display.
                  self.parent.display()
                  return
            if self.output is None:
                  __snake_case      :    List[str]  			=disp.display(disp.HTML(self.html_code						)							,   display_id=a						)
            else:
                  self.output.update(disp.HTML(self.html_code						)						)







      def _UpperCamelCase						(    self				:	List[str]						):





            """simple docstring"""







            if self.parent is None and self.output is not None:
                  self.output.update(disp.HTML(''''''						)						)



class      _lowercase (     lowerCAmelCase						):


      def __init__(    self				:	int							,   a				:	Union[str, Any]							,   a				:	Tuple=None						):





            """simple docstring"""







            super().__init__(a						)
            __snake_case      :    Tuple  			=None if column_names is None else [column_names]
            __snake_case      :    int  			=None


      def _UpperCamelCase						(    self				:	List[Any]						):





            """simple docstring"""







            __snake_case      :    Any  			=html_progress_bar(self.value							,   self.total							,   self.prefix							,   self.label							,   self.width						)
            if self.inner_table is not None:
                  self.html_code += text_to_html_table(self.inner_table						)
            if self.child_bar is not None:
                  self.html_code += self.child_bar.html_code
            if self.output is None:
                  __snake_case      :    int  			=disp.display(disp.HTML(self.html_code						)							,   display_id=a						)
            else:
                  self.output.update(disp.HTML(self.html_code						)						)


      def _UpperCamelCase						(    self				:	Dict							,   a				:	str						):





            """simple docstring"""







            if self.inner_table is None:
                  __snake_case      :    List[str]  			=[list(values.keys()						), list(values.values()						)]
            else:
                  __snake_case      :    List[Any]  			=self.inner_table[0]
                  if len(self.inner_table						) == 1:
                        # We give a chance to update the column names at the first iteration
                        for key in values.keys():
                              if key not in columns:
                                    columns.append(a						)
                        __snake_case      :    Any  			=columns
                  self.inner_table.append([values[c] for c in columns]						)


      def _UpperCamelCase						(    self				:	Tuple							,   a				:	Union[str, Any]							,   a				:	Dict=None							,   a				:	int=3_0_0						):





            """simple docstring"""







            __snake_case      :    List[Any]  			=NotebookProgressBar(a							,   prefix=a							,   parent=self							,   width=a						)
            return self.child_bar







      def _UpperCamelCase						(    self				:	Optional[int]						):





            """simple docstring"""







            __snake_case      :    Tuple  			=None
            self.display()



class      _lowercase (     lowerCAmelCase						):


      def __init__(    self				:	int						):





            """simple docstring"""







            __snake_case      :    List[Any]  			=None
            __snake_case      :    int  			=None
            __snake_case      :    Union[str, Any]  			=False


      def _UpperCamelCase						(    self				:	Tuple							,   a				:	Any							,   a				:	Union[str, Any]							,   a				:	Union[str, Any]							,   **a				:	Tuple						):





            """simple docstring"""







            __snake_case      :    Union[str, Any]  			='''Epoch''' if args.evaluation_strategy == IntervalStrategy.EPOCH else '''Step'''
            __snake_case      :    Optional[int]  			=0
            __snake_case      :    Optional[int]  			=0
            __snake_case      :    str  			=[self.first_column] + ['''Training Loss''']
            if args.evaluation_strategy != IntervalStrategy.NO:
                  column_names.append('''Validation Loss'''						)
            __snake_case      :    List[Any]  			=NotebookTrainingTracker(state.max_steps							,   a						)


      def _UpperCamelCase						(    self				:	Tuple							,   a				:	Optional[Any]							,   a				:	int							,   a				:	str							,   **a				:	List[Any]						):





            """simple docstring"""







            __snake_case      :    List[str]  			=int(state.epoch						) if int(state.epoch						) == state.epoch else f'''{state.epoch:.2f}'''
            self.training_tracker.update(
                state.global_step + 1							,   comment=f'''Epoch {epoch}/{state.num_train_epochs}'''							,   force_update=self._force_next_update							,   )
            __snake_case      :    int  			=False


      def _UpperCamelCase						(    self				:	List[str]							,   a				:	int							,   a				:	Dict							,   a				:	Dict							,   a				:	Dict=None							,   **a				:	Union[str, Any]						):





            """simple docstring"""







            if not has_length(a						):
                  return
            if self.prediction_bar is None:
                  if self.training_tracker is not None:
                        __snake_case      :    int  			=self.training_tracker.add_child(len(a						)						)
                  else:
                        __snake_case      :    Any  			=NotebookProgressBar(len(a						)						)
                  self.prediction_bar.update(1						)
            else:
                  self.prediction_bar.update(self.prediction_bar.value + 1						)


      def _UpperCamelCase						(    self				:	List[Any]							,   a				:	str							,   a				:	str							,   a				:	List[str]							,   **a				:	Union[str, Any]						):





            """simple docstring"""







            if self.prediction_bar is not None:
                  self.prediction_bar.close()
            __snake_case      :    Optional[Any]  			=None


      def _UpperCamelCase						(    self				:	str							,   a				:	int							,   a				:	Optional[Any]							,   a				:	Optional[int]							,   a				:	Dict=None							,   **a				:	List[str]						):





            """simple docstring"""







            if args.evaluation_strategy == IntervalStrategy.NO and "loss" in logs:
                  __snake_case      :    Optional[int]  			={'''Training Loss''': logs['''loss''']}
                  # First column is necessarily Step sine we're not in epoch eval strategy
                  __snake_case      :    Dict  			=state.global_step
                  self.training_tracker.write_line(a						)


      def _UpperCamelCase						(    self				:	Union[str, Any]							,   a				:	Optional[Any]							,   a				:	List[Any]							,   a				:	List[str]							,   a				:	Dict=None							,   **a				:	Dict						):





            """simple docstring"""







            if self.training_tracker is not None:
                  __snake_case      :    List[Any]  			={'''Training Loss''': '''No log''', '''Validation Loss''': '''No log'''}
                  for log in reversed(state.log_history						):
                        if "loss" in log:
                              __snake_case      :    Union[str, Any]  			=log['''loss''']
                              break

                  if self.first_column == "Epoch":
                        __snake_case      :    Dict  			=int(state.epoch						)
                  else:
                        __snake_case      :    Dict  			=state.global_step
                  __snake_case      :    Dict  			='''eval'''
                  for k in metrics:
                        if k.endswith('''_loss'''						):
                              __snake_case      :    Any  			=re.sub(R'''\_loss$'''							,   ''''''							,   a						)
                  __snake_case      :    Tuple  			=metrics.pop('''total_flos'''							,   a						)
                  __snake_case      :    int  			=metrics.pop('''epoch'''							,   a						)
                  __snake_case      :    Any  			=metrics.pop(f'''{metric_key_prefix}_runtime'''							,   a						)
                  __snake_case      :    Union[str, Any]  			=metrics.pop(f'''{metric_key_prefix}_samples_per_second'''							,   a						)
                  __snake_case      :    Union[str, Any]  			=metrics.pop(f'''{metric_key_prefix}_steps_per_second'''							,   a						)
                  __snake_case      :    List[str]  			=metrics.pop(f'''{metric_key_prefix}_jit_compilation_time'''							,   a						)
                  for k, v in metrics.items():
                        if k == f'''{metric_key_prefix}_loss''':
                              __snake_case      :    List[Any]  			=v
                        else:
                              __snake_case      :    str  			=k.split('''_'''						)
                              __snake_case      :    int  			=''' '''.join([part.capitalize() for part in splits[1:]]						)
                              __snake_case      :    Optional[Any]  			=v
                  self.training_tracker.write_line(a						)
                  self.training_tracker.remove_child()
                  __snake_case      :    Optional[int]  			=None
                  # Evaluation takes a long time so we should force the next update.
                  __snake_case      :    int  			=True







      def _UpperCamelCase						(    self				:	Union[str, Any]							,   a				:	int							,   a				:	Tuple							,   a				:	Tuple							,   **a				:	Optional[int]						):





            """simple docstring"""







            self.training_tracker.update(
                state.global_step							,   comment=f'''Epoch {int(state.epoch						)}/{state.num_train_epochs}'''							,   force_update=a						)
            __snake_case      :    int  			=None










"""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     __lowercase					(    a :				Any       ,		a :				List[Any] )					->					Dict:
      __snake_case      :    Any  			=checkpoint

      __snake_case      :    Dict  			={}

      __snake_case      :    List[Any]  			=vae_state_dict['''encoder.conv_in.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''encoder.conv_in.bias''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''encoder.conv_out.weight''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''encoder.conv_out.bias''']
      __snake_case      :    str  			=vae_state_dict['''encoder.norm_out.weight''']
      __snake_case      :    str  			=vae_state_dict['''encoder.norm_out.bias''']

      __snake_case      :    Tuple  			=vae_state_dict['''decoder.conv_in.weight''']
      __snake_case      :    str  			=vae_state_dict['''decoder.conv_in.bias''']
      __snake_case      :    List[str]  			=vae_state_dict['''decoder.conv_out.weight''']
      __snake_case      :    Tuple  			=vae_state_dict['''decoder.conv_out.bias''']
      __snake_case      :    Union[str, Any]  			=vae_state_dict['''decoder.norm_out.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''decoder.norm_out.bias''']

      __snake_case      :    Tuple  			=vae_state_dict['''quant_conv.weight''']
      __snake_case      :    List[str]  			=vae_state_dict['''quant_conv.bias''']
      __snake_case      :    Optional[int]  			=vae_state_dict['''post_quant_conv.weight''']
      __snake_case      :    Tuple  			=vae_state_dict['''post_quant_conv.bias''']

      # Retrieves the keys for the encoder down blocks only
      __snake_case      :    Union[str, Any]  			=len({'''.'''.join(layer.split('''.''' )[:3] ) for layer in vae_state_dict if '''encoder.down''' in layer} )
      __snake_case      :    Union[str, Any]  			={
          layer_id: [key for key in vae_state_dict if f'''down.{layer_id}''' in key] for layer_id in range(a )
      }

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

      for i in range(a ):
            __snake_case      :    Any  			=[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:
                  __snake_case      :    Optional[Any]  			=vae_state_dict.pop(
                      f'''encoder.down.{i}.downsample.conv.weight''' )
                  __snake_case      :    Optional[int]  			=vae_state_dict.pop(
                      f'''encoder.down.{i}.downsample.conv.bias''' )

            __snake_case      :    Optional[Any]  			=renew_vae_resnet_paths(a )
            __snake_case      :    Union[str, Any]  			={'''old''': f'''down.{i}.block''', '''new''': f'''down_blocks.{i}.resnets'''}
            assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )

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

            __snake_case      :    List[str]  			=renew_vae_resnet_paths(a )
            __snake_case      :    List[str]  			={'''old''': f'''mid.block_{i}''', '''new''': f'''mid_block.resnets.{i - 1}'''}
            assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )

      __snake_case      :    int  			=[key for key in vae_state_dict if '''encoder.mid.attn''' in key]
      __snake_case      :    List[Any]  			=renew_vae_attention_paths(a )
      __snake_case      :    Union[str, Any]  			={'''old''': '''mid.attn_1''', '''new''': '''mid_block.attentions.0'''}
      assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )
      conv_attn_to_linear(a )

      for i in range(a ):
            __snake_case      :    List[Any]  			=num_up_blocks - 1 - i
            __snake_case      :    Dict  			=[
                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:
                  __snake_case      :    int  			=vae_state_dict[
                      f'''decoder.up.{block_id}.upsample.conv.weight'''
                  ]
                  __snake_case      :    Optional[Any]  			=vae_state_dict[
                      f'''decoder.up.{block_id}.upsample.conv.bias'''
                  ]

            __snake_case      :    List[str]  			=renew_vae_resnet_paths(a )
            __snake_case      :    int  			={'''old''': f'''up.{block_id}.block''', '''new''': f'''up_blocks.{i}.resnets'''}
            assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )

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

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

      __snake_case      :    Optional[Any]  			=[key for key in vae_state_dict if '''decoder.mid.attn''' in key]
      __snake_case      :    Dict  			=renew_vae_attention_paths(a )
      __snake_case      :    List[str]  			={'''old''': '''mid.attn_1''', '''new''': '''mid_block.attentions.0'''}
      assign_to_checkpoint(a       ,		a       ,		a       ,		additional_replacements=[meta_path]       ,		config=a )
      conv_attn_to_linear(a )
      return new_checkpoint



def     __lowercase					(    a :				str       ,		a :				str       ,		)					->					Optional[int]:
      # Only support V1
      __snake_case      :    Union[str, Any]  			=requests.get(
          ''' https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml''' )
      __snake_case      :    Union[str, Any]  			=io.BytesIO(r.content )

      __snake_case      :    Any  			=OmegaConf.load(a )
      __snake_case      :    Union[str, Any]  			=512
      __snake_case      :    List[str]  			='''cuda''' if torch.cuda.is_available() else '''cpu'''
      if checkpoint_path.endswith('''safetensors''' ):
            from safetensors import safe_open

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

      # Convert the VAE model.
      __snake_case      :    Dict  			=create_vae_diffusers_config(a       ,		image_size=a )
      __snake_case      :    Union[str, Any]  			=custom_convert_ldm_vae_checkpoint(a       ,		a )

      __snake_case      :    str  			=AutoencoderKL(**a )
      vae.load_state_dict(a )
      vae.save_pretrained(a )


if __name__ == "__main__":
    UpperCamelCase_	:				Optional[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.""")

    UpperCamelCase_	:				Optional[Any]										= parser.parse_args()

    vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)


