




'''simple docstring'''





import numpy as np

_SCREAMING_SNAKE_CASE							:   List[Any]      		= [
    ["""a""", """b""", """c""", """d""", """e"""],
    ["""f""", """g""", """h""", """i""", """k"""],
    ["""l""", """m""", """n""", """o""", """p"""],
    ["""q""", """r""", """s""", """t""", """u"""],
    ["""v""", """w""", """x""", """y""", """z"""],
]




class  _snake_case		:




			def __init__(					self							)				->				Any:

										'''simple docstring'''




										snake_case_						=      np.array(a__							)




			def    lowerCAmelCase__				(					self				,							a__							)				->				Optional[Any]:

										'''simple docstring'''




										snake_case_						=      np.where(letter == self.SQUARE							)
										snake_case_						=      np.concatenate([indexa + 1, indexa + 1]							)
										return indexes




			def    lowerCAmelCase__				(					self				,							a__				,							a__							)				->				List[Any]:

										'''simple docstring'''




										snake_case_						=      self.SQUARE[indexa - 1, indexa - 1]
										return letter




			def    lowerCAmelCase__				(					self				,							a__							)				->				int:

										'''simple docstring'''




										snake_case_						=      message.lower()
										snake_case_						=      message.replace(" "				,							""							)
										snake_case_						=      message.replace("j"				,							"i"							)

										snake_case_						=      np.empty((2, len(a__							))							)
										for letter_index in range(len(a__							)							):
																	snake_case_						=      self.letter_to_numbers(message[letter_index]							)

																	snake_case_						=      numbers[0]
																	snake_case_						=      numbers[1]

										snake_case_						=      first_step.reshape(2 * len(a__							)							)
										snake_case_						=      """"""
										for numbers_index in range(len(a__							)							):
																	snake_case_						=      int(second_step[numbers_index * 2]							)
																	snake_case_						=      int(second_step[(numbers_index * 2) + 1]							)
																	snake_case_						=      self.numbers_to_letter(a__				,							a__							)
																	snake_case_						=      encoded_message + letter

										return encoded_message




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

										'''simple docstring'''




										snake_case_						=      message.lower()
										message.replace(" "				,							""							)
										snake_case_						=      np.empty(2 * len(a__							)							)
										for letter_index in range(len(a__							)							):
																	snake_case_						=      self.letter_to_numbers(message[letter_index]							)
																	snake_case_						=      numbers[0]
																	snake_case_						=      numbers[1]

										snake_case_						=      first_step.reshape((2, len(a__							))							)
										snake_case_						=      """"""
										for numbers_index in range(len(a__							)							):
																	snake_case_						=      int(second_step[0, numbers_index]							)
																	snake_case_						=      int(second_step[1, numbers_index]							)
																	snake_case_						=      self.numbers_to_letter(a__				,							a__							)
																	snake_case_						=      decoded_message + letter

										return decoded_message






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
