




'''simple docstring'''




import itertools
import random
import unittest

import numpy as np

from transformers import is_speech_available
from transformers.testing_utils import require_torch, require_torchaudio

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
						from transformers import SpeechaTextFeatureExtractor

_lowercase		:				List[str]	       =		random.Random()




def 							lowerCamelCase			(   UpperCAmelCase__ :      List[Any]   ,   UpperCAmelCase__ :      Optional[int]=1.0   ,   UpperCAmelCase__ :      Optional[Any]=None   ,   UpperCAmelCase__ :      Optional[Any]=None				)    ->   int:






							if rng is None:
														lowercase_      :    Optional[Any]       		= global_rng

							lowercase_      :    List[str]       		= []
							for batch_idx in range(shape[0]				):
														values.append([]				)
														for _ in range(shape[1]				):
																					values[-1].append(rng.random() * scale				)

							return values





@require_torch
@require_torchaudio
class 					__magic_name__		(					unittest.TestCase):







			def __init__(			self       :						Optional[Any]       ,    lowercase_       :						List[Any]       ,    lowercase_       :						Dict=7       ,    lowercase_       :						Dict=400       ,    lowercase_       :						Any=2000       ,    lowercase_       :						List[Any]=24       ,    lowercase_       :						Union[str, Any]=24       ,    lowercase_       :						str=0.0       ,    lowercase_       :						Dict=16000       ,    lowercase_       :						Optional[int]=True       ,    lowercase_       :						Dict=True       ,    ):
										lowercase_      :    Union[str, Any]       		= parent
										lowercase_      :    int       		= batch_size
										lowercase_      :    int       		= min_seq_length
										lowercase_      :    str       		= max_seq_length
										lowercase_      :    Union[str, Any]       		= (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
										lowercase_      :    List[Any]       		= feature_size
										lowercase_      :    List[Any]       		= num_mel_bins
										lowercase_      :    Optional[int]       		= padding_value
										lowercase_      :    Union[str, Any]       		= sampling_rate
										lowercase_      :    str       		= return_attention_mask
										lowercase_      :    Union[str, Any]       		= do_normalize







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Any					):
										return {
										    "feature_size": self.feature_size,
										    "num_mel_bins": self.num_mel_bins,
										    "padding_value": self.padding_value,
										    "sampling_rate": self.sampling_rate,
										    "return_attention_mask": self.return_attention_mask,
										    "do_normalize": self.do_normalize,
										}







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Union[str, Any]       ,    lowercase_       :						Tuple=False       ,    lowercase_       :						Optional[int]=False					):
										def _flatten(lowercase_       :						Optional[int]					):
																	return list(itertools.chain(*lowercase_					)					)

										if equal_length:
																	lowercase_      :    Any       		= [floats_list((self.max_seq_length, self.feature_size)					) for _ in range(self.batch_size					)]
										else:
																	# make sure that inputs increase in size
																	lowercase_      :    List[str]       		= [
																	    floats_list((x, self.feature_size)					)
																	    for x in range(self.min_seq_length       ,    self.max_seq_length       ,    self.seq_length_diff					)
																	]
										if numpify:
																	lowercase_      :    List[str]       		= [np.asarray(lowercase_					) for x in speech_inputs]
										return speech_inputs





@require_torch
@require_torchaudio
class 					__magic_name__		(					lowercase_,  unittest.TestCase):
			UpperCamelCase__       = SpeechaTextFeatureExtractor if is_speech_available() else None







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[Any]					):
										lowercase_      :    Any       		= SpeechaTextFeatureExtractionTester(self					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[Any]       ,    lowercase_       :						Dict					):
										self.assertTrue(np.all(np.mean(lowercase_       ,    axis=0					) < 1E-3					)					)
										self.assertTrue(np.all(np.abs(np.var(lowercase_       ,    axis=0					) - 1					) < 1E-3					)					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[int]					):
										# Tests that all call wrap to encode_plus and batch_encode_plus
										lowercase_      :    str       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
										# create three inputs of length 800, 1000, and 1200
										lowercase_      :    Optional[Any]       		= [floats_list((1, x)					)[0] for x in range(800       ,    1400       ,    200					)]
										lowercase_      :    Dict       		= [np.asarray(lowercase_					) for speech_input in speech_inputs]

										# Test feature size
										lowercase_      :    Any       		= feature_extractor(lowercase_       ,    padding=lowercase_       ,    return_tensors="""np"""					).input_features
										self.assertTrue(input_features.ndim == 3					)
										self.assertTrue(input_features.shape[-1] == feature_extractor.feature_size					)

										# Test not batched input
										lowercase_      :    str       		= feature_extractor(speech_inputs[0]       ,    return_tensors="""np"""					).input_features
										lowercase_      :    Any       		= feature_extractor(np_speech_inputs[0]       ,    return_tensors="""np"""					).input_features
										self.assertTrue(np.allclose(lowercase_       ,    lowercase_       ,    atol=1E-3					)					)

										# Test batched
										lowercase_      :    int       		= feature_extractor(lowercase_       ,    return_tensors="""np"""					).input_features
										lowercase_      :    Optional[Any]       		= feature_extractor(lowercase_       ,    return_tensors="""np"""					).input_features
										for enc_seq_a, enc_seq_a in zip(lowercase_       ,    lowercase_					):
																	self.assertTrue(np.allclose(lowercase_       ,    lowercase_       ,    atol=1E-3					)					)

										# Test 2-D numpy arrays are batched.
										lowercase_      :    Optional[Any]       		= [floats_list((1, x)					)[0] for x in (800, 800, 800)]
										lowercase_      :    int       		= np.asarray(lowercase_					)
										lowercase_      :    Tuple       		= feature_extractor(lowercase_       ,    return_tensors="""np"""					).input_features
										lowercase_      :    int       		= feature_extractor(lowercase_       ,    return_tensors="""np"""					).input_features
										for enc_seq_a, enc_seq_a in zip(lowercase_       ,    lowercase_					):
																	self.assertTrue(np.allclose(lowercase_       ,    lowercase_       ,    atol=1E-3					)					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Dict					):
										lowercase_      :    Dict       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
										lowercase_      :    int       		= [floats_list((1, x)					)[0] for x in range(800       ,    1400       ,    200					)]

										lowercase_      :    List[str]       		= ["""longest""", """max_length""", """do_not_pad"""]
										lowercase_      :    List[Any]       		= [None, 16, None]
										for max_length, padding in zip(lowercase_       ,    lowercase_					):
																	lowercase_      :    Any       		= feature_extractor(
																	    lowercase_       ,    padding=lowercase_       ,    max_length=lowercase_       ,    return_attention_mask=lowercase_					)
																	lowercase_      :    Any       		= inputs.input_features
																	lowercase_      :    str       		= inputs.attention_mask
																	lowercase_      :    Any       		= [np.sum(lowercase_					) for x in attention_mask]

																	self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]					)
																	self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]					)
																	self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Union[str, Any]					):
										lowercase_      :    Any       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
										lowercase_      :    Union[str, Any]       		= [floats_list((1, x)					)[0] for x in range(800       ,    1400       ,    200					)]

										lowercase_      :    Dict       		= ["""longest""", """max_length""", """do_not_pad"""]
										lowercase_      :    Optional[Any]       		= [None, 16, None]
										for max_length, padding in zip(lowercase_       ,    lowercase_					):
																	lowercase_      :    Tuple       		= feature_extractor(
																	    lowercase_       ,    max_length=lowercase_       ,    padding=lowercase_       ,    return_tensors="""np"""       ,    return_attention_mask=lowercase_					)
																	lowercase_      :    Optional[Any]       		= inputs.input_features
																	lowercase_      :    Optional[Any]       		= inputs.attention_mask
																	lowercase_      :    Union[str, Any]       		= [np.sum(lowercase_					) for x in attention_mask]

																	self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]					)
																	self.assertTrue(input_features[0][fbank_feat_lengths[0] :].sum() < 1E-6					)
																	self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]					)
																	self.assertTrue(input_features[0][fbank_feat_lengths[1] :].sum() < 1E-6					)
																	self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						str					):
										lowercase_      :    Dict       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
										lowercase_      :    Optional[Any]       		= [floats_list((1, x)					)[0] for x in range(800       ,    1400       ,    200					)]
										lowercase_      :    Any       		= feature_extractor(
										    lowercase_       ,    padding="""max_length"""       ,    max_length=4       ,    truncation=lowercase_       ,    return_tensors="""np"""       ,    return_attention_mask=lowercase_       ,    )
										lowercase_      :    Dict       		= inputs.input_features
										lowercase_      :    Union[str, Any]       		= inputs.attention_mask
										lowercase_      :    Union[str, Any]       		= np.sum(attention_mask == 1       ,    axis=1					)

										self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]					)
										self._check_zero_mean_unit_variance(input_features[1]					)
										self._check_zero_mean_unit_variance(input_features[2]					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Dict					):
										lowercase_      :    Dict       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
										lowercase_      :    Tuple       		= [floats_list((1, x)					)[0] for x in range(800       ,    1400       ,    200					)]
										lowercase_      :    Optional[int]       		= feature_extractor(
										    lowercase_       ,    padding="""longest"""       ,    max_length=4       ,    truncation=lowercase_       ,    return_tensors="""np"""       ,    return_attention_mask=lowercase_       ,    )
										lowercase_      :    Dict       		= inputs.input_features
										lowercase_      :    Tuple       		= inputs.attention_mask
										lowercase_      :    str       		= np.sum(attention_mask == 1       ,    axis=1					)

										self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]					)
										self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]					)
										self._check_zero_mean_unit_variance(input_features[2]					)

										# make sure that if max_length < longest -> then pad to max_length
										self.assertEqual(input_features.shape       ,    (3, 4, 24)					)

										lowercase_      :    Tuple       		= [floats_list((1, x)					)[0] for x in range(800       ,    1400       ,    200					)]
										lowercase_      :    Union[str, Any]       		= feature_extractor(
										    lowercase_       ,    padding="""longest"""       ,    max_length=16       ,    truncation=lowercase_       ,    return_tensors="""np"""       ,    return_attention_mask=lowercase_       ,    )
										lowercase_      :    List[Any]       		= inputs.input_features
										lowercase_      :    List[Any]       		= inputs.attention_mask
										lowercase_      :    Union[str, Any]       		= np.sum(attention_mask == 1       ,    axis=1					)

										self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]					)
										self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]					)
										self._check_zero_mean_unit_variance(input_features[2]					)

										# make sure that if max_length < longest -> then pad to max_length
										self.assertEqual(input_features.shape       ,    (3, 6, 24)					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Tuple					):
										import torch

										lowercase_      :    Union[str, Any]       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
										lowercase_      :    int       		= np.random.rand(100       ,    32					).astype(np.floataa					)
										lowercase_      :    Optional[int]       		= np_speech_inputs.tolist()

										for inputs in [py_speech_inputs, np_speech_inputs]:
																	lowercase_      :    List[Any]       		= feature_extractor.pad([{"""input_features""": inputs}]       ,    return_tensors="""np"""					)
																	self.assertTrue(np_processed.input_features.dtype == np.floataa					)
																	lowercase_      :    Union[str, Any]       		= feature_extractor.pad([{"""input_features""": inputs}]       ,    return_tensors="""pt"""					)
																	self.assertTrue(pt_processed.input_features.dtype == torch.floataa					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						List[Any]       ,    lowercase_       :						List[str]					):
										from datasets import load_dataset

										lowercase_      :    Dict       		= load_dataset("""hf-internal-testing/librispeech_asr_dummy"""       ,    """clean"""       ,    split="""validation"""					)
										# automatic decoding with librispeech
										lowercase_      :    Optional[int]       		= ds.sort("""id"""					).select(range(lowercase_					)					)[:num_samples]["""audio"""]

										return [x["array"] for x in speech_samples]







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Union[str, Any]					):
										# fmt: off
										lowercase_      :    Optional[int]       		= np.array([
										    -1.57_45, -1.77_13, -1.70_20, -1.60_69, -1.22_50, -1.11_05, -0.90_72, -0.82_41,
										    -1.23_10, -0.80_98, -0.33_20, -0.41_01, -0.79_85, -0.49_96, -0.82_13, -0.91_28,
										    -1.04_20, -1.12_86, -1.04_40, -0.79_99, -0.84_05, -1.22_75, -1.54_43, -1.46_25,
										]					)
										# fmt: on

										lowercase_      :    List[str]       		= self._load_datasamples(1					)
										lowercase_      :    List[Any]       		= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
										lowercase_      :    Union[str, Any]       		= feature_extractor(lowercase_       ,    return_tensors="""pt"""					).input_features
										self.assertEquals(input_features.shape       ,    (1, 584, 24)					)
										self.assertTrue(np.allclose(input_features[0, 0, :30]       ,    lowercase_       ,    atol=1E-4					)					)






'''simple docstring'''




import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


_lowercase		:				Optional[Any]	       =		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")
_lowercase		:				Dict	       =		parser.parse_args()


_lowercase		:				Dict	       =		"cpu"
_lowercase		:				str	       =		"a lovely <dicoo> in red dress and hat, in the snowly and brightly night, with many brighly buildings"

_lowercase		:				Any	       =		"path-to-your-trained-model"
_lowercase		:				str	       =		StableDiffusionPipeline.from_pretrained(model_id)
if args.dpm:
						_lowercase		:				Optional[int]	       =		DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
_lowercase		:				Any	       =		pipe.to(device)

# to channels last
_lowercase		:				Union[str, Any]	       =		pipe.unet.to(memory_format=torch.channels_last)
_lowercase		:				List[Any]	       =		pipe.vae.to(memory_format=torch.channels_last)
_lowercase		:				Union[str, Any]	       =		pipe.text_encoder.to(memory_format=torch.channels_last)
if pipe.requires_safety_checker:
						_lowercase		:				Tuple	       =		pipe.safety_checker.to(memory_format=torch.channels_last)

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

# compute
_lowercase		:				int	       =		666
_lowercase		:				Any	       =		torch.Generator(device).manual_seed(seed)
_lowercase		:				int	       =		{"generator": generator}
if args.steps is not None:
						_lowercase		:				Optional[int]	       =		args.steps

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

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





