| """Runtime tests that need no model files: fake graphs stand in for the bundle's sessions."""
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| import os
|
| from pathlib import Path
|
| import tempfile
|
| import unittest
|
| from unittest import mock
|
|
|
| import numpy as np
|
|
|
| from scyllasband.contract import (BundleError, CONTEXT_INPUTS, DURATION_INPUTS, GRAPH_CONTRACT, GRAPH_CONTRACT_V4, PITCH_INPUTS,
|
| VECTOR_INPUTS, VECTOR_INPUTS_V4, VOCODER_INPUTS, validate_bundle_layout)
|
| from scyllasband.delivery import CONTROLS, DEFAULT_RANGES, delivery_tensors, resolve_delivery, with_defaults
|
| from scyllasband.engine import Engine, OverlongError, Sentence, balanced_context
|
| from scyllasband.events import frame_events, frames_to_durations, modifier_bits, sentence_types
|
| from scyllasband.g2p import punctuated_segments
|
| from scyllasband.planner import parse_group_lines, plan_records, records_from_text, split_for_retry, split_sentences
|
| from scyllasband.streaming import StreamOptions, synthesize_plan_stream
|
|
|
| VOCAB = ["<pad>", "<unk>", "<sil>", "<pause_comma>", "<pause_semicolon>", "<pause_colon>", "<pause_dash>", "<ellipsis>", "<end_stmt>",
|
| "<end_question>", "<end_exclaim>", "a", "b", "h", "l", "o", "w", "ˈ", "ː"]
|
| PHONE_TO_ID = {p: i for i, p in enumerate(VOCAB)}
|
|
|
| V4_DELIVERY_MODES = dict(enabled=True, input="delivery_mode_id", ids=dict(spoken=0, sung=1, rap=2, ballad=3), default="spoken",
|
| singing=dict(modes=["sung", "ballad"], sampler="heun", steps=8, decoder="bigvgan",
|
| melody=dict(layout="note_phase_presence", midi_center=60, midi_scale=24),
|
| pitch_predictor="bypassed", pitch_soft_limit="bypassed"))
|
| V4_DECODERS = dict(vocos=dict(component="vocoder", samples_per_frame=512, output_samples="latent_frames * 512 - 256"),
|
| bigvgan=dict(component="vocoder_bigvgan", samples_per_frame=512, output_samples="latent_frames * 512", required=False))
|
|
|
|
|
| class PunctuationTest(unittest.TestCase):
|
| def test_segments_carry_their_punctuation(self):
|
| self.assertEqual(punctuated_segments("Wait — what was that? I heard it, too."),
|
| [("Wait", ["<pause_dash>"]), ("what was that", ["<end_question>"]), ("I heard it", ["<pause_comma>"]),
|
| ("too", ["<end_stmt>"])])
|
|
|
| def test_stacked_marks_stay_on_the_previous_phrase(self):
|
| self.assertEqual(punctuated_segments("for Death, — He kindly stopped —"),
|
| [("for Death", ["<pause_comma>", "<pause_dash>"]), ("He kindly stopped", ["<pause_dash>"])])
|
|
|
| def test_a_sentence_end_wins_over_other_stacked_marks(self):
|
| self.assertEqual(punctuated_segments("Really?! she asked."), [("Really", ["<end_question>"]), ("she asked", ["<end_stmt>"])])
|
|
|
| def test_long_phrases_split_between_words_without_punctuation(self):
|
| segments = punctuated_segments("one two three four five six.", max_chars=10)
|
| self.assertEqual([s for s, _ in segments], ["one two", "three four", "five six"])
|
| self.assertEqual([t for _, t in segments], [[], [], ["<end_stmt>"]])
|
|
|
|
|
| class FakeG2PTest(unittest.TestCase):
|
| def test_layout_word_starts_and_punctuation_silences(self):
|
| engine = _fake_engine()
|
| result = engine.phonemize("hello bob, ah.", language="en_us")
|
| phones = result["phones"]
|
| self.assertEqual(phones[0], "<sil>")
|
| self.assertEqual(phones[-1], "<sil>")
|
| self.assertIn("<pause_comma>", phones)
|
| comma = phones.index("<pause_comma>")
|
| self.assertEqual(phones[comma + 1], "<sil>")
|
| self.assertEqual([phones[i - 1] for i in result["word_starts"]], ["<sil>", "<sil>"])
|
| self.assertEqual(len(result["word_boundary_candidates"]), 1)
|
|
|
|
|
| class DurationAndEventTest(unittest.TestCase):
|
| def test_rounding_matches_training(self):
|
| phones = ["<sil>", "ˈ", "a", "<pause_comma>", "<sil>", "b", "<sil>"]
|
| self.assertEqual(frames_to_durations([0.4, 3, 0.2, 5, 2.5, 3.5, 0.0], phones), [0, 0, 1, 0, 2, 4, 0])
|
|
|
| def test_pause_scale_moves_only_silences(self):
|
| phones = ["<sil>", "a", "<sil>", "b", "<sil>"]
|
| self.assertEqual(frames_to_durations([4, 4, 0.4, 4, 2], phones, pause_scale=1.3), [5, 4, 1, 4, 3])
|
| self.assertEqual(frames_to_durations([4, 4, 0.4, 4, 2], phones, pause_scale=0.7), [3, 4, 0, 4, 1])
|
|
|
| def test_events(self):
|
| phones = ["<sil>", "ˈ", "a", "b", "<pause_comma>", "<sil>", "h", "ː", "<end_stmt>", "<sil>"]
|
| durations = [2, 0, 3, 2, 0, 3, 2, 0, 0, 2]
|
| bits = modifier_bits(PHONE_TO_ID)
|
| events = frame_events(phones, durations, word_starts=[6], following=[], token_to_id=PHONE_TO_ID, bits=bits)
|
| boundary = events["expanded_boundary_event_ids"]
|
| self.assertTrue((boundary[7:10] == PHONE_TO_ID["<pause_comma>"]).all())
|
| self.assertTrue((boundary[10:12] == len(PHONE_TO_ID)).all())
|
| self.assertTrue((boundary[12:14] == PHONE_TO_ID["<end_stmt>"]).all())
|
| modifier = events["expanded_modifier_event_ids"]
|
| self.assertTrue((modifier[2:5] == bits["ˈ"]).all())
|
| self.assertTrue((modifier[10:12] == bits["ː"]).all())
|
| np.testing.assert_allclose(events["expanded_phone_phase"][2:5], [1 / 6, 0.5, 5 / 6], rtol=1e-6)
|
| types = events["expanded_sentence_type_ids"]
|
| self.assertTrue((types[:-2] == 1).all())
|
| self.assertTrue((types[-2:] == 0).all())
|
|
|
| def test_zero_frame_silence_moves_the_event_to_the_next_phone(self):
|
| phones = ["<sil>", "a", "<pause_comma>", "<sil>", "b", "<sil>"]
|
| events = frame_events(phones, [1, 2, 0, 0, 2, 1], word_starts=[4], following=[], token_to_id=PHONE_TO_ID, bits={})
|
| self.assertEqual(events["expanded_boundary_event_ids"].tolist(), [0, 0, 0, PHONE_TO_ID["<pause_comma>"], 0, 0])
|
|
|
| def test_sentence_type_reads_the_following_context(self):
|
| self.assertEqual(sentence_types(["a", "<pause_comma>", "b"], ["c", "<end_question>"]), [2, 2, 2])
|
| self.assertEqual(sentence_types(["a"], []), [0])
|
|
|
|
|
| class ContextTest(unittest.TestCase):
|
| def test_balanced_context(self):
|
| span, segments = balanced_context(list(range(10)), [99] * 4, list(range(20, 30)), 10)
|
| self.assertEqual(segments, [0, 0, 0, 1, 1, 1, 1, 2, 2, 2])
|
| self.assertEqual(span[:3], [7, 8, 9])
|
| self.assertEqual(span[-3:], [20, 21, 22])
|
|
|
| def test_spare_room_goes_to_the_longer_side(self):
|
| span, segments = balanced_context([1, 2], [9], list(range(20, 30)), 8)
|
| self.assertEqual(segments, [0, 0, 1, 2, 2, 2, 2, 2])
|
|
|
|
|
| class NormalizerTest(unittest.TestCase):
|
| def test_decorative_asterisks_are_not_spoken(self):
|
| from scyllasband.text_normalizer import normalize_spoken_text
|
| self.assertEqual(normalize_spoken_text("* * * * * The Pool of Tears. It was *very* odd.*", language="en_gb"),
|
| "The Pool of Tears. It was very odd.")
|
|
|
| def test_letters_glued_to_digits_are_split(self):
|
| from scyllasband.text_normalizer import normalize_spoken_text
|
| self.assertEqual(normalize_spoken_text("Whisper Large V3 beat v3.5.", language="en_us"),
|
| "Whisper Large vee three beat vee three point five.")
|
| self.assertEqual(normalize_spoken_text("MP3 in 4K, H2O, Python3 and COVID19.", language="en_us"),
|
| "em pee three in four kay, aitch two oh, Python three and COVID nineteen.")
|
| self.assertEqual(normalize_spoken_text("La V3 y el MP3.", language="es"), "La uve tres y el eme pe tres.")
|
| self.assertEqual(normalize_spoken_text("test_voice2.py", language="en_us"), "test_voice2.py")
|
|
|
| def test_english_decades_clock_times_and_years(self):
|
| from scyllasband.text_normalizer import normalize_spoken_text
|
| self.assertEqual(normalize_spoken_text("The 1990s, the '80s and her 20s.", language="en_us"),
|
| "The nineteen nineties, the eighties and her twenties.")
|
| self.assertEqual(normalize_spoken_text("100s of 4s, two MP3s, Route 66's and 1.5s.", language="en_us"),
|
| "hundreds of fours, two em pee threes, Route sixty six's and one point five seconds.")
|
| self.assertEqual(normalize_spoken_text("68x faster on a 4x4 grid, 1.5x, 20ms, 24kHz, 1 GB and 60fps.", language="en_us"),
|
| "sixty eight times faster on a four by four grid, one point five times, twenty milliseconds, "
|
| "twenty four kilohertz, one gigabyte and sixty frames per second.")
|
| self.assertEqual(normalize_spoken_text("At 10am, 5 PM or 7:05 AM.", language="en_us"),
|
| "At ten ay em, five pee em or seven oh five ay em.")
|
| self.assertEqual(normalize_spoken_text("On March 4, 1990 and 3/4/2005.", language="en_us"),
|
| "On March fourth, nineteen ninety and March fourth, two thousand five.")
|
|
|
|
|
| class LexiconTest(unittest.TestCase):
|
| def test_entries_and_matching(self):
|
| from scyllasband.lexicon import LETTERS, WORDS, Lexicon, initialism_phones, split_phones
|
| self.assertEqual(initialism_phones("GPU", LETTERS["en_us"]), "dʒˌiːpˌiːjˈuː")
|
| spelled = "".join(t for table in LETTERS.values() for t in table.values()) + "".join(p for w in WORDS.values() for p in w.values())
|
| vocabulary = {ch: 1 for ch in spelled.replace("{s}", "") + "ˈˌzsˈɪlə"}
|
| self.assertEqual(split_phones("ɡˈɪt", vocabulary), ("ɡ", "ˈ", "ɪ", "t"))
|
| self.assertIsNone(split_phones("ɡˈɪt§", vocabulary))
|
| lexicon = Lexicon(vocabulary, {"en_us": {"Scylla": "sˈɪlə"}})
|
| self.assertEqual("".join(lexicon.lookup("GPUs", "en_us", initialisms=True)), "dʒˌiːpˌiːjˈuːz")
|
| self.assertEqual("".join(lexicon.lookup("Git", "en_us", initialisms=True)), "ɡˈɪt")
|
| self.assertEqual("".join(lexicon.lookup("scylla", "en_us", initialisms=True)), "sˈɪlə")
|
| self.assertIsNone(lexicon.lookup("GPU", "en_us", initialisms=False))
|
| self.assertIsNone(lexicon.lookup("US", "en_us", initialisms=True))
|
| self.assertIsNone(lexicon.lookup("GPU", "es", initialisms=True))
|
|
|
| def test_native_lexicon_table_is_current(self):
|
| import importlib.util
|
| path = Path(__file__).resolve().parents[1] / "libscyllasband" / "tools" / "generate_lexicon.py"
|
| spec = importlib.util.spec_from_file_location("generate_lexicon", path)
|
| module = importlib.util.module_from_spec(spec)
|
| spec.loader.exec_module(module)
|
| self.assertEqual(module.OUT.read_text(encoding="utf-8"), module.render())
|
|
|
|
|
| class InitialismTest(unittest.TestCase):
|
| def test_names_stay_in_their_sentence_and_times_can_end_one(self):
|
| from scyllasband.text_normalizer import normalize_spoken_text
|
| self.assertEqual(normalize_spoken_text("D.J. Stumpy met J.K. Rowling.", language="en_us"), "dee jay Stumpy met jay kay Rowling.")
|
| self.assertEqual(normalize_spoken_text("We left at 5 p.m. Then it rained.", language="en_us"),
|
| "We left at five pee em. Then it rained.")
|
| self.assertEqual(normalize_spoken_text("She lives in the U.S.", language="en_us"), "She lives in the you ess.")
|
|
|
|
|
| class PlannerTest(unittest.TestCase):
|
| def test_sentence_split(self):
|
| self.assertEqual(split_sentences('You. Are. Not. It. "Really?!" she asked... Fine'),
|
| ["You.", "Are.", "Not.", "It.", '"Really?!"', "she asked...", "Fine"])
|
|
|
| def test_retry_split_prefers_clause_punctuation(self):
|
| self.assertEqual(split_for_retry("one two three, four five six seven"), ["one two three,", "four five six seven"])
|
|
|
| def test_retry_split_leaves_each_piece_a_minimum_share(self):
|
| self.assertEqual(split_for_retry("Well, one two three four five six"), ["Well,", "one two three four five six"])
|
| self.assertEqual(split_for_retry("Well, one two three four five six", min_share=0.25), ["Well, one two", "three four five six"])
|
|
|
| def test_chains_follow_voice_language_and_delivery(self):
|
| records = parse_group_lines("[scylla:en_us] Hi there. How are you?\n[ink:en_gb:intensity=1,pace=-0.5] Fine. [scylla] Good.",
|
| default_voice=None, default_language=None)
|
| plan = plan_records(records, resolve_language=lambda voice, language: language or "en_us", normalize=False)
|
| self.assertEqual([c.chain_id for c in plan.chunks], ["chain_001", "chain_001", "chain_002", "chain_003"])
|
| self.assertEqual(plan.chunks[2].delivery, dict(intensity=1.0, pace=-0.5, pitch=None))
|
| self.assertIsNone(plan.chunks[0].delivery)
|
|
|
| def test_delivery_fields_or_string_not_both(self):
|
| plan = plan_records(records_from_text("Hi.", voice="scylla", language=None, pace=-1), resolve_language=lambda v, l: "en_us")
|
| self.assertEqual(plan.chunks[0].delivery, dict(intensity=None, pace=-1.0, pitch=None))
|
| with self.assertRaisesRegex(ValueError, "not both"):
|
| plan_records(records_from_text("Hi.", voice="scylla", language=None, delivery="pace=1", pace=-1),
|
| resolve_language=lambda v, l: "en_us")
|
| with self.assertRaisesRegex(ValueError, "emotion comes from the text"):
|
| plan_records([dict(text="Hi.", voice="scylla", emotion="joy")], resolve_language=lambda v, l: "en_us")
|
|
|
|
|
| class DeliveryTest(unittest.TestCase):
|
| def test_defaults(self):
|
| self.assertEqual(resolve_delivery(None), dict.fromkeys(CONTROLS))
|
| self.assertEqual(resolve_delivery(""), resolve_delivery("default"))
|
| self.assertEqual(with_defaults(resolve_delivery("pace=-1"), dict(pace=0.5, pitch=1)), dict(intensity=0.0, pace=-1.0, pitch=1.0))
|
| values, present = delivery_tensors(with_defaults({}, None))
|
| self.assertEqual(present[0].tolist(), [True, True, False, False, False])
|
| self.assertEqual(values[0].tolist(), [0, 0, 0, 0, 0])
|
|
|
| def test_values_are_z_over_five_and_clamped_into_the_ranges(self):
|
| resolved = resolve_delivery("intensity=5,pace=-3,pitch=2.5")
|
| self.assertEqual(resolved, dict(intensity=2.0, pace=-2.0, pitch=2.0))
|
| self.assertEqual(resolve_delivery({"pace": 1}, ranges=dict(pace=[-0.5, 0.5]))["pace"], 0.5)
|
| values, _ = delivery_tensors(dict(intensity=1.0, pace=-2.0, pitch=0.0))
|
| np.testing.assert_allclose(values[0, :2], [0.2, -0.4], rtol=1e-6)
|
| for low, high in DEFAULT_RANGES.values():
|
| self.assertTrue(low <= 0.0 <= high)
|
|
|
| def test_v2_axes_and_unknown_keys_are_rejected(self):
|
| for spec in ("energy=2", "tension=2,pace=1", {"whisper": False}, {"emotion": "joy"}):
|
| with self.assertRaisesRegex(ValueError, "intensity, pace and pitch"):
|
| resolve_delivery(spec)
|
| with self.assertRaisesRegex(ValueError, "only intensity"):
|
| resolve_delivery("tempo=1")
|
| with self.assertRaisesRegex(ValueError, "must be a number"):
|
| resolve_delivery("pace=fast")
|
|
|
|
|
| class ContractTest(unittest.TestCase):
|
| def test_valid_and_invalid_bundles(self):
|
| with tempfile.TemporaryDirectory() as tmp:
|
| bundle = _fake_bundle(Path(tmp))
|
| summary = validate_bundle_layout(bundle)
|
| self.assertEqual(summary["buckets"], [8, 16])
|
| manifest = json.loads((bundle / "manifest.json").read_text())
|
| manifest["components"]["vector_estimator"]["inputs"] = list(VECTOR_INPUTS[:-1])
|
| (bundle / "manifest.json").write_text(json.dumps(manifest))
|
| with self.assertRaises(BundleError):
|
| validate_bundle_layout(bundle)
|
|
|
| def test_v4_bundles_take_the_mode_input_and_optional_decoders(self):
|
| with tempfile.TemporaryDirectory() as tmp:
|
| bundle = _fake_bundle(Path(tmp), v4=True)
|
| summary = validate_bundle_layout(bundle)
|
| self.assertEqual((summary["graph_contract"], summary["singing"]), (GRAPH_CONTRACT_V4, True))
|
| manifest = json.loads((bundle / "manifest.json").read_text())
|
|
|
| broken = json.loads(json.dumps(manifest))
|
| broken["components"]["vector_estimator"]["inputs"] = list(VECTOR_INPUTS)
|
| (bundle / "manifest.json").write_text(json.dumps(broken))
|
| with self.assertRaisesRegex(BundleError, "vector_estimator"):
|
| validate_bundle_layout(bundle)
|
| broken = json.loads(json.dumps(manifest))
|
| broken["controls"]["graph_input_contract"] = GRAPH_CONTRACT
|
| (bundle / "manifest.json").write_text(json.dumps(broken))
|
| with self.assertRaisesRegex(BundleError, "vector_estimator"):
|
| validate_bundle_layout(bundle)
|
|
|
| tolerant = json.loads(json.dumps(manifest))
|
| tolerant["components"]["vocoder_bigvgan_8"]["artifacts"] = {}
|
| (bundle / "manifest.json").write_text(json.dumps(tolerant))
|
| self.assertTrue(validate_bundle_layout(bundle)["singing"])
|
| del tolerant["components"]["vocoder_bigvgan"]
|
| (bundle / "manifest.json").write_text(json.dumps(tolerant))
|
| with self.assertRaisesRegex(BundleError, "vocoder_bigvgan"):
|
| validate_bundle_layout(bundle)
|
|
|
| def test_v3_bundles_validate_without_the_mode_input(self):
|
| with tempfile.TemporaryDirectory() as tmp:
|
| summary = validate_bundle_layout(_fake_bundle(Path(tmp)))
|
| self.assertEqual((summary["graph_contract"], summary["singing"]), (GRAPH_CONTRACT, False))
|
|
|
|
|
| class LiteRTThreadsTest(unittest.TestCase):
|
| def test_default_threads_follow_the_native_runtime(self):
|
| from unittest import mock
|
| from scyllasband import litert
|
| with mock.patch.object(litert.os, "sched_getaffinity", return_value=set(range(16)), create=True):
|
| self.assertEqual(litert.default_threads(), 4)
|
| with mock.patch.object(litert.os, "sched_getaffinity", return_value={0, 1}, create=True):
|
| self.assertEqual(litert.default_threads(), 2)
|
|
|
| def test_interpreters_get_the_default_unless_threads_are_given(self):
|
| from unittest import mock
|
| from scyllasband import litert
|
| made = []
|
|
|
| class Interpreter:
|
| def __init__(self, model_path, num_threads):
|
| made.append(num_threads)
|
|
|
| spec = {"artifacts": {"litert": {"path": "g2p.tflite"}}, "inputs": []}
|
| with mock.patch.object(litert, "_interpreter_class", return_value=Interpreter), \
|
| mock.patch.object(litert, "LiteRTSession"), mock.patch.object(litert, "default_threads", return_value=3):
|
| litert.litert_session_factory(Path("."))("g2p", spec)
|
| litert.litert_session_factory(Path("."), threads=8)("g2p", spec)
|
| self.assertEqual(made, [3, 8])
|
|
|
|
|
| class OnnxProvidersTest(unittest.TestCase):
|
| def test_compute_units_pick_execution_providers(self):
|
| from scyllasband.engine import onnx_providers
|
| cpu, cuda = ["CPUExecutionProvider"], ["CUDAExecutionProvider", "CPUExecutionProvider"]
|
| for value in (None, "", "auto", "cpu"):
|
| self.assertEqual(onnx_providers(value)("vector_estimator_256"), cpu)
|
| per_graph = onnx_providers("gpu,g2p=cpu")
|
| self.assertEqual(per_graph("vector_estimator_256"), cuda)
|
| self.assertEqual(per_graph("g2p"), cpu)
|
| self.assertEqual(onnx_providers("cuda")("vocoder"), cuda)
|
| self.assertEqual(onnx_providers({"vocoder": "gpu"})("vocoder"), cuda)
|
| with self.assertRaises(ValueError):
|
| onnx_providers("ane")
|
|
|
| def test_a_cuda_session_that_falls_back_to_the_cpu_is_an_error(self):
|
| import sys
|
| import types
|
| from unittest import mock
|
| from scyllasband.engine import OnnxSession, _preload_cuda
|
|
|
| class Session:
|
| def __init__(self, path, options, providers):
|
| pass
|
|
|
| def get_providers(self):
|
| return ["CPUExecutionProvider"]
|
|
|
| def get_inputs(self):
|
| return []
|
|
|
| fake = types.SimpleNamespace(SessionOptions=lambda: types.SimpleNamespace(), InferenceSession=Session,
|
| preload_dlls=lambda: None)
|
| _preload_cuda.cache_clear()
|
| with mock.patch.dict(sys.modules, {"onnxruntime": fake}):
|
| OnnxSession(Path("g2p.onnx"))
|
| with self.assertRaisesRegex(RuntimeError, r"\.\[cuda\]"):
|
| OnnxSession(Path("g2p.onnx"), providers=["CUDAExecutionProvider", "CPUExecutionProvider"])
|
| _preload_cuda.cache_clear()
|
|
|
| def test_gpu_requests_use_the_webgpu_bundle(self):
|
| import argparse
|
| from scyllasband.cli import _default_bundle, wants_gpu
|
| self.assertTrue(wants_gpu("gpu") and wants_gpu("cuda") and wants_gpu("cpu,vector_estimator=gpu"))
|
| self.assertFalse(wants_gpu(None) or wants_gpu("cpu") or wants_gpu("gpu_like=cpu"))
|
| with tempfile.TemporaryDirectory() as tmp:
|
| for flavor in ("onnx", "webgpu"):
|
| Path(tmp, flavor).mkdir()
|
| Path(tmp, flavor, "manifest.json").write_text("{}")
|
| args = lambda units: argparse.Namespace(models_dir=tmp, backend="onnx", compute_units=units)
|
| self.assertEqual(_default_bundle(args("gpu")).name, "webgpu")
|
| self.assertEqual(_default_bundle(args(None)).name, "onnx")
|
|
|
|
|
| class EngineTest(unittest.TestCase):
|
| def test_bucket_padding_prefix_and_decode(self):
|
| engine = _fake_engine()
|
| sentence = Sentence(phones=["<sil>", "a", "b", "<end_stmt>", "<sil>"], word_starts=[], before_ids=[11, 12], after_ids=[13])
|
| delivery = engine.delivery_inputs("scylla", None)[:2]
|
| result = engine.synthesize_sentence(sentence, voice="scylla", language="en_us", delivery=delivery, steps=2,
|
| rng=np.random.default_rng(0), prefix=np.ones((24, 3), np.float32))
|
| self.assertEqual(result.latents.shape, (24, sum(result.durations)))
|
| self.assertEqual(result.bucket, 8)
|
| self.assertEqual(result.metadata["prefix_frames"], 3)
|
| calls = engine._fake.calls
|
| self.assertTrue(all(call["noise"].shape == (1, 24, 8) for call in calls if call["name"] == "vector_estimator_8"))
|
| audio = engine.decode(result.latents, voice="scylla", language="en_us", left=np.zeros((24, 20), np.float32))
|
| self.assertEqual(audio.shape, (result.latents.shape[1] * 512,))
|
|
|
| def test_decode_gives_left_context_way_to_long_sentences(self):
|
| engine = _fake_engine()
|
| left = np.zeros((24, 20), np.float32)
|
| for frames in (15, 16):
|
| audio = engine.decode(np.zeros((24, frames), np.float32), voice="scylla", language="en_us", left=left)
|
| self.assertEqual(audio.shape, (frames * 512,))
|
| self.assertEqual(engine._fake.calls[-1]["latents"].shape, (1, 24, 16))
|
| self.assertEqual(int(engine._fake.calls[-1]["latent_mask"].sum()), frames)
|
|
|
| def test_fused_flow_replaces_the_step_loop(self):
|
| engine = _fake_engine(apple=True)
|
| sentence = Sentence(phones=["<sil>", "a", "b", "<end_stmt>", "<sil>"], word_starts=[])
|
| delivery = engine.delivery_inputs("scylla", None)[:2]
|
| fused = engine.synthesize_sentence(sentence, voice="scylla", language="en_us", delivery=delivery, steps=8, sampler="heun",
|
| rng=np.random.default_rng(0))
|
| names = [call["name"] for call in engine._fake.calls]
|
| self.assertEqual(names.count("vector_flow_8"), 1)
|
| self.assertFalse(any(name.startswith("vector_estimator") for name in names))
|
| self.assertNotIn("time", engine._fake.calls[names.index("vector_flow_8")])
|
| self.assertTrue(fused.metadata["fused_flow"])
|
| engine._fake.calls.clear()
|
| stepped = engine.synthesize_sentence(sentence, voice="scylla", language="en_us", delivery=delivery, steps=2, sampler="heun",
|
| rng=np.random.default_rng(0))
|
| names = [call["name"] for call in engine._fake.calls]
|
| self.assertEqual(names.count("vector_estimator_8"), 4)
|
| self.assertFalse(stepped.metadata["fused_flow"])
|
|
|
| def test_g2p_runs_the_narrowest_bucket_that_fits(self):
|
| plain, bucketed = _fake_engine(), _fake_engine(apple=True)
|
| self.assertEqual(bucketed.phonemize("hello", language="en_us")["phones"], plain.phonemize("hello", language="en_us")["phones"])
|
| call = next(c for c in bucketed._fake.calls if c["name"].startswith("g2p"))
|
| self.assertEqual((call["name"], call["text"].shape), ("g2p_16", (1, 16)))
|
| bucketed.phonemize("hello hello hello hello", language="en_us")
|
| self.assertEqual(bucketed._fake.calls[-1]["name"], "g2p_32")
|
|
|
| def test_overlong(self):
|
| engine = _fake_engine(frames_per_phone=9.0)
|
| sentence = Sentence(phones=["<sil>", "a", "b", "a", "<sil>"], word_starts=[])
|
| with self.assertRaises(OverlongError):
|
| engine.synthesize_sentence(sentence, voice="scylla", language="en_us", delivery=engine.delivery_inputs("scylla", None)[:2],
|
| steps=1)
|
|
|
| def test_short_targets_leave_intensity_and_pace_unset(self):
|
| engine = _fake_engine()
|
| delivery = engine.delivery_inputs("scylla", dict(intensity=1.0, pace=-1.0))[:2]
|
| short = Sentence(phones=["<sil>", "ˈ", "a", "b", "a", "<end_stmt>", "<sil>"], word_starts=[])
|
| long = Sentence(phones=["<sil>", "a", "b", "a", "b", "<end_stmt>", "<sil>"], word_starts=[])
|
| for sentence, expected in ((short, [False] * 5), (long, [True, True, False, False, False])):
|
| engine._fake.calls.clear()
|
| engine.synthesize_sentence(sentence, voice="scylla", language="en_us", delivery=delivery, steps=1, rng=np.random.default_rng(0))
|
| for call in engine._fake.calls:
|
| if "delivery_present" in call:
|
| self.assertEqual(call["delivery_present"][0].tolist(), expected, call["name"])
|
|
|
| def test_pace_moves_pauses_even_when_unset(self):
|
| engine = _fake_engine(frames_per_phone=4.0)
|
| sentence = Sentence(phones=["<sil>", "a", "b", "<end_stmt>", "<sil>"], word_starts=[])
|
| normal = engine.durations(sentence, voice=0, language=0, delivery=engine.delivery_inputs("scylla", None)[:2])
|
| slow = engine.durations(sentence, voice=0, language=0, delivery=engine.delivery_inputs("scylla", dict(pace=-2.0))[:2])
|
| self.assertEqual(normal, [4, 4, 4, 0, 4])
|
| self.assertEqual(slow, [5, 4, 4, 0, 5])
|
|
|
| def test_pitch_contour_feeds_the_flow_melody(self):
|
| engine = _fake_engine()
|
| sentence = Sentence(phones=["<sil>", "a", "b", "a", "b", "<end_stmt>", "<sil>"], word_starts=[])
|
| result = engine.synthesize_sentence(sentence, voice="scylla", language="en_us", delivery=(*engine.delivery_inputs("scylla", None)[:2], 1.5),
|
| steps=1, rng=np.random.default_rng(0), pitch_noise=np.full((2, 6), 0.25, np.float32))
|
| calls = engine._fake.calls
|
| pitch = [c for c in calls if c["name"] == "pitch_estimator_8"]
|
| self.assertEqual(len(pitch), engine.pitch_steps)
|
| self.assertEqual(pitch[0]["noise"].shape, (1, 2, 8))
|
| self.assertEqual(pitch[0]["span_context_hidden"].shape, (1, 384))
|
| flow = next(c for c in calls if c["name"] == "vector_estimator_8")
|
| melody = flow["expanded_melody"]
|
| self.assertEqual(melody.shape, (1, 8, 3))
|
| self.assertTrue((melody[0, 6:] == 0).all())
|
| contour = np.full((2, 6), 0.25 * (1 - 1 / engine.pitch_steps) ** engine.pitch_steps, np.float32)
|
| np.testing.assert_allclose(melody[0, :6], engine.melody(contour, "scylla", 1.5), rtol=1e-5)
|
| self.assertEqual(result.metadata["pitch_shift"], 1.5)
|
|
|
| def test_melody_soft_limit_and_voicing(self):
|
| engine = _fake_engine()
|
| low, high = engine.pitch["edges"]["scylla"]
|
| contour = np.array([[(high + 5 - 10) / 6, (low - 5 - 10) / 6, 0.0], [1.0, 1.0, -1.0]], np.float32)
|
| melody = engine.melody(contour, "scylla")
|
| st = melody[:2, 0] * 24 + 16.65
|
| self.assertTrue(high - 2 < st[0] < high and low < st[1] < low + 2)
|
| self.assertEqual(melody[:, 2].tolist(), [1.0, 1.0, 0.0])
|
| self.assertEqual(float(melody[2, 0]), 0.0)
|
|
|
| def _spoken(self, engine, **kwargs):
|
| sentence = Sentence(phones=["<sil>", "a", "b", "a", "b", "<end_stmt>", "<sil>"], word_starts=[3], before_ids=[11], after_ids=[12])
|
| delivery = (*engine.delivery_inputs("scylla", dict(intensity=0.5))[:2], 1.0)
|
| return engine.synthesize_sentence(sentence, voice="scylla", language="en_us", delivery=delivery, steps=2, sampler="heun",
|
| rng=np.random.default_rng(7), prefix=np.ones((24, 2), np.float32), **kwargs)
|
|
|
| def test_v3_speech_feeds_exactly_the_v3_inputs_and_v4_adds_mode_zero(self):
|
| v3, v4 = _fake_engine(), _fake_engine(v4=True)
|
| spoken_v3, spoken_v4 = self._spoken(v3), self._spoken(v4)
|
| flows_v3 = [c for c in v3._fake.calls if c["name"].startswith("vector_estimator")]
|
| flows_v4 = [c for c in v4._fake.calls if c["name"].startswith("vector_estimator")]
|
| self.assertEqual(len(flows_v3), 4)
|
| for call in flows_v3:
|
| self.assertEqual(set(call) - {"name"}, set(VECTOR_INPUTS))
|
| self.assertFalse(any("delivery_mode_id" in call for call in v3._fake.calls))
|
| self.assertNotIn("delivery_mode", spoken_v3.metadata)
|
|
|
| self.assertEqual([c["name"] for c in v3._fake.calls], [c["name"] for c in v4._fake.calls])
|
| for call in flows_v4:
|
| self.assertEqual(set(call) - {"name"}, set(VECTOR_INPUTS_V4))
|
| self.assertEqual(call["delivery_mode_id"].tolist(), [0])
|
| self.assertEqual(call["delivery_mode_id"].dtype, np.int64)
|
| for a, b in zip(v3._fake.calls, v4._fake.calls):
|
| for key in a:
|
| if key != "name":
|
| np.testing.assert_array_equal(a[key], b[key], key)
|
| np.testing.assert_array_equal(spoken_v3.latents, spoken_v4.latents)
|
| self.assertEqual(spoken_v4.metadata["delivery_mode"], 0)
|
| self.assertEqual(spoken_v4.metadata["melody_source"], "pitch_predictor")
|
|
|
| def test_a_score_melody_replaces_the_pitch_predictor(self):
|
| engine = _fake_engine(v4=True)
|
| melody = np.zeros((6, 3), np.float32)
|
| melody[1:5] = [[2 / 24, 0.5, 1.0]] * 4
|
| result = self._spoken(engine, melody=melody, delivery_mode=3)
|
| names = [c["name"] for c in engine._fake.calls]
|
| self.assertFalse(any(n.startswith("pitch_") for n in names))
|
| flow = next(c for c in engine._fake.calls if c["name"].startswith("vector_estimator"))
|
| np.testing.assert_array_equal(flow["expanded_melody"][0, :6], melody)
|
| self.assertTrue((flow["expanded_melody"][0, 6:] == 0).all())
|
| self.assertEqual(flow["delivery_mode_id"].tolist(), [3])
|
| self.assertEqual((result.metadata["delivery_mode"], result.metadata["melody_source"]), (3, "score"))
|
| with self.assertRaisesRegex(ValueError, r"\[6, 3\]"):
|
| self._spoken(engine, melody=np.zeros((5, 3), np.float32))
|
| with self.assertRaisesRegex(ValueError, "v4"):
|
| self._spoken(_fake_engine(), delivery_mode=1)
|
|
|
| def test_guidance_contrasts_the_melody_and_mode_against_off(self):
|
| engine = _fake_engine(v4=True)
|
| melody = np.full((6, 3), 0.5, np.float32)
|
| self._spoken(engine, melody=melody, delivery_mode=1, guidance=2.0)
|
| flows = [c for c in engine._fake.calls if c["name"].startswith("vector_estimator")]
|
| self.assertEqual(len(flows), 8)
|
| self.assertEqual([c["delivery_mode_id"].tolist()[0] for c in flows], [1, 0] * 4)
|
| self.assertTrue((flows[1]["expanded_melody"] == 0).all() and (flows[0]["expanded_melody"][0, :6] == 0.5).all())
|
|
|
| def test_bigvgan_decoder_needs_no_spare_frame(self):
|
| engine = _fake_engine(v4=True)
|
| for frames in (9, 16):
|
| audio = engine.decode(np.zeros((24, frames), np.float32), voice="scylla", language="en_us", decoder="bigvgan")
|
| self.assertEqual(audio.shape, (frames * 512,))
|
| self.assertEqual(engine._fake.calls[-1]["name"], "vocoder_bigvgan")
|
| self.assertEqual(int(engine._fake.calls[-1]["latent_mask"].sum()), frames)
|
| engine.decode(np.zeros((24, 7), np.float32), voice="scylla", language="en_us", decoder="bigvgan")
|
| self.assertEqual(engine._fake.calls[-1]["name"], "vocoder_bigvgan_8")
|
| engine.decode(np.zeros((24, 7), np.float32), voice="scylla", language="en_us")
|
| self.assertEqual(engine._fake.calls[-1]["name"], "vocoder_8")
|
| engine.decode(np.zeros((24, 8), np.float32), voice="scylla", language="en_us")
|
| self.assertEqual(engine._fake.calls[-1]["name"], "vocoder")
|
| with self.assertRaisesRegex(ValueError, "Unknown decoder"):
|
| _fake_engine().decode(np.zeros((24, 4), np.float32), voice="scylla", language="en_us", decoder="bigvgan")
|
|
|
|
|
| class DecoderWindowTest(unittest.TestCase):
|
| """Windowed decoding (controls.decoders.<name>.window), as libscyllasband does it: with a BigVGAN stand-in whose
|
| samples are their frame's first latent channel, windows stitched from the frames each keeps equal one whole decode."""
|
|
|
| @staticmethod
|
| def _engine(window=None, *, single_bucket=False):
|
| tmp = tempfile.TemporaryDirectory()
|
| bundle = _fake_bundle(Path(tmp.name), v4=True)
|
| manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8"))
|
| if window is not None:
|
| manifest["controls"]["decoders"]["bigvgan"]["window"] = window
|
| if single_bucket:
|
| for entry in manifest["controls"]["target_buckets"]["buckets"]:
|
| if entry["latent_frames"] != 8:
|
| entry.pop("vocoder_bigvgan")
|
| del manifest["components"]["vocoder_bigvgan"]
|
| (bundle / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
| graphs = _FakeGraphs(1.0)
|
| graphs.bigvgan_by_frame = True
|
| engine = Engine(bundle, session_factory=graphs.factory)
|
| engine._fake, engine._tmp = graphs, tmp
|
| return engine
|
|
|
| @staticmethod
|
| def _latents(frames, base):
|
| latents = np.full((24, frames), 0.5, np.float32)
|
| latents[0] = base + np.arange(frames, dtype=np.float32) / 100
|
| return latents
|
|
|
| def test_windows_stitch_to_the_whole_decode(self):
|
| whole, windowed = self._engine(), self._engine(dict(frames=8, context=2))
|
| self.assertEqual(windowed.decoder_windows, {"bigvgan": (8, 2)})
|
| latents = self._latents(13, 0.1)
|
| want = whole.decode(latents, voice="scylla", language="en_us", decoder="bigvgan")
|
| got = windowed.decode(latents, voice="scylla", language="en_us", decoder="bigvgan")
|
| np.testing.assert_array_equal(got, want)
|
| self.assertEqual([c["name"] for c in windowed._fake.calls], ["vocoder_bigvgan_8"] * 3)
|
| left = self._latents(5, 0.7)
|
| np.testing.assert_array_equal(windowed.decode(latents, voice="scylla", language="en_us", left=left, decoder="bigvgan"),
|
| whole.decode(latents, voice="scylla", language="en_us", left=left, decoder="bigvgan"))
|
| short = self._latents(5, 0.3)
|
| windowed._fake.calls.clear()
|
| np.testing.assert_array_equal(windowed.decode(short, voice="scylla", language="en_us", decoder="bigvgan"),
|
| whole.decode(short, voice="scylla", language="en_us", decoder="bigvgan"))
|
| self.assertEqual([c["name"] for c in windowed._fake.calls], ["vocoder_bigvgan_8"])
|
|
|
| np.testing.assert_array_equal(windowed.decode(latents, voice="scylla", language="en_us"), whole.decode(latents, voice="scylla", language="en_us"))
|
|
|
| def test_a_single_bucket_decodes_any_length(self):
|
| engine = self._engine(dict(frames=8, context=2), single_bucket=True)
|
| latents = self._latents(13, 0.1)
|
| np.testing.assert_array_equal(engine.decode(latents, voice="scylla", language="en_us", decoder="bigvgan"),
|
| self._engine().decode(latents, voice="scylla", language="en_us", decoder="bigvgan"))
|
| with self.assertRaises(OverlongError):
|
| self._engine(single_bucket=True).decode(latents, voice="scylla", language="en_us", decoder="bigvgan")
|
|
|
| def test_bad_windows_and_the_environment(self):
|
| with self.assertRaisesRegex(ValueError, "leave frames to keep"):
|
| self._engine(dict(frames=8, context=4))
|
| with mock.patch.dict(os.environ, {"SCYLLASBAND_DECODER_WINDOW": "bigvgan=8:3"}):
|
| self.assertEqual(self._engine().decoder_windows, {"bigvgan": (8, 3)})
|
| with mock.patch.dict(os.environ, {"SCYLLASBAND_DECODER_WINDOW": "bigvgan=0"}):
|
| self.assertEqual(self._engine(dict(frames=8, context=2)).decoder_windows, {})
|
| with mock.patch.dict(os.environ, {"SCYLLASBAND_DECODER_WINDOW": "vocos=8"}), self.assertRaisesRegex(ValueError, "whole frames"):
|
| self._engine()
|
|
|
|
|
| class StreamingTargetTest(unittest.TestCase):
|
| """Fake durations: one frame per phone and silence, none for punctuation ("ab." 4 frames, "ab ab." 7); trained
|
| target range 6 to 12 frames."""
|
|
|
| def targets(self, records, frames_per_phone: float = 1.0):
|
| engine = _fake_engine(frames_per_phone, chunking=dict(min_target_frames=6, max_target_frames=12),
|
| span_conditioning=dict(context_max_phones=64, context_phones_each_side=8))
|
| plan = plan_records(records, resolve_language=lambda voice, language: "en_us")
|
| events = [e for e in synthesize_plan_stream(engine, plan, StreamOptions(steps=1, seed=0)) if e.type == "audio_chunk"]
|
| return [(e.chunk.chunk_id, e.chunk.text, e.metadata["latent_frames"]) for e in events]
|
|
|
| def test_short_sentences_join_the_following_ones_until_long_enough(self):
|
| self.assertEqual(self.targets(records_from_text("ab. ab. ab ab ab. ab ab.", voice="scylla", language=None)),
|
| [("chunk_0001+chunk_0002", "ab. ab.", 7), ("chunk_0003", "ab ab ab.", 10), ("chunk_0004", "ab ab.", 7)])
|
|
|
| def test_a_short_last_sentence_joins_the_one_before(self):
|
| self.assertEqual(self.targets(records_from_text("ab ab. ab.", voice="scylla", language=None)),
|
| [("chunk_0001+chunk_0002", "ab ab. ab.", 10)])
|
|
|
| def test_a_short_first_sentence_takes_the_next_word_rather_than_speak_alone(self):
|
|
|
| self.assertEqual(self.targets(records_from_text("ab. ab ab ab ab.", voice="scylla", language=None)),
|
| [("chunk_0001+chunk_0002.1", "ab. ab", 7), ("chunk_0002.2", "ab ab ab.", 10)])
|
|
|
| def test_later_targets_fill_toward_the_trained_maximum(self):
|
| self.assertEqual(self.targets(records_from_text("ab. ab. ab. ab. ab. ab. ab. ab.", voice="scylla", language=None)),
|
| [("chunk_0001+chunk_0002", "ab. ab.", 7), ("chunk_0003+chunk_0005", "ab. ab. ab.", 10),
|
| ("chunk_0006+chunk_0008", "ab. ab. ab.", 10)])
|
|
|
| def test_text_without_punctuation_splits_between_words(self):
|
| self.assertEqual(self.targets(records_from_text("ab ab ab ab ab ab ab ab", voice="scylla", language=None)),
|
| [("chunk_0001.1", "ab ab", 7), ("chunk_0001.2", "ab ab ab", 10), ("chunk_0001.3", "ab ab ab", 10)])
|
|
|
| def test_slow_delivery_never_exceeds_the_trained_maximum(self):
|
| frames = [f for _, _, f in self.targets(records_from_text("ab. ab. ab. ab. ab. ab.", voice="scylla", language=None),
|
| frames_per_phone=2.0)]
|
| self.assertEqual(frames, [8] * 6)
|
|
|
| def test_short_paragraphs_and_records_join_and_chains_never_do(self):
|
| paragraphs = records_from_text("ab.\n\nab.", voice="scylla", language=None)
|
| records = [dict(text="ab.", voice="scylla"), dict(text="ab.", voice="scylla")]
|
| chains = [dict(text="ab.", voice="scylla"), dict(text="ab.", voice="scylla", delivery="intensity=1")]
|
| self.assertEqual([text for _, text, _ in self.targets(paragraphs)], ["ab. ab."])
|
| self.assertEqual([text for _, text, _ in self.targets(records)], ["ab. ab."])
|
| self.assertEqual([text for _, text, _ in self.targets(chains)], ["ab.", "ab."])
|
|
|
| def test_a_long_enough_paragraph_ends_its_target(self):
|
| self.assertEqual([text for _, text, _ in self.targets(records_from_text("ab ab. ab ab.\n\nab ab.", voice="scylla",
|
| language=None))],
|
| ["ab ab.", "ab ab.", "ab ab."])
|
|
|
|
|
|
|
| class _FakeGraphs:
|
| """Graph stand-ins with the bundle's input names. The vocoder returns frames * 512 - 256 samples, like the real one;
|
| the BigVGAN decoder of a v4 bundle returns frames * 512."""
|
|
|
| def __init__(self, frames_per_phone: float):
|
| self.frames_per_phone, self.calls = frames_per_phone, []
|
| self.bigvgan_by_frame = False
|
|
|
| def factory(self, name, spec):
|
| graphs = self
|
|
|
| class Session:
|
| def run(self, feeds):
|
| graphs.calls.append(dict(feeds, name=name))
|
| if name.startswith("g2p"):
|
| return _fake_g2p_logits(feeds["text"])
|
| if name.startswith("vector_flow"):
|
| return feeds["noise"] * 0.5
|
| if name == "duration_predictor":
|
| return np.full((1, feeds["span_phone_ids"].shape[1]), graphs.frames_per_phone, np.float32)
|
| if name == "vector_context_encoder":
|
| return np.zeros((1, 512), np.float32)
|
| if name == "pitch_context_encoder":
|
| return np.zeros((1, 384), np.float32)
|
| if name.startswith("pitch_estimator"):
|
| return -feeds["noise"]
|
| if name.startswith("vector_estimator"):
|
| return -feeds["noise"]
|
| frames = feeds["latents"].shape[-1]
|
| if name.startswith("vocoder_bigvgan"):
|
| if graphs.bigvgan_by_frame:
|
| values = np.where(feeds["latent_mask"][0], feeds["latents"][0, 0], 0.0).astype(np.float32)
|
| return np.repeat(values, 512)[None]
|
| return np.ones((1, frames * 512), np.float32) * 0.2
|
| return np.ones((1, frames * 512 - 256), np.float32) * 0.1
|
| return Session()
|
|
|
|
|
|
|
| _PHONEME_SYMBOLS = {0: "<pad>", 1: "<end>", 2: " ", 3: "a", 4: "b", 5: "h", 6: "l", 7: "o", 8: "w"}
|
| _TEXT_SYMBOLS = {"<pad>": 0, "<end>": 1, "<en_us>": 2, " ": 3, "a": 4, "b": 5, "h": 6, "l": 7, "o": 8, "w": 9, "e": 10}
|
|
|
|
|
| def _fake_g2p_logits(encoded: np.ndarray) -> np.ndarray:
|
| inverse = {v: k for k, v in _TEXT_SYMBOLS.items()}
|
| phone_of = {s: i for i, s in _PHONEME_SYMBOLS.items()}
|
| frames = []
|
| for token in encoded[0].tolist():
|
| char = inverse.get(int(token), "<pad>")
|
| if char in ("<pad>", "<en_us>"):
|
| continue
|
| if char == "<end>":
|
| frames.append(1)
|
| break
|
| frames.append(phone_of.get(char, phone_of.get("a")) if char != "e" else phone_of["o"])
|
| frames.append(0)
|
| logits = np.full((1, len(frames), len(_PHONEME_SYMBOLS)), -10.0, np.float32)
|
| for index, symbol in enumerate(frames):
|
| logits[0, index, symbol] = 10.0
|
| return logits
|
|
|
|
|
| def _fake_bundle(root: Path, *, apple: bool = False, controls: dict | None = None, v4: bool = False) -> Path:
|
| """``apple``: also the optional fused-flow graphs and narrower G2P inputs of the Core ML and Core AI bundles.
|
| ``controls``: manifest controls to replace. ``v4``: the singing contract (the flow takes delivery_mode_id, delivery
|
| modes and decoders are declared, every bucket has a BigVGAN decoder)."""
|
| bundle = root / "bundle"
|
| (bundle / "assets/g2p").mkdir(parents=True)
|
| (bundle / "onnx").mkdir()
|
| flow_inputs = list(VECTOR_INPUTS_V4 if v4 else VECTOR_INPUTS)
|
| write = lambda rel, value: (bundle / rel).write_text(json.dumps(value))
|
| write("assets/phone_vocab.json", dict(token_to_id=PHONE_TO_ID))
|
| write("assets/voices.json", [dict(id="scylla", index=0)])
|
| write("assets/languages.json", [dict(id="en_us", index=0)])
|
| write("assets/g2p/export_config.json", dict(input_granularity="phrase", chunk_max_chars=100))
|
| write("assets/g2p/tokenizer.json", dict(text_symbols=_TEXT_SYMBOLS, phoneme_symbols={str(k): v for k, v in _PHONEME_SYMBOLS.items()},
|
| char_repeats=1, lowercase=True, text_pad_index=0, phoneme_pad_index=0, phoneme_end_index=1))
|
| write("assets/g2p/language_map.json", {"en_us": "en_us"})
|
| components = {}
|
| for name, inputs in (("g2p", ["text"]), ("duration_predictor", list(DURATION_INPUTS)), ("vector_context_encoder", list(CONTEXT_INPUTS)),
|
| ("pitch_context_encoder", list(CONTEXT_INPUTS)), ("vector_estimator_8", flow_inputs),
|
| ("pitch_estimator_8", list(PITCH_INPUTS)), ("vocoder_8", list(VOCODER_INPUTS)),
|
| ("vector_estimator", flow_inputs), ("pitch_estimator", list(PITCH_INPUTS)), ("vocoder", list(VOCODER_INPUTS)),
|
| *((("vocoder_bigvgan_8", list(VOCODER_INPUTS)), ("vocoder_bigvgan", list(VOCODER_INPUTS))) if v4 else ())):
|
| (bundle / f"onnx/{name}.onnx").write_bytes(b"")
|
| components[name] = dict(artifacts=dict(onnx=dict(path=f"onnx/{name}.onnx", format="onnx")), inputs=inputs, outputs=["out"])
|
| manifest = dict(contract_version="1.0.0", audio=dict(sample_rate=24000, latent_dim=24, latent_hop_length=512, hop_length=256),
|
| assets=dict(phone_vocab="assets/phone_vocab.json", voice_index="assets/voices.json", language_index="assets/languages.json",
|
| g2p_config="assets/g2p/export_config.json", g2p_tokenizer="assets/g2p/tokenizer.json",
|
| g2p_language_map="assets/g2p/language_map.json"),
|
| components=components,
|
| voices=[dict(id="scylla", languages=["en_us"], default_language="en_us")],
|
| controls=dict(graph_input_contract=GRAPH_CONTRACT, word_boundaries=dict(enabled=True, g2p_output_symbol=" "),
|
| fixed_shapes=dict(g2p_text_tokens=64), span_conditioning=dict(context_max_phones=16, context_phones_each_side=4),
|
| prefix_conditioning=dict(max_frames=6), decoding=dict(context_frames=4),
|
| delivery=dict(z_scale=5.0, z_limit=2.5, min_spoken_phones=4, ranges=dict(DEFAULT_RANGES),
|
| defaults=dict(scylla=dict(intensity=0.0, pace=0.0, pitch=0.0))),
|
| pace_pauses=dict(phone="<sil>", per_sd=-0.13),
|
| pitch=dict(channels=2, context_encoder="pitch_context_encoder", sampler=dict(default="euler"),
|
| steps=dict(default=2), center=10.0, scale=6.0, melody_center=16.65, melody_scale=24.0,
|
| voiced_threshold=0.5, knee=2.0, edges=dict(scylla=[5.0, 25.0])),
|
| target_buckets=dict(buckets=[dict(latent_frames=8, vector_estimator="vector_estimator_8", vocoder="vocoder_8",
|
| pitch_estimator="pitch_estimator_8"),
|
| dict(latent_frames=16, vector_estimator="vector_estimator", vocoder="vocoder",
|
| pitch_estimator="pitch_estimator")])))
|
| if apple:
|
| fused_inputs = [n for n in flow_inputs if n != "time"]
|
| for name, inputs in (("vector_flow_8", fused_inputs), ("vector_flow", fused_inputs), ("g2p_16", ["text"]), ("g2p_32", ["text"])):
|
| (bundle / f"onnx/{name}.onnx").write_bytes(b"")
|
| components[name] = dict(artifacts=dict(onnx=dict(path=f"onnx/{name}.onnx", format="onnx")), inputs=inputs, outputs=["out"])
|
| manifest["controls"]["fused_flow"] = dict(sampler="heun", steps=8)
|
| manifest["controls"]["g2p_buckets"] = [dict(text_tokens=16, component="g2p_16"), dict(text_tokens=32, component="g2p_32")]
|
| for entry, flow in zip(manifest["controls"]["target_buckets"]["buckets"], ("vector_flow_8", "vector_flow")):
|
| entry["vector_flow"] = flow
|
| if v4:
|
| manifest["controls"]["graph_input_contract"] = GRAPH_CONTRACT_V4
|
| manifest["controls"]["delivery_modes"] = json.loads(json.dumps(V4_DELIVERY_MODES))
|
| manifest["controls"]["decoders"] = json.loads(json.dumps(V4_DECODERS))
|
| for entry, decoder in zip(manifest["controls"]["target_buckets"]["buckets"], ("vocoder_bigvgan_8", "vocoder_bigvgan")):
|
| entry["vocoder_bigvgan"] = decoder
|
| manifest["controls"].update(controls or {})
|
| write("manifest.json", manifest)
|
| return bundle
|
|
|
|
|
| def _fake_engine(frames_per_phone: float = 1.0, *, apple: bool = False, v4: bool = False, **controls) -> Engine:
|
| tmp = tempfile.TemporaryDirectory()
|
| bundle = _fake_bundle(Path(tmp.name), apple=apple, controls=controls, v4=v4)
|
| graphs = _FakeGraphs(frames_per_phone)
|
| engine = Engine(bundle, session_factory=graphs.factory)
|
| engine._fake, engine._tmp = graphs, tmp
|
| return engine
|
|
|
|
|
| if __name__ == "__main__":
|
| unittest.main()
|
|
|
