File size: 6,169 Bytes
7ed86c3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """Encode the macron-data corpora into letter-plane npz stores.
python -m meter.encode [--out $METER_DATA/encoded] [--src $MACRONIZER_SRC] [--norma-source hf]
Sources (all under --src/data, except Norma -- see --norma-source):
macron TSVs plain \t marked (verse silver + OGA prose silver)
dev.txt marked lines (763 Aristophanic verses, macron dev set)
scanner/corpus_v3.tsv work \t meter \t bracketed verse
Anything whose letter stream overlaps the Norma benchmark (both tasks, dev+test) or
dev.txt is EXCLUDED from training stores: a record is dropped if any 20-letter
shingle of an eval line occurs in it (shorter eval lines: exact letter-stream match).
Slightly over-eager by design — dropping a few extra silver lines is free, leakage
is not.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import unicodedata
from pathlib import Path
from meter.backbone import ALPHABET # noqa: F401 (puts STOICHEIA_ROOT on sys.path)
from meter.dataset import encode_macron_line, encode_scan_line, save_records
from meter.marks import parse_macron_line, parse_scan_line
from meter.norma_data import add_norma_source_arg, load_norma
MACRON_TSVS = ["hypotactic", "drama_ia6", "drama_ia6_tet", "anthology",
"nonnus_quintus", "babrius_chol", "theocritus_doric",
"theocritus_other", "sweep1_hex", "sweep1_ia6", "sweep1_chol",
"sweep1_eleg", "oga_0", "oga_1", "oga_2", "oga_3"]
SHINGLE = 20
def letters_of(rec) -> str:
return "".join(ALPHABET[c] for c in rec.chars)
ALPHABET_SET = set(ALPHABET) | {"ς", "ϲ"}
def _letters_only(plain: str) -> str:
return "".join(ch for ch in unicodedata.normalize("NFD", plain).lower()
if ch in ALPHABET_SET)
def eval_letter_streams(src: Path, norma_source: str = "hf"):
"""Letter streams of every eval line (Norma macronize+syllabify -- dev+test both,
so the exclusion screen covers everything Norma could ever score us against --
plus dev.txt)."""
streams = []
norma = load_norma(norma_source)
for d in norma["dev"] + norma["test"]:
parsed = (parse_scan_line(d["text"]) if d["task"] == "syllabify"
else parse_macron_line(d["text"]))
if parsed is not None:
streams.append(_letters_only(parsed[0]))
for line in open(src / "data/dev.txt", encoding="utf-8"):
streams.append(_letters_only(parse_macron_line(line.rstrip("\n"))[0]))
return [s for s in streams if s]
def build_screen(streams):
shingles, exact = set(), set()
for s in streams:
s = s.replace("ς", "σ").replace("ϲ", "σ")
if len(s) >= SHINGLE:
for i in range(len(s) - SHINGLE + 1):
shingles.add(s[i:i + SHINGLE])
else:
exact.add(s)
return shingles, exact
def is_contaminated(letters: str, shingles, exact) -> bool:
if letters in exact:
return True
for i in range(len(letters) - SHINGLE + 1):
if letters[i:i + SHINGLE] in shingles:
return True
return False
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--src", default=os.path.expandvars(os.environ.get(
"MACRONIZER_SRC", "$MACRONIZER_SRC")))
ap.add_argument("--out", default=None)
add_norma_source_arg(ap)
a = ap.parse_args()
src = Path(a.src)
out = Path(a.out or os.path.join(os.environ["METER_DATA"], "encoded"))
out.mkdir(parents=True, exist_ok=True)
print("building eval exclusion screen ...", flush=True)
shingles, exact = build_screen(eval_letter_streams(src, a.norma_source))
print(f" {len(shingles):,} shingles, {len(exact)} exact keys", flush=True)
stats = {}
def finish(name, kept, dropped, excluded, works=None):
save_records(out / f"{name}.npz", kept, works)
n_mac = sum(int((r.y_mac != -100).sum()) for r in kept)
stats[name] = dict(records=len(kept), dropped=dropped, excluded=excluded,
letters=sum(len(r) for r in kept), mac_labels=n_mac)
print(f" {name}: kept={len(kept):,} dropped={dropped:,} "
f"excluded={excluded:,} mac_labels={n_mac:,}", flush=True)
# ---- macron TSVs (train)
for name in MACRON_TSVS:
path = src / "data" / f"{name}.tsv"
if not path.exists():
print(f" {name}: MISSING, skipped", flush=True)
continue
kept, dropped, excluded = [], 0, 0
for line in open(path, encoding="utf-8"):
parts = line.rstrip("\n").split("\t")
if len(parts) < 2:
dropped += 1
continue
rec = encode_macron_line(parts[1])
if rec is None:
dropped += 1
continue
if is_contaminated(letters_of(rec), shingles, exact):
excluded += 1
continue
kept.append(rec)
finish(name, kept, dropped, excluded)
# ---- macron dev (no exclusion screen — it IS an eval set)
kept, dropped = [], 0
for line in open(src / "data/dev.txt", encoding="utf-8"):
rec = encode_macron_line(line.rstrip("\n"))
if rec is None:
dropped += 1
continue
kept.append(rec)
finish("dev_aristophanes", kept, dropped, 0)
# ---- scanner corpus (train/dev/test split by work happens at load time)
kept, works, dropped, excluded = [], [], 0, 0
for line in open(src / "data/scanner/corpus_v3.tsv", encoding="utf-8"):
parts = line.rstrip("\n").split("\t")
if len(parts) != 3 or parts[0] == "?":
dropped += 1
continue
rec = encode_scan_line(parts[2])
if rec is None:
dropped += 1
continue
if is_contaminated(letters_of(rec), shingles, exact):
excluded += 1
continue
kept.append(rec)
works.append(parts[0])
finish("scan_corpus", kept, dropped, excluded, works)
(out / "stats.json").write_text(json.dumps(stats, indent=1))
print("done:", out, flush=True)
if __name__ == "__main__":
sys.exit(main())
|