





import numpy as np
import torch
from torch.utils.data import Dataset

from utils import logger







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   def __init__( self					,			snake_case					,			snake_case      )				->  List[Any]:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  params

      a__	:						List[str]      		=  np.array(snake_case      )
      a__	:						Dict      		=  np.array([len(snake_case      ) for t in data]      )

      self.check()
      self.remove_long_sequences()
      self.remove_empty_sequences()
      self.remove_unknown_sequences()
      self.check()
      self.print_statistics()





   def __getitem__( self					,			snake_case      )				->  int:
      """simple docstring"""





      return (self.token_ids[index], self.lengths[index])





   def __len__( self      )				->  Optional[int]:
      """simple docstring"""





      return len(self.lengths      )





   def 	_snake_case ( self      )				->  Dict:
      """simple docstring"""





      assert len(self.token_ids      ) == len(self.lengths      )
      assert all(self.lengths[i] == len(self.token_ids[i]      ) for i in range(len(self.lengths      )      )      )





   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Optional[Any]      		=  self.params.max_model_input_size
      a__	:						Tuple      		=  self.lengths > max_len
      logger.info(F"""Splitting {sum(snake_case      )} too long sequences."""      )

      def divide_chunks(snake_case					,			snake_case      ):
         return [l[i : i + n] for i in range(0					,			len(snake_case      )					,			snake_case      )]

      a__	:						Any      		=  []
      a__	:						List[Any]      		=  []
      if self.params.mlm:
         a__	:						Any      		=  self.params.special_tok_ids["cls_token"], self.params.special_tok_ids["sep_token"]
      else:
         a__	:						Any      		=  self.params.special_tok_ids["bos_token"], self.params.special_tok_ids["eos_token"]

      for seq_, len_ in zip(self.token_ids					,			self.lengths      ):
         assert (seq_[0] == cls_id) and (seq_[-1] == sep_id), seq_
         if len_ <= max_len:
            new_tok_ids.append(seq_      )
            new_lengths.append(len_      )
         else:
            a__	:						List[str]      		=  []
            for sub_s in divide_chunks(seq_					,			max_len - 2      ):
               if sub_s[0] != cls_id:
                  a__	:						Union[str, Any]      		=  np.insert(snake_case					,			0					,			snake_case      )
               if sub_s[-1] != sep_id:
                  a__	:						Tuple      		=  np.insert(snake_case					,			len(snake_case      )					,			snake_case      )
               assert len(snake_case      ) <= max_len
               assert (sub_s[0] == cls_id) and (sub_s[-1] == sep_id), sub_s
               sub_seqs.append(snake_case      )

            new_tok_ids.extend(snake_case      )
            new_lengths.extend([len(snake_case      ) for l in sub_seqs]      )

      a__	:						int      		=  np.array(snake_case      )
      a__	:						List[Any]      		=  np.array(snake_case      )





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      a__	:						List[str]      		=  len(self      )
      a__	:						Optional[int]      		=  self.lengths > 11
      a__	:						List[Any]      		=  self.token_ids[indices]
      a__	:						str      		=  self.lengths[indices]
      a__	:						Optional[Any]      		=  len(self      )
      logger.info(F"""Remove {init_size - new_size} too short (<=11 tokens) sequences."""      )





   def 	_snake_case ( self      )				->  Dict:
      """simple docstring"""





      if "unk_token" not in self.params.special_tok_ids:
         return
      else:
         a__	:						int      		=  self.params.special_tok_ids["unk_token"]
      a__	:						List[str]      		=  len(self      )
      a__	:						Dict      		=  np.array([np.count_nonzero(a == unk_token_id      ) for a in self.token_ids]      )
      a__	:						str      		=  (unk_occs / self.lengths) < 0.5
      a__	:						Any      		=  self.token_ids[indices]
      a__	:						Dict      		=  self.lengths[indices]
      a__	:						Union[str, Any]      		=  len(self      )
      logger.info(F"""Remove {init_size - new_size} sequences with a high level of unknown tokens (50%)."""      )





   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      if not self.params.is_master:
         return
      logger.info(F"""{len(self      )} sequences"""      )
      # data_len = sum(self.lengths)
      # nb_unique_tokens = len(Counter(list(chain(*self.token_ids))))
      # logger.info(f'{data_len} tokens ({nb_unique_tokens} unique)')

      # unk_idx = self.params.special_tok_ids['unk_token']
      # nb_unknown = sum([(t==unk_idx).sum() for t in self.token_ids])
      # logger.info(f'{nb_unknown} unknown tokens (covering {100*nb_unknown/data_len:.2f}% of the data)')





   def 	_snake_case ( self					,			snake_case      )				->  Any:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  [t[0] for t in batch]
      a__	:						List[str]      		=  [t[1] for t in batch]
      assert len(snake_case      ) == len(snake_case      )

      # Max for paddings
      a__	:						Any      		=  max(snake_case      )

      # Pad token ids
      if self.params.mlm:
         a__	:						Optional[int]      		=  self.params.special_tok_ids["pad_token"]
      else:
         a__	:						List[str]      		=  self.params.special_tok_ids["unk_token"]
      a__	:						Any      		=  [list(t.astype(snake_case      )      ) + [pad_idx] * (max_seq_len_ - len(snake_case      )) for t in token_ids]
      assert len(tk_      ) == len(snake_case      )
      assert all(len(snake_case      ) == max_seq_len_ for t in tk_      )

      a__	:						Optional[int]      		=  torch.tensor(tk_      )  # (bs, max_seq_len_)
      a__	:						Optional[int]      		=  torch.tensor(snake_case      )  # (bs)
      return tk_t, lg_t










import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_lxmert import LxmertTokenizer


SCREAMING_SNAKE_CASE__       :			Optional[Any]			      =						{"""vocab_file""": """vocab.txt""", """tokenizer_file""": """tokenizer.json"""}

SCREAMING_SNAKE_CASE__       :			List[Any]			      =						{
    """vocab_file""": {
        """unc-nlp/lxmert-base-uncased""": """https://huggingface.co/unc-nlp/lxmert-base-uncased/resolve/main/vocab.txt""",
    },
    """tokenizer_file""": {
        """unc-nlp/lxmert-base-uncased""": (
            """https://huggingface.co/unc-nlp/lxmert-base-uncased/resolve/main/tokenizer.json"""
        ),
    },
}

SCREAMING_SNAKE_CASE__       :			Any			      =						{
    """unc-nlp/lxmert-base-uncased""": 5_1_2,
}

SCREAMING_SNAKE_CASE__       :			Optional[int]			      =						{
    """unc-nlp/lxmert-base-uncased""": {"""do_lower_case""": True},
}







class   __lowerCAmelCase			(					_UpperCamelCase  ):
   _UpperCamelCase :						List[str]     		=				VOCAB_FILES_NAMES
   _UpperCamelCase :						str     		=				PRETRAINED_VOCAB_FILES_MAP
   _UpperCamelCase :						List[str]     		=				PRETRAINED_INIT_CONFIGURATION
   _UpperCamelCase :						Optional[int]     		=				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   _UpperCamelCase :						Optional[Any]     		=				LxmertTokenizer





   def __init__( self					,			snake_case=None					,			snake_case=None					,			snake_case=True					,			snake_case="[UNK]"					,			snake_case="[SEP]"					,			snake_case="[PAD]"					,			snake_case="[CLS]"					,			snake_case="[MASK]"					,			snake_case=True					,			snake_case=None					,			**snake_case					,			)				->  Any:
      """simple docstring"""





      super().__init__(
          snake_case					,			tokenizer_file=snake_case					,			do_lower_case=snake_case					,			unk_token=snake_case					,			sep_token=snake_case					,			pad_token=snake_case					,			cls_token=snake_case					,			mask_token=snake_case					,			tokenize_chinese_chars=snake_case					,			strip_accents=snake_case					,			**snake_case					,			)

      a__	:						Optional[Any]      		=  json.loads(self.backend_tokenizer.normalizer.__getstate__()      )
      if (
          normalizer_state.get("lowercase"					,			snake_case      ) != do_lower_case
          or normalizer_state.get("strip_accents"					,			snake_case      ) != strip_accents
          or normalizer_state.get("handle_chinese_chars"					,			snake_case      ) != tokenize_chinese_chars
      ):
         a__	:						str      		=  getattr(snake_case					,			normalizer_state.pop("type"      )      )
         a__	:						Tuple      		=  do_lower_case
         a__	:						Union[str, Any]      		=  strip_accents
         a__	:						str      		=  tokenize_chinese_chars
         a__	:						List[str]      		=  normalizer_class(**snake_case      )

      a__	:						str      		=  do_lower_case





   def 	_snake_case ( self					,			snake_case					,			snake_case=None      )				->  List[str]:
      """simple docstring"""





      a__	:						Optional[int]      		=  [self.cls_token_id] + token_ids_a + [self.sep_token_id]

      if token_ids_a:
         output += token_ids_a + [self.sep_token_id]

      return output





   def 	_snake_case ( self					,			snake_case					,			snake_case = None      )				->  List[int]:
      """simple docstring"""





      a__	:						Tuple      		=  [self.sep_token_id]
      a__	:						str      		=  [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      ) * [0] + len(token_ids_a + sep      ) * [1]





   def 	_snake_case ( self					,			snake_case					,			snake_case = None      )				->  Tuple[str]:
      """simple docstring"""





      a__	:						Optional[int]      		=  self._tokenizer.model.save(snake_case					,			name=snake_case      )
      return tuple(snake_case      )

