133 lines
5.4 KiB
Python
133 lines
5.4 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Prove: the vmatrix alone can reproduce the reference WAV — and verify its provenance.
|
|
|
|
1. DECODE author.pt ref_code -> waveform (matrix -> audio; the user's requirement)
|
|
2. VERIFY that decoded audio matches the WAV the matrix was captured from
|
|
3. PROVE provenance by re-encoding each candidate source WAV and comparing its
|
|
audio_codes to the stored ref_code. Identical codes => that file is
|
|
provably where this matrix came from (no trusting the notes).
|
|
|
|
Runs on CPU with only the 682 MB speech tokenizer, so the TTS server keeps its
|
|
MPS model resident and nothing else contends for memory.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import librosa
|
|
import numpy as np
|
|
import soundfile as sf
|
|
import torch
|
|
|
|
sys.path.insert(0, "/Volumes/1TB_APL_Storage/UmbraAudio")
|
|
from qwen_tts import Qwen3TTSTokenizer # noqa: E402
|
|
|
|
PROJECT = Path("/Volumes/1TB_APL_Storage/UmbraAudio")
|
|
MATRIX = PROJECT / "DealHunterShip_4.0e_whole-to-audio/.umbra/voice_prompts/author.pt"
|
|
TOKENIZER = PROJECT / "models/Qwen3-TTS-12Hz-1.7B-Base/speech_tokenizer"
|
|
A4 = Path.home() / "Documents/Audacity4"
|
|
OUT = Path("/private/var/folders/y8/cbvhds297918lrxt2wd3btsr0000gn/T/opencode/probe_out")
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
|
|
CANDIDATES = {
|
|
"NOBG_v2 (documented v2 source)": A4 / "BDMurphy_Voice_NOBG_v2.wav",
|
|
"NOBG (documented v1 source)": A4 / "BDMurphy_Voice_NOBG.wav",
|
|
"raw recording": A4 / "BDMurphy_Voice.wav",
|
|
}
|
|
|
|
|
|
def f0_median(y, sr):
|
|
f0, _, _ = librosa.pyin(y.astype(np.float32), fmin=60.0, fmax=400.0, sr=sr)
|
|
v = f0[~np.isnan(f0)]
|
|
return float(np.median(v)) if len(v) else float("nan")
|
|
|
|
|
|
def lsd(a, b, sr):
|
|
"""Log-spectral distance over the speech band (dB)."""
|
|
n = min(len(a), len(b))
|
|
A = np.abs(librosa.stft(a[:n].astype(np.float32), n_fft=1024, hop_length=256))
|
|
B = np.abs(librosa.stft(b[:n].astype(np.float32), n_fft=1024, hop_length=256))
|
|
m = min(A.shape[1], B.shape[1])
|
|
A, B = A[:, :m], B[:, :m]
|
|
f = librosa.fft_frequencies(sr=sr, n_fft=1024)
|
|
band = (f >= 80) & (f <= 8000)
|
|
la = 20 * np.log10(A[band] + 1e-8)
|
|
lb = 20 * np.log10(B[band] + 1e-8)
|
|
return float(np.sqrt(np.mean((la - lb) ** 2)))
|
|
|
|
|
|
def main() -> None:
|
|
item = torch.load(str(MATRIX), weights_only=False)["items"][0]
|
|
ref_code = item["ref_code"].cpu()
|
|
print(f"matrix ref_code: {tuple(ref_code.shape)} (~{ref_code.shape[0] / 12.5:.2f}s)")
|
|
print(f"icl_mode={item['icl_mode']} x_vector_only_mode={item['x_vector_only_mode']}")
|
|
|
|
tok = Qwen3TTSTokenizer.from_pretrained(str(TOKENIZER), device_map="cpu")
|
|
print(f"tokenizer loaded, output sr = {tok.feature_extractor.sampling_rate}")
|
|
|
|
# ---------- 1. DECODE: matrix -> wav ----------
|
|
wavs, sr = tok.decode({"audio_codes": ref_code})
|
|
decoded = wavs[0].astype(np.float32)
|
|
dec_path = OUT / "author_ref_DECODED_FROM_MATRIX.wav"
|
|
sf.write(str(dec_path), decoded, sr, subtype="PCM_16")
|
|
print(f"\nDECODED from matrix: {len(decoded)/sr:.2f}s @ {sr} Hz -> {dec_path.name}")
|
|
print(f" median F0 = {f0_median(decoded, sr):.1f} Hz")
|
|
|
|
# ---------- 2. VERIFY against each candidate ----------
|
|
print("\n--- decoded vs candidate source WAVs ---")
|
|
print(f"{'candidate':36} {'dur':>7} {'F0':>7} {'LSD_dB':>8} {'corr':>7}")
|
|
results = {}
|
|
for label, p in CANDIDATES.items():
|
|
if not p.exists():
|
|
print(f"{label:36} MISSING")
|
|
continue
|
|
y, ysr = sf.read(str(p), dtype="float32")
|
|
if y.ndim > 1:
|
|
y = y.mean(axis=1)
|
|
if ysr != sr:
|
|
y = librosa.resample(y, orig_sr=ysr, target_sr=sr)
|
|
n = min(len(y), len(decoded))
|
|
corr = float(np.corrcoef(y[:n], decoded[:n])[0, 1])
|
|
d = lsd(y, decoded, sr)
|
|
results[label] = {"dur": round(len(y) / sr, 2), "f0": round(f0_median(y, sr), 1),
|
|
"lsd_db": round(d, 2), "corr": round(corr, 4)}
|
|
print(f"{label:36} {len(y)/sr:7.2f} {f0_median(y, sr):7.1f} {d:8.2f} {corr:7.4f}")
|
|
|
|
# ---------- 3. PROVE provenance: re-encode and compare codes ----------
|
|
print("\n--- provenance: re-encode candidate, compare audio_codes to stored ref_code ---")
|
|
print(f"{'candidate':36} {'shape':>12} {'exact_match':>13} {'frame_agree':>13}")
|
|
prov = {}
|
|
for label, p in CANDIDATES.items():
|
|
if not p.exists():
|
|
continue
|
|
enc = tok.encode([str(p)])
|
|
codes = enc.audio_codes[0].cpu()
|
|
same_shape = tuple(codes.shape) == tuple(ref_code.shape)
|
|
if same_shape:
|
|
exact = bool(torch.equal(codes, ref_code))
|
|
agree = float((codes == ref_code).float().mean())
|
|
else:
|
|
exact, agree = False, float("nan")
|
|
prov[label] = {"shape": list(codes.shape), "exact_match": exact,
|
|
"frame_agreement": round(agree, 6)}
|
|
print(f"{label:36} {str(tuple(codes.shape)):>12} {str(exact):>13} {agree:>13.6f}")
|
|
|
|
out = {
|
|
"matrix": str(MATRIX),
|
|
"ref_code_shape": list(ref_code.shape),
|
|
"decoded": {"path": str(dec_path), "duration_s": round(len(decoded) / sr, 3),
|
|
"sample_rate": sr, "median_f0_hz": round(f0_median(decoded, sr), 1)},
|
|
"verified_against": results,
|
|
"provenance": prov,
|
|
}
|
|
(OUT / "matrix_roundtrip.json").write_text(json.dumps(out, indent=2) + "\n")
|
|
print("\nwrote", OUT / "matrix_roundtrip.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|