lemas_tts/infer/text_norm/tokenizer.py

# cp from https://github.com/lifeiteng/vall-e/blob/main/valle/data/tokenizer.py
# Copyright    2023                            (authors: Feiteng Li)
#
# 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.

import os
import re, logging
from dataclasses import asdict, dataclass
from typing import Any, Dict, List, Optional, Pattern, Union
import math
import numpy as np
import torch
import torchaudio
# from lhotse.features import FeatureExtractor
# from lhotse.utils import Seconds, compute_num_frames
from phonemizer.backend.espeak.wrapper import EspeakWrapper
from phonemizer.backend import EspeakBackend
from phonemizer.backend.espeak.language_switch import LanguageSwitch
from phonemizer.backend.espeak.words_mismatch import WordMismatch
from phonemizer.punctuation import Punctuation
from phonemizer.separator import Separator

# Configure espeak-ng via espeakng_loader if available.
# This provides a consistent libespeak-ng + data across environments (e.g. HF Spaces).
try:
    import espeakng_loader

    EspeakWrapper.set_library(espeakng_loader.get_library_path())
    data_path = espeakng_loader.get_data_path()
    # Export data path via environment so underlying espeak-ng uses it.
    os.environ["ESPEAK_DATA_PATH"] = data_path
    os.environ["ESPEAKNG_DATA_PATH"] = data_path
    print("[LEMAS-TTS] espeak-ng configured via espeakng_loader")
except Exception as e:  # ImportError or runtime errors
    # Fall back to system espeak-ng discovery.
    print(f"[LEMAS-TTS] espeakng_loader not available or failed ({e}); using system espeak-ng")



class TextTokenizer:
    """Phonemize Text."""

    def __init__(
        self,
        language="en-us",
        backend="espeak",
        separator=Separator(word="_", syllable="-", phone="|"),
        preserve_punctuation=True,
        punctuation_marks: Union[str, Pattern] = Punctuation.default_marks(),
        with_stress: bool = False,
        tie: Union[bool, str] = False,
        language_switch: LanguageSwitch = "keep-flags",
        words_mismatch: WordMismatch = "ignore",
    ) -> None:
        phonemizer = EspeakBackend(
            language,
            punctuation_marks=punctuation_marks,
            preserve_punctuation=preserve_punctuation,
            with_stress=with_stress,
            tie=tie,
            language_switch=language_switch,
            words_mismatch=words_mismatch,
        )
        
        self.backend = phonemizer
        self.separator = separator

    def to_list(self, phonemized: str) -> List[str]:
        fields = []
        for word in phonemized.split(self.separator.word):
            # "ɐ    m|iː|n?"    ɹ|ɪ|z|ɜː|v; h|ɪ|z.
            pp = re.findall(r"\w+|[^\w\s]", word, re.UNICODE)
            fields.extend(
                [p for p in pp if p != self.separator.phone]
                + [self.separator.word]
            )
        assert len("".join(fields[:-1])) == len(phonemized) - phonemized.count(
            self.separator.phone
        )
        return fields[:-1]

    def __call__(self, text, strip=True) -> List[List[str]]:
        if isinstance(text, str):
            text = [text]
        phones = []
        for txt in text:
            if txt == '':
                continue
            if txt[0] == '#':
                phones.append(txt)
            else:
                ipa = text_tokenizer.backend.phonemize([txt], separator=text_tokenizer.separator, strip=True, njobs=1, logger=logging.basicConfig(level=logging.ERROR))
                phones += text_tokenizer.to_list(ipa[0])
        return phones


def tokenize_text(tokenizer: TextTokenizer, text: str) -> List[str]:
    phonemes = tokenizer([text.strip()])
    return phonemes[0]  # k2symbols


_PAUSE_SYMBOL = {'、':',', '，':',', '。':',', '！':'!', '？':'?', '：':':'}
def _replace(match):
    word = match.group(0)
    return _PAUSE_SYMBOL[word]

def txt2phone(tokenizer: TextTokenizer, text: str):
    text = re.sub('|'.join(_PAUSE_SYMBOL.keys()), _replace, text)
    text = re.split(r"(#\d)", text)
    phones = []
    for txt in text:
        if txt == '':
            continue
        if txt[0] == '#':
            phones.append(txt)
        else:
            ipa = tokenizer.backend.phonemize([txt], separator=tokenizer.separator, strip=True, njobs=1)
            phones += tokenizer.to_list(ipa[0])
    phones = "|".join(phones).replace("(|", "(").replace("|)", ")")
    # phones = ["(cmn)"] + phones.split("|")
    return phones   


def convert_audio(wav: torch.Tensor, sr: int, target_sr: int, target_channels: int):
    assert wav.shape[0] in [1, 2], "Audio must be mono or stereo."
    if target_channels == 1:
        wav = wav.mean(0, keepdim=True)
    elif target_channels == 2:
        *shape, _, length = wav.shape
        wav = wav.expand(*shape, target_channels, length)
    elif wav.shape[0] == 1:
        wav = wav.expand(target_channels, -1)
    wav = torchaudio.transforms.Resample(sr, target_sr)(wav)
    return wav


class AudioTokenizer:
    """EnCodec audio."""

    def __init__(
        self,
        device: Any = None,
        signature = None
    ) -> None:
        from audiocraft.solvers import CompressionSolver
        model = CompressionSolver.model_from_checkpoint(signature)
        self.sample_rate = model.sample_rate
        self.channels = model.channels
        
        if not device:
            device = torch.device("cpu")
            if torch.cuda.is_available():
                device = torch.device("cuda:0")

        self._device = device

        self.codec = model.to(device)

    @property
    def device(self):
        return self._device

    def encode(self, wav: torch.Tensor) -> torch.Tensor:
        codes = self.codec.encode(wav.to(self.device))
        return [(codes[0], None)]

    def decode(self, frames: torch.Tensor) -> torch.Tensor:
        frames = frames[0][0] # [1,4,T]
        return self.codec.decode(frames)
    


def tokenize_audio(tokenizer: AudioTokenizer, audio, offset = -1, num_frames=-1):
    # Load and pre-process the audio waveform
    if type(audio) == str:
        if offset != -1 and num_frames!=-1:
            wav, sr = torchaudio.load(audio, frame_offset=offset, num_frames=num_frames)
        else:
            wav, sr = torchaudio.load(audio)
        wav = convert_audio(wav, sr, tokenizer.sample_rate, tokenizer.channels)
        wav = wav.unsqueeze(0)
    else:
        wav = audio.unsqueeze(0).unsqueeze(0)
    # Extract discrete codes from EnCodec
    with torch.no_grad():
        encoded_frames = tokenizer.encode(wav)
    return encoded_frames


class AudioSR:
    """EnCodec audio."""

    def __init__(
        self,
        model_path,
        device = "cpu",
    ) -> None:
        import dac
        self.codec = dac.DAC.load(model_path)
        self.codec.to(device)
        self.codec.eval()

        self.sample_rate = self.codec.sample_rate
        self.channels = 1
        self._device = device

    @property
    def device(self):
        return self._device

    def encode(self, wav: torch.Tensor) -> torch.Tensor:
        length = wav.shape[-1]
        right_pad = math.ceil(length / self.codec.hop_length) * self.codec.hop_length - length
        wav = torch.nn.functional.pad(wav, (0, right_pad))
        z, codes, _, _, _ = self.codec.encode(wav.to(self._device))
        return [(codes, z)]

    def decode(self, frames: torch.Tensor) -> torch.Tensor:
        # frames = frames[0][0] # [1,4,T]
        # with torch.no_grad():
        #     z = self.codec.quantizer.from_codes(frames)[0]
        #     y = self.codec.decode(z)
        z = frames[0][1] # [1, 2048, T]
        with torch.no_grad():
            y = self.codec.decode(z)
        return y