



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 A						(       __UpperCamelCase ,	__UpperCamelCase					)						->      Optional[int]:
			A__       =    checkpoint

			A__       =    {}

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

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

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

			# Retrieves the keys for the encoder down blocks only
			A__       =    len({'.'.join(layer.split('.'					)[:3]					) for layer in vae_state_dict if 'encoder.down' in layer}					)
			A__       =    {
			    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
			A__       =    len({'.'.join(layer.split('.'					)[:3]					) for layer in vae_state_dict if 'decoder.up' in layer}					)
			A__       =    {
			    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					):
						A__       =    [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:
									A__       =    vae_state_dict.pop(
									    f'''encoder.down.{i}.downsample.conv.weight'''					)
									A__       =    vae_state_dict.pop(
									    f'''encoder.down.{i}.downsample.conv.bias'''					)

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

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

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

			A__       =    [key for key in vae_state_dict if 'encoder.mid.attn' in key]
			A__       =    renew_vae_attention_paths(__UpperCamelCase					)
			A__       =    {'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					):
						A__       =    num_up_blocks - 1 - i
						A__       =    [
						    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:
									A__       =    vae_state_dict[
									    f'''decoder.up.{block_id}.upsample.conv.weight'''
									]
									A__       =    vae_state_dict[
									    f'''decoder.up.{block_id}.upsample.conv.bias'''
									]

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

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

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

			A__       =    [key for key in vae_state_dict if 'decoder.mid.attn' in key]
			A__       =    renew_vae_attention_paths(__UpperCamelCase					)
			A__       =    {'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 A						(       __UpperCamelCase ,	__UpperCamelCase ,	)						->      Tuple:
			# Only support V1
			A__       =    requests.get(
			    ' https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml'					)
			A__       =    io.BytesIO(r.content					)

			A__       =    OmegaConf.load(__UpperCamelCase					)
			A__       =    512
			A__       =    'cuda' if torch.cuda.is_available() else 'cpu'
			if checkpoint_path.endswith('safetensors'					):
						from safetensors import safe_open

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

			# Convert the VAE model.
			A__       =    create_vae_diffusers_config(__UpperCamelCase ,	image_size=__UpperCamelCase					)
			A__       =    custom_convert_ldm_vae_checkpoint(__UpperCamelCase ,	__UpperCamelCase					)

			A__       =    AutoencoderKL(**__UpperCamelCase					)
			vae.load_state_dict(__UpperCamelCase					)
			vae.save_pretrained(__UpperCamelCase					)


if __name__ == "__main__":
							SCREAMING_SNAKE_CASE__       =      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.''')

							SCREAMING_SNAKE_CASE__       =      parser.parse_args()

							vae_pt_to_vae_diffuser(args.vae_pt_path, args.dump_path)









import pytest

import datasets


# Import fixture modules as plugins
SCREAMING_SNAKE_CASE__       =      ['''tests.fixtures.files''', '''tests.fixtures.hub''', '''tests.fixtures.fsspec''']


def A						(       __UpperCamelCase ,	__UpperCamelCase					)						->      Optional[int]:
			# Mark tests as "unit" by default if not marked as "integration" (or already marked as "unit")
			for item in items:
						if any(marker in item.keywords for marker in ['integration', 'unit']					):
									continue
						item.add_marker(pytest.mark.unit					)


def A						(       __UpperCamelCase					)						->      str:
			config.addinivalue_line('markers' ,	'torchaudio_latest: mark test to run with torchaudio>=0.12'					)


@pytest.fixture(autouse=__UpperCamelCase					)
def A						(       __UpperCamelCase ,	__UpperCamelCase					)						->      Union[str, Any]:
			# test_hf_cache_home = tmp_path_factory.mktemp("cache")  # TODO: why a cache dir per test function does not work?
			A__       =    tmp_path_factory.getbasetemp() / 'cache'
			A__       =    test_hf_cache_home / 'datasets'
			A__       =    test_hf_cache_home / 'metrics'
			A__       =    test_hf_cache_home / 'modules'
			monkeypatch.setattr('datasets.config.HF_DATASETS_CACHE' ,	str(__UpperCamelCase					)					)
			monkeypatch.setattr('datasets.config.HF_METRICS_CACHE' ,	str(__UpperCamelCase					)					)
			monkeypatch.setattr('datasets.config.HF_MODULES_CACHE' ,	str(__UpperCamelCase					)					)
			A__       =    test_hf_datasets_cache / 'downloads'
			monkeypatch.setattr('datasets.config.DOWNLOADED_DATASETS_PATH' ,	str(__UpperCamelCase					)					)
			A__       =    test_hf_datasets_cache / 'downloads' / 'extracted'
			monkeypatch.setattr('datasets.config.EXTRACTED_DATASETS_PATH' ,	str(__UpperCamelCase					)					)


@pytest.fixture(autouse=__UpperCamelCase ,	scope='session'					)
def A						(       )						->      Union[str, Any]:
			datasets.disable_progress_bar()


@pytest.fixture(autouse=__UpperCamelCase					)
def A						(       __UpperCamelCase					)						->      int:
			# don't take tests into account when counting downloads
			monkeypatch.setattr('datasets.config.HF_UPDATE_DOWNLOAD_COUNTS' ,	__UpperCamelCase					)






@pytest.fixture
def A						(       __UpperCamelCase					)						->      Any:
			# Required to suppress RemovedIn20Warning when feature(s) are not compatible with SQLAlchemy 2.0
			# To be removed once SQLAlchemy 2.0 supported
			monkeypatch.setattr('sqlalchemy.util.deprecations.SILENCE_UBER_WARNING' ,	__UpperCamelCase					)




