
'''simple docstring'''
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   :







							'''simple docstring'''

							lowerCAmelCase_ 						=						None
							lowerCAmelCase_ 						=						True
							lowerCAmelCase_ 						=						True
							lowerCAmelCase_ 						=						None
							# Automatically constructed
							lowerCAmelCase_ 						=						"dict"
							lowerCAmelCase_ 						=						pa.struct({"bytes": pa.binary(), "path": pa.string()}				)
							lowerCAmelCase_ 						=						field(default="Audio"					,							init=__snake_case					,							repr=__snake_case				)

							def __call__(self  )					->       Any:
														return self.pa_type

							def 	lowercase							(self  ,  UpperCAmelCase  )					->       dict:
														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(UpperCAmelCase  ,  UpperCAmelCase  ):
																					return {"bytes": None, "path": value}
														elif isinstance(UpperCAmelCase  ,  UpperCAmelCase  ):
																					return {"bytes": value, "path": None}
														elif "array" in value:
																					# convert the audio array to wav bytes
																					_snake_case       =  BytesIO()
																					sf.write(UpperCAmelCase  ,  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       =  np.frombuffer(value["""bytes"""]  ,  dtype=np.intaa  ).astype(np.floataa  ) / 32767
																												else:
																																			_snake_case       =  np.memmap(value["""path"""]  ,  dtype="""h"""  ,  mode="""r"""  ).astype(np.floataa  ) / 32767

																												_snake_case       =  BytesIO(bytes()  )
																												sf.write(UpperCAmelCase  ,  UpperCAmelCase  ,  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 	lowercase							(self  ,  UpperCAmelCase  ,  UpperCAmelCase = None  )					->       dict:
														if not self.decode:
																					raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""  )

														_snake_case,							_snake_case       =  (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       =  xsplitext(UpperCAmelCase  )[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       =  token_per_repo_id or {}
																					_snake_case       =  path.split("""::"""  )[-1]
																					try:
																												_snake_case       =  string_to_dict(UpperCAmelCase  ,  config.HUB_DATASETS_URL  )["""repo_id"""]
																												_snake_case       =  token_per_repo_id[repo_id]
																					except (ValueError, KeyError):
																												_snake_case       =  None

																					with xopen(UpperCAmelCase  ,  """rb"""  ,  use_auth_token=UpperCAmelCase  ) as f:
																												_snake_case,							_snake_case       =  sf.read(UpperCAmelCase  )

														else:
																					_snake_case,							_snake_case       =  sf.read(UpperCAmelCase  )

														_snake_case       =  array.T
														if self.mono:
																					_snake_case       =  librosa.to_mono(UpperCAmelCase  )
														if self.sampling_rate and self.sampling_rate != sampling_rate:
																					_snake_case       =  librosa.resample(UpperCAmelCase  ,  orig_sr=UpperCAmelCase  ,  target_sr=self.sampling_rate  )
																					_snake_case       =  self.sampling_rate

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

							def 	lowercase							(self  )					->       Union["FeatureType", Dict[str, "FeatureType"]]:
														from .features import Value

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

							def 	lowercase							(self  ,  UpperCAmelCase  )					->       pa.StructArray:
														if pa.types.is_string(storage.type  ):
																					_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.binary()  )
																					_snake_case       =  pa.StructArray.from_arrays([bytes_array, storage]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()  )
														elif pa.types.is_binary(storage.type  ):
																					_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.string()  )
																					_snake_case       =  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       =  pa.array([Audio().encode_example(UpperCAmelCase  ) 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       =  storage.field("""bytes"""  )
																					else:
																												_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.binary()  )
																					if storage.type.get_field_index("""path"""  ) >= 0:
																												_snake_case       =  storage.field("""path"""  )
																					else:
																												_snake_case       =  pa.array([None] * len(UpperCAmelCase  )  ,  type=pa.string()  )
																					_snake_case       =  pa.StructArray.from_arrays([bytes_array, path_array]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()  )
														return array_cast(UpperCAmelCase  ,  self.pa_type  )





							def 	lowercase							(self  ,  UpperCAmelCase  )					->       pa.StructArray:

														@no_op_if_value_is_null
														def path_to_bytes(UpperCAmelCase  ):
																					with xopen(UpperCAmelCase  ,  """rb"""  ) as f:
																												_snake_case       =  f.read()
																					return bytes_

														_snake_case       =  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       =  pa.array(
														    [os.path.basename(UpperCAmelCase  ) if path is not None else None for path in storage.field("""path"""  ).to_pylist()]  ,  type=pa.string()  ,  )
														_snake_case       =  pa.StructArray.from_arrays([bytes_array, path_array]  ,  ["""bytes""", """path"""]  ,  mask=bytes_array.is_null()  )
														return array_cast(UpperCAmelCase  ,  self.pa_type  )

'''simple docstring'''
import unittest

from transformers import DebertaConfig, is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
				import torch

				from transformers import (
				    DebertaForMaskedLM,
				    DebertaForQuestionAnswering,
				    DebertaForSequenceClassification,
				    DebertaForTokenClassification,
				    DebertaModel,
				)
				from transformers.models.deberta.modeling_deberta import DEBERTA_PRETRAINED_MODEL_ARCHIVE_LIST


class       _lowerCAmelCase   (   __snake_case				):







							'''simple docstring'''

							def __init__(self  ,  UpperCAmelCase  ,  UpperCAmelCase=13  ,  UpperCAmelCase=7  ,  UpperCAmelCase=True  ,  UpperCAmelCase=True  ,  UpperCAmelCase=True  ,  UpperCAmelCase=True  ,  UpperCAmelCase=99  ,  UpperCAmelCase=32  ,  UpperCAmelCase=5  ,  UpperCAmelCase=4  ,  UpperCAmelCase=37  ,  UpperCAmelCase="gelu"  ,  UpperCAmelCase=0.1  ,  UpperCAmelCase=0.1  ,  UpperCAmelCase=512  ,  UpperCAmelCase=16  ,  UpperCAmelCase=2  ,  UpperCAmelCase=0.02  ,  UpperCAmelCase=False  ,  UpperCAmelCase=True  ,  UpperCAmelCase="None"  ,  UpperCAmelCase=3  ,  UpperCAmelCase=4  ,  UpperCAmelCase=None  ,  )					->       Optional[Any]:
														_snake_case       =  parent
														_snake_case       =  batch_size
														_snake_case       =  seq_length
														_snake_case       =  is_training
														_snake_case       =  use_input_mask
														_snake_case       =  use_token_type_ids
														_snake_case       =  use_labels
														_snake_case       =  vocab_size
														_snake_case       =  hidden_size
														_snake_case       =  num_hidden_layers
														_snake_case       =  num_attention_heads
														_snake_case       =  intermediate_size
														_snake_case       =  hidden_act
														_snake_case       =  hidden_dropout_prob
														_snake_case       =  attention_probs_dropout_prob
														_snake_case       =  max_position_embeddings
														_snake_case       =  type_vocab_size
														_snake_case       =  type_sequence_label_size
														_snake_case       =  initializer_range
														_snake_case       =  num_labels
														_snake_case       =  num_choices
														_snake_case       =  relative_attention
														_snake_case       =  position_biased_input
														_snake_case       =  pos_att_type
														_snake_case       =  scope

							def 	lowercase							(self  )					->       List[Any]:
														_snake_case       =  ids_tensor([self.batch_size, self.seq_length]  ,  self.vocab_size  )

														_snake_case       =  None
														if self.use_input_mask:
																					_snake_case       =  ids_tensor([self.batch_size, self.seq_length]  ,  vocab_size=2  )

														_snake_case       =  None
														if self.use_token_type_ids:
																					_snake_case       =  ids_tensor([self.batch_size, self.seq_length]  ,  self.type_vocab_size  )

														_snake_case       =  None
														_snake_case       =  None
														_snake_case       =  None
														if self.use_labels:
																					_snake_case       =  ids_tensor([self.batch_size]  ,  self.type_sequence_label_size  )
																					_snake_case       =  ids_tensor([self.batch_size, self.seq_length]  ,  self.num_labels  )
																					_snake_case       =  ids_tensor([self.batch_size]  ,  self.num_choices  )

														_snake_case       =  self.get_config()

														return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels

							def 	lowercase							(self  )					->       int:
														return DebertaConfig(
														    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  ,  initializer_range=self.initializer_range  ,  relative_attention=self.relative_attention  ,  position_biased_input=self.position_biased_input  ,  pos_att_type=self.pos_att_type  ,  )

							def 	lowercase							(self  )					->       int:
														_snake_case       =  self.get_config()
														_snake_case       =  300
														return config

							def 	lowercase							(self  ,  UpperCAmelCase  )					->       Dict:
														self.parent.assertListEqual(list(result.loss.size()  )  ,  []  )

							def 	lowercase							(self  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  )					->       Optional[Any]:
														_snake_case       =  DebertaModel(config=UpperCAmelCase  )
														model.to(UpperCAmelCase  )
														model.eval()
														_snake_case       =  model(UpperCAmelCase  ,  attention_mask=UpperCAmelCase  ,  token_type_ids=UpperCAmelCase  )[0]
														_snake_case       =  model(UpperCAmelCase  ,  token_type_ids=UpperCAmelCase  )[0]
														_snake_case       =  model(UpperCAmelCase  )[0]

														self.parent.assertListEqual(list(sequence_output.size()  )  ,  [self.batch_size, self.seq_length, self.hidden_size]  )

							def 	lowercase							(self  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  )					->       str:
														_snake_case       =  DebertaForMaskedLM(config=UpperCAmelCase  )
														model.to(UpperCAmelCase  )
														model.eval()
														_snake_case       =  model(UpperCAmelCase  ,  attention_mask=UpperCAmelCase  ,  token_type_ids=UpperCAmelCase  ,  labels=UpperCAmelCase  )

														self.parent.assertEqual(result.logits.shape  ,  (self.batch_size, self.seq_length, self.vocab_size)  )

							def 	lowercase							(self  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  )					->       List[str]:
														_snake_case       =  self.num_labels
														_snake_case       =  DebertaForSequenceClassification(UpperCAmelCase  )
														model.to(UpperCAmelCase  )
														model.eval()
														_snake_case       =  model(UpperCAmelCase  ,  attention_mask=UpperCAmelCase  ,  token_type_ids=UpperCAmelCase  ,  labels=UpperCAmelCase  )
														self.parent.assertListEqual(list(result.logits.size()  )  ,  [self.batch_size, self.num_labels]  )
														self.check_loss_output(UpperCAmelCase  )

							def 	lowercase							(self  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  )					->       Any:
														_snake_case       =  self.num_labels
														_snake_case       =  DebertaForTokenClassification(config=UpperCAmelCase  )
														model.to(UpperCAmelCase  )
														model.eval()
														_snake_case       =  model(UpperCAmelCase  ,  attention_mask=UpperCAmelCase  ,  token_type_ids=UpperCAmelCase  ,  labels=UpperCAmelCase  )
														self.parent.assertEqual(result.logits.shape  ,  (self.batch_size, self.seq_length, self.num_labels)  )

							def 	lowercase							(self  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  ,  UpperCAmelCase  )					->       Any:
														_snake_case       =  DebertaForQuestionAnswering(config=UpperCAmelCase  )
														model.to(UpperCAmelCase  )
														model.eval()
														_snake_case       =  model(
														    UpperCAmelCase  ,  attention_mask=UpperCAmelCase  ,  token_type_ids=UpperCAmelCase  ,  start_positions=UpperCAmelCase  ,  end_positions=UpperCAmelCase  ,  )
														self.parent.assertEqual(result.start_logits.shape  ,  (self.batch_size, self.seq_length)  )
														self.parent.assertEqual(result.end_logits.shape  ,  (self.batch_size, self.seq_length)  )





							def 	lowercase							(self  )					->       Tuple:
														_snake_case       =  self.prepare_config_and_inputs()
														(
														    (
														    _snake_case
														),							(
														    _snake_case
														),							(
														    _snake_case
														),							(
														    _snake_case
														),							(
														    _snake_case
														),							(
														    _snake_case
														),							(
														    _snake_case
														),							
														)       =  config_and_inputs
														_snake_case       =  {"""input_ids""": input_ids, """token_type_ids""": token_type_ids, """attention_mask""": input_mask}
														return config, inputs_dict


@require_torch
class       _lowerCAmelCase   (   __snake_case					,							__snake_case					,							unittest.TestCase				):







							'''simple docstring'''

							lowerCAmelCase_ 						=						(
							    (
							        DebertaModel,
							        DebertaForMaskedLM,
							        DebertaForSequenceClassification,
							        DebertaForTokenClassification,
							        DebertaForQuestionAnswering,
							    )
							    if is_torch_available()
							    else ()
							)
							lowerCAmelCase_ 						=						(
							    {
							        "feature-extraction": DebertaModel,
							        "fill-mask": DebertaForMaskedLM,
							        "question-answering": DebertaForQuestionAnswering,
							        "text-classification": DebertaForSequenceClassification,
							        "token-classification": DebertaForTokenClassification,
							        "zero-shot": DebertaForSequenceClassification,
							    }
							    if is_torch_available()
							    else {}
							)

							lowerCAmelCase_ 						=						True
							lowerCAmelCase_ 						=						False
							lowerCAmelCase_ 						=						False
							lowerCAmelCase_ 						=						False
							lowerCAmelCase_ 						=						False

							def 	lowercase							(self  )					->       Any:
														_snake_case       =  DebertaModelTester(self  )
														_snake_case       =  ConfigTester(self  ,  config_class=UpperCAmelCase  ,  hidden_size=37  )

							def 	lowercase							(self  )					->       List[str]:
														self.config_tester.run_common_tests()

							def 	lowercase							(self  )					->       Tuple:
														_snake_case       =  self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_deberta_model(*UpperCAmelCase  )

							def 	lowercase							(self  )					->       str:
														_snake_case       =  self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_deberta_for_sequence_classification(*UpperCAmelCase  )

							def 	lowercase							(self  )					->       int:
														_snake_case       =  self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_deberta_for_masked_lm(*UpperCAmelCase  )

							def 	lowercase							(self  )					->       Any:
														_snake_case       =  self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_deberta_for_question_answering(*UpperCAmelCase  )

							def 	lowercase							(self  )					->       str:
														_snake_case       =  self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_deberta_for_token_classification(*UpperCAmelCase  )





							@slow
							def 	lowercase							(self  )					->       Tuple:
														for model_name in DEBERTA_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																					_snake_case       =  DebertaModel.from_pretrained(UpperCAmelCase  )
																					self.assertIsNotNone(UpperCAmelCase  )


@require_torch
@require_sentencepiece
@require_tokenizers
class       _lowerCAmelCase   (   unittest.TestCase				):







							'''simple docstring'''

							@unittest.skip(reason="""Model not available yet"""  )
							def 	lowercase							(self  )					->       Any:
														pass





							@slow
							def 	lowercase							(self  )					->       Dict:
														_snake_case       =  DebertaModel.from_pretrained("""microsoft/deberta-base"""  )

														_snake_case       =  torch.tensor([[0, 31414, 232, 328, 740, 1140, 12695, 69, 46078, 1588, 2]]  )
														_snake_case       =  torch.tensor([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]  )
														with torch.no_grad():
																					_snake_case       =  model(UpperCAmelCase  ,  attention_mask=UpperCAmelCase  )[0]
														# compare the actual values for a slice.
														_snake_case       =  torch.tensor(
														    [[[-0.5986, -0.8055, -0.8462], [1.4484, -0.9348, -0.8059], [0.3123, 0.0032, -1.4131]]]  )
														self.assertTrue(torch.allclose(output[:, 1:4, 1:4]  ,  UpperCAmelCase  ,  atol=1e-4  )  ,  f"""{output[:, 1:4, 1:4]}"""  )