
"""simple docstring"""




import inspect
import os
import unittest
from dataclasses import dataclass

import torch

from accelerate import Accelerator, DistributedDataParallelKwargs, GradScalerKwargs
from accelerate.state import AcceleratorState
from accelerate.test_utils import execute_subprocess_async, require_cuda, require_multi_gpu
from accelerate.utils import KwargsHandler



@dataclass
class 	lowerCAmelCase_						(a__  ):


				"""simple docstring"""



				__UpperCamelCase      :						int					       =   0
				__UpperCamelCase      :						bool					       =   False
				__UpperCamelCase      :						float					       =   3.0






class 	lowerCAmelCase_						(unittest.TestCase  ):


				"""simple docstring"""



				def 	__magic_name__    (self    )					->		List[str]:



											"""simple docstring"""




											self.assertDictEqual(MockClass().to_kwargs()   ,       {}    )
											self.assertDictEqual(MockClass(a=2    ).to_kwargs()   ,       {"""a""": 2}    )
											self.assertDictEqual(MockClass(a=2   ,       b=SCREAMING_SNAKE_CASE__    ).to_kwargs()   ,       {"""a""": 2, """b""": True}    )
											self.assertDictEqual(MockClass(a=2   ,       c=2.25    ).to_kwargs()   ,       {"""a""": 2, """c""": 2.25}    )



				@require_cuda
				def 	__magic_name__    (self    )					->		int:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       Tuple           =   GradScalerKwargs(init_scale=10_24   ,       growth_factor=2    )
											AcceleratorState._reset_state()
											SCREAMING_SNAKE_CASE__						:       Tuple           =   Accelerator(mixed_precision="""fp16"""   ,       kwargs_handlers=[scaler_handler]    )
											print(accelerator.use_fpaa    )
											SCREAMING_SNAKE_CASE__						:       str           =   accelerator.scaler

											# Check the kwargs have been applied
											self.assertEqual(scaler._init_scale   ,       1024.0    )
											self.assertEqual(scaler._growth_factor   ,       2.0    )

											# Check the other values are at the default
											self.assertEqual(scaler._backoff_factor   ,       0.5    )
											self.assertEqual(scaler._growth_interval   ,       20_00    )
											self.assertEqual(scaler._enabled   ,       SCREAMING_SNAKE_CASE__    )



				@require_multi_gpu
				def 	__magic_name__    (self    )					->		Union[str, Any]:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       int           =   ["""torchrun""", F'''--nproc_per_node={torch.cuda.device_count()}''', inspect.getfile(self.__class__    )]
											execute_subprocess_async(SCREAMING_SNAKE_CASE__   ,       env=os.environ.copy()    )


if __name__ == "__main__":
					UpperCAmelCase__      :    Optional[Any]    							= DistributedDataParallelKwargs(bucket_cap_mb=1_5, find_unused_parameters=True)
					UpperCAmelCase__      :    Union[str, Any]    							= Accelerator(kwargs_handlers=[ddp_scaler])
					UpperCAmelCase__      :    Any    							= torch.nn.Linear(1_0_0, 2_0_0)
					UpperCAmelCase__      :    Optional[Any]    							= accelerator.prepare(model)

					# Check the values changed in kwargs
					UpperCAmelCase__      :    Tuple    							= ''
					UpperCAmelCase__      :    List[str]    							= model.bucket_bytes_cap // (1_0_2_4 * 1_0_2_4)
					if observed_bucket_cap_map != 1_5:
										error_msg += f"Kwargs badly passed, should have `15` but found {observed_bucket_cap_map}.\n"
					if model.find_unused_parameters is not True:
										error_msg += f"Kwargs badly passed, should have `True` but found {model.find_unused_parameters}.\n"

					# Check the values of the defaults
					if model.dim != 0:
										error_msg += f"Default value not respected, should have `0` but found {model.dim}.\n"
					if model.broadcast_buffers is not True:
										error_msg += f"Default value not respected, should have `True` but found {model.broadcast_buffers}.\n"
					if model.gradient_as_bucket_view is not False:
										error_msg += f"Default value not respected, should have `False` but found {model.gradient_as_bucket_view}.\n"

					# Raise error at the end to make sure we don't stop at the first failure.
					if len(error_msg) > 0:
										raise ValueError(error_msg)






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


lowercase  :					Optional[Any]						    =							logging.get_logger(__name__)

@add_end_docstrings(lowerCAmelCase     )
class __snake_case		(      lowerCAmelCase     ):
    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       _SCREAMING_SNAKE_CASE       (  self		,**snake_case      ):



           '''simple docstring'''







           lowercase							:     Union[str, Any]       	=      {}
           if "candidate_labels" in kwargs:
                  lowercase							:     List[str]       	=      kwargs["""candidate_labels"""]
           if "hypothesis_template" in kwargs:
                  lowercase							:     Dict       	=      kwargs["""hypothesis_template"""]

           return preprocess_params, {}, {}
    def       _SCREAMING_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
                         lowercase							:     Optional[Any]       	=      requests.get(snake_case      ).content
                  else:
                         with open(snake_case		,"""rb"""      ) as f:
                                lowercase							:     Union[str, Any]       	=      f.read()

           if isinstance(snake_case		,snake_case      ):
                  lowercase							:     int       	=      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"""      )

           lowercase							:     Dict       	=      self.feature_extractor(
               [audio]		,sampling_rate=self.feature_extractor.sampling_rate		,return_tensors="""pt"""      )
           lowercase							:     Tuple       	=      candidate_labels
           lowercase							:     Tuple       	=      [hypothesis_template.format(snake_case      ) for x in candidate_labels]
           lowercase							:     Optional[Any]       	=      self.tokenizer(snake_case		,return_tensors=self.framework		,padding=snake_case      )
           lowercase							:     Optional[Any]       	=      [text_inputs]
           return inputs
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     List[str]       	=      model_inputs.pop("""candidate_labels"""      )
           lowercase							:     Dict       	=      model_inputs.pop("""text_inputs"""      )
           if isinstance(text_inputs[0]		,snake_case      ):
                  lowercase							:     List[Any]       	=      text_inputs[0]
           else:
                  # Batching case.
                  lowercase							:     Dict       	=      text_inputs[0][0]

           lowercase							:     Optional[Any]       	=      self.model(**snake_case		,**snake_case      )

           lowercase							:     Any       	=      {
               """candidate_labels""": candidate_labels,
               """logits""": outputs.logits_per_audio,
           }
           return model_outputs
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     List[Any]       	=      model_outputs.pop("""candidate_labels"""      )
           lowercase							:     Any       	=      model_outputs["""logits"""][0]

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

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