diff --git a/docs/source/tts/magpietts-longform.rst b/docs/source/tts/magpietts-longform.rst index 33aef42a5abe..8bcf721bc299 100644 --- a/docs/source/tts/magpietts-longform.rst +++ b/docs/source/tts/magpietts-longform.rst @@ -211,14 +211,15 @@ Configuration Dataclasses ######################### -``ChunkedInferenceConfig`` --------------------------- +``ModelInferenceParameters`` +---------------------------- -Immutable tuning parameters (set in model): +Model inference parameters, including long-form and chunked-inference tuning values. These can be set in the model's +``inference_parameters`` configuration: .. literalinclude:: ../../../nemo/collections/tts/models/magpietts.py :language: python - :pyobject: ChunkedInferenceConfig + :pyobject: ModelInferenceParameters ``ChunkState`` @@ -255,4 +256,3 @@ See Also - :doc:`magpietts`: Main Magpie-TTS documentation - :doc:`magpietts-po`: Preference Optimization Guide - diff --git a/nemo/collections/tts/models/magpietts.py b/nemo/collections/tts/models/magpietts.py index ad472c45161e..79018231c6bf 100644 --- a/nemo/collections/tts/models/magpietts.py +++ b/nemo/collections/tts/models/magpietts.py @@ -181,41 +181,6 @@ class ChunkedDecoderState: attn_prior: Optional[torch.Tensor] = None -@dataclass -class ChunkedInferenceConfig: - """Immutable configuration for chunked inference tuning parameters. - - These parameters control the behavior of chunked (single- or multi-chunk) speech generation. - Initialized once in MagpieTTSModel.__init__ and accessed via self.chunked_inference_config. - - Attributes: - history_len_heuristic: Maximum history tokens to retain across chunks. - prior_weights_init: Attention prior weights for chunk initialization. - prior_weights: Attention prior weights during generation (history, current, +1, +2, +3, +4). - finished_limit_with_eot: Steps after text end before allowing EOS (multi-chunk). - finished_limit_without_eot: Steps after chunk end before allowing EOS (multi-chunk). - finished_limit_first_chunk: Steps near text end before forcing EOS for first/single chunk. - Matches the threshold used in infer_batch() for consistent single-chunk behavior. - forceful_chunk_end_threshold: Threshold for forceful chunk termination. - argmax_temperature: Temperature for argmax sampling in EOS detection. - short_sentence_threshold: Sentences shorter than this skip attention prior. - attention_sink_threshold: Times attended before position is considered a sink. - near_end_threshold: Positions from text end to consider "near end". - """ - - history_len_heuristic: int = 20 - prior_weights_init: Tuple[float, ...] = (0.5, 1.0, 0.8, 0.2, 0.2) - prior_weights: Tuple[float, ...] = (0.2, 1.0, 0.6, 0.4, 0.2, 0.2) - finished_limit_with_eot: int = 5 - finished_limit_without_eot: int = 1 - finished_limit_first_chunk: int = 20 - forceful_chunk_end_threshold: int = 3 - argmax_temperature: float = 0.01 - short_sentence_threshold: int = 35 - attention_sink_threshold: int = 10 - near_end_threshold: int = 3 - - @dataclass class ChunkState: """Mutable state persisting across chunks during chunked generation. @@ -276,6 +241,18 @@ class ModelInferenceParameters: eos_detection_method (str): EOS detection method. See the EOSDetectionMethod class. min_generated_frames (int): Setting this greater than 0 prevents rare cases of first-frame termination. Any number greater between 1 and 4 should work, but 4 lines up with the codec's minimum frame requirement. + attention_sink_threshold (int): Times a position may be attended before standard inference advances past it. + history_len_heuristic (int): Maximum history tokens retained across text chunks. + prior_weights_init (Tuple[float, ...]): Attention prior weights used when initializing a new chunk. + prior_weights (Tuple[float, ...]): Attention prior weights used during chunked generation. + finished_limit_with_eot (int): Near-end steps before allowing EOS in the final chunk. + finished_limit_without_eot (int): Near-end steps before allowing EOS in a non-final chunk. + finished_limit_first_chunk (int): Near-end steps before allowing EOS in the first chunk. + forceful_chunk_end_threshold (int): Near-end steps before forcibly ending a non-final chunk. + argmax_temperature (float): Temperature used for the argmax EOS-detection sample. + short_sentence_threshold (int): Texts at or below this length use a uniform chunked attention prior. + chunked_attention_sink_threshold (int): Times a position may be attended before the chunked prior penalizes it. + near_end_threshold (int): Positions from the text end that are treated as near the end. """ max_decoder_steps: int = 500 @@ -292,6 +269,18 @@ class ModelInferenceParameters: ignore_finished_sentence_tracking: bool = True eos_detection_method: str = "argmax_or_multinomial_any" min_generated_frames: int = 4 + attention_sink_threshold: int = 8 + history_len_heuristic: int = 20 + prior_weights_init: Tuple[float, ...] = (0.5, 1.0, 0.8, 0.2, 0.2) + prior_weights: Tuple[float, ...] = (0.2, 1.0, 0.6, 0.4, 0.2, 0.2) + finished_limit_with_eot: int = 5 + finished_limit_without_eot: int = 1 + finished_limit_first_chunk: int = 20 + forceful_chunk_end_threshold: int = 3 + argmax_temperature: float = 0.01 + short_sentence_threshold: int = 35 + chunked_attention_sink_threshold: int = 10 + near_end_threshold: int = 3 @classmethod def from_dict(cls, data: dict) -> 'ModelInferenceParameters': @@ -307,6 +296,9 @@ def from_dict(cls, data: dict) -> 'ModelInferenceParameters': filtered_data['attention_prior_epsilon'] = data['prior_epsilon'] if 'lookahead_window_size' in data: filtered_data['attention_prior_lookahead_window'] = data['lookahead_window_size'] + for field_name in ('prior_weights_init', 'prior_weights'): + if field_name in filtered_data: + filtered_data[field_name] = tuple(filtered_data[field_name]) return cls(**filtered_data) @@ -721,9 +713,6 @@ def __init__(self, cfg: DictConfig, trainer: 'Trainer' = None): # Class-level cache for text normalizers. Used during inference. self._text_normalizers: Dict[str, Any] = {} - # Chunked inference configuration (immutable tuning parameters) - self.chunked_inference_config = ChunkedInferenceConfig() - def _register_tokenizer_artifacts(self, cfg: DictConfig) -> None: """ Register tokenizer file artifacts (phoneme_dict, heteronyms, etc.) for .nemo packaging. @@ -759,7 +748,34 @@ def _register_tokenizer_artifacts(self, cfg: DictConfig) -> None: if hasattr(g2p_cfg, 'get') else getattr(g2p_cfg, 'phoneme_dict', None) ) - if phoneme_dict_path and isinstance(phoneme_dict_path, str) and phoneme_dict_path.strip(): + if phoneme_dict_path and isinstance(phoneme_dict_path, (list, ListConfig)): + # Handle list of phoneme dicts (e.g. Hindi code-switching: hi_prondict + ipa_cmudict) + registered = [] + for i, path_item in enumerate(phoneme_dict_path): + if isinstance(path_item, str) and path_item.strip(): + try: + # Use a list-index path (phoneme_dict.{i}, dot) so the connector's + # OmegaConf.update writes the element back into the list. With an + # underscore (phoneme_dict_{i}) the saved config gets sibling keys + # phoneme_dict_0/_1 (and phoneme_dict null), which IpaG2p rejects on + # restore ("unexpected keyword argument 'phoneme_dict_0'"). + artifact_path = self.register_artifact( + f'text_tokenizers.{tokenizer_name}.g2p.phoneme_dict.{i}', + path_item, + verify_src_exists=True, + ) + registered.append(artifact_path if artifact_path else path_item) + except FileNotFoundError: + logging.warning( + f"phoneme_dict[{i}] file not found for tokenizer '{tokenizer_name}': " + f"{path_item}. Artifact will not be packaged in .nemo file." + ) + registered.append(path_item) + else: + registered.append(path_item) + with open_dict(cfg): + cfg.text_tokenizers[tokenizer_name].g2p.phoneme_dict = registered + elif phoneme_dict_path and isinstance(phoneme_dict_path, str) and phoneme_dict_path.strip(): try: # register_artifact handles both: # - Local paths: registers for .nemo packaging, returns absolute path @@ -2641,7 +2657,7 @@ def get_most_attended_text_timestep( lookahead_window_size, attended_timestep_counter, batch_size, - left_offset=[], + left_offset=None, ): """ Returns the most attended timestep for each batch item @@ -2658,7 +2674,7 @@ def get_most_attended_text_timestep( text_lens (torch.Tensor): Length of text sequence for each batch item. Shape: (batch_size,). lookahead_window_size (int): Size of the forward-looking window to search for the next attended timestep. Determines how far ahead from the last attended timestep to look. - attended_timestep_counter (list): List of dictionaries (one per batch item) tracking how many + attended_timestep_counter (Optional[list]): List of dictionaries (one per batch item) tracking how many times each timestep has been attended. Used to detect attention sinks. batch_size (int): Number of items in the batch. left_offset (list, optional): List of offsets to adjust timestep indices for each batch item, @@ -2672,14 +2688,18 @@ def get_most_attended_text_timestep( - attended_timestep_counter (list): Updated counter tracking attendance frequency for each timestep across all batch items. """ - if len(left_offset) == 0: - left_offset = [0 for _ in range(batch_size)] + if left_offset is None: + left_offset = [0] * batch_size text_time_step_attended = [] for bidx in range(batch_size): last_attended_timestep = last_attended_timesteps[-1][bidx] - if attended_timestep_counter[bidx].get(last_attended_timestep, 0) >= 8: + if ( + attended_timestep_counter[bidx].get(last_attended_timestep, 0) + >= self.inference_parameters.attention_sink_threshold + ): # This is probably an attention sink! Move to the next timestep last_attended_timestep += 1 + last_attended_timestep = max(last_attended_timestep, left_offset[bidx]) last_attended_timestep_in_this_window = last_attended_timestep - left_offset[bidx] window_size = lookahead_window_size window_end = min( @@ -2729,10 +2749,12 @@ def construct_inference_prior( for ind in range(1, lookahead_window_size + 1): _attn_prior[bidx, 0, min(text_time_step_attended[bidx] + ind, _text_len - 1)] = 1.0 - # Penalize timesteps that have been attended to more than 10 times + # Penalize positions that have become attention sinks. for _timestep in attended_timestep_counter[bidx]: - if attended_timestep_counter[bidx][_timestep] >= 10: - # This means the timestep has been attended to more than 10 times (To avoid getting stuck) + if ( + attended_timestep_counter[bidx][_timestep] + >= self.inference_parameters.attention_sink_threshold + ): _attn_prior[bidx, 0, : _timestep + 1] = prior_epsilon unfinished_texts[bidx] = False @@ -3775,7 +3797,10 @@ def do_tts( # Determine tokenizer name based on language using centralized mapping available_tokenizers = list(self.tokenizer.tokenizers.keys()) - tokenizer_name = get_tokenizer_for_language(language, available_tokenizers) + available_mapping = self.cfg.get("language_to_tokenizer_mapping", None) + tokenizer_name = get_tokenizer_for_language( + language, available_tokenizers, language_tokenizer_map=available_mapping + ) logging.info(f"Using tokenizer '{tokenizer_name}' for language '{language}'") # Unified inference path: chunk_text_for_inference automatically decides @@ -3880,7 +3905,7 @@ def _set_attention_prior_weights( text_len: Length of text for this batch item. eps_sq: Squared epsilon for strong suppression. """ - prior_weights = self.chunked_inference_config.prior_weights + prior_weights = self.inference_parameters.prior_weights # Suppress history (before attended - 1) history_end = max(1, attended_pos - 1) @@ -3922,7 +3947,7 @@ def _penalize_attention_sinks( left_offset: Chunk offset for this batch item. eps_sq: Squared epsilon for strong suppression. """ - threshold = self.chunked_inference_config.attention_sink_threshold + threshold = self.inference_parameters.chunked_attention_sink_threshold for timestep, count in attended_timestep_counter.items(): if timestep > left_offset and count >= threshold: @@ -3953,7 +3978,7 @@ def _update_text_completion_state( unfinished_texts: Dict to update in-place. finished_texts_counter: Dict to update in-place. """ - is_near_end = attended_pos >= text_len - self.chunked_inference_config.near_end_threshold + is_near_end = attended_pos >= text_len - self.inference_parameters.near_end_threshold # Text is unfinished if not near end AND not already marked finished unfinished_texts[batch_idx] = not is_near_end and not is_finished @@ -4018,7 +4043,7 @@ def construct_multi_chunk_prior( is_finished = bidx in end_indices or bidx in chunk_end_dict # Short sentences: uniform prior (no guidance needed) - if text_len <= self.chunked_inference_config.short_sentence_threshold: + if text_len <= self.inference_parameters.short_sentence_threshold: attn_prior[bidx, 0, :] = 1.0 else: # Set attention weights around attended position @@ -4097,8 +4122,7 @@ def _check_eos_and_update_state( logging.info(f"Chunk end detected for item {item_idx} at local timestep {current_step}") elif ( not end_of_text[item_idx] - and finished_texts_counter.get(item_idx, -1) - >= self.chunked_inference_config.forceful_chunk_end_threshold + and finished_texts_counter.get(item_idx, -1) >= self.inference_parameters.forceful_chunk_end_threshold ): chunk_end_dict[item_idx] = current_step chunk_end_frame_lens[item_idx] = (current_step + 1) * self.frame_stacking_factor @@ -4265,9 +4289,9 @@ def _initialize_chunked_attn_prior( # Set prior weights for new chunk current_starting_point = batch_text_lens[_idx] - current_chunk_len[_idx] - prior_weights = self.chunked_inference_config.prior_weights_init + prior_weights = self.inference_parameters.prior_weights_init _attn_prior[_idx, :, :current_starting_point] = prior_epsilon * prior_epsilon - for offset, weight in enumerate(prior_weights[:5]): + for offset, weight in enumerate(prior_weights): current_offset_idx = current_starting_point + offset if current_offset_idx < max_text_len: _attn_prior[_idx, :, current_offset_idx] = weight @@ -4305,11 +4329,12 @@ def _update_context_from_history( continue if not beginning_of_text: pad_len_idx = max_text_len - batch_text_lens[_idx] - context_tensors.cond[_idx, : -current_chunk_len[_idx] - pad_len_idx] = ( - chunk_state.history_context_tensor[ - _idx, -(context_tensors.cond[_idx].shape[0] - current_chunk_len[_idx] - pad_len_idx) : - ] + history_context_len = self._to_int( + context_tensors.cond[_idx].shape[0] - current_chunk_len[_idx] - pad_len_idx ) + context_tensors.cond[_idx, :history_context_len] = chunk_state.history_context_tensor[ + _idx, -history_context_len - 1 : -1 + ] chunk_state.history_context_tensor = context_tensors.cond def _prepare_chunked_text_tensors( @@ -4348,9 +4373,10 @@ def _prepare_chunked_text_tensors( # Combine history with current chunk if chunk_state.history_text is not None: + history_text_len = self._to_int(chunk_state.history_text_lens[_idx]) - 1 current_text = torch.cat( [ - chunk_state.history_text[_idx][: chunk_state.history_text_lens[_idx]], + chunk_state.history_text[_idx][:history_text_len], batch["text"][_idx][: current_chunk_len[_idx]], ] ) @@ -4358,7 +4384,7 @@ def _prepare_chunked_text_tensors( current_text = batch["text"][_idx][: current_chunk_len[_idx]] # Apply sliding window - history_len = min(current_chunk_len[_idx], self.chunked_inference_config.history_len_heuristic) + history_len = min(current_chunk_len[_idx], self.inference_parameters.history_len_heuristic) true_window_size = current_chunk_len[_idx] + history_len if not beginning_of_text: current_text = current_text[max(0, current_text.shape[0] - true_window_size) :] @@ -4549,7 +4575,7 @@ def generate_speech( dummy_additional_decoder_input=dummy_additional_decoder_input, dummy_addition_dec_mask=dummy_addition_dec_mask, batch_size=batch_size, - ) + ) # (B, T, num_codebooks * num_tokens_per_codebook), (B, T, d_model) if self.inference_parameters.apply_attention_prior: # Get cross-attention scores (optionally from specific layers for alignment) @@ -4615,9 +4641,9 @@ def generate_speech( for key in state.finished_texts_counter: state.finished_texts_counter[key] += 1 limit = ( - self.chunked_inference_config.finished_limit_with_eot + self.inference_parameters.finished_limit_with_eot if end_of_text[key] - else self.chunked_inference_config.finished_limit_without_eot + else self.inference_parameters.finished_limit_without_eot ) if state.finished_texts_counter[key] > limit: state.unfinished_texts[key] = False @@ -4627,9 +4653,9 @@ def generate_speech( unfinished_items = {} else: finished_threshold = ( - self.chunked_inference_config.finished_limit_first_chunk + self.inference_parameters.finished_limit_first_chunk if beginning_of_text - else self.chunked_inference_config.finished_limit_with_eot + else self.inference_parameters.finished_limit_with_eot ) finished_items = {k: v for k, v in state.finished_texts_counter.items() if v >= finished_threshold} unfinished_items = {k: v for k, v in state.unfinished_texts.items() if v} @@ -4678,15 +4704,15 @@ def generate_speech( unfinished_items=unfinished_items, finished_items=finished_items, forbid_audio_eos=forbid_audio_eos, - ) # (B, num_codebooks) + ) # (B, num_codebooks, frame_stacking_factor) all_codes_next_argmax = self.sample_codes_from_logits( all_code_logits_t, - temperature=self.chunked_inference_config.argmax_temperature, + temperature=self.inference_parameters.argmax_temperature, topk=1, unfinished_items=unfinished_items, finished_items=finished_items, forbid_audio_eos=forbid_audio_eos, - ) # (B, num_codebooks) + ) # (B, num_codebooks, frame_stacking_factor) # Check for EOS and update state self._check_eos_and_update_state( diff --git a/nemo/collections/tts/parts/utils/tts_dataset_utils.py b/nemo/collections/tts/parts/utils/tts_dataset_utils.py index 9a26d15b50bf..dc0cb7aea1fd 100644 --- a/nemo/collections/tts/parts/utils/tts_dataset_utils.py +++ b/nemo/collections/tts/parts/utils/tts_dataset_utils.py @@ -21,7 +21,7 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union import librosa import numpy as np @@ -739,22 +739,30 @@ def get_tokenizer_for_language( language: str, available_tokenizers: List[str], default_tokenizer: str = "english_phoneme", + language_tokenizer_map: Optional[Mapping[str, Union[str, Sequence[str]]]] = None, ) -> str: """Get the appropriate tokenizer name for a language. - Searches LANGUAGE_TOKENIZER_MAP for candidate tokenizers and returns + Searches language_tokenizer_map for candidate tokenizers and returns the first one available. Falls back to default if no match found. Args: language: Language code (e.g., "en", "de", "zh"). available_tokenizers: List of tokenizer names available in the model. default_tokenizer: Fallback tokenizer if no match found. + language_tokenizer_map: Mapping of languages to a tokenizer or ordered tokenizer candidates. + Defaults to ``LANGUAGE_TOKENIZER_MAP``. Returns: Tokenizer name to use. """ - if language in LANGUAGE_TOKENIZER_MAP: - for candidate in LANGUAGE_TOKENIZER_MAP[language]: + if language_tokenizer_map is None: + language_tokenizer_map = LANGUAGE_TOKENIZER_MAP + if language in language_tokenizer_map: + candidates = language_tokenizer_map[language] + if isinstance(candidates, str): + candidates = [candidates] + for candidate in candidates: if candidate in available_tokenizers: return candidate diff --git a/tests/collections/tts/models/test_magpietts_tokenizer_artifacts.py b/tests/collections/tts/models/test_magpietts_tokenizer_artifacts.py new file mode 100644 index 000000000000..24921f564ac7 --- /dev/null +++ b/tests/collections/tts/models/test_magpietts_tokenizer_artifacts.py @@ -0,0 +1,176 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# 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. + +""" +Unit tests for MagpieTTSModel._register_tokenizer_artifacts (.nemo packaging of tokenizer files). + +A code-switching tokenizer (e.g. Hindi = hi_prondict + ipa_cmudict) stores ``g2p.phoneme_dict`` as +a LIST of files. These tests pin the two behaviors that make such a model round-trip through .nemo: + +1. List elements are registered under a DOT-indexed config path (``phoneme_dict.{i}``), so the + save/restore connector's ``OmegaConf.update`` writes each back into the list. An underscore + (``phoneme_dict_{i}``) would instead create sibling keys ``phoneme_dict_0/_1`` (and leave + ``phoneme_dict`` null), which ``IpaG2p`` rejects on restore. +2. A full ``save_to`` -> ``restore_from`` round-trip resolves every list entry to an existing, + packaged file (and produces no ``phoneme_dict_{i}`` sibling keys) -- and the common string case + still round-trips too. + +The tests exercise the real ``MagpieTTSModel._register_tokenizer_artifacts`` without building a full +model (no codec / encoders / GPU): a mock ``self`` for the registration-path test, and a minimal +``ModelPT`` subclass that reuses the method for the save/restore round-trip. +""" + +import os +from unittest.mock import MagicMock + +import pytest +import torch +from omegaconf import ListConfig, OmegaConf + +from nemo.collections.tts.models.magpietts import MagpieTTSModel +from nemo.core.classes import ModelPT + +_IPA_TOKENIZER = "nemo.collections.common.tokenizers.text_to_speech.tts_tokenizers.IPATokenizer" +_IPA_G2P = "nemo.collections.tts.g2p.models.i18n_ipa.IpaG2p" + + +def _g2p_cfg(phoneme_dict): + """Build a minimal text_tokenizers config whose single tokenizer carries `phoneme_dict`.""" + return OmegaConf.create( + { + "text_tokenizers": { + "hindi_phoneme": { + "_target_": _IPA_TOKENIZER, + "g2p": {"_target_": _IPA_G2P, "phoneme_dict": phoneme_dict}, + } + } + } + ) + + +class _TokenizerArtifactModel(ModelPT): + """Minimal ModelPT that reuses MagpieTTSModel's tokenizer-artifact registration, for round-trips. + + It deliberately builds no tokenizers/codec -- it only registers the tokenizer file artifacts, so + save_to/restore_from exercises exactly the artifact packaging/resolution path under test. + """ + + def __init__(self, cfg, trainer=None): + super().__init__(cfg=cfg, trainer=trainer) + self.w = torch.nn.Linear(1, 1) # ensure a non-empty state dict to save + MagpieTTSModel._register_tokenizer_artifacts(self, self.cfg) + # Mirror real usage (the tokenizer reads the dict during __init__) and consume the artifact + # now, while it is on disk. On restore NeMo extracts artifacts to a temp dir that is removed + # once restore_from returns, so reading here is what proves the registered path resolved to a + # real, packaged file. We record the basenames read so tests can assert on a restored model. + self.loaded_phoneme_dicts = self._read_phoneme_dicts(self.cfg) + + @staticmethod + def _read_phoneme_dicts(cfg): + g2p = cfg.text_tokenizers.hindi_phoneme.g2p + phoneme_dict = g2p.phoneme_dict + paths = list(phoneme_dict) if isinstance(phoneme_dict, (list, ListConfig)) else [phoneme_dict] + basenames = [] + for path in paths: + with open(path, "r", encoding="utf-8") as f: + f.read() + basenames.append(os.path.basename(path)) + return basenames + + def setup_training_data(self, train_data_config): + self._train_dl = None + + def setup_validation_data(self, val_data_config): + self._validation_dl = None + + def setup_test_data(self, test_data_config): + self._test_dl = None + + @classmethod + def list_available_models(cls): + return [] + + +class TestRegisterTokenizerArtifacts: + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_list_phoneme_dict_uses_dot_indexed_paths(self): + """A list phoneme_dict registers each element under `phoneme_dict.{i}` (dot) and stays a list.""" + cfg = _g2p_cfg(["/data/hi_prondict.dict", "/data/ipa_cmudict.txt"]) + + mock_model = MagicMock(spec=MagpieTTSModel) + # Mimic register_artifact's contract: return an absolute path for a given src. + mock_model.register_artifact.side_effect = lambda config_path, src, verify_src_exists=True: os.path.abspath( + src + ) + + MagpieTTSModel._register_tokenizer_artifacts(mock_model, cfg) + + registered_paths = [call.args[0] for call in mock_model.register_artifact.call_args_list] + assert "text_tokenizers.hindi_phoneme.g2p.phoneme_dict.0" in registered_paths + assert "text_tokenizers.hindi_phoneme.g2p.phoneme_dict.1" in registered_paths + # Regression guard: the underscore form is what broke .nemo restore. + assert not any("phoneme_dict_0" in p or "phoneme_dict_1" in p for p in registered_paths) + + # phoneme_dict must remain a 2-element list (not collapsed to sibling keys / null). + g2p = cfg.text_tokenizers.hindi_phoneme.g2p + assert isinstance(g2p.phoneme_dict, ListConfig) and len(g2p.phoneme_dict) == 2 + assert "phoneme_dict_0" not in g2p and "phoneme_dict_1" not in g2p + + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_list_phoneme_dict_survives_nemo_save_restore(self, tmp_path): + """End-to-end: a list phoneme_dict resolves to existing packaged files after save_to/restore_from.""" + d1 = tmp_path / "hi_prondict.dict" + d1.write_text("नमस्ते\tn a m a s t e\n", encoding="utf-8") + d2 = tmp_path / "ipa_cmudict.txt" + d2.write_text("HELLO\th ʌ l o\n", encoding="utf-8") + + model = _TokenizerArtifactModel(_g2p_cfg([str(d1), str(d2)])) + # After registration the config is a 2-element list of absolute paths. + pd = model.cfg.text_tokenizers.hindi_phoneme.g2p.phoneme_dict + assert isinstance(pd, ListConfig) and len(pd) == 2 + + nemo_path = str(tmp_path / "list_dict_model.nemo") + model.save_to(nemo_path) + restored = _TokenizerArtifactModel.restore_from(nemo_path, map_location="cpu") + + g2p = restored.cfg.text_tokenizers.hindi_phoneme.g2p + rpd = g2p.phoneme_dict + assert isinstance(rpd, ListConfig) and len(rpd) == 2 + # No sibling keys from the underscore bug, and the list is intact. + assert "phoneme_dict_0" not in g2p and "phoneme_dict_1" not in g2p + # Both list entries resolved to real packaged files at restore time (read in __init__). + restored_basenames = " ".join(restored.loaded_phoneme_dicts) + assert len(restored.loaded_phoneme_dicts) == 2 + assert "hi_prondict.dict" in restored_basenames + assert "ipa_cmudict.txt" in restored_basenames + + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_string_phoneme_dict_survives_nemo_save_restore(self, tmp_path): + """The common single-file (string) phoneme_dict still round-trips through .nemo.""" + d = tmp_path / "en.dict" + d.write_text("HELLO\th ʌ l o\n", encoding="utf-8") + + model = _TokenizerArtifactModel(_g2p_cfg(str(d))) + nemo_path = str(tmp_path / "string_dict_model.nemo") + model.save_to(nemo_path) + restored = _TokenizerArtifactModel.restore_from(nemo_path, map_location="cpu") + + rpd = restored.cfg.text_tokenizers.hindi_phoneme.g2p.phoneme_dict + assert isinstance(rpd, str) + # The single entry resolved to a real packaged file at restore time (read in __init__). + assert restored.loaded_phoneme_dicts == [os.path.basename(rpd)] + assert restored.loaded_phoneme_dicts[0].endswith("en.dict") diff --git a/tests/collections/tts/modules/test_magpietts_chunked_attn_prior.py b/tests/collections/tts/modules/test_magpietts_chunked_attn_prior.py index 4f0d06bae987..e066567e1948 100644 --- a/tests/collections/tts/modules/test_magpietts_chunked_attn_prior.py +++ b/tests/collections/tts/modules/test_magpietts_chunked_attn_prior.py @@ -22,14 +22,14 @@ import pytest import torch -from nemo.collections.tts.models.magpietts import ChunkedInferenceConfig, ChunkState +from nemo.collections.tts.models.magpietts import ChunkState, ModelInferenceParameters class _StubModel: """Minimal stub that exposes only what _initialize_chunked_attn_prior needs.""" def __init__(self): - self.chunked_inference_config = ChunkedInferenceConfig() + self.inference_parameters = ModelInferenceParameters() @staticmethod def _to_int(x): @@ -39,6 +39,7 @@ def _to_int(x): from nemo.collections.tts.models.magpietts import MagpieTTSModel _initialize_chunked_attn_prior = MagpieTTSModel._initialize_chunked_attn_prior + get_most_attended_text_timestep = MagpieTTSModel.get_most_attended_text_timestep def _make_chunk_state(batch_size, previous_attn_len, left_offset=None): @@ -53,7 +54,7 @@ class TestInitializeChunkedAttnPrior: """Tests for the bounds-check in _initialize_chunked_attn_prior.""" prior_epsilon = 1e-8 - prior_weights = ChunkedInferenceConfig().prior_weights_init # (0.5, 1.0, 0.8, 0.2, 0.2) + prior_weights = ModelInferenceParameters().prior_weights_init def _call(self, chunk_state, current_chunk_len, batch_text_lens, max_text_len, batch_size, use_cfg=False): model = _StubModel() @@ -135,3 +136,39 @@ def test_batch_mixed_boundary_conditions(self): # Item 1: only first weight at index 9; offsets 1..4 were out of bounds assert result[1, 0, 9].item() == pytest.approx(self.prior_weights[0]) + + +class TestModelInferenceParameters: + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_from_dict_loads_chunked_inference_parameters(self): + params = ModelInferenceParameters.from_dict( + { + "history_len_heuristic": 2, + "prior_weights_init": [0.5, 1.0], + "finished_limit_first_chunk": 7, + "chunked_attention_sink_threshold": 6, + } + ) + + assert params.history_len_heuristic == 2 + assert params.prior_weights_init == (0.5, 1.0) + assert params.finished_limit_first_chunk == 7 + assert params.chunked_attention_sink_threshold == 6 + + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_attention_sink_threshold_controls_attended_position_advance(self): + model = _StubModel() + model.inference_parameters.attention_sink_threshold = 2 + + attended, _ = model.get_most_attended_text_timestep( + alignment_attention_scores=torch.tensor([[0.0, 1.0, 0.5, 0.0, 0.0, 0.0]]), + last_attended_timesteps=[[1]], + text_lens=torch.tensor([6]), + lookahead_window_size=3, + attended_timestep_counter=[{1: 2}], + batch_size=1, + ) + + assert attended == [2] diff --git a/tests/collections/tts/parts/utils/test_tts_dataset_utils.py b/tests/collections/tts/parts/utils/test_tts_dataset_utils.py index 53cda1d9e14d..48e1ac613456 100644 --- a/tests/collections/tts/parts/utils/test_tts_dataset_utils.py +++ b/tests/collections/tts/parts/utils/test_tts_dataset_utils.py @@ -26,6 +26,7 @@ filter_dataset_by_duration, get_abs_rel_paths, get_audio_filepaths, + get_tokenizer_for_language, load_audio, normalize_volume, split_by_sentence, @@ -817,3 +818,34 @@ def test_word_count_unknown_language_fallback(self): """Test unknown language falls back to whitespace splitting.""" text = "word1 word2 word3" assert self.thresholds.get_word_count(text, "unknown") == 3 + + +class TestGetTokenizerForLanguage: + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_custom_mapping_uses_first_available_candidate(self): + mapping = {"hi": ["hindi_phoneme", "hindi_chartokenizer"]} + + result = get_tokenizer_for_language( + "hi", ["english_phoneme", "hindi_chartokenizer"], language_tokenizer_map=mapping + ) + + assert result == "hindi_chartokenizer" + + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_custom_mapping_accepts_single_tokenizer(self): + result = get_tokenizer_for_language( + "hi", ["english_phoneme", "hindi_phoneme"], language_tokenizer_map={"hi": "hindi_phoneme"} + ) + + assert result == "hindi_phoneme" + + @pytest.mark.run_only_on('CPU') + @pytest.mark.unit + def test_custom_mapping_preserves_default_fallback(self): + result = get_tokenizer_for_language( + "hi", ["english_phoneme"], language_tokenizer_map={"hi": ["hindi_phoneme"]} + ) + + assert result == "english_phoneme"