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


__lowercase				,		__lowercase				,		__lowercase							:   Union[str, Any]					=     False, False, False

@dataclass
class 	__lowercase				:
			lowerCamelCase			:		Any	      =			None
			lowerCamelCase			:		Union[str, Any]	      =			True
			lowerCamelCase			:		Dict	      =			True
			lowerCamelCase			:		Any	      =			None
			# Automatically constructed
			lowerCamelCase			:		Tuple	      =			"dict"
			lowerCamelCase			:		List[Any]	      =			pa.struct({"bytes": pa.binary(), "path": pa.string()}    )
			lowerCamelCase			:		List[Any]	      =			field(default="Audio"      ,     init=_A      ,     repr=_A    )

			def __call__(self      ):
								return self.pa_type

			def        UpperCAmelCase__ (self			,   A      ):
								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(__lowerCamelCase			,   __lowerCamelCase      ):
													return {"bytes": None, "path": value}
								elif isinstance(__lowerCamelCase			,   __lowerCamelCase      ):
													return {"bytes": value, "path": None}
								elif "array" in value:
													# convert the audio array to wav bytes
													lowerCamelCase_					:      Union[str, Any]     =		BytesIO()
													sf.write(__lowerCamelCase			,   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!)
																							lowerCamelCase_					:      Tuple     =		np.frombuffer(value['''bytes''']			,   dtype=np.intaa      ).astype(np.floataa      ) / 3_2_7_6_7
																		else:
																							lowerCamelCase_					:      Dict     =		np.memmap(value['''path''']			,   dtype='''h'''			,   mode='''r'''      ).astype(np.floataa      ) / 3_2_7_6_7

																		lowerCamelCase_					:      Optional[int]     =		BytesIO(bytes()      )
																		sf.write(__lowerCamelCase			,   __lowerCamelCase			,   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        UpperCAmelCase__ (self			,   A			,   A = None      ):
								if not self.decode:
													raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.'''      )

								lowerCamelCase_					:      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

								lowerCamelCase_					:      List[str]     =		xsplitext(__lowerCamelCase      )[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:
													lowerCamelCase_					:      int     =		token_per_repo_id or {}
													lowerCamelCase_					:      Tuple     =		path.split('''::'''      )[-1]
													try:
																		lowerCamelCase_					:      List[Any]     =		string_to_dict(__lowerCamelCase			,   config.HUB_DATASETS_URL      )["repo_id"]
																		lowerCamelCase_					:      Tuple     =		token_per_repo_id[repo_id]
													except (ValueError, KeyError):
																		lowerCamelCase_					:      int     =		None

													with xopen(__lowerCamelCase			,   '''rb'''			,   use_auth_token=__lowerCamelCase      ) as f:
																		lowerCamelCase_					:      List[str]     =		sf.read(__lowerCamelCase      )

								else:
													lowerCamelCase_					:      Tuple     =		sf.read(__lowerCamelCase      )

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

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

			def        UpperCAmelCase__ (self      ):
								from .features import Value

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

			def        UpperCAmelCase__ (self			,   A      ):
								if pa.types.is_string(storage.type      ):
													lowerCamelCase_					:      int     =		pa.array([None] * len(__lowerCamelCase      )			,   type=pa.binary()      )
													lowerCamelCase_					:      Any     =		pa.StructArray.from_arrays([bytes_array, storage]			,   ['''bytes''', '''path''']			,   mask=storage.is_null()      )
								elif pa.types.is_binary(storage.type      ):
													lowerCamelCase_					:      Any     =		pa.array([None] * len(__lowerCamelCase      )			,   type=pa.string()      )
													lowerCamelCase_					:      Union[str, 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'''      ):
													lowerCamelCase_					:      List[str]     =		pa.array([Audio().encode_example(__lowerCamelCase      ) 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:
																		lowerCamelCase_					:      Any     =		storage.field('''bytes'''      )
													else:
																		lowerCamelCase_					:      Dict     =		pa.array([None] * len(__lowerCamelCase      )			,   type=pa.binary()      )
													if storage.type.get_field_index('''path'''      ) >= 0:
																		lowerCamelCase_					:      Optional[Any]     =		storage.field('''path'''      )
													else:
																		lowerCamelCase_					:      Union[str, Any]     =		pa.array([None] * len(__lowerCamelCase      )			,   type=pa.string()      )
													lowerCamelCase_					:      int     =		pa.StructArray.from_arrays([bytes_array, path_array]			,   ['''bytes''', '''path''']			,   mask=storage.is_null()      )
								return array_cast(__lowerCamelCase			,   self.pa_type      )






			def        UpperCAmelCase__ (self			,   A      ):

								@no_op_if_value_is_null
								def path_to_bytes(A      ):
													with xopen(__lowerCamelCase			,   '''rb'''      ) as f:
																		lowerCamelCase_					:      int     =		f.read()
													return bytes_

								lowerCamelCase_					:      str     =		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()			,   )
								lowerCamelCase_					:      Optional[int]     =		pa.array(
								    [os.path.basename(__lowerCamelCase      ) if path is not None else None for path in storage.field('''path'''      ).to_pylist()]			,   type=pa.string()			,   )
								lowerCamelCase_					:      Tuple     =		pa.StructArray.from_arrays([bytes_array, path_array]			,   ['''bytes''', '''path''']			,   mask=bytes_array.is_null()      )
								return array_cast(__lowerCamelCase			,   self.pa_type      )




'''simple docstring'''


import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer

class 	__lowercase				(  _lowercase    ):

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Any     =		tempfile.mkdtemp()
								lowerCamelCase_					:      Optional[int]     =		5

								# Realm tok
								lowerCamelCase_					:      Any     =		[
								    '''[UNK]''',
								    '''[CLS]''',
								    '''[SEP]''',
								    '''[PAD]''',
								    '''[MASK]''',
								    '''test''',
								    '''question''',
								    '''this''',
								    '''is''',
								    '''the''',
								    '''first''',
								    '''second''',
								    '''third''',
								    '''fourth''',
								    '''fifth''',
								    '''record''',
								    '''want''',
								    '''##want''',
								    '''##ed''',
								    '''wa''',
								    '''un''',
								    '''runn''',
								    '''##ing''',
								    ''',''',
								    '''low''',
								    '''lowest''',
								]
								lowerCamelCase_					:      Optional[int]     =		os.path.join(self.tmpdirname			,   '''realm_tokenizer'''      )
								os.makedirs(A			,   exist_ok=A      )
								lowerCamelCase_					:      str     =		os.path.join(A			,   VOCAB_FILES_NAMES['''vocab_file''']      )
								with open(self.vocab_file			,   '''w'''			,   encoding='''utf-8'''      ) as vocab_writer:
													vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]      )      )

								lowerCamelCase_					:      List[Any]     =		os.path.join(self.tmpdirname			,   '''realm_block_records'''      )
								os.makedirs(A			,   exist_ok=A      )

			def        UpperCAmelCase__ (self      ):
								return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname			,   '''realm_tokenizer'''      )      )

			def        UpperCAmelCase__ (self      ):
								shutil.rmtree(self.tmpdirname      )

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      int     =		RealmConfig(num_block_records=self.num_block_records      )
								return config

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Optional[int]     =		Dataset.from_dict(
								    {
								        '''id''': ['''0''', '''1'''],
								        '''question''': ['''foo''', '''bar'''],
								        '''answers''': [['''Foo''', '''Bar'''], ['''Bar''']],
								    }      )
								return dataset

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Any     =		np.array(
								    [
								        B'''This is the first record''',
								        B'''This is the second record''',
								        B'''This is the third record''',
								        B'''This is the fourth record''',
								        B'''This is the fifth record''',
								        B'''This is a longer longer longer record''',
								    ]			,   dtype=A			,   )
								return block_records

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Dict     =		RealmRetriever(
								    block_records=self.get_dummy_block_records()			,   tokenizer=self.get_tokenizer()			,   )
								return retriever

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Optional[Any]     =		self.get_config()
								lowerCamelCase_					:      List[Any]     =		self.get_dummy_retriever()
								lowerCamelCase_					:      Dict     =		retriever.tokenizer

								lowerCamelCase_					:      List[Any]     =		np.array([0, 3]			,   dtype='''long'''      )
								lowerCamelCase_					:      int     =		tokenizer(['''Test question''']      ).input_ids
								lowerCamelCase_					:      Any     =		tokenizer(
								    ['''the fourth''']			,   add_special_tokens=A			,   return_token_type_ids=A			,   return_attention_mask=A			,   ).input_ids
								lowerCamelCase_					:      Any     =		config.reader_seq_len

								lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_					:      Tuple     =		retriever(
								    A			,   A			,   answer_ids=A			,   max_length=A			,   return_tensors='''np'''      )

								self.assertEqual(len(A      )			,   2      )
								self.assertEqual(len(A      )			,   2      )
								self.assertEqual(len(A      )			,   2      )
								self.assertEqual(concat_inputs.input_ids.shape			,   (2, 1_0)      )
								self.assertEqual(concat_inputs.attention_mask.shape			,   (2, 1_0)      )
								self.assertEqual(concat_inputs.token_type_ids.shape			,   (2, 1_0)      )
								self.assertEqual(concat_inputs.special_tokens_mask.shape			,   (2, 1_0)      )
								self.assertEqual(
								    tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]      )			,   ['''[CLS]''', '''test''', '''question''', '''[SEP]''', '''this''', '''is''', '''the''', '''first''', '''record''', '''[SEP]''']			,   )
								self.assertEqual(
								    tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]      )			,   ['''[CLS]''', '''test''', '''question''', '''[SEP]''', '''this''', '''is''', '''the''', '''fourth''', '''record''', '''[SEP]''']			,   )

			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      str     =		self.get_config()
								lowerCamelCase_					:      Optional[Any]     =		self.get_dummy_retriever()
								lowerCamelCase_					:      Optional[int]     =		retriever.tokenizer

								lowerCamelCase_					:      int     =		np.array([0, 3, 5]			,   dtype='''long'''      )
								lowerCamelCase_					:      List[Any]     =		tokenizer(['''Test question''']      ).input_ids
								lowerCamelCase_					:      int     =		tokenizer(
								    ['''the fourth''', '''longer longer''']			,   add_special_tokens=A			,   return_token_type_ids=A			,   return_attention_mask=A			,   ).input_ids
								lowerCamelCase_					:      Tuple     =		config.reader_seq_len

								lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_					:      int     =		retriever(
								    A			,   A			,   answer_ids=A			,   max_length=A			,   return_tensors='''np'''      )

								self.assertEqual([False, True, True]			,   A      )
								self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]			,   A      )
								self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]			,   A      )






			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Any     =		self.get_dummy_retriever()
								retriever.save_pretrained(os.path.join(self.tmpdirname			,   '''realm_block_records'''      )      )

								# Test local path
								lowerCamelCase_					:      str     =		retriever.from_pretrained(os.path.join(self.tmpdirname			,   '''realm_block_records'''      )      )
								self.assertEqual(retriever.block_records[0]			,   B'''This is the first record'''      )

								# Test mocked remote path
								with patch('''transformers.models.realm.retrieval_realm.hf_hub_download'''      ) as mock_hf_hub_download:
													lowerCamelCase_					:      Dict     =		os.path.join(
													    os.path.join(self.tmpdirname			,   '''realm_block_records'''      )			,   _REALM_BLOCK_RECORDS_FILENAME      )
													lowerCamelCase_					:      str     =		RealmRetriever.from_pretrained('''google/realm-cc-news-pretrained-openqa'''      )

								self.assertEqual(retriever.block_records[0]			,   B'''This is the first record'''      )


