





'''simple docstring'''
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

import numpy as np
import PIL
from PIL import Image

from ...utils import BaseOutput, is_torch_available, is_transformers_available







@dataclass
class 							A__   (	A__  ):
  A__					=    42
  A__					=    42


if is_transformers_available() and is_torch_available():
     from .pipeline_semantic_stable_diffusion import SemanticStableDiffusionPipeline



'''simple docstring'''
import itertools
import os
import random
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers import is_speech_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_speech_available():
     from transformers import WhisperFeatureExtractor

if is_torch_available():
     import torch

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


def 	_lowerCAmelCase (						_UpperCamelCase						:					Optional[Any]							,			_UpperCamelCase						:					List[str]=1.0							,			_UpperCamelCase						:					Optional[int]=None							,			_UpperCamelCase						:					Dict=None    )				->      Any:
     """simple docstring"""




     if rng is None:
          _SCREAMING_SNAKE_CASE							   =global_rng

     _SCREAMING_SNAKE_CASE							   =[]
     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 							A__   (	unittest.TestCase  ):





  def __init__(		self				:     List[Any]							, _a				:     Tuple							, _a				:     Dict=7							, _a				:     List[Any]=400							, _a				:     List[str]=2000							, _a				:     Optional[Any]=10							, _a				:     Dict=160							, _a				:     Tuple=8							, _a				:     Any=0.0							, _a				:     Optional[Any]=4000							, _a				:     List[Any]=False							, _a				:     Dict=True							, )     ->   Union[str, Any]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =parent
       _SCREAMING_SNAKE_CASE							   =batch_size
       _SCREAMING_SNAKE_CASE							   =min_seq_length
       _SCREAMING_SNAKE_CASE							   =max_seq_length
       _SCREAMING_SNAKE_CASE							   =(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
       _SCREAMING_SNAKE_CASE							   =padding_value
       _SCREAMING_SNAKE_CASE							   =sampling_rate
       _SCREAMING_SNAKE_CASE							   =return_attention_mask
       _SCREAMING_SNAKE_CASE							   =do_normalize
       _SCREAMING_SNAKE_CASE							   =feature_size
       _SCREAMING_SNAKE_CASE							   =chunk_length
       _SCREAMING_SNAKE_CASE							   =hop_length





  def 						A  (		self				:     Any			)     ->   Optional[Any]:





       '''simple docstring'''






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





  def 						A  (		self				:     Optional[Any]							, _a				:     Any=False							, _a				:     Union[str, Any]=False			)     ->   Optional[Any]:





       '''simple docstring'''






       def _flatten(_a				:     Union[str, Any]			):
            return list(itertools.chain(*_a			)			)

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







@require_torch
@require_torchaudio
class 							A__   (	A__					,       unittest.TestCase  ):
  A__					=    WhisperFeatureExtractor if is_speech_available() else None





  def 						A  (		self				:     Union[str, Any]			)     ->   Any:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =WhisperFeatureExtractionTester(self			)





  def 						A  (		self				:     str			)     ->   Tuple:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.feature_extraction_class(**self.feat_extract_dict			)

       with tempfile.TemporaryDirectory() as tmpdirname:
            _SCREAMING_SNAKE_CASE							   =feat_extract_first.save_pretrained(_a			)[0]
            check_json_file_has_correct_format(_a			)
            _SCREAMING_SNAKE_CASE							   =self.feature_extraction_class.from_pretrained(_a			)

       _SCREAMING_SNAKE_CASE							   =feat_extract_first.to_dict()
       _SCREAMING_SNAKE_CASE							   =feat_extract_second.to_dict()
       _SCREAMING_SNAKE_CASE							   =feat_extract_first.mel_filters
       _SCREAMING_SNAKE_CASE							   =feat_extract_second.mel_filters
       self.assertTrue(np.allclose(_a							, _a			)			)
       self.assertEqual(_a							, _a			)





  def 						A  (		self				:     Tuple			)     ->   Any:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.feature_extraction_class(**self.feat_extract_dict			)

       with tempfile.TemporaryDirectory() as tmpdirname:
            _SCREAMING_SNAKE_CASE							   =os.path.join(_a							, 'feat_extract.json'			)
            feat_extract_first.to_json_file(_a			)
            _SCREAMING_SNAKE_CASE							   =self.feature_extraction_class.from_json_file(_a			)

       _SCREAMING_SNAKE_CASE							   =feat_extract_first.to_dict()
       _SCREAMING_SNAKE_CASE							   =feat_extract_second.to_dict()
       _SCREAMING_SNAKE_CASE							   =feat_extract_first.mel_filters
       _SCREAMING_SNAKE_CASE							   =feat_extract_second.mel_filters
       self.assertTrue(np.allclose(_a							, _a			)			)
       self.assertEqual(_a							, _a			)





  def 						A  (		self				:     List[Any]			)     ->   Union[str, Any]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()			)
       # create three inputs of length 800, 1000, and 1200
       _SCREAMING_SNAKE_CASE							   =[floats_list((1, x)			)[0] for x in range(800							, 1400							, 200			)]
       _SCREAMING_SNAKE_CASE							   =[np.asarray(_a			) for speech_input in speech_inputs]

       # Test feature size
       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, padding='max_length'							, return_tensors='np'			).input_features
       self.assertTrue(input_features.ndim == 3			)
       self.assertTrue(input_features.shape[-1] == feature_extractor.nb_max_frames			)
       self.assertTrue(input_features.shape[-2] == feature_extractor.feature_size			)

       # Test not batched input
       _SCREAMING_SNAKE_CASE							   =feature_extractor(speech_inputs[0]							, return_tensors='np'			).input_features
       _SCREAMING_SNAKE_CASE							   =feature_extractor(np_speech_inputs[0]							, return_tensors='np'			).input_features
       self.assertTrue(np.allclose(_a							, _a							, atol=1e-3			)			)

       # Test batched
       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, return_tensors='np'			).input_features
       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, return_tensors='np'			).input_features
       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.
       _SCREAMING_SNAKE_CASE							   =[floats_list((1, x)			)[0] for x in (800, 800, 800)]
       _SCREAMING_SNAKE_CASE							   =np.asarray(_a			)
       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, return_tensors='np'			).input_features
       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, return_tensors='np'			).input_features
       for enc_seq_a, enc_seq_a in zip(_a							, _a			):
            self.assertTrue(np.allclose(_a							, _a							, atol=1e-3			)			)

       # Test truncation required
       _SCREAMING_SNAKE_CASE							   =[floats_list((1, x)			)[0] for x in range(200							, (feature_extractor.n_samples + 500)							, 200			)]
       _SCREAMING_SNAKE_CASE							   =[np.asarray(_a			) for speech_input in speech_inputs]

       _SCREAMING_SNAKE_CASE							   =[x[: feature_extractor.n_samples] for x in speech_inputs]
       _SCREAMING_SNAKE_CASE							   =[np.asarray(_a			) for speech_input in speech_inputs_truncated]

       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, return_tensors='np'			).input_features
       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, return_tensors='np'			).input_features
       for enc_seq_a, enc_seq_a in zip(_a							, _a			):
            self.assertTrue(np.allclose(_a							, _a							, atol=1e-3			)			)





  def 						A  (		self				:     Any			)     ->   List[Any]:





       '''simple docstring'''






       import torch

       _SCREAMING_SNAKE_CASE							   =self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()			)
       _SCREAMING_SNAKE_CASE							   =np.random.rand(100							, 32			).astype(np.floataa			)
       _SCREAMING_SNAKE_CASE							   =np_speech_inputs.tolist()

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





  def 						A  (		self				:     Tuple							, _a				:     str			)     ->   Optional[Any]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =load_dataset('hf-internal-testing/librispeech_asr_dummy'							, 'clean'							, split='validation'			)
       # automatic decoding with librispeech
       _SCREAMING_SNAKE_CASE							   =ds.sort('id'			).select(range(_a			)			)[:num_samples]['audio']

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





  def 						A  (		self				:     List[Any]			)     ->   Optional[int]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =torch.tensor(
           [
               0.11_93, -0.09_46, -0.10_98, -0.01_96, 0.02_25, -0.06_90, -0.17_36, 0.09_51,
               0.09_71, -0.08_17, -0.07_02, 0.01_62, 0.02_60, 0.00_17, -0.01_92, -0.16_78,
               0.07_09, -0.18_67, -0.06_55, -0.02_74, -0.02_34, -0.18_84, -0.05_16, -0.05_54,
               -0.02_74, -0.14_25, -0.14_23, 0.08_37, 0.03_77, -0.08_54
           ]			)
       # fmt: on

       _SCREAMING_SNAKE_CASE							   =self._load_datasamples(1			)
       _SCREAMING_SNAKE_CASE							   =WhisperFeatureExtractor()
       _SCREAMING_SNAKE_CASE							   =feature_extractor(_a							, return_tensors='pt'			).input_features
       self.assertEqual(input_features.shape							, (1, 80, 3000)			)
       self.assertTrue(torch.allclose(input_features[0, 0, :30]							, _a							, atol=1e-4			)			)





  def 						A  (		self				:     Union[str, Any]			)     ->   Tuple:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()			)
       _SCREAMING_SNAKE_CASE							   =self._load_datasamples(1			)[0]
       _SCREAMING_SNAKE_CASE							   =((audio - audio.min()) / (audio.max() - audio.min())) * 6_5535  # Rescale to [0, 65535] to show issue
       _SCREAMING_SNAKE_CASE							   =feat_extract.zero_mean_unit_var_norm([audio]							, attention_mask=_a			)[0]

       self.assertTrue(np.all(np.mean(_a			) < 1e-3			)			)
       self.assertTrue(np.all(np.abs(np.var(_a			) - 1			) < 1e-3			)			)
