codingutf8 2021 the facebook inc 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 class for speech2text2 import json import os from typing import dict list optional tuple from tokenizationutils import pretrainedtokenizer from utils import logging logger logging getloggername vocabfilesnames vocabfile vocab json tokenizerconfigfile tokenizerconfig json mergesfile merges txt pretrainedvocabfilesmap vocabfile facebooks2twav2vec2largeende https huggingface cofacebooks2twav2vec2largeenderesolvemainvocab json tokenizerconfigfile facebooks2twav2vec2largeende https huggingface cofacebooks2twav2vec2largeenderesolvemaintokenizerconfig json mergesfile facebooks2twav2vec2largeende https huggingface cofacebooks2twav2vec2largeenderesolvemainmerges txt bpetokenmerges w bpetokenvocab def getpairsword pairs set prevchar word0 for char in word1 pairs addprevchar char prevchar char return pairs speech2text2 has no max input length pretrainedpositionalembeddingssizes facebooks2twav2vec2largeende 1024 class speech2text2tokenizerpretrainedtokenizer vocabfilesnames vocabfilesnames pretrainedvocabfilesmap pretrainedvocabfilesmap maxmodelinputsizes pretrainedpositionalembeddingssizes modelinputnames inputids attentionmask def init self vocabfile bostokens padtokenpad eostokens unktokenunk dolowercasefalse mergesfilenone kwargs self dolowercase dolowercase with openvocabfile encodingutf8 as vocabhandle self encoder json loadvocabhandle self decoder v k for k v in self encoder items if mergesfile is none logger infofno merges files provided self class name can only be used for decoding self bperanks none self cache none else 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 unktokenunktoken bostokenbostoken eostokeneostoken padtokenpadtoken dolowercasedolowercase kwargs property def vocabsizeself int return lenself decoder def getvocabself dict return dictself encoder self addedtokensencoder def bpeself token word tupletoken 1 token1 bpetokenmerges if token in self cache return self cachetoken pairs getpairsword if not pairs return token 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 bpetokenmerges word n bpetokenmerges if word endswithbpetokenmerges word word replacebpetokenmerges word word replace bpetokenvocab self cachetoken word return word def tokenizeself text converts a token str in an index integer using the vocab return self encoder gettoken self encoder getself unktoken def convertidtotokenself index int str converts a list of output tokens into a single string combine tokens make sure tokens are concatenated coding utf 8 2021 the facebook inc 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 class for speech2text2 return set of symbol pairs in a word word is represented as tuple of symbols symbols being variable length strings speech2text2 has no max input length constructs a speech2text2tokenizer this tokenizer inherits from pretrainedtokenizer which contains some of the main methods users should refer to the superclass for more information regarding such methods args vocab_file str file containing the vocabulary bos_token str optional defaults to s the beginning of sentence token eos_token str optional defaults to s the end of sentence token 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 pad_token str optional defaults to pad the token used for padding for example when batching sequences of different lengths kwargs additional keyword arguments passed along to pretrainedtokenizer tokenize a string converts a token str in an index integer using the vocab converts an index integer in a token str using the vocab converts a list of output tokens into a single string combine tokens make sure tokens are concatenated
import json
import os
from typing import Dict, List, Optional, Tuple

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


logger = logging.get_logger(__name__)


VOCAB_FILES_NAMES = {
    "vocab_file": "vocab.json",
    "tokenizer_config_file": "tokenizer_config.json",
    "merges_file": "merges.txt",
}

PRETRAINED_VOCAB_FILES_MAP = {
    "vocab_file": {
        "facebook/s2t-wav2vec2-large-en-de": (
            "https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/vocab.json"
        ),
    },
    "tokenizer_config_file": {
        "facebook/s2t-wav2vec2-large-en-de": (
            "https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/tokenizer_config.json"
        ),
    },
    "merges_file": {
        "facebook/s2t-wav2vec2-large-en-de": (
            "https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/merges.txt"
        ),
    },
}

BPE_TOKEN_MERGES = "</w>"
BPE_TOKEN_VOCAB = "@@ "


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



PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES = {"facebook/s2t-wav2vec2-large-en-de": 1024}


class Speech2Text2Tokenizer(PreTrainedTokenizer):
    

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

    def __init__(
        self,
        vocab_file,
        bos_token="<s>",
        pad_token="<pad>",
        eos_token="</s>",
        unk_token="<unk>",
        do_lower_case=False,
        merges_file=None,
        **kwargs,
    ):
        self.do_lower_case = do_lower_case

        with open(vocab_file, encoding="utf-8") as vocab_handle:
            self.encoder = json.load(vocab_handle)
        self.decoder = {v: k for k, v in self.encoder.items()}

        if merges_file is None:
            logger.info(f"No merges files provided. {self.__class__.__name__} can only be used for decoding.")

            self.bpe_ranks = None
            self.cache = None
        else:
            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__(
            unk_token=unk_token,
            bos_token=bos_token,
            eos_token=eos_token,
            pad_token=pad_token,
            do_lower_case=do_lower_case,
            **kwargs,
        )

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

    def get_vocab(self) -> Dict:
        return dict(self.encoder, **self.added_tokens_encoder)

    def bpe(self, token):
        word = tuple(token[:-1]) + (token[-1] + BPE_TOKEN_MERGES,)
        if token in self.cache:
            return self.cache[token]
        pairs = get_pairs(word)

        if not pairs:
            return token

        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  " + BPE_TOKEN_MERGES:
            word = "\n" + BPE_TOKEN_MERGES

        if word.endswith(BPE_TOKEN_MERGES):
            word = word.replace(BPE_TOKEN_MERGES, "")

        word = word.replace(" ", BPE_TOKEN_VOCAB)
        self.cache[token] = word
        return word

    def _tokenize(self, text):
        

        if self.bpe_ranks is None:
            raise ValueError(
                "This tokenizer was instantiated without a `merges.txt` file, so"
                " that it can only be used for decoding, not for encoding. "
                "Make sure to provide `merges.txt` file at instantiation to enable "
                "encoding."
            )

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

        text = text.split()

        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: str) -> int:
        
        return self.encoder.get(token, self.encoder.get(self.unk_token))

    def _convert_id_to_token(self, index: int) -> str:
        
        result = self.decoder.get(index, self.unk_token)
        return result

    def convert_tokens_to_string(self, tokens: List[str]) -> str:
        
        
        string = " ".join(tokens)

        
        string = "".join(string.split(BPE_TOKEN_VOCAB))

        return string

    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
        vocab_file = os.path.join(
            save_directory, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]
        )
        merges_file = os.path.join(
            save_directory, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]
        )

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

        index = 0
        if self.bpe_ranks is None:
            return (vocab_file,)

        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 (vocab_file, merges_file)