File size: 8,169 Bytes
b9dc61d | 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 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | from __future__ import annotations
import hashlib
import json
import shutil
from pathlib import Path
import numpy as np
import pytest
import soundfile as sf
from music3lab import reference_guided_append as rga
from music3lab.__main__ import build_parser
def _sha(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def _write_float(path: Path, audio: np.ndarray) -> None:
sf.write(path, audio.T, rga.SAMPLE_RATE, subtype="FLOAT", format="WAV")
def _fixture_bundle(tmp_path: Path) -> tuple[Path, Path]:
source = tmp_path / "source.wav"
t = np.arange(5000, dtype=np.float32) / rga.SAMPLE_RATE
source_audio = np.stack((0.1 * np.sin(2*np.pi*110*t), 0.09*np.sin(2*np.pi*113*t)))
sf.write(source, source_audio.T, rga.SAMPLE_RATE, subtype="PCM_16", format="WAV")
bundle = tmp_path / "candidates.reference-style"
bundle.mkdir()
candidates = []
profiles = []
for index in range(8):
phase = np.arange(rga.CANDIDATE_FRAMES, dtype=np.float32) / rga.SAMPLE_RATE
audio = np.stack((
0.04 * np.sin(2*np.pi*(220+index)*phase),
0.035 * np.sin(2*np.pi*(223+index)*phase),
)).astype(np.float32)
path = bundle / f"candidate-{index:02d}-seed-{101+index}.wav"
_write_float(path, audio)
decoded = sf.read(path, dtype="float32", always_2d=True)[0].T.copy()
candidates.append({
"index": index, "seed": 101 + index, "wav_sha256": _sha(path),
"decoded_pcm_sha256": hashlib.sha256(decoded.astype("<f4").tobytes()).hexdigest(),
"eligible": index < 3, "rejections": [] if index < 3 else ["tempo drift"],
"final_score": [0.3, 0.2, 0.2][index] if index < 3 else 0.5 + index / 100,
"source_correlation": 0.1,
"metrics": {"rms": 0.03, "clipped_fraction": 0.0, "tempo_bpm": 120.0,
"key": "C", "mode": "major", "stereo_width": 0.5},
})
profiles.append({"profile_sha256": f"{index:064x}", "finite": True})
source_decoded = sf.read(source, dtype="float32", always_2d=True)[0].T.copy()
manifest = {
"schema_version": "music3lab.reference-style-render.v1",
"capability": "reference_style_direct_latent_scorer_internal_text_bridge",
"reference": {
"source_audio_sha256": _sha(source),
"decoded_pcm_sha256": hashlib.sha256(source_decoded.astype("<f4").tobytes()).hexdigest(),
"tempo_bpm": 120.0, "key": "C", "mode": "major", "stereo_width": 0.5,
"profile": {"profile_sha256": "f" * 64},
},
"candidates": candidates,
"candidate_profiles": profiles,
"semantic_digest": "a" * 64,
}
(bundle / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
return source, bundle
def test_seam_formula_is_bounded_deterministic_and_combines_both_terms() -> None:
x = np.zeros((2, rga.CROSSFADE_FRAMES), dtype=np.float32)
y = np.linspace(-0.2, 0.2, rga.CROSSFADE_FRAMES, dtype=np.float32)[None].repeat(2, 0)
first = rga.seam_cost(x, y)
second = rga.seam_cost(x, y)
assert first == second
assert first["seam_cost"] == pytest.approx(
0.5 * (first["waveform_mismatch"] + first["first_difference_mismatch"])
)
assert all(0.0 <= value <= 1.0 for value in first.values())
def test_equal_power_append_has_exact_geometry_and_exact_outside_guard() -> None:
rng = np.random.default_rng(7)
source = rng.normal(0, 0.05, (2, 6000)).astype(np.float32)
candidate = rng.normal(0, 0.04, (2, rga.CANDIDATE_FRAMES)).astype(np.float32)
output = rga.equal_power_append(source, candidate)
assert output.shape == (2, 6000 + rga.CANDIDATE_FRAMES - rga.CROSSFADE_FRAMES)
assert np.array_equal(output[:, : 6000-rga.CROSSFADE_FRAMES], source[:, :-rga.CROSSFADE_FRAMES])
assert np.array_equal(output[:, 6000:], candidate[:, rga.CROSSFADE_FRAMES:])
assert np.array_equal(output[:, 6000-rga.CROSSFADE_FRAMES], source[:, -rga.CROSSFADE_FRAMES])
assert np.allclose(output[:, 5999], candidate[:, rga.CROSSFADE_FRAMES-1], atol=2e-7)
def test_ranking_uses_frozen_composite_and_candidate_index_tie() -> None:
records = [
{"index": 2, "eligible": True, "composite_score": 0.3},
{"index": 1, "eligible": True, "composite_score": 0.3},
{"index": 0, "eligible": False, "composite_score": 0.1},
]
assert rga.select_evaluated(records)["index"] == 1
def test_manifest_source_and_candidate_swap_are_rejected(tmp_path: Path) -> None:
source, bundle = _fixture_bundle(tmp_path)
wrong_source = tmp_path / "wrong.wav"
sf.write(wrong_source, np.zeros((6000, 2), dtype=np.float32), rga.SAMPLE_RATE, subtype="PCM_16")
with pytest.raises(ValueError, match="provenance"):
rga.evaluate_bundle(source=wrong_source, bundle=bundle, negative="")
first = bundle / "candidate-00-seed-101.wav"
second = bundle / "candidate-01-seed-102.wav"
saved = first.read_bytes()
first.write_bytes(second.read_bytes())
with pytest.raises(ValueError, match="file hash"):
rga.evaluate_bundle(source=source, bundle=bundle, negative="")
first.write_bytes(saved)
def test_negative_truth_and_cli_surface(tmp_path: Path) -> None:
source, bundle = _fixture_bundle(tmp_path)
_, _, records, constraints, _ = rga.evaluate_bundle(
source=source, bundle=bundle, negative="no vocals, no clipping, no tempo drift"
)
status = {item["name"]: item["status"] for item in constraints.to_dict()["items"]}
assert status == {"vocals": "NOT_ENFORCEABLE", "clipping": "ENFORCEABLE", "tempo drift": "ENFORCEABLE"}
assert constraints.to_dict()["native_negative_prompt_used"] is False
assert [record["index"] for record in records if record["eligible"]] == [0, 1, 2]
args = build_parser().parse_args([
"reference-guided-append", "--audio", "a.wav", "--bundle", "b", "--output", "o.wav"
])
assert args.config == "configs/reference-guided-append-v1.yaml" and not hasattr(args, "prompt")
def test_synthetic_end_to_end_roundtrip_and_retained_originals(tmp_path: Path) -> None:
source, bundle = _fixture_bundle(tmp_path)
result = rga.run_reference_guided_append(
source_audio=source, candidate_bundle=bundle, output=tmp_path / "joined.wav",
negative="no vocals, no clipping, no source copy",
config_path="configs/reference-guided-append-v1.yaml",
)
payload = json.loads(result.sidecar.read_text())
assert payload["status"] == "PASS" and len(payload["candidates"]) == 8
assert result.selected_index == 1
assert payload["geometry"]["net_appended_duration_seconds"] == pytest.approx(7.952834467120181)
assert payload["validation"]["float_wav_roundtrip_sample_exact"]
report_lines = result.report.read_text().splitlines()
assert all(any(line.startswith(f"| {index} |") for line in report_lines) for index in range(8))
assert _sha(result.evidence_bundle / "source-original.wav") == _sha(source)
assert _sha(result.evidence_bundle / "candidate-01-seed-102.wav") == _sha(bundle / "candidate-01-seed-102.wav")
def test_retained_real_bundle_integrates_when_present() -> None:
source = Path("/home/ubuntu/minimax-user-audio/incoming/loveonme_x_osh.wav")
bundle = Path("/home/ubuntu/minimax-reference-style-evidence/user-loveonme-f040ab1-seed101.reference-style")
if not source.exists() or not bundle.exists():
pytest.skip("retained measured bundle unavailable")
_, _, records, constraints, provenance = rga.evaluate_bundle(
source=source, bundle=bundle,
negative="no clipping, no source copy, no tempo drift, no vocals",
)
assert provenance["source"]["wav_sha256"] == "ff777c257a9f8044bbe08eb2933d6f4073a22770cf04ae527b8fe768982e6fea"
assert provenance["manifest_sha256"] == "2ada3a88bc57936810e7c8b8cf920d722225d438ca697745d9869ff4eeaa7485"
assert [record["index"] for record in records if record["eligible"]] == [0, 1, 2]
assert rga.select_evaluated(records)["index"] == 0
assert {item.name: item.status for item in constraints.items}["vocals"] == "NOT_ENFORCEABLE"
|