
'''simple docstring'''





import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
 import torch

if is_datasets_available():
 from datasets import load_dataset

__lowerCAmelCase					:			Dict			      =		random.Random()


def     UpperCAmelCase_  (						__lowerCAmelCase	,					__lowerCAmelCase=1.0	,					__lowerCAmelCase=None	,					__lowerCAmelCase=None	) ->						List[str]:
    if rng is None:
        __lowercase		:			Any	    =       global_rng

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

    return values


class 						__lowerCAmelCase	(	unittest.TestCase      ):



      """simple docstring"""







      def __init__(      self		:       List[Any]				,       _snake_case		:       Optional[int]				,       _snake_case		:       str=7				,       _snake_case		:       Any=400				,       _snake_case		:       List[str]=2000				,       _snake_case		:       List[Any]=2048				,       _snake_case		:       Any=128				,       _snake_case		:       Union[str, Any]=1				,       _snake_case		:       List[Any]=512				,       _snake_case		:       str=30				,       _snake_case		:       Tuple=4_4100				,       ):
          __lowercase		:			str	    =       parent
          __lowercase		:			str	    =       batch_size
          __lowercase		:			str	    =       min_seq_length
          __lowercase		:			Dict	    =       max_seq_length
          __lowercase		:			Optional[int]	    =       (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
          __lowercase		:			Optional[Any]	    =       spectrogram_length
          __lowercase		:			Optional[int]	    =       feature_size
          __lowercase		:			Tuple	    =       num_audio_channels
          __lowercase		:			Union[str, Any]	    =       hop_length
          __lowercase		:			List[Any]	    =       chunk_length
          __lowercase		:			str	    =       sampling_rate




      def 		snake_case_			(      self		:       Optional[Any]				):
          return {
              "spectrogram_length": self.spectrogram_length,
              "feature_size": self.feature_size,
              "num_audio_channels": self.num_audio_channels,
              "hop_length": self.hop_length,
              "chunk_length": self.chunk_length,
              "sampling_rate": self.sampling_rate,
          }




      def 		snake_case_			(      self		:       Tuple				,       _snake_case		:       List[str]=False				,       _snake_case		:       Optional[int]=False				):





          def _flatten(_snake_case		:       str				):
              return list(itertools.chain(*UpperCamelCase__				)				)

          if equal_length:
              __lowercase		:			List[str]	    =       [floats_list((self.max_seq_length, self.feature_size)				) for _ in range(self.batch_size				)]
          else:
              # make sure that inputs increase in size
              __lowercase		:			int	    =       [
                  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		:			Optional[Any]	    =       [np.asarray(UpperCamelCase__				) for x in speech_inputs]
          return speech_inputs






@require_torch
@require_torchaudio
class 						__lowerCAmelCase	(	UpperCamelCase_ , unittest.TestCase      ):



      """simple docstring"""







      A__	:			int        =    TvltFeatureExtractor




      def 		snake_case_			(      self		:       List[str]				):
          __lowercase		:			Dict	    =       TvltFeatureExtractionTester(self				)




      def 		snake_case_			(      self		:       Union[str, Any]				):
          __lowercase		:			Any	    =       self.feature_extraction_class(**self.feat_extract_dict				)
          self.assertTrue(hasattr(UpperCamelCase__				,       '''spectrogram_length'''				)				)
          self.assertTrue(hasattr(UpperCamelCase__				,       '''feature_size'''				)				)
          self.assertTrue(hasattr(UpperCamelCase__				,       '''num_audio_channels'''				)				)
          self.assertTrue(hasattr(UpperCamelCase__				,       '''hop_length'''				)				)
          self.assertTrue(hasattr(UpperCamelCase__				,       '''chunk_length'''				)				)
          self.assertTrue(hasattr(UpperCamelCase__				,       '''sampling_rate'''				)				)




      def 		snake_case_			(      self		:       Union[str, Any]				):
          __lowercase		:			Union[str, Any]	    =       self.feature_extraction_class(**self.feat_extract_dict				)

          with tempfile.TemporaryDirectory() as tmpdirname:
              __lowercase		:			Optional[Any]	    =       feat_extract_first.save_pretrained(UpperCamelCase__				)[0]
              check_json_file_has_correct_format(UpperCamelCase__				)
              __lowercase		:			List[Any]	    =       self.feature_extraction_class.from_pretrained(UpperCamelCase__				)

          __lowercase		:			Optional[Any]	    =       feat_extract_first.to_dict()
          __lowercase		:			str	    =       feat_extract_second.to_dict()
          __lowercase		:			Optional[int]	    =       dict_first.pop('''mel_filters'''				)
          __lowercase		:			Optional[int]	    =       dict_second.pop('''mel_filters'''				)
          self.assertTrue(np.allclose(UpperCamelCase__				,       UpperCamelCase__				)				)
          self.assertEqual(UpperCamelCase__				,       UpperCamelCase__				)




      def 		snake_case_			(      self		:       int				):
          __lowercase		:			Dict	    =       self.feature_extraction_class(**self.feat_extract_dict				)

          with tempfile.TemporaryDirectory() as tmpdirname:
              __lowercase		:			Tuple	    =       os.path.join(UpperCamelCase__				,       '''feat_extract.json'''				)
              feat_extract_first.to_json_file(UpperCamelCase__				)
              __lowercase		:			List[str]	    =       self.feature_extraction_class.from_json_file(UpperCamelCase__				)

          __lowercase		:			Any	    =       feat_extract_first.to_dict()
          __lowercase		:			str	    =       feat_extract_second.to_dict()
          __lowercase		:			int	    =       dict_first.pop('''mel_filters'''				)
          __lowercase		:			Any	    =       dict_second.pop('''mel_filters'''				)
          self.assertTrue(np.allclose(UpperCamelCase__				,       UpperCamelCase__				)				)
          self.assertEqual(UpperCamelCase__				,       UpperCamelCase__				)




      def 		snake_case_			(      self		:       Any				):
          __lowercase		:			int	    =       self.feature_extraction_class(**self.feat_extract_dict				)

          # create three inputs of length 800, 1000, and 1200
          __lowercase		:			List[Any]	    =       [floats_list((1, x)				)[0] for x in range(800				,       1400				,       200				)]
          __lowercase		:			Optional[int]	    =       [np.asarray(UpperCamelCase__				) for speech_input in speech_inputs]

          # Test not batched input
          __lowercase		:			List[Any]	    =       feature_extractor(np_speech_inputs[0]				,       return_tensors='''np'''				,       sampling_rate=4_4100				).audio_values

          self.assertTrue(encoded_audios.ndim == 4				)
          self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
          self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
          self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

          # Test batched
          __lowercase		:			Optional[Any]	    =       feature_extractor(UpperCamelCase__				,       return_tensors='''np'''				,       sampling_rate=4_4100				).audio_values

          self.assertTrue(encoded_audios.ndim == 4				)
          self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
          self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
          self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

          # Test audio masking
          __lowercase		:			List[str]	    =       feature_extractor(
              UpperCamelCase__				,       return_tensors='''np'''				,       sampling_rate=4_4100				,       mask_audio=UpperCamelCase__				).audio_values

          self.assertTrue(encoded_audios.ndim == 4				)
          self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
          self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
          self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)

          # Test 2-D numpy arrays are batched.
          __lowercase		:			Optional[Any]	    =       [floats_list((1, x)				)[0] for x in (800, 800, 800)]
          __lowercase		:			Dict	    =       np.asarray(UpperCamelCase__				)
          __lowercase		:			str	    =       feature_extractor(UpperCamelCase__				,       return_tensors='''np'''				,       sampling_rate=4_4100				).audio_values
          self.assertTrue(encoded_audios.ndim == 4				)
          self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size				)
          self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length				)
          self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels				)




      def 		snake_case_			(      self		:       Optional[int]				,       _snake_case		:       int				):
          __lowercase		:			List[str]	    =       load_dataset('''hf-internal-testing/librispeech_asr_dummy'''				,       '''clean'''				,       split='''validation'''				)
          # automatic decoding with librispeech
          __lowercase		:			Dict	    =       ds.sort('''id'''				).select(range(UpperCamelCase__				)				)[:num_samples]['''audio''']

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




      def 		snake_case_			(      self		:       List[Any]				):
          __lowercase		:			Optional[Any]	    =       self._load_datasamples(1				)
          __lowercase		:			int	    =       TvltFeatureExtractor()
          __lowercase		:			List[Any]	    =       feature_extractor(UpperCamelCase__				,       return_tensors='''pt'''				).audio_values

          self.assertEquals(audio_values.shape				,       (1, 1, 192, 128)				)

          __lowercase		:			str	    =       torch.tensor([[-0.30_32, -0.27_08], [-0.44_34, -0.40_07]]				)
          self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]				,       UpperCamelCase__				,       atol=1E-4				)				)




import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np


class 						__lowerCAmelCase	:



      """simple docstring"""







      def 		snake_case_			(      self		:       str				):
          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Optional[Any]	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Tuple	    =       UNetaDConditionModel(
              sample_size=32				,       layers_per_block=1				,       block_out_channels=[32, 64]				,       down_block_types=[
                  '''ResnetDownsampleBlock2D''',
                  '''SimpleCrossAttnDownBlock2D''',
              ]				,       mid_block_type='''UNetMidBlock2DSimpleCrossAttn'''				,       up_block_types=['''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D''']				,       in_channels=3				,       out_channels=6				,       cross_attention_dim=32				,       encoder_hid_dim=32				,       attention_head_dim=8				,       addition_embed_type='''text'''				,       addition_embed_type_num_heads=2				,       cross_attention_norm='''group_norm'''				,       resnet_time_scale_shift='''scale_shift'''				,       act_fn='''gelu'''				,       )
          unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          torch.manual_seed(0				)
          __lowercase		:			List[str]	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       thresholding=_snake_case				,       dynamic_thresholding_ratio=0.95				,       sample_max_value=1.0				,       prediction_type='''epsilon'''				,       variance_type='''learned_range'''				,       )

          torch.manual_seed(0				)
          __lowercase		:			str	    =       IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }




      def 		snake_case_			(      self		:       Optional[int]				):
          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Tuple	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Dict	    =       UNetaDConditionModel(
              sample_size=32				,       layers_per_block=[1, 2]				,       block_out_channels=[32, 64]				,       down_block_types=[
                  '''ResnetDownsampleBlock2D''',
                  '''SimpleCrossAttnDownBlock2D''',
              ]				,       mid_block_type='''UNetMidBlock2DSimpleCrossAttn'''				,       up_block_types=['''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D''']				,       in_channels=6				,       out_channels=6				,       cross_attention_dim=32				,       encoder_hid_dim=32				,       attention_head_dim=8				,       addition_embed_type='''text'''				,       addition_embed_type_num_heads=2				,       cross_attention_norm='''group_norm'''				,       resnet_time_scale_shift='''scale_shift'''				,       act_fn='''gelu'''				,       class_embed_type='''timestep'''				,       mid_block_scale_factor=1.4_14				,       time_embedding_act_fn='''gelu'''				,       time_embedding_dim=32				,       )
          unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          torch.manual_seed(0				)
          __lowercase		:			Dict	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       thresholding=_snake_case				,       dynamic_thresholding_ratio=0.95				,       sample_max_value=1.0				,       prediction_type='''epsilon'''				,       variance_type='''learned_range'''				,       )

          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       )

          torch.manual_seed(0				)
          __lowercase		:			Any	    =       IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "image_noising_scheduler": image_noising_scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }




      def 		snake_case_			(      self		:       Any				):
          __lowercase		:			Tuple	    =       self.get_dummy_components()
          __lowercase		:			Any	    =       self.pipeline_class(**_snake_case				)
          pipe.to(_snake_case				)
          pipe.set_progress_bar_config(disable=_snake_case				)

          __lowercase		:			List[str]	    =       self.get_dummy_inputs(_snake_case				)

          __lowercase		:			List[Any]	    =       inputs['''prompt''']
          __lowercase		:			int	    =       inputs['''generator''']
          __lowercase		:			Dict	    =       inputs['''num_inference_steps''']
          __lowercase		:			Optional[int]	    =       inputs['''output_type''']

          if "image" in inputs:
              __lowercase		:			Tuple	    =       inputs['''image''']
          else:
              __lowercase		:			Dict	    =       None

          if "mask_image" in inputs:
              __lowercase		:			Tuple	    =       inputs['''mask_image''']
          else:
              __lowercase		:			List[Any]	    =       None

          if "original_image" in inputs:
              __lowercase		:			List[Any]	    =       inputs['''original_image''']
          else:
              __lowercase		:			Optional[int]	    =       None

          __lowercase    ,      __lowercase		:			Optional[int]	    =       pipe.encode_prompt(_snake_case				)

          # inputs with prompt converted to embeddings
          __lowercase		:			int	    =       {
              '''prompt_embeds''': prompt_embeds,
              '''negative_prompt_embeds''': negative_prompt_embeds,
              '''generator''': generator,
              '''num_inference_steps''': num_inference_steps,
              '''output_type''': output_type,
          }

          if image is not None:
              __lowercase		:			Tuple	    =       image

          if mask_image is not None:
              __lowercase		:			List[str]	    =       mask_image

          if original_image is not None:
              __lowercase		:			List[Any]	    =       original_image

          # set all optional components to None
          for optional_component in pipe._optional_components:
              setattr(_snake_case				,       _snake_case				,       _snake_case				)

          __lowercase		:			List[str]	    =       pipe(**_snake_case				)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_snake_case				)
              __lowercase		:			List[Any]	    =       self.pipeline_class.from_pretrained(_snake_case				)
              pipe_loaded.to(_snake_case				)
              pipe_loaded.set_progress_bar_config(disable=_snake_case				)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          for optional_component in pipe._optional_components:
              self.assertTrue(
                  getattr(_snake_case				,       _snake_case				) is None				,       F'`{optional_component}` did not stay set to None after loading.'				,       )

          __lowercase		:			int	    =       self.get_dummy_inputs(_snake_case				)

          __lowercase		:			Union[str, Any]	    =       inputs['''generator''']
          __lowercase		:			Any	    =       inputs['''num_inference_steps''']
          __lowercase		:			Tuple	    =       inputs['''output_type''']

          # inputs with prompt converted to embeddings
          __lowercase		:			List[str]	    =       {
              '''prompt_embeds''': prompt_embeds,
              '''negative_prompt_embeds''': negative_prompt_embeds,
              '''generator''': generator,
              '''num_inference_steps''': num_inference_steps,
              '''output_type''': output_type,
          }

          if image is not None:
              __lowercase		:			Dict	    =       image

          if mask_image is not None:
              __lowercase		:			Tuple	    =       mask_image

          if original_image is not None:
              __lowercase		:			Any	    =       original_image

          __lowercase		:			List[str]	    =       pipe_loaded(**_snake_case				)[0]

          __lowercase		:			Optional[int]	    =       np.abs(to_np(_snake_case				) - to_np(_snake_case				)				).max()
          self.assertLess(_snake_case				,       1E-4				)




      def 		snake_case_			(      self		:       Optional[int]				):
          __lowercase		:			Union[str, Any]	    =       self.get_dummy_components()
          __lowercase		:			Dict	    =       self.pipeline_class(**_snake_case				)
          pipe.to(_snake_case				)
          pipe.set_progress_bar_config(disable=_snake_case				)

          __lowercase		:			Dict	    =       self.get_dummy_inputs(_snake_case				)
          __lowercase		:			Any	    =       pipe(**_snake_case				)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_snake_case				)
              __lowercase		:			Dict	    =       self.pipeline_class.from_pretrained(_snake_case				)
              pipe_loaded.to(_snake_case				)
              pipe_loaded.set_progress_bar_config(disable=_snake_case				)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          __lowercase		:			Optional[Any]	    =       self.get_dummy_inputs(_snake_case				)
          __lowercase		:			str	    =       pipe_loaded(**_snake_case				)[0]

          __lowercase		:			int	    =       np.abs(to_np(_snake_case				) - to_np(_snake_case				)				).max()
          self.assertLess(_snake_case				,       1E-4				)


