





"""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


SCREAMING_SNAKE_CASE_					:	Optional[int]							  =						logging.get_logger(__name__)



@add_end_docstrings(_a				)
class 					a						(       _a				):



    """simple docstring"""



    def __init__(     self: Optional[Any]    ,						**UpperCamelCase: List[Any]       ):






         """simple docstring"""


         super().__init__(**_A       )

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

    def __call__(     self: Union[str, Any]    ,						UpperCamelCase: str    ,						**UpperCamelCase: Dict       ):






         """simple docstring"""


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

    def 				UpperCamelCase	(     self: Optional[Any]    ,						**UpperCamelCase: Optional[int]       ):






         """simple docstring"""


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

         return preprocess_params, {}, {}

    def 				UpperCamelCase	(     self: Optional[int]    ,						UpperCamelCase: Optional[int]    ,						UpperCamelCase: Tuple=None    ,						UpperCamelCase: str="This is a sound of {}."       ):






         """simple docstring"""


         if isinstance(_A    ,						_A       ):
              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
                   A__										=  requests.get(_A       ).content
              else:
                   with open(_A    ,						"""rb"""       ) as f:
                        A__										=  f.read()

         if isinstance(_A    ,						_A       ):
              A__										=  ffmpeg_read(_A    ,						self.feature_extractor.sampling_rate       )

         if not isinstance(_A    ,						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"""       )

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

    def 				UpperCamelCase	(     self: Optional[int]    ,						UpperCamelCase: Any       ):






         """simple docstring"""


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

         A__										=  self.model(**_A    ,						**_A       )

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





    def 				UpperCamelCase	(     self: Optional[int]    ,						UpperCamelCase: Optional[Any]       ):






         """simple docstring"""


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

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

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









"""simple docstring"""



import unittest

import numpy as np
import torch

from diffusers import PNDMPipeline, PNDMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()



class 					a						(       unittest.TestCase				):



    """simple docstring"""



    @property
    def 				UpperCamelCase	(     self: Any       ):






         """simple docstring"""


         torch.manual_seed(0       )
         A__										=  UNetaDModel(
             block_out_channels=(32, 64)    ,						layers_per_block=2    ,						sample_size=32    ,						in_channels=3    ,						out_channels=3    ,						down_block_types=("""DownBlock2D""", """AttnDownBlock2D""")    ,						up_block_types=("""AttnUpBlock2D""", """UpBlock2D""")    ,						)
         return model





    def 				UpperCamelCase	(     self: str       ):






         """simple docstring"""


         A__										=  self.dummy_uncond_unet
         A__										=  PNDMScheduler()

         A__										=  PNDMPipeline(unet=UpperCamelCase    ,						scheduler=UpperCamelCase       )
         pndm.to(UpperCamelCase       )
         pndm.set_progress_bar_config(disable=UpperCamelCase       )

         A__										=  torch.manual_seed(0       )
         A__										=  pndm(generator=UpperCamelCase    ,						num_inference_steps=20    ,						output_type="""numpy"""       ).images

         A__										=  torch.manual_seed(0       )
         A__										=  pndm(generator=UpperCamelCase    ,						num_inference_steps=20    ,						output_type="""numpy"""    ,						return_dict=UpperCamelCase       )[0]

         A__										=  image[0, -3:, -3:, -1]
         A__										=  image_from_tuple[0, -3:, -3:, -1]

         assert image.shape == (1, 32, 32, 3)
         A__										=  np.array([1.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0]       )

         assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2
         assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1e-2



@slow
@require_torch
class 					a						(       unittest.TestCase				):



    """simple docstring"""



    def 				UpperCamelCase	(     self: Dict       ):






         """simple docstring"""


         A__										=  """google/ddpm-cifar10-32"""

         A__										=  UNetaDModel.from_pretrained(UpperCamelCase       )
         A__										=  PNDMScheduler()

         A__										=  PNDMPipeline(unet=UpperCamelCase    ,						scheduler=UpperCamelCase       )
         pndm.to(UpperCamelCase       )
         pndm.set_progress_bar_config(disable=UpperCamelCase       )
         A__										=  torch.manual_seed(0       )
         A__										=  pndm(generator=UpperCamelCase    ,						output_type="""numpy"""       ).images

         A__										=  image[0, -3:, -3:, -1]

         assert image.shape == (1, 32, 32, 3)
         A__										=  np.array([0.1_564, 0.14_645, 0.1_406, 0.14_715, 0.12_425, 0.14_045, 0.13_115, 0.12_175, 0.125]       )

         assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2






