





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							:					int		    =		random.Random()

def    __snake_case  (    _lowerCAmelCase       :       Union[str, Any] , _lowerCAmelCase       :       Dict=1.0 , _lowerCAmelCase       :       str=None , _lowerCAmelCase       :       Any=None		)       ->  Optional[Any]:


   if rng is None:
      A_    : int								=	global_rng

   A_    : List[Any]								=	[]
   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     __magic_name__     (							unittest.TestCase							):





     """simple docstring"""



     def __init__(							self				:Optional[Any]	,					snake_case				:Dict	,					snake_case				:str=7	,					snake_case				:Any=400	,					snake_case				:Any=2_000	,					snake_case				:int=10	,					snake_case				:List[Any]=160	,					snake_case				:List[Any]=8	,					snake_case				:Optional[int]=0.0	,					snake_case				:Optional[Any]=4_000	,					snake_case				:List[str]=False	,					snake_case				:Union[str, Any]=True	,					):





        '''simple docstring'''




        A_    : Optional[Any]								=	parent
        A_    : int								=	batch_size
        A_    : int								=	min_seq_length
        A_    : Tuple								=	max_seq_length
        A_    : Optional[int]								=	(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        A_    : Dict								=	padding_value
        A_    : int								=	sampling_rate
        A_    : List[str]								=	return_attention_mask
        A_    : List[Any]								=	do_normalize
        A_    : List[Any]								=	feature_size
        A_    : Dict								=	chunk_length
        A_    : List[Any]								=	hop_length



     def   SCREAMING_SNAKE_CASE						(							self				: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   SCREAMING_SNAKE_CASE						(							self				:Optional[int]	,					snake_case				:str=False	,					snake_case				:List[Any]=False     ):





        '''simple docstring'''




        def _flatten(snake_case				:Any     ):
           return list(itertools.chain(*a_     )     )

        if equal_length:
           A_    : Union[str, Any]								=	[floats_list((self.max_seq_length, self.feature_size)     ) for _ in range(self.batch_size     )]
        else:
           # make sure that inputs increase in size
           A_    : Dict								=	[
               floats_list((x, self.feature_size)     )
               for x in range(self.min_seq_length	,					self.max_seq_length	,					self.seq_length_diff     )
           ]
        if numpify:
           A_    : int								=	[np.asarray(a_     ) for x in speech_inputs]
        return speech_inputs




@require_torch
@require_torchaudio
class     __magic_name__     (							_lowerCamelCase			, unittest.TestCase							):





     """simple docstring"""



     __UpperCamelCase    			=   WhisperFeatureExtractor if is_speech_available() else None



     def   SCREAMING_SNAKE_CASE						(							self				:List[Any]     ):





        '''simple docstring'''




        A_    : List[str]								=	WhisperFeatureExtractionTester(self     )



     def   SCREAMING_SNAKE_CASE						(							self				:int     ):





        '''simple docstring'''




        A_    : str								=	self.feature_extraction_class(**self.feat_extract_dict     )

        with tempfile.TemporaryDirectory() as tmpdirname:
           A_    : List[Any]								=	feat_extract_first.save_pretrained(a_     )[0]
           check_json_file_has_correct_format(a_     )
           A_    : Tuple								=	self.feature_extraction_class.from_pretrained(a_     )

        A_    : Any								=	feat_extract_first.to_dict()
        A_    : Optional[Any]								=	feat_extract_second.to_dict()
        A_    : Tuple								=	feat_extract_first.mel_filters
        A_    : Optional[int]								=	feat_extract_second.mel_filters
        self.assertTrue(np.allclose(a_	,					a_     )     )
        self.assertEqual(a_	,					a_     )



     def   SCREAMING_SNAKE_CASE						(							self				:Union[str, Any]     ):





        '''simple docstring'''




        A_    : int								=	self.feature_extraction_class(**self.feat_extract_dict     )

        with tempfile.TemporaryDirectory() as tmpdirname:
           A_    : Union[str, Any]								=	os.path.join(a_	,					"feat_extract.json"     )
           feat_extract_first.to_json_file(a_     )
           A_    : str								=	self.feature_extraction_class.from_json_file(a_     )

        A_    : List[str]								=	feat_extract_first.to_dict()
        A_    : Union[str, Any]								=	feat_extract_second.to_dict()
        A_    : str								=	feat_extract_first.mel_filters
        A_    : List[str]								=	feat_extract_second.mel_filters
        self.assertTrue(np.allclose(a_	,					a_     )     )
        self.assertEqual(a_	,					a_     )



     def   SCREAMING_SNAKE_CASE						(							self				:List[str]     ):





        '''simple docstring'''




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

        # Test feature size
        A_    : str								=	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
        A_    : str								=	feature_extractor(speech_inputs[0]	,					return_tensors="np"     ).input_features
        A_    : Union[str, Any]								=	feature_extractor(np_speech_inputs[0]	,					return_tensors="np"     ).input_features
        self.assertTrue(np.allclose(a_	,					a_	,					atol=1e-3     )     )

        # Test batched
        A_    : str								=	feature_extractor(a_	,					return_tensors="np"     ).input_features
        A_    : List[Any]								=	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.
        A_    : Union[str, Any]								=	[floats_list((1, x)     )[0] for x in (800, 800, 800)]
        A_    : Optional[int]								=	np.asarray(a_     )
        A_    : str								=	feature_extractor(a_	,					return_tensors="np"     ).input_features
        A_    : List[str]								=	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
        A_    : Optional[Any]								=	[floats_list((1, x)     )[0] for x in range(200	,					(feature_extractor.n_samples + 500)	,					200     )]
        A_    : List[Any]								=	[np.asarray(a_     ) for speech_input in speech_inputs]

        A_    : Union[str, Any]								=	[x[: feature_extractor.n_samples] for x in speech_inputs]
        A_    : str								=	[np.asarray(a_     ) for speech_input in speech_inputs_truncated]

        A_    : Dict								=	feature_extractor(a_	,					return_tensors="np"     ).input_features
        A_    : List[str]								=	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   SCREAMING_SNAKE_CASE						(							self				:List[Any]     ):





        '''simple docstring'''




        import torch

        A_    : Optional[int]								=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()     )
        A_    : int								=	np.random.rand(100	,					32     ).astype(np.floataa     )
        A_    : int								=	np_speech_inputs.tolist()

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



     def   SCREAMING_SNAKE_CASE						(							self				:Tuple	,					snake_case				:Dict     ):





        '''simple docstring'''




        A_    : List[str]								=	load_dataset("hf-internal-testing/librispeech_asr_dummy"	,					"clean"	,					split="validation"     )
        # automatic decoding with librispeech
        A_    : Optional[int]								=	ds.sort("id"     ).select(range(a_     )     )[:num_samples]["""audio"""]

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



     def   SCREAMING_SNAKE_CASE						(							self				:Optional[int]     ):





        '''simple docstring'''




        A_    : List[str]								=	torch.tensor(
            [
                0.1193, -0.0946, -0.1098, -0.0196, 0.0225, -0.0690, -0.1736, 0.0951,
                0.0971, -0.0817, -0.0702, 0.0162, 0.0260, 0.0017, -0.0192, -0.1678,
                0.0709, -0.1867, -0.0655, -0.0274, -0.0234, -0.1884, -0.0516, -0.0554,
                -0.0274, -0.1425, -0.1423, 0.0837, 0.0377, -0.0854
            ]     )
        # fmt: on

        A_    : str								=	self._load_datasamples(1     )
        A_    : Any								=	WhisperFeatureExtractor()
        A_    : Any								=	feature_extractor(a_	,					return_tensors="pt"     ).input_features
        self.assertEqual(input_features.shape	,					(1, 80, 3_000)     )
        self.assertTrue(torch.allclose(input_features[0, 0, :30]	,					a_	,					atol=1e-4     )     )





     def   SCREAMING_SNAKE_CASE						(							self				:Union[str, Any]     ):





        '''simple docstring'''




        A_    : Any								=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()     )
        A_    : Optional[Any]								=	self._load_datasamples(1     )[0]
        A_    : Any								=	((audio - audio.min()) / (audio.max() - audio.min())) * 65_535  # Rescale to [0, 65535] to show issue
        A_    : Dict								=	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     )     )





'''simple docstring'''






import tempfile

import torch

from diffusers import IPNDMScheduler

from .test_schedulers import SchedulerCommonTest



class 							lowercase       (  _lowerCamelCase				):


 """simple docstring"""


 UpperCAmelCase							=  (IPNDMScheduler,)
 UpperCAmelCase							=  (("""num_inference_steps""", 50),)







 def _snake_case (							self  ,**a_							)		->   Optional[int]:
        _UpperCAmelCase  :						str			     = {"""num_train_timesteps""": 1_000}
        config.update(**a_							)
        return config







 def _snake_case (							self  ,a_=0  ,**a_							)		->   List[str]:
        _UpperCAmelCase  :						Any			     = dict(self.forward_default_kwargs							)
        _UpperCAmelCase  :						Any			     = kwargs.pop("""num_inference_steps"""  ,a_							)
        _UpperCAmelCase  :						Union[str, Any]			     = self.dummy_sample
        _UpperCAmelCase  :						Union[str, Any]			     = 0.1 * sample
        _UpperCAmelCase  :						List[str]			     = [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

        for scheduler_class in self.scheduler_classes:
               _UpperCAmelCase  :						str			     = self.get_scheduler_config(**a_							)
               _UpperCAmelCase  :						List[str]			     = scheduler_class(**a_							)
               scheduler.set_timesteps(a_							)
               # copy over dummy past residuals
               _UpperCAmelCase  :						str			     = dummy_past_residuals[:]

               if time_step is None:
                      _UpperCAmelCase  :						List[str]			     = scheduler.timesteps[len(scheduler.timesteps							) // 2]

               with tempfile.TemporaryDirectory() as tmpdirname:
                      scheduler.save_config(a_							)
                      _UpperCAmelCase  :						List[str]			     = scheduler_class.from_pretrained(a_							)
                      new_scheduler.set_timesteps(a_							)
                      # copy over dummy past residuals
                      _UpperCAmelCase  :						Tuple			     = dummy_past_residuals[:]

               _UpperCAmelCase  :						Dict			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample
               _UpperCAmelCase  :						Any			     = new_scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample

               assert torch.sum(torch.abs(output - new_output							)							) < 1E-5, "Scheduler outputs are not identical"

               _UpperCAmelCase  :						str			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample
               _UpperCAmelCase  :						Any			     = new_scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample

               assert torch.sum(torch.abs(output - new_output							)							) < 1E-5, "Scheduler outputs are not identical"







 def _snake_case (							self							)		->   Any:
        pass







 def _snake_case (							self  ,a_=0  ,**a_							)		->   Dict:
        _UpperCAmelCase  :						str			     = dict(self.forward_default_kwargs							)
        _UpperCAmelCase  :						Union[str, Any]			     = kwargs.pop("""num_inference_steps"""  ,a_							)
        _UpperCAmelCase  :						Optional[int]			     = self.dummy_sample
        _UpperCAmelCase  :						Tuple			     = 0.1 * sample
        _UpperCAmelCase  :						Optional[Any]			     = [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

        for scheduler_class in self.scheduler_classes:
               _UpperCAmelCase  :						Optional[Any]			     = self.get_scheduler_config()
               _UpperCAmelCase  :						Dict			     = scheduler_class(**a_							)
               scheduler.set_timesteps(a_							)

               # copy over dummy past residuals (must be after setting timesteps)
               _UpperCAmelCase  :						Any			     = dummy_past_residuals[:]

               if time_step is None:
                      _UpperCAmelCase  :						Optional[int]			     = scheduler.timesteps[len(scheduler.timesteps							) // 2]

               with tempfile.TemporaryDirectory() as tmpdirname:
                      scheduler.save_config(a_							)
                      _UpperCAmelCase  :						str			     = scheduler_class.from_pretrained(a_							)
                      # copy over dummy past residuals
                      new_scheduler.set_timesteps(a_							)

                      # copy over dummy past residual (must be after setting timesteps)
                      _UpperCAmelCase  :						Dict			     = dummy_past_residuals[:]

               _UpperCAmelCase  :						Tuple			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample
               _UpperCAmelCase  :						str			     = new_scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample

               assert torch.sum(torch.abs(output - new_output							)							) < 1E-5, "Scheduler outputs are not identical"

               _UpperCAmelCase  :						str			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample
               _UpperCAmelCase  :						str			     = new_scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample

               assert torch.sum(torch.abs(output - new_output							)							) < 1E-5, "Scheduler outputs are not identical"







 def _snake_case (							self  ,**a_							)		->   List[Any]:
        _UpperCAmelCase  :						int			     = self.scheduler_classes[0]
        _UpperCAmelCase  :						List[str]			     = self.get_scheduler_config(**a_							)
        _UpperCAmelCase  :						List[Any]			     = scheduler_class(**a_							)

        _UpperCAmelCase  :						List[Any]			     = 10
        _UpperCAmelCase  :						Dict			     = self.dummy_model()
        _UpperCAmelCase  :						Dict			     = self.dummy_sample_deter
        scheduler.set_timesteps(a_							)

        for i, t in enumerate(scheduler.timesteps							):
               _UpperCAmelCase  :						Optional[Any]			     = model(a_  ,a_							)
               _UpperCAmelCase  :						Tuple			     = scheduler.step(a_  ,a_  ,a_							).prev_sample

        for i, t in enumerate(scheduler.timesteps							):
               _UpperCAmelCase  :						List[str]			     = model(a_  ,a_							)
               _UpperCAmelCase  :						int			     = scheduler.step(a_  ,a_  ,a_							).prev_sample

        return sample







 def _snake_case (							self							)		->   Optional[Any]:
        _UpperCAmelCase  :						int			     = dict(self.forward_default_kwargs							)

        _UpperCAmelCase  :						Optional[Any]			     = kwargs.pop("""num_inference_steps"""  ,a_							)

        for scheduler_class in self.scheduler_classes:
               _UpperCAmelCase  :						int			     = self.get_scheduler_config()
               _UpperCAmelCase  :						List[str]			     = scheduler_class(**a_							)

               _UpperCAmelCase  :						Optional[int]			     = self.dummy_sample
               _UpperCAmelCase  :						Tuple			     = 0.1 * sample

               if num_inference_steps is not None and hasattr(a_  ,"""set_timesteps"""							):
                      scheduler.set_timesteps(a_							)
               elif num_inference_steps is not None and not hasattr(a_  ,"""set_timesteps"""							):
                      _UpperCAmelCase  :						Optional[Any]			     = num_inference_steps

               # copy over dummy past residuals (must be done after set_timesteps)
               _UpperCAmelCase  :						Union[str, Any]			     = [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
               _UpperCAmelCase  :						Optional[int]			     = dummy_past_residuals[:]

               _UpperCAmelCase  :						Tuple			     = scheduler.timesteps[5]
               _UpperCAmelCase  :						Optional[Any]			     = scheduler.timesteps[6]

               _UpperCAmelCase  :						Optional[Any]			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample
               _UpperCAmelCase  :						Optional[int]			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample

               self.assertEqual(output_a.shape  ,sample.shape							)
               self.assertEqual(output_a.shape  ,output_a.shape							)

               _UpperCAmelCase  :						Dict			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample
               _UpperCAmelCase  :						Any			     = scheduler.step(a_  ,a_  ,a_  ,**a_							).prev_sample

               self.assertEqual(output_a.shape  ,sample.shape							)
               self.assertEqual(output_a.shape  ,output_a.shape							)







 def _snake_case (							self							)		->   List[Any]:
        for timesteps in [100, 1_000]:
               self.check_over_configs(num_train_timesteps=a_  ,time_step=a_							)







 def _snake_case (							self							)		->   int:
        for t, num_inference_steps in zip([1, 5, 10]  ,[10, 50, 100]							):
               self.check_over_forward(num_inference_steps=a_  ,time_step=a_							)







 def _snake_case (							self							)		->   Optional[Any]:
        _UpperCAmelCase  :						Union[str, Any]			     = self.full_loop()
        _UpperCAmelCase  :						Any			     = torch.mean(torch.abs(a_							)							)

        assert abs(result_mean.item() - 2_540_529							) < 10



