






import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
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

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





def 				snake_case  (     lowerCamelCase					):




 '''simple docstring'''







 __lowercase             =  [tensor.shape for tensor in tensor_list]
 return all(shape == shapes[0] for shape in shapes[1:]					)


class        __UpperCamelCase    (     __A   ,			__A   ,			__A   ,			unittest.TestCase   ):
 __snake_case	:str   =       StableDiffusionLatentUpscalePipeline
 __snake_case	:Dict   =       TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
     'height',
     'width',
     'cross_attention_kwargs',
     'negative_prompt_embeds',
     'prompt_embeds',
 }
 __snake_case	:int   =       PipelineTesterMixin.required_optional_params - {'num_images_per_prompt'}
 __snake_case	:Any   =       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 __snake_case	:Dict   =       frozenset(
     []   )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 __snake_case	:str   =       frozenset([]   )

 __snake_case	:Optional[Any]   =       True







 @property
 def  _a   (  self      :							Optional[Any]			)		-> Optional[Any]:
  """simple docstring"""







  __lowercase             =  1
  __lowercase             =  4
  __lowercase             =  (16, 16)

  __lowercase             =  floats_tensor((batch_size, num_channels) + sizes       ,				rng=random.Random(0			)			).to(_lowerCAmelCase			)
  return image







 def  _a   (  self      :							Optional[Any]			)		-> List[str]:
  """simple docstring"""







  torch.manual_seed(0			)
  __lowercase             =  UNetaDConditionModel(
      act_fn="""gelu"""       ,				attention_head_dim=8       ,				norm_num_groups=_lowerCAmelCase       ,				block_out_channels=[32, 32, 64, 64]       ,				time_cond_proj_dim=160       ,				conv_in_kernel=1       ,				conv_out_kernel=1       ,				cross_attention_dim=32       ,				down_block_types=(
          """KDownBlock2D""",
          """KCrossAttnDownBlock2D""",
          """KCrossAttnDownBlock2D""",
          """KCrossAttnDownBlock2D""",
      )       ,				in_channels=8       ,				mid_block_type=_lowerCAmelCase       ,				only_cross_attention=_lowerCAmelCase       ,				out_channels=5       ,				resnet_time_scale_shift="""scale_shift"""       ,				time_embedding_type="""fourier"""       ,				timestep_post_act="""gelu"""       ,				up_block_types=("""KCrossAttnUpBlock2D""", """KCrossAttnUpBlock2D""", """KCrossAttnUpBlock2D""", """KUpBlock2D""")       ,				)
  __lowercase             =  AutoencoderKL(
      block_out_channels=[32, 32, 64, 64]       ,				in_channels=3       ,				out_channels=3       ,				down_block_types=[
          """DownEncoderBlock2D""",
          """DownEncoderBlock2D""",
          """DownEncoderBlock2D""",
          """DownEncoderBlock2D""",
      ]       ,				up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D""", """UpDecoderBlock2D""", """UpDecoderBlock2D"""]       ,				latent_channels=4       ,				)
  __lowercase             =  EulerDiscreteScheduler(prediction_type="""sample"""			)
  __lowercase             =  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       ,				hidden_act="""quick_gelu"""       ,				projection_dim=512       ,				)
  __lowercase             =  CLIPTextModel(_lowerCAmelCase			)
  __lowercase             =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

  __lowercase             =  {
      '''unet''': model.eval(),
      '''vae''': vae.eval(),
      '''scheduler''': scheduler,
      '''text_encoder''': text_encoder,
      '''tokenizer''': tokenizer,
  }

  return components







 def  _a   (  self      :							Union[str, Any]       ,				_lowerCAmelCase      :							Optional[int]       ,				_lowerCAmelCase      :							Optional[int]=0			)		-> List[str]:
  """simple docstring"""







  if str(_lowerCAmelCase			).startswith("""mps"""			):
   __lowercase             =  torch.manual_seed(_lowerCAmelCase			)
  else:
   __lowercase             =  torch.Generator(device=_lowerCAmelCase			).manual_seed(_lowerCAmelCase			)
  __lowercase             =  {
      '''prompt''': '''A painting of a squirrel eating a burger''',
      '''image''': self.dummy_image.cpu(),
      '''generator''': generator,
      '''num_inference_steps''': 2,
      '''output_type''': '''numpy''',
  }
  return inputs







 def  _a   (  self      :							str			)		-> List[Any]:
  """simple docstring"""







  __lowercase             =  '''cpu'''

  __lowercase             =  self.get_dummy_components()
  __lowercase             =  self.pipeline_class(**_lowerCAmelCase			)
  pipe.to(_lowerCAmelCase			)
  pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

  __lowercase             =  self.get_dummy_inputs(_lowerCAmelCase			)
  __lowercase             =  pipe(**_lowerCAmelCase			).images
  __lowercase             =  image[0, -3:, -3:, -1]

  self.assertEqual(image.shape       ,				(1, 256, 256, 3)			)
  __lowercase             =  np.array(
      [0.47_222_412, 0.41_921_633, 0.44_717_434, 0.46_874_192, 0.42_588_258, 0.46_150_726, 0.4_677_534, 0.45_583_832, 0.48_579_055]			)
  __lowercase             =  np.abs(image_slice.flatten() - expected_slice			).max()
  self.assertLessEqual(_lowerCAmelCase       ,				1e-3			)







 def  _a   (  self      :							List[Any]			)		-> List[str]:
  """simple docstring"""







  super().test_attention_slicing_forward_pass(expected_max_diff=7e-3			)







 def  _a   (  self      :							str			)		-> Optional[Any]:
  """simple docstring"""







  super().test_cpu_offload_forward_pass(expected_max_diff=3e-3			)







 def  _a   (  self      :							str			)		-> int:
  """simple docstring"""







  super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3			)







 def  _a   (  self      :							str			)		-> List[str]:
  """simple docstring"""







  super().test_inference_batch_single_identical(expected_max_diff=7e-3			)







 def  _a   (  self      :							Dict			)		-> int:
  """simple docstring"""







  super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3e-3			)







 def  _a   (  self      :							int			)		-> Tuple:
  """simple docstring"""







  super().test_save_load_local(expected_max_difference=3e-3			)







 def  _a   (  self      :							List[Any]			)		-> Optional[int]:
  """simple docstring"""







  super().test_save_load_optional_components(expected_max_difference=3e-3			)







 def  _a   (  self      :							Dict			)		-> Dict:
  """simple docstring"""







  __lowercase             =  [
      '''DDIMScheduler''',
      '''DDPMScheduler''',
      '''PNDMScheduler''',
      '''HeunDiscreteScheduler''',
      '''EulerAncestralDiscreteScheduler''',
      '''KDPM2DiscreteScheduler''',
      '''KDPM2AncestralDiscreteScheduler''',
      '''DPMSolverSDEScheduler''',
  ]
  __lowercase             =  self.get_dummy_components()
  __lowercase             =  self.pipeline_class(**_lowerCAmelCase			)

  # make sure that PNDM does not need warm-up
  pipe.scheduler.register_to_config(skip_prk_steps=_lowerCAmelCase			)

  pipe.to(_lowerCAmelCase			)
  pipe.set_progress_bar_config(disable=_lowerCAmelCase			)
  __lowercase             =  self.get_dummy_inputs(_lowerCAmelCase			)
  __lowercase             =  2

  __lowercase             =  []
  for scheduler_enum in KarrasDiffusionSchedulers:
   if scheduler_enum.name in skip_schedulers:
    # no sigma schedulers are not supported
    # no schedulers
    continue

   __lowercase             =  getattr(_lowerCAmelCase       ,				scheduler_enum.name			)
   __lowercase             =  scheduler_cls.from_config(pipe.scheduler.config			)
   __lowercase             =  pipe(**_lowerCAmelCase			)[0]
   outputs.append(_lowerCAmelCase			)

  assert check_same_shape(_lowerCAmelCase			)


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







 def  _a   (  self      :							Tuple			)		-> Tuple:
  """simple docstring"""







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







 def  _a   (  self      :							Optional[Any]			)		-> List[Any]:
  """simple docstring"""







  __lowercase             =  torch.manual_seed(33			)

  __lowercase             =  StableDiffusionPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4"""       ,				torch_dtype=torch.floataa			)
  pipe.to("""cuda"""			)

  __lowercase             =  StableDiffusionLatentUpscalePipeline.from_pretrained(
      """stabilityai/sd-x2-latent-upscaler"""       ,				torch_dtype=torch.floataa			)
  upscaler.to("""cuda"""			)

  __lowercase             =  '''a photo of an astronaut high resolution, unreal engine, ultra realistic'''

  __lowercase             =  pipe(_lowerCAmelCase       ,				generator=_lowerCAmelCase       ,				output_type="""latent"""			).images

  __lowercase             =  upscaler(
      prompt=_lowerCAmelCase       ,				image=_lowerCAmelCase       ,				num_inference_steps=20       ,				guidance_scale=0       ,				generator=_lowerCAmelCase       ,				output_type="""np"""       ,				).images[0]

  __lowercase             =  load_numpy(
      """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy"""			)
  assert np.abs((expected_image - image).mean()			) < 5e-2







 def  _a   (  self      :							Union[str, Any]			)		-> List[Any]:
  """simple docstring"""







  __lowercase             =  torch.manual_seed(33			)

  __lowercase             =  StableDiffusionLatentUpscalePipeline.from_pretrained(
      """stabilityai/sd-x2-latent-upscaler"""       ,				torch_dtype=torch.floataa			)
  upscaler.to("""cuda"""			)

  __lowercase             =  '''the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas'''

  __lowercase             =  load_image(
      """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png"""			)

  __lowercase             =  upscaler(
      prompt=_lowerCAmelCase       ,				image=_lowerCAmelCase       ,				num_inference_steps=20       ,				guidance_scale=0       ,				generator=_lowerCAmelCase       ,				output_type="""np"""       ,				).images[0]

  __lowercase             =  load_numpy(
      """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy"""			)
  assert np.abs((expected_image - image).max()			) < 5e-2









def 				snake_case  (     lowerCamelCase					):




	'''simple docstring'''







	__lowercase             =  """"""
	for ch in key:
		if ch == " " or ch not in key_no_dups and ch.isalpha():
			key_no_dups += ch
	return key_no_dups





def 				snake_case  (     lowerCamelCase					):




	'''simple docstring'''







	__lowercase             =  [chr(i + 65					) for i in range(26					)]
	# Remove duplicate characters from key
	__lowercase             =  remove_duplicates(key.upper()					)
	__lowercase             =  len(lowerCamelCase					)
	# First fill cipher with key characters
	__lowercase             =  {alphabet[i]: char for i, char in enumerate(lowerCamelCase					)}
	# Then map remaining characters in alphabet to
	# the alphabet from the beginning
	for i in range(len(lowerCamelCase					)				,  26					):
		__lowercase             =  alphabet[i - offset]
		# Ensure we are not mapping letters to letters previously mapped
		while char in key:
			offset -= 1
			__lowercase             =  alphabet[i - offset]
		__lowercase             =  char
	return cipher_alphabet





def 				snake_case  (     lowerCamelCase				,  lowerCamelCase					):




	'''simple docstring'''







	return "".join(cipher_map.get(lowerCamelCase				,  lowerCamelCase					) for ch in message.upper()					)





def 				snake_case  (     lowerCamelCase				,  lowerCamelCase					):




	'''simple docstring'''







	__lowercase             =  {v: k for k, v in cipher_map.items()}
	return "".join(rev_cipher_map.get(lowerCamelCase				,  lowerCamelCase					) for ch in message.upper()					)





def 				snake_case  (     ):




	'''simple docstring'''







	__lowercase             =  input("""Enter message to encode or decode: """					).strip()
	__lowercase             =  input("""Enter keyword: """					).strip()
	__lowercase             =  input("""Encipher or decipher? E/D:"""					).strip()[0].lower()
	try:
		__lowercase             =  {"""e""": encipher, """d""": decipher}[option]
	except KeyError:
		raise KeyError("""invalid input option"""					)
	__lowercase             =  create_cipher_map(lowerCamelCase					)
	print(func(lowerCamelCase				,  lowerCamelCase					)					)


if __name__ == "__main__":
						import doctest

						doctest.testmod()
						main()

