
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 							UpperCAmelCase	(   __lowerCamelCase						, __lowerCamelCase						, unittest.TestCase					):
     a__: str			  =				StableDiffusionParadigmsPipeline
     a__: Optional[int]			  =				TEXT_TO_IMAGE_PARAMS
     a__: Dict			  =				TEXT_TO_IMAGE_BATCH_PARAMS
     a__: Union[str, Any]			  =				TEXT_TO_IMAGE_IMAGE_PARAMS
     a__: Tuple			  =				TEXT_TO_IMAGE_IMAGE_PARAMS

     def        _lowerCAmelCase	(			self      :					Union[str, Any]					):
           torch.manual_seed(0					)
           lowercase       :    Any											=  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=lowerCAmelCase ,      )
           lowercase       :    Optional[int]											=  DDIMParallelScheduler(
               beta_start=0.0_00_85 ,      beta_end=0.0_12 ,      beta_schedule='''scaled_linear''' ,      clip_sample=lowerCAmelCase ,      set_alpha_to_one=lowerCAmelCase ,      )
           torch.manual_seed(0					)
           lowercase       :    Optional[int]											=  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=128 ,      )
           torch.manual_seed(0					)
           lowercase       :    int											=  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='''gelu''' ,      projection_dim=512 ,      )
           lowercase       :    List[Any]											=  CLIPTextModel(lowerCAmelCase					)
           lowercase       :    Tuple											=  CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

           lowercase       :    int											=  {
               '''unet''': unet,
               '''scheduler''': scheduler,
               '''vae''': vae,
               '''text_encoder''': text_encoder,
               '''tokenizer''': tokenizer,
               '''safety_checker''': None,
               '''feature_extractor''': None,
           }
           return components

     def        _lowerCAmelCase	(			self      :					Union[str, Any] ,      lowerCAmelCase      :					List[Any] ,      lowerCAmelCase      :					List[Any]=0					):
           if str(lowerCAmelCase					).startswith('''mps'''					):
                 lowercase       :    Tuple											=  torch.manual_seed(lowerCAmelCase					)
           else:
                 lowercase       :    List[str]											=  torch.Generator(device=lowerCAmelCase					).manual_seed(lowerCAmelCase					)
           lowercase       :    Dict											=  {
               '''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        _lowerCAmelCase	(			self      :					Any					):
           lowercase       :    str											=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
           lowercase       :    int											=  self.get_dummy_components()
           lowercase       :    Tuple											=  StableDiffusionParadigmsPipeline(**lowerCAmelCase					)
           lowercase       :    List[Any]											=  sd_pipe.to(lowerCAmelCase					)
           sd_pipe.set_progress_bar_config(disable=lowerCAmelCase					)

           lowercase       :    Optional[int]											=  self.get_dummy_inputs(lowerCAmelCase					)
           lowercase       :    int											=  sd_pipe(**lowerCAmelCase					).images
           lowercase       :    Dict											=  image[0, -3:, -3:, -1]
           assert image.shape == (1, 64, 64, 3)

           lowercase       :    Dict											=  np.array([0.47_73, 0.54_17, 0.47_23, 0.49_25, 0.56_31, 0.47_52, 0.52_40, 0.49_35, 0.50_23]					)

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

     def        _lowerCAmelCase	(			self      :					List[Any]					):
           lowercase       :    Tuple											=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
           lowercase       :    List[str]											=  self.get_dummy_components()
           torch.manual_seed(0					)
           lowercase       :    List[Any]											=  DDPMParallelScheduler()
           torch.manual_seed(0					)
           lowercase       :    Tuple											=  StableDiffusionParadigmsPipeline(**lowerCAmelCase					)
           lowercase       :    Dict											=  sd_pipe.to(lowerCAmelCase					)
           sd_pipe.set_progress_bar_config(disable=lowerCAmelCase					)

           lowercase       :    str											=  self.get_dummy_inputs(lowerCAmelCase					)
           lowercase       :    Tuple											=  sd_pipe(**lowerCAmelCase					).images
           lowercase       :    Optional[int]											=  image[0, -3:, -3:, -1]
           assert image.shape == (1, 64, 64, 3)

           lowercase       :    str											=  np.array([0.35_73, 0.44_20, 0.49_60, 0.47_99, 0.37_96, 0.38_79, 0.48_19, 0.43_65, 0.44_68]					)

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

     def        _lowerCAmelCase	(			self      :					Optional[int]					):
           super().test_inference_batch_consistent(batch_sizes=[1, 2]					)

     def        _lowerCAmelCase	(			self      :					int					):
           super().test_inference_batch_single_identical(batch_size=2 ,      expected_max_diff=3E-3					)




     def        _lowerCAmelCase	(			self      :					Tuple					):
           lowercase       :    List[Any]											=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
           lowercase       :    Optional[Any]											=  self.get_dummy_components()
           lowercase       :    str											=  StableDiffusionParadigmsPipeline(**lowerCAmelCase					)
           lowercase       :    List[Any]											=  sd_pipe.to(lowerCAmelCase					)
           sd_pipe.set_progress_bar_config(disable=lowerCAmelCase					)

           lowercase       :    Union[str, Any]											=  self.get_dummy_inputs(lowerCAmelCase					)
           lowercase       :    Union[str, Any]											=  '''french fries'''
           lowercase       :    Tuple											=  sd_pipe(**lowerCAmelCase ,      negative_prompt=lowerCAmelCase					)
           lowercase       :    Union[str, Any]											=  output.images
           lowercase       :    List[str]											=  image[0, -3:, -3:, -1]

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

           lowercase       :    Optional[int]											=  np.array([0.47_71, 0.54_20, 0.46_83, 0.49_18, 0.56_36, 0.47_25, 0.52_30, 0.49_23, 0.50_15]					)

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







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

     def        _lowerCAmelCase	(			self      :					str					):
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()

     def        _lowerCAmelCase	(			self      :					Dict ,      lowerCAmelCase      :					Dict=0					):
           lowercase       :    Optional[Any]											=  torch.Generator(device=lowerCAmelCase					).manual_seed(lowerCAmelCase					)
           lowercase       :    List[Any]											=  {
               '''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        _lowerCAmelCase	(			self      :					Tuple					):
           lowercase       :    Tuple											=  '''stabilityai/stable-diffusion-2-base'''
           lowercase       :    Optional[int]											=  DDIMParallelScheduler.from_pretrained(lowerCAmelCase ,      subfolder='''scheduler'''					)
           lowercase       :    Optional[Any]											=  StableDiffusionParadigmsPipeline.from_pretrained(lowerCAmelCase ,      scheduler=lowerCAmelCase ,      safety_checker=lowerCAmelCase					)
           pipe.to(lowerCAmelCase					)
           pipe.set_progress_bar_config(disable=lowerCAmelCase					)
           pipe.enable_attention_slicing()

           lowercase       :    List[str]											=  self.get_inputs()
           lowercase       :    int											=  pipe(**lowerCAmelCase					).images
           lowercase       :    int											=  image[0, -3:, -3:, -1].flatten()

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

           lowercase       :    Optional[int]											=  np.array([0.96_22, 0.96_02, 0.97_48, 0.95_91, 0.96_30, 0.96_91, 0.96_61, 0.96_31, 0.97_41]					)

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




from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


snake_case__  							=							logging.get_logger(__name__)







class 							UpperCAmelCase	(   __lowerCamelCase					):
					a__: str			  =				["""input_values""", """attention_mask"""]

					def __init__(			self      :					Union[str, Any] ,      lowerCAmelCase      :					Tuple=1 ,      lowerCAmelCase      :					Dict=1_6000 ,      lowerCAmelCase      :					Optional[Any]=128 ,      lowerCAmelCase      :					Dict=1024 ,      lowerCAmelCase      :					List[Any]=0.0 ,      lowerCAmelCase      :					Any=True ,      lowerCAmelCase      :					Tuple=-4.2_67_73_93 ,      lowerCAmelCase      :					Dict=4.5_68_99_74 ,      lowerCAmelCase      :					int=False ,      **lowerCAmelCase      :					Optional[Any] ,      ):
											super().__init__(feature_size=lowerCAmelCase ,      sampling_rate=lowerCAmelCase ,      padding_value=lowerCAmelCase ,      **lowerCAmelCase					)
											lowercase       :    Optional[int]											=  num_mel_bins
											lowercase       :    List[Any]											=  max_length
											lowercase       :    Any											=  do_normalize
											lowercase       :    Any											=  mean
											lowercase       :    Union[str, Any]											=  std
											lowercase       :    int											=  return_attention_mask

					def        _lowerCAmelCase	(			self      :					int ,      lowerCAmelCase      :					np.ndarray ,      lowerCAmelCase      :					int ,      ):
											lowercase       :    List[Any]											=  torch.from_numpy(lowerCAmelCase					).unsqueeze(0					)
											lowercase       :    List[str]											=  ta_kaldi.fbank(
											    lowerCAmelCase ,      htk_compat=lowerCAmelCase ,      sample_frequency=self.sampling_rate ,      use_energy=lowerCAmelCase ,      window_type='''hanning''' ,      num_mel_bins=self.num_mel_bins ,      dither=0.0 ,      frame_shift=10 ,      )

											lowercase       :    Tuple											=  fbank.shape[0]
											lowercase       :    Optional[Any]											=  max_length - n_frames

											# pad or truncate, depending on difference
											if difference > 0:
																	lowercase       :    int											=  torch.nn.ZeroPadad((0, 0, 0, difference)					)
																	lowercase       :    str											=  pad_module(lowerCAmelCase					)
											elif difference < 0:
																	lowercase       :    Optional[int]											=  fbank[0:max_length, :]

											lowercase       :    List[str]											=  fbank.numpy()

											return fbank

					def        _lowerCAmelCase	(			self      :					Tuple ,      lowerCAmelCase      :					np.ndarray					):
											return (input_values - (self.mean)) / (self.std * 2)




					def __call__(			self      :					Any ,      lowerCAmelCase      :					Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]] ,      lowerCAmelCase      :					Optional[int] = None ,      lowerCAmelCase      :					Optional[Union[str, TensorType]] = None ,      **lowerCAmelCase      :					Tuple ,      ):

											if sampling_rate is not None:
																	if sampling_rate != self.sampling_rate:
																							raise ValueError(
																							    f'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
																							    f''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
																							    f''' {self.sampling_rate} and not {sampling_rate}.'''					)
											else:
																	logger.warning(
																	    '''It is strongly recommended to pass the `sampling_rate` argument to this function. '''
																	    '''Failing to do so can result in silent errors that might be hard to debug.'''					)

											lowercase       :    Optional[Any]											=  isinstance(lowerCAmelCase ,      np.ndarray					) and len(raw_speech.shape					) > 1
											if is_batched_numpy and len(raw_speech.shape					) > 2:
																	raise ValueError(f'''Only mono-channel audio is supported for input to {self}'''					)
											lowercase       :    Tuple											=  is_batched_numpy or (
											    isinstance(lowerCAmelCase ,      (list, tuple)					) and (isinstance(raw_speech[0] ,      (np.ndarray, tuple, list)					))
											)

											if is_batched:
																	lowercase       :    Optional[int]											=  [np.asarray(lowerCAmelCase ,      dtype=np.floataa					) for speech in raw_speech]
											elif not is_batched and not isinstance(lowerCAmelCase ,      np.ndarray					):
																	lowercase       :    Dict											=  np.asarray(lowerCAmelCase ,      dtype=np.floataa					)
											elif isinstance(lowerCAmelCase ,      np.ndarray					) and raw_speech.dtype is np.dtype(np.floataa					):
																	lowercase       :    int											=  raw_speech.astype(np.floataa					)

											# always return batch
											if not is_batched:
																	lowercase       :    Dict											=  [raw_speech]

											# extract fbank features and pad/truncate to max_length
											lowercase       :    Dict											=  [self._extract_fbank_features(lowerCAmelCase ,      max_length=self.max_length					) for waveform in raw_speech]

											# convert into BatchFeature
											lowercase       :    Any											=  BatchFeature({'''input_values''': features}					)

											# make sure list is in array format
											lowercase       :    str											=  padded_inputs.get('''input_values'''					)
											if isinstance(input_values[0] ,      lowerCAmelCase					):
																	lowercase       :    Any											=  [np.asarray(lowerCAmelCase ,      dtype=np.floataa					) for feature in input_values]

											# normalization
											if self.do_normalize:
																	lowercase       :    List[str]											=  [self.normalize(lowerCAmelCase					) for feature in input_values]

											if return_tensors is not None:
																	lowercase       :    str											=  padded_inputs.convert_to_tensors(lowerCAmelCase					)

											return padded_inputs


