mirror of
https://github.com/RVC-Project/Retrieval-based-Voice-Conversion-WebUI.git
synced 2026-08-29 10:09:32 +02:00
300 lines
10 KiB
Python
300 lines
10 KiB
Python
import argparse
|
||
import os
|
||
import sys
|
||
import warnings
|
||
from io import BytesIO
|
||
from pathlib import Path
|
||
|
||
|
||
warnings.filterwarnings(
|
||
"ignore",
|
||
message=r"`torch\.nn\.utils\.weight_norm` is deprecated.*",
|
||
category=FutureWarning,
|
||
)
|
||
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
os.chdir(PROJECT_ROOT)
|
||
os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")
|
||
os.environ.setdefault("weight_root", str(PROJECT_ROOT / "assets" / "weights"))
|
||
os.environ.setdefault("index_root", str(PROJECT_ROOT / "logs"))
|
||
os.environ.setdefault("outside_index_root", str(PROJECT_ROOT / "assets" / "indices"))
|
||
os.environ.setdefault("rmvpe_root", str(PROJECT_ROOT / "assets" / "rmvpe"))
|
||
|
||
AUDIO_EXTENSIONS = {
|
||
".wav",
|
||
".flac",
|
||
".mp3",
|
||
".m4a",
|
||
".ogg",
|
||
".opus",
|
||
".aac",
|
||
".wma",
|
||
".mp4",
|
||
".mkv",
|
||
".webm",
|
||
}
|
||
OUTPUT_FORMATS = {"wav", "flac", "mp3", "m4a"}
|
||
|
||
|
||
def build_parser():
|
||
parser = argparse.ArgumentParser(
|
||
description="Offline RVC inference for single-speaker and multi-speaker models."
|
||
)
|
||
parser.add_argument("--model", required=True, help="Model filename or .pth path.")
|
||
parser.add_argument("--input", help="Input audio file or directory.")
|
||
parser.add_argument("--output", help="Output audio file or directory.")
|
||
parser.add_argument(
|
||
"--speaker-id",
|
||
type=int,
|
||
help="Speaker ID. Multi-speaker models default to their smallest declared ID.",
|
||
)
|
||
parser.add_argument(
|
||
"--list-speakers",
|
||
action="store_true",
|
||
help="List the model's speaker metadata and exit.",
|
||
)
|
||
parser.add_argument("--pitch", type=int, default=0, help="Pitch shift in semitones.")
|
||
parser.add_argument(
|
||
"--f0-method", choices=["pm", "rmvpe"], default="rmvpe"
|
||
)
|
||
parser.add_argument(
|
||
"--index",
|
||
help="Explicit added index path. The matching speaker index is selected automatically when omitted.",
|
||
)
|
||
parser.add_argument("--index-rate", type=float, default=0.75)
|
||
parser.add_argument("--resample-sr", type=int, default=0)
|
||
parser.add_argument("--rms-mix-rate", type=float, default=1.0)
|
||
parser.add_argument("--protect", type=float, default=0.33)
|
||
parser.add_argument(
|
||
"--format", dest="output_format", choices=sorted(OUTPUT_FORMATS)
|
||
)
|
||
parser.add_argument(
|
||
"--recursive", action="store_true", help="Scan input subdirectories."
|
||
)
|
||
parser.add_argument("--overwrite", action="store_true")
|
||
return parser
|
||
|
||
|
||
def resolve_model(value):
|
||
candidate = Path(value).expanduser()
|
||
if not candidate.is_absolute():
|
||
candidate = (PROJECT_ROOT / candidate).resolve()
|
||
if not candidate.is_file():
|
||
candidate = (Path(os.environ["weight_root"]) / value).resolve()
|
||
if not candidate.is_file():
|
||
raise FileNotFoundError("Model not found: %s" % value)
|
||
if candidate.suffix.lower() != ".pth":
|
||
raise ValueError("Model must be a .pth file: %s" % candidate)
|
||
return candidate
|
||
|
||
|
||
def load_model_metadata(model_path):
|
||
import torch
|
||
|
||
checkpoint = torch.load(str(model_path), map_location="cpu")
|
||
weight = checkpoint.get("weight", {}) if isinstance(checkpoint, dict) else {}
|
||
embedding = weight.get("emb_g.weight")
|
||
if embedding is None:
|
||
raise ValueError("Model does not contain weight/emb_g.weight")
|
||
speaker_count = int(embedding.shape[0])
|
||
speakers = []
|
||
seen = set()
|
||
for item in checkpoint.get("speaker_info", []):
|
||
try:
|
||
speaker_id = int(item["id"])
|
||
speaker_name = str(item["name"])
|
||
except (KeyError, TypeError, ValueError):
|
||
continue
|
||
if (
|
||
0 <= speaker_id < speaker_count
|
||
and speaker_name
|
||
and speaker_id not in seen
|
||
):
|
||
speakers.append({"id": speaker_id, "name": speaker_name})
|
||
seen.add(speaker_id)
|
||
speakers.sort(key=lambda item: item["id"])
|
||
return speaker_count, speakers
|
||
|
||
|
||
def select_speaker(speaker_count, speakers, requested_id):
|
||
if speakers:
|
||
valid_ids = {item["id"] for item in speakers}
|
||
speaker_id = speakers[0]["id"] if requested_id is None else requested_id
|
||
if speaker_id not in valid_ids:
|
||
raise ValueError(
|
||
"Speaker ID %s is not declared by this multi-speaker model; available IDs: %s"
|
||
% (speaker_id, ", ".join(str(value) for value in sorted(valid_ids)))
|
||
)
|
||
return speaker_id
|
||
speaker_id = 0 if requested_id is None else requested_id
|
||
if speaker_id < 0 or speaker_id >= speaker_count:
|
||
raise ValueError(
|
||
"Speaker ID must be between 0 and %s for this model" % (speaker_count - 1)
|
||
)
|
||
return speaker_id
|
||
|
||
|
||
def resolve_index(value, model_name, speaker_id, index_rate):
|
||
from infer.vc.utils import get_index_path_from_model
|
||
|
||
if index_rate == 0:
|
||
return ""
|
||
if value:
|
||
candidate = Path(value).expanduser()
|
||
if not candidate.is_absolute():
|
||
candidate = (PROJECT_ROOT / candidate).resolve()
|
||
if "trained" in candidate.name:
|
||
candidate = candidate.with_name(candidate.name.replace("trained", "added"))
|
||
index_path = str(candidate)
|
||
else:
|
||
index_path = get_index_path_from_model(model_name, speaker_id)
|
||
if not index_path or not Path(index_path).is_file():
|
||
raise FileNotFoundError(
|
||
"No usable added index for speaker ID %s. Supply --index or use --index-rate 0."
|
||
% speaker_id
|
||
)
|
||
return str(Path(index_path).resolve())
|
||
|
||
|
||
def collect_jobs(input_value, output_value, output_format, recursive):
|
||
input_path = Path(input_value).expanduser().resolve()
|
||
output_path = Path(output_value).expanduser().resolve()
|
||
if not input_path.exists():
|
||
raise FileNotFoundError("Input not found: %s" % input_path)
|
||
|
||
if input_path.is_file():
|
||
suffix_format = output_path.suffix.lower().lstrip(".")
|
||
if output_path.suffix and suffix_format not in OUTPUT_FORMATS:
|
||
raise ValueError("Unsupported output suffix: %s" % output_path.suffix)
|
||
selected_format = output_format or suffix_format or "wav"
|
||
if output_path.suffix:
|
||
output_file = output_path.with_suffix(".%s" % selected_format)
|
||
else:
|
||
output_path.mkdir(parents=True, exist_ok=True)
|
||
output_file = output_path / (input_path.stem + "." + selected_format)
|
||
return [(input_path, output_file, selected_format)]
|
||
|
||
selected_format = output_format or "wav"
|
||
output_path.mkdir(parents=True, exist_ok=True)
|
||
iterator = input_path.rglob("*") if recursive else input_path.iterdir()
|
||
jobs = []
|
||
for path in sorted(iterator):
|
||
if not path.is_file() or path.suffix.lower() not in AUDIO_EXTENSIONS:
|
||
continue
|
||
relative_parent = path.relative_to(input_path).parent if recursive else Path()
|
||
output_file = output_path / relative_parent / (path.stem + "." + selected_format)
|
||
jobs.append((path, output_file, selected_format))
|
||
if not jobs:
|
||
raise ValueError("No supported audio files found in: %s" % input_path)
|
||
return jobs
|
||
|
||
|
||
def write_audio(path, audio, sample_rate, output_format):
|
||
import soundfile as sf
|
||
from infer.audio import wav2
|
||
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
if output_format in ("wav", "flac"):
|
||
sf.write(str(path), audio, sample_rate)
|
||
return
|
||
with BytesIO() as wav_file:
|
||
sf.write(wav_file, audio, sample_rate, format="wav")
|
||
wav_file.seek(0)
|
||
with open(path, "wb") as output_file:
|
||
wav2(wav_file, output_file, output_format)
|
||
|
||
|
||
def create_config():
|
||
from configs.config import Config
|
||
|
||
original_argv = sys.argv[:]
|
||
sys.argv = [sys.argv[0]]
|
||
try:
|
||
return Config()
|
||
finally:
|
||
sys.argv = original_argv
|
||
|
||
|
||
def main(argv=None):
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
if not args.list_speakers and (not args.input or not args.output):
|
||
parser.error("--input and --output are required unless --list-speakers is used")
|
||
if not 0 <= args.index_rate <= 1:
|
||
parser.error("--index-rate must be between 0 and 1")
|
||
if not 0 <= args.rms_mix_rate <= 1:
|
||
parser.error("--rms-mix-rate must be between 0 and 1")
|
||
if not 0 <= args.protect <= 0.5:
|
||
parser.error("--protect must be between 0 and 0.5")
|
||
if args.resample_sr and args.resample_sr < 16000:
|
||
parser.error("--resample-sr must be 0 or at least 16000")
|
||
|
||
model_path = resolve_model(args.model)
|
||
os.environ["weight_root"] = str(model_path.parent)
|
||
model_name = model_path.name
|
||
speaker_count, speakers = load_model_metadata(model_path)
|
||
if args.list_speakers:
|
||
if speakers:
|
||
for item in speakers:
|
||
print("%s\t%s" % (item["id"], item["name"]))
|
||
else:
|
||
print("0-%s" % (speaker_count - 1))
|
||
return 0
|
||
|
||
speaker_id = select_speaker(speaker_count, speakers, args.speaker_id)
|
||
index_path = resolve_index(
|
||
args.index, model_name, speaker_id, args.index_rate
|
||
)
|
||
jobs = collect_jobs(
|
||
args.input, args.output, args.output_format, args.recursive
|
||
)
|
||
existing = [str(output) for _, output, _ in jobs if output.exists()]
|
||
if existing and not args.overwrite:
|
||
raise FileExistsError(
|
||
"Output already exists; use --overwrite: %s" % existing[0]
|
||
)
|
||
|
||
from i18n.i18n import I18nAuto
|
||
from infer.vc.modules import VC
|
||
|
||
i18n = I18nAuto()
|
||
config = create_config()
|
||
print(i18n("当前设备：%s | 推理精度：%s") % (config.device, config.dtype))
|
||
print("%s: %s" % (i18n("选择模型"), model_name))
|
||
print("%s: %s" % (i18n("说话人ID（0~109）"), speaker_id))
|
||
print("%s: %s" % (i18n("选择索引"), index_path or i18n("未使用")))
|
||
|
||
vc = VC(config)
|
||
vc.get_vc(model_name)
|
||
failed = 0
|
||
for input_path, output_path, output_format in jobs:
|
||
status, result = vc.vc_single(
|
||
speaker_id,
|
||
str(input_path),
|
||
args.pitch,
|
||
args.f0_method,
|
||
index_path,
|
||
args.index_rate,
|
||
args.resample_sr,
|
||
args.rms_mix_rate,
|
||
args.protect,
|
||
)
|
||
print(status)
|
||
if not result or result[0] is None or result[1] is None:
|
||
failed += 1
|
||
continue
|
||
write_audio(output_path, result[1], result[0], output_format)
|
||
print(str(output_path))
|
||
return 1 if failed else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
raise SystemExit(main())
|
||
except KeyboardInterrupt:
|
||
raise SystemExit(130)
|
||
except Exception as error:
|
||
print("rvc-cli: error: %s" % error, file=sys.stderr)
|
||
raise SystemExit(1)
|