
"""simple docstring"""



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 PaddingStrategy, TensorType, logging


UpperCamelCase__          =	logging.get_logger(__name__)





class 				a			(					lowercase      ):
      UpperCamelCase							:   str           =     ["""input_features""", """attention_mask"""]



      def __init__(   self     ,	UpperCamelCase_=80     ,	UpperCamelCase_=16_000     ,	UpperCamelCase_=80     ,	UpperCamelCase_=0.0     ,	UpperCamelCase_=True     ,	UpperCamelCase_=True     ,	UpperCamelCase_=True     ,	**UpperCamelCase_     ,	):
          super().__init__(feature_size=UpperCamelCase_     ,	sampling_rate=UpperCamelCase_     ,	padding_value=UpperCamelCase_     ,	**UpperCamelCase_      )
          UpperCAmelCase__	:      Optional[int]     							=					num_mel_bins
          UpperCAmelCase__	:      Optional[Any]     							=					do_ceptral_normalize
          UpperCAmelCase__	:      List[Any]     							=					normalize_means
          UpperCAmelCase__	:      Union[str, Any]     							=					normalize_vars
          UpperCAmelCase__	:      Optional[int]     							=					True



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	):
          UpperCAmelCase__	:      List[Any]     							=					waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
          UpperCAmelCase__	:      str     							=					torch.from_numpy(UpperCamelCase_      ).unsqueeze(0      )
          UpperCAmelCase__	:      str     							=					ta_kaldi.fbank(UpperCamelCase_     ,	num_mel_bins=self.num_mel_bins     ,	sample_frequency=self.sampling_rate      )
          return features.numpy()



      @staticmethod
      def  __snake_case							(   UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_ = True     ,	UpperCamelCase_ = True     ,	UpperCamelCase_ = 0.0     ,	):
          # make sure we normalize float32 arrays
          if normalize_means:
              UpperCAmelCase__	:      Dict     							=					x[:input_length].mean(axis=0      )
              UpperCAmelCase__	:      Union[str, Any]     							=					np.subtract(UpperCamelCase_     ,	UpperCamelCase_      )
          if normalize_vars:
              UpperCAmelCase__	:      Any     							=					x[:input_length].std(axis=0      )
              UpperCAmelCase__	:      Dict     							=					np.divide(UpperCamelCase_     ,	UpperCamelCase_      )

          if input_length < x.shape[0]:
              UpperCAmelCase__	:      List[str]     							=					padding_value

          # make sure array is in float32
          UpperCAmelCase__	:      int     							=					x.astype(np.floataa      )

          return x



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_ = None      ):
          UpperCAmelCase__	:      Any     							=					attention_mask.sum(-1      ) if attention_mask is not None else [x.shape[0] for x in input_features]
          return [
              self.utterance_cmvn(UpperCamelCase_     ,	UpperCamelCase_     ,	self.normalize_means     ,	self.normalize_vars     ,	self.padding_value      )
              for x, n in zip(UpperCamelCase_     ,	UpperCamelCase_      )
          ]



      def __call__(   self     ,	UpperCamelCase_     ,	UpperCamelCase_ = False     ,	UpperCamelCase_ = None     ,	UpperCamelCase_ = False     ,	UpperCamelCase_ = None     ,	UpperCamelCase_ = None     ,	UpperCamelCase_ = None     ,	UpperCamelCase_ = None     ,	**UpperCamelCase_     ,	):

          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.'      )

          UpperCAmelCase__	:      str     							=					isinstance(UpperCamelCase_     ,	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}'''      )
          UpperCAmelCase__	:      Optional[Any]     							=					is_batched_numpy or (
              isinstance(UpperCamelCase_     ,	(list, tuple)      ) and (isinstance(raw_speech[0]     ,	(np.ndarray, tuple, list)      ))
          )

          if is_batched:
              UpperCAmelCase__	:      Dict     							=					[np.asarray(UpperCamelCase_     ,	dtype=np.floataa      ) for speech in raw_speech]
          elif not is_batched and not isinstance(UpperCamelCase_     ,	np.ndarray      ):
              UpperCAmelCase__	:      Union[str, Any]     							=					np.asarray(UpperCamelCase_     ,	dtype=np.floataa      )
          elif isinstance(UpperCamelCase_     ,	np.ndarray      ) and raw_speech.dtype is np.dtype(np.floataa      ):
              UpperCAmelCase__	:      List[str]     							=					raw_speech.astype(np.floataa      )

          # always return batch
          if not is_batched:
              UpperCAmelCase__	:      Tuple     							=					[raw_speech]

          # extract fbank features
          UpperCAmelCase__	:      Any     							=					[self._extract_fbank_features(UpperCamelCase_      ) for waveform in raw_speech]

          # convert into correct format for padding
          UpperCAmelCase__	:      Tuple     							=					BatchFeature({'input_features': features}      )

          UpperCAmelCase__	:      Any     							=					self.pad(
              UpperCamelCase_     ,	padding=UpperCamelCase_     ,	max_length=UpperCamelCase_     ,	truncation=UpperCamelCase_     ,	pad_to_multiple_of=UpperCamelCase_     ,	return_attention_mask=UpperCamelCase_     ,	**UpperCamelCase_     ,	)

          # make sure list is in array format
          UpperCAmelCase__	:      Union[str, Any]     							=					padded_inputs.get('input_features'      )
          if isinstance(input_features[0]     ,	UpperCamelCase_      ):
              UpperCAmelCase__	:      int     							=					[np.asarray(UpperCamelCase_     ,	dtype=np.floataa      ) for feature in input_features]

          UpperCAmelCase__	:      Optional[int]     							=					padded_inputs.get('attention_mask'      )
          if attention_mask is not None:
              UpperCAmelCase__	:      List[Any]     							=					[np.asarray(UpperCamelCase_     ,	dtype=np.intaa      ) for array in attention_mask]

          # Utterance-level cepstral mean and variance normalization
          if self.do_ceptral_normalize:
              UpperCAmelCase__	:      Optional[int]     							=					(
                  np.array(UpperCamelCase_     ,	dtype=np.intaa      )
                  if self._get_padding_strategies(UpperCamelCase_     ,	max_length=UpperCamelCase_      ) is not PaddingStrategy.DO_NOT_PAD
                  else None
              )
              UpperCAmelCase__	:      Union[str, Any]     							=					self.normalize(
                  padded_inputs['input_features']     ,	attention_mask=UpperCamelCase_      )

          if return_tensors is not None:
              UpperCAmelCase__	:      List[str]     							=					padded_inputs.convert_to_tensors(UpperCamelCase_      )

          return padded_inputs



"""simple docstring"""



import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()





class 				a			(					lowercase						,		unittest.TestCase      ):
      UpperCamelCase							:   Any           =     KandinskyImgaImgPipeline
      UpperCamelCase							:   int           =     ["""prompt""", """image_embeds""", """negative_image_embeds""", """image"""]
      UpperCamelCase							:   Optional[Any]           =     [
          """prompt""",
          """negative_prompt""",
          """image_embeds""",
          """negative_image_embeds""",
          """image""",
      ]
      UpperCamelCase							:   Union[str, Any]           =     [
          """generator""",
          """height""",
          """width""",
          """strength""",
          """guidance_scale""",
          """negative_prompt""",
          """num_inference_steps""",
          """return_dict""",
          """guidance_scale""",
          """num_images_per_prompt""",
          """output_type""",
          """return_dict""",
      ]
      UpperCamelCase							:   List[str]           =     False



      @property
      def  __snake_case							(   self      ):
          return 32



      @property
      def  __snake_case							(   self      ):
          return 32



      @property
      def  __snake_case							(   self      ):
          return self.time_input_dim



      @property
      def  __snake_case							(   self      ):
          return self.time_input_dim * 4



      @property
      def  __snake_case							(   self      ):
          return 100



      @property
      def  __snake_case							(   self      ):
          UpperCAmelCase__	:      str     							=					XLMRobertaTokenizerFast.from_pretrained('YiYiXu/tiny-random-mclip-base'      )
          return tokenizer



      @property
      def  __snake_case							(   self      ):
          torch.manual_seed(0      )
          UpperCAmelCase__	:      Optional[int]     							=					MCLIPConfig(
              numDims=self.cross_attention_dim     ,	transformerDimensions=self.text_embedder_hidden_size     ,	hidden_size=self.text_embedder_hidden_size     ,	intermediate_size=37     ,	num_attention_heads=4     ,	num_hidden_layers=5     ,	vocab_size=1_005     ,	)

          UpperCAmelCase__	:      Tuple     							=					MultilingualCLIP(UpperCamelCase_      )
          UpperCAmelCase__	:      List[Any]     							=					text_encoder.eval()

          return text_encoder



      @property
      def  __snake_case							(   self      ):
          torch.manual_seed(0      )

          UpperCAmelCase__	:      Optional[Any]     							=					{
              'in_channels': 4,
              # Out channels is double in channels because predicts mean and variance
              'out_channels': 8,
              'addition_embed_type': 'text_image',
              'down_block_types': ('ResnetDownsampleBlock2D', 'SimpleCrossAttnDownBlock2D'),
              'up_block_types': ('SimpleCrossAttnUpBlock2D', 'ResnetUpsampleBlock2D'),
              'mid_block_type': 'UNetMidBlock2DSimpleCrossAttn',
              'block_out_channels': (self.block_out_channels_a, self.block_out_channels_a * 2),
              'layers_per_block': 1,
              'encoder_hid_dim': self.text_embedder_hidden_size,
              'encoder_hid_dim_type': 'text_image_proj',
              'cross_attention_dim': self.cross_attention_dim,
              'attention_head_dim': 4,
              'resnet_time_scale_shift': 'scale_shift',
              'class_embed_type': None,
          }

          UpperCAmelCase__	:      Optional[Any]     							=					UNetaDConditionModel(**UpperCamelCase_      )
          return model



      @property
      def  __snake_case							(   self      ):
          return {
              "block_out_channels": [32, 64],
              "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
              "in_channels": 3,
              "latent_channels": 4,
              "layers_per_block": 1,
              "norm_num_groups": 8,
              "norm_type": "spatial",
              "num_vq_embeddings": 12,
              "out_channels": 3,
              "up_block_types": [
                  "AttnUpDecoderBlock2D",
                  "UpDecoderBlock2D",
              ],
              "vq_embed_dim": 4,
          }



      @property
      def  __snake_case							(   self      ):
          torch.manual_seed(0      )
          UpperCAmelCase__	:      Optional[Any]     							=					VQModel(**self.dummy_movq_kwargs      )
          return model



      def  __snake_case							(   self      ):
          UpperCAmelCase__	:      Union[str, Any]     							=					self.dummy_text_encoder
          UpperCAmelCase__	:      int     							=					self.dummy_tokenizer
          UpperCAmelCase__	:      int     							=					self.dummy_unet
          UpperCAmelCase__	:      Optional[Any]     							=					self.dummy_movq

          UpperCAmelCase__	:      Any     							=					{
              'num_train_timesteps': 1_000,
              'beta_schedule': 'linear',
              'beta_start': 0.00085,
              'beta_end': 0.012,
              'clip_sample': False,
              'set_alpha_to_one': False,
              'steps_offset': 0,
              'prediction_type': 'epsilon',
              'thresholding': False,
          }

          UpperCAmelCase__	:      List[Any]     							=					DDIMScheduler(**UpperCamelCase_      )

          UpperCAmelCase__	:      Tuple     							=					{
              'text_encoder': text_encoder,
              'tokenizer': tokenizer,
              'unet': unet,
              'scheduler': scheduler,
              'movq': movq,
          }

          return components



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_=0      ):
          UpperCAmelCase__	:      int     							=					floats_tensor((1, self.cross_attention_dim)     ,	rng=random.Random(UpperCamelCase_      )      ).to(UpperCamelCase_      )
          UpperCAmelCase__	:      Optional[Any]     							=					floats_tensor((1, self.cross_attention_dim)     ,	rng=random.Random(seed + 1      )      ).to(UpperCamelCase_      )
          # create init_image
          UpperCAmelCase__	:      Optional[int]     							=					floats_tensor((1, 3, 64, 64)     ,	rng=random.Random(UpperCamelCase_      )      ).to(UpperCamelCase_      )
          UpperCAmelCase__	:      int     							=					image.cpu().permute(0     ,	2     ,	3     ,	1      )[0]
          UpperCAmelCase__	:      Union[str, Any]     							=					Image.fromarray(np.uinta(UpperCamelCase_      )      ).convert('RGB'      ).resize((256, 256)      )

          if str(UpperCamelCase_      ).startswith('mps'      ):
              UpperCAmelCase__	:      Dict     							=					torch.manual_seed(UpperCamelCase_      )
          else:
              UpperCAmelCase__	:      Optional[int]     							=					torch.Generator(device=UpperCamelCase_      ).manual_seed(UpperCamelCase_      )
          UpperCAmelCase__	:      Tuple     							=					{
              'prompt': 'horse',
              'image': init_image,
              'image_embeds': image_embeds,
              'negative_image_embeds': negative_image_embeds,
              'generator': generator,
              'height': 64,
              'width': 64,
              'num_inference_steps': 10,
              'guidance_scale': 7.0,
              'strength': 0.2,
              'output_type': 'np',
          }
          return inputs



      def  __snake_case							(   self      ):
          UpperCAmelCase__	:      Optional[int]     							=					'cpu'

          UpperCAmelCase__	:      Any     							=					self.get_dummy_components()

          UpperCAmelCase__	:      Any     							=					self.pipeline_class(**UpperCamelCase_      )
          UpperCAmelCase__	:      Tuple     							=					pipe.to(UpperCamelCase_      )

          pipe.set_progress_bar_config(disable=UpperCamelCase_      )

          UpperCAmelCase__	:      List[Any]     							=					pipe(**self.get_dummy_inputs(UpperCamelCase_      )      )
          UpperCAmelCase__	:      Optional[int]     							=					output.images

          UpperCAmelCase__	:      Dict     							=					pipe(
              **self.get_dummy_inputs(UpperCamelCase_      )     ,	return_dict=UpperCamelCase_     ,	)[0]

          UpperCAmelCase__	:      Dict     							=					image[0, -3:, -3:, -1]
          UpperCAmelCase__	:      Any     							=					image_from_tuple[0, -3:, -3:, -1]

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

          UpperCAmelCase__	:      Any     							=					np.array(
              [0.61474943, 0.6073539, 0.43308544, 0.5928269, 0.47493595, 0.46755973, 0.4613838, 0.45368797, 0.50119233]      )
          assert (
              np.abs(image_slice.flatten() - expected_slice      ).max() < 1E-2
          ), F''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''
          assert (
              np.abs(image_from_tuple_slice.flatten() - expected_slice      ).max() < 1E-2
          ), F''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''





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



      def  __snake_case							(   self      ):
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()



      def  __snake_case							(   self      ):
          UpperCAmelCase__	:      str     							=					load_numpy(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
              '/kandinsky/kandinsky_img2img_frog.npy'      )

          UpperCAmelCase__	:      List[Any]     							=					load_image(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main' '/kandinsky/cat.png'      )
          UpperCAmelCase__	:      Any     							=					'A red cartoon frog, 4k'

          UpperCAmelCase__	:      Union[str, Any]     							=					KandinskyPriorPipeline.from_pretrained(
              'kandinsky-community/kandinsky-2-1-prior'     ,	torch_dtype=torch.floataa      )
          pipe_prior.to(UpperCamelCase_      )

          UpperCAmelCase__	:      Any     							=					KandinskyImgaImgPipeline.from_pretrained(
              'kandinsky-community/kandinsky-2-1'     ,	torch_dtype=torch.floataa      )
          UpperCAmelCase__	:      Optional[Any]     							=					pipeline.to(UpperCamelCase_      )

          pipeline.set_progress_bar_config(disable=UpperCamelCase_      )

          UpperCAmelCase__	:      List[Any]     							=					torch.Generator(device='cpu'      ).manual_seed(0      )
          UpperCAmelCase__							,							UpperCAmelCase__	:      Optional[int]     							=					pipe_prior(
              UpperCamelCase_     ,	generator=UpperCamelCase_     ,	num_inference_steps=5     ,	negative_prompt=''     ,	).to_tuple()

          UpperCAmelCase__	:      Optional[Any]     							=					pipeline(
              UpperCamelCase_     ,	image=UpperCamelCase_     ,	image_embeds=UpperCamelCase_     ,	negative_image_embeds=UpperCamelCase_     ,	generator=UpperCamelCase_     ,	num_inference_steps=100     ,	height=768     ,	width=768     ,	strength=0.2     ,	output_type='np'     ,	)

          UpperCAmelCase__	:      Union[str, Any]     							=					output.images[0]

          assert image.shape == (768, 768, 3)

          assert_mean_pixel_difference(UpperCamelCase_     ,	UpperCamelCase_      )

