Upload eval_darijabert_mix.py with huggingface_hub

eval_darijabert_mix.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3 -u
+"""eval_darijabert_mix.py — Evaluate DarijaBERT-mix on the same test set and append to external_comparison.csv."""
+
+import json, os, csv, gc, warnings
+from dataclasses import dataclass, asdict
+
+import numpy as np
+import regex
+warnings.filterwarnings("ignore")
+
+BASE = "/root/oiq_cc_tokenizer/results"
+CORPORA = os.path.join(BASE, "corpora")
+PLOTS_DIR = os.path.join(BASE, "plots")
+
+HF_TOKEN = os.environ.get("HF_TOKEN", "")
+
+_WORD_PAT = regex.compile(r"[\p{L}\p{M}\p{N}]+", regex.UNICODE)
+_AR_PAT = regex.compile(r"[\u0600-\u06FF\u0750-\u077F]")
+_SPECIAL = {"<unk>", "<s>", "</s>", "[CLS]", "[SEP]", "[PAD]", "[UNK]", "<pad>",
+            "<|endoftext|>", "<|im_start|>", "<|im_end|>"}
+
+def segment_words(t): return _WORD_PAT.findall(t)
+def count_graphemes(t): return len(regex.findall(r"\X", t))
+def detect_script(t): return "ar" if len(_AR_PAT.findall(t)) > len(t) * 0.3 else "az"
+def filter_sp(tokens): return [t for t in tokens if t not in _SPECIAL]
+def normalize_decode(s):
+    s = s.replace("##", "")
+    s = " ".join(s.split())
+    return s
+
+
+@dataclass
+class M:
+    name: str = ""
+    source: str = ""
+    algorithm: str = ""
+    architecture: str = ""
+    vocab_size: int = 0
+    fertility_ar: float = 0.0
+    fertility_az: float = 0.0
+    fertility_overall: float = 0.0
+    disparity: float = 0.0
+    cpt_ar: float = 0.0
+    cpt_az: float = 0.0
+    exact_match_ar: float = 0.0
+    exact_match_az: float = 0.0
+
+
+def evaluate(tok, name, source, algo, arch, vsz, texts):
+    m = M(name=name, source=source, algorithm=algo, architecture=arch, vocab_size=vsz)
+    ar_f, az_f, all_f = [], [], []
+    ar_c, az_c = [], []
+    ar_ok, az_ok, ar_n, az_n = 0, 0, 0, 0
+
+    for i, text in enumerate(texts):
+        if (i + 1) % 5000 == 0:
+            print(f"    [{i+1}/{len(texts)}] {name}", flush=True)
+        try:
+            ids = tok.encode(text, add_special_tokens=False)
+            tokens = tok.convert_ids_to_tokens(ids)
+            content = filter_sp(tokens)
+            words = segment_words(text)
+            if not words:
+                continue
+            fert = len(content) / len(words)
+            all_f.append(fert)
+            cpt = count_graphemes(text) / max(len(content), 1)
+            try:
+                dec = tok.decode(ids, skip_special_tokens=True)
+                exact = normalize_decode(dec) == normalize_decode(text)
+            except:
+                exact = False
+            script = detect_script(text)
+            if script == "ar":
+                ar_f.append(fert); ar_c.append(cpt); ar_n += 1
+                if exact: ar_ok += 1
+            else:
+                az_f.append(fert); az_c.append(cpt); az_n += 1
+                if exact: az_ok += 1
+        except:
+            pass
+
+    m.fertility_ar = float(np.mean(ar_f)) if ar_f else 0
+    m.fertility_az = float(np.mean(az_f)) if az_f else 0
+    m.fertility_overall = float(np.mean(all_f)) if all_f else 0
+    mx = max(m.fertility_ar, m.fertility_az, 1e-9)
+    m.disparity = abs(m.fertility_ar - m.fertility_az) / mx
+    m.cpt_ar = float(np.mean(ar_c)) if ar_c else 0
+    m.cpt_az = float(np.mean(az_c)) if az_c else 0
+    m.exact_match_ar = ar_ok / max(ar_n, 1)
+    m.exact_match_az = az_ok / max(az_n, 1)
+    return m
+
+
+def main():
+    from transformers import AutoTokenizer
+
+    # Load test texts
+    texts = []
+    for s in ("test_ar", "test_az", "test_mi"):
+        p = os.path.join(CORPORA, f"{s}.txt")
+        if os.path.exists(p):
+            with open(p) as f:
+                texts.extend(l.strip() for l in f if l.strip())
+    print(f"{len(texts)} test texts", flush=True)
+
+    # Load DarijaBERT-mix
+    repo = "SI2M-Lab/DarijaBERT-mix"
+    print(f"\nLoading {repo} ...", flush=True)
+    tok = AutoTokenizer.from_pretrained(repo, trust_remote_code=True)
+    vsz = tok.vocab_size
+    print(f"  vocab_size = {vsz}", flush=True)
+
+    # Evaluate
+    print(f"\nEvaluating DarijaBERT-mix ...", flush=True)
+    r = evaluate(tok, "DarijaBERT-mix", "external_darija", "WordPiece", "shared", vsz, texts)
+    print(f"  F={r.fertility_overall:.3f}  F_ar={r.fertility_ar:.3f}  F_az={r.fertility_az:.3f}", flush=True)
+    print(f"  D={r.disparity:.3f}  CPT_ar={r.cpt_ar:.3f}  CPT_az={r.cpt_az:.3f}", flush=True)
+    print(f"  EM_ar={r.exact_match_ar:.2%}  EM_az={r.exact_match_az:.2%}", flush=True)
+
+    # Append to external_comparison.csv
+    csv_path = os.path.join(BASE, "external_comparison.csv")
+    with open(csv_path, "a", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=list(asdict(r).keys()))
+        w.writerow(asdict(r))
+    print(f"\nAppended to {csv_path}", flush=True)
+
+    print("DONE!", flush=True)
+
+
+if __name__ == "__main__":
+    main()