#!/usr/bin/env python # -*- coding: utf-8 -*- """Generate a short A/B diagnostic matrix for an existing cover session.""" from __future__ import annotations import argparse import json import re import sys from pathlib import Path from typing import Dict, Iterable, List, Optional, Sequence, Tuple import numpy as np import soundfile as sf REPO_ROOT = Path(__file__).resolve().parent.parent if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from infer.cover_pipeline import CoverPipeline from infer.official_adapter import convert_vocals_official, convert_vocals_official_upstream def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Run several short VC variants on an existing session and export " "per-clip WAVs plus summary.json for side-by-side diagnosis." ) ) parser.add_argument("--session-dir", required=True, help="Session directory under temp/cover/") parser.add_argument( "--model-path", help=( "RVC .pth path. If omitted, try to infer from " "outputs/__all_files_." ), ) parser.add_argument("--index-path", help="Optional .index path. If omitted, try best-effort matching.") parser.add_argument("--output-dir", help="Where to write diagnostic_matrix outputs.") parser.add_argument("--stage", default="vc_final_state", help="Preferred stage in quality_debug.json for suspect times.") parser.add_argument("--times", help="Comma-separated times in seconds. If set, skip automatic time selection.") parser.add_argument("--max-clips", type=int, default=4, help="Maximum suspect clips to export.") parser.add_argument("--window-sec", type=float, default=1.8, help="Window size around each suspect time.") parser.add_argument("--gap-sec", type=float, default=0.35, help="Silence gap inserted between concatenated clips.") parser.add_argument( "--variants", default="existing_raw,existing_final,upstream_raw,upstream_post,upstream_index0,official_repair", help=( "Comma-separated variants. Supported: " "existing_raw, existing_final, upstream_raw, upstream_post, " "upstream_index0, official_repair" ), ) parser.add_argument("--f0-method", default="hybrid") parser.add_argument("--pitch-shift", type=int, default=0) parser.add_argument("--index-rate", type=float, default=0.5) parser.add_argument("--filter-radius", type=int, default=3) parser.add_argument("--rms-mix-rate", type=float, default=0.0) parser.add_argument("--protect", type=float, default=0.33) parser.add_argument("--speaker-id", type=int, default=0) parser.add_argument("--source-constraint-mode", default="auto") return parser.parse_args() def _normalize_name(text: str) -> str: return re.sub(r"[^a-z0-9]+", "", str(text or "").lower()) def _tokenize_name(text: str) -> List[str]: return [token for token in re.split(r"[^a-z0-9]+", str(text or "").lower()) if len(token) >= 2] def _load_quality_debug(session_dir: Path) -> Dict[str, object]: report_path = session_dir / "quality_debug.json" if not report_path.exists(): return {} with open(report_path, "r", encoding="utf-8") as f: return json.load(f) def _get_stage_entry(entries: Sequence[Dict[str, object]], stage: str) -> Optional[Dict[str, object]]: for entry in reversed(entries): if entry.get("stage") == stage: return entry return None def _dedupe_times(times: Iterable[float], min_delta_sec: float = 0.18) -> List[float]: ordered = sorted(float(t) for t in times if float(t) >= 0.0) deduped: List[float] = [] for time_sec in ordered: if not deduped or abs(time_sec - deduped[-1]) >= min_delta_sec: deduped.append(round(time_sec, 3)) return deduped def _collect_suspect_times( session_dir: Path, preferred_stage: str, max_clips: int, ) -> Tuple[List[float], Dict[str, object]]: payload = _load_quality_debug(session_dir) entries = payload.get("stages", []) if not isinstance(entries, list): entries = [] stage_order = [ preferred_stage, "vc_final_state", "vc_refined", "vc_source_constrained", "vc_raw", ] collected: List[float] = [] for stage in stage_order: entry = _get_stage_entry(entries, stage) if not entry: continue analysis = entry.get("analysis", {}) if not isinstance(analysis, dict): continue times = analysis.get("transition_spike_times_sec", []) if isinstance(times, list): collected.extend(float(t) for t in times[:12]) collected = _dedupe_times(collected) if len(collected) >= max_clips: break return collected[: max(1, int(max_clips))], payload if isinstance(payload, dict) else {} def _choose_model_file(folder: Path) -> Path: candidates = sorted(folder.glob("*.pth")) if not candidates: raise FileNotFoundError(f"No .pth model found in {folder}") if len(candidates) == 1: return candidates[0] folder_norm = _normalize_name(folder.name) best_path = candidates[0] best_score = -1 for path in candidates: stem_norm = _normalize_name(path.stem) score = 0 if stem_norm == folder_norm: score += 1000 if folder_norm and (folder_norm in stem_norm or stem_norm in folder_norm): score += 300 score += len(set(_tokenize_name(folder.name)) & set(_tokenize_name(path.stem))) * 40 if score > best_score: best_score = score best_path = path return best_path def _resolve_index_for_model(model_path: Path, explicit_index: Optional[str] = None) -> Optional[Path]: if explicit_index: explicit = Path(explicit_index) if explicit.exists(): return explicit candidates = list(model_path.parent.glob("*.index")) official_indexes = REPO_ROOT / "assets" / "weights" / "official_indexes" if official_indexes.exists(): candidates.extend(list(official_indexes.glob("*.index"))) if not candidates: return None if len(candidates) == 1: return candidates[0] model_norm = _normalize_name(model_path.stem) model_tokens = set(_tokenize_name(model_path.stem)) best_match = None best_score = -1 for candidate in candidates: idx_norm = _normalize_name(candidate.stem) idx_tokens = set(_tokenize_name(candidate.stem)) score = 0 if idx_norm == model_norm: score += 1000 if model_norm and (model_norm in idx_norm or idx_norm in model_norm): score += 300 score += len(model_tokens & idx_tokens) * 40 if "added" in candidate.stem.lower(): score += 10 if score > best_score: best_score = score best_match = candidate return best_match if best_score > 0 else None def _infer_model_from_session(session_dir: Path) -> Tuple[Path, Optional[Path], Optional[str]]: session_id = session_dir.name outputs_root = REPO_ROOT / "outputs" characters_root = REPO_ROOT / "assets" / "weights" / "characters" if not outputs_root.exists() or not characters_root.exists(): raise FileNotFoundError("Cannot infer model because outputs/ or assets/weights/characters/ is missing.") output_matches = sorted(outputs_root.glob(f"*_all_files_{session_id}")) if not output_matches: raise FileNotFoundError("Could not infer model from outputs/. Pass --model-path explicitly.") character_dirs = [path for path in characters_root.iterdir() if path.is_dir()] best_dir = None best_score = -1 for output_dir in output_matches: prefix = output_dir.name[: -len(f"_all_files_{session_id}")] for character_dir in character_dirs: char_name = character_dir.name if prefix == char_name or prefix.endswith(f"_{char_name}"): score = len(char_name) if score > best_score: best_score = score best_dir = character_dir if best_dir is None: raise FileNotFoundError( "Could not match the session output folder to a character model folder. " "Pass --model-path explicitly." ) model_path = _choose_model_file(best_dir) index_path = _resolve_index_for_model(model_path) return model_path, index_path, best_dir.name def _extract_bundle( source_path: Path, output_path: Path, times_sec: Sequence[float], window_sec: float, gap_sec: float, ) -> List[Dict[str, float]]: audio, sr = sf.read(str(source_path), always_2d=True) audio = np.asarray(audio, dtype=np.float32) if audio.shape[0] <= 0: raise ValueError(f"Empty audio: {source_path}") duration_sec = float(audio.shape[0] / max(sr, 1)) bundle_parts: List[np.ndarray] = [] manifest: List[Dict[str, float]] = [] cursor_sec = 0.0 gap_samples = int(round(max(0.0, gap_sec) * sr)) target_window = max(0.2, float(window_sec)) for index, time_sec in enumerate(times_sec): clip_start_sec = 0.0 if duration_sec > target_window: clip_start_sec = min(max(0.0, float(time_sec) - target_window / 2.0), duration_sec - target_window) clip_end_sec = min(duration_sec, clip_start_sec + target_window) start_sample = int(round(clip_start_sec * sr)) end_sample = max(start_sample + 1, int(round(clip_end_sec * sr))) clip = audio[start_sample:end_sample] if clip.size == 0: continue actual_duration = float(clip.shape[0] / sr) bundle_parts.append(clip.astype(np.float32)) manifest.append( { "clip_id": index, "source_time_sec": round(float(time_sec), 3), "clip_start_sec": round(float(clip_start_sec), 6), "clip_end_sec": round(float(clip_end_sec), 6), "bundle_start_sec": round(float(cursor_sec), 6), "bundle_end_sec": round(float(cursor_sec + actual_duration), 6), "duration_sec": round(actual_duration, 6), } ) cursor_sec += actual_duration if gap_samples > 0 and index < len(times_sec) - 1: bundle_parts.append(np.zeros((gap_samples, audio.shape[1]), dtype=np.float32)) cursor_sec += float(gap_samples / sr) if not bundle_parts: raise ValueError(f"No clips were extracted from {source_path}") output_path.parent.mkdir(parents=True, exist_ok=True) bundle_audio = np.concatenate(bundle_parts, axis=0).astype(np.float32) sf.write(str(output_path), bundle_audio, sr) return manifest def _split_bundle( bundle_path: Path, manifest: Sequence[Dict[str, float]], output_dir: Path, prefix: str, ) -> Dict[str, str]: audio, sr = sf.read(str(bundle_path), always_2d=True) audio = np.asarray(audio, dtype=np.float32) output_dir.mkdir(parents=True, exist_ok=True) exported: Dict[str, str] = {} for segment in manifest: bundle_start_sec = float(segment["bundle_start_sec"]) bundle_end_sec = float(segment["bundle_end_sec"]) start_sample = max(0, int(round(bundle_start_sec * sr))) end_sample = min(audio.shape[0], max(start_sample + 1, int(round(bundle_end_sec * sr)))) if end_sample <= start_sample: continue clip = audio[start_sample:end_sample].astype(np.float32) clip_key = f"{float(segment['source_time_sec']):07.3f}s" output_path = output_dir / f"{prefix}_{clip_key}.wav" sf.write(str(output_path), clip, sr) exported[clip_key] = str(output_path) return exported def _estimate_artifact_score(analysis: Dict[str, object]) -> float: quiet_rms = max(0.0, float(analysis.get("quiet_rms_ratio", 1.0) or 1.0) - 1.0) quiet_hf = max(0.0, float(analysis.get("quiet_hf_ratio", 1.0) or 1.0) - 1.0) midquiet_hf = max(0.0, float(analysis.get("midquiet_hf_ratio", 1.0) or 1.0) - 1.0) spike = max(0.0, float(analysis.get("transition_spike_ratio", 1.0) or 1.0) - 1.0) active = max(0.0, float(analysis.get("active_rms_ratio", 1.0) or 1.0) - 1.5) breaths = max(0.0, float(analysis.get("synthetic_breath_frames", 0) or 0) - 80.0) / 220.0 corr_penalty = max(0.0, 0.85 - float(analysis.get("corr_active", 0.0) or 0.0)) return float( 0.22 * quiet_rms + 0.28 * quiet_hf + 0.08 * midquiet_hf + 0.22 * spike + 0.05 * active + 0.10 * breaths + 0.05 * corr_penalty ) def _aggregate_variant_metrics(analyses: Sequence[Dict[str, object]]) -> Dict[str, object]: if not analyses: return {"clips": 0} keys = [ "active_rms_ratio", "quiet_rms_ratio", "quiet_hf_ratio", "midquiet_hf_ratio", "transition_spike_ratio", "corr_active", "synthetic_breath_frames", ] aggregate: Dict[str, object] = {"clips": len(analyses)} for key in keys: values = [float(item.get(key, 0.0) or 0.0) for item in analyses] aggregate[f"avg_{key}"] = float(sum(values) / max(len(values), 1)) aggregate["avg_artifact_score_estimate"] = float( sum(_estimate_artifact_score(item) for item in analyses) / max(len(analyses), 1) ) return aggregate def _preprocess_mode_from_payload(payload: Dict[str, object]) -> str: entries = payload.get("stages", []) if not isinstance(entries, list): return "direct" vc_input = _get_stage_entry(entries, "vc_input") if not vc_input: return "direct" mode = vc_input.get("preprocess_mode") if isinstance(mode, str) and mode.strip(): return mode.strip() extra = vc_input.get("extra", {}) if isinstance(extra, dict): mode = extra.get("effective_preprocess_mode") if isinstance(mode, str) and mode.strip(): return mode.strip() return "direct" def _run_upstream_bundle( source_bundle: Path, output_bundle: Path, model_path: Path, index_path: Optional[Path], args: argparse.Namespace, index_rate: Optional[float] = None, ) -> None: convert_vocals_official_upstream( vocals_path=str(source_bundle), output_path=str(output_bundle), model_path=str(model_path), index_path=str(index_path) if index_path else None, f0_method=args.f0_method, pitch_shift=args.pitch_shift, index_rate=float(args.index_rate if index_rate is None else index_rate), filter_radius=args.filter_radius, rms_mix_rate=args.rms_mix_rate, protect=args.protect, speaker_id=args.speaker_id, ) def _run_variant( variant: str, output_dir: Path, manifest: Sequence[Dict[str, float]], source_bundle: Path, lead_bundle: Optional[Path], session_dir: Path, model_path: Path, index_path: Optional[Path], args: argparse.Namespace, pipeline: CoverPipeline, ) -> Dict[str, object]: variant_dir = output_dir / variant variant_dir.mkdir(parents=True, exist_ok=True) bundle_path = variant_dir / "bundle.wav" result: Dict[str, object] = {"name": variant, "bundle_path": str(bundle_path)} clip_times = [float(item["source_time_sec"]) for item in manifest] if variant == "existing_raw": existing_path = session_dir / "debug_converted_raw.wav" if not existing_path.exists(): return {"name": variant, "skipped": "missing debug_converted_raw.wav"} _extract_bundle(existing_path, bundle_path, clip_times, args.window_sec, args.gap_sec) elif variant == "existing_final": existing_path = session_dir / "converted_vocals.wav" if not existing_path.exists(): return {"name": variant, "skipped": "missing converted_vocals.wav"} _extract_bundle(existing_path, bundle_path, clip_times, args.window_sec, args.gap_sec) elif variant == "upstream_raw": _run_upstream_bundle(source_bundle, bundle_path, model_path, index_path, args) elif variant == "upstream_index0": _run_upstream_bundle(source_bundle, bundle_path, model_path, index_path, args, index_rate=0.0) elif variant == "upstream_post": raw_bundle_path = variant_dir / "bundle_raw.wav" _run_upstream_bundle(source_bundle, raw_bundle_path, model_path, index_path, args) should_apply = pipeline._should_apply_source_constraint( vc_preprocessed=True, source_constraint_mode=args.source_constraint_mode, ) if should_apply: pipeline._constrain_converted_to_source( source_vocals_path=str(source_bundle), converted_vocals_path=str(raw_bundle_path), original_vocals_path=str(lead_bundle) if lead_bundle else None, output_path=str(bundle_path), ) pipeline._refine_source_constrained_output( source_vocals_path=str(source_bundle), converted_vocals_path=str(bundle_path), source_constraint_mode=args.source_constraint_mode, f0_method=args.f0_method, ) else: bundle_path.write_bytes(raw_bundle_path.read_bytes()) result["note"] = ( "postprocess skipped because current preprocess/source-constraint " "combination would not apply it" ) elif variant == "official_repair": saved_argv = sys.argv[:] try: sys.argv = [sys.argv[0]] convert_vocals_official( vocals_path=str(source_bundle), output_path=str(bundle_path), model_path=str(model_path), index_path=str(index_path) if index_path else None, f0_method=args.f0_method, pitch_shift=args.pitch_shift, index_rate=float(args.index_rate), filter_radius=args.filter_radius, rms_mix_rate=args.rms_mix_rate, protect=args.protect, speaker_id=args.speaker_id, repair_profile=True, ) finally: sys.argv = saved_argv else: return {"name": variant, "skipped": f"unknown variant: {variant}"} clips_dir = variant_dir / "clips" result["clips"] = _split_bundle(bundle_path, manifest, clips_dir, "candidate") return result def main() -> int: args = parse_args() session_dir = Path(args.session_dir) if not session_dir.is_absolute(): session_dir = (REPO_ROOT / session_dir).resolve() if not session_dir.exists(): raise FileNotFoundError(f"Session dir not found: {session_dir}") output_dir = Path(args.output_dir) if args.output_dir else session_dir / "diagnostic_matrix" if not output_dir.is_absolute(): output_dir = (REPO_ROOT / output_dir).resolve() output_dir.mkdir(parents=True, exist_ok=True) if args.times: times_sec = _dedupe_times(float(item.strip()) for item in args.times.split(",") if item.strip()) payload = _load_quality_debug(session_dir) else: times_sec, payload = _collect_suspect_times(session_dir, args.stage, args.max_clips) if not times_sec: raise ValueError( "No suspect times found. Pass --times explicitly, for example " '--times "22.48,23.04,24.17".' ) inferred_model_key = None if args.model_path: model_path = Path(args.model_path) if not model_path.is_absolute(): model_path = (REPO_ROOT / model_path).resolve() else: model_path, inferred_index_path, inferred_model_key = _infer_model_from_session(session_dir) if not args.index_path and inferred_index_path is not None: args.index_path = str(inferred_index_path) if not model_path.exists(): raise FileNotFoundError(f"Model path not found: {model_path}") index_path = _resolve_index_for_model(model_path, args.index_path) preprocess_mode = _preprocess_mode_from_payload(payload) source_full_path = session_dir / "vocals_for_vc.wav" if not source_full_path.exists(): raise FileNotFoundError(f"Missing source vocals for VC: {source_full_path}") lead_full_path = session_dir / "karaoke" / "lead_vocals.wav" if not lead_full_path.exists(): fallback_lead = session_dir / "lead_vocals.wav" lead_full_path = fallback_lead if fallback_lead.exists() else source_full_path reference_dir = output_dir / "reference" source_bundle = reference_dir / "vocals_for_vc_bundle.wav" lead_bundle = reference_dir / "lead_vocals_bundle.wav" manifest = _extract_bundle(source_full_path, source_bundle, times_sec, args.window_sec, args.gap_sec) _extract_bundle(lead_full_path, lead_bundle, times_sec, args.window_sec, args.gap_sec) reference_clips = _split_bundle(source_bundle, manifest, reference_dir / "clips", "reference") pipeline = CoverPipeline(device="cuda") pipeline._last_vc_preprocess_mode = preprocess_mode requested_variants = [item.strip() for item in args.variants.split(",") if item.strip()] variant_results: Dict[str, Dict[str, object]] = {} per_clip_results: List[Dict[str, object]] = [] aggregate_results: Dict[str, Dict[str, object]] = {} print(f"Session: {session_dir}") print(f"Output: {output_dir}") print(f"Model: {model_path}") print(f"Index: {index_path if index_path else '(none)'}") print(f"Preprocess mode from session: {preprocess_mode}") print("Suspect times:", ", ".join(f"{time_sec:.3f}s" for time_sec in times_sec)) for variant in requested_variants: print(f"[diagnose] running {variant} ...") variant_result = _run_variant( variant=variant, output_dir=output_dir, manifest=manifest, source_bundle=source_bundle, lead_bundle=lead_bundle, session_dir=session_dir, model_path=model_path, index_path=index_path, args=args, pipeline=pipeline, ) variant_results[variant] = variant_result for segment in manifest: clip_key = f"{float(segment['source_time_sec']):07.3f}s" clip_summary: Dict[str, object] = { "time_sec": float(segment["source_time_sec"]), "reference_clip": reference_clips.get(clip_key), "variants": {}, } for variant in requested_variants: variant_result = variant_results.get(variant, {}) variant_clips = variant_result.get("clips", {}) if isinstance(variant_result, dict) else {} if not isinstance(variant_clips, dict): continue candidate_path = variant_clips.get(clip_key) reference_path = reference_clips.get(clip_key) if not candidate_path or not reference_path: continue analysis = CoverPipeline._analyze_quality_stage(candidate_path=candidate_path, reference_path=reference_path) analysis["artifact_score_estimate"] = _estimate_artifact_score(analysis) clip_summary["variants"][variant] = { "path": candidate_path, "analysis": analysis, } ranking = [] variants_dict = clip_summary.get("variants", {}) if isinstance(variants_dict, dict): for variant_name, data in variants_dict.items(): if not isinstance(data, dict): continue analysis = data.get("analysis", {}) if not isinstance(analysis, dict): continue ranking.append( { "variant": variant_name, "artifact_score_estimate": float(analysis.get("artifact_score_estimate", 0.0) or 0.0), } ) ranking.sort(key=lambda item: item["artifact_score_estimate"]) clip_summary["ranking"] = ranking per_clip_results.append(clip_summary) for variant in requested_variants: analyses = [] for clip_summary in per_clip_results: variants_dict = clip_summary.get("variants", {}) if not isinstance(variants_dict, dict): continue data = variants_dict.get(variant) if not isinstance(data, dict): continue analysis = data.get("analysis") if isinstance(analysis, dict): analyses.append(analysis) aggregate = _aggregate_variant_metrics(analyses) variant_result = variant_results.get(variant, {}) if isinstance(variant_result, dict) and "skipped" in variant_result: aggregate["skipped"] = variant_result["skipped"] aggregate_results[variant] = aggregate ranking = [ { "variant": variant, "avg_artifact_score_estimate": float(data.get("avg_artifact_score_estimate", 9999.0) or 9999.0), "clips": int(data.get("clips", 0) or 0), } for variant, data in aggregate_results.items() if not data.get("skipped") ] ranking.sort(key=lambda item: item["avg_artifact_score_estimate"]) summary = { "session_dir": str(session_dir), "output_dir": str(output_dir), "model_path": str(model_path), "index_path": str(index_path) if index_path else None, "model_inferred_from_session": inferred_model_key, "preprocess_mode": preprocess_mode, "source_constraint_mode": args.source_constraint_mode, "times_sec": [float(time_sec) for time_sec in times_sec], "window_sec": float(args.window_sec), "gap_sec": float(args.gap_sec), "manifest": manifest, "variants": variant_results, "per_clip": per_clip_results, "aggregate": aggregate_results, "ranking": ranking, "score_note": ( "artifact_score_estimate is a heuristic based on quiet-region energy, " "high-frequency excess, transition spikes, and synthetic breath count. Lower is better." ), } summary_path = output_dir / "summary.json" with open(summary_path, "w", encoding="utf-8") as f: json.dump(summary, f, ensure_ascii=False, indent=2) print("\nAggregate ranking (lower artifact score is better):") if ranking: for item in ranking: print( f" {item['variant']}: score={item['avg_artifact_score_estimate']:.4f}, " f"clips={item['clips']}" ) else: print(" No completed variants.") print(f"\nSummary written to: {summary_path}") return 0 if __name__ == "__main__": raise SystemExit(main())