

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    _a (       __SCREAMING_SNAKE_CASE			,							__SCREAMING_SNAKE_CASE			,							unittest.TestCase							):




      """simple docstring"""

      _lowerCamelCase    :       Optional[int]							=    CycleDiffusionPipeline
      _lowerCamelCase    :       Any							=    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
          'negative_prompt',
          'height',
          'width',
          'negative_prompt_embeds',
      }
      _lowerCamelCase    :       int							=    PipelineTesterMixin.required_optional_params - {'latents'}
      _lowerCamelCase    :       Any							=    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'source_prompt'}							)
      _lowerCamelCase    :       Tuple							=    IMAGE_TO_IMAGE_IMAGE_PARAMS
      _lowerCamelCase    :       int							=    IMAGE_TO_IMAGE_IMAGE_PARAMS





      def 						__A     (					self				:							int     ):
            torch.manual_seed(0     )
            A_		   =							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		,       )
            A_		   =							DDIMScheduler(
                beta_start=0.00_085		,       beta_end=0.012		,       beta_schedule="scaled_linear"		,       num_train_timesteps=1000		,       clip_sample=_a		,       set_alpha_to_one=_a		,       )
            torch.manual_seed(0     )
            A_		   =							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     )
            A_		   =							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=1000		,       )
            A_		   =							CLIPTextModel(_a     )
            A_		   =							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

            A_		   =							{
                "unet": unet,
                "scheduler": scheduler,
                "vae": vae,
                "text_encoder": text_encoder,
                "tokenizer": tokenizer,
                "safety_checker": None,
                "feature_extractor": None,
            }
            return components





      def 						__A     (					self				:							List[Any]		,       UpperCAmelCase				:							int		,       UpperCAmelCase				:							Optional[int]=0     ):
            A_		   =							floats_tensor((1, 3, 32, 32)		,       rng=random.Random(_a     )     ).to(_a     )
            A_		   =							image / 2 + 0.5
            if str(_a     ).startswith("mps"     ):
                  A_		   =							torch.manual_seed(_a     )
            else:
                  A_		   =							torch.Generator(device=_a     ).manual_seed(_a     )
            A_		   =							{
                "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 						__A     (					self				:							Any     ):
            A_		   =							"cpu"  # ensure determinism for the device-dependent torch.Generator

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

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

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

            assert images.shape == (1, 32, 32, 3)
            A_		   =							np.array([0.4_459, 0.4_943, 0.4_544, 0.6_643, 0.5_474, 0.4_327, 0.5_701, 0.5_959, 0.5_179]     )

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





      @unittest.skipIf(torch_device != "cuda"		,       "This test requires a GPU"     )
      def 						__A     (					self				:							List[Any]     ):
            A_		   =							self.get_dummy_components()
            for name, module in components.items():
                  if hasattr(_a		,       "half"     ):
                        A_		   =							module.half()
            A_		   =							CycleDiffusionPipeline(**_a     )
            A_		   =							pipe.to(_a     )
            pipe.set_progress_bar_config(disable=_a     )

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

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

            assert images.shape == (1, 32, 32, 3)
            A_		   =							np.array([0.3_506, 0.4_543, 0.446, 0.4_575, 0.5_195, 0.4_155, 0.5_273, 0.518, 0.4_116]     )

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





      @skip_mps
      def 						__A     (					self				:							str     ):
            return super().test_save_load_local()





      @unittest.skip("non-deterministic pipeline"     )
      def 						__A     (					self				:							Any     ):
            return super().test_inference_batch_single_identical()





      @skip_mps
      def 						__A     (					self				:							Tuple     ):
            return super().test_dict_tuple_outputs_equivalent()





      @skip_mps
      def 						__A     (					self				:							List[str]     ):
            return super().test_save_load_optional_components()







      @skip_mps
      def 						__A     (					self				:							Optional[Any]     ):
            return super().test_attention_slicing_forward_pass()




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




      """simple docstring"""

      def 						__A     (					self				:							Tuple     ):
            # clean up the VRAM after each test
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()





      def 						__A     (					self				:							Union[str, Any]     ):
            A_		   =							load_image(
                "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                "/cycle-diffusion/black_colored_car.png"     )
            A_		   =							load_numpy(
                "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"     )
            A_		   =							init_image.resize((512, 512)     )

            A_		   =							"CompVis/stable-diffusion-v1-4"
            A_		   =							DDIMScheduler.from_pretrained(_a		,       subfolder="scheduler"     )
            A_		   =							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()

            A_		   =							"A black colored car"
            A_		   =							"A blue colored car"

            A_		   =							torch.manual_seed(0     )
            A_		   =							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"		,       )
            A_		   =							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 						__A     (					self				:							List[Any]     ):
            A_		   =							load_image(
                "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                "/cycle-diffusion/black_colored_car.png"     )
            A_		   =							load_numpy(
                "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"     )
            A_		   =							init_image.resize((512, 512)     )

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

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

            A_		   =							"A black colored car"
            A_		   =							"A blue colored car"

            A_		   =							torch.manual_seed(0     )
            A_		   =							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"		,       )
            A_		   =							output.images

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



"""simple docstring"""




import warnings

from ..trainer import Trainer
from ..utils import logging


lowercase_			     =  logging.get_logger(__name__)
class 					__lowerCAmelCase					(   __SCREAMING_SNAKE_CASE       ):


		'''simple docstring'''



		def __init__(  self							, _a=None							, **_a      ):
				warnings.warn(
				    '''`SageMakerTrainer` is deprecated and will be removed in v5 of Transformers. You can use `Trainer` '''
				    '''instead.'''							, _a							, )
				super().__init__(args=_a							, **_a      )






