






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

__UpperCamelCase      :       str      			=      random.Random()



def    __UpperCAmelCase	(				SCREAMING_SNAKE_CASE__:	int,					SCREAMING_SNAKE_CASE__:	List[str]=1.0,					SCREAMING_SNAKE_CASE__:	Optional[int]=None,					SCREAMING_SNAKE_CASE__:	Optional[int]=None    )  ->			List[Any]:

   """simple docstring"""
   if rng is None:
      __a				     =    global_rng

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




      def __init__(  self ,   lowerCamelCase ,   lowerCamelCase=7 ,   lowerCamelCase=400 ,   lowerCamelCase=2000 ,   lowerCamelCase=10 ,   lowerCamelCase=160 ,   lowerCamelCase=8 ,   lowerCamelCase=0.0 ,   lowerCamelCase=4000 ,   lowerCamelCase=False ,   lowerCamelCase=True ,   )							->List[Any]:



         '''simple docstring'''
         __a				     =    parent
         __a				     =    batch_size
         __a				     =    min_seq_length
         __a				     =    max_seq_length
         __a				     =    (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
         __a				     =    padding_value
         __a				     =    sampling_rate
         __a				     =    return_attention_mask
         __a				     =    do_normalize
         __a				     =    feature_size
         __a				     =    chunk_length
         __a				     =    hop_length




      def  __UpperCamelCase							(  self      )							->List[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  __UpperCamelCase							(  self ,   lowerCamelCase=False ,   lowerCamelCase=False      )							->Dict:



         '''simple docstring'''
         def _flatten(lowerCamelCase      ):
            return list(itertools.chain(*lowerCamelCase      )      )

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







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




      def  __UpperCamelCase							(  self      )							->str:



         '''simple docstring'''
         __a				     =    WhisperFeatureExtractionTester(self      )




      def  __UpperCamelCase							(  self      )							->Any:



         '''simple docstring'''
         __a				     =    self.feature_extraction_class(**self.feat_extract_dict      )

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

         __a				     =    feat_extract_first.to_dict()
         __a				     =    feat_extract_second.to_dict()
         __a				     =    feat_extract_first.mel_filters
         __a				     =    feat_extract_second.mel_filters
         self.assertTrue(np.allclose(lowerCamelCase ,   lowerCamelCase      )      )
         self.assertEqual(lowerCamelCase ,   lowerCamelCase      )




      def  __UpperCamelCase							(  self      )							->int:



         '''simple docstring'''
         __a				     =    self.feature_extraction_class(**self.feat_extract_dict      )

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

         __a				     =    feat_extract_first.to_dict()
         __a				     =    feat_extract_second.to_dict()
         __a				     =    feat_extract_first.mel_filters
         __a				     =    feat_extract_second.mel_filters
         self.assertTrue(np.allclose(lowerCamelCase ,   lowerCamelCase      )      )
         self.assertEqual(lowerCamelCase ,   lowerCamelCase      )




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



         '''simple docstring'''
         # Tests that all call wrap to encode_plus and batch_encode_plus
         __a				     =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
         # create three inputs of length 800, 1000, and 1200
         __a				     =    [floats_list((1, x)      )[0] for x in range(800 ,   1400 ,   200      )]
         __a				     =    [np.asarray(lowerCamelCase      ) for speech_input in speech_inputs]

         # Test feature size
         __a				     =    feature_extractor(lowerCamelCase ,   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				     =    feature_extractor(speech_inputs[0] ,   return_tensors='np'      ).input_features
         __a				     =    feature_extractor(np_speech_inputs[0] ,   return_tensors='np'      ).input_features
         self.assertTrue(np.allclose(lowerCamelCase ,   lowerCamelCase ,   atol=1e-3      )      )

         # Test batched
         __a				     =    feature_extractor(lowerCamelCase ,   return_tensors='np'      ).input_features
         __a				     =    feature_extractor(lowerCamelCase ,   return_tensors='np'      ).input_features
         for enc_seq_a, enc_seq_a in zip(lowerCamelCase ,   lowerCamelCase      ):
            self.assertTrue(np.allclose(lowerCamelCase ,   lowerCamelCase ,   atol=1e-3      )      )

         # Test 2-D numpy arrays are batched.
         __a				     =    [floats_list((1, x)      )[0] for x in (800, 800, 800)]
         __a				     =    np.asarray(lowerCamelCase      )
         __a				     =    feature_extractor(lowerCamelCase ,   return_tensors='np'      ).input_features
         __a				     =    feature_extractor(lowerCamelCase ,   return_tensors='np'      ).input_features
         for enc_seq_a, enc_seq_a in zip(lowerCamelCase ,   lowerCamelCase      ):
            self.assertTrue(np.allclose(lowerCamelCase ,   lowerCamelCase ,   atol=1e-3      )      )

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

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

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




      def  __UpperCamelCase							(  self      )							->List[str]:



         '''simple docstring'''
         import torch

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

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




      def  __UpperCamelCase							(  self ,   lowerCamelCase      )							->str:



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

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




      def  __UpperCamelCase							(  self      )							->Dict:



         '''simple docstring'''
         # fmt: off
         __a				     =    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				     =    self._load_datasamples(1      )
         __a				     =    WhisperFeatureExtractor()
         __a				     =    feature_extractor(lowerCamelCase ,   return_tensors='pt'      ).input_features
         self.assertEqual(input_features.shape ,   (1, 80, 3000)      )
         self.assertTrue(torch.allclose(input_features[0, 0, :30] ,   lowerCamelCase ,   atol=1e-4      )      )






      def  __UpperCamelCase							(  self      )							->Dict:



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

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







'''simple docstring'''



import inspect
from typing import List, Optional, Tuple, Union

import numpy as np
import PIL
import torch
import torch.utils.checkpoint

from ...models import UNetaDModel, VQModel
from ...schedulers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
)
from ...utils import PIL_INTERPOLATION, randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput



def    __UpperCAmelCase	(				SCREAMING_SNAKE_CASE__:	List[Any]    )  ->			Dict:

   """simple docstring"""
   __a     ,			__a				     =    image.size
   __a     ,			__a				     =    (x - x % 32 for x in (w, h))  # resize to integer multiple of 32
   __a				     =    image.resize((w, h),					resample=PIL_INTERPOLATION['lanczos']    )
   __a				     =    np.array(SCREAMING_SNAKE_CASE__    ).astype(np.floataa    ) / 2_5_5.0
   __a				     =    image[None].transpose(0,					3,					1,					2    )
   __a				     =    torch.from_numpy(SCREAMING_SNAKE_CASE__    )
   return 2.0 * image - 1.0



class 							__SCREAMING_SNAKE_CASE       (							_lowerCAmelCase     ):




      def __init__(  self ,   lowerCamelCase ,   lowerCamelCase ,   lowerCamelCase ,   )							->Dict:



         '''simple docstring'''
         super().__init__()
         self.register_modules(vqvae=lowerCamelCase ,   unet=lowerCamelCase ,   scheduler=lowerCamelCase      )






      @torch.no_grad()
      def __call__(  self ,   lowerCamelCase = None ,   lowerCamelCase = 1 ,   lowerCamelCase = 100 ,   lowerCamelCase = 0.0 ,   lowerCamelCase = None ,   lowerCamelCase = "pil" ,   lowerCamelCase = True ,   )							->Union[Tuple, ImagePipelineOutput]:



         '''simple docstring'''
         if isinstance(lowerCamelCase ,   PIL.Image.Image      ):
            __a				     =    1
         elif isinstance(lowerCamelCase ,   torch.Tensor      ):
            __a				     =    image.shape[0]
         else:
            raise ValueError(F"""`image` has to be of type `PIL.Image.Image` or `torch.Tensor` but is {type(lowerCamelCase      )}"""      )

         if isinstance(lowerCamelCase ,   PIL.Image.Image      ):
            __a				     =    preprocess(lowerCamelCase      )

         __a     ,			__a				     =    image.shape[-2:]

         # in_channels should be 6: 3 for latents, 3 for low resolution image
         __a				     =    (batch_size, self.unet.config.in_channels // 2, height, width)
         __a				     =    next(self.unet.parameters()      ).dtype

         __a				     =    randn_tensor(lowerCamelCase ,   generator=lowerCamelCase ,   device=self.device ,   dtype=lowerCamelCase      )

         __a				     =    image.to(device=self.device ,   dtype=lowerCamelCase      )

         # set timesteps and move to the correct device
         self.scheduler.set_timesteps(lowerCamelCase ,   device=self.device      )
         __a				     =    self.scheduler.timesteps

         # scale the initial noise by the standard deviation required by the scheduler
         __a				     =    latents * self.scheduler.init_noise_sigma

         # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature.
         # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
         # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
         # and should be between [0, 1]
         __a				     =    'eta' in set(inspect.signature(self.scheduler.step      ).parameters.keys()      )
         __a				     =    {}
         if accepts_eta:
            __a				     =    eta

         for t in self.progress_bar(lowerCamelCase      ):
            # concat latents and low resolution image in the channel dimension.
            __a				     =    torch.cat([latents, image] ,   dim=1      )
            __a				     =    self.scheduler.scale_model_input(lowerCamelCase ,   lowerCamelCase      )
            # predict the noise residual
            __a				     =    self.unet(lowerCamelCase ,   lowerCamelCase      ).sample
            # compute the previous noisy sample x_t -> x_t-1
            __a				     =    self.scheduler.step(lowerCamelCase ,   lowerCamelCase ,   lowerCamelCase ,   **lowerCamelCase      ).prev_sample

         # decode the image latents with the VQVAE
         __a				     =    self.vqvae.decode(lowerCamelCase      ).sample
         __a				     =    torch.clamp(lowerCamelCase ,   -1.0 ,   1.0      )
         __a				     =    image / 2 + 0.5
         __a				     =    image.cpu().permute(0 ,   2 ,   3 ,   1      ).numpy()

         if output_type == "pil":
            __a				     =    self.numpy_to_pil(lowerCamelCase      )

         if not return_dict:
            return (image,)

         return ImagePipelineOutput(images=lowerCamelCase      )