


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


UpperCamelCase						:				Tuple             =      logging.get_logger(__name__)



@add_end_docstrings(UpperCAmelCase_			)
class 			lowerCamelCase__						(   UpperCAmelCase_			):







						def __init__(							self   :    Tuple			,			**_lowercase   :    int		):
								super().__init__(**_lowercase		)

								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]			,			_lowercase   :    Union[np.ndarray, bytes, str]			,			**_lowercase   :    Any		):
								return super().__call__(_lowercase			,			**_lowercase		)







						def __a						(							self   :    Union[str, Any]			,			**_lowercase   :    Tuple		):
								A					       = {}
								if "candidate_labels" in kwargs:
										A					       = kwargs['candidate_labels']
								if "hypothesis_template" in kwargs:
										A					       = kwargs['hypothesis_template']

								return preprocess_params, {}, {}







						def __a						(							self   :    List[str]			,			_lowercase   :    List[str]			,			_lowercase   :    List[Any]=None			,			_lowercase   :    Optional[int]="This is a sound of {}."		):
								if isinstance(_lowercase			,			_lowercase		):
										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(_lowercase		).content
										else:
												with open(_lowercase			,			'rb'		) as f:
														A					       = f.read()

								if isinstance(_lowercase			,			_lowercase		):
										A					       = ffmpeg_read(_lowercase			,			self.feature_extractor.sampling_rate		)

								if not isinstance(_lowercase			,			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(_lowercase		) for x in candidate_labels]
								A					       = self.tokenizer(_lowercase			,			return_tensors=self.framework			,			padding=_lowercase		)
								A					       = [text_inputs]
								return inputs







						def __a						(							self   :    Dict			,			_lowercase   :    Dict		):
								A					       = model_inputs.pop('candidate_labels'		)
								A					       = model_inputs.pop('text_inputs'		)
								if isinstance(text_inputs[0]			,			_lowercase		):
										A					       = text_inputs[0]
								else:
										# Batching case.
										A					       = text_inputs[0][0]

								A					       = self.model(**_lowercase			,			**_lowercase		)

								A					       = {
								    'candidate_labels': candidate_labels,
								    'logits': outputs.logits_per_audio,
								}
								return model_outputs







						def __a						(							self   :    Union[str, Any]			,			_lowercase   :    Any		):
								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(_lowercase			,			_lowercase		)			,			key=lambda _lowercase		: -x[0]		)
								]
								return result









"""simple docstring"""


from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
	import tensorflow as tf


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



@dataclass
class 			lowerCamelCase__						(   UpperCAmelCase_			):
						lowerCAmelCase       =     [
						    """no_inference""",
						    """no_cuda""",
						    """no_tpu""",
						    """no_speed""",
						    """no_memory""",
						    """no_env_print""",
						    """no_multi_process""",
						]







						def __init__(							self   :    str			,			**_lowercase   :    Union[str, Any]		):
								for deprecated_arg in self.deprecated_args:
										if deprecated_arg in kwargs:
												A					       = deprecated_arg[3:]
												A					       = not kwargs.pop(_lowercase		)
												logger.warning(
												    f'{deprecated_arg} is depreciated. Please use --no-{positive_arg} or'
												    f' {positive_arg}={kwargs[positive_arg]}'		)
								A					       = kwargs.pop('tpu_name'			,			self.tpu_name		)
								A					       = kwargs.pop('device_idx'			,			self.device_idx		)
								A					       = kwargs.pop('eager_mode'			,			self.eager_mode		)
								A					       = kwargs.pop('use_xla'			,			self.use_xla		)
								super().__init__(**_lowercase		)

						lowerCAmelCase       =     field(
						    default=UpperCAmelCase_							,    metadata={"""help""": """Name of TPU"""}							,    )
						lowerCAmelCase       =     field(
						    default=0							,    metadata={"""help""": """CPU / GPU device index. Defaults to 0."""}							,    )
						lowerCAmelCase       =     field(default=UpperCAmelCase_							,    metadata={"""help""": """Benchmark models in eager model."""}			)
						lowerCAmelCase       =     field(
						    default=UpperCAmelCase_							,    metadata={
						        """help""": """Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`."""
						    }							,    )







						@cached_property
						def __a						(							self   :    Optional[Any]		):
								requires_backends(self			,			['tf']		)
								A					       = None
								if self.tpu:
										try:
												if self.tpu_name:
														A					       = tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name		)
												else:
														A					       = tf.distribute.cluster_resolver.TPUClusterResolver()
										except ValueError:
												A					       = None
								return tpu







						@cached_property
						def __a						(							self   :    Dict		):
								requires_backends(self			,			['tf']		)
								if self.is_tpu:
										tf.config.experimental_connect_to_cluster(self._setup_tpu		)
										tf.tpu.experimental.initialize_tpu_system(self._setup_tpu		)

										A					       = tf.distribute.TPUStrategy(self._setup_tpu		)
								else:
										# currently no multi gpu is allowed
										if self.is_gpu:
												# TODO: Currently only single GPU is supported
												tf.config.set_visible_devices(self.gpu_list[self.device_idx]			,			'GPU'		)
												A					       = tf.distribute.OneDeviceStrategy(device=f'/gpu:{self.device_idx}'		)
										else:
												tf.config.set_visible_devices([]			,			'GPU'		)  # disable GPU
												A					       = tf.distribute.OneDeviceStrategy(device=f'/cpu:{self.device_idx}'		)

								return strategy







						@property
						def __a						(							self   :    List[Any]		):
								requires_backends(self			,			['tf']		)
								return self._setup_tpu is not None







						@property
						def __a						(							self   :    Optional[Any]		):
								requires_backends(self			,			['tf']		)
								return self._setup_strategy







						@property
						def __a						(							self   :    str		):
								requires_backends(self			,			['tf']		)
								return tf.config.list_physical_devices('GPU'		)







						@property
						def __a						(							self   :    Any		):
								requires_backends(self			,			['tf']		)
								if self.cuda:
										return len(self.gpu_list		)
								return 0







						@property
						def __a						(							self   :    Dict		):
								return self.n_gpu > 0





