Buckets:

Interscript
/

isx-training

Files

xet

Interscript/isx-training / run-031-audio-probe /probe_stage2.py

rhtse

23 days ago

download

raw

6.22 kB

"""WO09 stage 2 — alignment viability (the gate).
For N FLEURS he_il dev clips WITH their gold transcripts:
  ASR phonemes  = universal phoneme CTC (facebook/wav2vec2-lv-60-espeak-cv-ft)
  reference     = transcript -> heb-diac-plane-2.0 (nikud) -> nikud_ipa
                 rules (espeak-style IPA)
  metric        = phoneme error rate (edit distance) raw + after light
                 symbol normalization.
Verdict rule: median PER <= 0.45 => GO (audio supervision is usable —
ReNikud trained on noisier); > 0.60 => investigate before campaign.
Run on an apt-enabled job (espeak-ng needed by phonemizer):
  python /ckpt/run-031-audio-probe/probe_stage2.py
"""

from __future__ import annotations

import io
import json
import sys
import tarfile
from pathlib import Path

sys.path.insert(0, "/code")
sys.path.insert(0, "/code/src")

CKPT = Path("/ckpt/run-031-audio-probe")
AUDIO_TAR = Path("/tmp/fleurs-dev.tar.gz")
TSV = Path("/tmp/fleurs-dev.tsv")
N = 20
ZIP = Path("/ckpt/run-022-heb-plane-base/export/heb-diac-plane-2.0.zip")

import csv

import numpy as np
import soundfile as sf
import torch
from huggingface_hub import hf_hub_download

from interscript.ml.nikud_ipa import plane_to_ipa
from interscript.ml.plane import PlaneModel


def norm_phones(s: str, light: bool = False) -> list[str]:
    # compare per-PHONE: strip all separators, split into characters
    # (phones are mostly single chars; digraphs handled by light mapping)
    s = s.replace("ˈ", "").replace("ˌ", "").replace(".", "").replace("-", "")
    s = s.replace(" ", "")
    s = s.replace("ɡ", "g").replace("ʀ", "ʁ").replace("ɣ", "ʁ")
    if light:
        s = s.replace("χ", "x").replace("t͡s", "T").replace("ʦ", "T").replace("ʃ", "S")
    toks = list(s)
    if light:
        toks = [{"T": "ts", "S": "ʃ"}.get(c, c) for c in toks]
    return toks


def lev(a: list[str], b: list[str]) -> int:
    prev = list(range(len(b) + 1))
    for i, ca in enumerate(a, 1):
        cur = [i]
        for j, cb in enumerate(b, 1):
            cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
        prev = cur
    return prev[-1]


def main() -> None:
    # ---- data: first N tsv rows + their wavs from the tar ----

    def fetch(repo_file: str) -> str:
        p = hf_hub_download("google/fleurs", repo_file, repo_type="dataset")
        print(f"[fetch] {repo_file} -> {p}", flush=True)
        return p

    tsv_path = fetch("data/he_il/dev.tsv")
    rows = []
    with open(tsv_path, encoding="utf-8") as f:
        for row in csv.reader(f, delimiter="\t"):
            if len(row) < 3:
                continue
            rows.append(row)
            if len(rows) >= N:
                break
    # match by the tsv id/filename columns against tar basenames;
    # print samples once so mismatches are diagnosable
    clips: dict[str, list] = {}
    with tarfile.open(fetch("data/he_il/audio/dev.tar.gz"), "r:gz") as tf:
        members = [m for m in tf.getmembers() if m.name.endswith(".wav")]
        print(f"[names] tsv r[0]={rows[0][0]!r} r[1]={rows[0][1]!r}; "
              f"tar first={members[0].name!r} second={members[1].name!r}", flush=True)
        by_base = {m.name.split("/")[-1]: m for m in members}
        for row in rows:
            for cand in (row[0] + ".wav", row[1] + ".wav",
                         row[0] + ".wav.wav", row[1].split("/")[-1] + ".wav"):
                m = by_base.get(cand) or by_base.get(cand.replace(".wav.wav", ".wav"))
                if m:
                    f = tf.extractfile(m)
                    if f:
                        clips[cand] = row + [f.read()]
                    break
            if len(clips) >= N:
                break
    print(f"[data] tsv rows={len(rows)} wavs matched={len(clips)}", flush=True)


    # ---- ASR ----
    from transformers import AutoModelForCTC, AutoProcessor

    proc = AutoProcessor.from_pretrained("facebook/wav2vec2-lv-60-espeak-cv-ft")
    asr = AutoModelForCTC.from_pretrained("facebook/wav2vec2-lv-60-espeak-cv-ft").eval()
    if torch.cuda.is_available():
        asr = asr.cuda()

    # ---- our text chain ----
    model = PlaneModel.from_zip(ZIP.read_bytes())

    results = []
    import time

    t0 = time.time()
    for name, item in clips.items():
        text = item[2].strip()
        if not text:
            continue
        speech, sr = sf.read(io.BytesIO(item[-1]))
        if speech.ndim > 1:
            speech = speech.mean(axis=1)
        inputs = proc(speech.astype(np.float32), sampling_rate=sr, return_tensors="pt")
        if torch.cuda.is_available():
            inputs = {k: v.cuda() for k, v in inputs.items()}
        with torch.no_grad():
            logits = asr(**inputs).logits
        asr_ph = proc.batch_decode(logits.argmax(-1))[0]

        ref_ph = plane_to_ipa(model, text)
        a_raw, r_raw = norm_phones(asr_ph), norm_phones(ref_ph)
        a_lit, r_lit = norm_phones(asr_ph, True), norm_phones(ref_ph, True)
        per_raw = lev(a_raw, r_raw) / max(1, len(r_raw))
        per_lit = lev(a_lit, r_lit) / max(1, len(r_lit))
        results.append({"clip": name, "text": text[:40], "ref": ref_ph[:60],
                        "asr": asr_ph[:60], "per_raw": round(per_raw, 3),
                        "per_norm": round(per_lit, 3)})

    percs = [r["per_raw"] for r in results]
    percs_n = [r["per_norm"] for r in results]
    percs.sort(); percs_n.sort()
    med = percs[len(percs) // 2] if percs else 1.0
    med_n = percs_n[len(percs_n) // 2] if percs_n else 1.0
    verdict = "GO" if med <= 0.45 else ("INVESTIGATE" if med <= 0.60 else "NO-GO")
    report = {
        "stage": 2, "clips": len(results),
        "median_per_raw": round(med, 3), "median_per_norm": round(med_n, 3),
        "verdict": verdict,
        "rule": "median PER <= 0.45 GO; <= 0.60 investigate; else NO-GO",
        "wall_s": round(time.time() - t0, 1),
        "samples": results[:4],
    }
    (CKPT / "probe_stage2_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=1))
    print(json.dumps({k: v for k, v in report.items() if k != "samples"},
                     ensure_ascii=False, indent=1), flush=True)


if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 6.22 kB
Xet hash:: 1de7d54796194c2527209410145583b19b7c1301eb959909788dd1dc29e98447

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""WO09 stage 2 — alignment viability (the gate).

	For N FLEURS he_il dev clips WITH their gold transcripts:
	ASR phonemes = universal phoneme CTC (facebook/wav2vec2-lv-60-espeak-cv-ft)
	reference = transcript -> heb-diac-plane-2.0 (nikud) -> nikud_ipa
	rules (espeak-style IPA)
	metric = phoneme error rate (edit distance) raw + after light
	symbol normalization.

	Verdict rule: median PER <= 0.45 => GO (audio supervision is usable —
	ReNikud trained on noisier); > 0.60 => investigate before campaign.

	Run on an apt-enabled job (espeak-ng needed by phonemizer):
	python /ckpt/run-031-audio-probe/probe_stage2.py
	"""

	from __future__ import annotations

	import io
	import json
	import sys
	import tarfile
	from pathlib import Path

	sys.path.insert(0, "/code")
	sys.path.insert(0, "/code/src")

	CKPT = Path("/ckpt/run-031-audio-probe")
	AUDIO_TAR = Path("/tmp/fleurs-dev.tar.gz")
	TSV = Path("/tmp/fleurs-dev.tsv")
	N = 20
	ZIP = Path("/ckpt/run-022-heb-plane-base/export/heb-diac-plane-2.0.zip")

	import csv

	import numpy as np
	import soundfile as sf
	import torch
	from huggingface_hub import hf_hub_download

	from interscript.ml.nikud_ipa import plane_to_ipa
	from interscript.ml.plane import PlaneModel


	def norm_phones(s: str, light: bool = False) -> list[str]:
	# compare per-PHONE: strip all separators, split into characters
	# (phones are mostly single chars; digraphs handled by light mapping)
	s = s.replace("ˈ", "").replace("ˌ", "").replace(".", "").replace("-", "")
	s = s.replace(" ", "")
	s = s.replace("ɡ", "g").replace("ʀ", "ʁ").replace("ɣ", "ʁ")
	if light:
	s = s.replace("χ", "x").replace("t͡s", "T").replace("ʦ", "T").replace("ʃ", "S")
	toks = list(s)
	if light:
	toks = [{"T": "ts", "S": "ʃ"}.get(c, c) for c in toks]
	return toks


	def lev(a: list[str], b: list[str]) -> int:
	prev = list(range(len(b) + 1))
	for i, ca in enumerate(a, 1):
	cur = [i]
	for j, cb in enumerate(b, 1):
	cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
	prev = cur
	return prev[-1]


	def main() -> None:
	# ---- data: first N tsv rows + their wavs from the tar ----

	def fetch(repo_file: str) -> str:
	p = hf_hub_download("google/fleurs", repo_file, repo_type="dataset")
	print(f"[fetch] {repo_file} -> {p}", flush=True)
	return p

	tsv_path = fetch("data/he_il/dev.tsv")
	rows = []
	with open(tsv_path, encoding="utf-8") as f:
	for row in csv.reader(f, delimiter="\t"):
	if len(row) < 3:
	continue
	rows.append(row)
	if len(rows) >= N:
	break
	# match by the tsv id/filename columns against tar basenames;
	# print samples once so mismatches are diagnosable
	clips: dict[str, list] = {}
	with tarfile.open(fetch("data/he_il/audio/dev.tar.gz"), "r:gz") as tf:
	members = [m for m in tf.getmembers() if m.name.endswith(".wav")]
	print(f"[names] tsv r[0]={rows[0][0]!r} r[1]={rows[0][1]!r}; "
	f"tar first={members[0].name!r} second={members[1].name!r}", flush=True)
	by_base = {m.name.split("/")[-1]: m for m in members}
	for row in rows:
	for cand in (row[0] + ".wav", row[1] + ".wav",
	row[0] + ".wav.wav", row[1].split("/")[-1] + ".wav"):
	m = by_base.get(cand) or by_base.get(cand.replace(".wav.wav", ".wav"))
	if m:
	f = tf.extractfile(m)
	if f:
	clips[cand] = row + [f.read()]
	break
	if len(clips) >= N:
	break
	print(f"[data] tsv rows={len(rows)} wavs matched={len(clips)}", flush=True)


	# ---- ASR ----
	from transformers import AutoModelForCTC, AutoProcessor

	proc = AutoProcessor.from_pretrained("facebook/wav2vec2-lv-60-espeak-cv-ft")
	asr = AutoModelForCTC.from_pretrained("facebook/wav2vec2-lv-60-espeak-cv-ft").eval()
	if torch.cuda.is_available():
	asr = asr.cuda()

	# ---- our text chain ----
	model = PlaneModel.from_zip(ZIP.read_bytes())

	results = []
	import time

	t0 = time.time()
	for name, item in clips.items():
	text = item[2].strip()
	if not text:
	continue
	speech, sr = sf.read(io.BytesIO(item[-1]))
	if speech.ndim > 1:
	speech = speech.mean(axis=1)
	inputs = proc(speech.astype(np.float32), sampling_rate=sr, return_tensors="pt")
	if torch.cuda.is_available():
	inputs = {k: v.cuda() for k, v in inputs.items()}
	with torch.no_grad():
	logits = asr(**inputs).logits
	asr_ph = proc.batch_decode(logits.argmax(-1))[0]

	ref_ph = plane_to_ipa(model, text)
	a_raw, r_raw = norm_phones(asr_ph), norm_phones(ref_ph)
	a_lit, r_lit = norm_phones(asr_ph, True), norm_phones(ref_ph, True)
	per_raw = lev(a_raw, r_raw) / max(1, len(r_raw))
	per_lit = lev(a_lit, r_lit) / max(1, len(r_lit))
	results.append({"clip": name, "text": text[:40], "ref": ref_ph[:60],
	"asr": asr_ph[:60], "per_raw": round(per_raw, 3),
	"per_norm": round(per_lit, 3)})

	percs = [r["per_raw"] for r in results]
	percs_n = [r["per_norm"] for r in results]
	percs.sort(); percs_n.sort()
	med = percs[len(percs) // 2] if percs else 1.0
	med_n = percs_n[len(percs_n) // 2] if percs_n else 1.0
	verdict = "GO" if med <= 0.45 else ("INVESTIGATE" if med <= 0.60 else "NO-GO")
	report = {
	"stage": 2, "clips": len(results),
	"median_per_raw": round(med, 3), "median_per_norm": round(med_n, 3),
	"verdict": verdict,
	"rule": "median PER <= 0.45 GO; <= 0.60 investigate; else NO-GO",
	"wall_s": round(time.time() - t0, 1),
	"samples": results[:4],
	}
	(CKPT / "probe_stage2_report.json").write_text(
	json.dumps(report, ensure_ascii=False, indent=1))
	print(json.dumps({k: v for k, v in report.items() if k != "samples"},
	ensure_ascii=False, indent=1), flush=True)


	if __name__ == "__main__":
	main()