
'''simple docstring'''






import random
import unittest

import torch

from diffusers import IFInpaintingSuperResolutionPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin

@skip_mps
class 			UpperCAmelCase__    (							A    ,						A    ,						unittest.TestCase			):
    lowerCAmelCase_					     =		IFInpaintingSuperResolutionPipeline
    lowerCAmelCase_					     =		TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {'width', 'height'}
    lowerCAmelCase_					     =		TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS.union({'original_image'}			)
    lowerCAmelCase_					     =		PipelineTesterMixin.required_optional_params - {'latents'}



    def 	lowerCamelCase_	(						self   :							List[str]  ):
          return self._get_superresolution_dummy_components()



    def 	lowerCamelCase_	(						self   :							str,__A   :							List[str],__A   :							List[str]=0  ):
          if str(__A  ).startswith("mps"  ):
                _lowerCamelCase					: List[str]		    = torch.manual_seed(__A  )
          else:
                _lowerCamelCase					: Optional[int]		    = torch.Generator(device=__A  ).manual_seed(__A  )

          _lowerCamelCase					: List[Any]		    = floats_tensor((1, 3, 1_6, 1_6),rng=random.Random(__A  )  ).to(__A  )
          _lowerCamelCase					: Any		    = floats_tensor((1, 3, 3_2, 3_2),rng=random.Random(__A  )  ).to(__A  )
          _lowerCamelCase					: Tuple		    = floats_tensor((1, 3, 3_2, 3_2),rng=random.Random(__A  )  ).to(__A  )

          _lowerCamelCase					: Dict		    = {
              "prompt": "A painting of a squirrel eating a burger",
              "image": image,
              "original_image": original_image,
              "mask_image": mask_image,
              "generator": generator,
              "num_inference_steps": 2,
              "output_type": "numpy",
          }

          return inputs



    @unittest.skipIf(
        torch_device != "cuda" or not is_xformers_available(),reason="XFormers attention is only available with CUDA and `xformers` installed",)
    def 	lowerCamelCase_	(						self   :							Optional[int]  ):
          self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3  )



    def 	lowerCamelCase_	(						self   :							Dict  ):
          self._test_save_load_optional_components()



    @unittest.skipIf(torch_device != "cuda",reason="float16 requires CUDA"  )
    def 	lowerCamelCase_	(						self   :							Optional[Any]  ):
          # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
          super().test_save_load_floataa(expected_max_diff=1e-1  )



    def 	lowerCamelCase_	(						self   :							Any  ):
          self._test_attention_slicing_forward_pass(expected_max_diff=1e-2  )



    def 	lowerCamelCase_	(						self   :							Dict  ):
          self._test_save_load_local()



    def 	lowerCamelCase_	(						self   :							Any  ):
          self._test_inference_batch_single_identical(
              expected_max_diff=1e-2,)




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


snake_case_					:      List[Any]    			=    logging.get_logger(__name__)






@add_end_docstrings(lowercase  )
class 				lowercase__      (				lowercase  ):





	def __init__(					self   :							Tuple					,**lowerCamelCase__   :							Union[str, Any]					):







							'''simple docstring'''
							super().__init__(**lowerCamelCase__					)

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





	def __call__(					self   :							Dict					,lowerCamelCase__   :							Union[np.ndarray, bytes, str]					,**lowerCamelCase__   :							Any					):







							'''simple docstring'''
							return super().__call__(lowerCamelCase__					,**lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							int					,**lowerCamelCase__   :							Optional[int]					):







							'''simple docstring'''
							_UpperCamelCase   :						List[Any]												=  {}
							if "candidate_labels" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['candidate_labels']
							if "hypothesis_template" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['hypothesis_template']

							return preprocess_params, {}, {}





	def       UpperCamelCase_    (					self   :							Any					,lowerCamelCase__   :							List[str]					,lowerCamelCase__   :							str=None					,lowerCamelCase__   :							List[Any]="This is a sound of {}."					):







							'''simple docstring'''
							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													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   :						Optional[Any]												=  requests.get(lowerCamelCase__					).content
													else:
																			with open(lowerCamelCase__					,'rb'					) as f:
																									_UpperCamelCase   :						List[str]												=  f.read()

							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													_UpperCamelCase   :						int												=  ffmpeg_read(lowerCamelCase__					,self.feature_extractor.sampling_rate					)

							if not isinstance(lowerCamelCase__					,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   :						Dict												=  self.feature_extractor(
							    [audio]					,sampling_rate=self.feature_extractor.sampling_rate					,return_tensors='pt'					)
							_UpperCamelCase   :						List[Any]												=  candidate_labels
							_UpperCamelCase   :						Tuple												=  [hypothesis_template.format(lowerCamelCase__					) for x in candidate_labels]
							_UpperCamelCase   :						Dict												=  self.tokenizer(lowerCamelCase__					,return_tensors=self.framework					,padding=lowerCamelCase__					)
							_UpperCamelCase   :						Optional[Any]												=  [text_inputs]
							return inputs





	def       UpperCamelCase_    (					self   :							str					,lowerCamelCase__   :							Tuple					):







							'''simple docstring'''
							_UpperCamelCase   :						Any												=  model_inputs.pop('candidate_labels'					)
							_UpperCamelCase   :						str												=  model_inputs.pop('text_inputs'					)
							if isinstance(text_inputs[0]					,lowerCamelCase__					):
													_UpperCamelCase   :						Union[str, Any]												=  text_inputs[0]
							else:
													# Batching case.
													_UpperCamelCase   :						int												=  text_inputs[0][0]

							_UpperCamelCase   :						List[str]												=  self.model(**lowerCamelCase__					,**lowerCamelCase__					)

							_UpperCamelCase   :						Optional[Any]												=  {
							    'candidate_labels': candidate_labels,
							    'logits': outputs.logits_per_audio,
							}
							return model_outputs





	def       UpperCamelCase_    (					self   :							int					,lowerCamelCase__   :							Optional[Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[Any]												=  model_outputs.pop('candidate_labels'					)
							_UpperCamelCase   :						int												=  model_outputs['logits'][0]

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

							_UpperCamelCase   :						Union[str, Any]												=  [
							    {'score': score, 'label': candidate_label}
							    for score, candidate_label in sorted(zip(lowerCamelCase__					,lowerCamelCase__					)					,key=lambda lowerCamelCase__					: -x[0]					)
							]
							return result


