
"""simple docstring"""

import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)






def 							A						(    snake_case				:Optional[Any]    , snake_case				:List[str]    , snake_case				:str		)				->     Optional[int]:





 if isinstance(lowercase_    , torch.Tensor		):
  return image
 elif isinstance(lowercase_    , PIL.Image.Image		):
  __UpperCamelCase      =	[image]

 if isinstance(image[0]    , PIL.Image.Image		):
  __UpperCamelCase      =	[np.array(i.resize((w, h)    , resample=PIL_INTERPOLATION['lanczos']		)		)[None, :] for i in image]
  __UpperCamelCase      =	np.concatenate(lowercase_    , axis=0		)
  __UpperCamelCase      =	np.array(lowercase_		).astype(np.floataa		) / 2_5_5.0
  __UpperCamelCase      =	image.transpose(0    , 3    , 1    , 2		)
  __UpperCamelCase      =	2.0 * image - 1.0
  __UpperCamelCase      =	torch.from_numpy(lowercase_		)
 elif isinstance(image[0]    , torch.Tensor		):
  __UpperCamelCase      =	torch.cat(lowercase_    , dim=0		)
 return image






def 							A						(    snake_case				:int    , snake_case				:List[str]    , snake_case				:List[Any]    , snake_case				:Tuple=0.9_995		)				->     Any:





 if not isinstance(lowercase_    , np.ndarray		):
  __UpperCamelCase      =	True
  __UpperCamelCase      =	va.device
  __UpperCamelCase      =	va.cpu().numpy()
  __UpperCamelCase      =	va.cpu().numpy()

 __UpperCamelCase      =	np.sum(va * va / (np.linalg.norm(lowercase_		) * np.linalg.norm(lowercase_		))		)
 if np.abs(lowercase_		) > DOT_THRESHOLD:
  __UpperCamelCase      =	(1 - t) * va + t * va
 else:
  __UpperCamelCase      =	np.arccos(lowercase_		)
  __UpperCamelCase      =	np.sin(lowercase_		)
  __UpperCamelCase      =	theta_a * t
  __UpperCamelCase      =	np.sin(lowercase_		)
  __UpperCamelCase      =	np.sin(theta_a - theta_t		) / sin_theta_a
  __UpperCamelCase      =	sin_theta_t / sin_theta_a
  __UpperCamelCase      =	sa * va + sa * va

 if inputs_are_torch:
  __UpperCamelCase      =	torch.from_numpy(lowercase_		).to(lowercase_		)

 return va






def 							A						(    snake_case				:Optional[int]    , snake_case				:Any		)				->     Union[str, Any]:
 __UpperCamelCase      =	F.normalize(lowercase_    , dim=-1		)
 __UpperCamelCase      =	F.normalize(lowercase_    , dim=-1		)
 return (x - y).norm(dim=-1		).div(2		).arcsin().pow(2		).mul(2		)






def 							A						(    snake_case				:Tuple    , snake_case				:Optional[int]		)				->     Union[str, Any]:





 for param in model.parameters():
  __UpperCamelCase      =	value





class __lowerCAmelCase		(				__snake_case      ):





     def __init__(     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase=None     ,					__UpperCAmelCase=None     ,					__UpperCAmelCase=None     ,					):

      '''simple docstring'''




      super().__init__()
      self.register_modules(
          vae=__UpperCAmelCase     ,					text_encoder=__UpperCAmelCase     ,					clip_model=__UpperCAmelCase     ,					tokenizer=__UpperCAmelCase     ,					unet=__UpperCAmelCase     ,					scheduler=__UpperCAmelCase     ,					feature_extractor=__UpperCAmelCase     ,					coca_model=__UpperCAmelCase     ,					coca_tokenizer=__UpperCAmelCase     ,					coca_transform=__UpperCAmelCase     ,					)
      __UpperCamelCase      =	(
          feature_extractor.size
          if isinstance(feature_extractor.size     ,					__UpperCAmelCase				)
          else feature_extractor.size['shortest_edge']
      )
      __UpperCamelCase      =	transforms.Normalize(mean=feature_extractor.image_mean     ,					std=feature_extractor.image_std				)
      set_requires_grad(self.text_encoder     ,					__UpperCAmelCase				)
      set_requires_grad(self.clip_model     ,					__UpperCAmelCase				)





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase = "auto"				):

      '''simple docstring'''




      if slice_size == "auto":
       # half the attention head size is usually a good trade-off between
       # speed and memory
       __UpperCamelCase      =	self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(__UpperCAmelCase				)





     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      self.enable_attention_slicing(__UpperCAmelCase				)





     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      set_requires_grad(self.vae     ,					__UpperCAmelCase				)





     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      set_requires_grad(self.vae     ,					__UpperCAmelCase				)





     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      set_requires_grad(self.unet     ,					__UpperCAmelCase				)





     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      set_requires_grad(self.unet     ,					__UpperCAmelCase				)





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				):

      '''simple docstring'''




      __UpperCamelCase      =	min(int(num_inference_steps * strength				)     ,					__UpperCAmelCase				)

      __UpperCamelCase      =	max(num_inference_steps - init_timestep     ,					0				)
      __UpperCamelCase      =	self.scheduler.timesteps[t_start:]

      return timesteps, num_inference_steps - t_start





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase=None				):

      '''simple docstring'''




      if not isinstance(__UpperCAmelCase     ,					torch.Tensor				):
       raise ValueError(F'`image` has to be of type `torch.Tensor` but is {type(__UpperCAmelCase				)}'				)

      __UpperCamelCase      =	image.to(device=__UpperCAmelCase     ,					dtype=__UpperCAmelCase				)

      if isinstance(__UpperCAmelCase     ,					__UpperCAmelCase				):
       __UpperCamelCase      =	[
           self.vae.encode(image[i : i + 1]				).latent_dist.sample(generator[i]				) for i in range(__UpperCAmelCase				)
       ]
       __UpperCamelCase      =	torch.cat(__UpperCAmelCase     ,					dim=0				)
      else:
       __UpperCamelCase      =	self.vae.encode(__UpperCAmelCase				).latent_dist.sample(__UpperCAmelCase				)

      # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      __UpperCamelCase      =	0.1_8_2_1_5 * init_latents
      __UpperCamelCase      =	init_latents.repeat_interleave(__UpperCAmelCase     ,					dim=0				)

      __UpperCamelCase      =	randn_tensor(init_latents.shape     ,					generator=__UpperCAmelCase     ,					device=__UpperCAmelCase     ,					dtype=__UpperCAmelCase				)

      # get latents
      __UpperCamelCase      =	self.scheduler.add_noise(__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)
      __UpperCamelCase      =	init_latents

      return latents





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase				):

      '''simple docstring'''




      __UpperCamelCase      =	self.coca_transform(__UpperCAmelCase				).unsqueeze(0				)
      with torch.no_grad(), torch.cuda.amp.autocast():
       __UpperCamelCase      =	self.coca_model.generate(transformed_image.to(device=self.device     ,					dtype=self.coca_model.dtype				)				)
      __UpperCamelCase      =	self.coca_tokenizer.decode(generated[0].cpu().numpy()				)
      return generated.split('<end_of_text>'				)[0].replace('<start_of_text>'     ,					''				).rstrip(' .,'				)





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase				):

      '''simple docstring'''




      __UpperCamelCase      =	self.feature_extractor.preprocess(__UpperCAmelCase				)
      __UpperCamelCase      =	torch.from_numpy(clip_image_input['pixel_values'][0]				).unsqueeze(0				).to(self.device				).half()
      __UpperCamelCase      =	self.clip_model.get_image_features(__UpperCAmelCase				)
      __UpperCamelCase      =	image_embeddings_clip / image_embeddings_clip.norm(p=2     ,					dim=-1     ,					keepdim=__UpperCAmelCase				)
      __UpperCamelCase      =	image_embeddings_clip.repeat_interleave(__UpperCAmelCase     ,					dim=0				)
      return image_embeddings_clip





     @torch.enable_grad()
     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					):

      '''simple docstring'''




      __UpperCamelCase      =	latents.detach().requires_grad_()

      __UpperCamelCase      =	self.scheduler.scale_model_input(__UpperCAmelCase     ,					__UpperCAmelCase				)

      # predict the noise residual
      __UpperCamelCase      =	self.unet(__UpperCAmelCase     ,					__UpperCAmelCase     ,					encoder_hidden_states=__UpperCAmelCase				).sample

      if isinstance(self.scheduler     ,					(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)				):
       __UpperCamelCase      =	self.scheduler.alphas_cumprod[timestep]
       __UpperCamelCase      =	1 - alpha_prod_t
       # compute predicted original sample from predicted noise also called
       # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
       __UpperCamelCase      =	(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

       __UpperCamelCase      =	torch.sqrt(__UpperCAmelCase				)
       __UpperCamelCase      =	pred_original_sample * (fac) + latents * (1 - fac)
      elif isinstance(self.scheduler     ,					__UpperCAmelCase				):
       __UpperCamelCase      =	self.scheduler.sigmas[index]
       __UpperCamelCase      =	latents - sigma * noise_pred
      else:
       raise ValueError(F'scheduler type {type(self.scheduler				)} not supported'				)

      # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      __UpperCamelCase      =	1 / 0.1_8_2_1_5 * sample
      __UpperCamelCase      =	self.vae.decode(__UpperCAmelCase				).sample
      __UpperCamelCase      =	(image / 2 + 0.5).clamp(0     ,					1				)

      __UpperCamelCase      =	transforms.Resize(self.feature_extractor_size				)(__UpperCAmelCase				)
      __UpperCamelCase      =	self.normalize(__UpperCAmelCase				).to(latents.dtype				)

      __UpperCamelCase      =	self.clip_model.get_image_features(__UpperCAmelCase				)
      __UpperCamelCase      =	image_embeddings_clip / image_embeddings_clip.norm(p=2     ,					dim=-1     ,					keepdim=__UpperCAmelCase				)

      __UpperCamelCase      =	spherical_dist_loss(__UpperCAmelCase     ,					__UpperCAmelCase				).mean() * clip_guidance_scale

      __UpperCamelCase      =	-torch.autograd.grad(__UpperCAmelCase     ,					__UpperCAmelCase				)[0]

      if isinstance(self.scheduler     ,					__UpperCAmelCase				):
       __UpperCamelCase      =	latents.detach() + grads * (sigma**2)
       __UpperCamelCase      =	noise_pred_original
      else:
       __UpperCamelCase      =	noise_pred_original - torch.sqrt(__UpperCAmelCase				) * grads
      return noise_pred, latents







     @torch.no_grad()
     def __call__(     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = 512     ,					__UpperCAmelCase = 512     ,					__UpperCAmelCase = 0.6     ,					__UpperCAmelCase = 50     ,					__UpperCAmelCase = 7.5     ,					__UpperCAmelCase = 1     ,					__UpperCAmelCase = 0.0     ,					__UpperCAmelCase = 100     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = "pil"     ,					__UpperCAmelCase = True     ,					__UpperCAmelCase = 0.8     ,					__UpperCAmelCase = 0.1     ,					__UpperCAmelCase = 0.1     ,					):

      '''simple docstring'''




      if isinstance(__UpperCAmelCase     ,					__UpperCAmelCase				) and len(__UpperCAmelCase				) != batch_size:
       raise ValueError(F'You have passed {batch_size} batch_size, but only {len(__UpperCAmelCase				)} generators.'				)

      if height % 8 != 0 or width % 8 != 0:
       raise ValueError(F'`height` and `width` have to be divisible by 8 but are {height} and {width}.'				)

      if isinstance(__UpperCAmelCase     ,					torch.Generator				) and batch_size > 1:
       __UpperCamelCase      =	[generator] + [None] * (batch_size - 1)

      __UpperCamelCase      =	[
          ('model', self.coca_model is None),
          ('tokenizer', self.coca_tokenizer is None),
          ('transform', self.coca_transform is None),
      ]
      __UpperCamelCase      =	[x[0] for x in coca_is_none if x[1]]
      __UpperCamelCase      =	', '.join(__UpperCAmelCase				)
      # generate prompts with coca model if prompt is None
      if content_prompt is None:
       if len(__UpperCAmelCase				):
        raise ValueError(
            F'Content prompt is None and CoCa [{coca_is_none_str}] is None.'
            F'Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'				)
       __UpperCamelCase      =	self.get_image_description(__UpperCAmelCase				)
      if style_prompt is None:
       if len(__UpperCAmelCase				):
        raise ValueError(
            F'Style prompt is None and CoCa [{coca_is_none_str}] is None.'
            F' Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'				)
       __UpperCamelCase      =	self.get_image_description(__UpperCAmelCase				)

      # get prompt text embeddings for content and style
      __UpperCamelCase      =	self.tokenizer(
          __UpperCAmelCase     ,					padding='max_length'     ,					max_length=self.tokenizer.model_max_length     ,					truncation=__UpperCAmelCase     ,					return_tensors='pt'     ,					)
      __UpperCamelCase      =	self.text_encoder(content_text_input.input_ids.to(self.device				)				)[0]

      __UpperCamelCase      =	self.tokenizer(
          __UpperCAmelCase     ,					padding='max_length'     ,					max_length=self.tokenizer.model_max_length     ,					truncation=__UpperCAmelCase     ,					return_tensors='pt'     ,					)
      __UpperCamelCase      =	self.text_encoder(style_text_input.input_ids.to(self.device				)				)[0]

      __UpperCamelCase      =	slerp(__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)

      # duplicate text embeddings for each generation per prompt
      __UpperCamelCase      =	text_embeddings.repeat_interleave(__UpperCAmelCase     ,					dim=0				)

      # set timesteps
      __UpperCamelCase      =	'offset' in set(inspect.signature(self.scheduler.set_timesteps				).parameters.keys()				)
      __UpperCamelCase      =	{}
      if accepts_offset:
       __UpperCamelCase      =	1

      self.scheduler.set_timesteps(__UpperCAmelCase     ,					**__UpperCAmelCase				)
      # Some schedulers like PNDM have timesteps as arrays
      # It's more optimized to move all timesteps to correct device beforehand
      self.scheduler.timesteps.to(self.device				)

      __UpperCamelCase  ,     __UpperCamelCase      =	self.get_timesteps(__UpperCAmelCase     ,					__UpperCAmelCase     ,					self.device				)
      __UpperCamelCase      =	timesteps[:1].repeat(__UpperCAmelCase				)

      # Preprocess image
      __UpperCamelCase      =	preprocess(__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)
      __UpperCamelCase      =	self.prepare_latents(
          __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					text_embeddings.dtype     ,					self.device     ,					__UpperCAmelCase				)

      __UpperCamelCase      =	preprocess(__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)
      __UpperCamelCase      =	self.prepare_latents(
          __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					text_embeddings.dtype     ,					self.device     ,					__UpperCAmelCase				)

      __UpperCamelCase      =	slerp(__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)

      if clip_guidance_scale > 0:
       __UpperCamelCase      =	self.get_clip_image_embeddings(__UpperCAmelCase     ,					__UpperCAmelCase				)
       __UpperCamelCase      =	self.get_clip_image_embeddings(__UpperCAmelCase     ,					__UpperCAmelCase				)
       __UpperCamelCase      =	slerp(
           __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)

      # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
      # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
      # corresponds to doing no classifier free guidance.
      __UpperCamelCase      =	guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
       __UpperCamelCase      =	content_text_input.input_ids.shape[-1]
       __UpperCamelCase      =	self.tokenizer(['']     ,					padding='max_length'     ,					max_length=__UpperCAmelCase     ,					return_tensors='pt'				)
       __UpperCamelCase      =	self.text_encoder(uncond_input.input_ids.to(self.device				)				)[0]
       # duplicate unconditional embeddings for each generation per prompt
       __UpperCamelCase      =	uncond_embeddings.repeat_interleave(__UpperCAmelCase     ,					dim=0				)

       # For classifier free guidance, we need to do two forward passes.
       # Here we concatenate the unconditional and text embeddings into a single batch
       # to avoid doing two forward passes
       __UpperCamelCase      =	torch.cat([uncond_embeddings, text_embeddings]				)

      # get the initial random noise unless the user supplied it

      # Unlike in other pipelines, latents need to be generated in the target device
      # for 1-to-1 results reproducibility with the CompVis implementation.
      # However this currently doesn't work in `mps`.
      __UpperCamelCase      =	(batch_size, self.unet.config.in_channels, height // 8, width // 8)
      __UpperCamelCase      =	text_embeddings.dtype
      if latents is None:
       if self.device.type == "mps":
        # randn does not work reproducibly on mps
        __UpperCamelCase      =	torch.randn(__UpperCAmelCase     ,					generator=__UpperCAmelCase     ,					device='cpu'     ,					dtype=__UpperCAmelCase				).to(
            self.device				)
       else:
        __UpperCamelCase      =	torch.randn(__UpperCAmelCase     ,					generator=__UpperCAmelCase     ,					device=self.device     ,					dtype=__UpperCAmelCase				)
      else:
       if latents.shape != latents_shape:
        raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'				)
       __UpperCamelCase      =	latents.to(self.device				)

      # scale the initial noise by the standard deviation required by the scheduler
      __UpperCamelCase      =	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]
      __UpperCamelCase      =	'eta' in set(inspect.signature(self.scheduler.step				).parameters.keys()				)
      __UpperCamelCase      =	{}
      if accepts_eta:
       __UpperCamelCase      =	eta

      # check if the scheduler accepts generator
      __UpperCamelCase      =	'generator' in set(inspect.signature(self.scheduler.step				).parameters.keys()				)
      if accepts_generator:
       __UpperCamelCase      =	generator

      with self.progress_bar(total=__UpperCAmelCase				):
       for i, t in enumerate(__UpperCAmelCase				):
        # expand the latents if we are doing classifier free guidance
        __UpperCamelCase      =	torch.cat([latents] * 2				) if do_classifier_free_guidance else latents
        __UpperCamelCase      =	self.scheduler.scale_model_input(__UpperCAmelCase     ,					__UpperCAmelCase				)

        # predict the noise residual
        __UpperCamelCase      =	self.unet(__UpperCAmelCase     ,					__UpperCAmelCase     ,					encoder_hidden_states=__UpperCAmelCase				).sample

        # perform classifier free guidance
        if do_classifier_free_guidance:
         __UpperCamelCase  ,     __UpperCamelCase      =	noise_pred.chunk(2				)
         __UpperCamelCase      =	noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

        # perform clip guidance
        if clip_guidance_scale > 0:
         __UpperCamelCase      =	(
             text_embeddings.chunk(2				)[1] if do_classifier_free_guidance else text_embeddings
         )
         __UpperCamelCase  ,     __UpperCamelCase      =	self.cond_fn(
             __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					)

        # compute the previous noisy sample x_t -> x_t-1
        __UpperCamelCase      =	self.scheduler.step(__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					**__UpperCAmelCase				).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      __UpperCamelCase      =	1 / 0.1_8_2_1_5 * latents
      __UpperCamelCase      =	self.vae.decode(__UpperCAmelCase				).sample

      __UpperCamelCase      =	(image / 2 + 0.5).clamp(0     ,					1				)
      __UpperCamelCase      =	image.cpu().permute(0     ,					2     ,					3     ,					1				).numpy()

      if output_type == "pil":
       __UpperCamelCase      =	self.numpy_to_pil(__UpperCAmelCase				)

      if not return_dict:
       return (image, None)

      return StableDiffusionPipelineOutput(images=__UpperCAmelCase     ,					nsfw_content_detected=__UpperCAmelCase				)







"""simple docstring"""

import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


UpperCamelCase     :    int       =			logging.get_logger(__name__)

class __lowerCAmelCase		(				__SCREAMING_SNAKE_CASE      ):
     lowercase						       =						["input_values", "attention_mask"]





     def __init__(     self     ,					__UpperCAmelCase = 1     ,					__UpperCAmelCase = 1_6000     ,					__UpperCAmelCase = 0.0     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = 80     ,					__UpperCAmelCase = 16     ,					__UpperCAmelCase = 64     ,					__UpperCAmelCase = "hann_window"     ,					__UpperCAmelCase = 1.0     ,					__UpperCAmelCase = 80     ,					__UpperCAmelCase = 7600     ,					__UpperCAmelCase = 1E-10     ,					__UpperCAmelCase = 2     ,					__UpperCAmelCase = True     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      super().__init__(feature_size=__UpperCAmelCase     ,					sampling_rate=__UpperCAmelCase     ,					padding_value=__UpperCAmelCase     ,					**__UpperCAmelCase				)
      __UpperCamelCase      =	do_normalize
      __UpperCamelCase      =	return_attention_mask

      __UpperCamelCase      =	num_mel_bins
      __UpperCamelCase      =	hop_length
      __UpperCamelCase      =	win_length
      __UpperCamelCase      =	win_function
      __UpperCamelCase      =	frame_signal_scale
      __UpperCamelCase      =	fmin
      __UpperCamelCase      =	fmax
      __UpperCamelCase      =	mel_floor
      __UpperCamelCase      =	reduction_factor

      __UpperCamelCase      =	win_length * sampling_rate // 1000
      __UpperCamelCase      =	hop_length * sampling_rate // 1000
      __UpperCamelCase      =	optimal_fft_length(self.sample_size				)
      __UpperCamelCase      =	(self.n_fft // 2) + 1

      __UpperCamelCase      =	window_function(window_length=self.sample_size     ,					name=self.win_function     ,					periodic=__UpperCAmelCase				)

      __UpperCamelCase      =	mel_filter_bank(
          num_frequency_bins=self.n_freqs     ,					num_mel_filters=self.num_mel_bins     ,					min_frequency=self.fmin     ,					max_frequency=self.fmax     ,					sampling_rate=self.sampling_rate     ,					norm='slaney'     ,					mel_scale='slaney'     ,					)

      if frame_signal_scale != 1.0:
       warnings.warn(
           'The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers'     ,					__UpperCAmelCase     ,					)
      if reduction_factor != 2.0:
       warnings.warn(
           'The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'     ,					__UpperCAmelCase     ,					)





     @staticmethod
     # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
     def 						UpperCAmelCase (     __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase = 0.0				):

      '''simple docstring'''




      if attention_mask is not None:
       __UpperCamelCase      =	np.array(__UpperCAmelCase     ,					np.intaa				)
       __UpperCamelCase      =	[]

       for vector, length in zip(__UpperCAmelCase     ,					attention_mask.sum(-1				)				):
        __UpperCamelCase      =	(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7				)
        if length < normed_slice.shape[0]:
         __UpperCamelCase      =	padding_value

        normed_input_values.append(__UpperCAmelCase				)
      else:
       __UpperCamelCase      =	[(x - x.mean()) / np.sqrt(x.var() + 1E-7				) for x in input_values]

      return normed_input_values





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					):

      '''simple docstring'''




      __UpperCamelCase      =	spectrogram(
          __UpperCAmelCase     ,					window=self.window     ,					frame_length=self.sample_size     ,					hop_length=self.sample_stride     ,					fft_length=self.n_fft     ,					mel_filters=self.mel_filters     ,					mel_floor=self.mel_floor     ,					log_mel='log10'     ,					)
      return log_mel_spec.T





     def __call__(     self     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      if audio is None and audio_target is None:
       raise ValueError('You must provide either `audio` or `audio_target` values.'				)

      if sampling_rate is not None:
       if sampling_rate != self.sampling_rate:
        raise ValueError(
            F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
            F' {self.sampling_rate}. Please make sure that the provided audio input was sampled with'
            F' {self.sampling_rate} and not {sampling_rate}.'				)
      else:
       logger.warning(
           'It is strongly recommended to pass the ``sampling_rate`` argument to this function. '
           'Failing to do so can result in silent errors that might be hard to debug.'				)

      if audio is not None:
       __UpperCamelCase      =	self._process_audio(
           __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)
      else:
       __UpperCamelCase      =	None

      if audio_target is not None:
       __UpperCamelCase      =	self._process_audio(
           __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)

       if inputs is None:
        return inputs_target
       else:
        __UpperCamelCase      =	inputs_target['input_values']
        __UpperCamelCase      =	inputs_target.get('attention_mask'				)
        if decoder_attention_mask is not None:
         __UpperCamelCase      =	decoder_attention_mask

      return inputs





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      __UpperCamelCase      =	isinstance(__UpperCAmelCase     ,					np.ndarray				) and len(speech.shape				) > 1
      if is_batched_numpy and len(speech.shape				) > 2:
       raise ValueError(F'Only mono-channel audio is supported for input to {self}'				)
      __UpperCamelCase      =	is_batched_numpy or (
          isinstance(__UpperCAmelCase     ,					(list, tuple)				) and (isinstance(speech[0]     ,					(np.ndarray, tuple, list)				))
      )

      if is_batched:
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				) for speech in speech]
      elif not is_batched and not isinstance(__UpperCAmelCase     ,					np.ndarray				):
       __UpperCamelCase      =	np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				)
      elif isinstance(__UpperCAmelCase     ,					np.ndarray				) and speech.dtype is np.dtype(np.floataa				):
       __UpperCamelCase      =	speech.astype(np.floataa				)

      # always return batch
      if not is_batched:
       __UpperCamelCase      =	[speech]

      # needed to make pad() work on spectrogram inputs
      __UpperCamelCase      =	self.feature_size

      # convert into correct format for padding
      if is_target:
       __UpperCamelCase      =	[self._extract_mel_features(__UpperCAmelCase				) for waveform in speech]
       __UpperCamelCase      =	BatchFeature({'input_values': features}				)
       __UpperCamelCase      =	self.num_mel_bins
      else:
       __UpperCamelCase      =	BatchFeature({'input_values': speech}				)

      __UpperCamelCase      =	self.pad(
          __UpperCAmelCase     ,					padding=__UpperCAmelCase     ,					max_length=__UpperCAmelCase     ,					truncation=__UpperCAmelCase     ,					pad_to_multiple_of=__UpperCAmelCase     ,					return_attention_mask=__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)

      __UpperCamelCase      =	feature_size_hack

      # convert input values to correct format
      __UpperCamelCase      =	padded_inputs['input_values']
      if not isinstance(input_values[0]     ,					np.ndarray				):
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				) for array in input_values]
      elif (
          not isinstance(__UpperCAmelCase     ,					np.ndarray				)
          and isinstance(input_values[0]     ,					np.ndarray				)
          and input_values[0].dtype is np.dtype(np.floataa				)
      ):
       __UpperCamelCase      =	[array.astype(np.floataa				) for array in input_values]
      elif isinstance(__UpperCAmelCase     ,					np.ndarray				) and input_values.dtype is np.dtype(np.floataa				):
       __UpperCamelCase      =	input_values.astype(np.floataa				)

      # convert attention_mask to correct format
      __UpperCamelCase      =	padded_inputs.get('attention_mask'				)
      if attention_mask is not None:
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.intaa				) for array in attention_mask]

      # zero-mean and unit-variance normalization
      if not is_target and self.do_normalize:
       __UpperCamelCase      =	(
           attention_mask
           if self._get_padding_strategies(__UpperCAmelCase     ,					max_length=__UpperCAmelCase				) is not PaddingStrategy.DO_NOT_PAD
           else None
       )
       __UpperCamelCase      =	self.zero_mean_unit_var_norm(
           padded_inputs['input_values']     ,					attention_mask=__UpperCAmelCase     ,					padding_value=self.padding_value				)

      if return_tensors is not None:
       __UpperCamelCase      =	padded_inputs.convert_to_tensors(__UpperCAmelCase				)

      return padded_inputs







     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      __UpperCamelCase      =	super().to_dict()

      # Don't serialize these as they are derived from the other properties.
      __UpperCamelCase      =	['window', 'mel_filters', 'sample_size', 'sample_stride', 'n_fft', 'n_freqs']
      for name in names:
       if name in output:
        del output[name]

      return output



