





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        __SCREAMING_SNAKE_CASE      (			lowerCAmelCase_					,lowerCAmelCase_					,unittest.TestCase ):
       _UpperCAmelCase				:  int			   =   CycleDiffusionPipeline
       _UpperCAmelCase				:  Union[str, Any]			   =   TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
           "negative_prompt",
           "height",
           "width",
           "negative_prompt_embeds",
       }
       _UpperCAmelCase				:  Union[str, Any]			   =   PipelineTesterMixin.required_optional_params - {"latents"}
       _UpperCAmelCase				:  Optional[int]			   =   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"source_prompt"} )
       _UpperCAmelCase				:  Optional[int]			   =   IMAGE_TO_IMAGE_IMAGE_PARAMS
       _UpperCAmelCase				:  List[Any]			   =   IMAGE_TO_IMAGE_IMAGE_PARAMS







       def 						__lowerCamelCase   (				self					:     List[str]      )       ->List[str]:
          torch.manual_seed(0      )
          lowerCamelCase__       :						Tuple        =							UNetaDConditionModel(
              block_out_channels=(3_2, 6_4)		, layers_per_block=2		, sample_size=3_2		, in_channels=4		, out_channels=4		, down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		, up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		, cross_attention_dim=3_2		, )
          lowerCamelCase__       :						Optional[int]        =							DDIMScheduler(
              beta_start=0.0_00_85		, beta_end=0.0_12		, beta_schedule='''scaled_linear'''		, num_train_timesteps=1_0_0_0		, clip_sample=A		, set_alpha_to_one=A		, )
          torch.manual_seed(0      )
          lowerCamelCase__       :						Dict        =							AutoencoderKL(
              block_out_channels=[3_2, 6_4]		, in_channels=3		, out_channels=3		, down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		, up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		, latent_channels=4		, )
          torch.manual_seed(0      )
          lowerCamelCase__       :						Union[str, Any]        =							CLIPTextConfig(
              bos_token_id=0		, eos_token_id=2		, hidden_size=3_2		, intermediate_size=3_7		, layer_norm_eps=1e-05		, num_attention_heads=4		, num_hidden_layers=5		, pad_token_id=1		, vocab_size=1_0_0_0		, )
          lowerCamelCase__       :						Tuple        =							CLIPTextModel(A      )
          lowerCamelCase__       :						Dict        =							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

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







       def 						__lowerCamelCase   (				self					:     int		, A					:     int		, A					:     List[Any]=0      )       ->List[str]:
          lowerCamelCase__       :						Dict        =							floats_tensor((1, 3, 3_2, 3_2)		, rng=random.Random(A      )      ).to(A      )
          lowerCamelCase__       :						Optional[Any]        =							image / 2 + 0.5
          if str(A      ).startswith('''mps'''      ):
             lowerCamelCase__       :						int        =							torch.manual_seed(A      )
          else:
             lowerCamelCase__       :						Union[str, Any]        =							torch.Generator(device=A      ).manual_seed(A      )
          lowerCamelCase__       :						str        =							{
              '''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 						__lowerCamelCase   (				self					:     str      )       ->Any:
          lowerCamelCase__       :						Tuple        =							'''cpu'''  # ensure determinism for the device-dependent torch.Generator

          lowerCamelCase__       :						Optional[Any]        =							self.get_dummy_components()
          lowerCamelCase__       :						Optional[int]        =							CycleDiffusionPipeline(**A      )
          lowerCamelCase__       :						List[Any]        =							pipe.to(A      )
          pipe.set_progress_bar_config(disable=A      )

          lowerCamelCase__       :						str        =							self.get_dummy_inputs(A      )
          lowerCamelCase__       :						Any        =							pipe(**A      )
          lowerCamelCase__       :						Optional[Any]        =							output.images

          lowerCamelCase__       :						Dict        =							images[0, -3:, -3:, -1]

          assert images.shape == (1, 3_2, 3_2, 3)
          lowerCamelCase__       :						List[Any]        =							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 						__lowerCamelCase   (				self					:     str      )       ->List[str]:
          lowerCamelCase__       :						List[str]        =							self.get_dummy_components()
          for name, module in components.items():
             if hasattr(A		, '''half'''      ):
                lowerCamelCase__       :						str        =							module.half()
          lowerCamelCase__       :						List[str]        =							CycleDiffusionPipeline(**A      )
          lowerCamelCase__       :						Any        =							pipe.to(A      )
          pipe.set_progress_bar_config(disable=A      )

          lowerCamelCase__       :						Dict        =							self.get_dummy_inputs(A      )
          lowerCamelCase__       :						Any        =							pipe(**A      )
          lowerCamelCase__       :						Union[str, Any]        =							output.images

          lowerCamelCase__       :						List[str]        =							images[0, -3:, -3:, -1]

          assert images.shape == (1, 3_2, 3_2, 3)
          lowerCamelCase__       :						Any        =							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 						__lowerCamelCase   (				self					:     int      )       ->Optional[int]:
          return super().test_save_load_local()







       @unittest.skip('''non-deterministic pipeline'''      )
       def 						__lowerCamelCase   (				self					:     Optional[Any]      )       ->Dict:
          return super().test_inference_batch_single_identical()







       @skip_mps
       def 						__lowerCamelCase   (				self					:     List[Any]      )       ->Optional[Any]:
          return super().test_dict_tuple_outputs_equivalent()







       @skip_mps
       def 						__lowerCamelCase   (				self					:     Any      )       ->int:
          return super().test_save_load_optional_components()







       @skip_mps
       def 						__lowerCamelCase   (				self					:     str      )       ->Union[str, Any]:
          return super().test_attention_slicing_forward_pass()


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







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







       def 						__lowerCamelCase   (				self					:     Optional[int]      )       ->Any:
          lowerCamelCase__       :						Optional[Any]        =							load_image(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
              '''/cycle-diffusion/black_colored_car.png'''      )
          lowerCamelCase__       :						Any        =							load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'''      )
          lowerCamelCase__       :						Any        =							init_image.resize((5_1_2, 5_1_2)      )

          lowerCamelCase__       :						int        =							'''CompVis/stable-diffusion-v1-4'''
          lowerCamelCase__       :						Tuple        =							DDIMScheduler.from_pretrained(A		, subfolder='''scheduler'''      )
          lowerCamelCase__       :						str        =							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()

          lowerCamelCase__       :						Dict        =							'''A black colored car'''
          lowerCamelCase__       :						Optional[Any]        =							'''A blue colored car'''

          lowerCamelCase__       :						List[str]        =							torch.manual_seed(0      )
          lowerCamelCase__       :						Tuple        =							pipe(
              prompt=A		, source_prompt=A		, image=A		, num_inference_steps=1_0_0		, eta=0.1		, strength=0.85		, guidance_scale=3		, source_guidance_scale=1		, generator=A		, output_type='''np'''		, )
          lowerCamelCase__       :						List[str]        =							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 						__lowerCamelCase   (				self					:     Tuple      )       ->int:
          lowerCamelCase__       :						Any        =							load_image(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
              '''/cycle-diffusion/black_colored_car.png'''      )
          lowerCamelCase__       :						Optional[Any]        =							load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'''      )
          lowerCamelCase__       :						List[Any]        =							init_image.resize((5_1_2, 5_1_2)      )

          lowerCamelCase__       :						Tuple        =							'''CompVis/stable-diffusion-v1-4'''
          lowerCamelCase__       :						Any        =							DDIMScheduler.from_pretrained(A		, subfolder='''scheduler'''      )
          lowerCamelCase__       :						Dict        =							CycleDiffusionPipeline.from_pretrained(A		, scheduler=A		, safety_checker=A      )

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

          lowerCamelCase__       :						int        =							'''A black colored car'''
          lowerCamelCase__       :						Tuple        =							'''A blue colored car'''

          lowerCamelCase__       :						Optional[int]        =							torch.manual_seed(0      )
          lowerCamelCase__       :						Union[str, Any]        =							pipe(
              prompt=A		, source_prompt=A		, image=A		, num_inference_steps=1_0_0		, eta=0.1		, strength=0.85		, guidance_scale=3		, source_guidance_scale=1		, generator=A		, output_type='''np'''		, )
          lowerCamelCase__       :						Tuple        =							output.images

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










import os
from pathlib import Path



def 	_a						(   )    ->    Tuple:







   """simple docstring"""
   from torch.utils.cpp_extension import load

   lowerCamelCase__       :						List[Any]        =							Path(UpperCAmelCase					).resolve().parent.parent.parent / '''kernels''' / '''deformable_detr'''
   lowerCamelCase__       :						Any        =							[
       root / filename
       for filename in [
           '''vision.cpp''',
           os.path.join('''cpu'''      ,  '''ms_deform_attn_cpu.cpp'''					),
           os.path.join('''cuda'''      ,  '''ms_deform_attn_cuda.cu'''					),
       ]
   ]

   load(
       '''MultiScaleDeformableAttention'''      ,  UpperCAmelCase      ,  with_cuda=UpperCAmelCase      ,  extra_include_paths=[str(UpperCAmelCase					)]      ,  extra_cflags=['''-DWITH_CUDA=1''']      ,  extra_cuda_cflags=[
           '''-DCUDA_HAS_FP16=1''',
           '''-D__CUDA_NO_HALF_OPERATORS__''',
           '''-D__CUDA_NO_HALF_CONVERSIONS__''',
           '''-D__CUDA_NO_HALF2_OPERATORS__''',
       ]      ,  )

   import MultiScaleDeformableAttention as MSDA

   return MSDA



