

import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionInstructPixaPixPipeline,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.utils import floats_tensor, load_image, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class 				a__				(      _UpperCAmelCase		,	_UpperCAmelCase		,	_UpperCAmelCase		,	unittest.TestCase				):







  """simple docstring"""




  UpperCAmelCase__					:       Dict					=StableDiffusionInstructPixaPixPipeline
  UpperCAmelCase__					:       List[Any]					=TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width""", """cross_attention_kwargs"""}
  UpperCAmelCase__					:       Optional[Any]					=TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
  UpperCAmelCase__					:       List[str]					=IMAGE_TO_IMAGE_IMAGE_PARAMS
  UpperCAmelCase__					:       str					=IMAGE_TO_IMAGE_IMAGE_PARAMS


  def        _lowercase					(	self			:       Tuple						)    ->Tuple:
        """simple docstring"""







        torch.manual_seed(0						)
        SCREAMING_SNAKE_CASE       :					Tuple        =      UNetaDConditionModel(
            block_out_channels=(3_2, 6_4) , layers_per_block=2 , sample_size=3_2 , in_channels=8 , out_channels=4 , down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""") , up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""") , cross_attention_dim=3_2 , )
        SCREAMING_SNAKE_CASE       :					List[Any]        =      PNDMScheduler(skip_prk_steps=A_						)
        torch.manual_seed(0						)
        SCREAMING_SNAKE_CASE       :					Tuple        =      AutoencoderKL(
            block_out_channels=[3_2, 6_4] , in_channels=3 , out_channels=3 , down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""] , up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""] , latent_channels=4 , )
        torch.manual_seed(0						)
        SCREAMING_SNAKE_CASE       :					Any        =      CLIPTextConfig(
            bos_token_id=0 , eos_token_id=2 , hidden_size=3_2 , intermediate_size=3_7 , layer_norm_eps=1e-05 , num_attention_heads=4 , num_hidden_layers=5 , pad_token_id=1 , vocab_size=1_0_0_0 , )
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      CLIPTextModel(A_						)
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)

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


  def        _lowercase					(	self			:       Dict , UpperCAmelCase__			:       Union[str, Any] , UpperCAmelCase__			:       Optional[Any]=0						)    ->List[str]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					Any        =      floats_tensor((1, 3, 3_2, 3_2) , rng=random.Random(A_						)						).to(A_						)
        SCREAMING_SNAKE_CASE       :					Optional[int]        =      image.cpu().permute(0 , 2 , 3 , 1						)[0]
        SCREAMING_SNAKE_CASE       :					Any        =      Image.fromarray(np.uinta(A_						)						).convert("""RGB"""						)
        if str(A_						).startswith("""mps"""						):
              SCREAMING_SNAKE_CASE       :					Dict        =      torch.manual_seed(A_						)
        else:
              SCREAMING_SNAKE_CASE       :					int        =      torch.Generator(device=A_						).manual_seed(A_						)
        SCREAMING_SNAKE_CASE       :					List[Any]        =      {
            '''prompt''': '''A painting of a squirrel eating a burger''',
            '''image''': image,
            '''generator''': generator,
            '''num_inference_steps''': 2,
            '''guidance_scale''': 6.0,
            '''image_guidance_scale''': 1,
            '''output_type''': '''numpy''',
        }
        return inputs


  def        _lowercase					(	self			:       int						)    ->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					int        =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        SCREAMING_SNAKE_CASE       :					Dict        =      self.get_dummy_components()
        SCREAMING_SNAKE_CASE       :					List[str]        =      StableDiffusionInstructPixaPixPipeline(**A_						)
        SCREAMING_SNAKE_CASE       :					int        =      sd_pipe.to(A_						)
        sd_pipe.set_progress_bar_config(disable=A_						)

        SCREAMING_SNAKE_CASE       :					List[str]        =      self.get_dummy_inputs(A_						)
        SCREAMING_SNAKE_CASE       :					Union[str, Any]        =      sd_pipe(**A_						).images
        SCREAMING_SNAKE_CASE       :					Dict        =      image[0, -3:, -3:, -1]
        assert image.shape == (1, 3_2, 3_2, 3)
        SCREAMING_SNAKE_CASE       :					List[Any]        =      np.array([0.75_26, 0.37_50, 0.45_47, 0.61_17, 0.58_66, 0.50_16, 0.43_27, 0.56_42, 0.48_15]						)

        assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-3


  def        _lowercase					(	self			:       List[Any]						)    ->List[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					Tuple        =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        SCREAMING_SNAKE_CASE       :					str        =      self.get_dummy_components()
        SCREAMING_SNAKE_CASE       :					List[str]        =      StableDiffusionInstructPixaPixPipeline(**A_						)
        SCREAMING_SNAKE_CASE       :					List[str]        =      sd_pipe.to(A_						)
        sd_pipe.set_progress_bar_config(disable=A_						)

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

        assert image.shape == (1, 3_2, 3_2, 3)
        SCREAMING_SNAKE_CASE       :					int        =      np.array([0.75_11, 0.36_42, 0.45_53, 0.62_36, 0.57_97, 0.50_13, 0.43_43, 0.56_11, 0.48_31]						)

        assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-3


  def        _lowercase					(	self			:       List[Any]						)    ->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        SCREAMING_SNAKE_CASE       :					Any        =      self.get_dummy_components()
        SCREAMING_SNAKE_CASE       :					int        =      StableDiffusionInstructPixaPixPipeline(**A_						)
        SCREAMING_SNAKE_CASE       :					Optional[int]        =      sd_pipe.to(A_						)
        sd_pipe.set_progress_bar_config(disable=A_						)

        SCREAMING_SNAKE_CASE       :					List[Any]        =      self.get_dummy_inputs(A_						)
        SCREAMING_SNAKE_CASE       :					Tuple        =      [inputs['''prompt''']] * 2

        SCREAMING_SNAKE_CASE       :					List[Any]        =      np.array(inputs["""image"""]						).astype(np.floataa						) / 2_55.0
        SCREAMING_SNAKE_CASE       :					Tuple        =      torch.from_numpy(A_						).unsqueeze(0						).to(A_						)
        SCREAMING_SNAKE_CASE       :					int        =      image / 2 + 0.5
        SCREAMING_SNAKE_CASE       :					int        =      image.permute(0 , 3 , 1 , 2						)
        SCREAMING_SNAKE_CASE       :					Optional[int]        =      image.repeat(2 , 1 , 1 , 1						)

        SCREAMING_SNAKE_CASE       :					int        =      sd_pipe(**A_						).images
        SCREAMING_SNAKE_CASE       :					List[Any]        =      image[-1, -3:, -3:, -1]

        assert image.shape == (2, 3_2, 3_2, 3)
        SCREAMING_SNAKE_CASE       :					Any        =      np.array([0.58_12, 0.57_48, 0.52_22, 0.59_08, 0.56_95, 0.71_74, 0.68_04, 0.55_23, 0.55_79]						)

        assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-3


  def        _lowercase					(	self			:       int						)    ->List[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					Dict        =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        SCREAMING_SNAKE_CASE       :					List[Any]        =      self.get_dummy_components()
        SCREAMING_SNAKE_CASE       :					Any        =      EulerAncestralDiscreteScheduler(
            beta_start=0.0_00_85 , beta_end=0.0_12 , beta_schedule="""scaled_linear"""						)
        SCREAMING_SNAKE_CASE       :					int        =      StableDiffusionInstructPixaPixPipeline(**A_						)
        SCREAMING_SNAKE_CASE       :					Dict        =      sd_pipe.to(A_						)
        sd_pipe.set_progress_bar_config(disable=A_						)

        SCREAMING_SNAKE_CASE       :					Optional[int]        =      self.get_dummy_inputs(A_						)
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      sd_pipe(**A_						).images
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      image[0, -3:, -3:, -1]

        SCREAMING_SNAKE_CASE       :					List[str]        =      [round(A_ , 4						) for x in image_slice.flatten().tolist()]
        print(""",""".join([str(A_						) for x in slice]						)						)

        assert image.shape == (1, 3_2, 3_2, 3)
        SCREAMING_SNAKE_CASE       :					List[str]        =      np.array([0.74_17, 0.38_42, 0.47_32, 0.57_76, 0.58_91, 0.51_39, 0.40_52, 0.56_73, 0.49_86]						)

        assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-3


  def        _lowercase					(	self			:       Optional[Any]						)    ->str:
        """simple docstring"""







        super().test_inference_batch_single_identical(expected_max_diff=3e-3						)




  def        _lowercase					(	self			:       List[str]						)    ->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					List[Any]        =      self.get_dummy_components()
        SCREAMING_SNAKE_CASE       :					Any        =      StableDiffusionInstructPixaPixPipeline(**A_						)
        SCREAMING_SNAKE_CASE       :					Dict        =      VaeImageProcessor(do_resize=A_ , do_normalize=A_						)
        SCREAMING_SNAKE_CASE       :					List[str]        =      pipe.to(A_						)
        pipe.set_progress_bar_config(disable=A_						)

        SCREAMING_SNAKE_CASE       :					List[str]        =      pipe(**self.get_dummy_inputs_by_type(A_ , input_image_type="""pt"""						)						)[0]

        SCREAMING_SNAKE_CASE       :					Any        =      components['''vae''']
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      self.get_dummy_inputs_by_type(A_ , input_image_type="""pt"""						)

        for image_param in self.image_latents_params:
              if image_param in inputs.keys():
                    SCREAMING_SNAKE_CASE       :					Any        =      vae.encode(inputs[image_param]						).latent_dist.mode()

        SCREAMING_SNAKE_CASE       :					Optional[int]        =      pipe(**A_						)[0]

        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      np.abs(out - out_latents_inputs						).max()
        self.assertLess(A_ , 1e-4 , """passing latents as image input generate different result from passing image"""						)







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







  """simple docstring"""




  def        _lowercase					(	self			:       Tuple						)    ->Dict:
        """simple docstring"""







        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()


  def        _lowercase					(	self			:       str , UpperCAmelCase__			:       Union[str, Any]=0						)    ->List[str]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					Optional[int]        =      torch.manual_seed(A_						)
        SCREAMING_SNAKE_CASE       :					Optional[int]        =      load_image(
            """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg"""						)
        SCREAMING_SNAKE_CASE       :					Optional[int]        =      {
            '''prompt''': '''turn him into a cyborg''',
            '''image''': image,
            '''generator''': generator,
            '''num_inference_steps''': 3,
            '''guidance_scale''': 7.5,
            '''image_guidance_scale''': 1.0,
            '''output_type''': '''numpy''',
        }
        return inputs


  def        _lowercase					(	self			:       Dict						)    ->List[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					str        =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
            """timbrooks/instruct-pix2pix""" , safety_checker=A_						)
        pipe.to(A_						)
        pipe.set_progress_bar_config(disable=A_						)
        pipe.enable_attention_slicing()

        SCREAMING_SNAKE_CASE       :					int        =      self.get_inputs()
        SCREAMING_SNAKE_CASE       :					Union[str, Any]        =      pipe(**A_						).images
        SCREAMING_SNAKE_CASE       :					str        =      image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 5_1_2, 5_1_2, 3)
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      np.array([0.59_02, 0.60_15, 0.60_27, 0.59_83, 0.60_92, 0.60_61, 0.57_65, 0.57_85, 0.55_55]						)

        assert np.abs(expected_slice - image_slice						).max() < 1e-3


  def        _lowercase					(	self			:       int						)    ->List[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					str        =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
            """timbrooks/instruct-pix2pix""" , safety_checker=A_						)
        SCREAMING_SNAKE_CASE       :					Tuple        =      LMSDiscreteScheduler.from_config(pipe.scheduler.config						)
        pipe.to(A_						)
        pipe.set_progress_bar_config(disable=A_						)
        pipe.enable_attention_slicing()

        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      self.get_inputs()
        SCREAMING_SNAKE_CASE       :					Dict        =      pipe(**A_						).images
        SCREAMING_SNAKE_CASE       :					Tuple        =      image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 5_1_2, 5_1_2, 3)
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      np.array([0.65_78, 0.68_17, 0.69_72, 0.67_61, 0.68_56, 0.69_16, 0.64_28, 0.65_16, 0.63_01]						)

        assert np.abs(expected_slice - image_slice						).max() < 1e-3


  def        _lowercase					(	self			:       Tuple						)    ->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					List[str]        =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
            """timbrooks/instruct-pix2pix""" , safety_checker=A_						)
        SCREAMING_SNAKE_CASE       :					List[Any]        =      DDIMScheduler.from_config(pipe.scheduler.config						)
        pipe.to(A_						)
        pipe.set_progress_bar_config(disable=A_						)
        pipe.enable_attention_slicing()

        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      self.get_inputs()
        SCREAMING_SNAKE_CASE       :					Union[str, Any]        =      pipe(**A_						).images
        SCREAMING_SNAKE_CASE       :					Union[str, Any]        =      image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 5_1_2, 5_1_2, 3)
        SCREAMING_SNAKE_CASE       :					Optional[Any]        =      np.array([0.38_28, 0.38_34, 0.38_18, 0.37_92, 0.38_65, 0.37_52, 0.37_92, 0.38_47, 0.37_53]						)

        assert np.abs(expected_slice - image_slice						).max() < 1e-3


  def        _lowercase					(	self			:       int						)    ->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					Dict        =      0

        def callback_fn(UpperCAmelCase__			:       Optional[int] , UpperCAmelCase__			:       str , UpperCAmelCase__			:       Union[str, Any]						) -> None:
              SCREAMING_SNAKE_CASE       :					Union[str, Any]        =      True
              nonlocal number_of_steps
              number_of_steps += 1
              if step == 1:
                    SCREAMING_SNAKE_CASE       :					Any        =      latents.detach().cpu().numpy()
                    assert latents.shape == (1, 4, 6_4, 6_4)
                    SCREAMING_SNAKE_CASE       :					List[str]        =      latents[0, -3:, -3:, -1]
                    SCREAMING_SNAKE_CASE       :					Any        =      np.array([-0.24_63, -0.46_44, -0.97_56, 1.51_76, 1.44_14, 0.78_66, 0.98_97, 0.85_21, 0.79_83]						)

                    assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5e-2
              elif step == 2:
                    SCREAMING_SNAKE_CASE       :					int        =      latents.detach().cpu().numpy()
                    assert latents.shape == (1, 4, 6_4, 6_4)
                    SCREAMING_SNAKE_CASE       :					List[str]        =      latents[0, -3:, -3:, -1]
                    SCREAMING_SNAKE_CASE       :					Any        =      np.array([-0.26_44, -0.46_26, -0.96_53, 1.51_76, 1.45_51, 0.76_86, 0.98_05, 0.84_52, 0.81_15]						)

                    assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5e-2

        SCREAMING_SNAKE_CASE       :					List[str]        =      False

        SCREAMING_SNAKE_CASE       :					Union[str, Any]        =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
            """timbrooks/instruct-pix2pix""" , safety_checker=A_ , torch_dtype=torch.floataa						)
        SCREAMING_SNAKE_CASE       :					List[Any]        =      pipe.to(A_						)
        pipe.set_progress_bar_config(disable=A_						)
        pipe.enable_attention_slicing()

        SCREAMING_SNAKE_CASE       :					Optional[int]        =      self.get_inputs()
        pipe(**A_ , callback=A_ , callback_steps=1						)
        assert callback_fn.has_been_called
        assert number_of_steps == 3


  def        _lowercase					(	self			:       List[str]						)    ->List[str]:
        """simple docstring"""







        torch.cuda.empty_cache()
        torch.cuda.reset_max_memory_allocated()
        torch.cuda.reset_peak_memory_stats()

        SCREAMING_SNAKE_CASE       :					Dict        =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
            """timbrooks/instruct-pix2pix""" , safety_checker=A_ , torch_dtype=torch.floataa						)
        SCREAMING_SNAKE_CASE       :					str        =      pipe.to(A_						)
        pipe.set_progress_bar_config(disable=A_						)
        pipe.enable_attention_slicing(1						)
        pipe.enable_sequential_cpu_offload()

        SCREAMING_SNAKE_CASE       :					Union[str, Any]        =      self.get_inputs()
        SCREAMING_SNAKE_CASE       :					int        =      pipe(**A_						)

        SCREAMING_SNAKE_CASE       :					Dict        =      torch.cuda.max_memory_allocated()
        # make sure that less than 2.2 GB is allocated
        assert mem_bytes < 2.2 * 1_0**9




  def        _lowercase					(	self			:       Optional[int]						)    ->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE       :					Optional[int]        =      self.get_inputs()
        # resize to resolution that is divisible by 8 but not 16 or 32
        SCREAMING_SNAKE_CASE       :					List[str]        =      inputs['''image'''].resize((5_0_4, 5_0_4)						)

        SCREAMING_SNAKE_CASE       :					Tuple        =      '''timbrooks/instruct-pix2pix'''
        SCREAMING_SNAKE_CASE       :					List[str]        =      StableDiffusionInstructPixaPixPipeline.from_pretrained(
            A_ , safety_checker=A_ , )
        pipe.to(A_						)
        pipe.set_progress_bar_config(disable=A_						)
        pipe.enable_attention_slicing()

        SCREAMING_SNAKE_CASE       :					List[Any]        =      pipe(**A_						)
        SCREAMING_SNAKE_CASE       :					Optional[int]        =      output.images[0]

        SCREAMING_SNAKE_CASE       :					Any        =      image[2_5_5:2_5_8, 3_8_3:3_8_6, -1]

        assert image.shape == (5_0_4, 5_0_4, 3)
        SCREAMING_SNAKE_CASE       :					int        =      np.array([0.27_26, 0.25_29, 0.26_64, 0.26_55, 0.26_41, 0.26_42, 0.25_91, 0.26_49, 0.25_90]						)

        assert np.abs(image_slice.flatten() - expected_slice						).max() < 5e-3







import itertools
import random
import unittest

import numpy as np

from transformers import WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST, WavaVecaConfig, WavaVecaFeatureExtractor
from transformers.testing_utils import require_torch, slow

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


_UpperCamelCase 			=		random.Random()







def 		_lowercase (     lowercase__   ,	lowercase__=1.0   ,	lowercase__=None   ,	lowercase__=None						):
      if rng is None:
            __lowerCAmelCase					:						Optional[Any]					=					global_rng

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

      return values







class   __lowercase      (unittest.TestCase ):
 def __init__(							self      ,							A_      ,							A_=7      ,							A_=400      ,							A_=2000      ,							A_=1      ,							A_=0.0      ,							A_=1_6000      ,							A_=True      ,							A_=True      ,							)					->List[Any]:



       '''simple docstring'''





       __lowerCAmelCase					:						List[Any]					=					parent
       __lowerCAmelCase					:						Optional[int]					=					batch_size
       __lowerCAmelCase					:						Any					=					min_seq_length
       __lowerCAmelCase					:						Tuple					=					max_seq_length
       __lowerCAmelCase					:						Tuple					=					(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
       __lowerCAmelCase					:						Dict					=					feature_size
       __lowerCAmelCase					:						Optional[int]					=					padding_value
       __lowerCAmelCase					:						Tuple					=					sampling_rate
       __lowerCAmelCase					:						Union[str, Any]					=					return_attention_mask
       __lowerCAmelCase					:						Dict					=					do_normalize
 def 	UpperCamelCase__  (							self					)					->Optional[Any]:



       '''simple docstring'''





       return {
           "feature_size": self.feature_size,
           "padding_value": self.padding_value,
           "sampling_rate": self.sampling_rate,
           "return_attention_mask": self.return_attention_mask,
           "do_normalize": self.do_normalize,
       }





 def 	UpperCamelCase__  (							self      ,							A_=False      ,							A_=False					)					->Union[str, Any]:



       '''simple docstring'''





       def _flatten(A_					):
             return list(itertools.chain(*A_					)					)

       if equal_length:
             __lowerCAmelCase					:						Dict					=					floats_list((self.batch_size, self.max_seq_length)					)
       else:
             # make sure that inputs increase in size
             __lowerCAmelCase					:						Tuple					=					[
                 _flatten(floats_list((x, self.feature_size)					)					)
                 for x in range(self.min_seq_length      ,							self.max_seq_length      ,							self.seq_length_diff					)
             ]

       if numpify:
             __lowerCAmelCase					:						Tuple					=					[np.asarray(A_					) for x in speech_inputs]

       return speech_inputs







class   __lowercase      (_UpperCAmelCase		,   unittest.TestCase ):
 _UpperCamelCase			  =  WavaVecaFeatureExtractor
 def 	UpperCamelCase__  (							self					)					->Dict:



       '''simple docstring'''





       __lowerCAmelCase					:						List[Any]					=					WavaVecaFeatureExtractionTester(self					)
 def 	UpperCamelCase__  (							self      ,							A_					)					->Optional[Any]:



       '''simple docstring'''





       self.assertTrue(np.all(np.mean(A_      ,							axis=0					) < 1e-3					)					)
       self.assertTrue(np.all(np.abs(np.var(A_      ,							axis=0					) - 1					) < 1e-3					)					)
 def 	UpperCamelCase__  (							self					)					->Tuple:



       '''simple docstring'''





       __lowerCAmelCase					:						List[str]					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       # create three inputs of length 800, 1000, and 1200
       __lowerCAmelCase					:						Optional[Any]					=					[floats_list((1, x)					)[0] for x in range(800      ,							1400      ,							200					)]
       __lowerCAmelCase					:						Any					=					[np.asarray(A_					) for speech_input in speech_inputs]

       # Test not batched input
       __lowerCAmelCase					:						Optional[Any]					=					feat_extract(speech_inputs[0]      ,							return_tensors='''np'''					).input_values
       __lowerCAmelCase					:						Dict					=					feat_extract(np_speech_inputs[0]      ,							return_tensors='''np'''					).input_values
       self.assertTrue(np.allclose(A_      ,							A_      ,							atol=1e-3					)					)

       # Test batched
       __lowerCAmelCase					:						Dict					=					feat_extract(A_      ,							return_tensors='''np'''					).input_values
       __lowerCAmelCase					:						Dict					=					feat_extract(A_      ,							return_tensors='''np'''					).input_values
       for enc_seq_a, enc_seq_a in zip(A_      ,							A_					):
             self.assertTrue(np.allclose(A_      ,							A_      ,							atol=1e-3					)					)

       # Test 2-D numpy arrays are batched.
       __lowerCAmelCase					:						int					=					[floats_list((1, x)					)[0] for x in (800, 800, 800)]
       __lowerCAmelCase					:						List[Any]					=					np.asarray(A_					)
       __lowerCAmelCase					:						Any					=					feat_extract(A_      ,							return_tensors='''np'''					).input_values
       __lowerCAmelCase					:						Union[str, Any]					=					feat_extract(A_      ,							return_tensors='''np'''					).input_values
       for enc_seq_a, enc_seq_a in zip(A_      ,							A_					):
             self.assertTrue(np.allclose(A_      ,							A_      ,							atol=1e-3					)					)
 def 	UpperCamelCase__  (							self					)					->Union[str, Any]:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[int]					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       __lowerCAmelCase					:						str					=					[floats_list((1, x)					)[0] for x in range(800      ,							1400      ,							200					)]

       __lowerCAmelCase					:						str					=					['''longest''', '''max_length''', '''do_not_pad''']
       __lowerCAmelCase					:						str					=					[None, 1600, None]
       for max_length, padding in zip(A_      ,							A_					):
             __lowerCAmelCase					:						Optional[int]					=					feat_extract(A_      ,							padding=A_      ,							max_length=A_      ,							return_tensors='''np'''					)
             __lowerCAmelCase					:						Optional[Any]					=					processed.input_values

             self._check_zero_mean_unit_variance(input_values[0][:800]					)
             self.assertTrue(input_values[0][800:].sum() < 1e-6					)
             self._check_zero_mean_unit_variance(input_values[1][:1000]					)
             self.assertTrue(input_values[0][1000:].sum() < 1e-6					)
             self._check_zero_mean_unit_variance(input_values[2][:1200]					)
 def 	UpperCamelCase__  (							self					)					->Dict:



       '''simple docstring'''





       __lowerCAmelCase					:						List[str]					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       __lowerCAmelCase					:						Optional[int]					=					range(800      ,							1400      ,							200					)
       __lowerCAmelCase					:						Union[str, Any]					=					[floats_list((1, x)					)[0] for x in lengths]

       __lowerCAmelCase					:						int					=					['''longest''', '''max_length''', '''do_not_pad''']
       __lowerCAmelCase					:						List[str]					=					[None, 1600, None]

       for max_length, padding in zip(A_      ,							A_					):
             __lowerCAmelCase					:						Union[str, Any]					=					feat_extract(A_      ,							max_length=A_      ,							padding=A_					)
             __lowerCAmelCase					:						Union[str, Any]					=					processed.input_values

             self._check_zero_mean_unit_variance(input_values[0][:800]					)
             self._check_zero_mean_unit_variance(input_values[1][:1000]					)
             self._check_zero_mean_unit_variance(input_values[2][:1200]					)
 def 	UpperCamelCase__  (							self					)					->Tuple:



       '''simple docstring'''





       __lowerCAmelCase					:						Dict					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       __lowerCAmelCase					:						Optional[Any]					=					[floats_list((1, x)					)[0] for x in range(800      ,							1400      ,							200					)]
       __lowerCAmelCase					:						List[str]					=					feat_extract(
           A_      ,							truncation=A_      ,							max_length=1000      ,							padding='''max_length'''      ,							return_tensors='''np'''					)
       __lowerCAmelCase					:						int					=					processed.input_values

       self._check_zero_mean_unit_variance(input_values[0, :800]					)
       self._check_zero_mean_unit_variance(input_values[1]					)
       self._check_zero_mean_unit_variance(input_values[2]					)
 def 	UpperCamelCase__  (							self					)					->List[str]:



       '''simple docstring'''





       __lowerCAmelCase					:						Dict					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       __lowerCAmelCase					:						int					=					[floats_list((1, x)					)[0] for x in range(800      ,							1400      ,							200					)]
       __lowerCAmelCase					:						int					=					feat_extract(
           A_      ,							truncation=A_      ,							max_length=1000      ,							padding='''longest'''      ,							return_tensors='''np'''					)
       __lowerCAmelCase					:						Optional[Any]					=					processed.input_values

       self._check_zero_mean_unit_variance(input_values[0, :800]					)
       self._check_zero_mean_unit_variance(input_values[1, :1000]					)
       self._check_zero_mean_unit_variance(input_values[2]					)

       # make sure that if max_length < longest -> then pad to max_length
       self.assertTrue(input_values.shape == (3, 1000)					)

       __lowerCAmelCase					:						Any					=					[floats_list((1, x)					)[0] for x in range(800      ,							1400      ,							200					)]
       __lowerCAmelCase					:						Optional[int]					=					feat_extract(
           A_      ,							truncation=A_      ,							max_length=2000      ,							padding='''longest'''      ,							return_tensors='''np'''					)
       __lowerCAmelCase					:						List[str]					=					processed.input_values

       self._check_zero_mean_unit_variance(input_values[0, :800]					)
       self._check_zero_mean_unit_variance(input_values[1, :1000]					)
       self._check_zero_mean_unit_variance(input_values[2]					)

       # make sure that if max_length > longest -> then pad to longest
       self.assertTrue(input_values.shape == (3, 1200)					)
 @require_torch
 def 	UpperCamelCase__  (							self					)					->Any:



       '''simple docstring'''





       import torch

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

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





 @slow
 @require_torch
 def 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





       for model_id in WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST:
             __lowerCAmelCase					:						Any					=					WavaVecaConfig.from_pretrained(A_					)
             __lowerCAmelCase					:						Tuple					=					WavaVecaFeatureExtractor.from_pretrained(A_					)

             # only "layer" feature extraction norm should make use of
             # attention_mask
             self.assertEqual(feat_extract.return_attention_mask      ,							config.feat_extract_norm == '''layer'''					)


