






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 			snake_case__   (							lowercase_	,    lowercase_	,    unittest.TestCase):

       '''simple docstring'''
       lowerCamelCase							:				Union[str, Any] 				= CycleDiffusionPipeline
       lowerCamelCase							:				Optional[int] 				= TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
           'negative_prompt',
           'height',
           'width',
           'negative_prompt_embeds',
       }
       lowerCamelCase							:				Union[str, Any] 				= PipelineTesterMixin.required_optional_params - {'latents'}
       lowerCamelCase							:				Dict 				= TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"source_prompt"})
       lowerCamelCase							:				str 				= IMAGE_TO_IMAGE_IMAGE_PARAMS
       lowerCamelCase							:				List[str] 				= IMAGE_TO_IMAGE_IMAGE_PARAMS



       def 					__lowercase						(							self    ) ->      List[str]:






            '''simple docstring'''





            torch.manual_seed(0    )
            __snake_case			:Optional[int]          =						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			,						)
            __snake_case			:int          =						DDIMScheduler(
                beta_start=0.0_00_85			,						beta_end=0.0_12			,						beta_schedule="""scaled_linear"""			,						num_train_timesteps=10_00			,						clip_sample=UpperCAmelCase__			,						set_alpha_to_one=UpperCAmelCase__			,						)
            torch.manual_seed(0    )
            __snake_case			:Tuple          =						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    )
            __snake_case			:Any          =						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=10_00			,						)
            __snake_case			:List[str]          =						CLIPTextModel(UpperCAmelCase__    )
            __snake_case			:Union[str, Any]          =						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

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



       def 					__lowercase						(							self			,						a__			,						a__=0    ) ->      Optional[Any]:






            '''simple docstring'''





            __snake_case			:List[Any]          =						floats_tensor((1, 3, 32, 32)			,						rng=random.Random(UpperCAmelCase__    )    ).to(UpperCAmelCase__    )
            __snake_case			:str          =						image / 2 + 0.5
            if str(UpperCAmelCase__    ).startswith("""mps"""    ):
                 __snake_case			:Any          =						torch.manual_seed(UpperCAmelCase__    )
            else:
                 __snake_case			:Dict          =						torch.Generator(device=UpperCAmelCase__    ).manual_seed(UpperCAmelCase__    )
            __snake_case			:Tuple          =						{
                '''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 					__lowercase						(							self    ) ->      List[str]:






            '''simple docstring'''





            __snake_case			:Optional[int]          =						'''cpu'''  # ensure determinism for the device-dependent torch.Generator

            __snake_case			:Dict          =						self.get_dummy_components()
            __snake_case			:List[str]          =						CycleDiffusionPipeline(**UpperCAmelCase__    )
            __snake_case			:Dict          =						pipe.to(UpperCAmelCase__    )
            pipe.set_progress_bar_config(disable=UpperCAmelCase__    )

            __snake_case			:int          =						self.get_dummy_inputs(UpperCAmelCase__    )
            __snake_case			:Optional[int]          =						pipe(**UpperCAmelCase__    )
            __snake_case			:str          =						output.images

            __snake_case			:str          =						images[0, -3:, -3:, -1]

            assert images.shape == (1, 32, 32, 3)
            __snake_case			:Optional[int]          =						np.array([0.44_59, 0.49_43, 0.45_44, 0.66_43, 0.54_74, 0.43_27, 0.57_01, 0.59_59, 0.51_79]    )

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



       @unittest.skipIf(torch_device != """cuda"""			,						"""This test requires a GPU"""    )
       def 					__lowercase						(							self    ) ->      Optional[int]:






            '''simple docstring'''





            __snake_case			:Optional[int]          =						self.get_dummy_components()
            for name, module in components.items():
                 if hasattr(UpperCAmelCase__			,						"""half"""    ):
                      __snake_case			:List[str]          =						module.half()
            __snake_case			:List[str]          =						CycleDiffusionPipeline(**UpperCAmelCase__    )
            __snake_case			:Tuple          =						pipe.to(UpperCAmelCase__    )
            pipe.set_progress_bar_config(disable=UpperCAmelCase__    )

            __snake_case			:Tuple          =						self.get_dummy_inputs(UpperCAmelCase__    )
            __snake_case			:Optional[int]          =						pipe(**UpperCAmelCase__    )
            __snake_case			:Optional[int]          =						output.images

            __snake_case			:List[Any]          =						images[0, -3:, -3:, -1]

            assert images.shape == (1, 32, 32, 3)
            __snake_case			:str          =						np.array([0.35_06, 0.45_43, 0.4_46, 0.45_75, 0.51_95, 0.41_55, 0.52_73, 0.5_18, 0.41_16]    )

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



       @skip_mps
       def 					__lowercase						(							self    ) ->      Union[str, Any]:






            '''simple docstring'''





            return super().test_save_load_local()



       @unittest.skip("""non-deterministic pipeline"""    )
       def 					__lowercase						(							self    ) ->      int:






            '''simple docstring'''





            return super().test_inference_batch_single_identical()



       @skip_mps
       def 					__lowercase						(							self    ) ->      Optional[int]:






            '''simple docstring'''





            return super().test_dict_tuple_outputs_equivalent()



       @skip_mps
       def 					__lowercase						(							self    ) ->      Dict:






            '''simple docstring'''





            return super().test_save_load_optional_components()



       @skip_mps
       def 					__lowercase						(							self    ) ->      Tuple:






            '''simple docstring'''





            return super().test_attention_slicing_forward_pass()




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

       '''simple docstring'''



       def 					__lowercase						(							self    ) ->      Dict:






            '''simple docstring'''





            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()



       def 					__lowercase						(							self    ) ->      Any:






            '''simple docstring'''





            __snake_case			:Any          =						load_image(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                """/cycle-diffusion/black_colored_car.png"""    )
            __snake_case			:Tuple          =						load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy"""    )
            __snake_case			:List[Any]          =						init_image.resize((5_12, 5_12)    )

            __snake_case			:Dict          =						'''CompVis/stable-diffusion-v1-4'''
            __snake_case			:Any          =						DDIMScheduler.from_pretrained(UpperCAmelCase__			,						subfolder="""scheduler"""    )
            __snake_case			:List[Any]          =						CycleDiffusionPipeline.from_pretrained(
                UpperCAmelCase__			,						scheduler=UpperCAmelCase__			,						safety_checker=UpperCAmelCase__			,						torch_dtype=torch.floataa			,						revision="""fp16"""    )

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

            __snake_case			:Tuple          =						'''A black colored car'''
            __snake_case			:str          =						'''A blue colored car'''

            __snake_case			:Union[str, Any]          =						torch.manual_seed(0    )
            __snake_case			:List[Any]          =						pipe(
                prompt=UpperCAmelCase__			,						source_prompt=UpperCAmelCase__			,						image=UpperCAmelCase__			,						num_inference_steps=1_00			,						eta=0.1			,						strength=0.85			,						guidance_scale=3			,						source_guidance_scale=1			,						generator=UpperCAmelCase__			,						output_type="""np"""			,						)
            __snake_case			:Any          =						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 					__lowercase						(							self    ) ->      List[str]:






            '''simple docstring'''





            __snake_case			:Optional[int]          =						load_image(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                """/cycle-diffusion/black_colored_car.png"""    )
            __snake_case			:Optional[int]          =						load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"""    )
            __snake_case			:str          =						init_image.resize((5_12, 5_12)    )

            __snake_case			:Optional[Any]          =						'''CompVis/stable-diffusion-v1-4'''
            __snake_case			:Union[str, Any]          =						DDIMScheduler.from_pretrained(UpperCAmelCase__			,						subfolder="""scheduler"""    )
            __snake_case			:Tuple          =						CycleDiffusionPipeline.from_pretrained(UpperCAmelCase__			,						scheduler=UpperCAmelCase__			,						safety_checker=UpperCAmelCase__    )

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

            __snake_case			:List[Any]          =						'''A black colored car'''
            __snake_case			:Tuple          =						'''A blue colored car'''

            __snake_case			:Optional[int]          =						torch.manual_seed(0    )
            __snake_case			:Tuple          =						pipe(
                prompt=UpperCAmelCase__			,						source_prompt=UpperCAmelCase__			,						image=UpperCAmelCase__			,						num_inference_steps=1_00			,						eta=0.1			,						strength=0.85			,						guidance_scale=3			,						source_guidance_scale=1			,						generator=UpperCAmelCase__			,						output_type="""np"""			,						)
            __snake_case			:str          =						output.images

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










def   UpperCamelCase  (       snake_case__     :   str	,snake_case__     :   int  ):




     '''simple docstring'''






     __snake_case			:list[list[str]]          =						[[] for _ in range(snake_case__  )]
     __snake_case			:Union[str, Any]          =						key - 1

     if key <= 0:
          raise ValueError("""Height of grid can't be 0 or negative"""  )
     if key == 1 or len(snake_case__  ) <= key:
          return input_string

     for position, character in enumerate(snake_case__  ):
          __snake_case			:Any          =						position % (lowest * 2)  # puts it in bounds
          __snake_case			:Optional[int]          =						min(snake_case__	,lowest * 2 - num  )  # creates zigzag pattern
          temp_grid[num].append(snake_case__  )
     __snake_case			:List[Any]          =						["""""".join(snake_case__  ) for row in temp_grid]
     __snake_case			:Any          =						"""""".join(snake_case__  )

     return output_string






def   UpperCamelCase  (       snake_case__     :   str	,snake_case__     :   int  ):




     '''simple docstring'''






     __snake_case			:Union[str, Any]          =						[]
     __snake_case			:Optional[int]          =						key - 1

     if key <= 0:
          raise ValueError("""Height of grid can't be 0 or negative"""  )
     if key == 1:
          return input_string

     __snake_case			:list[list[str]]          =						[[] for _ in range(snake_case__  )]  # generates template
     for position in range(len(snake_case__  )  ):
          __snake_case			:List[str]          =						position % (lowest * 2)  # puts it in bounds
          __snake_case			:int          =						min(snake_case__	,lowest * 2 - num  )  # creates zigzag pattern
          temp_grid[num].append("""*"""  )

     __snake_case			:str          =						0
     for row in temp_grid:  # fills in the characters
          __snake_case			:str          =						input_string[counter : counter + len(snake_case__  )]
          grid.append(list(snake_case__  )  )
          counter += len(snake_case__  )

     __snake_case			:Any          =						""""""  # reads as zigzag
     for position in range(len(snake_case__  )  ):
          __snake_case			:Optional[int]          =						position % (lowest * 2)  # puts it in bounds
          __snake_case			:int          =						min(snake_case__	,lowest * 2 - num  )  # creates zigzag pattern
          output_string += grid[num][0]
          grid[num].pop(0  )
     return output_string






def   UpperCamelCase  (       snake_case__     :   str  ):




     '''simple docstring'''






     __snake_case			:Optional[Any]          =						{}
     for key_guess in range(1	,len(snake_case__  )  ):  # tries every key
          __snake_case			:Dict          =						decrypt(snake_case__	,snake_case__  )
     return results


if __name__ == "__main__":
     import doctest

     doctest.testmod()


