



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

import torch
from transformers import CLIPTextModelWithProjection, CLIPTokenizer
from transformers.models.clip.modeling_clip import CLIPTextModelOutput

from diffusers.models import PriorTransformer
from diffusers.pipelines import DiffusionPipeline, StableDiffusionImageVariationPipeline
from diffusers.schedulers import UnCLIPScheduler
from diffusers.utils import logging, randn_tensor


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





def 						__lowerCamelCase       (      self	:  int     , __magic_name__	:  Any     , __magic_name__	:  Dict     , __magic_name__	:  Union[str, Any]     , __magic_name__	:  Union[str, Any]						):
							a__:       Dict											=image.to(device=__magic_name__						)
							a__:       Tuple											=image  # take image as image_embeddings
							a__:       int											=image_embeddings.unsqueeze(1						)

							# duplicate image embeddings for each generation per prompt, using mps friendly method
							a__							,			a__							,			a__:       Any											=image_embeddings.shape
							a__:       Union[str, Any]											=image_embeddings.repeat(1     , __magic_name__     , 1						)
							a__:       Optional[Any]											=image_embeddings.view(bs_embed * num_images_per_prompt     , __magic_name__     , -1						)

							if do_classifier_free_guidance:
														a__:       Tuple											=torch.zeros_like(__magic_name__						)

														# 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
														a__:       int											=torch.cat([uncond_embeddings, image_embeddings]						)

							return image_embeddings







class 	lowerCamelCase__ (		_a   ):


						def __init__( self						:  Any      ,       _a						:  PriorTransformer      ,       _a						:  CLIPTokenizer      ,       _a						:  CLIPTextModelWithProjection      ,       _a						:  UnCLIPScheduler      ,       _a						:  Optional[dict] = None      ,       ):
													super().__init__()

													a__:       List[Any]											={"image_encoder": None} if decoder_pipe_kwargs is None else decoder_pipe_kwargs

													a__:       Any											=decoder_pipe_kwargs.get("torch_dtype"      ,       _a			) or prior.dtype

													a__:       Optional[int]											=StableDiffusionImageVariationPipeline.from_pretrained(
													    "lambdalabs/sd-image-variations-diffusers"      ,       **_a			)

													# replace `_encode_image` method
													a__:       Union[str, Any]											=types.MethodType(_encode_image      ,       self.decoder_pipe			)

													self.register_modules(
													    prior=_a      ,       tokenizer=_a      ,       text_encoder=_a      ,       prior_scheduler=_a      ,       )


						def   _lowerCamelCase			( self						:  List[str]      ,       _a						:  Tuple      ,       _a						:  Any      ,       _a						:  Optional[int]      ,       _a						:  Any      ,       _a						:  Optional[Union[CLIPTextModelOutput, Tuple]] = None      ,       _a						:  Optional[torch.Tensor] = None      ,       ):
													if text_model_output is None:
																				a__:       Union[str, Any]											=len(_a			) if isinstance(_a      ,       _a			) else 1
																				# get prompt text embeddings
																				a__:       Any											=self.tokenizer(
																				    _a      ,       padding="max_length"      ,       max_length=self.tokenizer.model_max_length      ,       return_tensors="pt"      ,       )
																				a__:       Tuple											=text_inputs.input_ids
																				a__:       Optional[Any]											=text_inputs.attention_mask.bool().to(_a			)

																				if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
																											a__:       str											=self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]			)
																											logger.warning(
																											    "The following part of your input was truncated because CLIP can only handle sequences up to"
																											    F" {self.tokenizer.model_max_length} tokens: {removed_text}"			)
																											a__:       str											=text_input_ids[:, : self.tokenizer.model_max_length]

																				a__:       List[str]											=self.text_encoder(text_input_ids.to(_a			)			)

																				a__:       str											=text_encoder_output.text_embeds
																				a__:       Optional[Any]											=text_encoder_output.last_hidden_state

													else:
																				a__:       Optional[Any]											=text_model_output[0].shape[0]
																				a__							,			a__:       Union[str, Any]											=text_model_output[0], text_model_output[1]
																				a__:       Dict											=text_attention_mask

													a__:       Union[str, Any]											=text_embeddings.repeat_interleave(_a      ,       dim=0			)
													a__:       int											=text_encoder_hidden_states.repeat_interleave(_a      ,       dim=0			)
													a__:       Dict											=text_mask.repeat_interleave(_a      ,       dim=0			)

													if do_classifier_free_guidance:
																				a__:       List[str]											=[""] * batch_size

																				a__:       List[Any]											=self.tokenizer(
																				    _a      ,       padding="max_length"      ,       max_length=self.tokenizer.model_max_length      ,       truncation=_a      ,       return_tensors="pt"      ,       )
																				a__:       Any											=uncond_input.attention_mask.bool().to(_a			)
																				a__:       Optional[Any]											=self.text_encoder(uncond_input.input_ids.to(_a			)			)

																				a__:       Optional[int]											=uncond_embeddings_text_encoder_output.text_embeds
																				a__:       int											=uncond_embeddings_text_encoder_output.last_hidden_state

																				# duplicate unconditional embeddings for each generation per prompt, using mps friendly method

																				a__:       List[str]											=uncond_embeddings.shape[1]
																				a__:       Union[str, Any]											=uncond_embeddings.repeat(1      ,       _a			)
																				a__:       str											=uncond_embeddings.view(batch_size * num_images_per_prompt      ,       _a			)

																				a__:       Optional[Any]											=uncond_text_encoder_hidden_states.shape[1]
																				a__:       Union[str, Any]											=uncond_text_encoder_hidden_states.repeat(1      ,       _a      ,       1			)
																				a__:       Dict											=uncond_text_encoder_hidden_states.view(
																				    batch_size * num_images_per_prompt      ,       _a      ,       -1			)
																				a__:       Optional[Any]											=uncond_text_mask.repeat_interleave(_a      ,       dim=0			)

																				# done duplicates

																				# 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
																				a__:       int											=torch.cat([uncond_embeddings, text_embeddings]			)
																				a__:       int											=torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]			)

																				a__:       Any											=torch.cat([uncond_text_mask, text_mask]			)

													return text_embeddings, text_encoder_hidden_states, text_mask


						@property
						def   _lowerCamelCase			( self						:  Any			):
													if self.device != torch.device("meta"			) or not hasattr(self.prior      ,       "_hf_hook"			):
																				return self.device
													for module in self.prior.modules():
																				if (
																				    hasattr(_a      ,       "_hf_hook"			)
																				    and hasattr(module._hf_hook      ,       "execution_device"			)
																				    and module._hf_hook.execution_device is not None
																				):
																											return torch.device(module._hf_hook.execution_device			)
													return self.device


						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  Optional[Any]      ,       _a						:  str      ,       _a						:  List[str]      ,       _a						:  Tuple      ,       _a						:  int      ,       _a						:  Union[str, Any]			):
													if latents is None:
																				a__:       Any											=randn_tensor(_a      ,       generator=_a      ,       device=_a      ,       dtype=_a			)
													else:
																				if latents.shape != shape:
																											raise ValueError(F"Unexpected latents shape, got {latents.shape}, expected {shape}"			)
																				a__:       Optional[Any]											=latents.to(_a			)

													a__:       Dict											=latents * scheduler.init_noise_sigma
													return latents


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  Optional[Union[str, torch.device]] = None			):
													self.decoder_pipe.to(_a			)
													super().to(_a			)







						@torch.no_grad()
						def __call__( self						:  List[Any]      ,       _a						:  Optional[Union[str, List[str]]] = None      ,       _a						:  Optional[int] = None      ,       _a						:  Optional[int] = None      ,       _a						:  int = 1      ,       _a						:  int = 2_5      ,       _a						:  Optional[torch.Generator] = None      ,       _a						:  Optional[torch.FloatTensor] = None      ,       _a						:  Optional[Union[CLIPTextModelOutput, Tuple]] = None      ,       _a						:  Optional[torch.Tensor] = None      ,       _a						:  float = 4.0      ,       _a						:  float = 8.0      ,       _a						:  int = 5_0      ,       _a						:  Optional[int] = 1      ,       _a						:  float = 0.0      ,       _a						:  Optional[str] = "pil"      ,       _a						:  bool = True      ,       ):
													if prompt is not None:
																				if isinstance(_a      ,       _a			):
																											a__:       str											=1
																				elif isinstance(_a      ,       _a			):
																											a__:       str											=len(_a			)
																				else:
																											raise ValueError(F"`prompt` has to be of type `str` or `list` but is {type(_a			)}"			)
													else:
																				a__:       Optional[int]											=text_model_output[0].shape[0]

													a__:       Optional[Any]											=self._execution_device

													a__:       int											=batch_size * num_images_per_prompt

													a__:       str											=prior_guidance_scale > 1.0 or decoder_guidance_scale > 1.0

													a__							,			a__							,			a__:       List[str]											=self._encode_prompt(
													    _a      ,       _a      ,       _a      ,       _a      ,       _a      ,       _a			)

													# prior

													self.prior_scheduler.set_timesteps(_a      ,       device=_a			)
													a__:       Optional[int]											=self.prior_scheduler.timesteps

													a__:       Optional[int]											=self.prior.config.embedding_dim

													a__:       Any											=self.prepare_latents(
													    (batch_size, embedding_dim)      ,       text_embeddings.dtype      ,       _a      ,       _a      ,       _a      ,       self.prior_scheduler      ,       )

													for i, t in enumerate(self.progress_bar(_a			)			):
																				# expand the latents if we are doing classifier free guidance
																				a__:       List[str]											=torch.cat([prior_latents] * 2			) if do_classifier_free_guidance else prior_latents

																				a__:       Optional[int]											=self.prior(
																				    _a      ,       timestep=_a      ,       proj_embedding=_a      ,       encoder_hidden_states=_a      ,       attention_mask=_a      ,       ).predicted_image_embedding

																				if do_classifier_free_guidance:
																											a__							,			a__:       str											=predicted_image_embedding.chunk(2			)
																											a__:       Tuple											=predicted_image_embedding_uncond + prior_guidance_scale * (
																											    predicted_image_embedding_text - predicted_image_embedding_uncond
																											)

																				if i + 1 == prior_timesteps_tensor.shape[0]:
																											a__:       Tuple											=None
																				else:
																											a__:       Optional[Any]											=prior_timesteps_tensor[i + 1]

																				a__:       Any											=self.prior_scheduler.step(
																				    _a      ,       timestep=_a      ,       sample=_a      ,       generator=_a      ,       prev_timestep=_a      ,       ).prev_sample

													a__:       List[str]											=self.prior.post_process_latents(_a			)

													a__:       Optional[int]											=prior_latents

													a__:       Any											=self.decoder_pipe(
													    image=_a      ,       height=_a      ,       width=_a      ,       num_inference_steps=_a      ,       guidance_scale=_a      ,       generator=_a      ,       output_type=_a      ,       return_dict=_a      ,       num_images_per_prompt=_a      ,       eta=_a      ,       )
													return output










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





