Spaces:
Running
Running
File size: 4,818 Bytes
10bebcd | 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | #!/usr/bin/env python3
"""Score generated speech with speechmos DNSMOS.
This script intentionally keeps DNSMOS as an auxiliary non-intrusive proxy.
It requires the optional `speechmos` package and is not part of the default
test environment. Install it in a disposable target/venv before running.
"""
from __future__ import annotations
import argparse
import json
import statistics
import sys
from pathlib import Path
from typing import Any
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
source = parser.add_mutually_exclusive_group(required=True)
source.add_argument("--listing", help="wav_res_ref_text path")
source.add_argument("--wav_dir", help="Directory containing generated wav files")
parser.add_argument("--out_json", required=True)
parser.add_argument("--max_items", type=int, default=0)
return parser.parse_args()
def load_listing(path: Path, max_items: int = 0) -> list[Path]:
wavs: list[Path] = []
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
wavs.append(Path(line.split("|", 1)[0]))
if max_items > 0:
wavs = wavs[:max_items]
return wavs
def load_wav_dir(path: Path, max_items: int = 0) -> list[Path]:
wavs = sorted(path.glob("*.wav"))
if max_items > 0:
wavs = wavs[:max_items]
return wavs
def _mean(values: list[float]) -> float | None:
return statistics.mean(values) if values else None
def _std(values: list[float]) -> float | None:
if len(values) <= 1:
return 0.0 if values else None
return statistics.stdev(values)
def _as_float(value: Any) -> float:
return float(value.item() if hasattr(value, "item") else value)
def main() -> int:
args = parse_args()
if args.listing:
wavs = load_listing(Path(args.listing), args.max_items)
else:
wavs = load_wav_dir(Path(args.wav_dir), args.max_items)
print(f"[dnsmos] {len(wavs)} wavs", file=sys.stderr)
try:
import librosa
from speechmos import dnsmos
except Exception as exc: # pragma: no cover - optional dependency path
raise SystemExit(
"speechmos DNSMOS dependencies are missing. Install in a disposable "
"environment, for example: pip install --target /tmp/speechmos_target "
"speechmos==0.0.1.1 librosa==0.9.1 onnxruntime pandas tqdm"
) from exc
per_utt: list[dict[str, Any]] = []
n_skip = 0
for idx, wav in enumerate(wavs, start=1):
try:
audio, _ = librosa.load(str(wav), sr=16_000, mono=True)
# speechmos rejects samples outside [-1, 1]. Some generated wavs
# contain small overshoots, so clamp at metric input time instead
# of dropping the utterance.
audio = audio.clip(-1.0, 1.0)
scores = dnsmos.run(audio, 16_000, return_df=False, verbose=False)
row = {
"wav": str(wav),
"ovrl_mos": _as_float(scores["ovrl_mos"]),
"sig_mos": _as_float(scores["sig_mos"]),
"bak_mos": _as_float(scores["bak_mos"]),
"p808_mos": _as_float(scores["p808_mos"]),
}
per_utt.append(row)
except Exception as exc: # pragma: no cover - data-dependent path
n_skip += 1
print(f"[dnsmos-skip] {wav}: {exc}", file=sys.stderr)
if idx % 50 == 0 or idx == len(wavs):
print(f"[dnsmos] {idx}/{len(wavs)}", file=sys.stderr)
values = {
key: [float(row[key]) for row in per_utt]
for key in ("ovrl_mos", "sig_mos", "bak_mos", "p808_mos")
}
out = {
"n_items": len(wavs),
"n_scored": len(per_utt),
"n_skipped": n_skip,
"metric": "speechmos_dnsmos",
"sample_rate": 16_000,
"max_items": args.max_items,
"ovrl_mos_mean": _mean(values["ovrl_mos"]),
"ovrl_mos_std": _std(values["ovrl_mos"]),
"sig_mos_mean": _mean(values["sig_mos"]),
"sig_mos_std": _std(values["sig_mos"]),
"bak_mos_mean": _mean(values["bak_mos"]),
"bak_mos_std": _std(values["bak_mos"]),
"p808_mos_mean": _mean(values["p808_mos"]),
"p808_mos_std": _std(values["p808_mos"]),
}
out_path = Path(args.out_json)
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(json.dumps(out, indent=2, sort_keys=True) + "\n", encoding="utf-8")
(out_path.parent / f"{out_path.stem}_per_utt_dnsmos_objective.jsonl").write_text(
"\n".join(json.dumps(row, sort_keys=True) for row in per_utt) + "\n",
encoding="utf-8",
)
print(json.dumps(out, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|