"""Oswald compare: test and compare Oswald V1, V2 and V3 on Datamio recordings or your own audio.
Datamio clips come from jsbeaudry/datamio-kreyol-voices (approved phone takes). Pick one clip or many, run any of the
models, and read word and character error rates, numbers found whole, and each model's output with its errors marked.
The scoring is the V3 evaluation's (scoring.py): numbers in words and in digits count the same, accents never decide.
V1 and V2 are Qwen3-ASR 0.6B fine-tunes (transformers, beam 4, as in the Trio Space); V3 is Parakeet-TDT 0.6B v3
fine-tuned (transformers ParakeetForTDT, converted from the NeMo checkpoint, greedy as NeMo scores it). All run on
ZeroGPU. The dataset and V3 are private, so the Space reads them with its HF_TOKEN secret.
"""
import csv
import glob
import io
import json
import os
import random
import re
import tempfile
import time
import spaces # ZeroGPU: before torch
import gradio as gr
import librosa
import numpy as np
import pandas as pd
import pyarrow.parquet as pq
import soundfile as sf
import torch
from huggingface_hub import snapshot_download
from transformers import AutoModelForTDT, AutoProcessor
from oswald import Oswald
from scoring import clip_wer, diff_html, number_accuracy, rates, target
TOKEN = os.environ.get("HF_TOKEN")
def token_report():
"""Whose token the Space has and what it may read; never the token itself."""
if not TOKEN:
return "HF_TOKEN is not set"
try:
from huggingface_hub import whoami
w = whoami(token=TOKEN)
auth = w.get("auth", {}).get("accessToken", {})
scopes = sorted({p for s in auth.get("fineGrained", {}).get("scoped", []) for p in s.get("permissions", [])})
return (f"HF_TOKEN belongs to {w.get('name')}, role {auth.get('role')}"
+ (f", fine-grained: {scopes or 'no repo permissions'}" if auth.get("role") == "fineGrained" else ""))
except Exception as e:
return f"HF_TOKEN is set but not accepted: {type(e).__name__}"
print(token_report(), flush=True)
DEVICE = os.environ.get("DEVICE", "cuda") # "cpu" only to run the app off the Space
QWEN_DTYPE = torch.bfloat16 if DEVICE == "cuda" else torch.float32
DATASET = "jsbeaudry/datamio-kreyol-voices"
SR = 16000
MAX_CLIPS = 500 # per run
CHUNK = 40 # clips per ZeroGPU call, so each call stays well under its time limit
MAX_ROWS_SHOWN = 300
MODELS = {
"Oswald V1": ("qwen", "jsbeaudry/Oswald-ASR-0.6-V1"),
"Oswald V2": ("qwen", "jsbeaudry/Oswald-ASR-0.6-V2"),
"Oswald V3": ("tdt", "jsbeaudry/Oswald-ASR-0.6-V3"),
"Oswald V3 no-JW": ("tdt", "jsbeaudry/Oswald-ASR-0.6-V3-nojw"),
}
NOTES = {
"Oswald V1": "Qwen3-ASR · numbers in words · never saw Datamio",
"Oswald V2": "Qwen3-ASR · digits · trained on Datamio training speakers (takes to Oct 15)",
"Oswald V3": "Parakeet-TDT · digits · trained with JW narration (private: jw.org terms)",
"Oswald V3 no-JW": "Parakeet-TDT · digits · same mix without JW",
}
# MODELS_LOAD="Oswald V2,Oswald V3 no-JW" loads a subset, for running off the Space with little memory.
WANTED = [m for m in os.environ.get("MODELS_LOAD", ",".join(MODELS)).split(",") if m in MODELS]
# ---------------------------------------------------------------- models
class Parakeet:
def __init__(self, repo):
self.processor = AutoProcessor.from_pretrained(repo, token=TOKEN)
self.model = AutoModelForTDT.from_pretrained(repo, token=TOKEN, dtype=torch.float32).to(DEVICE).eval()
@torch.inference_mode()
def transcribe_many(self, audios, batch=8):
out = []
for i in range(0, len(audios), batch):
inputs = self.processor(audios[i:i + batch], sampling_rate=SR)
inputs = inputs.to(self.model.device, dtype=self.model.dtype)
ids = self.model.generate(**inputs, return_dict_in_generate=True)
out += [t.strip() for t in self.processor.batch_decode(ids.sequences, skip_special_tokens=True)]
return out
def load_model(name):
kind, repo = MODELS[name]
if kind == "qwen":
return Oswald(repo, DEVICE, QWEN_DTYPE)
return Parakeet(repo)
models, load_errors = {}, {}
for name in WANTED:
try:
models[name] = load_model(name)
print(f"loaded {name}", flush=True)
except Exception as e: # a private repo without HF_TOKEN must not take the whole Space down
load_errors[name] = f"{type(e).__name__}: {str(e)[:200]}"
print(f"could not load {name}: {load_errors[name]}", flush=True)
AVAILABLE = [m for m in MODELS if m in models]
def gpu_seconds(name, audios):
total = sum(len(a) for a in audios) / SR
if MODELS[name][0] == "qwen":
need = 15 + 0.2 * total + 0.5 * len(audios)
else:
need = 10 + 0.03 * total + 0.1 * len(audios)
return int(min(120, need))
@spaces.GPU(duration=gpu_seconds)
def transcribe(name, audios):
m = models[name]
if isinstance(m, Parakeet):
return m.transcribe_many(audios)
return [m.transcribe(a) for a in audios]
# ---------------------------------------------------------------- dataset
splits = json.load(open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "datamio_v2_splits.json")))
SPLIT_OF = {s: "test" for s in splits["test"]} | {s: "validation" for s in splits["val"]}
SPLIT_LABEL = {"test": "held-out test speaker", "validation": "held-out validation speaker",
"train": "training speaker"}
META_COLS = ["id", "speaker", "gender", "age", "department", "duration_s", "snr_db", "text", "oswald_score",
"recorded_at"]
meta, dataset_error, files = pd.DataFrame(), "", []
try:
# DATA_PATTERN loads fewer shards, only to run the app off the Space
local = snapshot_download(DATASET, repo_type="dataset", token=TOKEN,
allow_patterns=[os.environ.get("DATA_PATTERN", "data/*.parquet")])
pattern = os.environ.get("DATA_PATTERN", "data/*.parquet").split("/")[-1]
files = sorted(f for f in glob.glob(os.path.join(local, "data", pattern)))
parts = []
for fi, path in enumerate(files):
pf = pq.ParquetFile(path)
cols = [c for c in META_COLS if c in pf.schema_arrow.names]
for g in range(pf.num_row_groups):
t = pf.read_row_group(g, columns=cols).to_pandas()
t["_file"], t["_group"], t["_row"] = fi, g, range(len(t))
parts.append(t)
meta = pd.concat(parts, ignore_index=True)
meta["split"] = meta["speaker"].map(SPLIT_OF).fillna("train")
meta["duration_s"] = meta["duration_s"].astype(float).round(1)
meta = meta.sort_values(["split", "speaker", "recorded_at"], key=lambda s: s.map(
{"test": 0, "validation": 1, "train": 2}) if s.name == "split" else s).reset_index(drop=True)
print(f"dataset: {len(meta):,} clips, {meta.speaker.nunique()} speakers", flush=True)
except Exception as e:
dataset_error = f"{type(e).__name__}: {str(e)[:300]}"
print("dataset not loaded:", dataset_error, flush=True)
_pq = {}
def clip_audio(i):
"""16 kHz mono float audio of meta row i, read from its parquet row group."""
r = meta.iloc[i]
pf = _pq.setdefault(int(r["_file"]), pq.ParquetFile(files[int(r["_file"])]))
b = pf.read_row_group(int(r["_group"]), columns=["audio"]).column("audio")[int(r["_row"])].as_py()["bytes"]
a, sr = sf.read(io.BytesIO(b), dtype="float32")
a = a.mean(1) if a.ndim > 1 else a
return librosa.resample(a, orig_sr=sr, target_sr=SR) if sr != SR else a
def clip_audios(rows):
"""clip_audio for many rows, reading each parquet row group once."""
out, by_group = {}, {}
for i in rows:
r = meta.iloc[i]
by_group.setdefault((int(r["_file"]), int(r["_group"])), []).append(i)
for (fi, g), members in by_group.items():
col = _pq.setdefault(fi, pq.ParquetFile(files[fi])).read_row_group(g, columns=["audio"]).column("audio")
for i in members:
a, sr = sf.read(io.BytesIO(col[int(meta.iloc[i]["_row"])].as_py()["bytes"]), dtype="float32")
a = a.mean(1) if a.ndim > 1 else a
out[i] = librosa.resample(a, orig_sr=sr, target_sr=SR) if sr != SR else a
return [out[i] for i in rows]
def choices(col):
if meta.empty or col not in meta:
return ["All"]
counts = meta[col].fillna("").replace("", "(not given)").value_counts()
return ["All"] + [f"{k} ({v})" for k, v in counts.sort_index().items()]
def strip_count(v):
return re.sub(r" \(\d+\)$", "", v or "All")
def filter_clips(split, speaker, gender, department, search):
if meta.empty:
return pd.DataFrame(), [], "No dataset loaded."
m = meta
want = {"Held-out test speakers": ["test"], "Held-out validation speakers": ["validation"],
"Training speakers": ["train"], "All speakers": ["test", "validation", "train"]}[split]
m = m[m.split.isin(want)]
for col, v in (("speaker", speaker), ("gender", gender), ("department", department)):
v = strip_count(v)
if v != "All":
m = m[m[col].fillna("").replace("", "(not given)") == v]
if search and search.strip():
m = m[m.text.str.contains(re.escape(search.strip()), case=False, na=False)]
idx = list(m.index)
shown = m.head(MAX_ROWS_SHOWN)
table = pd.DataFrame({"#": range(1, len(shown) + 1), "speaker": shown.speaker, "split": shown.split,
"s": shown.duration_s, "text": shown.text})
note = f"{len(idx):,} clips match" + (f" (first {MAX_ROWS_SHOWN} shown; random picks use all)"
if len(idx) > MAX_ROWS_SHOWN else "")
return table, idx, note
def parse_selection(text, idx):
"""'1, 4, 7-9' -> meta rows, by position (#) in the current filtered list."""
picked = []
for part in re.split(r"[,\s]+", text or ""):
if not part:
continue
a, _, b = part.partition("-")
try:
lo, hi = int(a), int(b or a)
except ValueError:
continue
picked += [idx[k - 1] for k in range(lo, hi + 1) if 1 <= k <= len(idx)]
return list(dict.fromkeys(picked))
def add_clicked(evt: gr.SelectData, selection, idx):
k = evt.index[0] + 1
nums = [s for s in re.split(r"[,\s]+", selection or "") if s]
if str(k) not in nums:
nums.append(str(k))
i = idx[k - 1]
r = meta.loc[i]
info = f"**#{k}** · {r.speaker} · {SPLIT_LABEL[r.split]} · {r.duration_s} s\n\n{r.text}"
return ", ".join(nums), (SR, clip_audio(i)), info
def random_pick(n, idx):
if not idx:
return ""
ks = sorted(random.sample(range(1, len(idx) + 1), min(int(n), len(idx), MAX_CLIPS)))
return ", ".join(map(str, ks))
# ---------------------------------------------------------------- running and reporting
CSS = """
del.miss {color:#c0392b; text-decoration:line-through;}
ins.extra {color:#d35400; text-decoration:none; font-weight:600;}
.clip {border:1px solid var(--border-color-primary); border-radius:8px; padding:10px 12px; margin:8px 0;}
.clip .ref {font-weight:600; margin-bottom:6px;}
.clip table {width:100%; border-collapse:collapse;}
.clip td {padding:3px 6px; vertical-align:top; border-top:1px solid var(--border-color-primary);}
.clip td.m {white-space:nowrap; width:1%; font-weight:600;}
.clip td.w {white-space:nowrap; width:1%; text-align:right;}
.clip .scored {color:var(--body-text-color-subdued); font-size:0.9em;}
"""
def summary_table(names, refs, outs, secs, audio_s):
rows = []
for n in names:
wer, cer = rates(refs, outs[n]) if refs else (None, None)
acc, spans = number_accuracy(refs, outs[n]) if refs else (None, 0)
rows.append({"model": n, "clips": len(outs[n]),
"WER %": None if wer is None else round(100 * wer, 2),
"CER %": None if cer is None else round(100 * cer, 2),
"numbers whole": "-" if acc is None else f"{100 * acc:.0f}% of {spans}",
"GPU s": round(secs[n], 1), "RTF": round(secs[n] / max(audio_s, 1e-6), 4)})
return pd.DataFrame(rows)
def cards(names, labels, refs, outs):
html = []
for k, (label, ref) in enumerate(zip(labels, refs)):
lines = []
for n in names:
hyp = outs[n][k]
w = clip_wer(ref, hyp) if ref else None
scored = f'scored as: {diff_html(ref, hyp)}
' if ref else ""
lines.append(f'| {n} | {hyp or "(nothing)"}{scored} | '
f'{"" if w is None else f"{100 * w:.0f}%"} |
')
head = f'{label}
' + (f"{ref}
" if ref else "")
html.append(f'')
return "".join(html) or "No clips."
def write_csv(names, labels, refs, outs):
f = tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False, newline="", encoding="utf-8")
w = csv.writer(f)
w.writerow(["clip", "reference"] + [c for n in names for c in (n, f"{n} WER")])
for k, (label, ref) in enumerate(zip(labels, refs)):
row = [label, ref]
for n in names:
cw = clip_wer(ref, outs[n][k]) if ref else None
row += [outs[n][k], "" if cw is None else round(100 * cw, 2)]
w.writerow(row)
f.close()
return f.name
def run_models(names, audios):
outs, secs = {}, {}
for n in names:
t0 = time.time()
outs[n] = transcribe(n, audios)
secs[n] = time.time() - t0
return outs, secs
def start_job(selection, idx, names):
"""A run is a job: one step per (model, 40-clip chunk), each its own request. ZeroGPU's per-request token
expires after a few minutes, and 500 clips on V1 and V2 take about ten."""
names = [n for n in MODELS if n in names and n in models]
rows = parse_selection(selection, idx)
if not names:
raise gr.Error("Choose at least one model.")
if not rows:
raise gr.Error("Pick clips first: click rows in the table, type numbers like 1, 4, 7-9, or use Random.")
if len(rows) > MAX_CLIPS:
gr.Warning(f"Only the first {MAX_CLIPS} clips are run.")
rows = rows[:MAX_CLIPS]
pos = {i: k + 1 for k, i in enumerate(idx)}
job = {"rows": rows, "names": names, "refs": [meta.loc[i, "text"] for i in rows],
"labels": [f"#{pos[i]} · {meta.loc[i, 'speaker']} · {SPLIT_LABEL[meta.loc[i, 'split']]} · "
f"{meta.loc[i, 'duration_s']} s" for i in rows],
"audio_s": float(meta.loc[rows, "duration_s"].sum()),
"todo": [(n, c) for n in names for c in range(0, len(rows), CHUNK)],
"outs": {n: [] for n in names}, "secs": {n: 0.0 for n in names}, "step": 0}
job["steps"] = len(job["todo"])
return job, progress_text(job), None, "", None
def progress_text(job):
if not job["todo"]:
return f"Done: {len(job['rows'])} clips × {len(job['names'])} models."
n, c = job["todo"][0]
return (f"Running **{n}** on clips {c + 1}-{min(c + CHUNK, len(job['rows']))} of {len(job['rows'])} "
f"(step {job['step'] + 1} of {job['steps']})…")
def step_job(job):
"""One chunk of one model; scores appear for each model as soon as it has finished all its clips."""
if not job or not job["todo"]:
return gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip()
n, c = job["todo"].pop(0)
part = job["rows"][c:c + CHUNK]
o, s = run_models([n], clip_audios(part))
job["outs"][n] += o[n]
job["secs"][n] += s[n]
job["step"] += 1
finished = [m for m in job["names"] if len(job["outs"][m]) == len(job["rows"])]
table = summary_table(finished, job["refs"], job["outs"], job["secs"], job["audio_s"]) if finished else None
if job["todo"]:
return job, progress_text(job), table, gr.skip(), gr.skip()
names, labels, refs, outs = job["names"], job["labels"], job["refs"], job["outs"]
return job, progress_text(job), table, cards(names, labels, refs, outs), write_csv(names, labels, refs, outs)
def run_own(audio, reference, names):
names = [n for n in MODELS if n in names and n in models]
if audio is None:
raise gr.Error("Record or upload audio first.")
if not names:
raise gr.Error("Choose at least one model.")
sr, a = audio
a = a.astype(np.float32)
if a.ndim > 1:
a = a.mean(1)
if np.abs(a).max() > 1.5: # integer PCM from the microphone
a = a / 32768.0
if sr != SR:
a = librosa.resample(a, orig_sr=sr, target_sr=SR)
a = a[: 120 * SR]
ref = (reference or "").strip()
outs, secs = run_models(names, [a])
refs = [ref] if ref else []
table = summary_table(names, refs, outs, secs, len(a) / SR)
return table, cards(names, ["Your audio"], [ref], outs)
# ---------------------------------------------------------------- interface
def status_md():
lines = []
if dataset_error:
lines.append(f"⚠️ Dataset not loaded ({dataset_error}). Add an `HF_TOKEN` secret with read access.")
for n, e in load_errors.items():
lines.append(f"⚠️ {n} not loaded ({e}).")
return "\n\n".join(lines)
with gr.Blocks(title="Oswald compare", css=CSS, theme=gr.themes.Soft()) as demo:
gr.Markdown("# Oswald compare\nTest Oswald V1, V2 and V3 on Datamio recordings or on your own audio. "
"Scores read numbers in words and in digits the same way and ignore case, punctuation and accents.")
if status_md():
gr.Markdown(status_md())
model_box = gr.CheckboxGroup(AVAILABLE, value=AVAILABLE, label="Models",
info=" · ".join(f"**{n}**: {NOTES[n]}" for n in AVAILABLE))
with gr.Tab("Datamio clips"):
idx_state = gr.State([])
with gr.Row():
split_box = gr.Dropdown(["Held-out test speakers", "Held-out validation speakers", "Training speakers",
"All speakers"], value="Held-out test speakers", label="Speakers",
info="V2 and V3 never trained on held-out speakers")
speaker_box = gr.Dropdown(choices("speaker"), value="All", label="Speaker")
gender_box = gr.Dropdown(choices("gender"), value="All", label="Gender")
dept_box = gr.Dropdown(choices("department"), value="All", label="Department")
with gr.Row():
search_box = gr.Textbox(label="Text contains", placeholder="e.g. goud, 2027, lekòl", scale=3)
show_btn = gr.Button("Show clips", scale=1)
count_md = gr.Markdown()
table = gr.Dataframe(interactive=False, wrap=True, max_height=380,
label="Clips: click a row to add it to the selection and hear it")
with gr.Row():
with gr.Column(scale=2):
selection = gr.Textbox(label="Selected clips (# from the table)", placeholder="1, 4, 7-9")
with gr.Row():
n_random = gr.Slider(1, MAX_CLIPS, value=10, step=1, label="Random clips",
info="500 clips on all four models take about 12 GPU minutes")
random_btn = gr.Button("Pick random")
clear_btn = gr.Button("Clear")
with gr.Column(scale=2):
preview = gr.Audio(label="Last clicked clip", interactive=False)
preview_text = gr.Markdown()
run_btn = gr.Button(f"Transcribe and compare (up to {MAX_CLIPS} clips)", variant="primary")
job_state = gr.State(None)
progress_md = gr.Markdown()
summary = gr.Dataframe(label="Scores (each model as soon as it has finished)", interactive=False)
report = gr.HTML()
csv_file = gr.File(label="All outputs (CSV)")
filters = [split_box, speaker_box, gender_box, dept_box, search_box]
for comp in filters[:4]:
comp.change(filter_clips, filters, [table, idx_state, count_md]).then(lambda: "", None, selection)
show_btn.click(filter_clips, filters, [table, idx_state, count_md])
search_box.submit(filter_clips, filters, [table, idx_state, count_md])
table.select(add_clicked, [selection, idx_state], [selection, preview, preview_text])
random_btn.click(random_pick, [n_random, idx_state], selection)
clear_btn.click(lambda: "", None, selection)
outputs = [job_state, progress_md, summary, report, csv_file]
event = run_btn.click(start_job, [selection, idx_state, model_box], outputs)
for _ in range(len(MODELS) * -(-MAX_CLIPS // CHUNK)):
event = event.success(step_job, job_state, outputs) # stops at the first failed step
demo.load(filter_clips, filters, [table, idx_state, count_md])
with gr.Tab("Your audio"):
with gr.Row():
own_audio = gr.Audio(sources=["microphone", "upload"], type="numpy", label="Kreyòl audio (up to 2 min)")
own_ref = gr.Textbox(label="What was said (optional, for scores)", lines=4)
own_btn = gr.Button("Transcribe and compare", variant="primary")
own_summary = gr.Dataframe(label="Scores", interactive=False)
own_report = gr.HTML()
own_btn.click(run_own, [own_audio, own_ref, model_box], [own_summary, own_report])
with gr.Tab("About"):
gr.Markdown(f"""
**Datamio clips.** `{DATASET}`: approved phone recordings of sentences read aloud. The 14 speakers V2 held out
(10 test, 4 validation) stayed out of V3's training too, so only they are fair ground for comparing V2 and V3.
V1 never trained on Datamio at all. Training speakers' clips were heard in training by V2 (takes to Oct 15) and V3.
**Scores.** Word and character error rate after reading digits back as words, lowercasing and dropping punctuation
and accents, as in the V3 evaluation. *Numbers whole* counts a number right only when all its words come out.
In each clip's card, *scored as* shows the output in that form: struck red = missed, orange = extra or wrong.
**Models.** {"; ".join(f"{n}: `{MODELS[n][1]}`" for n in MODELS)}. V1/V2 decode with beam 4 and drop phrases
repeated more than three times (as in the Trio Space); V3 decodes greedily (TDT). V3 with JW is private by the
jw.org terms; the no-JW arm has no jw.org data.
""")
demo.queue(max_size=16).launch()