
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      =logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase_ )
class 				_lowerCamelCase (      UpperCamelCase_ ):





     """simple docstring"""







     def __init__(							self						,						**__SCREAMING_SNAKE_CASE  )  ->   List[str]:




           """simple docstring"""







           super().__init__(**__SCREAMING_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						,						__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   Optional[int]:




           """simple docstring"""







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






     def    __SCREAMING_SNAKE_CASE			(							self						,						**__SCREAMING_SNAKE_CASE  )  ->   Dict:




           """simple docstring"""







           UpperCamelCase__    :   int					=				{}
           if "candidate_labels" in kwargs:
                 UpperCamelCase__    :   List[str]					=				kwargs['''candidate_labels''']
           if "hypothesis_template" in kwargs:
                 UpperCamelCase__    :   str					=				kwargs['''hypothesis_template''']

           return preprocess_params, {}, {}






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE=None						,						__SCREAMING_SNAKE_CASE="This is a sound of {}."  )  ->   Dict:




           """simple docstring"""







           if isinstance(__SCREAMING_SNAKE_CASE						,						__SCREAMING_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
                       UpperCamelCase__    :   Optional[Any]					=				requests.get(__SCREAMING_SNAKE_CASE  ).content
                 else:
                       with open(__SCREAMING_SNAKE_CASE						,						'''rb'''  ) as f:
                             UpperCamelCase__    :   int					=				f.read()

           if isinstance(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  ):
                 UpperCamelCase__    :   Union[str, Any]					=				ffmpeg_read(__SCREAMING_SNAKE_CASE						,						self.feature_extractor.sampling_rate  )

           if not isinstance(__SCREAMING_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'''  )

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






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




           """simple docstring"""







           UpperCamelCase__    :   Optional[int]					=				model_inputs.pop('''candidate_labels'''  )
           UpperCamelCase__    :   Tuple					=				model_inputs.pop('''text_inputs'''  )
           if isinstance(text_inputs[0]						,						__SCREAMING_SNAKE_CASE  ):
                 UpperCamelCase__    :   List[Any]					=				text_inputs[0]
           else:
                 # Batching case.
                 UpperCamelCase__    :   Optional[Any]					=				text_inputs[0][0]

           UpperCamelCase__    :   List[Any]					=				self.model(**__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )

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






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE  )  ->   str:




           """simple docstring"""







           UpperCamelCase__    :   Optional[Any]					=				model_outputs.pop('''candidate_labels'''  )
           UpperCamelCase__    :   List[Any]					=				model_outputs['''logits'''][0]

           if self.framework == "pt":
                 UpperCamelCase__    :   Optional[int]					=				logits.softmax(dim=0  )
                 UpperCamelCase__    :   Any					=				probs.tolist()
           else:
                 raise ValueError('''`tf` framework not supported.'''  )

           UpperCamelCase__    :   Dict					=				[
               {'''score''': score, '''label''': candidate_label}
               for score, candidate_label in sorted(zip(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )						,						key=lambda __SCREAMING_SNAKE_CASE  : -x[0]  )
           ]
           return result




import torch
from transformers import PreTrainedModel, XLMRobertaConfig, XLMRobertaModel




class 				_lowerCamelCase (      UpperCamelCase_ ):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		   =  '''M-CLIP'''






     def __init__(							self						,						__SCREAMING_SNAKE_CASE=1_0_2_4						,						__SCREAMING_SNAKE_CASE=7_6_8						,						**__SCREAMING_SNAKE_CASE  )  ->   List[Any]:




           """simple docstring"""







           UpperCamelCase__    :   Dict					=				transformerDimSize
           UpperCamelCase__    :   Any					=				imageDimSize
           super().__init__(**__SCREAMING_SNAKE_CASE  )




class 				_lowerCamelCase (      UpperCamelCase_ ):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		   =  MCLIPConfig






     def __init__(							self						,						__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   Optional[int]:




           """simple docstring"""







           super().__init__(__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Union[str, Any]					=				XLMRobertaModel(__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Tuple					=				torch.nn.Linear(
               in_features=config.transformerDimensions						,						out_features=config.numDims  )






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




           """simple docstring"""







           UpperCamelCase__    :   Tuple					=				self.transformer(input_ids=__SCREAMING_SNAKE_CASE						,						attention_mask=__SCREAMING_SNAKE_CASE  )[0]
           UpperCamelCase__    :   Tuple					=				(embs * attention_mask.unsqueeze(2  )).sum(dim=1  ) / attention_mask.sum(dim=1  )[:, None]
           return self.LinearTransformation(__SCREAMING_SNAKE_CASE  ), embs


