






import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowercase    				=   get_tests_dir("fixtures")




class 					UpperCamelCase_ (     unittest.TestCase							):



	'''simple docstring'''




	def 				_UpperCamelCase     (			self   )			->       Optional[int]:
								# A mock response for an HTTP head request to emulate server down
								snake_case_           =				mock.Mock()
								snake_case_           =				5_00
								snake_case_           =				{}
								snake_case_           =				HTTPError
								snake_case_           =				{}

								# Download this model to make sure it's in the cache.
								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'   )
								# Under the mock environment we get a 500 error when trying to reach the model.
								with mock.patch('requests.Session.request'			,     return_value=a   ) as mock_head:
															snake_case_           =				WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'   )
															# This check we did call the fake head request
															mock_head.assert_called()






	def 				_UpperCamelCase     (			self   )			->       Dict:
								# This test is for deprecated behavior and can be removed in v5
								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained(
								    'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'   )






@is_staging_test
class 					UpperCamelCase_ (     unittest.TestCase							):



	'''simple docstring'''




	@classmethod
	def 				_UpperCamelCase     (			cls   )			->       Union[str, Any]:
								snake_case_           =				TOKEN
								HfFolder.save_token(a   )




	@classmethod
	def 				_UpperCamelCase     (			cls   )			->       Any:
								try:
															delete_repo(token=cls._token			,     repo_id='test-feature-extractor'   )
								except HTTPError:
															pass

								try:
															delete_repo(token=cls._token			,     repo_id='valid_org/test-feature-extractor-org'   )
								except HTTPError:
															pass

								try:
															delete_repo(token=cls._token			,     repo_id='test-dynamic-feature-extractor'   )
								except HTTPError:
															pass




	def 				_UpperCamelCase     (			self   )			->       Optional[int]:
								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained(a   )
								feature_extractor.push_to_hub('test-feature-extractor'			,     use_auth_token=self._token   )

								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''   )
								for k, v in feature_extractor.__dict__.items():
															self.assertEqual(a			,     getattr(a			,     a   )   )

								# Reset repo
								delete_repo(token=self._token			,     repo_id='test-feature-extractor'   )

								# Push to hub via save_pretrained
								with tempfile.TemporaryDirectory() as tmp_dir:
															feature_extractor.save_pretrained(
															    a			,     repo_id='test-feature-extractor'			,     push_to_hub=a			,     use_auth_token=self._token   )

								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''   )
								for k, v in feature_extractor.__dict__.items():
															self.assertEqual(a			,     getattr(a			,     a   )   )




	def 				_UpperCamelCase     (			self   )			->       str:
								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained(a   )
								feature_extractor.push_to_hub('valid_org/test-feature-extractor'			,     use_auth_token=self._token   )

								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor'   )
								for k, v in feature_extractor.__dict__.items():
															self.assertEqual(a			,     getattr(a			,     a   )   )

								# Reset repo
								delete_repo(token=self._token			,     repo_id='valid_org/test-feature-extractor'   )

								# Push to hub via save_pretrained
								with tempfile.TemporaryDirectory() as tmp_dir:
															feature_extractor.save_pretrained(
															    a			,     repo_id='valid_org/test-feature-extractor-org'			,     push_to_hub=a			,     use_auth_token=self._token   )

								snake_case_           =				WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor-org'   )
								for k, v in feature_extractor.__dict__.items():
															self.assertEqual(a			,     getattr(a			,     a   )   )






	def 				_UpperCamelCase     (			self   )			->       List[Any]:
								CustomFeatureExtractor.register_for_auto_class()
								snake_case_           =				CustomFeatureExtractor.from_pretrained(a   )

								feature_extractor.push_to_hub('test-dynamic-feature-extractor'			,     use_auth_token=self._token   )

								# This has added the proper auto_map field to the config
								self.assertDictEqual(
								    feature_extractor.auto_map			,     {'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor'}			,     )

								snake_case_           =				AutoFeatureExtractor.from_pretrained(
								    F'''{USER}/test-dynamic-feature-extractor'''			,     trust_remote_code=a   )
								# Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
								self.assertEqual(new_feature_extractor.__class__.__name__			,     'CustomFeatureExtractor'   )













from typing import Any, Callable, Dict, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker


lowercase    				=   "CompVis/stable-diffusion-v1-1"
lowercase    				=   "CompVis/stable-diffusion-v1-2"
lowercase    				=   "CompVis/stable-diffusion-v1-3"
lowercase    				=   "CompVis/stable-diffusion-v1-4"




class 					UpperCamelCase_ (     snake_case_							):



	'''simple docstring'''




	def __init__(			self			,     a			,     a			,     a			,     a			,     a			,     a			,     a			,     a = True			,     )			->       int:
								super()._init_()

								snake_case_           =				StableDiffusionPipeline.from_pretrained(a   )
								snake_case_           =				StableDiffusionPipeline.from_pretrained(a   )
								snake_case_           =				StableDiffusionPipeline.from_pretrained(a   )
								snake_case_           =				StableDiffusionPipeline(
								    vae=a			,     text_encoder=a			,     tokenizer=a			,     unet=a			,     scheduler=a			,     safety_checker=a			,     feature_extractor=a			,     requires_safety_checker=a			,     )

								self.register_modules(pipelinea=self.pipea			,     pipelinea=self.pipea			,     pipelinea=self.pipea			,     pipelinea=self.pipea   )




	@property
	def 				_UpperCamelCase     (			self   )			->       Dict[str, Any]:
								return {k: getattr(self			,     a   ) for k in self.config.keys() if not k.startswith('_'   )}




	def 				_UpperCamelCase     (			self			,     a = "auto"   )			->       Dict:
								if slice_size == "auto":
															# half the attention head size is usually a good trade-off between
															# speed and memory
															snake_case_           =				self.unet.config.attention_head_dim // 2
								self.unet.set_attention_slice(a   )




	def 				_UpperCamelCase     (			self   )			->       List[Any]:
								self.enable_attention_slicing(a   )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Optional[Any]:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       int:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Optional[int]:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Any:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )






	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Optional[Any]:
								snake_case_           =				'cuda' if torch.cuda.is_available() else 'cpu'
								self.to(a   )

								# Checks if the height and width are divisible by 8 or not
								if height % 8 != 0 or width % 8 != 0:
															raise ValueError(F'''`height` and `width` must be divisible by 8 but are {height} and {width}.'''   )

								# Get first result from Stable Diffusion Checkpoint v1.1
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get first result from Stable Diffusion Checkpoint v1.2
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get first result from Stable Diffusion Checkpoint v1.3
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get first result from Stable Diffusion Checkpoint v1.4
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get all result images into a single list and pass it via StableDiffusionPipelineOutput for final result
								return StableDiffusionPipelineOutput([resa[0], resa[0], resa[0], resa[0]]   )





