





import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
	from .features import FeatureType


lowerCAmelCase_  , lowerCAmelCase_  , lowerCAmelCase_    =    False, False, False



@dataclass
class  __lowerCAmelCase	:
			lowerCamelCase_					:					int	      =				None
			lowerCamelCase_					:					str	      =				True
			lowerCamelCase_					:					List[str]	      =				True
			lowerCamelCase_					:					List[str]	      =				None
			# Automatically constructed
			lowerCamelCase_					:					Optional[int]	      =				'''dict'''
			lowerCamelCase_					:					Any	      =				pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}					)
			lowerCamelCase_					:					Optional[int]	      =				field(default='''Audio''',					init=_a,					repr=_a					)






			def __call__(self   )					->       str:


							'''simple docstring'''



							return self.pa_type






			def       lowerCamelCase  (self ,							__magic_name__   )					->       dict:


							'''simple docstring'''



							try:
											import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
							except ImportError as err:
											raise ImportError('''To support encoding audio data, please install \'soundfile\'.'''   ) from err
							if isinstance(_SCREAMING_SNAKE_CASE ,							_SCREAMING_SNAKE_CASE   ):
											return {"bytes": None, "path": value}
							elif isinstance(_SCREAMING_SNAKE_CASE ,							_SCREAMING_SNAKE_CASE   ):
											return {"bytes": value, "path": None}
							elif "array" in value:
											# convert the audio array to wav bytes
											snake_case_						:   int           =  BytesIO()
											sf.write(_SCREAMING_SNAKE_CASE ,							value['''array'''] ,							value['''sampling_rate'''] ,							format='''wav'''   )
											return {"bytes": buffer.getvalue(), "path": None}
							elif value.get('''path'''   ) is not None and os.path.isfile(value['''path''']   ):
											# we set "bytes": None to not duplicate the data if they're already available locally
											if value["path"].endswith('''pcm'''   ):
															# "PCM" only has raw audio bytes
															if value.get('''sampling_rate'''   ) is None:
																			# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
																			raise KeyError('''To use PCM files, please specify a \'sampling_rate\' in Audio object'''   )
															if value.get('''bytes'''   ):
																			# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
																			snake_case_						:   Dict           =  np.frombuffer(value['''bytes'''] ,							dtype=np.intaa   ).astype(np.floataa   ) / 3_2767
															else:
																			snake_case_						:   Any           =  np.memmap(value['''path'''] ,							dtype='''h''' ,							mode='''r'''   ).astype(np.floataa   ) / 3_2767

															snake_case_						:   Union[str, Any]           =  BytesIO(bytes()   )
															sf.write(_SCREAMING_SNAKE_CASE ,							_SCREAMING_SNAKE_CASE ,							value['''sampling_rate'''] ,							format='''wav'''   )
															return {"bytes": buffer.getvalue(), "path": None}
											else:
															return {"bytes": None, "path": value.get('''path'''   )}
							elif value.get('''bytes'''   ) is not None or value.get('''path'''   ) is not None:
											# store the audio bytes, and path is used to infer the audio format using the file extension
											return {"bytes": value.get('''bytes'''   ), "path": value.get('''path'''   )}
							else:
											raise ValueError(
											    F'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''   )






			def       lowerCamelCase  (self ,							__magic_name__ ,							__magic_name__ = None   )					->       dict:


							'''simple docstring'''



							if not self.decode:
											raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.'''   )

							snake_case_						:   Optional[int]           =  (value['''path'''], BytesIO(value['''bytes''']   )) if value['''bytes'''] is not None else (value['''path'''], None)
							if path is None and file is None:
											raise ValueError(F'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.'''   )

							try:
											import librosa
											import soundfile as sf
							except ImportError as err:
											raise ImportError('''To support decoding audio files, please install \'librosa\' and \'soundfile\'.'''   ) from err

							snake_case_						:   Union[str, Any]           =  xsplitext(_SCREAMING_SNAKE_CASE   )[1][1:].lower() if path is not None else None
							if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
											raise RuntimeError(
											    '''Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '''
											    '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '''   )
							elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
											raise RuntimeError(
											    '''Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '''
											    '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '''   )

							if file is None:
											snake_case_						:   Optional[Any]           =  token_per_repo_id or {}
											snake_case_						:   int           =  path.split('''::'''   )[-1]
											try:
															snake_case_						:   Tuple           =  string_to_dict(_SCREAMING_SNAKE_CASE ,							config.HUB_DATASETS_URL   )['''repo_id''']
															snake_case_						:   int           =  token_per_repo_id[repo_id]
											except (ValueError, KeyError):
															snake_case_						:   int           =  None

											with xopen(_SCREAMING_SNAKE_CASE ,							'''rb''' ,							use_auth_token=_SCREAMING_SNAKE_CASE   ) as f:
															snake_case_						:   Any           =  sf.read(_SCREAMING_SNAKE_CASE   )

							else:
											snake_case_						:   Optional[int]           =  sf.read(_SCREAMING_SNAKE_CASE   )

							snake_case_						:   Optional[int]           =  array.T
							if self.mono:
											snake_case_						:   int           =  librosa.to_mono(_SCREAMING_SNAKE_CASE   )
							if self.sampling_rate and self.sampling_rate != sampling_rate:
											snake_case_						:   List[Any]           =  librosa.resample(_SCREAMING_SNAKE_CASE ,							orig_sr=_SCREAMING_SNAKE_CASE ,							target_sr=self.sampling_rate   )
											snake_case_						:   List[str]           =  self.sampling_rate

							return {"path": path, "array": array, "sampling_rate": sampling_rate}






			def       lowerCamelCase  (self   )					->       Union["FeatureType", Dict[str, "FeatureType"]]:


							'''simple docstring'''



							from .features import Value

							if self.decode:
											raise ValueError('''Cannot flatten a decoded Audio feature.'''   )
							return {
							    "bytes": Value('''binary'''   ),
							    "path": Value('''string'''   ),
							}






			def       lowerCamelCase  (self ,							__magic_name__   )					->       pa.StructArray:


							'''simple docstring'''



							if pa.types.is_string(storage.type   ):
											snake_case_						:   Optional[Any]           =  pa.array([None] * len(_SCREAMING_SNAKE_CASE   ) ,							type=pa.binary()   )
											snake_case_						:   str           =  pa.StructArray.from_arrays([bytes_array, storage] ,							['''bytes''', '''path'''] ,							mask=storage.is_null()   )
							elif pa.types.is_binary(storage.type   ):
											snake_case_						:   int           =  pa.array([None] * len(_SCREAMING_SNAKE_CASE   ) ,							type=pa.string()   )
											snake_case_						:   Optional[Any]           =  pa.StructArray.from_arrays([storage, path_array] ,							['''bytes''', '''path'''] ,							mask=storage.is_null()   )
							elif pa.types.is_struct(storage.type   ) and storage.type.get_all_field_indices('''array'''   ):
											snake_case_						:   List[str]           =  pa.array([Audio().encode_example(_SCREAMING_SNAKE_CASE   ) if x is not None else None for x in storage.to_pylist()]   )
							elif pa.types.is_struct(storage.type   ):
											if storage.type.get_field_index('''bytes'''   ) >= 0:
															snake_case_						:   List[str]           =  storage.field('''bytes'''   )
											else:
															snake_case_						:   Optional[int]           =  pa.array([None] * len(_SCREAMING_SNAKE_CASE   ) ,							type=pa.binary()   )
											if storage.type.get_field_index('''path'''   ) >= 0:
															snake_case_						:   List[str]           =  storage.field('''path'''   )
											else:
															snake_case_						:   Optional[int]           =  pa.array([None] * len(_SCREAMING_SNAKE_CASE   ) ,							type=pa.string()   )
											snake_case_						:   int           =  pa.StructArray.from_arrays([bytes_array, path_array] ,							['''bytes''', '''path'''] ,							mask=storage.is_null()   )
							return array_cast(_SCREAMING_SNAKE_CASE ,							self.pa_type   )






			def       lowerCamelCase  (self ,							__magic_name__   )					->       pa.StructArray:


							'''simple docstring'''



							@no_op_if_value_is_null
							def path_to_bytes(__magic_name__   ):
											with xopen(_SCREAMING_SNAKE_CASE ,							'''rb'''   ) as f:
															snake_case_						:   str           =  f.read()
											return bytes_

							snake_case_						:   Dict           =  pa.array(
							    [
							        (path_to_bytes(x['''path''']   ) if x['''bytes'''] is None else x['''bytes''']) if x is not None else None
							        for x in storage.to_pylist()
							    ] ,							type=pa.binary() ,							)
							snake_case_						:   Union[str, Any]           =  pa.array(
							    [os.path.basename(_SCREAMING_SNAKE_CASE   ) if path is not None else None for path in storage.field('''path'''   ).to_pylist()] ,							type=pa.string() ,							)
							snake_case_						:   Tuple           =  pa.StructArray.from_arrays([bytes_array, path_array] ,							['''bytes''', '''path'''] ,							mask=bytes_array.is_null()   )
							return array_cast(_SCREAMING_SNAKE_CASE ,							self.pa_type   )







import unittest

import numpy as np

from transformers import AlbertConfig, is_flax_available
from transformers.testing_utils import require_flax, slow

from ...test_modeling_flax_common import FlaxModelTesterMixin, ids_tensor, random_attention_mask


if is_flax_available():
	import jax.numpy as jnp

	from transformers.models.albert.modeling_flax_albert import (
	    FlaxAlbertForMaskedLM,
	    FlaxAlbertForMultipleChoice,
	    FlaxAlbertForPreTraining,
	    FlaxAlbertForQuestionAnswering,
	    FlaxAlbertForSequenceClassification,
	    FlaxAlbertForTokenClassification,
	    FlaxAlbertModel,
	)



class  __lowerCAmelCase	(      unittest.TestCase					):






			def __init__(self ,							__magic_name__ ,							__magic_name__=13 ,							__magic_name__=7 ,							__magic_name__=True ,							__magic_name__=True ,							__magic_name__=True ,							__magic_name__=True ,							__magic_name__=99 ,							__magic_name__=32 ,							__magic_name__=5 ,							__magic_name__=4 ,							__magic_name__=37 ,							__magic_name__="gelu" ,							__magic_name__=0.1 ,							__magic_name__=0.1 ,							__magic_name__=512 ,							__magic_name__=16 ,							__magic_name__=2 ,							__magic_name__=0.02 ,							__magic_name__=4 ,							)					->       Tuple:


							'''simple docstring'''



							snake_case_						:   Union[str, Any]           =  parent
							snake_case_						:   Optional[Any]           =  batch_size
							snake_case_						:   List[Any]           =  seq_length
							snake_case_						:   Tuple           =  is_training
							snake_case_						:   List[str]           =  use_attention_mask
							snake_case_						:   Any           =  use_token_type_ids
							snake_case_						:   Dict           =  use_labels
							snake_case_						:   Optional[Any]           =  vocab_size
							snake_case_						:   Dict           =  hidden_size
							snake_case_						:   List[Any]           =  num_hidden_layers
							snake_case_						:   Union[str, Any]           =  num_attention_heads
							snake_case_						:   Any           =  intermediate_size
							snake_case_						:   Optional[int]           =  hidden_act
							snake_case_						:   Optional[int]           =  hidden_dropout_prob
							snake_case_						:   Optional[Any]           =  attention_probs_dropout_prob
							snake_case_						:   Optional[int]           =  max_position_embeddings
							snake_case_						:   Optional[int]           =  type_vocab_size
							snake_case_						:   List[Any]           =  type_sequence_label_size
							snake_case_						:   Dict           =  initializer_range
							snake_case_						:   Dict           =  num_choices






			def       lowerCamelCase  (self   )					->       Tuple:


							'''simple docstring'''



							snake_case_						:   int           =  ids_tensor([self.batch_size, self.seq_length] ,							self.vocab_size   )

							snake_case_						:   Any           =  None
							if self.use_attention_mask:
											snake_case_						:   List[Any]           =  random_attention_mask([self.batch_size, self.seq_length]   )

							snake_case_						:   List[Any]           =  None
							if self.use_token_type_ids:
											snake_case_						:   Dict           =  ids_tensor([self.batch_size, self.seq_length] ,							self.type_vocab_size   )

							snake_case_						:   List[Any]           =  AlbertConfig(
							    vocab_size=self.vocab_size ,							hidden_size=self.hidden_size ,							num_hidden_layers=self.num_hidden_layers ,							num_attention_heads=self.num_attention_heads ,							intermediate_size=self.intermediate_size ,							hidden_act=self.hidden_act ,							hidden_dropout_prob=self.hidden_dropout_prob ,							attention_probs_dropout_prob=self.attention_probs_dropout_prob ,							max_position_embeddings=self.max_position_embeddings ,							type_vocab_size=self.type_vocab_size ,							is_decoder=__magic_name__ ,							initializer_range=self.initializer_range ,							)

							return config, input_ids, token_type_ids, attention_mask






			def       lowerCamelCase  (self   )					->       List[str]:


							'''simple docstring'''



							snake_case_						:   Tuple           =  self.prepare_config_and_inputs()
							snake_case_      ,							snake_case_      ,							snake_case_      ,							snake_case_						:   Optional[int]           =  config_and_inputs
							snake_case_						:   int           =  {'''input_ids''': input_ids, '''token_type_ids''': token_type_ids, '''attention_mask''': attention_mask}
							return config, inputs_dict




@require_flax
class  __lowerCAmelCase	(      _a,					unittest.TestCase					):
			lowerCamelCase_					:					Optional[int]	      =				(
			    (
			        FlaxAlbertModel,
			        FlaxAlbertForPreTraining,
			        FlaxAlbertForMaskedLM,
			        FlaxAlbertForMultipleChoice,
			        FlaxAlbertForQuestionAnswering,
			        FlaxAlbertForSequenceClassification,
			        FlaxAlbertForTokenClassification,
			        FlaxAlbertForQuestionAnswering,
			    )
			    if is_flax_available()
			    else ()
			)






			def       lowerCamelCase  (self   )					->       List[str]:


							'''simple docstring'''



							snake_case_						:   Optional[Any]           =  FlaxAlbertModelTester(self   )






			@slow
			def       lowerCamelCase  (self   )					->       Tuple:


							'''simple docstring'''



							for model_class_name in self.all_model_classes:
											snake_case_						:   Dict           =  model_class_name.from_pretrained('''albert-base-v2'''   )
											snake_case_						:   Union[str, Any]           =  model(np.ones((1, 1)   )   )
											self.assertIsNotNone(__magic_name__   )




@require_flax
class  __lowerCAmelCase	(      unittest.TestCase					):






			@slow
			def       lowerCamelCase  (self   )					->       Optional[Any]:


							'''simple docstring'''



							snake_case_						:   Optional[Any]           =  FlaxAlbertModel.from_pretrained('''albert-base-v2'''   )
							snake_case_						:   Optional[int]           =  np.array([[0, 345, 232, 328, 740, 140, 1695, 69, 6078, 1588, 2]]   )
							snake_case_						:   Dict           =  np.array([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]   )
							snake_case_						:   Union[str, Any]           =  model(__magic_name__ ,							attention_mask=__magic_name__   )[0]
							snake_case_						:   Tuple           =  (1, 11, 768)
							self.assertEqual(output.shape ,							__magic_name__   )
							snake_case_						:   str           =  np.array(
							    [[[-0.6_513, 1.5_035, -0.2_766], [-0.6_515, 1.5_046, -0.2_780], [-0.6_512, 1.5_049, -0.2_784]]]   )

							self.assertTrue(jnp.allclose(output[:, 1:4, 1:4] ,							__magic_name__ ,							atol=1e-4   )   )
