# ============================================================================= # NADI 2026 - Subtask 1.2 Mixed ASR - COMPLETE PIPELINE IN ONE CELL # # Paste this whole file into a single Colab cell and run it. Nothing else is # needed: no other file, no notebook, no repo artifact. Runtime > GPU. # Token: Colab > Secrets > HF_TOKEN (or it will prompt). # # Fixed here, with what each was worth: # 1. line order - 0.9 WER. predictions.txt line k must be DATASET row k, # not sorted(ids). This took the blind score 1.3757 -> 0.48. # 2. repeat loops - two clips decoded 320- and 62-word loops, ~380 insertions. # 3. long audio - ~7 WER, NOT YET CLAIMED. max_audio_clip_s=35 makes the # processor auto-chunk and return MORE rows than you passed; # zip(batch, texts) silently dropped them, so clips over 35 s # kept only their first chunk. 8 blind clips came out at # 0.03-1.73 words/sec against a 2.19 median: ~500 words lost. # 4. blanking norm - a literal Arabic char class resolved to [U+0617-U+064B], # covering every Arabic letter, and deleted 2980/3152 # transcripts (WER 0.9996). All Arabic below is \uXXXX. # # Text post-processing is exhausted: 0.7 WER points total, all already applied. # Rejected on dev - split leading waw (0.5872 vs 0.5382), ya->ya (0.5493), # ta-marbuta->ha (0.5632), alli->ali (0.5402). Do not re-add them. # ============================================================================= # ------------------------------- CONFIG -------------------------------------- QUICK_MODE = True # True : 1 model, 200 dev clips -> valid submission in ~30 min # False: 4-model ensemble, 500 dev clips (~2-3 h on an A100) RUN_DEV = True # score on dev before submitting. Leave True. HF_REPO = "FatimahEmadEldin/Nadi" # None to skip pushing PUSH_PREFIX = "claude/standalone" RESTORE_CACHE = True # pull previously computed transcripts from claude/cache/ # ----------------------------------------------------------------------------- import sys, subprocess, os, json, gc, time, re, random, glob, zipfile, getpass subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", "transformers", "accelerate", "datasets", "soundfile", "librosa", "jiwer", "huggingface_hub", "pyarrow", "fsspec"], check=True) import torch, numpy as np import fsspec, pyarrow.parquet as pq, jiwer from jiwer import wer, cer from tqdm.auto import tqdm from huggingface_hub import login, HfApi, hf_hub_download, list_repo_files from datasets import load_dataset, Audio DEV_REPO = "UBC-NLP/NADI2026_subtask2_MixedASR" TEST_REPO = "UBC-NLP/NADI2026_subtask1.2_MixedASR_test" N_TEST, SR = 427, 16000 DEV_EVAL_N = 200 if QUICK_MODE else 500 HAS_GPU = torch.cuda.is_available() print("torch", torch.__version__, "| GPU:", torch.cuda.get_device_name(0) if HAS_GPU else "NONE (dev scoring still works)") if HAS_GPU: torch.backends.cuda.matmul.allow_tf32 = True torch.backends.cudnn.allow_tf32 = True os.makedirs("out", exist_ok=True) # ============================== 1. LOGIN ===================================== _tok = None try: from google.colab import userdata _tok = userdata.get("HF_TOKEN") except Exception: _tok = os.environ.get("HF_TOKEN") if not _tok: _tok = getpass.getpass("HF token: ").strip() login(token=_tok, add_to_git_credential=False) os.environ["HF_TOKEN"] = _tok api = HfApi(token=_tok) print("logged in |", "QUICK mode" if QUICK_MODE else "FULL ensemble mode") if RESTORE_CACHE and HF_REPO: try: hits = [f for f in list_repo_files(HF_REPO, token=_tok) if f.startswith("claude/cache/") and f.endswith(".jsonl")] for f in hits: dst = "out/" + os.path.basename(f) if not os.path.exists(dst): open(dst, "wb").write(open(hf_hub_download(HF_REPO, f, token=_tok), "rb").read()) print(" restored", dst) print(f"cache: {len(hits)} file(s) on the hub") except Exception as e: print("cache restore skipped:", type(e).__name__) # ========================== 2. TEXT PIPELINE ================================= # Arabic is \uXXXX ON PURPOSE. Combining marks reorder between visual and logical # order; a literal class became [U+0617-U+064B] once, swallowing U+0621-U+064A # (every Arabic letter), and the normalizer deleted whole transcripts. _DIAC = re.compile("[\u0617-\u061a\u064b-\u0652\u0670\u0640]") # marks, tanween..sukun, superscript alef, tatweel _PUNCT_CHARS = "\u061b\u060c\u061f" + "`\u00f7\u00d7_()*^%][/:\"{}~\u00a6+|!\u2026\u2013\u2019\u201d\u201c;?.," _PUNCT = re.compile("[" + re.escape(_PUNCT_CHARS) + "]") _BRACKETED = re.compile(r"[\[\(\{<][^\]\)\}>]{0,40}[\]\)\}>]") _TAGWORDS = re.compile(r"\b[Ss][Pp][Ee][Aa][Kk][Ee][Rr]\s*[A-Za-z0-9]?\b") _JUNK = re.compile("[#@\u266a\u2669\u266b\u266c\\*\\$&<>\\\\]+") _LATIN = re.compile(r"[A-Za-z]+") _ELONG = re.compile(r"(.)\1{3,}") _ALEF = re.compile("[\u0623\u0625\u0622\u0671]") _PLAIN_ALEF = "\u0627" _FILLER = "\u0646\u0639\u0645" # never ship a blank line LEARNED_MAP = {"\u0634\u064a\u0621": "\u0634\u064a"} # mined on dev, 2-fold CV def norm_ws(t): return re.sub(r"\s+", " ", t).strip() def BEST_NORM(t): return norm_ws(_PUNCT.sub(" ", _DIAC.sub("", t))) def strip_junk(t): t = _BRACKETED.sub(" ", t); t = _TAGWORDS.sub(" ", t) t = _JUNK.sub(" ", t); t = _LATIN.sub(" ", t) return norm_ws(t) def collapse_repeats(text, max_run=3, max_ngram=5, max_phrase_run=3): """Kill decoder loops. Grid-searched on dev: strip=1, max_run=3, max_phrase_run=3.""" w = text.split() if not w: return text out, run = [], 0 for tok in w: if out and tok == out[-1]: run += 1 if run >= max_run: continue else: run = 0 out.append(tok) w = out for n in range(max_ngram, 1, -1): i, out = 0, [] while i < len(w): gram = w[i:i+n] if len(gram) < n: out.extend(w[i:]); break reps, j = 1, i + n while w[j:j+n] == gram: reps += 1; j += n if reps > max_phrase_run: out.extend(gram * max_phrase_run); i = j else: out.extend(gram); i += n w = out return " ".join(w) def fold_alef(t): """dev short subset WER 0.5382 -> 0.5320. A no-op if the scorer already folds alef, a gain if it does not, so it is free either way.""" return _ALEF.sub(_PLAIN_ALEF, t) def squeeze_elong(t, keep=3): """216-char 'hhhhh...' -> 'hhh'. Word-level collapse cannot catch it: one token, not a repeated token. CER 0.2425 -> 0.2317, WER unchanged.""" return _ELONG.sub(lambda m: m.group(1) * keep, t) def apply_learned(t): return " ".join(LEARNED_MAP.get(w, w) for w in t.split()) def CLEAN_NORM(t): return norm_ws(collapse_repeats(strip_junk(BEST_NORM(t)))) def FINAL_NORM(t): return norm_ws(apply_learned(squeeze_elong(fold_alef(CLEAN_NORM(t))))) def max_word_run(text): w = text.split(); best = cur = 1 for a, b in zip(w, w[1:]): cur = cur + 1 if a == b else 1; best = max(best, cur) return best if w else 0 # STOP here if the normalizer eats Arabic - that failure scores 0.9996 and looks # like a model problem for hours. _ANA = "\u0627\u0646\u0627" assert BEST_NORM(_ANA) == _ANA, "normalizer is deleting Arabic letters - check _DIAC" assert fold_alef("\u0623") == _PLAIN_ALEF _t1 = "###" + "\u0623\u0646\u0627" + " " + "\u0647" * 19 + " [Speaker B ] " + "\u0634\u064a\u0621" assert FINAL_NORM(_t1) == _ANA + " \u0647\u0647\u0647 \u0634\u064a", FINAL_NORM(_t1) assert FINAL_NORM("") == "" print("text pipeline verified |", FINAL_NORM(_t1)) # ===================== 3. DEV REFERENCES + SCORING =========================== DEV_SHARDS = ["dev-00000-of-00009-051641e73732d26d", "dev-00001-of-00009-f2505c9f0c41e739", "dev-00002-of-00009-574478cbf86a090c", "dev-00003-of-00009-f2d81e6bc3114920", "dev-00004-of-00009-797f0063ead966d6", "dev-00005-of-00009-c68f72797dac98b6", "dev-00006-of-00009-e6e6831129f93d1e", "dev-00007-of-00009-7fea04d4729de732", "dev-00008-of-00009-f28e4000cf5e2fa3"] if os.path.exists("out/dev_ids_refs.json"): _d = json.load(open("out/dev_ids_refs.json", encoding="utf-8")) dev_ids, dev_refs = _d["ids"], _d["refs"] else: _fs = fsspec.filesystem("http", client_kwargs={"headers": {"Authorization": "Bearer " + _tok}}) dev_ids, dev_refs = [], [] for nm in DEV_SHARDS: # text columns only - no audio downloaded url = f"https://huggingface.co/datasets/{DEV_REPO}/resolve/main/data/{nm}.parquet" t0 = time.time() with _fs.open(url, "rb", block_size=4 * 1024 * 1024) as fh: t = pq.ParquetFile(fh).read(columns=["id", "transcription"]) dev_ids += t.column("id").to_pylist(); dev_refs += t.column("transcription").to_pylist() print(f" {nm[:12]} {t.num_rows} rows {time.time()-t0:.0f}s total {len(dev_ids)}", flush=True) json.dump({"ids": dev_ids, "refs": dev_refs}, open("out/dev_ids_refs.json", "w", encoding="utf-8"), ensure_ascii=False) DEV_N = len(dev_refs) DEV_REFS = [norm_ws(r) for r in dev_refs] _sh = [i for i in range(DEV_N) if len(DEV_REFS[i].split()) <= 40] _lo = [i for i in range(DEV_N) if len(DEV_REFS[i].split()) > 60] random.Random(0).shuffle(_sh) DEV_SEL = sorted(_sh[:DEV_EVAL_N]) # blind-like clips: THE selection metric DEV_LONG = sorted(_lo) # long clips: windowing diagnostic only DEV_EVAL = sorted(set(DEV_SEL) | set(DEV_LONG)) _ws = sum(len(DEV_REFS[i].split()) for i in DEV_SEL) _wl = sum(len(DEV_REFS[i].split()) for i in DEV_LONG) print(f"dev {DEV_N} clips") print(f" selection set : {len(DEV_SEL):4d} clips, {_ws:6d} ref words, " f"{_ws/max(len(DEV_SEL),1):.1f} w/clip (blind: 427 clips, 6519 words, 15.3 w/clip)") print(f" long diagnostic:{len(DEV_LONG):4d} clips, {_wl:6d} ref words - transcribed to verify the") print(f" windowing fix, but NOT scored for selection: WER is") print(f" sum(errors)/sum(ref_words), so these {_wl*100//max(_ws+_wl,1)}% of words would") print(f" swamp the metric with clips 4x longer than any blind clip.") def score(hyp_by_index, idxs, label, norm=FINAL_NORM): """References stay in DATASET order; only the hypothesis is ever permuted. wer(ref[perm], hyp[perm]) is identical for every perm - that invariance is exactly what hid the ordering bug. Never write a check that way.""" h = [norm(hyp_by_index.get(i, "")) for i in idxs] r = [DEV_REFS[i] for i in idxs] w, c = wer(r, h), cer(r, h) empty = sum(1 for x in h if not x.strip()) print(f" {label:46s} WER {w:.4f} CER {c:.4f}" + (f" [{empty} EMPTY]" if empty else "")) raw_w = sum(len(str(hyp_by_index.get(i, "")).split()) for i in idxs) out_w = sum(len(x.split()) for x in h) # a blanking normalizer scores 0.9997 and slips past a bare `wer < 1.0` assert empty < 0.25 * len(h), ( f"{label}: {empty}/{len(h)} hypotheses EMPTY after normalization. The normalizer " f"is deleting text - not a model problem.") assert raw_w == 0 or out_w > 0.5 * raw_w, ( f"{label}: normalization dropped {raw_w - out_w} of {raw_w} words. Check _DIAC.") assert w < 1.0, f"{label}: WER >= 1 means the pairing is wrong, not the model" return w, c # ===================== 4. MODELS + LONG-AUDIO-SAFE ASR ======================= # Ensembling only pays when systems make DIFFERENT mistakes. The int4/int8/full/ # dialectal Cohere builds are the same 2B model, so their errors correlate and # combining them gains little. This list mixes architectures on purpose. MODELS = { "cohere_base": dict(id="CohereLabs/cohere-transcribe-arabic-07-2026", family="cohere"), "whisper_dialect": dict(id="oddadmix/whisper-large-v3-turbo-arabic-dialectal", family="whisper"), "whisper_lv3": dict(id="openai/whisper-large-v3", family="whisper"), "cohere_dialect": dict(id="oddadmix/cohere-transcribe-arabic-07-2026-dialectal-v2", family="cohere"), "cohere_int8": dict(id="NAMAA-Space/cohere-transcribe-arabic-07-2026-int8", family="cohere"), "w2v2_xlsr": dict(id="jonatasgrosman/wav2vec2-large-xlsr-53-arabic", family="ctc"), } ACTIVE = ["cohere_base"] if QUICK_MODE else \ ["cohere_base", "whisper_dialect", "whisper_lv3", "cohere_dialect"] print("active systems:", ACTIVE) GEN = dict(max_new_tokens=440, num_beams=5, repetition_penalty=1.1, no_repeat_ngram_size=8) GEN_STRICT = dict(max_new_tokens=440, num_beams=1, repetition_penalty=1.3, no_repeat_ngram_size=5) BATCH = 8 _S = {} def unload(): global _S _S = {}; gc.collect() if HAS_GPU: torch.cuda.empty_cache() def load_model(key): global _S unload() mid, fam = MODELS[key]["id"], MODELS[key]["family"] if fam == "cohere": from transformers import AutoProcessor, CohereAsrForConditionalGeneration proc = AutoProcessor.from_pretrained(mid) mdl = CohereAsrForConditionalGeneration.from_pretrained(mid, device_map="auto").eval() dt = next((p.dtype for p in mdl.parameters() if p.dtype in (torch.float16, torch.bfloat16)), torch.float32) fe = proc.feature_extractor _S = dict(key=key, fam=fam, proc=proc, mdl=mdl, dt=dt, max_s=float(getattr(fe, "max_audio_clip_s", 35.0)), ov_s=float(getattr(fe, "overlap_chunk_second", 5.0))) print(f"{key}: {mid} | {dt} | max_clip {_S['max_s']}s overlap {_S['ov_s']}s") else: from transformers import pipeline _S = dict(key=key, fam=fam, pipe=pipeline("automatic-speech-recognition", model=mid, device=0, torch_dtype=torch.float16, chunk_length_s=30, stride_length_s=5)) print(f"{key}: {mid} | pipeline, 30s/5s chunking") return _S def _cast(d, dt): return {k: (v.to(dt) if (torch.is_tensor(v) and torch.is_floating_point(v)) else v) for k, v in d.items()} @torch.no_grad() def _cohere_decode(auds, gen): a = [np.asarray(x, dtype="float32") for x in auds] try: inp = _S["proc"](a, sampling_rate=SR, language="ar", return_tensors="pt", padding=True).to(_S["mdl"].device) inp.pop("audio_chunk_index", None) out = _S["mdl"].generate(**_cast(inp, _S["dt"]), **gen) txt = [t.strip() for t in _S["proc"].batch_decode(out, skip_special_tokens=True)] if len(txt) != len(a): # THE BUG: the processor chunked, so there are more rows than inputs. # zip() would silently drop them. Fall back to one clip at a time. raise RuntimeError("row count mismatch - per-clip fallback") return txt except Exception: res = [] for x in a: inp = _S["proc"](x, sampling_rate=SR, language="ar", return_tensors="pt").to(_S["mdl"].device) inp.pop("audio_chunk_index", None) o = _S["mdl"].generate(**_cast(inp, _S["dt"]), **gen) res.append(" ".join(t.strip() for t in _S["proc"].batch_decode(o, skip_special_tokens=True)).strip()) return res def _stitch(a, b, max_overlap_words=25): """Drop the longest word n-gram that ends `a` and starts `b` (the overlap).""" aw, bw = a.split(), b.split() if not aw or not bw: return " ".join(aw + bw) for k in range(min(max_overlap_words, len(aw), len(bw)), 1, -1): if aw[-k:] == bw[:k]: return " ".join(aw + bw[k:]) return " ".join(aw + bw) def _cohere_long(arr, gen): """Window past max_audio_clip_s explicitly and stitch, so nothing depends on the processor's row count lining up with the batch.""" win = int((_S["max_s"] - 1.0) * SR) hop = int((_S["max_s"] - 1.0 - _S["ov_s"]) * SR) pieces = [arr[s:s + win] for s in range(0, max(len(arr) - win + hop, 1), hop)] pieces = [p for p in pieces if len(p) > int(0.4 * SR)] outs = [] for k in range(0, len(pieces), 2): outs += _cohere_decode(pieces[k:k + 2], gen) text = outs[0] if outs else "" for nxt in outs[1:]: text = _stitch(text, nxt) return text def transcribe(ds, indices, jsonl, tag=""): """{dataset_index: raw_text}. Resume-safe: rerun after a disconnect and it continues.""" done = {} if os.path.exists(jsonl): for line in open(jsonl, encoding="utf-8"): r = json.loads(line); done[r["i"]] = r["raw"] todo = [i for i in indices if i not in done] if not todo: print(f"{tag}: all {len(indices)} cached"); return done dur = {i: ds[i]["audio"]["array"].shape[0] / SR for i in todo} f = open(jsonl, "a", encoding="utf-8") def _save(i, t): done[i] = t f.write(json.dumps({"i": i, "raw": t}, ensure_ascii=False) + "\n") f.flush(); os.fsync(f.fileno()) if _S["fam"] == "cohere": mx = _S["max_s"] long_i = [i for i in todo if dur[i] > mx] short_i = sorted([i for i in todo if dur[i] <= mx], key=lambda i: dur[i]) print(f"{tag}: {len(short_i)} short + {len(long_i)} long (windowed), {len(done)} cached") for k in tqdm(range(0, len(short_i), BATCH), desc=f"{tag} short"): b = short_i[k:k + BATCH] for i, t in zip(b, _cohere_decode([ds[i]["audio"]["array"] for i in b], GEN)): _save(i, t) for i in tqdm(long_i, desc=f"{tag} long"): _save(i, _cohere_long(ds[i]["audio"]["array"], GEN)) bad = [i for i in indices if max_word_run(FINAL_NORM(done.get(i, ""))) >= 4 or not FINAL_NORM(done.get(i, "")).strip()] if bad: print(f"{tag}: re-decoding {len(bad)} degenerate/empty") for i in bad: arr = ds[i]["audio"]["array"] alt = (_cohere_long(arr, GEN_STRICT) if arr.shape[0] / SR > mx else _cohere_decode([arr], GEN_STRICT)[0]) a, b2 = FINAL_NORM(done.get(i, "")), FINAL_NORM(alt) if b2.strip() and (not a.strip() or max_word_run(b2) < max_word_run(a)): _save(i, alt) else: kw = dict(generate_kwargs={"language": "ar", "task": "transcribe"}) \ if _S["fam"] == "whisper" else {} for i in tqdm(todo, desc=tag): out = _S["pipe"]({"raw": np.asarray(ds[i]["audio"]["array"], dtype="float32"), "sampling_rate": SR}, **kw) _save(i, (out["text"] or "").strip()) f.close() return done # =========================== 5. COMBINERS ==================================== def _ed(a, b): if a == b: return 0 la, lb = len(a), len(b) if not la or not lb: return max(la, lb) prev = list(range(lb + 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 mbr_pick(cands): """Keep the candidate closest to the others. Output is always a real hypothesis.""" cands = [c for c in cands if c.strip()] or [""] if len(cands) == 1: return cands[0] costs = [sum(_ed(c, o) / max(len(c), len(o), 1) for o in cands if o is not c) for c in cands] return cands[costs.index(min(costs))] def rover(cands, weights=None): """Align to a pivot, majority-vote each slot, deletions voting as epsilon.""" toks = [c.split() for c in cands] if weights is None: weights = [1.0] * len(toks) live = [(t, w) for t, w in zip(toks, weights) if t] if not live: return "" if len(live) == 1: return " ".join(live[0][0]) piv = max(range(len(live)), key=lambda k: len(live[k][0]) * live[k][1]) pt = live[piv][0] slots = [[(w, live[piv][1])] for w in pt] for k, (tk, wt) in enumerate(live): if k == piv: continue al = jiwer.process_words([" ".join(pt)], [" ".join(tk)]).alignments[0] covered = set() for ch in al: if ch.type in ("equal", "substitute"): for off, r in enumerate(range(ch.ref_start_idx, ch.ref_end_idx)): h = ch.hyp_start_idx + off if r < len(slots) and h < len(tk): slots[r].append((tk[h], wt)); covered.add(r) elif ch.type == "delete": for r in range(ch.ref_start_idx, ch.ref_end_idx): if r < len(slots): slots[r].append((None, wt)); covered.add(r) for r in range(len(slots)): if r not in covered: slots[r].append((None, wt)) out = [] for s in slots: tally = {} for w, wt in s: tally[w] = tally.get(w, 0.0) + wt best = max(tally.items(), key=lambda kv: (kv[1], kv[0] is not None)) if best[0] is not None: out.append(best[0]) return " ".join(out) assert rover(["a b c", "a b c", "a x c"]) == "a b c" assert rover(["a b c", "a c", "a b c"]) == "a b c" assert rover(["a b c", "a b z c", "a b c"]) == "a b c" assert rover(["a b c", "", "a b c"]) == "a b c" assert mbr_pick(["a b c", "a b c", "z z z"]) == "a b c" print("combiners verified") def combine(sysdict, idxs, method, keys, dev_wer=None): wts = [1.0 / max(dev_wer[k][0], 1e-3) for k in keys] if (method == "wrover" and dev_wer) else None out = {} for i in idxs: c = [FINAL_NORM(sysdict[k].get(i, "")) for k in keys] out[i] = mbr_pick(c) if method == "mbr" else rover(c, wts) return out # ====================== 6. RUN ON DEV, PICK A WINNER ========================= DEV_SYS, DEV_SCORE = {}, {} BEST_SINGLE, WINNER, RESULTS = ACTIVE[0], ACTIVE[0], {} if RUN_DEV: dev_ds = load_dataset(DEV_REPO, split="dev").cast_column("audio", Audio(sampling_rate=SR)) assert len(dev_ds) == DEV_N and list(dev_ds["id"]) == dev_ids, "dev row order changed" for key in ACTIVE: load_model(key) DEV_SYS[key] = transcribe(dev_ds, DEV_EVAL, f"out/dev_{key}.jsonl", tag=f"dev/{key}") unload() print("\n=== single systems on the dev subset ===") for key in ACTIVE: DEV_SCORE[key] = score(DEV_SYS[key], DEV_SEL, key + " [blind-like]") BEST_SINGLE = min(DEV_SCORE, key=lambda k: DEV_SCORE[k][0]) WINNER = BEST_SINGLE print(f"\nbest single: {BEST_SINGLE} WER {DEV_SCORE[BEST_SINGLE][0]:.4f}") _lo_eval = DEV_LONG if _lo_eval: hw = sum(len(FINAL_NORM(DEV_SYS[BEST_SINGLE][i]).split()) for i in _lo_eval) rw = sum(len(DEV_REFS[i].split()) for i in _lo_eval) print(f"long-clip length ratio {hw/rw:.2f} over {len(_lo_eval)} clips " f"(0.71 before the windowing fix, 1.00 is the target)") if len(ACTIVE) > 1: RESULTS = {k: DEV_SCORE[k][0] for k in ACTIVE} print("\n=== combinations ===") for m in ("mbr", "rover", "wrover"): RESULTS[m] = score(combine(DEV_SYS, DEV_SEL, m, ACTIVE, DEV_SCORE), DEV_SEL, f"[{m}] all {len(ACTIVE)} systems", norm=lambda t: t)[0] if len(ACTIVE) > 2: print() for drop in ACTIVE: keys = [k for k in ACTIVE if k != drop] RESULTS["mbr::" + ",".join(keys)] = score( combine(DEV_SYS, DEV_SEL, "mbr", keys, DEV_SCORE), DEV_SEL, f"[mbr] without {drop}", norm=lambda t: t)[0] orc = {i: min([FINAL_NORM(DEV_SYS[k].get(i, "")) for k in ACTIVE], key=lambda c: jiwer.process_words([DEV_REFS[i]], [c or " "]).wer) for i in DEV_SEL} print() score(orc, DEV_SEL, "[oracle] best candidate per clip (CEILING)", norm=lambda t: t) WINNER = min(RESULTS, key=lambda k: RESULTS[k]) print(f"\n>>> winner: {WINNER} WER {RESULTS[WINNER]:.4f}" f" (best single {BEST_SINGLE} {DEV_SCORE[BEST_SINGLE][0]:.4f})") if RESULTS[WINNER] >= DEV_SCORE[BEST_SINGLE][0] - 0.002: WINNER = BEST_SINGLE print(" gain is inside this subset's noise - falling back to the single model") print("\nsubmitting with:", WINNER) # ==================== 7. BLIND SET + SUBMISSION ============================== test_ds = load_dataset(TEST_REPO, split="test").cast_column("audio", Audio(sampling_rate=SR)) test_ids = list(test_ds["id"]) assert len(test_ds) == N_TEST, f"expected {N_TEST} clips, got {len(test_ds)}" if WINNER in ("mbr", "rover", "wrover"): NEEDED, METHOD = ACTIVE, WINNER elif WINNER.startswith("mbr::"): NEEDED, METHOD = WINNER[5:].split(","), "mbr" else: NEEDED, METHOD = [WINNER], None print("systems needed:", NEEDED, "| method:", METHOD or "single") TEST_SYS = {} for key in NEEDED: load_model(key) TEST_SYS[key] = transcribe(test_ds, list(range(N_TEST)), f"out/test_{key}.jsonl", tag=f"blind/{key}") unload() if METHOD: merged = combine(TEST_SYS, range(N_TEST), METHOD, NEEDED, DEV_SCORE) lines = [merged[i] for i in range(N_TEST)] else: lines = [FINAL_NORM(TEST_SYS[WINNER][i]) for i in range(N_TEST)] lines = [l if l.strip() else _FILLER for l in lines] _BUILT = set() def build(name, lines, note): txt, zp = f"out/{name}.txt", f"out/{name}.zip" _BUILT.add(f"{name}.txt"); _BUILT.add(f"{name}.zip") with open(txt, "w", encoding="utf-8", newline="\n") as f: f.write("\n".join(lines) + "\n") if os.path.exists(zp): os.remove(zp) with zipfile.ZipFile(zp, "w", zipfile.ZIP_DEFLATED) as z: z.write(txt, "predictions.txt") # exactly one file, at the zip root with zipfile.ZipFile(zp) as z: inside = [(i.filename, i.file_size) for i in z.infolist()] assert len(inside) == 1 and inside[0][0] == "predictions.txt" and inside[0][1] > 2000, inside assert len(lines) == N_TEST, f"{len(lines)} != {N_TEST}" assert all(l.strip() for l in lines), "blank line" assert all(l == l.strip() for l in lines), "untrimmed line" print(f" {name:38s} lines={len(lines)} words={sum(len(l.split()) for l in lines):5d} " f"zip={os.path.getsize(zp)}B [{note}]") return zp # line k is the transcript of DATASET row k - never sorted(ids). That was the 0.9-WER bug. FINAL = build(f"predictions_{WINNER.replace(':', '_').replace(',', '-')}", lines, "UPLOAD THIS") for key in TEST_SYS: build(f"single_{key}", [FINAL_NORM(TEST_SYS[key][i]) for i in range(N_TEST)], "fallback") durs = [test_ds[i]["audio"]["array"].shape[0] / SR for i in range(N_TEST)] _r = sorted(len(lines[i].split()) / durs[i] for i in range(N_TEST) if durs[i] <= 35) _med = _r[len(_r) // 2] print(f"\nlong-clip check (median {_med:.2f} w/s under 35 s):") for i in [i for i in range(N_TEST) if durs[i] > 35]: w = len(lines[i].split()) print(f" row {i:3d} {durs[i]:5.1f}s -> {w:3d} words ({w/durs[i]:.2f} w/s)" + (" <-- STILL TRUNCATED" if w / durs[i] < 0.5 * _med else " ok")) _txt = open(FINAL.replace(".zip", ".txt"), encoding="utf-8").read() print(f"junk check: char runs 4+ {len(re.findall(chr(40)+chr(46)+chr(41)+chr(92)+'1{3,}', _txt))} | " f"'#' {_txt.count('#')} | latin {len(re.findall('[A-Za-z]', _txt))} | " f"word-repeat runs>=4 {sum(1 for l in _txt.splitlines() if max_word_run(l) >= 4)}") # ============================== 8. PUSH ====================================== if HF_REPO: for fn in sorted(os.listdir("out")): p = f"out/{fn}" # only this run's artifacts - earlier runs can leave blanked 374-byte zips # in out/, and pushing them makes it easy to upload the wrong file if fn.endswith((".txt", ".zip")) and fn not in _BUILT: print("skipped stale", fn); continue if os.path.isfile(p) and os.path.getsize(p) < 200 * 1024 * 1024: dest = f"claude/cache/{fn}" if fn.endswith(".jsonl") else f"{PUSH_PREFIX}/{fn}" api.upload_file(path_or_fileobj=p, path_in_repo=dest, repo_id=HF_REPO, commit_message=f"one-cell pipeline, winner={WINNER}") print("pushed", dest) try: from google.colab import files; files.download(FINAL) except Exception: print("download manually:", FINAL) print("\nDONE. Upload", FINAL, "to Codabench.") print("Next: set QUICK_MODE = False and rerun - cached transcripts are reused, " "so only the new models cost GPU time.")