
import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class 				lowercase					(		_SCREAMING_SNAKE_CASE  ,   _SCREAMING_SNAKE_CASE  ,   unittest.TestCase   ):
		__lowercase       :       Any	     =				CycleDiffusionPipeline
		__lowercase       :       Tuple	     =				TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
		    "negative_prompt",
		    "height",
		    "width",
		    "negative_prompt_embeds",
		}
		__lowercase       :       Union[str, Any]	     =				PipelineTesterMixin.required_optional_params - {"latents"}
		__lowercase       :       List[Any]	     =				TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"source_prompt"}   )
		__lowercase       :       Any	     =				IMAGE_TO_IMAGE_IMAGE_PARAMS
		__lowercase       :       Optional[Any]	     =				IMAGE_TO_IMAGE_IMAGE_PARAMS



		def __UpperCamelCase ( self       )  ->					List[str]:
				"""simple docstring"""
				torch.manual_seed(0       )
				UpperCamelCase		=     UNetaDConditionModel(
				    block_out_channels=(32, 64)				,      layers_per_block=2				,      sample_size=32				,      in_channels=4				,      out_channels=4				,      down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')				,      up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')				,      cross_attention_dim=32				,      )
				UpperCamelCase		=     DDIMScheduler(
				    beta_start=0.0_0085				,      beta_end=0.012				,      beta_schedule='scaled_linear'				,      num_train_timesteps=1_000				,      clip_sample=A_				,      set_alpha_to_one=A_				,      )
				torch.manual_seed(0       )
				UpperCamelCase		=     AutoencoderKL(
				    block_out_channels=[32, 64]				,      in_channels=3				,      out_channels=3				,      down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']				,      up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']				,      latent_channels=4				,      )
				torch.manual_seed(0       )
				UpperCamelCase		=     CLIPTextConfig(
				    bos_token_id=0				,      eos_token_id=2				,      hidden_size=32				,      intermediate_size=37				,      layer_norm_eps=1e-05				,      num_attention_heads=4				,      num_hidden_layers=5				,      pad_token_id=1				,      vocab_size=1_000				,      )
				UpperCamelCase		=     CLIPTextModel(A_       )
				UpperCamelCase		=     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       )

				UpperCamelCase		=     {
				    'unet': unet,
				    'scheduler': scheduler,
				    'vae': vae,
				    'text_encoder': text_encoder,
				    'tokenizer': tokenizer,
				    'safety_checker': None,
				    'feature_extractor': None,
				}
				return components



		def __UpperCamelCase ( self				,      A_				,      A_=0       )  ->					Optional[Any]:
				"""simple docstring"""
				UpperCamelCase		=     floats_tensor((1, 3, 32, 32)				,      rng=random.Random(A_       )       ).to(A_       )
				UpperCamelCase		=     image / 2 + 0.5
				if str(A_       ).startswith('mps'       ):
						UpperCamelCase		=     torch.manual_seed(A_       )
				else:
						UpperCamelCase		=     torch.Generator(device=A_       ).manual_seed(A_       )
				UpperCamelCase		=     {
				    'prompt': 'An astronaut riding an elephant',
				    'source_prompt': 'An astronaut riding a horse',
				    'image': image,
				    'generator': generator,
				    'num_inference_steps': 2,
				    'eta': 0.1,
				    'strength': 0.8,
				    'guidance_scale': 3,
				    'source_guidance_scale': 1,
				    'output_type': 'numpy',
				}
				return inputs



		def __UpperCamelCase ( self       )  ->					int:
				"""simple docstring"""
				UpperCamelCase		=     'cpu'  # ensure determinism for the device-dependent torch.Generator

				UpperCamelCase		=     self.get_dummy_components()
				UpperCamelCase		=     CycleDiffusionPipeline(**A_       )
				UpperCamelCase		=     pipe.to(A_       )
				pipe.set_progress_bar_config(disable=A_       )

				UpperCamelCase		=     self.get_dummy_inputs(A_       )
				UpperCamelCase		=     pipe(**A_       )
				UpperCamelCase		=     output.images

				UpperCamelCase		=     images[0, -3:, -3:, -1]

				assert images.shape == (1, 32, 32, 3)
				UpperCamelCase		=     np.array([0.4459, 0.4943, 0.4544, 0.6643, 0.5474, 0.4327, 0.5701, 0.5959, 0.5179]       )

				assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2



		@unittest.skipIf(torch_device != 'cuda'				,      'This test requires a GPU'       )
		def __UpperCamelCase ( self       )  ->					Union[str, Any]:
				"""simple docstring"""
				UpperCamelCase		=     self.get_dummy_components()
				for name, module in components.items():
						if hasattr(A_				,      'half'       ):
								UpperCamelCase		=     module.half()
				UpperCamelCase		=     CycleDiffusionPipeline(**A_       )
				UpperCamelCase		=     pipe.to(A_       )
				pipe.set_progress_bar_config(disable=A_       )

				UpperCamelCase		=     self.get_dummy_inputs(A_       )
				UpperCamelCase		=     pipe(**A_       )
				UpperCamelCase		=     output.images

				UpperCamelCase		=     images[0, -3:, -3:, -1]

				assert images.shape == (1, 32, 32, 3)
				UpperCamelCase		=     np.array([0.3506, 0.4543, 0.446, 0.4575, 0.5195, 0.4155, 0.5273, 0.518, 0.4116]       )

				assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2



		@skip_mps
		def __UpperCamelCase ( self       )  ->					Union[str, Any]:
				"""simple docstring"""
				return super().test_save_load_local()



		@unittest.skip('non-deterministic pipeline'       )
		def __UpperCamelCase ( self       )  ->					List[Any]:
				"""simple docstring"""
				return super().test_inference_batch_single_identical()



		@skip_mps
		def __UpperCamelCase ( self       )  ->					Any:
				"""simple docstring"""
				return super().test_dict_tuple_outputs_equivalent()



		@skip_mps
		def __UpperCamelCase ( self       )  ->					List[str]:
				"""simple docstring"""
				return super().test_save_load_optional_components()



		@skip_mps
		def __UpperCamelCase ( self       )  ->					Union[str, Any]:
				"""simple docstring"""
				return super().test_attention_slicing_forward_pass()






@slow
@require_torch_gpu
class 				lowercase					(		unittest.TestCase   ):



		def __UpperCamelCase ( self       )  ->					Optional[int]:
				"""simple docstring"""
				# clean up the VRAM after each test
				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()



		def __UpperCamelCase ( self       )  ->					Tuple:
				"""simple docstring"""
				UpperCamelCase		=     load_image(
				    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
				    '/cycle-diffusion/black_colored_car.png'       )
				UpperCamelCase		=     load_numpy(
				    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'       )
				UpperCamelCase		=     init_image.resize((512, 512)       )

				UpperCamelCase		=     'CompVis/stable-diffusion-v1-4'
				UpperCamelCase		=     DDIMScheduler.from_pretrained(A_				,      subfolder='scheduler'       )
				UpperCamelCase		=     CycleDiffusionPipeline.from_pretrained(
				    A_				,      scheduler=A_				,      safety_checker=A_				,      torch_dtype=torch.floataa				,      revision='fp16'       )

				pipe.to(A_       )
				pipe.set_progress_bar_config(disable=A_       )
				pipe.enable_attention_slicing()

				UpperCamelCase		=     'A black colored car'
				UpperCamelCase		=     'A blue colored car'

				UpperCamelCase		=     torch.manual_seed(0       )
				UpperCamelCase		=     pipe(
				    prompt=A_				,      source_prompt=A_				,      image=A_				,      num_inference_steps=100				,      eta=0.1				,      strength=0.85				,      guidance_scale=3				,      source_guidance_scale=1				,      generator=A_				,      output_type='np'				,      )
				UpperCamelCase		=     output.images

				# the values aren't exactly equal, but the images look the same visually
				assert np.abs(image - expected_image       ).max() < 5e-1



		def __UpperCamelCase ( self       )  ->					str:
				"""simple docstring"""
				UpperCamelCase		=     load_image(
				    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
				    '/cycle-diffusion/black_colored_car.png'       )
				UpperCamelCase		=     load_numpy(
				    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'       )
				UpperCamelCase		=     init_image.resize((512, 512)       )

				UpperCamelCase		=     'CompVis/stable-diffusion-v1-4'
				UpperCamelCase		=     DDIMScheduler.from_pretrained(A_				,      subfolder='scheduler'       )
				UpperCamelCase		=     CycleDiffusionPipeline.from_pretrained(A_				,      scheduler=A_				,      safety_checker=A_       )

				pipe.to(A_       )
				pipe.set_progress_bar_config(disable=A_       )
				pipe.enable_attention_slicing()

				UpperCamelCase		=     'A black colored car'
				UpperCamelCase		=     'A blue colored car'

				UpperCamelCase		=     torch.manual_seed(0       )
				UpperCamelCase		=     pipe(
				    prompt=A_				,      source_prompt=A_				,      image=A_				,      num_inference_steps=100				,      eta=0.1				,      strength=0.85				,      guidance_scale=3				,      source_guidance_scale=1				,      generator=A_				,      output_type='np'				,      )
				UpperCamelCase		=     output.images

				assert np.abs(image - expected_image       ).max() < 2e-2


import argparse

import torch
from torch import nn

from transformers import SpeechaTextConfig, SpeechaTextForConditionalGeneration


def       A       (					lowercase )   ->  Optional[Any]:
		'''simple docstring'''

		UpperCamelCase		=     [
		    'encoder.version',
		    'decoder.version',
		    'model.encoder.version',
		    'model.decoder.version',
		    'decoder.output_projection.weight',
		    '_float_tensor',
		    'encoder.embed_positions._float_tensor',
		    'decoder.embed_positions._float_tensor',
		]
		for k in ignore_keys:
				state_dict.pop(lowercase    ,       lowercase )


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

		UpperCamelCase		=     list(s_dict.keys() )
		for key in keys:
				if "transformer_layers" in key:
						UpperCamelCase		=     s_dict.pop(lowercase )
				elif "subsample" in key:
						UpperCamelCase		=     s_dict.pop(lowercase )


def       A       (					lowercase )   ->  Tuple:
		'''simple docstring'''

		UpperCamelCase	,  UpperCamelCase		=     emb.weight.shape
		UpperCamelCase		=     nn.Linear(lowercase    ,       lowercase    ,       bias=lowercase )
		UpperCamelCase		=     emb.weight.data
		return lin_layer


def       A       (					lowercase    ,       lowercase )   ->  Dict:
		'''simple docstring'''

		UpperCamelCase		=     torch.load(lowercase    ,       map_location='cpu' )
		UpperCamelCase		=     mam_aaa['args']
		UpperCamelCase		=     mam_aaa['model']
		UpperCamelCase		=     state_dict['decoder.output_projection.weight']

		remove_ignore_keys_(lowercase )
		rename_keys(lowercase )

		UpperCamelCase		=     state_dict['decoder.embed_tokens.weight'].shape[0]

		UpperCamelCase		=     args.share_decoder_input_output_embed

		UpperCamelCase		=     [int(lowercase ) for i in args.conv_kernel_sizes.split(',' )]
		UpperCamelCase		=     SpeechaTextConfig(
		    vocab_size=lowercase    ,       max_source_positions=args.max_source_positions    ,       max_target_positions=args.max_target_positions    ,       encoder_layers=args.encoder_layers    ,       decoder_layers=args.decoder_layers    ,       encoder_attention_heads=args.encoder_attention_heads    ,       decoder_attention_heads=args.decoder_attention_heads    ,       encoder_ffn_dim=args.encoder_ffn_embed_dim    ,       decoder_ffn_dim=args.decoder_ffn_embed_dim    ,       d_model=args.encoder_embed_dim    ,       dropout=args.dropout    ,       attention_dropout=args.attention_dropout    ,       activation_dropout=args.activation_dropout    ,       activation_function='relu'    ,       num_conv_layers=len(lowercase )    ,       conv_channels=args.conv_channels    ,       conv_kernel_sizes=lowercase    ,       input_feat_per_channel=args.input_feat_per_channel    ,       input_channels=args.input_channels    ,       tie_word_embeddings=lowercase    ,       num_beams=5    ,       max_length=200    ,       use_cache=lowercase    ,       decoder_start_token_id=2    ,       early_stopping=lowercase    ,       )

		UpperCamelCase		=     SpeechaTextForConditionalGeneration(lowercase )
		UpperCamelCase	,  UpperCamelCase		=     model.model.load_state_dict(lowercase    ,       strict=lowercase )
		if len(lowercase ) > 0 and not set(lowercase ) <= {
		    "encoder.embed_positions.weights",
		    "decoder.embed_positions.weights",
		}:
				raise ValueError(
				    'Only `encoder.embed_positions.weights` and `decoder.embed_positions.weights`  are allowed to be missing,'
				    f''' but all the following weights are missing {missing}''' )

		if tie_embeds:
				UpperCamelCase		=     make_linear_from_emb(model.model.decoder.embed_tokens )
		else:
				UpperCamelCase		=     lm_head_weights

		model.save_pretrained(lowercase )


if __name__ == "__main__":
					_UpperCAmelCase					: Optional[Any]      =       argparse.ArgumentParser()
					# Required parameters
					parser.add_argument("--fairseq_path", type=str, help="Path to the fairseq model (.pt) file.")
					parser.add_argument("--pytorch_dump_folder_path", default=None, type=str, help="Path to the output PyTorch model.")
					_UpperCAmelCase					: Dict      =       parser.parse_args()
					convert_fairseq_sat_checkpoint_to_tfms(args.fairseq_path, args.pytorch_dump_folder_path)
