



import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, logging, randn_tensor
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin


__UpperCAmelCase						  =     logging.get_logger(__name__)  # pylint: disable=invalid-name


@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->EulerDiscrete
class 	lowerCamelCase__ (		_a   ):
						_lowerCAmelCase								= 42
						_lowerCAmelCase								= None





def 						__lowerCamelCase       (      __magic_name__	:  Any     , __magic_name__	:  Any=0.999     , __magic_name__	:  str="cosine"     , ):
							if alpha_transform_type == "cosine":

														def alpha_bar_fn(__magic_name__	:  Optional[int]						):
																					return math.cos((t + 0.008) / 1.008 * math.pi / 2						) ** 2

							elif alpha_transform_type == "exp":

														def alpha_bar_fn(__magic_name__	:  Union[str, Any]						):
																					return math.exp(t * -12.0						)

							else:
														raise ValueError(F"Unsupported alpha_tranform_type: {alpha_transform_type}"						)

							a__:       Any											=[]
							for i in range(__magic_name__						):
														a__:       Any											=i / num_diffusion_timesteps
														a__:       int											=(i + 1) / num_diffusion_timesteps
														betas.append(min(1 - alpha_bar_fn(__magic_name__						) / alpha_bar_fn(__magic_name__						)     , __magic_name__						)						)
							return torch.tensor(__magic_name__     , dtype=torch.floataa						)







class 	lowerCamelCase__ (		_a				,					_a   ):
						_lowerCAmelCase								= [e.name for e in KarrasDiffusionSchedulers]
						_lowerCAmelCase								= 1


						@register_to_config
						def __init__( self						:  Optional[Any]      ,       _a						:  int = 1_0_0_0      ,       _a						:  float = 0.0_0_0_1      ,       _a						:  float = 0.0_2      ,       _a						:  str = "linear"      ,       _a						:  Optional[Union[np.ndarray, List[float]]] = None      ,       _a						:  str = "epsilon"      ,       _a						:  str = "linear"      ,       _a						:  Optional[bool] = False      ,       _a						:  str = "linspace"      ,       _a						:  int = 0      ,       ):
													if trained_betas is not None:
																				a__:       Optional[int]											=torch.tensor(_a      ,       dtype=torch.floataa			)
													elif beta_schedule == "linear":
																				a__:       List[str]											=torch.linspace(_a      ,       _a      ,       _a      ,       dtype=torch.floataa			)
													elif beta_schedule == "scaled_linear":
																				# this schedule is very specific to the latent diffusion model.
																				a__:       str											=(
																				    torch.linspace(beta_start**0.5      ,       beta_end**0.5      ,       _a      ,       dtype=torch.floataa			) ** 2
																				)
													elif beta_schedule == "squaredcos_cap_v2":
																				# Glide cosine schedule
																				a__:       Any											=betas_for_alpha_bar(_a			)
													else:
																				raise NotImplementedError(F"{beta_schedule} does is not implemented for {self.__class__}"			)

													a__:       Optional[Any]											=1.0 - self.betas
													a__:       Tuple											=torch.cumprod(self.alphas      ,       dim=0			)

													a__:       Any											=np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5			)
													a__:       Union[str, Any]											=np.concatenate([sigmas[::-1], [0.0]]			).astype(np.floataa			)
													a__:       int											=torch.from_numpy(_a			)

													# setable values
													a__:       Dict											=None
													a__:       str											=np.linspace(0      ,       num_train_timesteps - 1      ,       _a      ,       dtype=_a			)[::-1].copy()
													a__:       Tuple											=torch.from_numpy(_a			)
													a__:       List[Any]											=False
													a__:       List[str]											=use_karras_sigmas


						@property
						def   _lowerCamelCase			( self						:  Optional[int]			):
													# standard deviation of the initial noise distribution
													if self.config.timestep_spacing in ["linspace", "trailing"]:
																				return self.sigmas.max()

													return (self.sigmas.max() ** 2 + 1) ** 0.5


						def   _lowerCamelCase			( self						:  str      ,       _a						:  torch.FloatTensor      ,       _a						:  Union[float, torch.FloatTensor]			):
													if isinstance(_a      ,       torch.Tensor			):
																				a__:       List[Any]											=timestep.to(self.timesteps.device			)
													a__:       List[Any]											=(self.timesteps == timestep).nonzero().item()
													a__:       List[str]											=self.sigmas[step_index]

													a__:       List[Any]											=sample / ((sigma**2 + 1) ** 0.5)

													a__:       Any											=True
													return sample


						def   _lowerCamelCase			( self						:  str      ,       _a						:  int      ,       _a						:  Union[str, torch.device] = None			):
													a__:       List[Any]											=num_inference_steps

													# "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
													if self.config.timestep_spacing == "linspace":
																				a__:       List[str]											=np.linspace(0      ,       self.config.num_train_timesteps - 1      ,       _a      ,       dtype=_a			)[
																				    ::-1
																				].copy()
													elif self.config.timestep_spacing == "leading":
																				a__:       Dict											=self.config.num_train_timesteps // self.num_inference_steps
																				# creates integer timesteps by multiplying by ratio
																				# casting to int to avoid issues when num_inference_step is power of 3
																				a__:       Dict											=(np.arange(0      ,       _a			) * step_ratio).round()[::-1].copy().astype(_a			)
																				timesteps += self.config.steps_offset
													elif self.config.timestep_spacing == "trailing":
																				a__:       Optional[int]											=self.config.num_train_timesteps / self.num_inference_steps
																				# creates integer timesteps by multiplying by ratio
																				# casting to int to avoid issues when num_inference_step is power of 3
																				a__:       List[Any]											=(np.arange(self.config.num_train_timesteps      ,       0      ,       -step_ratio			)).round().copy().astype(_a			)
																				timesteps -= 1
													else:
																				raise ValueError(
																				    F"{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'."			)

													a__:       Tuple											=np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5			)
													a__:       Union[str, Any]											=np.log(_a			)

													if self.config.interpolation_type == "linear":
																				a__:       Union[str, Any]											=np.interp(_a      ,       np.arange(0      ,       len(_a			)			)      ,       _a			)
													elif self.config.interpolation_type == "log_linear":
																				a__:       List[Any]											=torch.linspace(np.log(sigmas[-1]			)      ,       np.log(sigmas[0]			)      ,       num_inference_steps + 1			).exp()
													else:
																				raise ValueError(
																				    F"{self.config.interpolation_type} is not implemented. Please specify interpolation_type to either"
																				    " 'linear' or 'log_linear'"			)

													if self.use_karras_sigmas:
																				a__:       List[str]											=self._convert_to_karras(in_sigmas=_a      ,       num_inference_steps=self.num_inference_steps			)
																				a__:       Optional[Any]											=np.array([self._sigma_to_t(_a      ,       _a			) for sigma in sigmas]			)

													a__:       List[str]											=np.concatenate([sigmas, [0.0]]			).astype(np.floataa			)
													a__:       str											=torch.from_numpy(_a			).to(device=_a			)
													if str(_a			).startswith("mps"			):
																				# mps does not support float64
																				a__:       Tuple											=torch.from_numpy(_a			).to(_a      ,       dtype=torch.floataa			)
													else:
																				a__:       List[Any]											=torch.from_numpy(_a			).to(device=_a			)


						def   _lowerCamelCase			( self						:  Optional[int]      ,       _a						:  List[str]      ,       _a						:  Optional[int]			):
													# get log sigma
													a__:       Optional[Any]											=np.log(_a			)

													# get distribution
													a__:       int											=log_sigma - log_sigmas[:, np.newaxis]

													# get sigmas range
													a__:       Optional[Any]											=np.cumsum((dists >= 0)      ,       axis=0			).argmax(axis=0			).clip(max=log_sigmas.shape[0] - 2			)
													a__:       Dict											=low_idx + 1

													a__:       Union[str, Any]											=log_sigmas[low_idx]
													a__:       Union[str, Any]											=log_sigmas[high_idx]

													# interpolate sigmas
													a__:       Optional[int]											=(low - log_sigma) / (low - high)
													a__:       str											=np.clip(_a      ,       0      ,       1			)

													# transform interpolation to time range
													a__:       List[Any]											=(1 - w) * low_idx + w * high_idx
													a__:       str											=t.reshape(sigma.shape			)
													return t


						def   _lowerCamelCase			( self						:  int      ,       _a						:  torch.FloatTensor      ,       _a						:  Tuple			):
													a__:       float											=in_sigmas[-1].item()
													a__:       float											=in_sigmas[0].item()

													a__:       Dict											=7.0  # 7.0 is the value used in the paper
													a__:       Dict											=np.linspace(0      ,       1      ,       _a			)
													a__:       Tuple											=sigma_min ** (1 / rho)
													a__:       Optional[int]											=sigma_max ** (1 / rho)
													a__:       Dict											=(max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
													return sigmas


						def   _lowerCamelCase			( self						:  List[str]      ,       _a						:  torch.FloatTensor      ,       _a						:  Union[float, torch.FloatTensor]      ,       _a						:  torch.FloatTensor      ,       _a						:  float = 0.0      ,       _a						:  float = 0.0      ,       _a						:  float = float("inf"			)      ,       _a						:  float = 1.0      ,       _a						:  Optional[torch.Generator] = None      ,       _a						:  bool = True      ,       ):

													if (
													    isinstance(_a      ,       _a			)
													    or isinstance(_a      ,       torch.IntTensor			)
													    or isinstance(_a      ,       torch.LongTensor			)
													):
																				raise ValueError(
																				    (
																				        "Passing integer indices (e.g. from `enumerate(timesteps)`) as timesteps to"
																				        " `EulerDiscreteScheduler.step()` is not supported. Make sure to pass"
																				        " one of the `scheduler.timesteps` as a timestep."
																				    )      ,       )

													if not self.is_scale_input_called:
																				logger.warning(
																				    "The `scale_model_input` function should be called before `step` to ensure correct denoising. "
																				    "See `StableDiffusionPipeline` for a usage example."			)

													if isinstance(_a      ,       torch.Tensor			):
																				a__:       int											=timestep.to(self.timesteps.device			)

													a__:       str											=(self.timesteps == timestep).nonzero().item()
													a__:       Optional[int]											=self.sigmas[step_index]

													a__:       Optional[Any]											=min(s_churn / (len(self.sigmas			) - 1)      ,       2**0.5 - 1			) if s_tmin <= sigma <= s_tmax else 0.0

													a__:       List[Any]											=randn_tensor(
													    model_output.shape      ,       dtype=model_output.dtype      ,       device=model_output.device      ,       generator=_a			)

													a__:       Optional[Any]											=noise * s_noise
													a__:       List[Any]											=sigma * (gamma + 1)

													if gamma > 0:
																				a__:       Dict											=sample + eps * (sigma_hat**2 - sigma**2) ** 0.5

													# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
													# NOTE: "original_sample" should not be an expected prediction_type but is left in for
													# backwards compatibility
													if self.config.prediction_type == "original_sample" or self.config.prediction_type == "sample":
																				a__:       str											=model_output
													elif self.config.prediction_type == "epsilon":
																				a__:       Union[str, Any]											=sample - sigma_hat * model_output
													elif self.config.prediction_type == "v_prediction":
																				# * c_out + input * c_skip
																				a__:       Optional[Any]											=model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
													else:
																				raise ValueError(
																				    F"prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"			)

													# 2. Convert to an ODE derivative
													a__:       List[Any]											=(sample - pred_original_sample) / sigma_hat

													a__:       Tuple											=self.sigmas[step_index + 1] - sigma_hat

													a__:       List[Any]											=sample + derivative * dt

													if not return_dict:
																				return (prev_sample,)

													return EulerDiscreteSchedulerOutput(prev_sample=_a      ,       pred_original_sample=_a			)


						def   _lowerCamelCase			( self						:  int      ,       _a						:  torch.FloatTensor      ,       _a						:  torch.FloatTensor      ,       _a						:  torch.FloatTensor      ,       ):
													# Make sure sigmas and timesteps have the same device and dtype as original_samples
													a__:       int											=self.sigmas.to(device=original_samples.device      ,       dtype=original_samples.dtype			)
													if original_samples.device.type == "mps" and torch.is_floating_point(_a			):
																				# mps does not support float64
																				a__:       str											=self.timesteps.to(original_samples.device      ,       dtype=torch.floataa			)
																				a__:       Optional[Any]											=timesteps.to(original_samples.device      ,       dtype=torch.floataa			)
													else:
																				a__:       Optional[Any]											=self.timesteps.to(original_samples.device			)
																				a__:       int											=timesteps.to(original_samples.device			)

													a__:       Any											=[(schedule_timesteps == t).nonzero().item() for t in timesteps]

													a__:       str											=sigmas[step_indices].flatten()
													while len(sigma.shape			) < len(original_samples.shape			):
																				a__:       Union[str, Any]											=sigma.unsqueeze(-1			)

													a__:       str											=original_samples + noise * sigma
													return noisy_samples







						def __len__( self						:  Dict			):
													return self.config.num_train_timesteps










import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
		from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

__UpperCAmelCase						  =     logging.get_logger(__name__)





def 						__lowerCamelCase       (      __magic_name__	:  bytes     , __magic_name__	:  int						):
							a__:       Optional[int]											=F"{sampling_rate}"
							a__:       Union[str, Any]											="1"
							a__:       List[str]											="f32le"
							a__:       Dict											=[
							    "ffmpeg",
							    "-i",
							    "pipe:0",
							    "-ac",
							    ac,
							    "-ar",
							    ar,
							    "-f",
							    format_for_conversion,
							    "-hide_banner",
							    "-loglevel",
							    "quiet",
							    "pipe:1",
							]

							try:
														a__:       List[Any]											=subprocess.Popen(__magic_name__     , stdin=subprocess.PIPE     , stdout=subprocess.PIPE						)
							except FileNotFoundError:
														raise ValueError("ffmpeg was not found but is required to load audio files from filename"						)
							a__:       int											=ffmpeg_process.communicate(__magic_name__						)
							a__:       List[str]											=output_stream[0]

							a__:       str											=np.frombuffer(__magic_name__     , np.floataa						)
							if audio.shape[0] == 0:
														raise ValueError("Malformed soundfile"						)
							return audio







@add_end_docstrings(_a   )
class 	lowerCamelCase__ (		_a   ):


						def __init__( self						:  Union[str, Any]      ,       *_a						:  Dict      ,       **_a						:  Tuple			):
													# Default, might be overriden by the model.config.
													a__:       List[Any]											=5
													super().__init__(*_a      ,       **_a			)

													if self.framework != "pt":
																				raise ValueError(F"The {self.__class__} is only available in PyTorch."			)

													self.check_model_type(_a			)


						def __call__( self						:  Tuple      ,       _a						:  Union[np.ndarray, bytes, str]      ,       **_a						:  Union[str, Any]      ,       ):
													return super().__call__(_a      ,       **_a			)


						def   _lowerCamelCase			( self						:  int      ,       _a						:  Tuple=None      ,       **_a						:  int			):
													# No parameters on this pipeline right now
													a__:       Optional[int]											={}
													if top_k is not None:
																				if top_k > self.model.config.num_labels:
																											a__:       Union[str, Any]											=self.model.config.num_labels
																				a__:       Union[str, Any]											=top_k
													return {}, {}, postprocess_params


						def   _lowerCamelCase			( self						:  Any      ,       _a						:  Union[str, Any]			):
													if isinstance(_a      ,       _a			):
																				if inputs.startswith("http://"			) or inputs.startswith("https://"			):
																											# We need to actually check for a real protocol, otherwise it's impossible to use a local file
																											# like http_huggingface_co.png
																											a__:       List[str]											=requests.get(_a			).content
																				else:
																											with open(_a      ,       "rb"			) as f:
																																		a__:       Dict											=f.read()

													if isinstance(_a      ,       _a			):
																				a__:       Tuple											=ffmpeg_read(_a      ,       self.feature_extractor.sampling_rate			)

													if isinstance(_a      ,       _a			):
																				# Accepting `"array"` which is the key defined in `datasets` for
																				# better integration
																				if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
																											raise ValueError(
																											    "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
																											    "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
																											    "containing the sampling_rate associated with that array"			)

																				a__:       Tuple											=inputs.pop("raw"      ,       _a			)
																				if _inputs is None:
																											# Remove path which will not be used from `datasets`.
																											inputs.pop("path"      ,       _a			)
																											a__:       int											=inputs.pop("array"      ,       _a			)
																				a__:       Optional[Any]											=inputs.pop("sampling_rate"			)
																				a__:       Optional[int]											=_inputs
																				if in_sampling_rate != self.feature_extractor.sampling_rate:
																											import torch

																											if is_torchaudio_available():
																																		from torchaudio import functional as F
																											else:
																																		raise ImportError(
																																		    "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
																																		    "The torchaudio package can be installed through: `pip install torchaudio`."			)

																											a__:       Union[str, Any]											=F.resample(
																											    torch.from_numpy(_a			)      ,       _a      ,       self.feature_extractor.sampling_rate			).numpy()

													if not isinstance(_a      ,       np.ndarray			):
																				raise ValueError("We expect a numpy ndarray as input"			)
													if len(inputs.shape			) != 1:
																				raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"			)

													a__:       Union[str, Any]											=self.feature_extractor(
													    _a      ,       sampling_rate=self.feature_extractor.sampling_rate      ,       return_tensors="pt"			)
													return processed


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  Tuple			):
													a__:       str											=self.model(**_a			)
													return model_outputs







						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  Tuple      ,       _a						:  Tuple=5			):
													a__:       Tuple											=model_outputs.logits[0].softmax(-1			)
													a__							,			a__:       Any											=probs.topk(_a			)

													a__:       Union[str, Any]											=scores.tolist()
													a__:       Dict											=ids.tolist()

													a__:       Optional[int]											=[{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(_a      ,       _a			)]

													return labels





