Spaces:
Sleeping
Sleeping
File size: 3,321 Bytes
eb1a122 b8fa9bf eb1a122 b8fa9bf eb1a122 b8fa9bf eb1a122 b8fa9bf eb1a122 b8fa9bf eb1a122 b8fa9bf eb1a122 b8fa9bf eb1a122 | 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 | #!/usr/bin/env python3
"""Benchmark significant sample-extraction subprocesses using synthetic fixtures.
This intentionally defaults to `stem=all` so the DSP stages can be measured without
Demucs download/runtime noise. Use `--include-demucs` with a real input file if you
want to benchmark stem separation on the current machine.
"""
from __future__ import annotations
import argparse
import json
import statistics
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import soundfile as sf
from pipeline_runner import PipelineParams, run_extraction_pipeline
from sample_extractor import cache_clear
from synth_generator import generate_test_song
def run_case(pattern: str, bars: int, bpm: float, run_index: int, clustering_mode: str) -> dict:
tmp = Path(tempfile.mkdtemp(prefix="dse-bench-"))
song = generate_test_song(pattern_name=pattern, bars=bars, bpm=bpm, add_bass=False, seed=42 + run_index)
src = tmp / f"{pattern}-{bars}bars.wav"
sf.write(src, song.drums_only, song.sr)
cache_clear()
params = PipelineParams(stem="all", clustering_mode=clustering_mode, target_min=4, target_max=12, synthesize=True)
result = run_extraction_pipeline(src, tmp / "out", params)
return {
"pattern": pattern,
"bars": bars,
"bpm": bpm,
"run_index": run_index,
"clustering_mode": clustering_mode,
"audio_duration_sec": result.audio_duration_sec,
"total_duration_sec": result.duration_sec,
"realtime_factor": result.realtime_factor,
"hit_count": result.hit_count,
"cluster_count": result.cluster_count,
"stages": result.stages,
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--runs", type=int, default=2)
parser.add_argument("--bars", type=int, default=4)
parser.add_argument("--bpm", type=float, default=120.0)
parser.add_argument("--output", default="docs/benchmark-subprocesses.json")
parser.add_argument("--clustering-mode", choices=["batch_quality", "online_preview"], default="batch_quality")
args = parser.parse_args()
# Warm imports/JIT and discard the result.
run_case("rock", 1, args.bpm, -1, args.clustering_mode)
rows = []
for run_index in range(args.runs):
for pattern in ["rock", "funk", "halftime"]:
rows.append(run_case(pattern, args.bars, args.bpm, run_index, args.clustering_mode))
stage_keys = [stage["key"] for stage in rows[0]["stages"]]
summary = []
for key in stage_keys:
values = [next(stage for stage in row["stages"] if stage["key"] == key)["duration_sec"] for row in rows]
summary.append({
"stage": key,
"mean_sec": round(statistics.mean(values), 6),
"median_sec": round(statistics.median(values), 6),
"min_sec": round(min(values), 6),
"max_sec": round(max(values), 6),
})
payload = {"clustering_mode": args.clustering_mode, "runs": rows, "summary": summary}
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(payload, indent=2), encoding="utf-8")
print(json.dumps(payload, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|