


from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


__UpperCamelCase     		=							logging.get_logger(__name__)







@add_end_docstrings(UpperCAmelCase   )
class 							lowerCamelCase__						(				UpperCAmelCase   ):






			"""simple docstring"""






			def __init__(				self   ,    **snake_case  ):

						'''simple docstring'''


						super().__init__(**snake_case  )

						if self.framework != "pt":
									raise ValueError(F'''The {self.__class__} is only available in PyTorch.'''  )
						# No specific FOR_XXX available yet




			def __call__(				self   ,    snake_case   ,    **snake_case  ):

						'''simple docstring'''


						return super().__call__(snake_case   ,    **snake_case  )




			def 					snake_case__					(				self   ,    **snake_case  ):

						'''simple docstring'''


						UpperCamelCase__       =							{}
						if "candidate_labels" in kwargs:
									UpperCamelCase__       =							kwargs["candidate_labels"]
						if "hypothesis_template" in kwargs:
									UpperCamelCase__       =							kwargs["hypothesis_template"]

						return preprocess_params, {}, {}




			def 					snake_case__					(				self   ,    snake_case   ,    snake_case=None   ,    snake_case="This is a sound of {}."  ):

						'''simple docstring'''


						if isinstance(snake_case   ,    snake_case  ):
									if audio.startswith("http://"  ) or audio.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
												UpperCamelCase__       =							requests.get(snake_case  ).content
									else:
												with open(snake_case   ,    "rb"  ) as f:
															UpperCamelCase__       =							f.read()

						if isinstance(snake_case   ,    snake_case  ):
									UpperCamelCase__       =							ffmpeg_read(snake_case   ,    self.feature_extractor.sampling_rate  )

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

						UpperCamelCase__       =							self.feature_extractor(
						    [audio]   ,    sampling_rate=self.feature_extractor.sampling_rate   ,    return_tensors="pt"  )
						UpperCamelCase__       =							candidate_labels
						UpperCamelCase__       =							[hypothesis_template.format(snake_case  ) for x in candidate_labels]
						UpperCamelCase__       =							self.tokenizer(snake_case   ,    return_tensors=self.framework   ,    padding=snake_case  )
						UpperCamelCase__       =							[text_inputs]
						return inputs




			def 					snake_case__					(				self   ,    snake_case  ):

						'''simple docstring'''


						UpperCamelCase__       =							model_inputs.pop("candidate_labels"  )
						UpperCamelCase__       =							model_inputs.pop("text_inputs"  )
						if isinstance(text_inputs[0]   ,    snake_case  ):
									UpperCamelCase__       =							text_inputs[0]
						else:
									# Batching case.
									UpperCamelCase__       =							text_inputs[0][0]

						UpperCamelCase__       =							self.model(**snake_case   ,    **snake_case  )

						UpperCamelCase__       =							{
						    "candidate_labels": candidate_labels,
						    "logits": outputs.logits_per_audio,
						}
						return model_outputs







			def 					snake_case__					(				self   ,    snake_case  ):

						'''simple docstring'''


						UpperCamelCase__       =							model_outputs.pop("candidate_labels"  )
						UpperCamelCase__       =							model_outputs["logits"][0]

						if self.framework == "pt":
									UpperCamelCase__       =							logits.softmax(dim=0  )
									UpperCamelCase__       =							probs.tolist()
						else:
									raise ValueError("`tf` framework not supported."  )

						UpperCamelCase__       =							[
						    {"score": score, "label": candidate_label}
						    for score, candidate_label in sorted(zip(snake_case   ,    snake_case  )   ,    key=lambda snake_case  : -x[0]  )
						]
						return result






# limitations under the License.


from typing import Optional, Tuple, Union

import torch

from diffusers import DiffusionPipeline, ImagePipelineOutput







class 							lowerCamelCase__						(				UpperCAmelCase   ):






			"""simple docstring"""






			def __init__(				self   ,    snake_case   ,    snake_case  ):

						'''simple docstring'''


						super().__init__()
						self.register_modules(unet=snake_case   ,    scheduler=snake_case  )







			@torch.no_grad()
			def __call__(				self   ,    snake_case = 1   ,    snake_case = None   ,    snake_case = 50   ,    snake_case = "pil"   ,    snake_case = True   ,    **snake_case   ,    ):

						'''simple docstring'''


						UpperCamelCase__       =							torch.randn(
						    (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)   ,    generator=snake_case   ,    )
						UpperCamelCase__       =							image.to(self.device  )

						# set step values
						self.scheduler.set_timesteps(snake_case  )

						for t in self.progress_bar(self.scheduler.timesteps  ):
									# 1. predict noise model_output
									UpperCamelCase__       =							self.unet(snake_case   ,    snake_case  ).sample

									# 2. predict previous mean of image x_t-1 and add variance depending on eta
									# eta corresponds to η in paper and should be between [0, 1]
									# do x_t -> x_t-1
									UpperCamelCase__       =							self.scheduler.step(snake_case   ,    snake_case   ,    snake_case  ).prev_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(snake_case  )

						if not return_dict:
									return (image,), "This is a local test"

						return ImagePipelineOutput(images=snake_case  ), "This is a local test"


