






import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map




def   __A       (  _lowercase			):

					'''simple docstring'''







					_A										=			[]
					if isinstance(_lowercase       ,      _lowercase			):
										for v in tree.values():
															shapes.extend(_fetch_dims(_lowercase			)			)
					elif isinstance(_lowercase       ,      (list, tuple)			):
										for t in tree:
															shapes.extend(_fetch_dims(_lowercase			)			)
					elif isinstance(_lowercase       ,      torch.Tensor			):
										shapes.append(tree.shape			)
					else:
										raise ValueError('''Not supported'''			)

					return shapes




@torch.jit.ignore
def   __A       (  _lowercase       ,      _lowercase			):

					'''simple docstring'''







					_A										=			[]
					for d in reversed(_lowercase			):
										idx.append(flat_idx % d			)
										_A										=			flat_idx // d

					return tuple(reversed(_lowercase			)			)




@torch.jit.ignore
def   __A       (  _lowercase       ,      _lowercase       ,      _lowercase       ,      _lowercase = None       ,      _lowercase = None       ,      ):

					'''simple docstring'''







					def reduce_edge_list(_lowercase			) -> None:
										_A										=			True
										for i in range(len(_lowercase			)			):
															_A										=			-1 * (i + 1)
															l[reversed_idx] &= tally
															_A										=			l[reversed_idx]

					if start_edges is None:
										_A										=			[s == 0 for s in start]
										reduce_edge_list(_lowercase			)
					if end_edges is None:
										_A										=			[e == (d - 1) for e, d in zip(_lowercase       ,      _lowercase			)]
										reduce_edge_list(_lowercase			)

					# Base cases. Either start/end are empty and we're done, or the final,
					# one-dimensional tensor can be simply sliced
					if len(_lowercase			) == 0:
										return [()]
					elif len(_lowercase			) == 1:
										return [(slice(start[0]       ,      end[0] + 1			),)]

					_A										=			[]
					_A										=			[]

					# Dimensions common to start and end can be selected directly
					for s, e in zip(_lowercase       ,      _lowercase			):
										if s == e:
															path_list.append(slice(_lowercase       ,      s + 1			)			)
										else:
															break

					_A										=			tuple(_lowercase			)
					_A										=			len(_lowercase			)

					# start == end, and we're done
					if divergence_idx == len(_lowercase			):
										return [path]

					def upper() -> Tuple[Tuple[slice, ...], ...]:
										assert start_edges is not None
										assert end_edges is not None

										_A										=			start[divergence_idx]
										return tuple(
										    path + (slice(_lowercase       ,      sdi + 1			),) + s
										    for s in _get_minimal_slice_set(
										        start[divergence_idx + 1 :]       ,      [d - 1 for d in dims[divergence_idx + 1 :]]       ,      dims[divergence_idx + 1 :]       ,      start_edges=start_edges[divergence_idx + 1 :]       ,      end_edges=[True for _ in end_edges[divergence_idx + 1 :]]       ,      )			)

					def lower() -> Tuple[Tuple[slice, ...], ...]:
										assert start_edges is not None
										assert end_edges is not None

										_A										=			end[divergence_idx]
										return tuple(
										    path + (slice(_lowercase       ,      edi + 1			),) + s
										    for s in _get_minimal_slice_set(
										        [0 for _ in start[divergence_idx + 1 :]]       ,      end[divergence_idx + 1 :]       ,      dims[divergence_idx + 1 :]       ,      start_edges=[True for _ in start_edges[divergence_idx + 1 :]]       ,      end_edges=end_edges[divergence_idx + 1 :]       ,      )			)

					# If both start and end are at the edges of the subtree rooted at
					# divergence_idx, we can just select the whole subtree at once
					if start_edges[divergence_idx] and end_edges[divergence_idx]:
										slices.append(path + (slice(start[divergence_idx]       ,      end[divergence_idx] + 1			),)			)
					# If just start is at the edge, we can grab almost all of the subtree,
					# treating only the ragged bottom edge as an edge case
					elif start_edges[divergence_idx]:
										slices.append(path + (slice(start[divergence_idx]       ,      end[divergence_idx]			),)			)
										slices.extend(lower()			)
					# Analogous to the previous case, but the top is ragged this time
					elif end_edges[divergence_idx]:
										slices.extend(upper()			)
										slices.append(path + (slice(start[divergence_idx] + 1       ,      end[divergence_idx] + 1			),)			)
					# If both sides of the range are ragged, we need to handle both sides
					# separately. If there's contiguous meat in between them, we can index it
					# in one big chunk
					else:
										slices.extend(upper()			)
										_A										=			end[divergence_idx] - start[divergence_idx]
										if middle_ground > 1:
															slices.append(path + (slice(start[divergence_idx] + 1       ,      end[divergence_idx]			),)			)
										slices.extend(lower()			)

					return slices




@torch.jit.ignore
def   __A       (  _lowercase       ,      _lowercase       ,      _lowercase       ,      _lowercase			):

					'''simple docstring'''







					_A										=			t.shape[:no_batch_dims]
					_A										=			list(_flat_idx_to_idx(_lowercase       ,      _lowercase			)			)
					# _get_minimal_slice_set is inclusive
					_A										=			list(_flat_idx_to_idx(flat_end - 1       ,      _lowercase			)			)

					# Get an ordered list of slices to perform
					_A										=			_get_minimal_slice_set(
					    _lowercase       ,      _lowercase       ,      _lowercase       ,      )

					_A										=			[t[s] for s in slices]

					return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]			) for s in sliced_tensors]			)




def   __A       (  _lowercase       ,      _lowercase       ,      _lowercase       ,      _lowercase       ,      _lowercase = False       ,      _lowercase = None       ,      _lowercase = False       ,      ):

					'''simple docstring'''







					if not (len(_lowercase			) > 0):
										raise ValueError('''Must provide at least one input'''			)

					_A										=			[shape[:no_batch_dims] for shape in _fetch_dims(_lowercase			)]
					_A										=			tuple([max(_lowercase			) for s in zip(*_lowercase			)]			)

					def _prep_inputs(_lowercase			) -> torch.Tensor:
										if not low_mem:
															if not sum(t.shape[:no_batch_dims]			) == no_batch_dims:
																				_A										=			t.expand(orig_batch_dims + t.shape[no_batch_dims:]			)
															_A										=			t.reshape(-1       ,      *t.shape[no_batch_dims:]			)
										else:
															_A										=			t.expand(orig_batch_dims + t.shape[no_batch_dims:]			)
										return t

					_A										=			tensor_tree_map(_prep_inputs       ,      _lowercase			)
					_A										=			None
					if _out is not None:
										_A										=			tensor_tree_map(lambda _lowercase			: t.view([-1] + list(t.shape[no_batch_dims:]			)			)       ,      _out			)

					_A										=			1
					for d in orig_batch_dims:
										flat_batch_dim *= d

					_A										=			flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

					def _select_chunk(_lowercase			) -> torch.Tensor:
										return t[i : i + chunk_size] if t.shape[0] != 1 else t

					_A										=			0
					_A										=			prepped_outputs
					for _ in range(_lowercase			):
										# Chunk the input
										if not low_mem:
															_A										=			_select_chunk
										else:
															_A										=			partial(
															    _chunk_slice       ,      flat_start=_lowercase       ,      flat_end=min(_lowercase       ,      i + chunk_size			)       ,      no_batch_dims=len(_lowercase			)       ,      )

										_A										=			tensor_tree_map(_lowercase       ,      _lowercase			)

										# Run the layer on the chunk
										_A										=			layer(**_lowercase			)

										# Allocate space for the output
										if out is None:
															_A										=			tensor_tree_map(lambda _lowercase			: t.new_zeros((flat_batch_dim,) + t.shape[1:]			)       ,      _lowercase			)

										# Put the chunk in its pre-allocated space
										if isinstance(_lowercase       ,      _lowercase			):

															def assign(_lowercase       ,      _lowercase			) -> None:
																				for k, v in da.items():
																									if isinstance(_lowercase       ,      _lowercase			):
																														assign(_lowercase       ,      da[k]			)
																									else:
																														if _add_into_out:
																																			v[i : i + chunk_size] += da[k]
																														else:
																																			_A										=			da[k]

															assign(_lowercase       ,      _lowercase			)
										elif isinstance(_lowercase       ,      _lowercase			):
															for xa, xa in zip(_lowercase       ,      _lowercase			):
																				if _add_into_out:
																									xa[i : i + chunk_size] += xa
																				else:
																									_A										=			xa
										elif isinstance(_lowercase       ,      torch.Tensor			):
															if _add_into_out:
																				out[i : i + chunk_size] += output_chunk
															else:
																				_A										=			output_chunk
										else:
															raise ValueError('''Not supported'''			)

										i += chunk_size

					_A										=			tensor_tree_map(lambda _lowercase			: t.view(orig_batch_dims + t.shape[1:]			)       ,      _lowercase			)

					return out





class     SCREAMING_SNAKE_CASE					:





							"""simple docstring"""

							def __init__(	self: Optional[Any]		,  __A: List[Any] = 5_12		,  )     ->					Any:
												_A										=			max_chunk_size
												_A										=			None
												_A										=			None

							def    __A			(	self: Union[str, Any]		,  __A: Optional[int]		,  __A: Optional[int]		,  __A: Optional[Any] )     ->					int:
												logging.info('''Tuning chunk size...''' )

												if min_chunk_size >= self.max_chunk_size:
																	return min_chunk_size

												_A										=			[2**l for l in range(int(math.log(self.max_chunk_size		,  2 ) ) + 1 )]
												_A										=			[c for c in candidates if c > min_chunk_size]
												_A										=			[min_chunk_size] + candidates
												candidates[-1] += 4

												def test_chunk_size(__A: Any ) -> bool:
																	try:
																						with torch.no_grad():
																											fn(*a_		,  chunk_size=a_ )
																						return True
																	except RuntimeError:
																						return False

												_A										=			0
												_A										=			len(a_ ) - 1
												while i > min_viable_chunk_size_index:
																	_A										=			test_chunk_size(candidates[i] )
																	if not viable:
																						_A										=			(min_viable_chunk_size_index + i) // 2
																	else:
																						_A										=			i
																						_A										=			(i + len(a_ ) - 1) // 2

												return candidates[min_viable_chunk_size_index]

							def    __A			(	self: List[Any]		,  __A: List[Any]		,  __A: Optional[Any] )     ->					bool:
												_A										=			True
												for aa, aa in zip(a_		,  a_ ):
																	assert type(a_ ) == type(a_ )
																	if isinstance(a_		,  (list, tuple) ):
																						consistent &= self._compare_arg_caches(a_		,  a_ )
																	elif isinstance(a_		,  a_ ):
																						_A										=			[v for _, v in sorted(aa.items()		,  key=lambda __A : x[0] )]
																						_A										=			[v for _, v in sorted(aa.items()		,  key=lambda __A : x[0] )]
																						consistent &= self._compare_arg_caches(a_		,  a_ )
																	else:
																						consistent &= aa == aa

												return consistent



							def    __A			(	self: Union[str, Any]		,  __A: List[Any]		,  __A: List[str]		,  __A: str		,  )     ->					int:
												_A										=			True
												_A										=			tree_map(lambda __A : a.shape if isinstance(a_		,  torch.Tensor ) else a		,  a_		,  a_ )
												if self.cached_arg_data is not None:
																	# If args have changed shape/value, we need to re-tune
																	assert len(self.cached_arg_data ) == len(a_ )
																	_A										=			self._compare_arg_caches(self.cached_arg_data		,  a_ )
												else:
																	# Otherwise, we can reuse the precomputed value
																	_A										=			False

												if not consistent:
																	_A										=			self._determine_favorable_chunk_size(
																	    a_		,  a_		,  a_		,  )
																	_A										=			arg_data

												assert self.cached_chunk_size is not None

												return self.cached_chunk_size






'''simple docstring'''




import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


_a					:			Tuple							=    logging.get_logger(__name__)

_a					:			Optional[int]							=    '▁'

_a					:			str							=    {'vocab_file': 'sentencepiece.bpe.model', 'monolingual_vocab_file': 'dict.txt'}

_a					:			Optional[Any]							=    {
    'vocab_file': {
        'vinai/bartpho-syllable': 'https://huggingface.co/vinai/bartpho-syllable/resolve/main/sentencepiece.bpe.model',
    },
    'monolingual_vocab_file': {
        'vinai/bartpho-syllable': 'https://huggingface.co/vinai/bartpho-syllable/resolve/main/dict.txt',
    },
}

_a					:			Union[str, Any]							=    {'vinai/bartpho-syllable': 1_024}



class 							lowercase_						(		a			):

  '''simple docstring'''





  __lowerCAmelCase       :			Dict     	= VOCAB_FILES_NAMES
  __lowerCAmelCase       :			str     	= PRETRAINED_VOCAB_FILES_MAP
  __lowerCAmelCase       :			Any     	= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  __lowerCAmelCase       :			int     	= ["input_ids", "attention_mask"]
  def __init__(  self ,			a_ ,			a_ ,			a_="<s>" ,			a_="</s>" ,			a_="</s>" ,			a_="<s>" ,			a_="<unk>" ,			a_="<pad>" ,			a_="<mask>" ,			a_ = None ,			**a_ ,			)				->					None:


    """simple docstring"""
    # Mask token behave like a normal word, i.e. include the space before it
    UpperCAmelCase    				=						AddedToken(a_ ,			lstrip=a_ ,			rstrip=a_  ) if isinstance(a_ ,			a_  ) else mask_token

    UpperCAmelCase    				=						{} if sp_model_kwargs is None else sp_model_kwargs

    super().__init__(
        bos_token=a_ ,			eos_token=a_ ,			unk_token=a_ ,			sep_token=a_ ,			cls_token=a_ ,			pad_token=a_ ,			mask_token=a_ ,			sp_model_kwargs=self.sp_model_kwargs ,			**a_ ,			)

    UpperCAmelCase    				=						vocab_file
    UpperCAmelCase    				=						monolingual_vocab_file
    UpperCAmelCase    				=						spm.SentencePieceProcessor(**self.sp_model_kwargs  )
    self.sp_model.Load(str(a_  )  )

    # Load the reduced vocab

    # Keep order of special tokens for backward compatibility
    UpperCAmelCase    				=						{}
    UpperCAmelCase    				=						0
    for token in [bos_token, pad_token, eos_token, unk_token, sep_token, cls_token]:
      if str(a_  ) not in self.fairseq_tokens_to_ids:
        UpperCAmelCase    				=						cnt
        cnt += 1
    with open(a_ ,			'r' ,			encoding='utf-8'  ) as f:
      for line in f.readlines():
        UpperCAmelCase    				=						line.strip().split()[0]
        UpperCAmelCase    				=						len(self.fairseq_tokens_to_ids  )
    if str(a_  ) not in self.fairseq_tokens_to_ids:
      UpperCAmelCase    				=						len(self.fairseq_tokens_to_ids  )

    UpperCAmelCase    				=						{v: k for k, v in self.fairseq_tokens_to_ids.items()}
  def __getstate__(  self  )				->					Any:


    """simple docstring"""
    UpperCAmelCase    				=						self.__dict__.copy()
    UpperCAmelCase    				=						None
    UpperCAmelCase    				=						self.sp_model.serialized_model_proto()
    return state
  def __setstate__(  self ,			a_  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						d

    # for backward compatibility
    if not hasattr(self ,			'sp_model_kwargs'  ):
      UpperCAmelCase    				=						{}

    UpperCAmelCase    				=						spm.SentencePieceProcessor(**self.sp_model_kwargs  )
    self.sp_model.LoadFromSerializedProto(self.sp_model_proto  )
  def     snake_case_      (  self ,			a_ ,			a_ = None  )				->					List[int]:


    """simple docstring"""

    if token_ids_a is None:
      return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
    UpperCAmelCase    				=						[self.cls_token_id]
    UpperCAmelCase    				=						[self.sep_token_id]
    return cls + token_ids_a + sep + sep + token_ids_a + sep
  def     snake_case_      (  self ,			a_ ,			a_ = None ,			a_ = False  )				->					List[int]:


    """simple docstring"""

    if already_has_special_tokens:
      return super().get_special_tokens_mask(
          token_ids_a=a_ ,			token_ids_a=a_ ,			already_has_special_tokens=a_  )

    if token_ids_a is None:
      return [1] + ([0] * len(a_  )) + [1]
    return [1] + ([0] * len(a_  )) + [1, 1] + ([0] * len(a_  )) + [1]
  def     snake_case_      (  self ,			a_ ,			a_ = None  )				->					List[int]:


    """simple docstring"""
    UpperCAmelCase    				=						[self.sep_token_id]
    UpperCAmelCase    				=						[self.cls_token_id]

    if token_ids_a is None:
      return len(cls + token_ids_a + sep  ) * [0]
    return len(cls + token_ids_a + sep + sep + token_ids_a + sep  ) * [0]
  @property
  def     snake_case_      (  self  )				->					Union[str, Any]:


    """simple docstring"""
    return len(self.fairseq_ids_to_tokens  )
  def     snake_case_      (  self  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						{self.convert_ids_to_tokens(a_  ): i for i in range(self.vocab_size  )}
    vocab.update(self.added_tokens_encoder  )
    return vocab
  def     snake_case_      (  self ,			a_  )				->					List[str]:


    """simple docstring"""
    return self.sp_model.encode(a_ ,			out_type=a_  )
  def     snake_case_      (  self ,			a_  )				->					Any:


    """simple docstring"""
    if token in self.fairseq_tokens_to_ids:
      return self.fairseq_tokens_to_ids[token]
    else:
      return self.unk_token_id
  def     snake_case_      (  self ,			a_  )				->					Dict:


    """simple docstring"""
    return self.fairseq_ids_to_tokens[index]
  def     snake_case_      (  self ,			a_  )				->					Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase    				=						''.join(a_  ).replace(a_ ,			' '  ).strip()
    return out_string





  def     snake_case_      (  self ,			a_ ,			a_ = None  )				->					Tuple[str]:


    """simple docstring"""
    if not os.path.isdir(a_  ):
      logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''  )
      return
    UpperCAmelCase    				=						os.path.join(
        a_ ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
    UpperCAmelCase    				=						os.path.join(
        a_ ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['monolingual_vocab_file'] ,			)

    if os.path.abspath(self.vocab_file  ) != os.path.abspath(a_  ) and os.path.isfile(self.vocab_file  ):
      copyfile(self.vocab_file ,			a_  )
    elif not os.path.isfile(self.vocab_file  ):
      with open(a_ ,			'wb'  ) as fi:
        UpperCAmelCase    				=						self.sp_model.serialized_model_proto()
        fi.write(a_  )

    if os.path.abspath(self.monolingual_vocab_file  ) != os.path.abspath(
        a_  ) and os.path.isfile(self.monolingual_vocab_file  ):
      copyfile(self.monolingual_vocab_file ,			a_  )
    elif not os.path.isfile(self.monolingual_vocab_file  ):
      with open(a_ ,			'w' ,			encoding='utf-8'  ) as fp:
        for token in self.fairseq_tokens_to_ids:
          if token not in self.all_special_tokens:
            fp.write(F'''{str(a_  )} \n'''  )

    return out_vocab_file, out_monolingual_vocab_file


