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      lowerCamelCase    (SCREAMING_SNAKE_CASE__     ,					SCREAMING_SNAKE_CASE__     ,					unittest.TestCase    ):
  """simple docstring"""







  lowerCamelCase__   =  CycleDiffusionPipeline
  lowerCamelCase__   =  TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
      '''negative_prompt''',
      '''height''',
      '''width''',
      '''negative_prompt_embeds''',
  }
  lowerCamelCase__   =  PipelineTesterMixin.required_optional_params - {'''latents'''}
  lowerCamelCase__   =  TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''source_prompt'''}    )
  lowerCamelCase__   =  IMAGE_TO_IMAGE_IMAGE_PARAMS
  lowerCamelCase__   =  IMAGE_TO_IMAGE_IMAGE_PARAMS







  def       __A   (							self       :				List[Any]      )     ->		List[Any]:
        torch.manual_seed(0      )
        SCREAMING_SNAKE_CASE_      =     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     ,   )
        SCREAMING_SNAKE_CASE_      =     DDIMScheduler(
            beta_start=0.0_0085     ,   beta_end=0.012     ,   beta_schedule="scaled_linear"     ,   num_train_timesteps=1_000     ,   clip_sample=__magic_name__     ,   set_alpha_to_one=__magic_name__     ,   )
        torch.manual_seed(0      )
        SCREAMING_SNAKE_CASE_      =     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      )
        SCREAMING_SNAKE_CASE_      =     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     ,   )
        SCREAMING_SNAKE_CASE_      =     CLIPTextModel(__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      )

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







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

        SCREAMING_SNAKE_CASE_      =     self.get_dummy_components()
        SCREAMING_SNAKE_CASE_      =     CycleDiffusionPipeline(**__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     pipe.to(__magic_name__      )
        pipe.set_progress_bar_config(disable=__magic_name__      )

        SCREAMING_SNAKE_CASE_      =     self.get_dummy_inputs(__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     pipe(**__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     output.images

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

        assert images.shape == (1, 32, 32, 3)
        SCREAMING_SNAKE_CASE_      =     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       __A   (							self       :				Any      )     ->		Optional[int]:
        SCREAMING_SNAKE_CASE_      =     self.get_dummy_components()
        for name, module in components.items():
              if hasattr(__magic_name__     ,   "half"      ):
                    SCREAMING_SNAKE_CASE_      =     module.half()
        SCREAMING_SNAKE_CASE_      =     CycleDiffusionPipeline(**__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     pipe.to(__magic_name__      )
        pipe.set_progress_bar_config(disable=__magic_name__      )

        SCREAMING_SNAKE_CASE_      =     self.get_dummy_inputs(__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     pipe(**__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     output.images

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

        assert images.shape == (1, 32, 32, 3)
        SCREAMING_SNAKE_CASE_      =     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       __A   (							self       :				Optional[int]      )     ->		Union[str, Any]:
        return super().test_save_load_local()







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







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







  @skip_mps
  def       __A   (							self       :				List[Any]      )     ->		List[Any]:
        return super().test_save_load_optional_components()







  @skip_mps
  def       __A   (							self       :				int      )     ->		Any:
        return super().test_attention_slicing_forward_pass()



@slow
@require_torch_gpu
class      lowerCamelCase    (unittest.TestCase    ):
  """simple docstring"""







  def       __A   (							self       :				List[str]      )     ->		List[Any]:
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()







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

        SCREAMING_SNAKE_CASE_      =     "CompVis/stable-diffusion-v1-4"
        SCREAMING_SNAKE_CASE_      =     DDIMScheduler.from_pretrained(__magic_name__     ,   subfolder="scheduler"      )
        SCREAMING_SNAKE_CASE_      =     CycleDiffusionPipeline.from_pretrained(
            __magic_name__     ,   scheduler=__magic_name__     ,   safety_checker=__magic_name__     ,   torch_dtype=torch.floataa     ,   revision="fp16"      )

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

        SCREAMING_SNAKE_CASE_      =     "A black colored car"
        SCREAMING_SNAKE_CASE_      =     "A blue colored car"

        SCREAMING_SNAKE_CASE_      =     torch.manual_seed(0      )
        SCREAMING_SNAKE_CASE_      =     pipe(
            prompt=__magic_name__     ,   source_prompt=__magic_name__     ,   image=__magic_name__     ,   num_inference_steps=100     ,   eta=0.1     ,   strength=0.85     ,   guidance_scale=3     ,   source_guidance_scale=1     ,   generator=__magic_name__     ,   output_type="np"     ,   )
        SCREAMING_SNAKE_CASE_      =     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       :				Dict      )     ->		int:
        SCREAMING_SNAKE_CASE_      =     load_image(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
            "/cycle-diffusion/black_colored_car.png"      )
        SCREAMING_SNAKE_CASE_      =     load_numpy(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy"      )
        SCREAMING_SNAKE_CASE_      =     init_image.resize((512, 512)      )

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

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

        SCREAMING_SNAKE_CASE_      =     "A black colored car"
        SCREAMING_SNAKE_CASE_      =     "A blue colored car"

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

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



def    a__       (						__UpperCamelCase ):
      if not head:
            return True
      # split the list to two parts
      SCREAMING_SNAKE_CASE_   ,       SCREAMING_SNAKE_CASE_      =     head.next, head
      while fast and fast.next:
            SCREAMING_SNAKE_CASE_      =     fast.next.next
            SCREAMING_SNAKE_CASE_      =     slow.next
      SCREAMING_SNAKE_CASE_      =     slow.next
      SCREAMING_SNAKE_CASE_      =     None  # Don't forget here! But forget still works!
      # reverse the second part
      SCREAMING_SNAKE_CASE_      =     None
      while second:
            SCREAMING_SNAKE_CASE_      =     second.next
            SCREAMING_SNAKE_CASE_      =     node
            SCREAMING_SNAKE_CASE_      =     second
            SCREAMING_SNAKE_CASE_      =     nxt
      # compare two parts
      # second part has the same or one less node
      while node:
            if node.val != head.val:
                  return False
            SCREAMING_SNAKE_CASE_      =     node.next
            SCREAMING_SNAKE_CASE_      =     head.next
      return True


def    a__       (						__UpperCamelCase ):
      if not head or not head.next:
            return True

      # 1. Get the midpoint (slow)
      SCREAMING_SNAKE_CASE_      =     SCREAMING_SNAKE_CASE_      =     SCREAMING_SNAKE_CASE_      =     head
      while fast and fast.next:
            SCREAMING_SNAKE_CASE_   ,       SCREAMING_SNAKE_CASE_      =     fast.next.next, slow.next

      # 2. Push the second half into the stack
      SCREAMING_SNAKE_CASE_      =     [slow.val]
      while slow.next:
            SCREAMING_SNAKE_CASE_      =     slow.next
            stack.append(slow.val )

      # 3. Comparison
      while stack:
            if stack.pop() != cur.val:
                  return False
            SCREAMING_SNAKE_CASE_      =     cur.next

      return True





def    a__       (						__UpperCamelCase ):
      if not head or not head.next:
            return True
      SCREAMING_SNAKE_CASE_      =     {}
      SCREAMING_SNAKE_CASE_      =     0
      while head:
            if head.val in d:
                  d[head.val].append(__UpperCamelCase )
            else:
                  SCREAMING_SNAKE_CASE_      =     [pos]
            SCREAMING_SNAKE_CASE_      =     head.next
            pos += 1
      SCREAMING_SNAKE_CASE_      =     pos - 1
      SCREAMING_SNAKE_CASE_      =     0
      for v in d.values():
            if len(__UpperCamelCase ) % 2 != 0:
                  middle += 1
            else:
                  SCREAMING_SNAKE_CASE_      =     0
                  for i in range(0   ,					len(__UpperCamelCase ) ):
                        if v[i] + v[len(__UpperCamelCase ) - 1 - step] != checksum:
                              return False
                        step += 1
            if middle > 1:
                  return False
      return True


