import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class 			SCREAMING_SNAKE_CASE__ (   __SCREAMING_SNAKE_CASE					,				__SCREAMING_SNAKE_CASE					,				unittest.TestCase  ):
       A				:     Dict					    =     StableDiffusionParadigmsPipeline
       A				:     str					    =     TEXT_TO_IMAGE_PARAMS
       A				:     Tuple					    =     TEXT_TO_IMAGE_BATCH_PARAMS
       A				:     Optional[int]					    =     TEXT_TO_IMAGE_IMAGE_PARAMS
       A				:     List[Any]					    =     TEXT_TO_IMAGE_IMAGE_PARAMS

       def        snake_case__  (				self :			List[str]       ):
           torch.manual_seed(0       )
           __snake_case    :							str             =  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       ,       attention_head_dim=(2, 4)       ,       use_linear_projection=_snake_case       ,       )
           __snake_case    :							List[Any]             =  DDIMParallelScheduler(
               beta_start=0.00085       ,       beta_end=0.012       ,       beta_schedule="""scaled_linear"""       ,       clip_sample=_snake_case       ,       set_alpha_to_one=_snake_case       ,       )
           torch.manual_seed(0       )
           __snake_case    :							str             =  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       ,       sample_size=1_28       ,       )
           torch.manual_seed(0       )
           __snake_case    :							Optional[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       ,       hidden_act="""gelu"""       ,       projection_dim=5_12       ,       )
           __snake_case    :							Tuple             =  CLIPTextModel(_snake_case       )
           __snake_case    :							Optional[Any]             =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

           __snake_case    :							Union[str, Any]             =  {
               "unet": unet,
               "scheduler": scheduler,
               "vae": vae,
               "text_encoder": text_encoder,
               "tokenizer": tokenizer,
               "safety_checker": None,
               "feature_extractor": None,
           }
           return components

       def        snake_case__  (				self :			Optional[Any]       ,       _lowerCAmelCase :			int       ,       _lowerCAmelCase :			Tuple=0       ):
           if str(_snake_case       ).startswith("""mps"""       ):
               __snake_case    :							List[Any]             =  torch.manual_seed(_snake_case       )
           else:
               __snake_case    :							List[str]             =  torch.Generator(device=_snake_case       ).manual_seed(_snake_case       )
           __snake_case    :							str             =  {
               "prompt": "a photograph of an astronaut riding a horse",
               "generator": generator,
               "num_inference_steps": 10,
               "guidance_scale": 6.0,
               "output_type": "numpy",
               "parallel": 3,
               "debug": True,
           }
           return inputs

       def        snake_case__  (				self :			Optional[int]       ):
           __snake_case    :							Any             =  "cpu"  # ensure determinism for the device-dependent torch.Generator
           __snake_case    :							Dict             =  self.get_dummy_components()
           __snake_case    :							Tuple             =  StableDiffusionParadigmsPipeline(**_snake_case       )
           __snake_case    :							Dict             =  sd_pipe.to(_snake_case       )
           sd_pipe.set_progress_bar_config(disable=_snake_case       )

           __snake_case    :							Optional[Any]             =  self.get_dummy_inputs(_snake_case       )
           __snake_case    :							List[str]             =  sd_pipe(**_snake_case       ).images
           __snake_case    :							str             =  image[0, -3:, -3:, -1]
           assert image.shape == (1, 64, 64, 3)

           __snake_case    :							Dict             =  np.array([0.4773, 0.5417, 0.4723, 0.4925, 0.5631, 0.4752, 0.5240, 0.4935, 0.5023]       )

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

       def        snake_case__  (				self :			List[str]       ):
           __snake_case    :							List[str]             =  "cpu"  # ensure determinism for the device-dependent torch.Generator
           __snake_case    :							Tuple             =  self.get_dummy_components()
           torch.manual_seed(0       )
           __snake_case    :							Union[str, Any]             =  DDPMParallelScheduler()
           torch.manual_seed(0       )
           __snake_case    :							Dict             =  StableDiffusionParadigmsPipeline(**_snake_case       )
           __snake_case    :							List[str]             =  sd_pipe.to(_snake_case       )
           sd_pipe.set_progress_bar_config(disable=_snake_case       )

           __snake_case    :							Union[str, Any]             =  self.get_dummy_inputs(_snake_case       )
           __snake_case    :							List[Any]             =  sd_pipe(**_snake_case       ).images
           __snake_case    :							str             =  image[0, -3:, -3:, -1]
           assert image.shape == (1, 64, 64, 3)

           __snake_case    :							int             =  np.array([0.3573, 0.4420, 0.4960, 0.4799, 0.3796, 0.3879, 0.4819, 0.4365, 0.4468]       )

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

       def        snake_case__  (				self :			Dict       ):
           super().test_inference_batch_consistent(batch_sizes=[1, 2]       )

       def        snake_case__  (				self :			List[str]       ):
           super().test_inference_batch_single_identical(batch_size=2       ,       expected_max_diff=3e-3       )





       def        snake_case__  (				self :			Union[str, Any]       ):
           __snake_case    :							Any             =  "cpu"  # ensure determinism for the device-dependent torch.Generator
           __snake_case    :							Optional[Any]             =  self.get_dummy_components()
           __snake_case    :							Tuple             =  StableDiffusionParadigmsPipeline(**_snake_case       )
           __snake_case    :							List[str]             =  sd_pipe.to(_snake_case       )
           sd_pipe.set_progress_bar_config(disable=_snake_case       )

           __snake_case    :							Tuple             =  self.get_dummy_inputs(_snake_case       )
           __snake_case    :							Any             =  "french fries"
           __snake_case    :							Union[str, Any]             =  sd_pipe(**_snake_case       ,       negative_prompt=_snake_case       )
           __snake_case    :							Optional[Any]             =  output.images
           __snake_case    :							Union[str, Any]             =  image[0, -3:, -3:, -1]

           assert image.shape == (1, 64, 64, 3)

           __snake_case    :							str             =  np.array([0.4771, 0.5420, 0.4683, 0.4918, 0.5636, 0.4725, 0.5230, 0.4923, 0.5015]       )

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







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

       def        snake_case__  (				self :			int       ):
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()

       def        snake_case__  (				self :			int       ,       _lowerCAmelCase :			List[str]=0       ):
           __snake_case    :							int             =  torch.Generator(device=_snake_case       ).manual_seed(_snake_case       )
           __snake_case    :							Tuple             =  {
               "prompt": "a photograph of an astronaut riding a horse",
               "generator": generator,
               "num_inference_steps": 10,
               "guidance_scale": 7.5,
               "output_type": "numpy",
               "parallel": 3,
               "debug": True,
           }
           return inputs





       def        snake_case__  (				self :			List[str]       ):
           __snake_case    :							int             =  "stabilityai/stable-diffusion-2-base"
           __snake_case    :							Any             =  DDIMParallelScheduler.from_pretrained(_snake_case       ,       subfolder="""scheduler"""       )
           __snake_case    :							int             =  StableDiffusionParadigmsPipeline.from_pretrained(_snake_case       ,       scheduler=_snake_case       ,       safety_checker=_snake_case       )
           pipe.to(_snake_case       )
           pipe.set_progress_bar_config(disable=_snake_case       )
           pipe.enable_attention_slicing()

           __snake_case    :							Optional[int]             =  self.get_inputs()
           __snake_case    :							int             =  pipe(**_snake_case       ).images
           __snake_case    :							Any             =  image[0, -3:, -3:, -1].flatten()

           assert image.shape == (1, 5_12, 5_12, 3)

           __snake_case    :							Union[str, Any]             =  np.array([0.9622, 0.9602, 0.9748, 0.9591, 0.9630, 0.9691, 0.9661, 0.9631, 0.9741]       )

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






'''simple docstring'''
def       a__    (  _SCREAMING_SNAKE_CASE					:    list     ,							_SCREAMING_SNAKE_CASE					:    int							)			-> int:


	"""simple docstring"""





	for index, item in enumerate(_SCREAMING_SNAKE_CASE							):
		if item == target:
			return index
	return -1


def       a__    (  _SCREAMING_SNAKE_CASE					:    list     ,							_SCREAMING_SNAKE_CASE					:    int     ,							_SCREAMING_SNAKE_CASE					:    int     ,							_SCREAMING_SNAKE_CASE					:    int							)			-> int:


	"""simple docstring"""





	if not (0 <= high < len(_SCREAMING_SNAKE_CASE							) and 0 <= low < len(_SCREAMING_SNAKE_CASE							)):
		raise Exception("Invalid upper or lower bound!"							)
	if high < low:
		return -1
	if sequence[low] == target:
		return low
	if sequence[high] == target:
		return high
	return rec_linear_search(_SCREAMING_SNAKE_CASE     ,							low + 1     ,							high - 1     ,							_SCREAMING_SNAKE_CASE							)


if __name__ == "__main__":
					_lowerCamelCase	 =			input("""Enter numbers separated by comma:\n""").strip()
					_lowerCamelCase	 =			[int(item.strip()) for item in user_input.split(""",""")]

					_lowerCamelCase	 =			int(input("""Enter a single number to be found in the list:\n""").strip())
					_lowerCamelCase	 =			linear_search(sequence, target)
					if result != -1:
										print(f"""linear_search({sequence}, {target}) = {result}""")
					else:
										print(f"""{target} was not found in {sequence}""")




