




'''simple docstring'''





import dataclasses
import json
import warnings
from dataclasses import dataclass, field
from time import time
from typing import List

from ..utils import logging


_SCREAMING_SNAKE_CASE							:   int      		= logging.get_logger(__name__)


def      UpperCamelCase_(	snake_case							:       Union[str, Any]=None							,				snake_case							:       Dict=None     ):

							'''simple docstring'''






							return field(default_factory=lambda: default							,				metadata=SCREAMING_SNAKE_CASE__     )




@dataclass
class  _snake_case		:
			lowerCAmelCase_     :      List[str]						       =    list_field(
			    default=[]						,						metadata={
			        "help": (
			            "Model checkpoints to be provided to the AutoModel classes. Leave blank to benchmark the base version"
			            " of all available models"
			        )
			    }						,						)

			lowerCAmelCase_     :      List[int]						       =    list_field(
			    default=[8]						,						metadata={"help": "List of batch sizes for which memory and time performance will be evaluated"}							)

			lowerCAmelCase_     :      List[int]						       =    list_field(
			    default=[8, 32, 128, 512]						,						metadata={"help": "List of sequence lengths for which memory and time performance will be evaluated"}						,						)

			lowerCAmelCase_     :      bool						       =    field(
			    default=lowercase_						,						metadata={"help": "Whether to benchmark inference of model. Inference can be disabled via --no-inference."}						,						)
			lowerCAmelCase_     :      bool						       =    field(
			    default=lowercase_						,						metadata={"help": "Whether to run on available cuda devices. Cuda can be disabled via --no-cuda."}						,						)
			lowerCAmelCase_     :      bool						       =    field(
			    default=lowercase_						,						metadata={"help": "Whether to run on available tpu devices. TPU can be disabled via --no-tpu."}							)
			lowerCAmelCase_     :      bool						       =    field(default=lowercase_						,						metadata={"help": "Use FP16 to accelerate inference."}							)
			lowerCAmelCase_     :      bool						       =    field(default=lowercase_						,						metadata={"help": "Benchmark training of model"}							)
			lowerCAmelCase_     :      bool						       =    field(default=lowercase_						,						metadata={"help": "Verbose memory tracing"}							)
			lowerCAmelCase_     :      bool						       =    field(
			    default=lowercase_						,						metadata={"help": "Whether to perform speed measurements. Speed measurements can be disabled via --no-speed."}						,						)
			lowerCAmelCase_     :      bool						       =    field(
			    default=lowercase_						,						metadata={
			        "help": "Whether to perform memory measurements. Memory measurements can be disabled via --no-memory"
			    }						,						)
			lowerCAmelCase_     :      bool						       =    field(default=lowercase_						,						metadata={"help": "Trace memory line by line"}							)
			lowerCAmelCase_     :      bool						       =    field(default=lowercase_						,						metadata={"help": "Save result to a CSV file"}							)
			lowerCAmelCase_     :      bool						       =    field(default=lowercase_						,						metadata={"help": "Save all print statements in a log file"}							)
			lowerCAmelCase_     :      bool						       =    field(default=lowercase_						,						metadata={"help": "Whether to print environment information"}							)
			lowerCAmelCase_     :      bool						       =    field(
			    default=lowercase_						,						metadata={
			        "help": (
			            "Whether to use multiprocessing for memory and speed measurement. It is highly recommended to use"
			            " multiprocessing for accurate CPU and GPU memory measurements. This option should only be disabled"
			            " for debugging / testing and on TPU."
			        )
			    }						,						)
			lowerCAmelCase_     :      str						       =    field(
			    default=F'''inference_time_{round(time()							)}.csv'''						,						metadata={"help": "CSV filename used if saving time results to csv."}						,						)
			lowerCAmelCase_     :      str						       =    field(
			    default=F'''inference_memory_{round(time()							)}.csv'''						,						metadata={"help": "CSV filename used if saving memory results to csv."}						,						)
			lowerCAmelCase_     :      str						       =    field(
			    default=F'''train_time_{round(time()							)}.csv'''						,						metadata={"help": "CSV filename used if saving time results to csv for training."}						,						)
			lowerCAmelCase_     :      str						       =    field(
			    default=F'''train_memory_{round(time()							)}.csv'''						,						metadata={"help": "CSV filename used if saving memory results to csv for training."}						,						)
			lowerCAmelCase_     :      str						       =    field(
			    default=F'''env_info_{round(time()							)}.csv'''						,						metadata={"help": "CSV filename used if saving environment information."}						,						)
			lowerCAmelCase_     :      str						       =    field(
			    default=F'''log_{round(time()							)}.csv'''						,						metadata={"help": "Log filename used if print statements are saved in log."}						,						)
			lowerCAmelCase_     :      int						       =    field(default=3						,						metadata={"help": "Times an experiment will be run."}							)
			lowerCAmelCase_     :      bool						       =    field(
			    default=lowercase_						,						metadata={
			        "help": (
			            "Instead of loading the model as defined in `config.architectures` if exists, just load the pretrain"
			            " model weights."
			        )
			    }						,						)




			def    lowerCAmelCase__				(					self							)				->				Tuple:

										'''simple docstring'''




										warnings.warn(
										    F'The class {self.__class__} is deprecated. Hugging Face Benchmarking utils'
										    " are deprecated in general and it is advised to use external Benchmarking libraries "
										    " to benchmark Transformer models."				,							a__				,							)




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

										'''simple docstring'''




										return json.dumps(dataclasses.asdict(self							)				,							indent=2							)




			@property
			def    lowerCAmelCase__				(					self							)				->				str:

										'''simple docstring'''




										if len(self.models							) <= 0:
																	raise ValueError(
																	    "Please make sure you provide at least one model name / model identifier, *e.g.* `--models"
																	    " bert-base-cased` or `args.models = ['bert-base-cased']."							)
										return self.models




			@property
			def    lowerCAmelCase__				(					self							)				->				str:

										'''simple docstring'''




										if not self.multi_process:
																	return False
										elif self.is_tpu:
																	logger.info("Multiprocessing is currently not possible on TPU."							)
																	return False
										else:
																	return True






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
