





import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

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


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

UpperCAmelCase	:	List[str]		     =       """▁"""

UpperCAmelCase	:	Dict		     =       {
    """vocab_file""": """vocab.json""",
    """spm_file""": """sentencepiece.bpe.model""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

UpperCAmelCase	:	List[Any]		     =       {
    """vocab_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json""",
    },
    """spm_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model""",
    },
    """tokenizer_config_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json""",
    },
}

UpperCAmelCase	:	Any		     =       {
    """facebook/m2m100_418M""": 1024,
}

# fmt: off
UpperCAmelCase	:	Optional[int]		     =       {
    """m2m100""": ["""af""", """am""", """ar""", """ast""", """az""", """ba""", """be""", """bg""", """bn""", """br""", """bs""", """ca""", """ceb""", """cs""", """cy""", """da""", """de""", """el""", """en""", """es""", """et""", """fa""", """ff""", """fi""", """fr""", """fy""", """ga""", """gd""", """gl""", """gu""", """ha""", """he""", """hi""", """hr""", """ht""", """hu""", """hy""", """id""", """ig""", """ilo""", """is""", """it""", """ja""", """jv""", """ka""", """kk""", """km""", """kn""", """ko""", """lb""", """lg""", """ln""", """lo""", """lt""", """lv""", """mg""", """mk""", """ml""", """mn""", """mr""", """ms""", """my""", """ne""", """nl""", """no""", """ns""", """oc""", """or""", """pa""", """pl""", """ps""", """pt""", """ro""", """ru""", """sd""", """si""", """sk""", """sl""", """so""", """sq""", """sr""", """ss""", """su""", """sv""", """sw""", """ta""", """th""", """tl""", """tn""", """tr""", """uk""", """ur""", """uz""", """vi""", """wo""", """xh""", """yi""", """yo""", """zh""", """zu"""],
    """wmt21""": ["""en""", """ha""", """is""", """ja""", """cs""", """ru""", """zh""", """de"""]
}







class 				__lowerCAmelCase       (    UpperCamelCase__):
      _lowercase				:					Union[str, Any]						=			VOCAB_FILES_NAMES
      _lowercase				:					Optional[Any]						=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _lowercase				:					int						=			PRETRAINED_VOCAB_FILES_MAP
      _lowercase				:					Dict						=			["""input_ids""", """attention_mask"""]

      _lowercase				:					List[int]						=			[]
      _lowercase				:					List[int]						=			[]







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__=None		,      lowerCAmelCase__=None		,      lowerCAmelCase__="<s>"		,      lowerCAmelCase__="</s>"		,      lowerCAmelCase__="</s>"		,      lowerCAmelCase__="<pad>"		,      lowerCAmelCase__="<unk>"		,      lowerCAmelCase__="m2m100"		,      lowerCAmelCase__ = None		,      lowerCAmelCase__=8		,      **lowerCAmelCase__		,      )   ->						None:



        '''simple docstring'''




        a__	:					Union[str, Any]										={} if sp_model_kwargs is None else sp_model_kwargs

        a__	:					List[str]										=language_codes
        a__	:					str										=FAIRSEQ_LANGUAGE_CODES[language_codes]
        a__	:					Union[str, Any]										={lang_code: F'''__{lang_code}__''' for lang_code in fairseq_language_code}

        a__	:					Union[str, Any]										=kwargs.get("additional_special_tokens"		,      []							)
        kwargs["additional_special_tokens"] += [
            self.get_lang_token(lowerCAmelCase__							)
            for lang_code in fairseq_language_code
            if self.get_lang_token(lowerCAmelCase__							) not in kwargs["additional_special_tokens"]
        ]

        super().__init__(
            src_lang=lowerCAmelCase__		,      tgt_lang=lowerCAmelCase__		,      bos_token=lowerCAmelCase__		,      eos_token=lowerCAmelCase__		,      sep_token=lowerCAmelCase__		,      unk_token=lowerCAmelCase__		,      pad_token=lowerCAmelCase__		,      language_codes=lowerCAmelCase__		,      sp_model_kwargs=self.sp_model_kwargs		,      num_madeup_words=lowerCAmelCase__		,      **lowerCAmelCase__		,      )

        a__	:					Dict										=vocab_file
        a__	:					Optional[int]										=load_json(lowerCAmelCase__							)
        a__	:					Optional[Any]										={v: k for k, v in self.encoder.items()}
        a__	:					Union[str, Any]										=spm_file
        a__	:					Optional[Any]										=load_spm(lowerCAmelCase__		,      self.sp_model_kwargs							)

        a__	:					int										=len(self.encoder							)

        a__	:					Union[str, Any]										={
            self.get_lang_token(lowerCAmelCase__							): self.encoder_size + i for i, lang_code in enumerate(lowerCAmelCase__							)
        }
        a__	:					int										={lang_code: self.encoder_size + i for i, lang_code in enumerate(lowerCAmelCase__							)}
        a__	:					Optional[Any]										={v: k for k, v in self.lang_token_to_id.items()}

        a__	:					Dict										=src_lang if src_lang is not None else "en"
        a__	:					List[Any]										=tgt_lang
        a__	:					Dict										=self.get_lang_id(self._src_lang							)
        self.set_src_lang_special_tokens(self._src_lang							)

        a__	:					Optional[Any]										=num_madeup_words







      @property
      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        return len(self.encoder							) + len(self.lang_token_to_id							)







      @property
      def    _lowercase  (    self							)   ->						str:



        '''simple docstring'''




        return self._src_lang







      @src_lang.setter
      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						None:



        '''simple docstring'''




        a__	:					Optional[Any]										=new_src_lang
        self.set_src_lang_special_tokens(self._src_lang							)







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						List[str]:



        '''simple docstring'''




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







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						Tuple:



        '''simple docstring'''




        if token in self.lang_token_to_id:
          return self.lang_token_to_id[token]
        return self.encoder.get(lowerCAmelCase__		,      self.encoder[self.unk_token]							)







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						str:



        '''simple docstring'''




        if index in self.id_to_lang_token:
          return self.id_to_lang_token[index]
        return self.decoder.get(lowerCAmelCase__		,      self.unk_token							)







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						Any:



        '''simple docstring'''




        a__	:					int										=[]
        a__	:					str										=""
        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(lowerCAmelCase__							) + token
            a__	:					Optional[int]										=[]
          else:
            current_sub_tokens.append(lowerCAmelCase__							)
        out_string += self.sp_model.decode(lowerCAmelCase__							)
        return out_string.strip()







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = False							)   ->						List[int]:



        '''simple docstring'''




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

        a__	:					List[str]										=[1] * len(self.prefix_tokens							)
        a__	:					Any										=[1] * len(self.suffix_tokens							)
        if token_ids_a is None:
          return prefix_ones + ([0] * len(lowerCAmelCase__							)) + suffix_ones
        return prefix_ones + ([0] * len(lowerCAmelCase__							)) + ([0] * len(lowerCAmelCase__							)) + suffix_ones







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = None							)   ->						List[int]:



        '''simple docstring'''




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







      def    _lowercase  (    self							)   ->						Dict:



        '''simple docstring'''




        a__	:					Any										={self.convert_ids_to_tokens(lowerCAmelCase__							): i for i in range(self.vocab_size							)}
        vocab.update(self.added_tokens_encoder							)
        return vocab







      def __getstate__(    self							)   ->						Dict:



        '''simple docstring'''




        a__	:					Optional[int]										=self.__dict__.copy()
        a__	:					Dict										=None
        return state







      def __setstate__(    self		,      lowerCAmelCase__							)   ->						None:



        '''simple docstring'''




        a__	:					Tuple										=d

        # for backward compatibility
        if not hasattr(self		,      "sp_model_kwargs"							):
          a__	:					List[Any]										={}

        a__	:					List[str]										=load_spm(self.spm_file		,      self.sp_model_kwargs							)







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = None							)   ->						Tuple[str]:



        '''simple docstring'''




        a__	:					Any										=Path(lowerCAmelCase__							)
        if not save_dir.is_dir():
          raise OSError(F'''{save_directory} should be a directory'''							)
        a__	:					str										=save_dir / (
            (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["vocab_file"]
        )
        a__	:					List[str]										=save_dir / (
            (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["spm_file"]
        )

        save_json(self.encoder		,      lowerCAmelCase__							)

        if os.path.abspath(self.spm_file							) != os.path.abspath(lowerCAmelCase__							) and os.path.isfile(self.spm_file							):
          copyfile(self.spm_file		,      lowerCAmelCase__							)
        elif not os.path.isfile(self.spm_file							):
          with open(lowerCAmelCase__		,      "wb"							) as fi:
            a__	:					Optional[Any]										=self.sp_model.serialized_model_proto()
            fi.write(lowerCAmelCase__							)

        return (str(lowerCAmelCase__							), str(lowerCAmelCase__							))







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = "en"		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = "ro"		,      **lowerCAmelCase__		,      )   ->						BatchEncoding:



        '''simple docstring'''




        a__	:					List[Any]										=src_lang
        a__	:					Any										=tgt_lang
        self.set_src_lang_special_tokens(self.src_lang							)
        return super().prepare_seqaseq_batch(lowerCAmelCase__		,      lowerCAmelCase__		,      **lowerCAmelCase__							)







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__		,      **lowerCAmelCase__							)   ->						str:



        '''simple docstring'''




        if src_lang is None or tgt_lang is None:
          raise ValueError("Translation requires a `src_lang` and a `tgt_lang` for this model"							)
        a__	:					int										=src_lang
        a__	:					Any										=self(lowerCAmelCase__		,      add_special_tokens=lowerCAmelCase__		,      **lowerCAmelCase__							)
        a__	:					Union[str, Any]										=self.get_lang_id(lowerCAmelCase__							)
        a__	:					Tuple										=tgt_lang_id
        return inputs







      def    _lowercase  (    self							)   ->						Tuple:



        '''simple docstring'''




        self.set_src_lang_special_tokens(self.src_lang							)







      def    _lowercase  (    self							)   ->						List[Any]:



        '''simple docstring'''




        self.set_tgt_lang_special_tokens(self.tgt_lang							)







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						None:



        '''simple docstring'''




        a__	:					Union[str, Any]										=self.get_lang_token(lowerCAmelCase__							)
        a__	:					str										=self.lang_token_to_id[lang_token]
        a__	:					Optional[int]										=[self.cur_lang_id]
        a__	:					List[Any]										=[self.eos_token_id]







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						None:



        '''simple docstring'''




        a__	:					int										=self.get_lang_token(lowerCAmelCase__							)
        a__	:					Optional[Any]										=self.lang_token_to_id[lang_token]
        a__	:					int										=[self.cur_lang_id]
        a__	:					Optional[Any]										=[self.eos_token_id]







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						str:



        '''simple docstring'''




        return self.lang_code_to_token[lang]







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						int:



        '''simple docstring'''




        a__	:					Optional[int]										=self.get_lang_token(lowerCAmelCase__							)
        return self.lang_token_to_id[lang_token]
def      _A       (		SCREAMING_SNAKE_CASE     :   str       ,       SCREAMING_SNAKE_CASE     :   Dict[str, Any]						):





  """simple docstring"""



  a__	:					int										=sentencepiece.SentencePieceProcessor(**SCREAMING_SNAKE_CASE						)
  spm.Load(str(SCREAMING_SNAKE_CASE						)						)
  return spm
def      _A       (		SCREAMING_SNAKE_CASE     :   str						):





  """simple docstring"""



  with open(SCREAMING_SNAKE_CASE       ,       "r"						) as f:
    return json.load(SCREAMING_SNAKE_CASE						)





def      _A       (		SCREAMING_SNAKE_CASE     :   int       ,       SCREAMING_SNAKE_CASE     :   str						):





  """simple docstring"""



  with open(SCREAMING_SNAKE_CASE       ,       "w"						) as f:
    json.dump(SCREAMING_SNAKE_CASE       ,       SCREAMING_SNAKE_CASE       ,       indent=2						)













import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
     import torch

if is_vision_available():
     from PIL import Image

     from transformers import MobileViTImageProcessor







class 				__lowerCAmelCase       (    unittest.TestCase):







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__=7		,      lowerCAmelCase__=3		,      lowerCAmelCase__=1_8		,      lowerCAmelCase__=3_0		,      lowerCAmelCase__=4_0_0		,      lowerCAmelCase__=True		,      lowerCAmelCase__=None		,      lowerCAmelCase__=True		,      lowerCAmelCase__=None		,      lowerCAmelCase__=True		,      )   ->						Optional[Any]:



        '''simple docstring'''




        a__	:					Union[str, Any]										=size if size is not None else {"shortest_edge": 2_0}
        a__	:					List[str]										=crop_size if crop_size is not None else {"height": 1_8, "width": 1_8}
        a__	:					Tuple										=parent
        a__	:					Union[str, Any]										=batch_size
        a__	:					List[str]										=num_channels
        a__	:					List[Any]										=image_size
        a__	:					str										=min_resolution
        a__	:					Optional[int]										=max_resolution
        a__	:					Tuple										=do_resize
        a__	:					Union[str, Any]										=size
        a__	:					List[Any]										=do_center_crop
        a__	:					List[str]										=crop_size
        a__	:					Optional[int]										=do_flip_channel_order







      def    _lowercase  (    self							)   ->						Optional[int]:



        '''simple docstring'''




        return {
            "do_resize": self.do_resize,
            "size": self.size,
            "do_center_crop": self.do_center_crop,
            "crop_size": self.crop_size,
            "do_flip_channel_order": self.do_flip_channel_order,
        }







@require_torch
@require_vision
class 				__lowerCAmelCase       (    UpperCamelCase__						,				unittest.TestCase):
      _lowercase				:					int						=			MobileViTImageProcessor if is_vision_available() else None







      def    _lowercase  (    self							)   ->						Optional[Any]:



        '''simple docstring'''




        a__	:					Tuple										=MobileViTImageProcessingTester(self							)







      @property
      def    _lowercase  (    self							)   ->						List[str]:



        '''simple docstring'''




        return self.image_processor_tester.prepare_image_processor_dict()







      def    _lowercase  (    self							)   ->						List[str]:



        '''simple docstring'''




        a__	:					str										=self.image_processing_class(**self.image_processor_dict							)
        self.assertTrue(hasattr(lowerCAmelCase__		,      "do_resize"							)							)
        self.assertTrue(hasattr(lowerCAmelCase__		,      "size"							)							)
        self.assertTrue(hasattr(lowerCAmelCase__		,      "do_center_crop"							)							)
        self.assertTrue(hasattr(lowerCAmelCase__		,      "center_crop"							)							)
        self.assertTrue(hasattr(lowerCAmelCase__		,      "do_flip_channel_order"							)							)







      def    _lowercase  (    self							)   ->						Optional[int]:



        '''simple docstring'''




        a__	:					List[Any]										=self.image_processing_class.from_dict(self.image_processor_dict							)
        self.assertEqual(image_processor.size		,      {"shortest_edge": 2_0}							)
        self.assertEqual(image_processor.crop_size		,      {"height": 1_8, "width": 1_8}							)

        a__	:					Union[str, Any]										=self.image_processing_class.from_dict(self.image_processor_dict		,      size=4_2		,      crop_size=8_4							)
        self.assertEqual(image_processor.size		,      {"shortest_edge": 4_2}							)
        self.assertEqual(image_processor.crop_size		,      {"height": 8_4, "width": 8_4}							)







      def    _lowercase  (    self							)   ->						Optional[Any]:



        '''simple docstring'''




        pass







      def    _lowercase  (    self							)   ->						Tuple:



        '''simple docstring'''




        a__	:					List[str]										=self.image_processing_class(**self.image_processor_dict							)
        # create random PIL images
        a__	:					Any										=prepare_image_inputs(self.image_processor_tester		,      equal_resolution=lowerCAmelCase__							)
        for image in image_inputs:
          self.assertIsInstance(lowerCAmelCase__		,      Image.Image							)

        # Test not batched input
        a__	:					Tuple										=image_processing(image_inputs[0]		,      return_tensors="pt"							).pixel_values
        self.assertEqual(
            encoded_images.shape		,      (
                1,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size["height"],
                self.image_processor_tester.crop_size["width"],
            )		,      )

        # Test batched
        a__	:					List[Any]										=image_processing(lowerCAmelCase__		,      return_tensors="pt"							).pixel_values
        self.assertEqual(
            encoded_images.shape		,      (
                self.image_processor_tester.batch_size,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size["height"],
                self.image_processor_tester.crop_size["width"],
            )		,      )







      def    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        a__	:					Optional[Any]										=self.image_processing_class(**self.image_processor_dict							)
        # create random numpy tensors
        a__	:					List[Any]										=prepare_image_inputs(self.image_processor_tester		,      equal_resolution=lowerCAmelCase__		,      numpify=lowerCAmelCase__							)
        for image in image_inputs:
          self.assertIsInstance(lowerCAmelCase__		,      np.ndarray							)

        # Test not batched input
        a__	:					Tuple										=image_processing(image_inputs[0]		,      return_tensors="pt"							).pixel_values
        self.assertEqual(
            encoded_images.shape		,      (
                1,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size["height"],
                self.image_processor_tester.crop_size["width"],
            )		,      )

        # Test batched
        a__	:					int										=image_processing(lowerCAmelCase__		,      return_tensors="pt"							).pixel_values
        self.assertEqual(
            encoded_images.shape		,      (
                self.image_processor_tester.batch_size,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size["height"],
                self.image_processor_tester.crop_size["width"],
            )		,      )







      def    _lowercase  (    self							)   ->						Optional[Any]:



        '''simple docstring'''




        a__	:					int										=self.image_processing_class(**self.image_processor_dict							)
        # create random PyTorch tensors
        a__	:					Tuple										=prepare_image_inputs(self.image_processor_tester		,      equal_resolution=lowerCAmelCase__		,      torchify=lowerCAmelCase__							)
        for image in image_inputs:
          self.assertIsInstance(lowerCAmelCase__		,      torch.Tensor							)

        # Test not batched input
        a__	:					List[str]										=image_processing(image_inputs[0]		,      return_tensors="pt"							).pixel_values
        self.assertEqual(
            encoded_images.shape		,      (
                1,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size["height"],
                self.image_processor_tester.crop_size["width"],
            )		,      )

        # Test batched
        a__	:					List[str]										=image_processing(lowerCAmelCase__		,      return_tensors="pt"							).pixel_values
        self.assertEqual(
            encoded_images.shape		,      (
                self.image_processor_tester.batch_size,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size["height"],
                self.image_processor_tester.crop_size["width"],
            )		,      )






