






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 		lowerCAmelCase_							(     UpperCamelCase   ):
     __lowerCamelCase					:						List[Any]     =    '''config.json'''





     @register_to_config
     def __init__(  self       ,   _UpperCAmelCase=2       ,   _UpperCAmelCase=5       ,   _UpperCAmelCase=(2, 5)       ,   _UpperCAmelCase="for diffusion"       ,   _UpperCAmelCase=[1, 3]       ,   )		->							Optional[Any]:
        pass




class 		lowerCAmelCase_							(     UpperCamelCase   ):
     __lowerCamelCase					:						List[Any]     =    '''config.json'''





     @register_to_config
     def __init__(  self       ,   _UpperCAmelCase=2       ,   _UpperCAmelCase=5       ,   _UpperCAmelCase=(2, 5)       ,   _UpperCAmelCase="for diffusion"       ,   _UpperCAmelCase=[1, 3]       ,   )		->							Tuple:
        pass




class 		lowerCAmelCase_							(     UpperCamelCase   ):
     __lowerCamelCase					:						Dict     =    '''config.json'''





     @register_to_config
     def __init__(  self       ,   _UpperCAmelCase=2       ,   _UpperCAmelCase=5       ,   _UpperCAmelCase=(2, 5)       ,   _UpperCAmelCase="for diffusion"       ,   _UpperCAmelCase=[1, 3]       ,   _UpperCAmelCase=[1, 3]       ,   )		->							int:
        pass




class 		lowerCAmelCase_							(     UpperCamelCase   ):
     __lowerCamelCase					:						str     =    '''config.json'''





     @register_to_config
     def __init__(  self       ,   _UpperCAmelCase=2       ,   _UpperCAmelCase=5       ,   _UpperCAmelCase=(2, 5)       ,   _UpperCAmelCase="for diffusion"       ,   _UpperCAmelCase=[1, 5]       ,   _UpperCAmelCase=[5, 4]       ,   )		->							Tuple:
        pass




class 		lowerCAmelCase_							(     unittest.TestCase   ):





     def 	lowerCamelCase       (  self     )		->							List[str]:
        with self.assertRaises(_UpperCAmelCase     ):
           ConfigMixin.load_config("""dummy_path"""     )





     def 	lowerCamelCase       (  self     )		->							Union[str, Any]:
        lowerCamelCase											=SampleObject()
        lowerCamelCase											=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
        lowerCamelCase											=SampleObject(_name_or_path="""lalala"""     )
        lowerCamelCase											=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
        lowerCamelCase											=SampleObject(c=6     )
        lowerCamelCase											=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.
        lowerCamelCase											=SampleObject(1       ,   c=6     )
        lowerCamelCase											=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 	lowerCamelCase       (  self     )		->							Tuple:
        lowerCamelCase											=SampleObject()
        lowerCamelCase											=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     )
           lowerCamelCase											=SampleObject.from_config(SampleObject.load_config(_UpperCAmelCase     )     )
           lowerCamelCase											=new_obj.config

        # unfreeze configs
        lowerCamelCase											=dict(_UpperCAmelCase     )
        lowerCamelCase											=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 	lowerCamelCase       (  self     )		->							List[str]:
        lowerCamelCase											=logging.get_logger("""diffusers.configuration_utils"""     )
        # 30 for warning
        logger.setLevel(30     )

        with CaptureLogger(_UpperCAmelCase     ) as cap_logger:
           lowerCamelCase											=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 	lowerCamelCase       (  self     )		->							int:
        lowerCamelCase											=logging.get_logger("""diffusers.configuration_utils"""     )
        # 30 for warning
        logger.setLevel(30     )

        with CaptureLogger(_UpperCAmelCase     ) as cap_logger:
           lowerCamelCase											=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 	lowerCamelCase       (  self     )		->							str:
        lowerCamelCase											=logging.get_logger("""diffusers.configuration_utils"""     )
        # 30 for warning
        logger.setLevel(30     )

        with CaptureLogger(_UpperCAmelCase     ) as cap_logger:
           lowerCamelCase											=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 	lowerCamelCase       (  self     )		->							Union[str, Any]:
        lowerCamelCase											=logging.get_logger("""diffusers.configuration_utils"""     )
        # 30 for warning
        logger.setLevel(30     )

        with CaptureLogger(_UpperCAmelCase     ) as cap_logger:
           lowerCamelCase											=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 	lowerCamelCase       (  self     )		->							str:
        lowerCamelCase											=logging.get_logger("""diffusers.configuration_utils"""     )
        # 30 for warning
        logger.setLevel(30     )

        with CaptureLogger(_UpperCAmelCase     ) as cap_logger:
           lowerCamelCase											=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:
           lowerCamelCase											=DDPMScheduler.from_pretrained("""google/ddpm-celebahq-256"""       ,   beta_start=88     )

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

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





     def 	lowerCamelCase       (  self     )		->							List[Any]:
        lowerCamelCase											=logging.get_logger("""diffusers.configuration_utils"""     )
        # 30 for warning
        logger.setLevel(30     )

        with CaptureLogger(_UpperCAmelCase     ) as cap_logger:
           lowerCamelCase											=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 	lowerCamelCase       (  self     )		->							List[Any]:
        # let's first save a config that should be in the form
        #    a=2,
        #    b=5,
        #    c=(2, 5),
        #    d="for diffusion",
        #    e=[1, 3],

        lowerCamelCase											=SampleObject()

        lowerCamelCase											={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`
           lowerCamelCase											=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`
        lowerCamelCase											=SampleObjecta.from_config(config.config     )
        assert new_config.f == [5, 4]

        config.config._use_default_values.pop()
        lowerCamelCase											=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]







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				:	Optional[int]    						=    logging.get_logger(__name__)






def    UpperCAmelCase__	(						lowerCamelCase  : bytes ,lowerCamelCase  : int						):
   lowerCamelCase											=f'''{sampling_rate}'''
   lowerCamelCase											="""1"""
   lowerCamelCase											="""f32le"""
   lowerCamelCase											=[
       """ffmpeg""",
       """-i""",
       """pipe:0""",
       """-ac""",
       ac,
       """-ar""",
       ar,
       """-f""",
       format_for_conversion,
       """-hide_banner""",
       """-loglevel""",
       """quiet""",
       """pipe:1""",
   ]

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

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







@add_end_docstrings(UpperCamelCase   )
class 		lowerCAmelCase_							(     UpperCamelCase   ):





     def __init__(  self       ,   *_UpperCAmelCase       ,   **_UpperCAmelCase     )		->							Dict:
        # Default, might be overriden by the model.config.
        lowerCamelCase											=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       ,   _UpperCAmelCase       ,   **_UpperCAmelCase       ,   )		->							Union[str, Any]:
        return super().__call__(_UpperCAmelCase       ,   **_UpperCAmelCase     )





     def 	lowerCamelCase       (  self       ,   _UpperCAmelCase=None       ,   **_UpperCAmelCase     )		->							Optional[int]:
        # No parameters on this pipeline right now
        lowerCamelCase											={}
        if top_k is not None:
           if top_k > self.model.config.num_labels:
              lowerCamelCase											=self.model.config.num_labels
           lowerCamelCase											=top_k
        return {}, {}, postprocess_params





     def 	lowerCamelCase       (  self       ,   _UpperCAmelCase     )		->							Dict:
        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
              lowerCamelCase											=requests.get(_UpperCAmelCase     ).content
           else:
              with open(_UpperCAmelCase       ,   """rb"""     ) as f:
                 lowerCamelCase											=f.read()

        if isinstance(_UpperCAmelCase       ,   _UpperCAmelCase     ):
           lowerCamelCase											=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"""     )

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

              lowerCamelCase											=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"""     )

        lowerCamelCase											=self.feature_extractor(
            _UpperCAmelCase       ,   sampling_rate=self.feature_extractor.sampling_rate       ,   return_tensors="""pt"""     )
        return processed





     def 	lowerCamelCase       (  self       ,   _UpperCAmelCase     )		->							int:
        lowerCamelCase											=self.model(**_UpperCAmelCase     )
        return model_outputs





     def 	lowerCamelCase       (  self       ,   _UpperCAmelCase       ,   _UpperCAmelCase=5     )		->							Tuple:
        lowerCamelCase											=model_outputs.logits[0].softmax(-1     )
        lowerCamelCase							,					lowerCamelCase											=probs.topk(_UpperCAmelCase     )

        lowerCamelCase											=scores.tolist()
        lowerCamelCase											=ids.tolist()

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

        return labels