





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

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_lowerCamelCase   				=			'▁'

_lowerCamelCase   				=			{'vocab_file': 'spiece.model'}

_lowerCamelCase   				=			{
    'vocab_file': {'google/pegasus-xsum': 'https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model'}
}

_lowerCamelCase   				=			{
    'google/pegasus-xsum': 512,
}


_lowerCamelCase   				=			logging.get_logger(__name__)




class   UpperCamelCase_       (     UpperCamelCase__    ):
							lowerCamelCase_           =					VOCAB_FILES_NAMES

							lowerCamelCase_           =					VOCAB_FILES_NAMES
							lowerCamelCase_           =					PRETRAINED_VOCAB_FILES_MAP
							lowerCamelCase_           =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							lowerCamelCase_           =					["input_ids", "attention_mask"]


							def __init__(							self    :int							,				__A    :List[Any]							,				__A    :str="<pad>"							,				__A    :List[Any]="</s>"							,				__A    :Optional[int]="<unk>"							,				__A    :int="<mask_2>"							,				__A    :Dict="<mask_1>"							,				__A    :Tuple=None							,				__A    :Optional[int]=103							,				__A    :Optional[Dict[str, Any]] = None							,				**__A    :List[str]							,				)   ->   None:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      offset
									if additional_special_tokens is not None:
											if not isinstance(__A							,				__A		):
													raise TypeError(
													    f'''additional_special_tokens should be of type {type(__A		)}, but is'''
													    f''' {type(__A		)}'''		)

											SCREAMING_SNAKE_CASE__			=      (
											    ([mask_token_sent] + additional_special_tokens)
											    if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
											    else additional_special_tokens
											)
											# fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
											additional_special_tokens_extended += [
											    f'''<unk_{i}>''' for i in range(len(__A		)							,				self.offset - 1		)
											]

											if len(set(__A		)		) != len(__A		):
													raise ValueError(
													    """Please make sure that the provided additional_special_tokens do not contain an incorrectly"""
													    f''' shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}.'''		)
											SCREAMING_SNAKE_CASE__			=      additional_special_tokens_extended
									else:
											SCREAMING_SNAKE_CASE__			=      [mask_token_sent] if mask_token_sent is not None else []
											additional_special_tokens += [f'''<unk_{i}>''' for i in range(2							,				self.offset		)]

									SCREAMING_SNAKE_CASE__			=      {} if sp_model_kwargs is None else sp_model_kwargs

									super().__init__(
									    eos_token=__A							,				unk_token=__A							,				mask_token=__A							,				pad_token=__A							,				mask_token_sent=__A							,				offset=__A							,				additional_special_tokens=__A							,				sp_model_kwargs=self.sp_model_kwargs							,				**__A							,				)
									SCREAMING_SNAKE_CASE__			=      mask_token_sent
									SCREAMING_SNAKE_CASE__			=      vocab_file
									SCREAMING_SNAKE_CASE__			=      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
									self.sp_model.Load(__A		)

									# add special tokens to encoder dict
									SCREAMING_SNAKE_CASE__			=      {
									    0: self.pad_token,
									    1: self.eos_token,
									}

									if self.mask_token_sent is not None:
											self.encoder.update(
											    {
											        2: self.mask_token_sent,
											        3: self.mask_token,
											    }		)

									if self.offset > 0:
											# entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
											# mask_token_sent is already added to list -> so start at 1
											self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1							,				self.offset - 1		)}		)

									SCREAMING_SNAKE_CASE__			=      {v: k for k, v in self.encoder.items()}


							@property
							def      _snake_case  (							self    :Tuple		)   ->   int:



									"""simple docstring"""



									return len(self.sp_model		) + self.offset


							def      _snake_case  (							self    :str		)   ->   Dict[str, int]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      {self.convert_ids_to_tokens(__A		): i for i in range(self.vocab_size		)}
									vocab.update(self.added_tokens_encoder		)
									return vocab


							def __getstate__(							self    :Optional[int]		)   ->   Union[str, Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      self.__dict__.copy()
									SCREAMING_SNAKE_CASE__			=      None
									return state


							def __setstate__(							self    :int							,				__A    :int		)   ->   str:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      d

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

									SCREAMING_SNAKE_CASE__			=      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
									self.sp_model.Load(self.vocab_file		)


							def      _snake_case  (							self    :Union[str, Any]							,				__A    :str		)   ->   List[str]:



									"""simple docstring"""



									return self.sp_model.encode(__A							,				out_type=__A		)


							def      _snake_case  (							self    :str							,				__A    :str		)   ->   int:



									"""simple docstring"""



									if token in self.decoder:
											return self.decoder[token]
									elif token in self.added_tokens_decoder:
											return self.added_tokens_decoder[token]
									SCREAMING_SNAKE_CASE__			=      self.sp_model.piece_to_id(__A		)
									return sp_id + self.offset


							def      _snake_case  (							self    :Tuple							,				__A    :int		)   ->   str:



									"""simple docstring"""



									if index in self.encoder:
											return self.encoder[index]
									elif index in self.added_tokens_encoder:
											return self.added_tokens_encoder[index]
									else:
											SCREAMING_SNAKE_CASE__			=      self.sp_model.IdToPiece(index - self.offset		)
									return token


							def      _snake_case  (							self    :Optional[int]							,				__A    :List[Any]		)   ->   Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      []
									SCREAMING_SNAKE_CASE__			=      """"""
									for token in tokens:
											# make sure that special tokens are not decoded using sentencepiece model
											if token in self.all_special_tokens:
													out_string += self.sp_model.decode(__A		) + token
													SCREAMING_SNAKE_CASE__			=      []
											else:
													current_sub_tokens.append(__A		)
									out_string += self.sp_model.decode(__A		)
									return out_string.strip()


							def      _snake_case  (							self    :List[str]							,				__A    :Union[str, Any]=False		)   ->   Any:



									"""simple docstring"""



									return 1


							def      _snake_case  (							self    :Dict							,				__A    :int		)   ->   List[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      set(self.all_special_ids		)  # call it once instead of inside list comp
									all_special_ids.remove(self.unk_token_id		)  # <unk> is only sometimes special

									return [1 if x in all_special_ids else 0 for x in seq]


							def      _snake_case  (							self    :Optional[int]							,				__A    :List							,				__A    :Optional[List] = None							,				__A    :bool = False		)   ->   List[int]:



									"""simple docstring"""



									if already_has_special_tokens:
											return self._special_token_mask(__A		)
									elif token_ids_a is None:
											return self._special_token_mask(__A		) + [1]
									else:
											return self._special_token_mask(token_ids_a + token_ids_a		) + [1]


							def      _snake_case  (							self    :List[str]							,				__A    :Union[str, Any]							,				__A    :List[Any]=None		)   ->   List[int]:



									"""simple docstring"""



									if token_ids_a is None:
											return token_ids_a + [self.eos_token_id]
									# We don't expect to process pairs, but leave the pair logic for API consistency
									return token_ids_a + token_ids_a + [self.eos_token_id]


							def      _snake_case  (							self    :str							,				__A    :str							,				__A    :Optional[str] = None		)   ->   Tuple[str]:



									"""simple docstring"""



									if not os.path.isdir(__A		):
											logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''		)
											return
									SCREAMING_SNAKE_CASE__			=      os.path.join(
									    __A							,				(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""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:
													SCREAMING_SNAKE_CASE__			=      self.sp_model.serialized_model_proto()
													fi.write(__A		)

									return (out_vocab_file,)






import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import ClassLabel, Features, Image
from .base import TaskTemplate




@dataclass(frozen=UpperCamelCase__    )
class   UpperCamelCase_       (     UpperCamelCase__    ):
							lowerCamelCase_           =					field(default="image-classification"				,  metadata={"include_in_asdict_even_if_is_default": True}    )
							lowerCamelCase_           =					Features({"image": Image()}    )
							lowerCamelCase_           =					Features({"labels": ClassLabel}    )
							lowerCamelCase_           =					"image"
							lowerCamelCase_           =					"labels"


							def      _snake_case  (							self    :List[str]							,				__A    :Tuple		)   ->   Tuple:



									"""simple docstring"""



									if self.label_column not in features:
											raise ValueError(f'''Column {self.label_column} is not present in features.'''		)
									if not isinstance(features[self.label_column]							,				__A		):
											raise ValueError(f'''Column {self.label_column} is not a ClassLabel.'''		)
									SCREAMING_SNAKE_CASE__			=      copy.deepcopy(self		)
									SCREAMING_SNAKE_CASE__			=      self.label_schema.copy()
									SCREAMING_SNAKE_CASE__			=      features[self.label_column]
									SCREAMING_SNAKE_CASE__			=      label_schema
									return task_template


							@property
							def      _snake_case  (							self    :Dict		)   ->   Dict[str, str]:



									"""simple docstring"""



									return {
									    self.image_column: "image",
									    self.label_column: "labels",
									}