



"""simple docstring"""


import copy
import os
from typing import TYPE_CHECKING, List, Union


if TYPE_CHECKING:
						pass

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__A   :							str										=   logging.get_logger(__name__)

__A   :							List[str]										=   {
    'kakaobrain/align-base': 'https://huggingface.co/kakaobrain/align-base/resolve/main/config.json',
}
class 			_a (					lowerCAmelCase):
			"""simple docstring"""
			UpperCamelCase__			      =	"""align_text_model"""



			def __init__(		self     :    Optional[int]				,   __UpperCamelCase     :    Optional[int]=3_0_5_2_2				,   __UpperCamelCase     :    int=7_6_8				,   __UpperCamelCase     :    Optional[Any]=1_2				,   __UpperCamelCase     :    Any=1_2				,   __UpperCamelCase     :    List[Any]=3_0_7_2				,   __UpperCamelCase     :    Any="gelu"				,   __UpperCamelCase     :    List[Any]=0.1				,   __UpperCamelCase     :    Dict=0.1				,   __UpperCamelCase     :    str=5_1_2				,   __UpperCamelCase     :    str=2				,   __UpperCamelCase     :    List[Any]=0.0_2				,   __UpperCamelCase     :    List[Any]=1e-12				,   __UpperCamelCase     :    List[Any]=0				,   __UpperCamelCase     :    List[Any]="absolute"				,   __UpperCamelCase     :    Dict=True				,   **__UpperCamelCase     :    int				,   )->List[Any]:
							super().__init__(**_lowerCAmelCase   )

							_UpperCAmelCase					=    vocab_size
							_UpperCAmelCase					=    hidden_size
							_UpperCAmelCase					=    num_hidden_layers
							_UpperCAmelCase					=    num_attention_heads
							_UpperCAmelCase					=    hidden_act
							_UpperCAmelCase					=    intermediate_size
							_UpperCAmelCase					=    hidden_dropout_prob
							_UpperCAmelCase					=    attention_probs_dropout_prob
							_UpperCAmelCase					=    max_position_embeddings
							_UpperCAmelCase					=    type_vocab_size
							_UpperCAmelCase					=    initializer_range
							_UpperCAmelCase					=    layer_norm_eps
							_UpperCAmelCase					=    position_embedding_type
							_UpperCAmelCase					=    use_cache
							_UpperCAmelCase					=    pad_token_id



			@classmethod
			def  lowercase__						(		cls     :    Optional[Any]				,   __UpperCamelCase     :    Optional[Any]				,   **__UpperCamelCase     :    List[str]   )->int:
							cls._set_token_in_kwargs(_lowerCAmelCase   )

							_UpperCAmelCase		,						_UpperCAmelCase					=    cls.get_config_dict(_lowerCAmelCase				,   **_lowerCAmelCase   )

							# get the text config dict if we are loading from AlignConfig
							if config_dict.get('''model_type'''   ) == "align":
											_UpperCAmelCase					=    config_dict['''text_config''']

							if "model_type" in config_dict and hasattr(cls				,   '''model_type'''   ) and config_dict["model_type"] != cls.model_type:
											logger.warning(
											    F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
											    F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'   )

							return cls.from_dict(_lowerCAmelCase				,   **_lowerCAmelCase   )



class 			_a (					lowerCAmelCase):
			"""simple docstring"""
			UpperCamelCase__			      =	"""align_vision_model"""



			def __init__(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[str, Any] = 3				,   __UpperCamelCase     :    List[str] = 6_0_0				,   __UpperCamelCase     :    Dict = 2.0				,   __UpperCamelCase     :    Dict = 3.1				,   __UpperCamelCase     :    List[Any] = 8				,   __UpperCamelCase     :    Any = [3, 3, 5, 3, 5, 5, 3]				,   __UpperCamelCase     :    Optional[Any] = [3_2, 1_6, 2_4, 4_0, 8_0, 1_1_2, 1_9_2]				,   __UpperCamelCase     :    Any = [1_6, 2_4, 4_0, 8_0, 1_1_2, 1_9_2, 3_2_0]				,   __UpperCamelCase     :    Any = []				,   __UpperCamelCase     :    str = [1, 2, 2, 2, 1, 2, 1]				,   __UpperCamelCase     :    List[Any] = [1, 2, 2, 3, 3, 4, 1]				,   __UpperCamelCase     :    Optional[int] = [1, 6, 6, 6, 6, 6, 6]				,   __UpperCamelCase     :    List[str] = 0.2_5				,   __UpperCamelCase     :    List[Any] = "swish"				,   __UpperCamelCase     :    List[str] = 2_5_6_0				,   __UpperCamelCase     :    Union[str, Any] = "mean"				,   __UpperCamelCase     :    Tuple = 0.0_2				,   __UpperCamelCase     :    int = 0.0_0_1				,   __UpperCamelCase     :    Tuple = 0.9_9				,   __UpperCamelCase     :    str = 0.2				,   **__UpperCamelCase     :    List[str]				,   )->Optional[int]:
							super().__init__(**_lowerCAmelCase   )

							_UpperCAmelCase					=    num_channels
							_UpperCAmelCase					=    image_size
							_UpperCAmelCase					=    width_coefficient
							_UpperCAmelCase					=    depth_coefficient
							_UpperCAmelCase					=    depth_divisor
							_UpperCAmelCase					=    kernel_sizes
							_UpperCAmelCase					=    in_channels
							_UpperCAmelCase					=    out_channels
							_UpperCAmelCase					=    depthwise_padding
							_UpperCAmelCase					=    strides
							_UpperCAmelCase					=    num_block_repeats
							_UpperCAmelCase					=    expand_ratios
							_UpperCAmelCase					=    squeeze_expansion_ratio
							_UpperCAmelCase					=    hidden_act
							_UpperCAmelCase					=    hidden_dim
							_UpperCAmelCase					=    pooling_type
							_UpperCAmelCase					=    initializer_range
							_UpperCAmelCase					=    batch_norm_eps
							_UpperCAmelCase					=    batch_norm_momentum
							_UpperCAmelCase					=    drop_connect_rate
							_UpperCAmelCase					=    sum(_lowerCAmelCase   ) * 4



			@classmethod
			def  lowercase__						(		cls     :    Any				,   __UpperCamelCase     :    Union[str, Any]				,   **__UpperCamelCase     :    Union[str, Any]   )->Optional[Any]:
							cls._set_token_in_kwargs(_lowerCAmelCase   )

							_UpperCAmelCase		,						_UpperCAmelCase					=    cls.get_config_dict(_lowerCAmelCase				,   **_lowerCAmelCase   )

							# get the vision config dict if we are loading from AlignConfig
							if config_dict.get('''model_type'''   ) == "align":
											_UpperCAmelCase					=    config_dict['''vision_config''']

							if "model_type" in config_dict and hasattr(cls				,   '''model_type'''   ) and config_dict["model_type"] != cls.model_type:
											logger.warning(
											    F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
											    F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'   )

							return cls.from_dict(_lowerCAmelCase				,   **_lowerCAmelCase   )



class 			_a (					lowerCAmelCase):
			"""simple docstring"""
			UpperCamelCase__			      =	"""align"""
			UpperCamelCase__			      =	True



			def __init__(		self     :    str				,   __UpperCamelCase     :    Dict=None				,   __UpperCamelCase     :    Union[str, Any]=None				,   __UpperCamelCase     :    Optional[int]=6_4_0				,   __UpperCamelCase     :    str=1.0				,   __UpperCamelCase     :    Any=0.0_2				,   **__UpperCamelCase     :    Union[str, Any]				,   )->Any:
							super().__init__(**_lowerCAmelCase   )

							if text_config is None:
											_UpperCAmelCase					=    {}
											logger.info('''text_config is None. Initializing the AlignTextConfig with default values.'''   )

							if vision_config is None:
											_UpperCAmelCase					=    {}
											logger.info('''vision_config is None. Initializing the AlignVisionConfig with default values.'''   )

							_UpperCAmelCase					=    AlignTextConfig(**_lowerCAmelCase   )
							_UpperCAmelCase					=    AlignVisionConfig(**_lowerCAmelCase   )

							_UpperCAmelCase					=    projection_dim
							_UpperCAmelCase					=    temperature_init_value
							_UpperCAmelCase					=    initializer_range



			@classmethod
			def  lowercase__						(		cls     :    List[Any]				,   __UpperCamelCase     :    Optional[int]				,   __UpperCamelCase     :    Optional[int]				,   **__UpperCamelCase     :    int   )->Tuple:
							return cls(text_config=text_config.to_dict()				,   vision_config=vision_config.to_dict()				,   **_lowerCAmelCase   )



			def  lowercase__						(		self     :    str   )->Optional[Any]:
							_UpperCAmelCase					=    copy.deepcopy(self.__dict__   )
							_UpperCAmelCase					=    self.text_config.to_dict()
							_UpperCAmelCase					=    self.vision_config.to_dict()
							_UpperCAmelCase					=    self.__class__.model_type
							return output








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


__A   :							List[str]										=   logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase)
class 			_a (					lowerCAmelCase):
			"""simple docstring"""



			def __init__(		self     :    Union[str, Any]				,   **__UpperCamelCase     :    Optional[Any]   )->List[Any]:
							super().__init__(**__UpperCamelCase   )

							if self.framework != "pt":
											raise ValueError(F'The {self.__class__} is only available in PyTorch.'   )
							# No specific FOR_XXX available yet



			def __call__(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[np.ndarray, bytes, str]				,   **__UpperCamelCase     :    Tuple   )->List[str]:
							return super().__call__(__UpperCamelCase				,   **__UpperCamelCase   )



			def  lowercase__						(		self     :    Any				,   **__UpperCamelCase     :    Any   )->Union[str, Any]:
							_UpperCAmelCase					=    {}
							if "candidate_labels" in kwargs:
											_UpperCAmelCase					=    kwargs['''candidate_labels''']
							if "hypothesis_template" in kwargs:
											_UpperCAmelCase					=    kwargs['''hypothesis_template''']

							return preprocess_params, {}, {}



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

							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    ffmpeg_read(__UpperCamelCase				,   self.feature_extractor.sampling_rate   )

							if not isinstance(__UpperCamelCase				,   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					=    self.feature_extractor(
							    [audio]				,   sampling_rate=self.feature_extractor.sampling_rate				,   return_tensors='''pt'''   )
							_UpperCAmelCase					=    candidate_labels
							_UpperCAmelCase					=    [hypothesis_template.format(__UpperCamelCase   ) for x in candidate_labels]
							_UpperCAmelCase					=    self.tokenizer(__UpperCamelCase				,   return_tensors=self.framework				,   padding=__UpperCamelCase   )
							_UpperCAmelCase					=    [text_inputs]
							return inputs



			def  lowercase__						(		self     :    Union[str, Any]				,   __UpperCamelCase     :    Optional[int]   )->Any:
							_UpperCAmelCase					=    model_inputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_inputs.pop('''text_inputs'''   )
							if isinstance(text_inputs[0]				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    text_inputs[0]
							else:
											# Batching case.
											_UpperCAmelCase					=    text_inputs[0][0]

							_UpperCAmelCase					=    self.model(**__UpperCamelCase				,   **__UpperCamelCase   )

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



			def  lowercase__						(		self     :    List[str]				,   __UpperCamelCase     :    Any   )->List[Any]:
							_UpperCAmelCase					=    model_outputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_outputs['''logits'''][0]

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

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



