112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
"""audio.cpp adapter for the Suite's unified Voice Changer node."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from typing import Any, Dict, Mapping
|
|
|
|
import torch
|
|
|
|
from engines.adapters.audio_cpp_adapter import AudioCppEngineAdapter
|
|
from utils.audio.processing import AudioProcessingUtils
|
|
|
|
|
|
def _advanced_options(config: Mapping[str, Any]) -> Dict[str, Any]:
|
|
value = config.get("advanced_options", config.get("request_options", {}))
|
|
if value in (None, ""):
|
|
return {}
|
|
if isinstance(value, str):
|
|
try:
|
|
value = json.loads(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"Invalid audio.cpp advanced JSON: {exc.msg}") from exc
|
|
if not isinstance(value, Mapping):
|
|
raise ValueError("audio.cpp advanced options must be a JSON object")
|
|
return dict(value)
|
|
|
|
|
|
def _materialize(audio: Mapping[str, Any], label: str) -> str:
|
|
waveform = audio.get("waveform")
|
|
sample_rate = audio.get("sample_rate")
|
|
if not torch.is_tensor(waveform):
|
|
raise TypeError(f"audio.cpp {label} must contain a waveform tensor")
|
|
if sample_rate is None or int(sample_rate) <= 0:
|
|
raise ValueError(f"audio.cpp {label} must contain a positive sample_rate")
|
|
return os.path.abspath(
|
|
AudioProcessingUtils.save_audio_to_temp_file(waveform, int(sample_rate))
|
|
)
|
|
|
|
|
|
class AudioCppVoiceConversionAdapter:
|
|
"""Convert source audio toward a target reference using an audio.cpp VC task."""
|
|
|
|
def __init__(self, config: Dict[str, Any]):
|
|
self.config = dict(config)
|
|
|
|
def _session_config(self) -> Dict[str, Any]:
|
|
config = dict(self.config)
|
|
if str(config.get("connection_mode", "auto")).lower() != "external_server":
|
|
config["requested_task"] = "vc"
|
|
config["task"] = "vc"
|
|
return config
|
|
|
|
def convert_voice(
|
|
self,
|
|
source_audio: Dict[str, Any],
|
|
target_audio: Dict[str, Any],
|
|
refinement_passes: int = 1,
|
|
) -> tuple[Dict[str, Any], str]:
|
|
from utils.audio_cpp.session import get_audio_cpp_session
|
|
|
|
config = self._session_config()
|
|
family = str(config.get("family", "")).strip()
|
|
passes = max(1, int(refinement_passes))
|
|
current = source_audio
|
|
output_rate = int(source_audio["sample_rate"])
|
|
|
|
session = get_audio_cpp_session(config)
|
|
if str(getattr(session, "task", "vc")) != "vc":
|
|
raise ValueError(
|
|
f"audio.cpp model '{session.model_id}' is configured for task "
|
|
f"'{session.task}', not voice conversion"
|
|
)
|
|
|
|
for pass_index in range(passes):
|
|
source_path = _materialize(current, "source audio")
|
|
target_path = _materialize(target_audio, "target reference audio")
|
|
try:
|
|
request = {
|
|
"audio": source_path,
|
|
"voice_ref": target_path,
|
|
"source_audio": source_path,
|
|
"target_voice": target_path,
|
|
"options": _advanced_options(config),
|
|
}
|
|
print(
|
|
f"🔄 audio.cpp VC: {family or 'external model'} pass "
|
|
f"{pass_index + 1}/{passes}..."
|
|
)
|
|
result = session.run(request)
|
|
waveform, output_rate = AudioCppEngineAdapter._normalize_result(result)
|
|
current = {"waveform": waveform.unsqueeze(0), "sample_rate": output_rate}
|
|
finally:
|
|
for path in (source_path, target_path):
|
|
try:
|
|
os.remove(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
info = (
|
|
f"Model family: {family or getattr(session, 'family', 'external')}\n"
|
|
f"Model ID: {session.model_id}\n"
|
|
f"Task: voice conversion\n"
|
|
f"Refinement passes: {passes}\n"
|
|
f"Output sample rate: {output_rate} Hz\n"
|
|
"Conversion completed successfully"
|
|
)
|
|
return current, info
|
|
|
|
|
|
__all__ = ["AudioCppVoiceConversionAdapter"]
|