Upload indus_ngram.py

indus_ngram.py

@@ -0,0 +1,234 @@
+"""
+indus_ngram.py — Standalone module for InduNgramModel
+======================================================
+This file MUST exist so that pickle can import InduNgramModel
+when loading ngram_model.pkl in 07_ensemble.py and 08_electra_train.py.
+
+The pickle fix:
+  When you save a class with pickle, Python records the module path.
+  If the class was defined in __main__ (i.e. inside 06_ngram_model.py
+  when run directly), pickle saves it as __main__.InduNgramModel.
+  When another script tries to load it, __main__ refers to THAT script,
+  which doesn't have InduNgramModel — hence the AttributeError.
+
+  Solution: define the class in THIS standalone module (indus_ngram.py).
+  Both 06_ngram_model.py and 07_ensemble.py import from here.
+  Pickle records the path as indus_ngram.InduNgramModel — always findable.
+
+Do not rename or move this file.
+"""
+
+import math
+import pickle
+from pathlib import Path
+from collections import Counter, defaultdict
+
+
+class InduNgramModel:
+    """
+    Kneser-Ney smoothed N-gram LM for Indus Script.
+
+    RTL mode (default, recommended):
+      Sequences reversed before training/scoring.
+      RTL bigram entropy (3.18) < LTR (3.72) → supports RTL hypothesis.
+
+    Sign roles in RTL reading direction:
+      INITIAL  = reading-start sign (data position [-1])
+      TERMINAL = reading-end sign   (data position [0])
+      MEDIAL   = appears in middle positions
+    """
+
+    def __init__(self, rtl=True):
+        self.rtl             = rtl
+        self.unigram         = Counter()
+        self.bigram          = defaultdict(Counter)
+        self.trigram         = defaultdict(Counter)
+        self.start_cnt       = Counter()
+        self.end_cnt         = Counter()
+        self.total_seqs      = 0
+        self.total_tokens    = 0
+        self.vocab_size      = 0
+        self.D               = 0.75
+        self.score_mean      = 0.0
+        self.score_std       = 1.0
+        self.score_min       = -20.0
+        self.score_max       = 0.0
+        self._pairwise_acc   = 0.0
+        self._cont_right     = Counter()
+        self._total_bi_types = 0
+
+    def _orient(self, seq):
+        cleaned = [t for t in seq if t is not None]
+        return list(reversed(cleaned)) if self.rtl else list(cleaned)
+
+    def train(self, sequences):
+        mode = "RTL" if self.rtl else "LTR"
+        print(f"  Training [{mode}] on {len(sequences):,} sequences...")
+
+        for seq in sequences:
+            s = self._orient(seq)
+            if not s:
+                continue
+            self.total_seqs += 1
+            self.unigram.update(s)
+            self.start_cnt[s[0]]  += 1
+            self.end_cnt[s[-1]]   += 1
+            for i in range(len(s) - 1):
+                self.bigram[s[i]][s[i+1]] += 1
+            for i in range(len(s) - 2):
+                self.trigram[(s[i], s[i+1])][s[i+2]] += 1
+
+        self.total_tokens    = sum(self.unigram.values())
+        self.vocab_size      = len(self.unigram)
+
+        for a, followers in self.bigram.items():
+            for b in followers:
+                self._cont_right[b] += 1
+        self._total_bi_types = sum(self._cont_right.values())
+
+        self._calibrate(sequences)
+        print(f"  Vocab      : {self.vocab_size}")
+        print(f"  Pairwise   : {self._pairwise_acc*100:.1f}%")
+        print(f"  Score range: [{self.score_min:.3f}, {self.score_max:.3f}]")
+
+    def _calibrate(self, sequences):
+        import random, statistics
+        random.seed(42)
+        all_toks = list(self.unigram.keys())
+
+        def corrupt(seq):
+            r = random.randint(0, 3)
+            c = list(seq)
+            if r == 0:
+                random.shuffle(c)
+            elif r == 1:
+                c[0] = random.choice(list(self.end_cnt.keys()))
+            elif r == 2:
+                c[-1] = random.choice(list(self.start_cnt.keys()))
+            else:
+                for p in random.sample(range(len(c)), max(1, len(c)//2)):
+                    c[p] = random.choice(all_toks)
+            return c
+
+        sample = sequences[:500]
+        good   = [self._raw_score(s) for s in sample]
+        bad    = [self._raw_score(corrupt(s)) for s in sample]
+        all_s  = good + bad
+
+        self.score_mean    = statistics.mean(all_s)
+        self.score_std     = statistics.stdev(all_s)
+        self.score_min     = min(all_s)
+        self.score_max     = max(all_s)
+        self._pairwise_acc = sum(g > b for g, b in zip(good, bad)) / len(good)
+
+    def _p_uni_kn(self, w):
+        return (self._cont_right[w] + 1) / (self._total_bi_types + self.vocab_size)
+
+    def _p_bi_kn(self, w, given):
+        gt = sum(self.bigram[given].values())
+        if gt == 0:
+            return self._p_uni_kn(w)
+        cnt   = self.bigram[given].get(w, 0)
+        first = max(cnt - self.D, 0) / gt
+        lam   = (self.D / gt) * len(self.bigram[given])
+        return first + lam * self._p_uni_kn(w)
+
+    def _p_tri_kn(self, w, a, b):
+        gt = sum(self.trigram[(a, b)].values())
+        if gt == 0:
+            return self._p_bi_kn(w, b)
+        cnt   = self.trigram[(a, b)].get(w, 0)
+        first = max(cnt - self.D, 0) / gt
+        lam   = (self.D / gt) * len(self.trigram[(a, b)])
+        return first + lam * self._p_bi_kn(w, b)
+
+    def _p_initial(self, w):
+        return (self.start_cnt[w] + 0.1) / (self.total_seqs + 0.1 * self.vocab_size)
+
+    def _p_terminal(self, w):
+        return (self.end_cnt[w] + 0.1) / (self.total_seqs + 0.1 * self.vocab_size)
+
+    def _raw_score(self, seq):
+        if not seq:
+            return self.score_min
+        s   = self._orient(seq)
+        eps = 1e-10
+        lp  = math.log(self._p_initial(s[0])  + eps)
+        lp += math.log(self._p_terminal(s[-1]) + eps)
+        for i, w in enumerate(s):
+            if i == 0:
+                p = self._p_uni_kn(w)
+            elif i == 1:
+                p = self._p_bi_kn(w, s[i-1])
+            else:
+                p = self._p_tri_kn(w, s[i-2], s[i-1])
+            lp += math.log(p + eps)
+        return lp / (len(s) + 2)
+
+    def validity_score(self, seq):
+        raw  = self._raw_score(seq)
+        norm = (raw - self.score_min) / (self.score_max - self.score_min + 1e-10)
+        return float(max(0.02, min(0.98, norm)))
+
+    def predict_masked(self, seq_with_none, top_k=10):
+        masked = [i for i, t in enumerate(seq_with_none) if t is None]
+        results = {}
+        n = len(seq_with_none)
+
+        for orig_pos in masked:
+            ort_pos  = (n - 1 - orig_pos) if self.rtl else orig_pos
+            oriented = self._orient(seq_with_none)
+            if ort_pos >= len(oriented):
+                continue
+
+            prev  = oriented[ort_pos-1] if ort_pos > 0 and oriented[ort_pos-1] is not None else None
+            prev2 = oriented[ort_pos-2] if ort_pos > 1 and oriented[ort_pos-2] is not None else None
+
+            cands = []
+            for cand in self.unigram:
+                if prev2 is not None and prev is not None:
+                    p = self._p_tri_kn(cand, prev2, prev)
+                elif prev is not None:
+                    p = self._p_bi_kn(cand, prev)
+                else:
+                    p = self._p_uni_kn(cand)
+
+                if ort_pos == 0:
+                    p *= max(self._p_initial(cand) * self.vocab_size, 0.01)
+                elif ort_pos == n - 1:
+                    p *= max(self._p_terminal(cand) * self.vocab_size, 0.01)
+
+                cands.append((cand, p))
+
+            cands.sort(key=lambda x: -x[1])
+            total = sum(p for _, p in cands[:top_k * 3]) or 1
+            results[orig_pos] = [
+                {"id": c, "prob": p / total, "rank": i + 1}
+                for i, (c, p) in enumerate(cands[:top_k])
+            ]
+
+        return results
+
+    def sign_role(self, sign_id):
+        """Positional role in reading direction."""
+        init_p = self.start_cnt[sign_id] / (self.total_seqs + 1)
+        term_p = self.end_cnt[sign_id]   / (self.total_seqs + 1)
+        if init_p > 0.05 and init_p > term_p * 2:
+            return "INITIAL"
+        elif term_p > 0.05 and term_p > init_p * 2:
+            return "TERMINAL"
+        elif self.unigram[sign_id] > 5:
+            return "MEDIAL"
+        return "RARE"
+
+    def save(self, path):
+        path = Path(path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(path, "wb") as f:
+            pickle.dump(self, f, protocol=pickle.HIGHEST_PROTOCOL)
+        print(f"  Saved → {path}")
+
+    @staticmethod
+    def load(path):
+        with open(path, "rb") as f:
+            return pickle.load(f)