Espeak phonemizer pipeline for spanish (and also english)

tts/.gitignore

@@ -0,0 +1,2 @@
+.ipynb_checkpoints/
+**/*.pyc

tts/text/pipeline/espeak_pipeline.py

@@ -0,0 +1,81 @@
+from typing import List
+from text.pipeline.base_pipeline import BasePipeline
+from text.pipeline_components.input_parsing.parser import ArpabetSequence, GraphemeSequence, ParsedSequence, parse_grapheme_and_arpabet_sequence
+from text.pipeline_components.normalization.espeak_normalization import normalize
+from text.pipeline_components.phonetization.espeak_phonemization import EspeakPhonetization
+from text.pipeline_components.sequencing.espeak_symbols import Symbols
+
+class EspeakPipeline(BasePipeline):
+    """
+    This is a new text pipeline for FakeYou that will be introduced in July 2022.
+    It introduces Arpabet and individually upgradable components.
+    The code is not 1:1 with our notebooks, but we'll go to efforts to make sure that 
+    the model input expectations on both sides match one another.
+    """
+
+    def __init__(self, lang='en'):
+        self.lang = lang
+        symbols = Symbols(lang)
+        self.symbols_to_sequence = symbols.symbols_to_sequence
+        self.arpabet_to_sequence = symbols.arpabet_to_sequence
+        self.sequence_to_text = symbols.sequence_to_text
+        self.SPACE_INTEGER_ENCODING = symbols.SPACE_INTEGER_ENCODING
+
+        self.phonetization = EspeakPhonetization(lang)
+
+    def user_input_to_sequence(self, input_text: str) -> List[int]:
+        """Convert raw user input into a numerical sequence for ML inference."""
+
+        parsed_sequence = parse_grapheme_and_arpabet_sequence(input_text)
+        self._normalize_graphemes(parsed_sequence)
+
+        parsed_sequence = self._convert_phonemes(parsed_sequence)
+        
+        return self._convert_to_integer_sequence(parsed_sequence)
+
+    def sequence_to_text(self, input_sequence: List[int]) -> str:
+        return self.sequence_to_text(input_sequence)
+
+    def _normalize_graphemes(self, parsed_sequence: ParsedSequence):
+        for i in range(len(parsed_sequence)):
+            subsequence = parsed_sequence[i]
+            if isinstance(subsequence, GraphemeSequence):
+                normalized = normalize(subsequence.sequence, self.lang)
+                parsed_sequence[i] = GraphemeSequence(normalized)
+
+
+    def _convert_phonemes(self, parsed_sequence: ParsedSequence) -> ParsedSequence:
+        # NB: We will likely break grapheme runs into multiple arpabet sequences, extending the 
+        # output list length.
+        updated_sequence = []
+
+        for subsequence in parsed_sequence:
+            if isinstance(subsequence, ArpabetSequence):
+                updated_sequence.append(subsequence)
+
+            elif isinstance(subsequence, GraphemeSequence):
+                converted_sequence = self.phonetization.try_convert_to_phonemes(subsequence.sequence)
+                updated_sequence.extend(converted_sequence)
+
+        return updated_sequence
+
+    def _convert_to_integer_sequence(self, parsed_sequence: ParsedSequence) -> List[int]:
+        integer_sequence = []
+        last_index = len(parsed_sequence) - 1
+
+        for i, subsequence in enumerate(parsed_sequence):
+            integer_subsequence = []
+
+            if isinstance(subsequence, ArpabetSequence):
+                integer_subsequence = self.arpabet_to_sequence(subsequence.sequence)
+            elif isinstance(subsequence, GraphemeSequence):
+                integer_subsequence = self.symbols_to_sequence(subsequence.sequence)
+                
+            if integer_subsequence:
+                integer_sequence.extend(integer_subsequence)
+
+                if i < last_index and integer_sequence[-1] != self.SPACE_INTEGER_ENCODING:
+                    # Separate words with single spaces if not already present.
+                    integer_sequence.append(self.SPACE_INTEGER_ENCODING)
+
+        return integer_sequence

tts/text/pipeline_components/normalization/espeak_cleaners.py

@@ -0,0 +1,107 @@
+""" from https://github.com/keithito/tacotron """
+
+import re
+from unidecode import unidecode
+
+from num2words import num2words
+
+# Regular expression matching whitespace:
+_whitespace_re = re.compile(r'\s+')
+
+class Cleaner:
+  def __init__(self, lang='en'):
+    self.lang=lang
+    if lang == 'en':
+      from text.pipeline_components.normalization.lang.en import ZERO_DOLLARS, CENT_SINGULAR, CENT_PLURAL, DOLLAR_SINGULAR, DOLLAR_PLURAL, FLOAT_COMMA,_abbreviations_list, EUROS_PLURAL, _ordinal_re
+    elif lang == 'es':
+      from text.pipeline_components.normalization.lang.es import ZERO_DOLLARS, CENT_SINGULAR, CENT_PLURAL, DOLLAR_SINGULAR, DOLLAR_PLURAL, FLOAT_COMMA,_abbreviations_list, EUROS_PLURAL, _ordinal_re
+    else:
+      raise Exception("Language not supported")
+    
+    # List of (regular expression, replacement) pairs for abbreviations:
+    self._abbreviations = [(re.compile('\\b%s\\.' % x[0], re.IGNORECASE), x[1]) for x in _abbreviations_list]
+    self._comma_number_re = re.compile(r'([0-9][0-9\,]+[0-9])')
+    self._decimal_number_re = re.compile(r'([0-9]+\,[0-9]+)')
+    self._euros_re = re.compile(r'€([0-9\,]*[0-9]+)')
+    self._dollars_re = re.compile(r'\$([0-9\.\,]*[0-9]+)')
+    self._ordinal_re = _ordinal_re
+    self._number_re = re.compile(r'[0-9]+')
+    self.ZERO_DOLLARS = ZERO_DOLLARS
+    self.CENT_SINGULAR = CENT_SINGULAR
+    self.CENT_PLURAL = CENT_PLURAL
+    self.DOLLAR_SINGULAR = DOLLAR_SINGULAR
+    self.DOLLAR_PLURAL = DOLLAR_PLURAL
+    self.FLOAT_COMMA = FLOAT_COMMA
+    self.EUROS_PLURAL = EUROS_PLURAL
+
+  def lowercase(self, text: str) -> str:
+    return text.lower()
+
+  def collapse_whitespace(self, text) -> str:
+    return re.sub(_whitespace_re, ' ', text)
+
+  def convert_to_ascii(self, text: str) -> str:
+    return unidecode(text)
+
+  def basic_cleaners(self, text: str) -> str:
+    '''Basic pipeline that lowercases and collapses whitespace without transliteration.'''
+    text = Cleaner.lowercase(text)
+    text = Cleaner.collapse_whitespace(text)
+    return text
+
+  def transliteration_cleaners(self, text: str) -> str:
+    '''Pipeline for non-English text that transliterates to ASCII.'''
+    text = Cleaner.convert_to_ascii(text)
+    text = Cleaner.lowercase(text)
+    text = Cleaner.collapse_whitespace(text)
+    return text
+
+  def expand_abbreviations(self, text):
+    for regex, replacement in self._abbreviations:
+      text = re.sub(regex, replacement, text)
+    return text
+
+  def _remove_commas(self, m):
+    return m.group(1).replace('.', '')
+
+  def _expand_decimal_point(self, m):
+    return m.group(1).replace(',', f' {self.FLOAT_COMMA} ')
+
+  def _expand_dollars(self, m):
+    match = m.group(1)
+    parts = match.split('.')
+    if len(parts) > 2:
+      return match + f' {self.DOLLAR_PLURAL}'  # Unexpected format
+    dollars = int(parts[0]) if parts[0] else 0
+    cents = int(parts[1]) if len(parts) > 1 and parts[1] else 0
+    if dollars and cents:
+      dollar_unit = self.DOLLAR_SINGULAR if dollars == 1 else self.DOLLAR_PLURAL
+      cent_unit = self.CENT_SINGULAR if cents == 1 else self.CENT_PLURAL
+      return '%s %s, %s %s' % (dollars, dollar_unit, cents, cent_unit)
+    elif dollars:
+      dollar_unit = self.DOLLAR_SINGULAR if dollars == 1 else self.DOLLAR_PLURAL
+      return '%s %s' % (dollars, dollar_unit)
+    elif cents:
+      cent_unit = self.CENT_SINGULAR if cents == 1 else self.CENT_PLURAL
+      return '%s %s' % (cents, cent_unit)
+    else:
+      return self.ZERO_DOLLARS
+
+  def _expand_ordinal(self, m):
+    return num2words(m.group(0), to='ordinal',lang=self.lang)
+
+  def _expand_number(self, m):
+    num = int(m.group(0))
+    return num2words(num,lang=self.lang)
+
+  def normalize_numbers(self, text):
+    text = re.sub(self._comma_number_re, self._remove_commas, text)
+    text = re.sub(self._euros_re, r'\1 EUROS_PLURAL', text).replace("EUROS_PLURAL", self.EUROS_PLURAL)
+    text = re.sub(self._dollars_re, self._expand_dollars, text)
+    text = re.sub(self._decimal_number_re, self._expand_decimal_point, text)
+    text = re.sub(self._ordinal_re, self._expand_ordinal, text)
+    text = re.sub(self._number_re, self._expand_number, text)
+    return text
+
+  def expand_numbers(self, text):
+    return self.normalize_numbers(text)

tts/text/pipeline_components/normalization/espeak_normalization.py

@@ -0,0 +1,8 @@
+from text.pipeline_components.normalization.espeak_cleaners import Cleaner
+
+def normalize(raw_input_text: str, lang='en') -> str:
+    cleaner = Cleaner(lang)
+    text = cleaner.expand_numbers(raw_input_text)
+    text = cleaner.expand_abbreviations(text)
+    text = cleaner.collapse_whitespace(text)
+    return text

tts/text/pipeline_components/normalization/lang/en.py

@@ -0,0 +1,32 @@
+import re
+
+ZERO_DOLLARS = "zero dollars"
+CENT_SINGULAR = "cent"
+CENT_PLURAL = "cents"
+DOLLAR_SINGULAR = "dollar"
+DOLLAR_PLURAL = "dollars"
+FLOAT_COMMA = "point"
+EUROS_PLURAL = "euros"
+
+_ordinal_re = re.compile(r'[0-9]+(st|nd|rd|th)')
+
+_abbreviations_list = [
+  ('mrs', 'misess'),
+  ('mr', 'mister'),
+  ('dr', 'doctor'),
+  ('st', 'saint'),
+  ('co', 'company'),
+  ('jr', 'junior'),
+  ('maj', 'major'),
+  ('gen', 'general'),
+  ('drs', 'doctors'),
+  ('rev', 'reverend'),
+  ('lt', 'lieutenant'),
+  ('hon', 'honorable'),
+  ('sgt', 'sergeant'),
+  ('capt', 'captain'),
+  ('esq', 'esquire'),
+  ('ltd', 'limited'),
+  ('col', 'colonel'),
+  ('ft', 'fort'),
+]

tts/text/pipeline_components/normalization/lang/es.py

@@ -0,0 +1,20 @@
+import re
+
+ZERO_DOLLARS = "cero dólares"
+CENT_SINGULAR = "centavo"
+CENT_PLURAL = "centavos"
+DOLLAR_SINGULAR = "dólar"
+DOLLAR_PLURAL = "dólares"
+FLOAT_COMMA = "coma"
+EUROS_PLURAL = "euros"
+
+_ordinal_re = re.compile(r'[0-9]+(\º|\ª|er|do|os|as)')
+
+_abbreviations_list = [
+  ('srs', 'señores'),
+  ('sr', 'señor'),
+  ('dr', 'doctor'),
+  ('drs', 'doctores'),
+  ('jr', 'yunior'),
+]
+

tts/text/pipeline_components/normalization/spanish_normalization.py

@@ -1,14 +0,0 @@
-from text.pipeline_components.normalization.common_cleaners import collapse_whitespace
-from text.pipeline_components.normalization.common_cleaners import convert_to_ascii
-from text.pipeline_components.normalization.common_cleaners import lowercase
-
-def normalize_spanish(raw_input_text: str) -> str:
-    text = convert_to_ascii(raw_input_text)
-    text = lowercase(text)
-
-    # TODO: Expand abbreviations (Spanish)
-    # TODO: Expand numbers (Spanish)
-
-    text = collapse_whitespace(text)
-
-    return text

tts/text/pipeline_components/phonetization/espeak_phonemization.py

@@ -0,0 +1,127 @@
+import re
+
+from phonemizer.backend import EspeakBackend
+from phonemizer.separator import Separator
+
+from typing import List
+from text.pipeline_components.input_parsing.parser import ArpabetSequence, GraphemeSequence, ParsedSequence
+from text.pipeline_components.phonetization.base_phonetization import BasePhonetization
+from text.pipeline_components.input_parsing.parser import parse_grapheme_and_arpabet_sequence
+# IPA to ARPABET
+
+basic_replacements = {
+  'aɪ': 'AA0 Y',
+  'aʊ': 'AA0 W',
+  'b': 'B',
+  'd': 'D',
+  'dʒ': 'JH',
+  'eɪ': 'EY0',
+  'f': 'F',
+  'h': 'HH',
+  'i': 'IY0',
+  'j': 'Y',
+  'k': 'K',
+  'l': 'L',
+  'l̩': 'EL',
+  'm': 'M',
+  'm̩': 'EM',
+  'n': 'N',
+  'n̩': 'EN',
+  'oʊ': 'OW',
+  'p': 'P',
+  's': 'S',
+  't': 'T',
+  'tʃ': 'CH',
+  'u': 'UW0',
+  'v': 'V',
+  'w': 'W',
+  'z': 'Z',
+  'æ': 'AE0',
+  'ð': 'DH',
+  'ŋ': 'NG',
+  'ɑ': 'AA0',
+  'ɔ': 'AO0',
+  'ɔɪ': 'OY0',
+  'ə': 'AX',
+  'ɚ': 'AXR',
+  'ɛ': 'EH0',
+  'ɝ': 'ER',
+  'ɡ': 'G',
+  'ɨ': 'IX',
+  'ɪ': 'IH0',
+  'ɹ': 'R',
+  'ɾ': 'R',
+  'ɾ̃': 'NX',
+  'ʃ': 'SH',
+  'ʉ': 'UX',
+  'ʊ': 'UH0',
+  'ʌ': 'AH0',
+  'ʍ': 'WH',
+  'ʒ': 'ZH',
+  'ʔ': 'Q',
+  'θ': 'TH'
+}
+
+class EspeakPhonetization(BasePhonetization):
+
+    def __init__(self, lang="en"):
+        super().__init__(None)
+        if lang == 'en':
+            from text.pipeline_components.phonetization.lang.es import valid_symbols, extra_replacements
+        elif lang == 'es':
+            from text.pipeline_components.phonetization.lang.es import valid_symbols, extra_replacements
+        else:
+            raise Exception("Not supported language")
+
+
+        self.replacements = {**basic_replacements, **extra_replacements}
+        self.separator = Separator(phone='|', word=None)
+
+        lang2espeaklang = {"en": "en-us", "es": "es", "fr": "fr-fr"}
+        self.backend = EspeakBackend(lang2espeaklang[lang],with_stress=True, preserve_punctuation=False)
+
+    def grapheme2ipa(self, word):
+        return self.backend.phonemize([word],separator=self.separator)[0].strip().strip("|")
+
+    def ipa2arpabet(self, ipa_word):
+        # print(ipa_word)
+        ipa_word = ipa_word.replace("ɾ|ɾ", "r")
+        if ipa_word.endswith("j|j"):
+            ipa_word = ipa_word.replace("j|j", "i")
+        else:
+            ipa_word = ipa_word.replace("j|j", "ʎ")
+        # Remove secondary accents, they are not important in spanish
+        ipa_word = ipa_word.replace("ˌ", "")
+        # Remove long consonant indication
+        ipa_word = ipa_word.replace("ː", "")
+        arpabet_word = ""
+        for c in ipa_word.split("|"):
+            accent = False
+            if c.startswith("ˈ"):
+                c = c[1:]
+                accent = True
+            if c in self.replacements:
+                replacement = self.replacements[c]
+                if accent:
+                    replacement = replacement.replace("0", "1")
+                arpabet_word += replacement + " "
+            else:
+                print(ipa_word)
+                print(f"'{c}' is not in the replacements dictionary")
+                raise Exception()
+        return "{" + arpabet_word.strip() + "}"
+
+    def word2phonemes(self, word):
+        return self.ipa2arpabet(self.grapheme2ipa(word))
+
+    def __len__(self):
+        return len(self._entries)
+
+    def try_convert_to_phonemes(self, grapheme_sequence: str) -> ParsedSequence:
+        '''Converts word to spanish adapted ARPAbet'''
+
+        # Apply it only on words, keep punctuation unchanged
+        for match in re.findall(r"[A-Za-zÀ-ÿ]+", grapheme_sequence):
+            grapheme_sequence = grapheme_sequence.replace(match, self.word2phonemes(match), 1)
+
+        return parse_grapheme_and_arpabet_sequence(grapheme_sequence)

tts/text/pipeline_components/phonetization/lang/en.py

@@ -0,0 +1,19 @@
+
+_pad        = '_'
+_punctuation = '!\'(),.:;? ' #NB: Includes space
+_special = '-'
+_letters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'
+
+# https://en.wikipedia.org/wiki/ARPABET
+valid_symbols = [
+  'AA', 'AA0', 'AA1', 'AA2', 'AE', 'AE0', 'AE1', 'AE2', 'AH', 'AH0', 'AH1', 'AH2',
+  'AO', 'AO0', 'AO1', 'AO2', 'AW', 'AW0', 'AW1', 'AW2', 'AY', 'AY0', 'AY1', 'AY2',
+  'B', 'CH', 'D', 'DH', 'EH', 'EH0', 'EH1', 'EH2', 'ER', 'ER0', 'ER1', 'ER2', 'EY',
+  'EY0', 'EY1', 'EY2', 'F', 'G', 'HH', 'IH', 'IH0', 'IH1', 'IH2', 'IY', 'IY0', 'IY1',
+  'IY2', 'JH', 'K', 'L', 'M', 'N', 'NG', 'OW', 'OW0', 'OW1', 'OW2', 'OY', 'OY0',
+  'OY1', 'OY2', 'P', 'R', 'S', 'SH', 'T', 'TH', 'UH', 'UH0', 'UH1', 'UH2', 'UW',
+  'UW0', 'UW1', 'UW2', 'V', 'W', 'Y', 'Z', 'ZH'
+]
+
+extra_replacements = {}
+

tts/text/pipeline_components/phonetization/lang/es.py

@@ -0,0 +1,34 @@
+# https://en.wikipedia.org/wiki/ARPABET
+valid_symbols = [
+  'AA', 'AA0', 'AA1', 'AA2', 'AE', 'AE0', 'AE1', 'AE2', 'AH', 'AH0', 'AH1', 'AH2',
+  'AO', 'AO0', 'AO1', 'AO2', 'AW', 'AW0', 'AW1', 'AW2', 'AY', 'AY0', 'AY1', 'AY2',
+  'B', 'CH', 'D', 'DH', 'EH', 'EH0', 'EH1', 'EH2', 'ER', 'ER0', 'ER1', 'ER2', 'EY',
+  'EY0', 'EY1', 'EY2', 'F', 'G', 'HH', 'IH', 'IH0', 'IH1', 'IH2', 'IY', 'IY0', 'IY1',
+  'IY2', 'JH', 'K', 'L', 'M', 'N', 'NG', 'OW', 'OW0', 'OW1', 'OW2', 'OY', 'OY0',
+  'OY1', 'OY2', 'P', 'R', 'S', 'SH', 'T', 'TH', 'UH', 'UH0', 'UH1', 'UH2', 'UW',
+  'UW0', 'UW1', 'UW2', 'V', 'W', 'Y', "RR", "NI"
+]
+
+extra_replacements = {
+    "a": "AA0",
+    "o": "OW0",
+    "β": "V",
+    "e": "EY0",
+    "ɲ": "NI",
+    "r": "RR",
+    "ɣ": "G",
+    "ʎ": "SH",
+    "x": "HH",
+    "ts": "TH", # jacuzzi / pizza
+    "oɪ": "OW0 Y", # acoitar
+    "eʊ": "EY0 W", # adeuda
+    "ʝ": "SH" # ya
+}
+
+# NB: DO NOT CHANGE THE ORDER OF SYMBOLS! 
+# Models are trained on and depend upon the order.
+pad        = '_'
+punctuation = '!\'(),.:;? '
+special = '-'
+# Little hack, since I'm using arpabet letters are not useful anymore, I just replace it with spanish symbols that I need like ¡ and ¿
+letters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwx¡¿'

tts/text/pipeline_components/sequencing/english_symbols_v1.py

@@ -47,4 +47,4 @@ def arpabet_to_sequence(arpabet_string: str) -> List[int]:
 
 
 def _should_keep_symbol(s):
-    return s in _symbol_to_id and s is not '_' and s is not '~'
+    return s in _symbol_to_id and s != '_' and s != '~'

tts/text/pipeline_components/sequencing/espeak_symbols.py

@@ -0,0 +1,60 @@
+"""
+Originally adapted from https://github.com/keithito/tacotron
+
+This defines the set of symbols used in text input to the model.
+
+This is the set of character encodings that FakeYou and UberDuck models have
+been trained on. We freeze it here as "V1", because these may change in future
+notebooks.
+
+Original documentation: 
+    The default is a set of ASCII characters that works well for English or 
+    text that has been run through Unidecode. For other data, you can modify 
+    _characters. See TRAINING_DATA.md for details.
+"""
+
+from typing import List
+
+class Symbols:
+    def __init__(self, lang='en'):
+        if lang == 'en':
+            from text.pipeline_components.phonetization.lang.es import pad, special, punctuation, letters, valid_symbols
+        elif lang == 'es':
+            from text.pipeline_components.phonetization.lang.es import pad, special, punctuation, letters, valid_symbols
+        else:
+            raise Exception("Not supported language")
+
+        # Prepend "@" to ARPAbet symbols to ensure uniqueness (some are the same as uppercase letters):
+        arpabet = ['@' + s for s in valid_symbols]
+
+        # Export all symbols:
+        symbols = [pad] + list(special) + list(punctuation) + list(letters) + arpabet
+
+        # Mappings from symbol to numeric ID and vice versa:
+        # The ordering **CANNOT** change.
+        self._symbol_to_id = {s: i for i, s in enumerate(symbols)}
+        self._id_to_symbol = {i: s for i, s in enumerate(symbols)}
+
+        # Spaces can be injected when joining words.
+        self.SPACE_INTEGER_ENCODING = self._symbol_to_id[' ']
+
+    def _should_keep_symbol(self, s):
+        return s in self._symbol_to_id and s != '_' and s != '~'
+
+    def symbols_to_sequence(self, symbols: str) -> List[int]:
+        return [self._symbol_to_id[s] for s in symbols if self._should_keep_symbol(s)]
+
+    def arpabet_to_sequence(self, arpabet_string: str) -> List[int]:
+        return self.symbols_to_sequence(['@' + s for s in arpabet_string.split()])
+
+    def sequence_to_text(self, sequence):
+        '''Converts a sequence of IDs back to a string'''
+        result = ''
+        for symbol_id in sequence:
+            if symbol_id in self._id_to_symbol:
+                s = self._id_to_symbol[symbol_id]
+                # Enclose ARPAbet back in curly braces:
+                if len(s) > 1 and s[0] == '@':
+                    s = '{%s}' % s[1:]
+                result += s
+        return result.replace('}{', ' ')