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


__lowerCamelCase			: List[str]							   =		logging.get_logger(__name__)







@add_end_docstrings(lowerCamelCase_		)
class __snake_case				(						lowerCamelCase_		):




   def __init__(    self			:				int       ,	**_lowercase			:				Optional[int]      ):



     """simple docstring"""






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



     """simple docstring"""






     return super().__call__(_lowercase       ,	**_lowercase      )




   def 	__a						(    self			:				List[str]       ,	**_lowercase			:				int      ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__										=      {}
     if "candidate_labels" in kwargs:
       SCREAMING_SNAKE_CASE__										=      kwargs["""candidate_labels"""]
     if "hypothesis_template" in kwargs:
       SCREAMING_SNAKE_CASE__										=      kwargs["""hypothesis_template"""]

     return preprocess_params, {}, {}




   def 	__a						(    self			:				Any       ,	_lowercase			:				Any       ,	_lowercase			:				List[str]=None       ,	_lowercase			:				Dict="This is a sound of {}."      ):



     """simple docstring"""






     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
         SCREAMING_SNAKE_CASE__										=      requests.get(_lowercase      ).content
       else:
         with open(_lowercase       ,	"""rb"""      ) as f:
           SCREAMING_SNAKE_CASE__										=      f.read()

     if isinstance(_lowercase       ,	_lowercase      ):
       SCREAMING_SNAKE_CASE__										=      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"""      )

     SCREAMING_SNAKE_CASE__										=      self.feature_extractor(
         [audio]       ,	sampling_rate=self.feature_extractor.sampling_rate       ,	return_tensors="""pt"""      )
     SCREAMING_SNAKE_CASE__										=      candidate_labels
     SCREAMING_SNAKE_CASE__										=      [hypothesis_template.format(_lowercase      ) for x in candidate_labels]
     SCREAMING_SNAKE_CASE__										=      self.tokenizer(_lowercase       ,	return_tensors=self.framework       ,	padding=_lowercase      )
     SCREAMING_SNAKE_CASE__										=      [text_inputs]
     return inputs




   def 	__a						(    self			:				Tuple       ,	_lowercase			:				Any      ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__										=      model_inputs.pop("""candidate_labels"""      )
     SCREAMING_SNAKE_CASE__										=      model_inputs.pop("""text_inputs"""      )
     if isinstance(text_inputs[0]       ,	_lowercase      ):
       SCREAMING_SNAKE_CASE__										=      text_inputs[0]
     else:
       # Batching case.
       SCREAMING_SNAKE_CASE__										=      text_inputs[0][0]

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

     SCREAMING_SNAKE_CASE__										=      {
         """candidate_labels""": candidate_labels,
         """logits""": outputs.logits_per_audio,
     }
     return model_outputs




   def 	__a						(    self			:				Union[str, Any]       ,	_lowercase			:				int      ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__										=      model_outputs.pop("""candidate_labels"""      )
     SCREAMING_SNAKE_CASE__										=      model_outputs["""logits"""][0]

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

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

import doctest
from collections import deque

import numpy as np







class __snake_case				:




   def __init__(    self			:				Dict      ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__										=      [2, 1, 2, -1]
     SCREAMING_SNAKE_CASE__										=      [1, 2, 3, 4]




   def 	__a						(    self			:				int      ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__										=      len(self.first_signal      )
     SCREAMING_SNAKE_CASE__										=      len(self.second_signal      )

     SCREAMING_SNAKE_CASE__										=      max(_lowercase       ,	_lowercase      )

     # create a zero matrix of max_length x max_length
     SCREAMING_SNAKE_CASE__										=      [[0] * max_length for i in range(_lowercase      )]

     # fills the smaller signal with zeros to make both signals of same length
     if length_first_signal < length_second_signal:
       self.first_signal += [0] * (max_length - length_first_signal)
     elif length_first_signal > length_second_signal:
       self.second_signal += [0] * (max_length - length_second_signal)
     for i in range(_lowercase      ):
       SCREAMING_SNAKE_CASE__										=      deque(self.second_signal      )
       rotated_signal.rotate(_lowercase      )
       for j, item in enumerate(_lowercase      ):
         matrix[i][j] += item

        # multiply the matrix with the first signal
     SCREAMING_SNAKE_CASE__										=      np.matmul(np.transpose(_lowercase      )       ,	np.transpose(self.first_signal      )      )

     # rounding-off to two decimal places
     return [round(_lowercase       ,	2      ) for i in final_signal]


if __name__ == "__main__":
   doctest.testmod()
