






import json
from typing import TYPE_CHECKING, List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_blenderbot import BlenderbotTokenizer


if TYPE_CHECKING:
	from transformers.pipelines.conversational import Conversation

UpperCAmelCase_ :						List[Any]      =      logging.get_logger(__name__)


UpperCAmelCase_ :						List[str]      =      {
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

UpperCAmelCase_ :						Optional[Any]      =      {
    '''vocab_file''': {'''facebook/blenderbot-3B''': '''https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json'''},
    '''merges_file''': {'''facebook/blenderbot-3B''': '''https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt'''},
    '''tokenizer_config_file''': {
        '''facebook/blenderbot-3B''': '''https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json'''
    },
}

UpperCAmelCase_ :						Optional[int]      =      {'''facebook/blenderbot-3B''': 1_28}

class     _SCREAMING_SNAKE_CASE       (			_a      ):
	snake_case__							:     Optional[Any]     	=  VOCAB_FILES_NAMES
	snake_case__							:     List[str]     	=  PRETRAINED_VOCAB_FILES_MAP
	snake_case__							:     int     	=  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	snake_case__							:     Dict     	=  ["""input_ids""", """attention_mask"""]
	snake_case__							:     Union[str, Any]     	=  BlenderbotTokenizer





	def __init__(		self				:					Union[str, Any] ,  __lowerCamelCase				:					Optional[Any]=None ,  __lowerCamelCase				:					str=None ,  __lowerCamelCase				:					Tuple=None ,  __lowerCamelCase				:					int="replace" ,  __lowerCamelCase				:					Optional[Any]="<s>" ,  __lowerCamelCase				:					List[Any]="</s>" ,  __lowerCamelCase				:					Dict="</s>" ,  __lowerCamelCase				:					List[str]="<s>" ,  __lowerCamelCase				:					int="<unk>" ,  __lowerCamelCase				:					Union[str, Any]="<pad>" ,  __lowerCamelCase				:					List[Any]="<mask>" ,  __lowerCamelCase				:					Dict=False ,  __lowerCamelCase				:					Dict=True ,  **__lowerCamelCase				:					Dict ,  ):
		super().__init__(
		    __lowerCamelCase ,  __lowerCamelCase ,  tokenizer_file=__lowerCamelCase ,  errors=__lowerCamelCase ,  bos_token=__lowerCamelCase ,  eos_token=__lowerCamelCase ,  sep_token=__lowerCamelCase ,  cls_token=__lowerCamelCase ,  unk_token=__lowerCamelCase ,  pad_token=__lowerCamelCase ,  mask_token=__lowerCamelCase ,  add_prefix_space=__lowerCamelCase ,  trim_offsets=__lowerCamelCase ,  **__lowerCamelCase ,  )

		UpperCamelCase     :Union[str, Any]          =	json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()	)
		if pre_tok_state.get("""add_prefix_space""" ,  __lowerCamelCase	) != add_prefix_space:
			UpperCamelCase     :Tuple          =	getattr(__lowerCamelCase ,  pre_tok_state.pop("""type"""	)	)
			UpperCamelCase     :List[str]          =	add_prefix_space
			UpperCamelCase     :Optional[int]          =	pre_tok_class(**__lowerCamelCase	)

		UpperCamelCase     :Any          =	add_prefix_space

		UpperCamelCase     :int          =	"""post_processor"""
		UpperCamelCase     :str          =	getattr(self.backend_tokenizer ,  __lowerCamelCase ,  __lowerCamelCase	)
		if tokenizer_component_instance:
			UpperCamelCase     :str          =	json.loads(tokenizer_component_instance.__getstate__()	)

			# The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
			if "sep" in state:
				UpperCamelCase     :Tuple          =	tuple(state["""sep"""]	)
			if "cls" in state:
				UpperCamelCase     :int          =	tuple(state["""cls"""]	)

			UpperCamelCase     :List[Any]          =	False

			if state.get("""add_prefix_space""" ,  __lowerCamelCase	) != add_prefix_space:
				UpperCamelCase     :int          =	add_prefix_space
				UpperCamelCase     :Dict          =	True

			if state.get("""trim_offsets""" ,  __lowerCamelCase	) != trim_offsets:
				UpperCamelCase     :Optional[Any]          =	trim_offsets
				UpperCamelCase     :Union[str, Any]          =	True

			if changes_to_apply:
				UpperCamelCase     :str          =	getattr(__lowerCamelCase ,  state.pop("""type"""	)	)
				UpperCamelCase     :Any          =	component_class(**__lowerCamelCase	)
				setattr(self.backend_tokenizer ,  __lowerCamelCase ,  __lowerCamelCase	)





	@property
	# Copied from transformers.models.roberta.tokenization_roberta_fast.RobertaTokenizerFast.mask_token with Roberta->Blenderbot, RoBERTa->Blenderbot
	def 						_A (		self				:					Union[str, Any]	):
		if self._mask_token is None:
			if self.verbose:
				logger.error("""Using mask_token, but it is not set yet."""	)
			return None
		return str(self._mask_token	)





	@mask_token.setter
	def 						_A (		self				:					str ,  __lowerCamelCase				:					int	):
		UpperCamelCase     :int          =	AddedToken(__lowerCamelCase ,  lstrip=__lowerCamelCase ,  rstrip=__lowerCamelCase	) if isinstance(__lowerCamelCase ,  __lowerCamelCase	) else value
		UpperCamelCase     :str          =	value





	def 						_A (		self				:					Union[str, Any] ,  *__lowerCamelCase				:					List[Any] ,  **__lowerCamelCase				:					List[Any]	):
		UpperCamelCase     :int          =	kwargs.get("""is_split_into_words""" ,  __lowerCamelCase	)
		assert self.add_prefix_space or not is_split_into_words, (
		    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
		    "to use it with pretokenized inputs."
		)

		return super()._batch_encode_plus(*__lowerCamelCase ,  **__lowerCamelCase	)





	def 						_A (		self				:					Tuple ,  *__lowerCamelCase				:					Any ,  **__lowerCamelCase				:					str	):
		UpperCamelCase     :Tuple          =	kwargs.get("""is_split_into_words""" ,  __lowerCamelCase	)

		assert self.add_prefix_space or not is_split_into_words, (
		    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
		    "to use it with pretokenized inputs."
		)

		return super()._encode_plus(*__lowerCamelCase ,  **__lowerCamelCase	)





	def 						_A (		self				:					str ,  __lowerCamelCase				:					str ,  __lowerCamelCase				:					Optional[str] = None	):
		UpperCamelCase     :List[Any]          =	self._tokenizer.model.save(__lowerCamelCase ,  name=__lowerCamelCase	)
		return tuple(__lowerCamelCase	)





	def 						_A (		self				:					Optional[int] ,  __lowerCamelCase				:					List[int] ,  __lowerCamelCase				:					Optional[List[int]] = None	):
		UpperCamelCase     :Optional[Any]          =	[self.sep_token_id]
		UpperCamelCase     :Tuple          =	[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]





	def 						_A (		self				:					List[Any] ,  __lowerCamelCase				:					List[int] ,  __lowerCamelCase				:					Optional[List[int]] = None	):
		return token_ids_a + [self.eos_token_id]





	def 						_A (		self				:					Union[str, Any] ,  __lowerCamelCase				:					"Conversation"	):
		UpperCamelCase     :Optional[int]          =	[]
		for is_user, text in conversation.iter_texts():
			if is_user:
				# We need to space prefix as it's being done within blenderbot
				inputs.append(""" """ + text	)
			else:
				# Generated responses should contain them already.
				inputs.append(__lowerCamelCase	)

		UpperCamelCase     :Union[str, Any]          =	"""  """.join(__lowerCamelCase	)
		UpperCamelCase     :Tuple          =	self.encode(__lowerCamelCase	)
		if len(__lowerCamelCase	) > self.model_max_length:
			UpperCamelCase     :List[str]          =	input_ids[-self.model_max_length :]
			logger.warning(F"""Trimmed input from conversation as it was longer than {self.model_max_length} tokens."""	)
		return input_ids








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

from ..features import Audio, Features, Value
from .base import TaskTemplate

@dataclass(frozen=_a      )
class     _SCREAMING_SNAKE_CASE       (			_a      ):
	snake_case__							:     str     	=  field(default="""automatic-speech-recognition"""		,   metadata={"""include_in_asdict_even_if_is_default""": True}      )
	snake_case__							:     ClassVar[Features]     	=  Features({"""audio""": Audio()}      )
	snake_case__							:     ClassVar[Features]     	=  Features({"""transcription""": Value("""string"""      )}      )
	snake_case__							:     str     	=  "audio"
	snake_case__							:     str     	=  "transcription"





	def 						_A (		self				:					List[str] ,  __lowerCamelCase				:					Dict	):
		if self.audio_column not in features:
			raise ValueError(F"""Column {self.audio_column} is not present in features."""	)
		if not isinstance(features[self.audio_column] ,  __lowerCamelCase	):
			raise ValueError(F"""Column {self.audio_column} is not an Audio type."""	)
		UpperCamelCase     :int          =	copy.deepcopy(self	)
		UpperCamelCase     :Any          =	self.input_schema.copy()
		UpperCamelCase     :List[str]          =	features[self.audio_column]
		UpperCamelCase     :List[Any]          =	input_schema
		return task_template





	@property
	def 						_A (		self				:					Optional[int]	):
		return {self.audio_column: "audio", self.transcription_column: "transcription"}
