codingutf8 2019 the open ai team s and the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache orglicenseslicense2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license tokenization classes for fsmt import json import os import re import unicodedata from typing import dict list optional tuple from tokenizationutils import pretrainedtokenizer from utils import logging logger logging getloggername vocabfilesnames srcvocabfile vocabsrc json tgtvocabfile vocabtgt json mergesfile merges txt pretrainedvocabfilesmap srcvocabfile stastinywmt19ende https huggingface costastinywmt19enderesolvemainvocabsrc json tgtvocabfile stastinywmt19ende https huggingface costastinywmt19enderesolvemainvocabtgt json mergesfile stastinywmt19ende https huggingface costastinywmt19enderesolvemainmerges txt pretrainedpositionalembeddingssizes stastinywmt19ende 1024 pretrainedinitconfiguration stastinywmt19ende langs en de modelmaxlength 1024 specialtokensmapfile none fulltokenizerfile none def getpairsword pairs set prevchar word0 for char in word1 pairs addprevchar char prevchar char return pairs def replaceunicodepuncttext text text replace text re subrs text text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace 1 text text replace text text replace text text replace 0 text text replace 3 text text replace 2 text text replace 5 text text replace 6 text text replace 9 text text replace 7 text text replace 8 text text replace 4 text re subrs text text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace text text replace return text def removenonprintingchartext output for char in text cat unicodedata categorychar if cat startswithc continue output appendchar return joinoutput porting notes this one is modeled after xlmtokenizer added srcvocabfile tgtvocabfile langs class fsmttokenizerpretrainedtokenizer vocabfilesnames vocabfilesnames pretrainedvocabfilesmap pretrainedvocabfilesmap pretrainedinitconfiguration pretrainedinitconfiguration maxmodelinputsizes pretrainedpositionalembeddingssizes modelinputnames inputids attentionmask def init self langsnone srcvocabfilenone tgtvocabfilenone mergesfilenone dolowercasefalse unktokenunk bostokens septokens padtokenpad kwargs try import sacremoses except importerror raise importerror you need to install sacremoses to use xlmtokenizer see https pypi orgprojectsacremoses for installation self sm sacremoses self srcvocabfile srcvocabfile self tgtvocabfile tgtvocabfile self mergesfile mergesfile self dolowercase dolowercase cache of sm mosespunctnormalizer instance self cachemosespunctnormalizer cache of sm mosestokenizer instance self cachemosestokenizer self cachemosesdetokenizer if langs and lenlangs 2 self srclang self tgtlang langs else raise valueerror farg langs needs to be a list of 2 langs e g en ru but got langs usually that means that tokenizer can t find a mapping for the given model path in pretrainedvocabfilesmap and other maps of this tokenizer with opensrcvocabfile encodingutf8 as srcvocabhandle self encoder json loadsrcvocabhandle with opentgtvocabfile encodingutf8 as tgtvocabhandle tgtvocab json loadtgtvocabhandle self decoder v k for k v in tgtvocab items with openmergesfile encodingutf8 as mergeshandle merges mergeshandle read splitn 1 merges tuplemerge split 2 for merge in merges self bperanks dictzipmerges rangelenmerges self cache super init langslangs srcvocabfilesrcvocabfile tgtvocabfiletgtvocabfile mergesfilemergesfile dolowercasedolowercase unktokenunktoken bostokenbostoken septokenseptoken padtokenpadtoken kwargs hack override def getvocabself dictstr int return self getsrcvocab hack override property def vocabsizeself int return self srcvocabsize def mosespunctnormself text lang if lang not in self cachemosespunctnormalizer punctnormalizer self sm mosespunctnormalizerlanglang self cachemosespunctnormalizerlang punctnormalizer return self cachemosespunctnormalizerlang normalizetext def mosestokenizeself text lang if lang not in self cachemosestokenizer mosestokenizer self sm mosestokenizerlanglang self cachemosestokenizerlang mosestokenizer return self cachemosestokenizerlang tokenize text aggressivedashsplitstrue returnstrfalse escapetrue def mosesdetokenizeself tokens lang if lang not in self cachemosesdetokenizer mosesdetokenizer self sm mosesdetokenizerlanglang self cachemosesdetokenizerlang mosesdetokenizer return self cachemosesdetokenizerlang detokenizetokens def mosespipelineself text lang text replaceunicodepuncttext text self mosespunctnormtext lang text removenonprintingchartext return text property def srcvocabsizeself return lenself encoder property def tgtvocabsizeself return lenself decoder def getsrcvocabself return dictself encoder self addedtokensencoder def gettgtvocabself return dictself decoder self addedtokensdecoder def bpeself token word tupletoken 1 token1 w if token in self cache return self cachetoken pairs getpairsword if not pairs return token w while true bigram minpairs keylambda pair self bperanks getpair floatinf if bigram not in self bperanks break first second bigram newword i 0 while i lenword try j word indexfirst i except valueerror newword extendwordi break else newword extendwordi j i j if wordi first and i lenword 1 and wordi 1 second newword appendfirst second i 2 else newword appendwordi i 1 newword tuplenewword word newword if lenword 1 break else pairs getpairsword word joinword if word n w word nw self cachetoken word return word def tokenizeself text langen bypasstokenizerfalse ignore lang which is currently isn t explicitly passed in tokenizationutils py and always results in langen if lang self srclang raise valueerrorfexpected langself srclang but got lang lang self srclang if self dolowercase text text lower if bypasstokenizer text text split else text self mosespipelinetext langlang text self mosestokenizetext langlang splittokens for token in text if token splittokens extendlistself bpetoken split return splittokens def converttokentoidself token converts an index integer in a token str using the vocab return self decoder getindex self unktoken def converttokenstostringself tokens remove bpe detokenize build model inputs from a sequence or a pair of sequence for sequence classification tasks by concatenating and adding special tokens a fairseq transformer sequence has the following format single sequence s x s pair of sequences s a s b s args tokenids0 listint list of ids to which the special tokens will be added tokenids1 listint optional optional second list of ids for sequence pairs returns listint list of input ids glossaryinputids with the appropriate special tokens no bos used in fairseq retrieve sequence ids from a token list that has no special tokens added this method is called when adding special tokens using the tokenizer prepareformodel method args tokenids0 listint list of ids tokenids1 listint optional optional second list of ids for sequence pairs alreadyhasspecialtokens bool optional defaults to false whether or not the token list is already formatted with special tokens for the model returns listint a list of integers in the range 0 1 1 for a special token 0 for a sequence token no bos used in fairseq create a mask from the two sequences passed to be used in a sequencepair classification task a fairseq transformer sequence pair mask has the following format 0 0 0 0 0 0 0 0 0 0 0 1 1 1 1 1 1 1 1 1 first sequence second sequence if tokenids1 is none this method only returns the first portion of the mask 0s args tokenids0 listint list of ids tokenids1 listint optional optional second list of ids for sequence pairs returns listint list of token type ids glossarytokentypeids according to the given sequences creates a mask from the two sequences passed to be used in a sequencepair classification task an fairseqtransformer sequence pair mask has the following format no bos used in fairseq coding utf 8 2019 the open ai team s and the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache org licenses license 2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license tokenization classes for fsmt return set of symbol pairs in a word word is represented as tuple of symbols symbols being variable length strings port of https github com moses smt mosesdecoder blob master scripts tokenizer replace unicode punctuation perl port of https github com moses smt mosesdecoder blob master scripts tokenizer remove non printing char perl porting notes this one is modeled after xlmtokenizer added src_vocab_file tgt_vocab_file langs construct an fairseq transformer tokenizer based on byte pair encoding the tokenization process is the following moses preprocessing and tokenization normalizing all inputs text the arguments special_tokens and the function set_special_tokens can be used to add additional symbols like __classify__ to a vocabulary the argument langs defines a pair of languages this tokenizer inherits from pretrainedtokenizer which contains most of the main methods users should refer to this superclass for more information regarding those methods args langs list str optional a list of two languages to translate from and to for instance en ru src_vocab_file str optional file containing the vocabulary for the source language tgt_vocab_file st optional file containing the vocabulary for the target language merges_file str optional file containing the merges do_lower_case bool optional defaults to false whether or not to lowercase the input when tokenizing unk_token str optional defaults to unk the unknown token a token that is not in the vocabulary cannot be converted to an id and is set to be this token instead bos_token str optional defaults to s the beginning of sequence token that was used during pretraining can be used a sequence classifier token tip when building a sequence using special tokens this is not the token that is used for the beginning of sequence the token used is the cls_token tip sep_token str optional defaults to s the separator token which is used when building a sequence from multiple sequences e g two sequences for sequence classification or for a text and a question for question answering it is also used as the last token of a sequence built with special tokens pad_token str optional defaults to pad the token used for padding for example when batching sequences of different lengths cache of sm mosespunctnormalizer instance cache of sm mosestokenizer instance hack override hack override tokenize a string given language code using moses details of tokenization sacremoses https github com alvations sacremoses port of moses install with pip install sacremoses args lang iso language code default en string languages should belong of the model supported languages however we don t enforce it bypass_tokenizer allow users to preprocess and tokenize the sentences externally default false bool if true we only apply bpe returns list of tokens ignore lang which is currently isn t explicitly passed in tokenization_utils py and always results in lang en if lang self src_lang raise valueerror f expected lang self src_lang but got lang converts a token str in an id using the vocab converts an index integer in a token str using the vocab converts a sequence of tokens string in a single string remove bpe detokenize build model inputs from a sequence or a pair of sequence for sequence classification tasks by concatenating and adding special tokens a fairseq transformer sequence has the following format single sequence s x s pair of sequences s a s b s args token_ids_0 list int list of ids to which the special tokens will be added token_ids_1 list int optional optional second list of ids for sequence pairs returns list int list of input ids glossary input ids with the appropriate special tokens no bos used in fairseq retrieve sequence ids from a token list that has no special tokens added this method is called when adding special tokens using the tokenizer prepare_for_model method args token_ids_0 list int list of ids token_ids_1 list int optional optional second list of ids for sequence pairs already_has_special_tokens bool optional defaults to false whether or not the token list is already formatted with special tokens for the model returns list int a list of integers in the range 0 1 1 for a special token 0 for a sequence token no bos used in fairseq create a mask from the two sequences passed to be used in a sequence pair classification task a fairseq transformer sequence pair mask has the following format 0 0 0 0 0 0 0 0 0 0 0 1 1 1 1 1 1 1 1 1 first sequence second sequence if token_ids_1 is none this method only returns the first portion of the mask 0s args token_ids_0 list int list of ids token_ids_1 list int optional optional second list of ids for sequence pairs returns list int list of token type ids glossary token type ids according to the given sequence s creates a mask from the two sequences passed to be used in a sequence pair classification task an fairseq_transformer sequence pair mask has the following format no bos used in fairseq
import json
import os
import re
import unicodedata
from typing import Dict, List, Optional, Tuple

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


logger = logging.get_logger(__name__)

VOCAB_FILES_NAMES = {
    "src_vocab_file": "vocab-src.json",
    "tgt_vocab_file": "vocab-tgt.json",
    "merges_file": "merges.txt",
}

PRETRAINED_VOCAB_FILES_MAP = {
    "src_vocab_file": {
        "stas/tiny-wmt19-en-de": "https://huggingface.co/stas/tiny-wmt19-en-de/resolve/main/vocab-src.json"
    },
    "tgt_vocab_file": {
        "stas/tiny-wmt19-en-de": "https://huggingface.co/stas/tiny-wmt19-en-de/resolve/main/vocab-tgt.json"
    },
    "merges_file": {"stas/tiny-wmt19-en-de": "https://huggingface.co/stas/tiny-wmt19-en-de/resolve/main/merges.txt"},
}

PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES = {"stas/tiny-wmt19-en-de": 1024}
PRETRAINED_INIT_CONFIGURATION = {
    "stas/tiny-wmt19-en-de": {
        "langs": ["en", "de"],
        "model_max_length": 1024,
        "special_tokens_map_file": None,
        "full_tokenizer_file": None,
    }
}


def get_pairs(word):
    
    pairs = set()
    prev_char = word[0]
    for char in word[1:]:
        pairs.add((prev_char, char))
        prev_char = char
    return pairs


def replace_unicode_punct(text):
    
    text = text.replace("，", ",")
    text = re.sub(r"。\s*", ". ", text)
    text = text.replace("、", ",")
    text = text.replace("”", '"')
    text = text.replace("“", '"')
    text = text.replace("∶", ":")
    text = text.replace("：", ":")
    text = text.replace("？", "?")
    text = text.replace("《", '"')
    text = text.replace("》", '"')
    text = text.replace("）", ")")
    text = text.replace("！", "!")
    text = text.replace("（", "(")
    text = text.replace("；", ";")
    text = text.replace("１", "1")
    text = text.replace("」", '"')
    text = text.replace("「", '"')
    text = text.replace("０", "0")
    text = text.replace("３", "3")
    text = text.replace("２", "2")
    text = text.replace("５", "5")
    text = text.replace("６", "6")
    text = text.replace("９", "9")
    text = text.replace("７", "7")
    text = text.replace("８", "8")
    text = text.replace("４", "4")
    text = re.sub(r"．\s*", ". ", text)
    text = text.replace("～", "~")
    text = text.replace("’", "'")
    text = text.replace("…", "...")
    text = text.replace("━", "-")
    text = text.replace("〈", "<")
    text = text.replace("〉", ">")
    text = text.replace("【", "[")
    text = text.replace("】", "]")
    text = text.replace("％", "%")
    return text


def remove_non_printing_char(text):
    
    output = []
    for char in text:
        cat = unicodedata.category(char)
        if cat.startswith("C"):
            continue
        output.append(char)
    return "".join(output)











class FSMTTokenizer(PreTrainedTokenizer):
    

    vocab_files_names = VOCAB_FILES_NAMES
    pretrained_vocab_files_map = PRETRAINED_VOCAB_FILES_MAP
    pretrained_init_configuration = PRETRAINED_INIT_CONFIGURATION
    max_model_input_sizes = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    model_input_names = ["input_ids", "attention_mask"]

    def __init__(
        self,
        langs=None,
        src_vocab_file=None,
        tgt_vocab_file=None,
        merges_file=None,
        do_lower_case=False,
        unk_token="<unk>",
        bos_token="<s>",
        sep_token="</s>",
        pad_token="<pad>",
        **kwargs,
    ):
        try:
            import sacremoses
        except ImportError:
            raise ImportError(
                "You need to install sacremoses to use XLMTokenizer. "
                "See https://pypi.org/project/sacremoses/ for installation."
            )

        self.sm = sacremoses

        self.src_vocab_file = src_vocab_file
        self.tgt_vocab_file = tgt_vocab_file
        self.merges_file = merges_file
        self.do_lower_case = do_lower_case

        
        self.cache_moses_punct_normalizer = {}
        
        self.cache_moses_tokenizer = {}
        self.cache_moses_detokenizer = {}

        if langs and len(langs) == 2:
            self.src_lang, self.tgt_lang = langs
        else:
            raise ValueError(
                f"arg `langs` needs to be a list of 2 langs, e.g. ['en', 'ru'], but got {langs}. "
                "Usually that means that tokenizer can't find a mapping for the given model path "
                "in PRETRAINED_VOCAB_FILES_MAP, and other maps of this tokenizer."
            )

        with open(src_vocab_file, encoding="utf-8") as src_vocab_handle:
            self.encoder = json.load(src_vocab_handle)
        with open(tgt_vocab_file, encoding="utf-8") as tgt_vocab_handle:
            tgt_vocab = json.load(tgt_vocab_handle)
            self.decoder = {v: k for k, v in tgt_vocab.items()}
        with open(merges_file, encoding="utf-8") as merges_handle:
            merges = merges_handle.read().split("\n")[:-1]
        merges = [tuple(merge.split()[:2]) for merge in merges]
        self.bpe_ranks = dict(zip(merges, range(len(merges))))
        self.cache = {}
        super().__init__(
            langs=langs,
            src_vocab_file=src_vocab_file,
            tgt_vocab_file=tgt_vocab_file,
            merges_file=merges_file,
            do_lower_case=do_lower_case,
            unk_token=unk_token,
            bos_token=bos_token,
            sep_token=sep_token,
            pad_token=pad_token,
            **kwargs,
        )

    
    def get_vocab(self) -> Dict[str, int]:
        return self.get_src_vocab()

    
    @property
    def vocab_size(self) -> int:
        return self.src_vocab_size

    def moses_punct_norm(self, text, lang):
        if lang not in self.cache_moses_punct_normalizer:
            punct_normalizer = self.sm.MosesPunctNormalizer(lang=lang)
            self.cache_moses_punct_normalizer[lang] = punct_normalizer
        return self.cache_moses_punct_normalizer[lang].normalize(text)

    def moses_tokenize(self, text, lang):
        if lang not in self.cache_moses_tokenizer:
            moses_tokenizer = self.sm.MosesTokenizer(lang=lang)
            self.cache_moses_tokenizer[lang] = moses_tokenizer
        return self.cache_moses_tokenizer[lang].tokenize(
            text, aggressive_dash_splits=True, return_str=False, escape=True
        )

    def moses_detokenize(self, tokens, lang):
        if lang not in self.cache_moses_detokenizer:
            moses_detokenizer = self.sm.MosesDetokenizer(lang=lang)
            self.cache_moses_detokenizer[lang] = moses_detokenizer
        return self.cache_moses_detokenizer[lang].detokenize(tokens)

    def moses_pipeline(self, text, lang):
        text = replace_unicode_punct(text)
        text = self.moses_punct_norm(text, lang)
        text = remove_non_printing_char(text)
        return text

    @property
    def src_vocab_size(self):
        return len(self.encoder)

    @property
    def tgt_vocab_size(self):
        return len(self.decoder)

    def get_src_vocab(self):
        return dict(self.encoder, **self.added_tokens_encoder)

    def get_tgt_vocab(self):
        return dict(self.decoder, **self.added_tokens_decoder)

    def bpe(self, token):
        word = tuple(token[:-1]) + (token[-1] + "</w>",)
        if token in self.cache:
            return self.cache[token]
        pairs = get_pairs(word)

        if not pairs:
            return token + "</w>"

        while True:
            bigram = min(pairs, key=lambda pair: self.bpe_ranks.get(pair, float("inf")))
            if bigram not in self.bpe_ranks:
                break
            first, second = bigram
            new_word = []
            i = 0
            while i < len(word):
                try:
                    j = word.index(first, i)
                except ValueError:
                    new_word.extend(word[i:])
                    break
                else:
                    new_word.extend(word[i:j])
                    i = j

                if word[i] == first and i < len(word) - 1 and word[i + 1] == second:
                    new_word.append(first + second)
                    i += 2
                else:
                    new_word.append(word[i])
                    i += 1
            new_word = tuple(new_word)
            word = new_word
            if len(word) == 1:
                break
            else:
                pairs = get_pairs(word)
        word = " ".join(word)
        if word == "\n  </w>":
            word = "\n</w>"
        self.cache[token] = word
        return word

    def _tokenize(self, text, lang="en", bypass_tokenizer=False):
        
        
        
        
        lang = self.src_lang

        if self.do_lower_case:
            text = text.lower()

        if bypass_tokenizer:
            text = text.split()
        else:
            text = self.moses_pipeline(text, lang=lang)
            text = self.moses_tokenize(text, lang=lang)

        split_tokens = []
        for token in text:
            if token:
                split_tokens.extend(list(self.bpe(token).split(" ")))

        return split_tokens

    def _convert_token_to_id(self, token):
        
        return self.encoder.get(token, self.encoder.get(self.unk_token))

    def _convert_id_to_token(self, index):
        
        return self.decoder.get(index, self.unk_token)

    def convert_tokens_to_string(self, tokens):
        

        
        tokens = [t.replace(" ", "").replace("</w>", " ") for t in tokens]
        tokens = "".join(tokens).split()
        
        text = self.moses_detokenize(tokens, self.tgt_lang)
        return text

    def build_inputs_with_special_tokens(
        self, token_ids_0: List[int], token_ids_1: Optional[List[int]] = None
    ) -> List[int]:
        
        sep = [self.sep_token_id]

        
        if token_ids_1 is None:
            return token_ids_0 + sep
        return token_ids_0 + sep + token_ids_1 + sep

    def get_special_tokens_mask(
        self, token_ids_0: List[int], token_ids_1: Optional[List[int]] = None, already_has_special_tokens: bool = False
    ) -> List[int]:
        

        if already_has_special_tokens:
            return super().get_special_tokens_mask(
                token_ids_0=token_ids_0, token_ids_1=token_ids_1, already_has_special_tokens=True
            )
        
        if token_ids_1 is not None:
            return ([0] * len(token_ids_0)) + [1] + ([0] * len(token_ids_1)) + [1]
        return ([0] * len(token_ids_0)) + [1]

    def create_token_type_ids_from_sequences(
        self, token_ids_0: List[int], token_ids_1: Optional[List[int]] = None
    ) -> List[int]:
        
        sep = [self.sep_token_id]

        
        if token_ids_1 is None:
            return len(token_ids_0 + sep) * [0]
        return len(token_ids_0 + sep) * [0] + len(token_ids_1 + sep) * [1]

    def save_vocabulary(self, save_directory: str, filename_prefix: Optional[str] = None) -> Tuple[str]:
        if not os.path.isdir(save_directory):
            logger.error(f"Vocabulary path ({save_directory}) should be a directory")
            return

        src_vocab_file = os.path.join(
            save_directory, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["src_vocab_file"]
        )
        tgt_vocab_file = os.path.join(
            save_directory, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["tgt_vocab_file"]
        )
        merges_file = os.path.join(
            save_directory, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]
        )

        with open(src_vocab_file, "w", encoding="utf-8") as f:
            f.write(json.dumps(self.encoder, indent=2, sort_keys=True, ensure_ascii=False) + "\n")

        with open(tgt_vocab_file, "w", encoding="utf-8") as f:
            tgt_vocab = {v: k for k, v in self.decoder.items()}
            f.write(json.dumps(tgt_vocab, indent=2, sort_keys=True, ensure_ascii=False) + "\n")

        index = 0
        with open(merges_file, "w", encoding="utf-8") as writer:
            for bpe_tokens, token_index in sorted(self.bpe_ranks.items(), key=lambda kv: kv[1]):
                if index != token_index:
                    logger.warning(
                        f"Saving vocabulary to {merges_file}: BPE merge indices are not consecutive."
                        " Please check that the tokenizer is not corrupted!"
                    )
                    index = token_index
                writer.write(" ".join(bpe_tokens) + "\n")
                index += 1

        return src_vocab_file, tgt_vocab_file, merges_file

    def __getstate__(self):
        state = self.__dict__.copy()
        state["sm"] = None
        return state

    def __setstate__(self, d):
        self.__dict__ = d

        try:
            import sacremoses
        except ImportError:
            raise ImportError(
                "You need to install sacremoses to use XLMTokenizer. "
                "See https://pypi.org/project/sacremoses/ for installation."
            )

        self.sm = sacremoses