Blame for processing_emosphere_pp.py

# Copyright 2026 The EmoSphere++ Authors, The NATSpeech Authors, TICTAP-11 and b-re-w. All rights reserved.
#
# Licensed under the MIT License. You may obtain a copy of the License at
#
#     https://opensource.org/licenses/MIT
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
# documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
# Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
# WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""Processor class for EmoSphere++."""

import json
import math
import os
import re
import unicodedata
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any

import numpy as np
import torch
from transformers.feature_extraction_utils import BatchFeature
from transformers.processing_utils import ProcessorMixin

from .g2p_en import G2pEn, download_g2p_en_resources, reject_numbers, tweet_tokenize
from .loudness import normalize_loudness
from .modeling_emotion2vec import Emotion2vecModel
from .modeling_w2v2_emotion_dim import Wav2Vec2EmotionDimModel
from .modeling_emosphere_pp import EMOSPHERE_PP_VAD_CENTERS, emotion_adaptive_spherical_vector


__all__ = ["EmoSpherePPProcessor"]

PROCESSOR_CONFIG = "processor_config.json"
SAMPLING_RATE = 16000

WAVLM_BASE_SV = ("microsoft/wavlm-base-sv", "0a23162ffc49adcf42bdf836a00cb2eb45af3601")
EMOTION2VEC_PLUS_BASE = ("emotion2vec/emotion2vec_plus_base", "b318240bfe67db81a8c572ecb37ce9c3759b81c9")
AUDEERING_VAD = ("audeering/wav2vec2-large-robust-12-ft-emotion-msp-dim", "6eba34a2485ea31cb03600241787c3a5edab8626")

PUNCS = "!,.?;:"
RESERVED_TOKENS = ("<pad>", "<EOS>", "<UNK>")

SOXR_HQ_CUTOFF = 0.9568718
SOXR_HQ_HALF_WIDTH = 290.5 / 3
SOXR_HQ_BETA = 13.04


# Copied from vad_weaver.utils.reproductions.emosphere_pp.is_sil_phoneme
def is_sil_phoneme(phoneme: str) -> bool:
    """
    Tells whether a token is silence for NATSpeech: empty or not starting with a letter.

    Args:
        phoneme (`str`):
            The token.

    Returns:
        `bool`: Whether it is a silence token.
    """
    return phoneme == "" or not phoneme[0].isalpha()


# Copied from vad_weaver.utils.reproductions.emosphere_pp.preprocess_text
def preprocess_text(text: str) -> str:
    r"""
    Normalizes text as NATSpeech's English `TxtProcessor.preprocess_text`: numbers (refused here), accents stripped,
    lowercase, quotes and brackets removed, hyphens to spaces, every character other than letters, spaces and
    `!,.?;:` removed, repeated punctuation merged and each punctuation mark set off by spaces.

    Args:
        text (`str`):
            The raw transcript.

    Returns:
        `str`: The normalized text, words and punctuation separated by single spaces.
    """
    text = reject_numbers(text)
    text = "".join(char for char in unicodedata.normalize("NFD", text) if unicodedata.category(char) != "Mn")
    text = text.lower()
    text = re.sub("['\"()]+", "", text)
    text = re.sub("[-]+", " ", text)
    text = re.sub(f"[^ a-z{PUNCS}]", "", text)
    text = re.sub(f" ?([{PUNCS}]) ?", r"\1", text)
    text = re.sub(f"([{PUNCS}])+", r"\1", text)
    text = text.replace("i.e.", "that is")
    text = text.replace("etc.", "etc")
    text = re.sub(f"([{PUNCS}])", r" \1 ", text)
    return re.sub(r"\s+", " ", text)


def phonemes(g2p: G2pEn, text: str) -> list[str]:
    r"""
    The phonemes of a sentence as NATSpeech's `en` text processor with `with_phsep` and `add_eos_bos` builds them, as
    [`~vad_weaver.utils.reproductions.emosphere_pp.EmoSpherePPTextFrontend.phonemes`] does.

    Args:
        g2p ([`~vad_weaver.utils.reproductions.g2p_en.G2pEn`]):
            The converter.
        text (`str`):
            The raw transcript.

    Returns:
        `list[str]`: The phonemes, `|` between words and `<BOS>`, `<EOS>` around the sentence.

    Raises:
        ValueError: If the tokenizer splits the normalized text differently from its spaces.
    """
    normalized = preprocess_text(text).strip()
    words: list[tuple[str, list[str]]] = [(word, []) for word in normalized.split(" ")]
    index = 0
    for phoneme in g2p.pronounce(tweet_tokenize(normalized)):
        if phoneme == " ":
            index += 1
        else:
            if index >= len(words):
                raise ValueError(f"tokenization of {normalized!r} does not follow its spaces")
            words[index][1].append(phoneme)
    while words and is_sil_phoneme(words[0][0]):
        words = words[1:]
    while words and is_sil_phoneme(words[-1][0]):
        words = words[:-1]
    result = ["<BOS>"]
    for i, (word, word_phonemes) in enumerate(words):
        result.extend(word_phonemes)
        if i != len(words) - 1 and not is_sil_phoneme(word) and not is_sil_phoneme(words[i + 1][0]):
            result.append("|")
    return result + ["<EOS>"]


# Copied from vad_weaver.metrics.aro_val_sim.soxr_hq_kernel
def soxr_hq_kernel(source_rate: int, sampling_rate: int) -> tuple[torch.Tensor, int]:
    r"""
    Builds the polyphase kernel of [`resample_like_soxr_hq`]. Output sample $m$ lies at time $m / r_{out}$ and input
    sample $n$ at $n / r_{in}$; with $\tau = m / r_{out} - n / r_{in}$, $f_c$ = [`SOXR_HQ_CUTOFF`] $\cdot
    \min(r_{in}, r_{out}) / 2$ and $T$ = [`SOXR_HQ_HALF_WIDTH`] $/ \min(r_{in}, r_{out})$, the weight of input $n$ is

    $$k(\tau) = \frac{2 f_c}{r_{in}} \operatorname{sinc}(2 f_c \tau) \frac{I_0\left(\beta \sqrt{1 - (\tau / T)^2}\right)}{I_0(\beta)}, \quad \vert\tau\vert \le T,$$

    and 0 beyond $T$. The three constants were fitted to the impulse response of `soxr` 1.1.0 (`soxr_hq`) measured
    from 48 and 24 kHz to 16 kHz, which is one 581 tap linear phase filter on the 48 kHz grid for both rates; the fit
    reproduces it to $2 \cdot 10^{-8}$ per tap, the float32 rounding of soxr's output.

    Args:
        source_rate (`int`):
            Input rate.
        sampling_rate (`int`):
            Output rate.

    Returns:
        `tuple[torch.Tensor, int]`: The kernel of shape `(new, 1, 2 * width + orig)`, float64, where `orig` and `new`
        are the two rates divided by their greatest common divisor and one row holds the input weights of one output
        phase, and `width`, the number of input samples the window reaches on each side.
    """
    divisor = math.gcd(source_rate, sampling_rate)
    orig, new = source_rate // divisor, sampling_rate // divisor
    lower = min(source_rate, sampling_rate)
    cutoff = SOXR_HQ_CUTOFF * lower / 2
    half_width = SOXR_HQ_HALF_WIDTH / lower
    width = math.ceil(half_width * source_rate)
    offsets = torch.arange(-width, width + orig, dtype=torch.float64)
    phases = torch.arange(new, dtype=torch.float64)[:, None]
    tau = (phases * orig / new - offsets[None, :]) / source_rate
    inside = tau.abs() <= half_width
    radius = torch.sqrt(torch.clamp(1 - (tau / half_width) ** 2, min=0))
    beta = torch.tensor(SOXR_HQ_BETA, dtype=torch.float64)
    window = torch.where(inside, torch.special.i0(beta * radius) / torch.special.i0(beta), torch.zeros_like(tau))
    kernel = 2 * cutoff / source_rate * torch.sinc(2 * cutoff * tau) * window
    return kernel[:, None, :], width


# Copied from vad_weaver.metrics.aro_val_sim.resample_like_soxr_hq
def resample_like_soxr_hq(waveform: np.ndarray, source_rate: int, sampling_rate: int = 16000) -> np.ndarray:
    r"""
    Resamples a waveform as `librosa.resample(..., res_type="soxr_hq")` does, which `librosa.load` uses: the linear
    phase Kaiser windowed sinc of [`soxr_hq_kernel`], with output sample $m$ centered on input time
    $m \cdot r_{in} / r_{out}$ and the signal taken as zero outside the clip. soxr returns
    $\lfloor n \cdot r_{out} / r_{in} + 1/2 \rfloor$ samples and librosa pads them with zeros to
    $\lceil n \cdot r_{out} / r_{in} \rceil$. From 48 and 24 kHz to 16 kHz the output matches soxr to float32
    rounding (130 dB SNR or more on speech); other rates match to 85 dB or more, since soxr builds their filter in
    several stages.

    Args:
        waveform (`np.ndarray` of shape `(samples,)`):
            Mono waveform.
        source_rate (`int`):
            Rate of `waveform`.
        sampling_rate (`int`, *optional*, defaults to 16000):
            Rate to resample to.

    Returns:
        `np.ndarray`, dtype `float32`: The resampled waveform, $\lceil n \cdot r_{out} / r_{in} \rceil$ samples long.
    """
    if source_rate == sampling_rate:
        return waveform.astype(np.float32)
    kernel, width = soxr_hq_kernel(source_rate, sampling_rate)
    new, _, taps = kernel.shape
    orig = taps - 2 * width
    samples = len(waveform)
    length = -(-samples * new // orig)
    kept = (2 * samples * new + orig) // (2 * orig)
    frames = -(-length // new)
    padded = torch.zeros(frames * orig + 2 * width + orig, dtype=torch.float64)
    padded[width : width + samples] = torch.from_numpy(np.asarray(waveform, dtype=np.float64))
    resampled = torch.nn.functional.conv1d(padded[None, None], kernel, stride=orig)[0, :, :frames]
    resampled = resampled.transpose(0, 1).reshape(-1)[:length]
    resampled[kept:] = 0
    return resampled.float().numpy()


def _read_mono(source: str | os.PathLike | tuple[np.ndarray, int]) -> tuple[np.ndarray, int]:
    if isinstance(source, tuple):
        waveform, rate = source
        waveform = np.asarray(waveform.cpu() if isinstance(waveform, torch.Tensor) else waveform, dtype=np.float32)
        return (waveform.mean(-1) if waveform.ndim > 1 else waveform), rate
    import soundfile

    waveform, rate = soundfile.read(str(source), dtype="float32", always_2d=True)
    return waveform.mean(1), rate


def _convert_emotion2vec_plus(snapshot: Path, dtype: torch.dtype = torch.bfloat16) -> Emotion2vecModel:
    from .convert_emotion2vec_original_pytorch_checkpoint_to_pytorch import (
        convert_state_dict,
        load_emotion2vec_config,
    )

    config = load_emotion2vec_config(snapshot / "config.yaml", snapshot / "tokens.txt")
    original = torch.load(snapshot / "model.pt", map_location="cpu", weights_only=True)["model"]
    original = {key.removeprefix("d2v_model."): value for key, value in original.items()}
    converted = convert_state_dict(original, config.num_prenet_layers)
    model = Emotion2vecModel(config).to(dtype=dtype)
    state = {key.removeprefix("emotion2vec."): value for key, value in converted.items() if key.startswith("emotion2vec.")}
    missing, unexpected = model.load_state_dict(state, strict=False)
    if missing or unexpected:
        raise RuntimeError(f"Missing keys {missing}, unexpected keys {unexpected}.")
    return model.eval()


class EmoSpherePPProcessor(ProcessorMixin):
    r"""
    Constructs the EmoSphere++ processor, which builds every input of
    [`~vad_weaver.models.emosphere_pp.EmoSpherePPForConditionalGeneration`] from a text and a reference recording, as
    the authors' offline pipeline does (`embedding_extract.py`, `data_gen/tts/base_binarizer.py`):

    - `input_ids`: the NATSpeech English front end with g2p_en, mapped through the phoneme table and interspersed with
      the blank id (the vocabulary size).
    - `speaker_embeddings`: the `microsoft/wavlm-base-sv` x-vector of the reference, padded to at least one second.
    - `emotion_embeddings`: the time average of the emotion2vec+ base frame features of the reference.
    - `emotion_intensity`, `emotion_style`: the emotion-adaptive spherical vector of the audEERING arousal, dominance
      and valence of the reference, measured against the center and the quartiles of the emotion; neutral is
      $(0, 0, 0)$.

    The encoders are downloaded from their own repositories at pinned commits on first use.

    Args:
        phone_set (`Sequence[str]`):
            The phoneme table (`phone_set.json` of the training data).
        vad_quartiles (`Mapping[str, Sequence[float]]`):
            First and third quartile of the distance to the emotion center, per emotion.
        vad_centers (`Mapping[str, Sequence[float]]`, *optional*):
            Emotion centers in arousal, dominance, valence order. Defaults to [`EMOSPHERE_PP_VAD_CENTERS`].
        g2p_dir (`str` or `os.PathLike`, *optional*):
            Folder with the g2p_en resources. Downloaded from their pinned sources when left out.
        device (`str`, *optional*):
            Device of the encoders. Defaults to CUDA where it is available.
        dtype (`torch.dtype` or `str`, *optional*):
            Floating-point dtype of the encoders and model inputs. Defaults to float32.
    """

    attributes: list[str] = []

    def __init__(
        self,
        phone_set: Sequence[str],
        vad_quartiles: Mapping[str, Sequence[float]],
        vad_centers: Mapping[str, Sequence[float]] | None = None,
        g2p_dir: str | os.PathLike | None = None,
        device: str | None = None,
        dtype: torch.dtype | str | None = None,
        **kwargs,
    ):
        self.phone_set = list(phone_set)
        self.vad_quartiles = {key: [float(v) for v in value] for key, value in vad_quartiles.items()}
        self.vad_centers = {k: [float(v) for v in c] for k, c in (vad_centers or EMOSPHERE_PP_VAD_CENTERS).items()}
        tokens = list(RESERVED_TOKENS) + [token for token in self.phone_set if token not in RESERVED_TOKENS]
        self.token_to_id = {token: index for index, token in enumerate(tokens)}
        self.blank_id = len(tokens)
        self.g2p_dir = g2p_dir
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.dtype = self._resolve_dtype(dtype)
        self._g2p = None
        self._encoders: dict[str, Any] = {}

    @staticmethod
    def _resolve_dtype(dtype: torch.dtype | str | None) -> torch.dtype:
        if dtype is None:
            return torch.float32
        if isinstance(dtype, str):
            dtype = {"float32": torch.float32, "bfloat16": torch.bfloat16}.get(dtype)
        if dtype not in (torch.float32, torch.bfloat16):
            raise ValueError("encoder dtype must be float32 or bfloat16")
        return dtype

    @classmethod
    def get_attributes(cls) -> list[str]:
        return []

    @classmethod
    def from_pretrained(cls, pretrained_model_name_or_path: str | os.PathLike, **kwargs) -> "EmoSpherePPProcessor":
        r"""
        Reads `processor_config.json` from a local folder or a Hub repository.

        Args:
            pretrained_model_name_or_path (`str` or `os.PathLike`):
                Folder or repository id.
            kwargs:
                `revision`, `cache_dir` and `token` go to the download; the rest overrides the saved arguments.

        Returns:
            [`EmoSpherePPProcessor`]: The processor.
        """
        from transformers.utils.hub import cached_file

        download = {key: kwargs.pop(key) for key in ("revision", "cache_dir", "token") if key in kwargs}
        for key in ("trust_remote_code", "_from_auto", "_commit_hash", "local_files_only", "subfolder"):
            kwargs.pop(key, None)
        path = cached_file(str(pretrained_model_name_or_path), PROCESSOR_CONFIG, **download)
        saved = json.loads(Path(path).read_text())
        for key in ("processor_class", "auto_map"):
            saved.pop(key, None)
        return cls(**{**saved, **kwargs})

    def save_pretrained(self, save_directory: str | os.PathLike, **kwargs) -> list[str]:
        r"""
        Writes `processor_config.json`.

        Args:
            save_directory (`str` or `os.PathLike`):
                Folder.
            kwargs:
                Unused.

        Returns:
            `list[str]`: The written file.
        """
        save_directory = Path(save_directory)
        save_directory.mkdir(parents=True, exist_ok=True)
        content = {
            "processor_class": type(self).__name__,
            "phone_set": self.phone_set,
            "vad_quartiles": self.vad_quartiles,
            "vad_centers": self.vad_centers,
        }
        content["dtype"] = str(self.dtype).removeprefix("torch.")
        if getattr(self, "_auto_map", None):
            content["auto_map"] = self._auto_map
        path = save_directory / PROCESSOR_CONFIG
        path.write_text(json.dumps(content, indent=2) + "\n")
        return [str(path)]

    def to(self, device: str, dtype: torch.dtype | str | None = None) -> "EmoSpherePPProcessor":
        """Moves loaded encoders to `device` and `dtype` and loads the next ones there."""
        self.device = device
        if dtype is not None:
            self.dtype = self._resolve_dtype(dtype)
        for model in self._encoders.values():
            if isinstance(model, torch.nn.Module):
                model.to(device, dtype=self.dtype)
        return self

    @property
    def g2p(self) -> G2pEn:
        """The g2p_en converter, built on first use."""
        if self._g2p is None:
            self._g2p = G2pEn(self.g2p_dir if self.g2p_dir is not None else download_g2p_en_resources())
        return self._g2p

    def _encoder(self, name: str) -> Any:
        if name not in self._encoders:
            from transformers import Wav2Vec2FeatureExtractor, WavLMForXVector

            if name == "speaker":
                model = WavLMForXVector.from_pretrained(
                    WAVLM_BASE_SV[0], revision=WAVLM_BASE_SV[1], dtype=self.dtype
                )
                extractor = Wav2Vec2FeatureExtractor.from_pretrained(WAVLM_BASE_SV[0], revision=WAVLM_BASE_SV[1])
                self._encoders["speaker_extractor"] = extractor
            elif name == "emotion":
                from huggingface_hub import snapshot_download

                snapshot = snapshot_download(
                    EMOTION2VEC_PLUS_BASE[0],
                    revision=EMOTION2VEC_PLUS_BASE[1],
                    allow_patterns=["config.yaml", "tokens.txt", "model.pt"],
                )
                model = _convert_emotion2vec_plus(Path(snapshot), dtype=self.dtype)
            else:
                model = Wav2Vec2EmotionDimModel.from_pretrained(
                    AUDEERING_VAD[0], revision=AUDEERING_VAD[1], dtype=self.dtype
                )
                extractor = Wav2Vec2FeatureExtractor.from_pretrained(AUDEERING_VAD[0], revision=AUDEERING_VAD[1])
                self._encoders["vad_extractor"] = extractor
            self._encoders[name] = model.to(self.device, dtype=self.dtype).eval()
        return self._encoders[name]

    def input_ids(self, text: str) -> list[int]:
        """
        The phoneme ids of a sentence with the blank id interspersed.

        Args:
            text (`str`):
                The raw transcript. Digits are refused, as the training front end has no number expansion.

        Returns:
            `list[int]`: The ids.
        """
        unknown = self.token_to_id["<UNK>"]
        ids = [self.token_to_id.get(token, unknown) for token in phonemes(self.g2p, text)]
        result = [self.blank_id] * (len(ids) * 2 + 1)
        result[1::2] = ids
        return result

    @torch.no_grad()
    def speaker_embedding(self, audio: str | os.PathLike | tuple[np.ndarray, int]) -> torch.Tensor:
        """
        The WavLM x-vector of a recording: read mono, resampled to 16 kHz as `librosa.load` does, padded with zeros
        to at least one second.

        Args:
            audio (`str`, `os.PathLike` or `tuple[np.ndarray, int]`):
                A file, or a waveform with its sampling rate.

        Returns:
            `torch.Tensor` of shape `(512,)`: The x-vector, float32 on the CPU.
        """
        model = self._encoder("speaker")
        waveform, rate = _read_mono(audio)
        waveform = resample_like_soxr_hq(waveform, rate, SAMPLING_RATE)
        if len(waveform) < SAMPLING_RATE:
            waveform = np.pad(waveform, (0, SAMPLING_RATE - len(waveform)))
        values = self._encoders["speaker_extractor"](waveform, return_tensors="pt", sampling_rate=SAMPLING_RATE)
        return model(values.input_values.to(self.device, dtype=self.dtype)).embeddings[0].float().cpu()

    @torch.no_grad()
    def emotion_embedding(self, audio: str | os.PathLike | tuple[np.ndarray, int]) -> torch.Tensor:
        """
        The time average of the emotion2vec+ base frame features of a recording, read as FunASR reads it (mono,
        `torchaudio` resampling to 16 kHz).

        Args:
            audio (`str`, `os.PathLike` or `tuple[np.ndarray, int]`):
                A file, or a waveform with its sampling rate.

        Returns:
            `torch.Tensor` of shape `(768,)`: The embedding, float32 on the CPU.
        """
        import torchaudio.functional as audio_functional

        model = self._encoder("emotion")
        waveform, rate = _read_mono(audio)
        waveform = torch.from_numpy(np.ascontiguousarray(waveform))
        if rate != SAMPLING_RATE:
            waveform = audio_functional.resample(waveform, rate, SAMPLING_RATE)
        frames = model(waveform[None].to(self.device, dtype=self.dtype)).last_hidden_state[0]
        # numpy's float32 mean, which the authors' extraction takes over the FunASR features
        return torch.from_numpy(frames.float().cpu().numpy().mean(0))

    @torch.no_grad()
    def vad(self, audio: str | os.PathLike | tuple[np.ndarray, int]) -> torch.Tensor:
        """
        The audEERING utterance arousal, dominance and valence of a recording, read as `librosa.load` reads it.

        Args:
            audio (`str`, `os.PathLike` or `tuple[np.ndarray, int]`):
                A file, or a waveform with its sampling rate.

        Returns:
            `torch.Tensor` of shape `(3,)`: Arousal, dominance and valence, float32 on the CPU.
        """
        model = self._encoder("vad")
        waveform, rate = _read_mono(audio)
        waveform = resample_like_soxr_hq(waveform, rate, SAMPLING_RATE)
        values = self._encoders["vad_extractor"](waveform, sampling_rate=SAMPLING_RATE, return_tensors="pt")
        return model(values.input_values.to(self.device, dtype=self.dtype)).logits[0].float().cpu()

    def spherical_vector(self, vad: torch.Tensor, emotion: str) -> list[float]:
        r"""
        The emotion-adaptive spherical vector $(r, \vartheta, \phi)$ of a VAD reading for an emotion.

        Args:
            vad (`torch.Tensor` of shape `(3,)`):
                Arousal, dominance and valence.
            emotion (`str`):
                `neutral`, `angry`, `happy`, `sad` or `surprise`, in any case.

        Returns:
            `list[float]`: The three values; `[0, 0, 0]` for neutral.

        Raises:
            `KeyError`: If the emotion is unknown.
        """
        emotion = emotion.lower()
        if emotion not in self.vad_centers:
            raise KeyError(f"unknown emotion {emotion!r}; use one of {sorted(self.vad_centers)}")
        if emotion == "neutral":
            return [0.0, 0.0, 0.0]
        center = torch.tensor(self.vad_centers[emotion])
        first, third = self.vad_quartiles[emotion]
        return emotion_adaptive_spherical_vector(vad[None].float(), center, first, third)[0].tolist()

    def __call__(
        self,
        text: str | Sequence[str],
        reference_audio: Any,
        emotion: str | Sequence[str] = "neutral",
        emotion_intensity: float | Sequence[float] | None = None,
        emotion_style: Sequence[float] | Sequence[Sequence[float]] | None = None,
        return_tensors: str | None = "pt",
        **kwargs,
    ) -> BatchFeature:
        r"""
        Builds the model inputs of one or several sentences.

        Args:
            text (`str` or `Sequence[str]`):
                Sentences to synthesize.
            reference_audio (`str`, `os.PathLike`, `tuple[np.ndarray, int]`, or a sequence of them):
                Reference recording per sentence, giving the voice, the emotion embedding and the VAD reading.
            emotion (`str` or `Sequence[str]`, *optional*, defaults to `"neutral"`):
                Emotion per sentence.
            emotion_intensity (`float` or `Sequence[float]`, *optional*):
                Radius $r \in [0, 1]$ per sentence, overriding the one read from the reference.
            emotion_style (`Sequence[float]` or `Sequence[Sequence[float]]`, *optional*):
                Elevation and azimuth in radians per sentence, overriding the ones read from the reference.
            return_tensors (`str`, *optional*, defaults to `"pt"`):
                Only `"pt"` is supported.
            kwargs:
                Unused.

        Returns:
            [`~transformers.BatchFeature`]: `input_ids` (padded with 0), `input_lengths`, `speaker_embeddings`,
            `emotion_embeddings`, `emotion_intensity` and `emotion_style`.
        """
        single = isinstance(text, str)
        texts = [text] if single else list(text)
        references = [reference_audio] if single or isinstance(reference_audio, (str, os.PathLike, tuple)) else list(reference_audio)
        references = references * len(texts) if len(references) == 1 else references
        emotions = [emotion] * len(texts) if isinstance(emotion, str) else list(emotion)
        intensities = [emotion_intensity] * len(texts) if not isinstance(emotion_intensity, Sequence) else list(emotion_intensity)
        if emotion_style is None or not isinstance(emotion_style[0], Sequence):
            styles = [emotion_style] * len(texts)
        else:
            styles = list(emotion_style)
        ids, spheres, speakers, emotion_embeddings = [], [], [], []
        for sentence, reference, name, intensity, style in zip(texts, references, emotions, intensities, styles):
            ids.append(self.input_ids(sentence))
            speakers.append(self.speaker_embedding(reference))
            emotion_embeddings.append(self.emotion_embedding(reference))
            sphere = self.spherical_vector(self.vad(reference), name) if name.lower() != "neutral" else [0.0] * 3
            if intensity is not None:
                sphere[0] = float(intensity)
            if style is not None:
                sphere[1:] = [float(v) for v in style]
            spheres.append(sphere)
        lengths = torch.tensor([len(row) for row in ids])
        input_ids = torch.zeros(len(ids), int(lengths.max()), dtype=torch.long)
        for row, sequence in enumerate(ids):
            input_ids[row, : len(sequence)] = torch.tensor(sequence)
        sphere_tensor = torch.tensor(spheres, dtype=self.dtype)
        return BatchFeature(
            {
                "input_ids": input_ids,
                "input_lengths": lengths,
                "speaker_embeddings": torch.stack(speakers).to(self.dtype),
                "emotion_embeddings": torch.stack(emotion_embeddings).to(self.dtype),
                "emotion_intensity": sphere_tensor[:, :1],
                "emotion_style": sphere_tensor[:, 1:],
            }
        )

    @staticmethod
    def postprocess(audio: np.ndarray, sampling_rate: int = SAMPLING_RATE, target_loudness: float = -20.0) -> np.ndarray:
        r"""
        Scales a vocoder waveform as EmoSphere++'s `BigVGAN.spec2wav` does: $\times 32768$ cast to int16, divided by
        its largest sample times 0.95, then BS.1770 loudness normalized to -20 LUFS.

        Args:
            audio (`np.ndarray`):
                The vocoder output in $[-1, 1]$.
            sampling_rate (`int`, *optional*, defaults to 16000):
                Its rate.
            target_loudness (`float`, *optional*, defaults to -20.0):
                Target loudness in LUFS.

        Returns:
            `np.ndarray`: The float64 waveform.
        """
        pcm = (np.asarray(audio, dtype=np.float32) * 32768.0).astype("int16")
        scaled = 0.95 * (pcm / pcm.max())
        return normalize_loudness(scaled, sampling_rate, target_loudness)