





"""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					      =   logging.get_logger(__name__)
@add_end_docstrings(snake_case__ )
class 		SCREAMING_SNAKE_CASE_				(     snake_case__ ):
				"""simple docstring"""




				def __init__(  self	:Union[str, Any]    ,     **__lowercase	:List[Any]      ):
											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	:Tuple    ,     __lowercase	:Union[np.ndarray, bytes, str]    ,     **__lowercase	:Union[str, Any]      ):
											return super().__call__(__lowercase    ,     **__lowercase      )




				def        __lowercase	(  self	:Optional[Any]    ,     **__lowercase	:Any      ):
											__lowerCamelCase   :   Any       ={}
											if "candidate_labels" in kwargs:
																		__lowerCamelCase   :   Optional[Any]       =kwargs['''candidate_labels''']
											if "hypothesis_template" in kwargs:
																		__lowerCamelCase   :   Tuple       =kwargs['''hypothesis_template''']

											return preprocess_params, {}, {}




				def        __lowercase	(  self	:List[Any]    ,     __lowercase	:Optional[Any]    ,     __lowercase	:Optional[int]=None    ,     __lowercase	:List[str]="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
																									__lowerCamelCase   :   Optional[int]       =requests.get(__lowercase      ).content
																		else:
																									with open(__lowercase    ,     '''rb'''      ) as f:
																																__lowerCamelCase   :   str       =f.read()

											if isinstance(__lowercase    ,     __lowercase      ):
																		__lowerCamelCase   :   Optional[Any]       =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'''      )

											__lowerCamelCase   :   Dict       =self.feature_extractor(
											    [audio]    ,     sampling_rate=self.feature_extractor.sampling_rate    ,     return_tensors='''pt'''      )
											__lowerCamelCase   :   Optional[int]       =candidate_labels
											__lowerCamelCase   :   Any       =[hypothesis_template.format(__lowercase      ) for x in candidate_labels]
											__lowerCamelCase   :   List[str]       =self.tokenizer(__lowercase    ,     return_tensors=self.framework    ,     padding=__lowercase      )
											__lowerCamelCase   :   Any       =[text_inputs]
											return inputs




				def        __lowercase	(  self	:Any    ,     __lowercase	:Any      ):
											__lowerCamelCase   :   Dict       =model_inputs.pop('''candidate_labels'''      )
											__lowerCamelCase   :   List[str]       =model_inputs.pop('''text_inputs'''      )
											if isinstance(text_inputs[0]    ,     __lowercase      ):
																		__lowerCamelCase   :   int       =text_inputs[0]
											else:
																		# Batching case.
																		__lowerCamelCase   :   Union[str, Any]       =text_inputs[0][0]

											__lowerCamelCase   :   List[Any]       =self.model(**__lowercase    ,     **__lowercase      )

											__lowerCamelCase   :   List[str]       ={
											    '''candidate_labels''': candidate_labels,
											    '''logits''': outputs.logits_per_audio,
											}
											return model_outputs




				def        __lowercase	(  self	:Tuple    ,     __lowercase	:List[str]      ):
											__lowerCamelCase   :   List[Any]       =model_outputs.pop('''candidate_labels'''      )
											__lowerCamelCase   :   Dict       =model_outputs['''logits'''][0]

											if self.framework == "pt":
																		__lowerCamelCase   :   Dict       =logits.softmax(dim=0      )
																		__lowerCamelCase   :   str       =probs.tolist()
											else:
																		raise ValueError('''`tf` framework not supported.'''      )

											__lowerCamelCase   :   List[str]       =[
											    {'''score''': score, '''label''': candidate_label}
											    for score, candidate_label in sorted(zip(__lowercase    ,     __lowercase      )    ,     key=lambda __lowercase      : -x[0]      )
											]
											return result











"""simple docstring"""





import inspect
import os
import unittest

import torch

import accelerate
from accelerate import debug_launcher
from accelerate.test_utils import (
    execute_subprocess_async,
    require_cpu,
    require_huggingface_suite,
    require_multi_gpu,
    require_single_gpu,
)
from accelerate.utils import patch_environment
@require_huggingface_suite
class 		SCREAMING_SNAKE_CASE_				(     unittest.TestCase ):
				"""simple docstring"""




				def        __lowercase	(  self	:Dict      ):
											__lowerCamelCase   :   Union[str, Any]       =inspect.getfile(accelerate.test_utils      )
											__lowerCamelCase   :   Union[str, Any]       =os.path.sep.join(
											    mod_file.split(os.path.sep      )[:-1] + ['''scripts''', '''external_deps''', '''test_metrics.py''']      )

											from accelerate.test_utils.scripts.external_deps import test_metrics  # noqa: F401

											__lowerCamelCase   :   List[str]       =test_metrics




				@require_cpu
				def        __lowercase	(  self	:Optional[Any]      ):
											debug_launcher(self.test_metrics.main    ,     num_processes=1      )




				@require_cpu
				def        __lowercase	(  self	:int      ):
											debug_launcher(self.test_metrics.main      )




				@require_single_gpu
				def        __lowercase	(  self	:int      ):
											self.test_metrics.main()




				@require_multi_gpu
				def        __lowercase	(  self	:Dict      ):
											print(f'Found {torch.cuda.device_count()} devices.'      )
											__lowerCamelCase   :   Union[str, Any]       =['''torchrun''', f'--nproc_per_node={torch.cuda.device_count()}', self.test_file_path]
											with patch_environment(omp_num_threads=1      ):
																		execute_subprocess_async(__lowercase    ,     env=os.environ.copy()      )




