

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


A											=     get_tests_dir('fixtures')


class SCREAMING_SNAKE_CASE  (		unittest.TestCase							):






			"""simple docstring"""




			def 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				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=__UpperCamelCase							) 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 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				snake_case_      		= WavaVecaFeatureExtractor.from_pretrained(
				    'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'							)







@is_staging_test
class SCREAMING_SNAKE_CASE  (		unittest.TestCase							):






			"""simple docstring"""




			@classmethod
			def 			__lowerCAmelCase (    cls							):

				"""simple docstring"""


				snake_case_      		= TOKEN
				HfFolder.save_token(__UpperCamelCase							)


			@classmethod
			def 			__lowerCAmelCase (    cls							):

				"""simple docstring"""


				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 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				snake_case_      		= WavaVecaFeatureExtractor.from_pretrained(__UpperCamelCase							)
				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(__UpperCamelCase ,				getattr(__UpperCamelCase ,				__UpperCamelCase							)							)

				# 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(
					    __UpperCamelCase ,				repo_id='test-feature-extractor' ,				push_to_hub=__UpperCamelCase ,				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(__UpperCamelCase ,				getattr(__UpperCamelCase ,				__UpperCamelCase							)							)


			def 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				snake_case_      		= WavaVecaFeatureExtractor.from_pretrained(__UpperCamelCase							)
				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(__UpperCamelCase ,				getattr(__UpperCamelCase ,				__UpperCamelCase							)							)

				# 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(
					    __UpperCamelCase ,				repo_id='valid_org/test-feature-extractor-org' ,				push_to_hub=__UpperCamelCase ,				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(__UpperCamelCase ,				getattr(__UpperCamelCase ,				__UpperCamelCase							)							)







			def 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				CustomFeatureExtractor.register_for_auto_class()
				snake_case_      		= CustomFeatureExtractor.from_pretrained(__UpperCamelCase							)

				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=__UpperCamelCase							)
				# 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'							)



import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


A											=     argparse.ArgumentParser('Stable Diffusion script with intel optimization', add_help=False)
parser.add_argument('--dpm', action='store_true', help='Enable DPMSolver or not')
parser.add_argument('--steps', default=None, type=int, help='Num inference steps')
A											=     parser.parse_args()


A											=     'cpu'
A											=     'a lovely <dicoo> in red dress and hat, in the snowly and brightly night, with many brighly buildings'

A											=     'path-to-your-trained-model'
A											=     StableDiffusionPipeline.from_pretrained(model_id)
if args.dpm:
		A											=     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
A											=     pipe.to(device)

# to channels last
A											=     pipe.unet.to(memory_format=torch.channels_last)
A											=     pipe.vae.to(memory_format=torch.channels_last)
A											=     pipe.text_encoder.to(memory_format=torch.channels_last)
if pipe.requires_safety_checker:
		A											=     pipe.safety_checker.to(memory_format=torch.channels_last)

# optimize with ipex
A											=     torch.randn(2, 4, 64, 64)
A											=     torch.rand(1) * 999
A											=     torch.randn(2, 77, 768)
A											=     (sample, timestep, encoder_hidden_status)
try:
		A											=     ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True, sample_input=input_example)
except Exception:
		A											=     ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True)
A											=     ipex.optimize(pipe.vae.eval(), dtype=torch.bfloataa, inplace=True)
A											=     ipex.optimize(pipe.text_encoder.eval(), dtype=torch.bfloataa, inplace=True)
if pipe.requires_safety_checker:
		A											=     ipex.optimize(pipe.safety_checker.eval(), dtype=torch.bfloataa, inplace=True)

# compute
A											=     666
A											=     torch.Generator(device).manual_seed(seed)
A											=     {'generator': generator}
if args.steps is not None:
		A											=     args.steps

with torch.cpu.amp.autocast(enabled=True, dtype=torch.bfloataa):
		A											=     pipe(prompt, **generate_kwargs).images[0]

# save image
image.save('generated.png')
