




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


if TYPE_CHECKING:
  pass

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


_snake_case            = logging.get_logger(__name__)

_snake_case            = {
    "kakaobrain/align-base": "https://huggingface.co/kakaobrain/align-base/resolve/main/config.json",
}







class 		lowercase							(     UpperCamelCase__	):
     _a											=		"align_text_model"

     def __init__(						self	,  _a=3_0522	,  _a=768	,  _a=12	,  _a=12	,  _a=3072	,  _a="gelu"	,  _a=0.1	,  _a=0.1	,  _a=512	,  _a=2	,  _a=0.02	,  _a=1e-12	,  _a=0	,  _a="absolute"	,  _a=True	,  **_a	,  )		->							Dict:
           super().__init__(**_a       )

           _A							:  Any    =		vocab_size
           _A							:  Optional[Any]    =		hidden_size
           _A							:  List[Any]    =		num_hidden_layers
           _A							:  Union[str, Any]    =		num_attention_heads
           _A							:  List[str]    =		hidden_act
           _A							:  Union[str, Any]    =		intermediate_size
           _A							:  Dict    =		hidden_dropout_prob
           _A							:  Optional[int]    =		attention_probs_dropout_prob
           _A							:  Optional[int]    =		max_position_embeddings
           _A							:  Any    =		type_vocab_size
           _A							:  List[Any]    =		initializer_range
           _A							:  Optional[Any]    =		layer_norm_eps
           _A							:  str    =		position_embedding_type
           _A							:  Any    =		use_cache
           _A							:  int    =		pad_token_id



     @classmethod
     def       a__			(						cls	,  _a	,  **_a       )		->							"PretrainedConfig":
           cls._set_token_in_kwargs(_a       )

           _A    ,     _A							:  Union[str, Any]    =		cls.get_config_dict(_a	,  **_a       )

           # get the text config dict if we are loading from AlignConfig
           if config_dict.get("""model_type"""       ) == "align":
                 _A							:  List[Any]    =		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(_a	,  **_a       )







class 		lowercase							(     UpperCamelCase__	):
     _a											=		"align_vision_model"

     def __init__(						self	,  _a = 3	,  _a = 600	,  _a = 2.0	,  _a = 3.1	,  _a = 8	,  _a = [3, 3, 5, 3, 5, 5, 3]	,  _a = [32, 16, 24, 40, 80, 112, 192]	,  _a = [16, 24, 40, 80, 112, 192, 320]	,  _a = []	,  _a = [1, 2, 2, 2, 1, 2, 1]	,  _a = [1, 2, 2, 3, 3, 4, 1]	,  _a = [1, 6, 6, 6, 6, 6, 6]	,  _a = 0.25	,  _a = "swish"	,  _a = 2560	,  _a = "mean"	,  _a = 0.02	,  _a = 0.001	,  _a = 0.99	,  _a = 0.2	,  **_a	,  )		->							str:
           super().__init__(**_a       )

           _A							:  Optional[Any]    =		num_channels
           _A							:  int    =		image_size
           _A							:  int    =		width_coefficient
           _A							:  Any    =		depth_coefficient
           _A							:  List[str]    =		depth_divisor
           _A							:  int    =		kernel_sizes
           _A							:  Union[str, Any]    =		in_channels
           _A							:  List[Any]    =		out_channels
           _A							:  Tuple    =		depthwise_padding
           _A							:  Union[str, Any]    =		strides
           _A							:  int    =		num_block_repeats
           _A							:  Any    =		expand_ratios
           _A							:  Optional[int]    =		squeeze_expansion_ratio
           _A							:  str    =		hidden_act
           _A							:  Optional[Any]    =		hidden_dim
           _A							:  List[Any]    =		pooling_type
           _A							:  List[str]    =		initializer_range
           _A							:  Union[str, Any]    =		batch_norm_eps
           _A							:  Any    =		batch_norm_momentum
           _A							:  str    =		drop_connect_rate
           _A							:  List[str]    =		sum(_a       ) * 4



     @classmethod
     def       a__			(						cls	,  _a	,  **_a       )		->							"PretrainedConfig":
           cls._set_token_in_kwargs(_a       )

           _A    ,     _A							:  Any    =		cls.get_config_dict(_a	,  **_a       )

           # get the vision config dict if we are loading from AlignConfig
           if config_dict.get("""model_type"""       ) == "align":
                 _A							:  Tuple    =		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(_a	,  **_a       )







class 		lowercase							(     UpperCamelCase__	):
     _a											=		"align"
     _a											=		True

     def __init__(						self	,  _a=None	,  _a=None	,  _a=640	,  _a=1.0	,  _a=0.02	,  **_a	,  )		->							List[str]:
           super().__init__(**_a       )

           if text_config is None:
                 _A							:  List[Any]    =		{}
                 logger.info("""text_config is None. Initializing the AlignTextConfig with default values."""       )

           if vision_config is None:
                 _A							:  str    =		{}
                 logger.info("""vision_config is None. Initializing the AlignVisionConfig with default values."""       )

           _A							:  str    =		AlignTextConfig(**_a       )
           _A							:  Optional[int]    =		AlignVisionConfig(**_a       )

           _A							:  Dict    =		projection_dim
           _A							:  Union[str, Any]    =		temperature_init_value
           _A							:  Any    =		initializer_range

     @classmethod
     def       a__			(						cls	,  _a	,  _a	,  **_a       )		->							Tuple:
           return cls(text_config=text_config.to_dict()	,  vision_config=vision_config.to_dict()	,  **_a       )



     def       a__			(						self       )		->							str:
           _A							:  Optional[int]    =		copy.deepcopy(self.__dict__       )
           _A							:  List[str]    =		self.text_config.to_dict()
           _A							:  Any    =		self.vision_config.to_dict()
           _A							:  Tuple    =		self.__class__.model_type
           return output







import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
  import torch

if is_datasets_available():
  from datasets import load_dataset

_snake_case            = random.Random()







def        lowerCAmelCase_			(  snake_case_,snake_case_=1.0,snake_case_=None,snake_case_=None  ):
      if rng is None:
            _A							:  str    =		global_rng

      _A							:  List[Any]    =		[]
      for batch_idx in range(shape[0]  ):
            values.append([]  )
            for _ in range(shape[1]  ):
                  values[-1].append(rng.random() * scale  )

      return values







class 		lowercase							(     unittest.TestCase	):

     def __init__(						self	,  _a	,  _a=7	,  _a=400	,  _a=2000	,  _a=2048	,  _a=128	,  _a=1	,  _a=512	,  _a=30	,  _a=4_4100	,  )		->							Tuple:
           _A							:  Any    =		parent
           _A							:  str    =		batch_size
           _A							:  Union[str, Any]    =		min_seq_length
           _A							:  int    =		max_seq_length
           _A							:  List[str]    =		(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
           _A							:  Tuple    =		spectrogram_length
           _A							:  int    =		feature_size
           _A							:  str    =		num_audio_channels
           _A							:  Tuple    =		hop_length
           _A							:  List[str]    =		chunk_length
           _A							:  Union[str, Any]    =		sampling_rate

     def       a__			(						self       )		->							Tuple:
           return {
               "spectrogram_length": self.spectrogram_length,
               "feature_size": self.feature_size,
               "num_audio_channels": self.num_audio_channels,
               "hop_length": self.hop_length,
               "chunk_length": self.chunk_length,
               "sampling_rate": self.sampling_rate,
           }



     def       a__			(						self	,  _a=False	,  _a=False       )		->							Optional[int]:
           def _flatten(_a       ):
                 return list(itertools.chain(*_a       )       )

           if equal_length:
                 _A							:  List[str]    =		[floats_list((self.max_seq_length, self.feature_size)       ) for _ in range(self.batch_size       )]
           else:
                 # make sure that inputs increase in size
                 _A							:  Union[str, Any]    =		[
                     floats_list((x, self.feature_size)       )
                     for x in range(self.min_seq_length	,  self.max_seq_length	,  self.seq_length_diff       )
                 ]
           if numpify:
                 _A							:  List[Any]    =		[np.asarray(_a       ) for x in speech_inputs]
           return speech_inputs







@require_torch
@require_torchaudio
class 		lowercase							(     UpperCamelCase__,unittest.TestCase	):
     _a											=		TvltFeatureExtractor

     def       a__			(						self       )		->							Any:
           _A							:  int    =		TvltFeatureExtractionTester(self       )

     def       a__			(						self       )		->							List[Any]:
           _A							:  Tuple    =		self.feature_extraction_class(**self.feat_extract_dict       )
           self.assertTrue(hasattr(_a	,  """spectrogram_length"""       )       )
           self.assertTrue(hasattr(_a	,  """feature_size"""       )       )
           self.assertTrue(hasattr(_a	,  """num_audio_channels"""       )       )
           self.assertTrue(hasattr(_a	,  """hop_length"""       )       )
           self.assertTrue(hasattr(_a	,  """chunk_length"""       )       )
           self.assertTrue(hasattr(_a	,  """sampling_rate"""       )       )

     def       a__			(						self       )		->							Optional[int]:
           _A							:  str    =		self.feature_extraction_class(**self.feat_extract_dict       )

           with tempfile.TemporaryDirectory() as tmpdirname:
                 _A							:  Tuple    =		feat_extract_first.save_pretrained(_a       )[0]
                 check_json_file_has_correct_format(_a       )
                 _A							:  Optional[int]    =		self.feature_extraction_class.from_pretrained(_a       )

           _A							:  Optional[Any]    =		feat_extract_first.to_dict()
           _A							:  int    =		feat_extract_second.to_dict()
           _A							:  int    =		dict_first.pop("""mel_filters"""       )
           _A							:  Optional[int]    =		dict_second.pop("""mel_filters"""       )
           self.assertTrue(np.allclose(_a	,  _a       )       )
           self.assertEqual(_a	,  _a       )

     def       a__			(						self       )		->							int:
           _A							:  str    =		self.feature_extraction_class(**self.feat_extract_dict       )

           with tempfile.TemporaryDirectory() as tmpdirname:
                 _A							:  List[str]    =		os.path.join(_a	,  """feat_extract.json"""       )
                 feat_extract_first.to_json_file(_a       )
                 _A							:  Union[str, Any]    =		self.feature_extraction_class.from_json_file(_a       )

           _A							:  Optional[Any]    =		feat_extract_first.to_dict()
           _A							:  Union[str, Any]    =		feat_extract_second.to_dict()
           _A							:  List[Any]    =		dict_first.pop("""mel_filters"""       )
           _A							:  Optional[Any]    =		dict_second.pop("""mel_filters"""       )
           self.assertTrue(np.allclose(_a	,  _a       )       )
           self.assertEqual(_a	,  _a       )

     def       a__			(						self       )		->							Optional[Any]:
           # Initialize feature_extractor
           _A							:  Tuple    =		self.feature_extraction_class(**self.feat_extract_dict       )

           # create three inputs of length 800, 1000, and 1200
           _A							:  Optional[int]    =		[floats_list((1, x)       )[0] for x in range(800	,  1400	,  200       )]
           _A							:  List[str]    =		[np.asarray(_a       ) for speech_input in speech_inputs]

           # Test not batched input
           _A							:  Optional[int]    =		feature_extractor(np_speech_inputs[0]	,  return_tensors="""np"""	,  sampling_rate=4_4100       ).audio_values

           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

           # Test batched
           _A							:  str    =		feature_extractor(_a	,  return_tensors="""np"""	,  sampling_rate=4_4100       ).audio_values

           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

           # Test audio masking
           _A							:  Optional[Any]    =		feature_extractor(
               _a	,  return_tensors="""np"""	,  sampling_rate=4_4100	,  mask_audio=_a       ).audio_values

           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

           # Test 2-D numpy arrays are batched.
           _A							:  Dict    =		[floats_list((1, x)       )[0] for x in (800, 800, 800)]
           _A							:  Optional[int]    =		np.asarray(_a       )
           _A							:  Optional[Any]    =		feature_extractor(_a	,  return_tensors="""np"""	,  sampling_rate=4_4100       ).audio_values
           self.assertTrue(encoded_audios.ndim == 4       )
           self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size       )
           self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length       )
           self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels       )

     def       a__			(						self	,  _a       )		->							str:
           _A							:  Optional[Any]    =		load_dataset("""hf-internal-testing/librispeech_asr_dummy"""	,  """clean"""	,  split="""validation"""       )
           # automatic decoding with librispeech
           _A							:  Dict    =		ds.sort("""id"""       ).select(range(_a       )       )[:num_samples]["""audio"""]

           return [x["array"] for x in speech_samples]



     def       a__			(						self       )		->							Optional[Any]:
           _A							:  List[str]    =		self._load_datasamples(1       )
           _A							:  List[str]    =		TvltFeatureExtractor()
           _A							:  List[Any]    =		feature_extractor(_a	,  return_tensors="""pt"""       ).audio_values

           self.assertEquals(audio_values.shape	,  (1, 1, 192, 128)       )

           _A							:  int    =		torch.tensor([[-0.3032, -0.2708], [-0.4434, -0.4007]]       )
           self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]	,  _a	,  atol=1e-4       )       )

