File size: 5,262 Bytes
7e8f83f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | from __future__ import annotations
import base64
from contextlib import redirect_stdout
import io
import json
import math
from pathlib import Path
import sys
def load_reference_audio(path):
# torchaudio.load on the base requires TorchCodec. Decode this local input
# with libsndfile instead; torchaudio functional resampling remains intact.
import soundfile as sf
import torch
info = sf.info(str(path))
if not 0 < info.duration <= 120 or not 1 <= info.channels <= 2 or not 8000 <= info.samplerate <= 192000:
raise ValueError('Reference audio must be 0-120 seconds, mono/stereo, 8-192 kHz')
audio, rate = sf.read(str(path), dtype='float32', always_2d=True)
if not len(audio) or not rate or not torch.isfinite(torch.from_numpy(audio)).all():
raise ValueError('Reference audio is empty or invalid')
return torch.from_numpy(audio.T.copy()), int(rate)
def synthesize(data):
import soundfile as sf
import torch
sys.path.insert(0, '/opt/irodori-tts')
from irodori_tts import inference_runtime as runtime_module
runtime_module._load_audio = load_reference_audio
runtime_module._resolve_tokenizer_source = lambda checkpoint, fallback: (data['tokenizer'], True)
checkpoint = Path(data['checkpoint'])
if not checkpoint.is_file():
raise ValueError('Irodori checkpoint does not exist')
reference = data.get('reference_wav')
references = data.get('reference_wavs')
if reference and references:
raise ValueError('Use one reference representation')
paths = references if references is not None else ([reference] if reference else [])
if not isinstance(paths, list) or len(paths) > 20 or any(not isinstance(path, str) or not path for path in paths):
raise ValueError('Invalid reference audio paths')
total_seconds = 0
for path in paths:
info = sf.info(path)
if not 0 < info.duration <= 120 or not 1 <= info.channels <= 2 or not 8000 <= info.samplerate <= 192000:
raise ValueError('Reference audio must be 0-120 seconds, mono/stereo, 8-192 kHz')
total_seconds += info.duration
if total_seconds > 120:
raise ValueError('Reference audio must be at most 120 seconds combined')
decoded = {str(path): load_reference_audio(path) for path in paths}
if decoded:
runtime_module._load_audio = lambda path: decoded[str(path)] if str(path) in decoded else load_reference_audio(path)
has_reference = bool(paths)
caption = data.get('caption') or None
runtime, reloaded = runtime_module.get_cached_runtime(runtime_module.RuntimeKey(
checkpoint=str(checkpoint), model_device='cuda', model_precision='bf16',
codec_repo=data['codec'], codec_device='cuda',
codec_precision='fp32', codec_deterministic_encode=True,
codec_deterministic_decode=True, compile_model=False, compile_dynamic=False,
))
if caption and not runtime.model_cfg.use_caption_condition:
raise ValueError('Selected model does not support voice descriptions')
if has_reference and not runtime.model_cfg.use_speaker_condition_resolved:
raise ValueError('Selected model does not support reference audio')
text_cfg, caption_cfg, speaker_cfg, messages = runtime_module.resolve_cfg_scales(
cfg_guidance_mode='independent', cfg_scale_text=data['cfg_scale_text'],
cfg_scale_caption=data['cfg_scale_caption'] if caption else 0,
cfg_scale_speaker=data['cfg_scale_speaker'] if has_reference else 0,
cfg_scale=None, use_caption_condition=bool(caption), use_speaker_condition=has_reference,
)
request = runtime_module.SamplingRequest(
text=data['text'], caption=caption, ref_wav=reference, ref_wavs=references, no_ref=not has_reference,
seconds=data.get('seconds'), duration_scale=data['duration_scale'],
min_seconds=0.5, max_seconds=45, max_ref_seconds=120,
num_candidates=1, decode_mode='sequential', num_steps=data['num_steps'],
seed=data['seed'], cfg_guidance_mode='independent', cfg_scale_text=text_cfg,
cfg_scale_caption=caption_cfg, cfg_scale_speaker=speaker_cfg,
context_kv_cache=True, trim_tail=True,
)
result = runtime.synthesize(request)
audio = result.audio.detach().float().cpu().squeeze()
if audio.ndim != 1 or not audio.numel() or not torch.isfinite(audio).all():
raise ValueError('Irodori generated invalid audio')
buffer = io.BytesIO()
sf.write(buffer, audio.numpy(), int(result.sample_rate), format='WAV', subtype='PCM_16')
Path(data['output']).write_bytes(buffer.getvalue())
return {
'backend': 'irodori',
'audio_mime': 'audio/wav', 'sample_rate': int(result.sample_rate),
'duration_seconds': audio.numel() / int(result.sample_rate),
'seed': int(result.used_seed), 'num_steps': data['num_steps'],
'model': checkpoint.name, 'runtime_reloaded': bool(reloaded),
'messages': [*messages, *result.messages],
}
if __name__ == '__main__':
try:
with redirect_stdout(sys.stderr):
result = synthesize(json.load(sys.stdin))
except Exception as exc:
result = {'error': str(exc)}
json.dump(result, sys.stdout, ensure_ascii=False)
|