




import json
import os
from collections.abc import Mapping
from typing import Dict, List, Optional, Union

import numpy as np
import torch

from .imports import is_safetensors_available
def        lowerCamelCase_     (     SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE=None ):
			'''simple docstring'''
			SCREAMING_SNAKE_CASE     =       None
			# Check the string instead of the dtype to be compatible with versions of PyTorch that don't have bfloat16.
			if str(weight.dtype ) == "torch.bfloat16":
						# Need to reinterpret the underlined data as int16 since NumPy does not handle bfloat16s.
						SCREAMING_SNAKE_CASE     =       weight.view(torch.intaa )
						SCREAMING_SNAKE_CASE     =       """bfloat16"""
			SCREAMING_SNAKE_CASE     =       weight.cpu().numpy()
			SCREAMING_SNAKE_CASE     =       os.path.join(_UpperCamelCase       ,						f"""{weight_name}.dat""" )
			if index is not None:
						if dtype is None:
									SCREAMING_SNAKE_CASE     =       str(array.dtype )
						SCREAMING_SNAKE_CASE     =       {"""dtype""": dtype, """shape""": list(array.shape )}
			if array.ndim == 0:
						SCREAMING_SNAKE_CASE     =       array[None]
			SCREAMING_SNAKE_CASE     =       np.memmap(_UpperCamelCase       ,						dtype=array.dtype       ,						mode="""w+"""       ,						shape=array.shape )
			SCREAMING_SNAKE_CASE     =       array[:]
			file_array.flush()
			return index
def        lowerCamelCase_     (     SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE ):
			'''simple docstring'''
			SCREAMING_SNAKE_CASE     =       tuple(weight_info["""shape"""] )
			if shape == ():
						# NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
						SCREAMING_SNAKE_CASE     =       (1,)

			SCREAMING_SNAKE_CASE     =       weight_info["""dtype"""]
			if dtype == "bfloat16":
						# NumPy does not support bfloat16 so this was saved as a int16
						SCREAMING_SNAKE_CASE     =       """int16"""

			SCREAMING_SNAKE_CASE     =       np.memmap(_UpperCamelCase       ,						dtype=_UpperCamelCase       ,						shape=_UpperCamelCase       ,						mode="""r""" )

			if len(weight_info["""shape"""] ) == 0:
						SCREAMING_SNAKE_CASE     =       weight[0]
			SCREAMING_SNAKE_CASE     =       torch.tensor(_UpperCamelCase )
			if weight_info["dtype"] == "bfloat16":
						SCREAMING_SNAKE_CASE     =       weight.view(torch.bfloataa )

			return weight
def        lowerCamelCase_     (     SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE ):
			'''simple docstring'''

			if index is None or len(_UpperCamelCase ) == 0:
						# Nothing to save
						return

			SCREAMING_SNAKE_CASE     =       os.path.join(_UpperCamelCase       ,						"""index.json""" )
			if os.path.isfile(_UpperCamelCase ):
						with open(_UpperCamelCase       ,						"""r"""       ,						encoding="""utf-8""" ) as f:
									SCREAMING_SNAKE_CASE     =       json.load(_UpperCamelCase )
			else:
						SCREAMING_SNAKE_CASE     =       {}
			current_index.update(_UpperCamelCase )

			with open(_UpperCamelCase       ,						"""w"""       ,						encoding="""utf-8""" ) as f:
						json.dump(_UpperCamelCase       ,						_UpperCamelCase       ,						indent=2 )
def        lowerCamelCase_     (     SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE ):
			'''simple docstring'''
			os.makedirs(_UpperCamelCase       ,						exist_ok=_UpperCamelCase )
			SCREAMING_SNAKE_CASE     =       {}
			for name, parameter in state_dict.items():
						SCREAMING_SNAKE_CASE     =       offload_weight(_UpperCamelCase       ,						_UpperCamelCase       ,						_UpperCamelCase       ,						index=_UpperCamelCase )

			# Update index
			save_offload_index(_UpperCamelCase       ,						_UpperCamelCase )




class       UpperCAmelCase_   ( __SCREAMING_SNAKE_CASE							):




							'''simple docstring'''



							def __init__(       self							:						Dict							,  a							:						Mapping							,  a							:						str      )      ->       Any:
										SCREAMING_SNAKE_CASE     =       dataset
										SCREAMING_SNAKE_CASE     =       prefix





							def __getitem__(       self							:						List[Any]							,  a							:						str      )      ->       str:
										return self.dataset[f"""{self.prefix}{key}"""]





							def __iter__(       self							:						Tuple      )      ->       Optional[Any]:
										return iter([key for key in self.dataset if key.startswith(self.prefix      )]      )







							def __len__(       self							:						Tuple      )      ->       Dict:
										return len(self.dataset      )





class       UpperCAmelCase_   ( __SCREAMING_SNAKE_CASE							):




							'''simple docstring'''



							def __init__(       self							:						Any							,  a							:						Dict[str, torch.Tensor] = None							,  a							:						Optional[Union[str, os.PathLike]] = None							,  a							:						Mapping = None							,  a							:						Tuple=None							,  )      ->       List[str]:
										if state_dict is None and save_folder is None:
													raise ValueError("""Need either a `state_dict` or a `save_folder` containing offloaded weights."""      )

										SCREAMING_SNAKE_CASE     =       {} if state_dict is None else state_dict
										SCREAMING_SNAKE_CASE     =       save_folder
										if index is None and save_folder is not None:
													with open(os.path.join(lowerCamelCase_							,  """index.json"""      )      ) as f:
																SCREAMING_SNAKE_CASE     =       json.load(lowerCamelCase_      )
										SCREAMING_SNAKE_CASE     =       {} if index is None else index
										SCREAMING_SNAKE_CASE     =       list(self.state_dict.keys()      )
										self.all_keys.extend([key for key in self.index if key not in self.all_keys]      )
										SCREAMING_SNAKE_CASE     =       device





							def __getitem__(       self							:						List[Any]							,  a							:						str      )      ->       Union[str, Any]:
										# State dict gets priority
										if key in self.state_dict:
													return self.state_dict[key]
										SCREAMING_SNAKE_CASE     =       self.index[key]
										if weight_info.get("""safetensors_file"""      ) is not None:
													if not is_safetensors_available():
																raise ImportError("""These offloaded weights require the use of safetensors: `pip install safetensors`."""      )

													from safetensors import safe_open

													SCREAMING_SNAKE_CASE     =       """cpu""" if self.device is None else self.device
													with safe_open(weight_info["""safetensors_file"""]							,  framework="""pt"""							,  device=lowerCamelCase_      ) as f:
																SCREAMING_SNAKE_CASE     =       f.get_tensor(weight_info.get("""weight_name"""							,  lowerCamelCase_      )      )

													if "dtype" in weight_info:
																return tensor.to(getattr(lowerCamelCase_							,  weight_info["""dtype"""]      )      )
													else:
																return tensor

										SCREAMING_SNAKE_CASE     =       os.path.join(self.save_folder							,  f"""{key}.dat"""      )
										return load_offloaded_weight(lowerCamelCase_							,  lowerCamelCase_      )





							def __iter__(       self							:						Dict      )      ->       int:
										return iter(self.all_keys      )







							def __len__(       self							:						Tuple      )      ->       List[Any]:
										return len(self.all_keys      )





def        lowerCamelCase_     (     SCREAMING_SNAKE_CASE       ,						SCREAMING_SNAKE_CASE ):
			'''simple docstring'''
			SCREAMING_SNAKE_CASE     =       {}
			for module_name in submodule_names:
						# We want to catch module_name parameter (module_name.xxx) or potentially module_name, but not any of the
						# submodules that could being like module_name (transformers.h.1 and transformers.h.10 for instance)
						result.update(
						    {
						        key: param
						        for key, param in state_dict.items()
						        if key == module_name or key.startswith(module_name + """.""" )
						    } )
			return result










"""simple docstring"""







import inspect
import json
import os
import tempfile
import unittest

from transformers.models.speech_to_text_a import SpeechaTextaTokenizer
from transformers.models.speech_to_text_a.tokenization_speech_to_text_a import VOCAB_FILES_NAMES

from ...test_tokenization_common import TokenizerTesterMixin
class 				lowerCamelCase							(      __SCREAMING_SNAKE_CASE					,       unittest.TestCase			):


							"""simple docstring"""
							SCREAMING_SNAKE_CASE              =   SpeechaTextaTokenizer
							SCREAMING_SNAKE_CASE              =   False






							def        lowerCAmelCase  (    self			:  Dict					):
													super().setUp()

													UpperCAmelCase__												=     '''<s> <pad> </s> <unk> here@@ a couple of@@ words for the he@@ re@@ vocab'''.split(''' '''					)
													UpperCAmelCase__												=     ['''he re</w> 123''', '''here a 1456''']
													UpperCAmelCase__												=     dict(zip(lowerCamelCase_		,			range(len(lowerCamelCase_					)					)					)					)

													UpperCAmelCase__												=     {'''pad_token''': '''<pad>''', '''unk_token''': '''<unk>''', '''bos_token''': '''<s>''', '''eos_token''': '''</s>'''}

													UpperCAmelCase__												=     tempfile.mkdtemp()
													UpperCAmelCase__												=     os.path.join(self.tmpdirname		,			VOCAB_FILES_NAMES['''vocab_file''']					)
													UpperCAmelCase__												=     os.path.join(self.tmpdirname		,			VOCAB_FILES_NAMES['''merges_file''']					)

													with open(self.vocab_file		,			'''w'''		,			encoding='''utf-8'''					) as fp:
																			fp.write(json.dumps(lowerCamelCase_					) + '''\n'''					)

													with open(self.merges_file		,			'''w'''					) as fp:
																			fp.write('''\n'''.join(lowerCamelCase_					)					)






							def        lowerCAmelCase  (    self			:  Union[str, Any]					):
													UpperCAmelCase__												=     list(self.get_tokenizer().get_vocab().keys()					)

													self.assertEqual(vocab_keys[0]		,			'''<s>'''					)
													self.assertEqual(vocab_keys[1]		,			'''<pad>'''					)
													self.assertEqual(vocab_keys[-1]		,			'''vocab'''					)
													self.assertEqual(len(lowerCamelCase_					)		,			14					)






							def        lowerCAmelCase  (    self			:  Optional[Any]					):
													self.assertEqual(self.get_tokenizer().vocab_size		,			14					)






							def        lowerCAmelCase  (    self			:  Any					):
													UpperCAmelCase__												=     SpeechaTextaTokenizer.from_pretrained(self.tmpdirname					)

													# make sure @@ is correctly concatenated
													UpperCAmelCase__												=     [4, 6, 8, 7, 10]  # ["here@@", "couple", "words", "of@@", "the"]
													UpperCAmelCase__												=     tokenizer.decode(lowerCamelCase_					)

													self.assertTrue(output_string == '''herecouple words ofthe'''					)






							def        lowerCAmelCase  (    self			:  int					):
													UpperCAmelCase__												=     SpeechaTextaTokenizer.from_pretrained(self.tmpdirname					)

													with tempfile.TemporaryDirectory() as tmp_dirname:
																			tokenizer.save_pretrained(lowerCamelCase_					)
																			os.remove(os.path.join(lowerCamelCase_		,			'''merges.txt'''					)					)

																			# load tokenizer without merges file should not throw an error
																			UpperCAmelCase__												=     SpeechaTextaTokenizer.from_pretrained(lowerCamelCase_					)

													with tempfile.TemporaryDirectory() as tmp_dirname:
																			# save tokenizer and load again
																			tokenizer.save_pretrained(lowerCamelCase_					)
																			UpperCAmelCase__												=     SpeechaTextaTokenizer.from_pretrained(lowerCamelCase_					)

													self.assertIsNotNone(lowerCamelCase_					)






							def        lowerCAmelCase  (    self			:  int					):
													if not self.test_slow_tokenizer:
																			return

													UpperCAmelCase__												=     inspect.signature(self.tokenizer_class.__init__					)
													UpperCAmelCase__												=     self.get_tokenizer()

													for parameter_name, parameter in signature.parameters.items():
																			if parameter.default != inspect.Parameter.empty and parameter_name != "merges_file":
																									self.assertIn(lowerCamelCase_		,			tokenizer.init_kwargs					)