




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

lowerCamelCase_  						=       logging.get_logger(__name__)



def      UpperCAmelCase_   (     __UpperCamelCase,    __UpperCamelCase     ):
    SCREAMING_SNAKE_CASE__       =f"""{sampling_rate}"""
    SCREAMING_SNAKE_CASE__       ="""1"""
    SCREAMING_SNAKE_CASE__       ="""f32le"""
    SCREAMING_SNAKE_CASE__       =[
        """ffmpeg""",
        """-i""",
        """pipe:0""",
        """-ac""",
        ac,
        """-ar""",
        ar,
        """-f""",
        format_for_conversion,
        """-hide_banner""",
        """-loglevel""",
        """quiet""",
        """pipe:1""",
    ]

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

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




@add_end_docstrings(__lowerCamelCase					)
class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     def __init__(     self      :			Any    ,*_UpperCamelCase      :			Dict    ,**_UpperCamelCase      :			Optional[Any]						)		->       Any:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =5
         super().__init__(*_UpperCamelCase    ,**_UpperCamelCase						)

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

         self.check_model_type(_UpperCamelCase						)




     def __call__(     self      :			str    ,_UpperCamelCase      :			Union[np.ndarray, bytes, str]    ,**_UpperCamelCase      :			Any    ,)		->       int:


         '''simple docstring'''





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




     def   __A    (     self      :			str    ,_UpperCamelCase      :			Tuple=None    ,**_UpperCamelCase      :			Optional[int]						)		->       Any:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       ={}
         if top_k is not None:
             if top_k > self.model.config.num_labels:
                 SCREAMING_SNAKE_CASE__       =self.model.config.num_labels
             SCREAMING_SNAKE_CASE__       =top_k
         return {}, {}, postprocess_params




     def   __A    (     self      :			Any    ,_UpperCamelCase      :			List[str]						)		->       List[Any]:


         '''simple docstring'''





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

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

         if isinstance(_UpperCamelCase    ,_UpperCamelCase						):
             # 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"""						)

             SCREAMING_SNAKE_CASE__       =inputs.pop("""raw"""    ,_UpperCamelCase						)
             if _inputs is None:
                 # Remove path which will not be used from `datasets`.
                 inputs.pop("""path"""    ,_UpperCamelCase						)
                 SCREAMING_SNAKE_CASE__       =inputs.pop("""array"""    ,_UpperCamelCase						)
             SCREAMING_SNAKE_CASE__       =inputs.pop("""sampling_rate"""						)
             SCREAMING_SNAKE_CASE__       =_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`."""						)

                 SCREAMING_SNAKE_CASE__       =F.resample(
                     torch.from_numpy(_UpperCamelCase						)    ,_UpperCamelCase    ,self.feature_extractor.sampling_rate						).numpy()

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

         SCREAMING_SNAKE_CASE__       =self.feature_extractor(
             _UpperCamelCase    ,sampling_rate=self.feature_extractor.sampling_rate    ,return_tensors="""pt"""						)
         return processed




     def   __A    (     self      :			Union[str, Any]    ,_UpperCamelCase      :			List[Any]						)		->       str:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =self.model(**_UpperCamelCase						)
         return model_outputs




     def   __A    (     self      :			List[Any]    ,_UpperCamelCase      :			str    ,_UpperCamelCase      :			List[Any]=5						)		->       Optional[int]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =model_outputs.logits[0].softmax(-1						)
         SCREAMING_SNAKE_CASE__      ,		SCREAMING_SNAKE_CASE__       =probs.topk(_UpperCamelCase						)

         SCREAMING_SNAKE_CASE__       =scores.tolist()
         SCREAMING_SNAKE_CASE__       =ids.tolist()

         SCREAMING_SNAKE_CASE__       =[{"""score""": score, """label""": self.model.config.idalabel[_id]} for score, _id in zip(_UpperCamelCase    ,_UpperCamelCase						)]

         return labels












import tempfile
import unittest

from diffusers import (
    DDIMScheduler,
    DDPMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    PNDMScheduler,
    logging,
)
from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.utils.testing_utils import CaptureLogger
class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			int    ,_UpperCamelCase      :			int=2    ,_UpperCamelCase      :			Optional[Any]=5    ,_UpperCamelCase      :			List[str]=(2, 5)    ,_UpperCamelCase      :			List[Any]="for diffusion"    ,_UpperCamelCase      :			Optional[int]=[1, 3]    ,)		->       Any:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[str]     =  "config.json"




     @register_to_config
     def __init__(     self      :			List[Any]    ,_UpperCamelCase      :			str=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			Optional[Any]=(2, 5)    ,_UpperCamelCase      :			Union[str, Any]="for diffusion"    ,_UpperCamelCase      :			Tuple=[1, 3]    ,)		->       List[str]:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   Optional[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			str    ,_UpperCamelCase      :			Optional[int]=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			List[Any]=(2, 5)    ,_UpperCamelCase      :			Dict="for diffusion"    ,_UpperCamelCase      :			Union[str, Any]=[1, 3]    ,_UpperCamelCase      :			Dict=[1, 3]    ,)		->       Any:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			Union[str, Any]    ,_UpperCamelCase      :			str=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			List[Any]=(2, 5)    ,_UpperCamelCase      :			int="for diffusion"    ,_UpperCamelCase      :			Any=[1, 5]    ,_UpperCamelCase      :			Optional[Any]=[5, 4]    ,)		->       List[Any]:


         '''simple docstring'''





         pass






class 				__a (			unittest.TestCase					):


     """simple docstring"""


     def   __A    (     self      :			Dict						)		->       Optional[Any]:


         '''simple docstring'''





         with self.assertRaises(_UpperCamelCase						):
             ConfigMixin.load_config("""dummy_path"""						)




     def   __A    (     self      :			int						)		->       List[str]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # init ignore private arguments
         SCREAMING_SNAKE_CASE__       =SampleObject(_name_or_path="""lalala"""						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # can override default
         SCREAMING_SNAKE_CASE__       =SampleObject(c=6						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == 6
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # can use positional arguments.
         SCREAMING_SNAKE_CASE__       =SampleObject(1    ,c=6						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 1
         assert config["b"] == 5
         assert config["c"] == 6
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]




     def   __A    (     self      :			List[Any]						)		->       Tuple:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()
         SCREAMING_SNAKE_CASE__       =obj.config

         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         with tempfile.TemporaryDirectory() as tmpdirname:
             obj.save_config(_UpperCamelCase						)
             SCREAMING_SNAKE_CASE__       =SampleObject.from_config(SampleObject.load_config(_UpperCamelCase						)						)
             SCREAMING_SNAKE_CASE__       =new_obj.config

         # unfreeze configs
         SCREAMING_SNAKE_CASE__       =dict(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =dict(_UpperCamelCase						)

         assert config.pop("""c"""						) == (2, 5)  # instantiated as tuple
         assert new_config.pop("""c"""						) == [2, 5]  # saved & loaded as list because of json
         config.pop("""_use_default_values"""						)
         assert config == new_config




     def   __A    (     self      :			Union[str, Any]						)		->       Union[str, Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DDIMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert ddim.__class__ == DDIMScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			Dict						)		->       Optional[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =EulerDiscreteScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert euler.__class__ == EulerDiscreteScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       int:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =EulerAncestralDiscreteScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert euler.__class__ == EulerAncestralDiscreteScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       Optional[int]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =PNDMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert pndm.__class__ == PNDMScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			Optional[int]						)		->       List[str]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DDPMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""    ,prediction_type="""sample"""    ,beta_end=8    ,)

         with CaptureLogger(_UpperCamelCase						) as cap_logger_a:
             SCREAMING_SNAKE_CASE__       =DDPMScheduler.from_pretrained("""google/ddpm-celebahq-256"""    ,beta_start=8_8						)

         assert ddpm.__class__ == DDPMScheduler
         assert ddpm.config.prediction_type == "sample"
         assert ddpm.config.beta_end == 8
         assert ddpm_a.config.beta_start == 8_8

         # no warning should be thrown
         assert cap_logger.out == ""
         assert cap_logger_a.out == ""




     def   __A    (     self      :			Any						)		->       Optional[int]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DPMSolverMultistepScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert dpm.__class__ == DPMSolverMultistepScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       str:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()

         SCREAMING_SNAKE_CASE__       ={k: v for k, v in config.config.items() if not k.startswith("""_"""						)}

         # make sure that default config has all keys in `_use_default_values`
         assert set(config_dict.keys()						) == set(config.config._use_default_values						)

         with tempfile.TemporaryDirectory() as tmpdirname:
             config.save_config(_UpperCamelCase						)

             # now loading it with SampleObject2 should put f into `_use_default_values`
             SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(_UpperCamelCase						)

             assert "f" in config._use_default_values
             assert config.f == [1, 3]

         # now loading the config, should **NOT** use [1, 3] for `f`, but the default [1, 4] value
         # **BECAUSE** it is part of `config._use_default_values`
         SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(config.config						)
         assert new_config.f == [5, 4]

         config.config._use_default_values.pop()
         SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(config.config						)
         assert new_config_a.f == [1, 3]

         # Nevertheless "e" should still be correctly loaded to [1, 3] from SampleObject2 instead of defaulting to [1, 5]
         assert new_config_a.e == [1, 3]






