






"""simple docstring"""


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       :					Tuple									=		logging.get_logger(__name__)

_UpperCamelCase       :					str									=		'▁'

_UpperCamelCase       :					str									=		{
    'vocab_file': 'vocab.json',
    'spm_file': 'sentencepiece.bpe.model',
    'tokenizer_config_file': 'tokenizer_config.json',
}

_UpperCamelCase       :					Optional[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       :					List[str]									=		{
    'facebook/m2m100_418M': 1024,
}

# fmt: off
_UpperCamelCase       :					Optional[Any]									=		{
    '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 		snake_case						(						UpperCAmelCase ):
    __magic_name__					  =  VOCAB_FILES_NAMES
    __magic_name__					  =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    __magic_name__					  =  PRETRAINED_VOCAB_FILES_MAP
    __magic_name__					  =  ['''input_ids''', '''attention_mask''']

    __magic_name__					  =  []
    __magic_name__					  =  []





    def __init__(							self			:      str       ,	A			:      Union[str, Any]       ,	A			:      Any       ,	A			:      List[str]=None       ,	A			:      Tuple=None       ,	A			:      Optional[Any]="<s>"       ,	A			:      List[str]="</s>"       ,	A			:      Tuple="</s>"       ,	A			:      Dict="<pad>"       ,	A			:      List[str]="<unk>"       ,	A			:      List[str]="m2m100"       ,	A			:      Optional[Dict[str, Any]] = None       ,	A			:      Tuple=8       ,	**A			:      Union[str, Any]       ,	):


      '''simple docstring'''




      a  :    List[str]       =     {} if sp_model_kwargs is None else sp_model_kwargs

      a  :    Any       =     language_codes
      a  :    Optional[int]       =     FAIRSEQ_LANGUAGE_CODES[language_codes]
      a  :    Optional[Any]       =     {lang_code: F'''__{lang_code}__''' for lang_code in fairseq_language_code}

      a  :    Tuple       =     kwargs.get('additional_special_tokens'       ,	[]    )
      kwargs["additional_special_tokens"] += [
          self.get_lang_token(A    )
          for lang_code in fairseq_language_code
          if self.get_lang_token(A    ) not in kwargs["additional_special_tokens"]
      ]

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

      a  :    Union[str, Any]       =     vocab_file
      a  :    str       =     load_json(A    )
      a  :    Dict       =     {v: k for k, v in self.encoder.items()}
      a  :    Tuple       =     spm_file
      a  :    List[str]       =     load_spm(A       ,	self.sp_model_kwargs    )

      a  :    Optional[Any]       =     len(self.encoder    )

      a  :    Tuple       =     {
          self.get_lang_token(A    ): self.encoder_size + i for i, lang_code in enumerate(A    )
      }
      a  :    Tuple       =     {lang_code: self.encoder_size + i for i, lang_code in enumerate(A    )}
      a  :    Optional[Any]       =     {v: k for k, v in self.lang_token_to_id.items()}

      a  :    Tuple       =     src_lang if src_lang is not None else 'en'
      a  :    Union[str, Any]       =     tgt_lang
      a  :    Optional[int]       =     self.get_lang_id(self._src_lang    )
      self.set_src_lang_special_tokens(self._src_lang    )

      a  :    Optional[Any]       =     num_madeup_words





    @property
    def        lowerCamelCase__   (							self			:      Dict    ):


      '''simple docstring'''




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





    @property
    def        lowerCamelCase__   (							self			:      Any    ):


      '''simple docstring'''




      return self._src_lang





    @src_lang.setter
    def        lowerCamelCase__   (							self			:      Optional[int]       ,	A			:      str    ):


      '''simple docstring'''




      a  :    List[str]       =     new_src_lang
      self.set_src_lang_special_tokens(self._src_lang    )





    def        lowerCamelCase__   (							self			:      Optional[Any]       ,	A			:      str    ):


      '''simple docstring'''




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





    def        lowerCamelCase__   (							self			:      Tuple       ,	A			:      Dict    ):


      '''simple docstring'''




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





    def        lowerCamelCase__   (							self			:      str       ,	A			:      int    ):


      '''simple docstring'''




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





    def        lowerCamelCase__   (							self			:      List[str]       ,	A			:      Tuple    ):


      '''simple docstring'''




      a  :    Optional[Any]       =     []
      a  :    Dict       =     ''
      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
          a  :    List[Any]       =     []
        else:
          current_sub_tokens.append(A    )
      out_string += self.sp_model.decode(A    )
      return out_string.strip()





    def        lowerCamelCase__   (							self			:      List[Any]       ,	A			:      List[int]       ,	A			:      Optional[List[int]] = None       ,	A			:      bool = False    ):


      '''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    )

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





    def        lowerCamelCase__   (							self			:      Any       ,	A			:      List[int]       ,	A			:      Optional[List[int]] = None    ):


      '''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        lowerCamelCase__   (							self			:      Union[str, Any]    ):


      '''simple docstring'''




      a  :    List[str]       =     {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[Any]    ):


      '''simple docstring'''




      a  :    Optional[int]       =     self.__dict__.copy()
      a  :    Optional[Any]       =     None
      return state





    def __setstate__(							self			:      Any       ,	A			:      Dict    ):


      '''simple docstring'''




      a  :    Tuple       =     d

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

      a  :    Tuple       =     load_spm(self.spm_file       ,	self.sp_model_kwargs    )





    def        lowerCamelCase__   (							self			:      Optional[Any]       ,	A			:      str       ,	A			:      Optional[str] = None    ):


      '''simple docstring'''




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

      save_json(self.encoder       ,	A    )

      if os.path.abspath(self.spm_file    ) != os.path.abspath(A    ) and os.path.isfile(self.spm_file    ):
        copyfile(self.spm_file       ,	A    )
      elif not os.path.isfile(self.spm_file    ):
        with open(A       ,	'wb'    ) as fi:
          a  :    Union[str, Any]       =     self.sp_model.serialized_model_proto()
          fi.write(A    )

      return (str(A    ), str(A    ))





    def        lowerCamelCase__   (							self			:      int       ,	A			:      List[str]       ,	A			:      str = "en"       ,	A			:      Optional[List[str]] = None       ,	A			:      str = "ro"       ,	**A			:      List[Any]       ,	):


      '''simple docstring'''




      a  :    Optional[int]       =     src_lang
      a  :    Dict       =     tgt_lang
      self.set_src_lang_special_tokens(self.src_lang    )
      return super().prepare_seqaseq_batch(A       ,	A       ,	**A    )





    def        lowerCamelCase__   (							self			:      List[Any]       ,	A			:      str       ,	A			:      Optional[str]       ,	A			:      Optional[str]       ,	**A			:      Union[str, Any]    ):


      '''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  :    List[str]       =     src_lang
      a  :    str       =     self(A       ,	add_special_tokens=A       ,	**A    )
      a  :    Optional[Any]       =     self.get_lang_id(A    )
      a  :    Optional[Any]       =     tgt_lang_id
      return inputs





    def        lowerCamelCase__   (							self			:      str    ):


      '''simple docstring'''




      self.set_src_lang_special_tokens(self.src_lang    )





    def        lowerCamelCase__   (							self			:      List[Any]    ):


      '''simple docstring'''




      self.set_tgt_lang_special_tokens(self.tgt_lang    )





    def        lowerCamelCase__   (							self			:      Optional[Any]       ,	A			:      str    ):


      '''simple docstring'''




      a  :    Dict       =     self.get_lang_token(A    )
      a  :    str       =     self.lang_token_to_id[lang_token]
      a  :    int       =     [self.cur_lang_id]
      a  :    List[str]       =     [self.eos_token_id]





    def        lowerCamelCase__   (							self			:      Any       ,	A			:      str    ):


      '''simple docstring'''




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





    def        lowerCamelCase__   (							self			:      int       ,	A			:      str    ):


      '''simple docstring'''




      return self.lang_code_to_token[lang]





    def        lowerCamelCase__   (							self			:      List[Any]       ,	A			:      str    ):


      '''simple docstring'''




      a  :    Optional[Any]       =     self.get_lang_token(A    )
      return self.lang_token_to_id[lang_token]




def     snake_case						(A_   :str			,    A_   :Dict[str, Any]   ):







  '''simple docstring'''

  a  :    Optional[int]       =     sentencepiece.SentencePieceProcessor(**A_   )
  spm.Load(str(A_   )   )
  return spm




def     snake_case						(A_   :str   ):







  '''simple docstring'''

  with open(A_			,    'r'   ) as f:
    return json.load(A_   )






def     snake_case						(A_   :Tuple			,    A_   :str   ):







  '''simple docstring'''

  with open(A_			,    'w'   ) as f:
    json.dump(A_			,    A_			,    indent=2   )












"""simple docstring"""


import torch
from torch import nn

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin






class 		snake_case						(						UpperCAmelCase     ,				UpperCAmelCase ):





    @register_to_config
    def __init__(							self			:      Dict       ,	*,
        A			:      int = 4       ,	A			:      int = 7_6_8       ,	A			:      int       ,	A			:      Any       ,	):


      '''simple docstring'''




      super().__init__()

      a  :    Optional[Any]       =     nn.Parameter(torch.zeros(A    )    )

      # parameters for additional clip time embeddings
      a  :    str       =     nn.Linear(A       ,	A    )
      a  :    List[Any]       =     nn.Linear(A       ,	A    )

      # parameters for encoder hidden states
      a  :    Optional[Any]       =     clip_extra_context_tokens
      a  :    Union[str, Any]       =     nn.Linear(
          A       ,	self.clip_extra_context_tokens * cross_attention_dim    )
      a  :    Optional[Any]       =     nn.Linear(A       ,	A    )
      a  :    Any       =     nn.LayerNorm(A    )





    def        lowerCamelCase__   (							self			:      Dict       ,	*, A			:      str       ,	A			:      Optional[Any]       ,	A			:      int       ,	A			:      Dict    ):


      '''simple docstring'''




      if do_classifier_free_guidance:
        # Add the classifier free guidance embeddings to the image embeddings
        a  :    Dict       =     image_embeddings.shape[0]
        a  :    Tuple       =     self.learned_classifier_free_guidance_embeddings.unsqueeze(0    )
        a  :    List[str]       =     classifier_free_guidance_embeddings.expand(
            A       ,	-1    )
        a  :    Union[str, Any]       =     torch.cat([classifier_free_guidance_embeddings, image_embeddings]       ,	dim=0    )

      # The image embeddings batch size and the text embeddings batch size are equal
      assert image_embeddings.shape[0] == prompt_embeds.shape[0]

      a  :    Dict       =     prompt_embeds.shape[0]

      # "Specifically, we modify the architecture described in Nichol et al. (2021) by projecting and
      # adding CLIP embeddings to the existing timestep embedding, ...
      a  :    Optional[int]       =     self.embedding_proj(A    )
      a  :    Optional[int]       =     self.clip_image_embeddings_project_to_time_embeddings(A    )
      a  :    Tuple       =     time_projected_image_embeddings + time_projected_prompt_embeds

      # ... and by projecting CLIP embeddings into four
      # extra tokens of context that are concatenated to the sequence of outputs from the GLIDE text encoder"
      a  :    Union[str, Any]       =     self.clip_extra_context_tokens_proj(A    )
      a  :    Dict       =     clip_extra_context_tokens.reshape(A       ,	-1       ,	self.clip_extra_context_tokens    )
      a  :    Union[str, Any]       =     clip_extra_context_tokens.permute(0       ,	2       ,	1    )

      a  :    Tuple       =     self.encoder_hidden_states_proj(A    )
      a  :    Optional[Any]       =     self.text_encoder_hidden_states_norm(A    )
      a  :    Optional[int]       =     torch.cat([clip_extra_context_tokens, text_encoder_hidden_states]       ,	dim=1    )

      return text_encoder_hidden_states, additive_clip_time_embeddings




