#!/usr/bin/env python3 """ Sylheti Dialectometer — Production Inference Engine v1.0 Phase 3, Step 6 — HuggingFace Deployment Build Self-contained: all assets loaded from Path(__file__).parent No global state: all lookup tables live inside DialectometerEngine.__init__ """ import json, math, re, unicodedata from pathlib import Path from typing import Dict, List, Tuple, Optional class DialectometerEngine: """ Self-contained inference engine for Sylheti dialect intensity scoring. Architecture: 5-stage pipeline Stage 1 : Pre-processing (Regex cleaning, NFC normalisation, word limit) Stage 2 : BPE Tokenisation (Byte-level with fallback) Stage 3 : Feature Extraction (weight lookup, position flags) Stage 4 : Granular Scoring (IG-blended, suffix-boosted, log-damped) Stage 5 : Calibrated Output (piecewise scaling, confidence damping) """ version = "1.0.0" MAX_WORDS = 100 # ────────────────────────────────────────────────────────────────── # §0 Constructor — loads ALL assets and builds lookup tables # ────────────────────────────────────────────────────────────────── def __init__(self, config_path: Optional[str] = None, tokenizer_path: Optional[str] = None): base = Path(__file__).parent cfg_path = Path(config_path) if config_path else base / "final_production_bundle.json" tok_path = Path(tokenizer_path) if tokenizer_path else base / "tokenizer.json" with open(cfg_path, "r", encoding="utf-8") as f: model_config = json.load(f) with open(tok_path, "r", encoding="utf-8") as f: tok_data = json.load(f) # ── Scoring parameters ──────────────────────────────────────── sp = model_config["scoring_parameters"] self.SIGNAL_THRESHOLD = sp["signal_threshold"] self.SUFFIX_AMP = sp["suffix_amplification"] self.SUFFIX_STACK_CAP = sp["suffix_stack_cap"] self.SIGMOID_K = sp["sigmoid_k"] self.SIGMOID_MID = sp["sigmoid_midpoint"] self.DETECTION_THRESH = sp["detection_threshold"] self.MIN_SIGNAL_TOKENS = sp["min_signal_tokens"] self.IG_ALPHA = sp["ig_alpha"] self.CONF_MIN_SIGNALS = sp["confidence_min_signals"] self.WORD_FINAL_MILD = sp["word_final_mild_boost"] # ── Calibration parameters ──────────────────────────────────── cp = model_config["calibration_parameters"] self.DISPLAY_FLOOR = cp["display_floor"] self.SATURATION_POINT = cp["saturation_point"] self.CONF_DAMPING = cp["confidence_damping"] self.CAT_BOUNDARIES = cp["category_boundaries"] # ── Weight & ID lookups ─────────────────────────────────────── self.WEIGHTS = {int(k): float(v) for k, v in model_config["weights"].items()} self.IG_WEIGHTS = {int(k): float(v) for k, v in model_config["ig_weights"].items()} self.STRICT_IDS = set(model_config["strict_ids"]) self.SUFFIX_IDS = set(model_config["suffix_match_ids"]) # ── ByteLevel BPE byte↔unicode mapping (GPT-2 standard) ─────── _bs = (list(range(ord('!'), ord('~') + 1)) + list(range(ord('¡'), ord('¬') + 1)) + list(range(ord('®'), ord('ÿ') + 1))) _cs = _bs[:] _n = 0 for b in range(256): if b not in _bs: _bs.append(b) _cs.append(256 + _n) _n += 1 self.BYTE_TO_UNICODE = {b: chr(c) for b, c in zip(_bs, _cs)} self.UNICODE_TO_BYTE = {chr(c): b for b, c in zip(_bs, _cs)} # ── Vocabulary & merge table ────────────────────────────────── raw_vocab = tok_data["model"]["vocab"] merges_list = tok_data["model"]["merges"] self.MERGE_RANKS: Dict[Tuple[str, str], int] = {} for i, m in enumerate(merges_list): pair = tuple(m) if isinstance(m, list) else tuple(m.split()) self.MERGE_RANKS[pair] = i self.RAW_TO_ID = {tok: vid for tok, vid in raw_vocab.items()} self.DECODED_VOCAB = {vid: self._decode_bpe_token_static(tok_str, self.UNICODE_TO_BYTE) for tok_str, vid in raw_vocab.items()} self.VOCAB_SIZE = len(raw_vocab) # ── Token type classification sets ──────────────────────────── BANGLA_RANGE = lambda c: '\u0980' <= c <= '\u09FF' self.LATIN_IDS = set() self.SPECIAL_IDS = set() self.PUNCT_IDS = set() self.BANGLA_IDS = set() for vid, d in self.DECODED_VOCAB.items(): if d.startswith('[') and d.endswith(']'): self.SPECIAL_IDS.add(vid) elif all(c.isascii() for c in d): if all(c.isalpha() for c in d) and len(d) > 0: self.LATIN_IDS.add(vid) elif any(c in d for c in '.,;:!?।'): self.PUNCT_IDS.add(vid) elif any(BANGLA_RANGE(c) for c in d): self.BANGLA_IDS.add(vid) # ── Pre-compiled regex patterns ─────────────────────────────── self.RE_ZERO_WIDTH = re.compile( r'[\u200B\u200C\u200D\uFEFF\u00AD\u200E\u200F\u202A-\u202E\u2060-\u2064]' ) self.RE_EMOJI = re.compile( "[" "\U0001F600-\U0001F64F" "\U0001F300-\U0001F5FF" "\U0001F680-\U0001F6FF" "\U0001F1E0-\U0001F1FF" "\U00002702-\U000027B0" "\U0001F900-\U0001F9FF" "\U0001FA00-\U0001FA6F" "\U0001FA70-\U0001FAFF" "\U00002600-\U000026FF" "\U0000FE00-\U0000FE0F" "]+", flags=re.UNICODE, ) self.RE_MULTI_SPACE = re.compile(r'\s+') self._inference_count = 0 # ────────────────────────────────────────────────────────────────── # §1 BPE Helper (static, used during __init__) # ────────────────────────────────────────────────────────────────── @staticmethod def _decode_bpe_token_static(tok_str: str, unicode_to_byte: dict) -> str: try: return bytes([unicode_to_byte[c] for c in tok_str]).decode('utf-8', errors='replace') except (KeyError, UnicodeDecodeError): return tok_str # ────────────────────────────────────────────────────────────────── # §2 Stage 1 — Pre-processing # ────────────────────────────────────────────────────────────────── def _preprocess(self, text: str) -> dict: trace = { "original_length": len(text), "original_words": len(text.split()) if text.strip() else 0, "modifications": [], } # NFC normalisation text_nfc = unicodedata.normalize('NFC', text) if text_nfc != text: trace["modifications"].append("nfc_normalized") text = text_nfc # Zero-width character removal cleaned = self.RE_ZERO_WIDTH.sub('', text) if cleaned != text: trace["modifications"].append(f"zwc_removed:{len(text)-len(cleaned)}_chars") text = cleaned # Emoji removal cleaned = self.RE_EMOJI.sub('', text) if cleaned != text: trace["modifications"].append("emoji_removed") text = cleaned # Whitespace collapse cleaned = self.RE_MULTI_SPACE.sub(' ', text).strip() if cleaned != text.strip(): trace["modifications"].append("whitespace_collapsed") text = cleaned # Word-count hard limit words = text.split() truncated = False if len(words) > self.MAX_WORDS: words = words[:self.MAX_WORDS] text = ' '.join(words) truncated = True trace["modifications"].append(f"truncated_to_{self.MAX_WORDS}_words") trace["cleaned_length"] = len(text) trace["cleaned_words"] = len(words) trace["truncated"] = truncated return {"text": text, "trace": trace} # ────────────────────────────────────────────────────────────────── # §3 Stage 2 — BPE Tokenisation # ────────────────────────────────────────────────────────────────── def _decode_bpe_token(self, tok_str: str) -> str: try: return bytes([self.UNICODE_TO_BYTE[c] for c in tok_str]).decode('utf-8', errors='replace') except (KeyError, UnicodeDecodeError): return tok_str def _bpe_encode_word(self, word: str) -> List[str]: try: bpe_chars = ''.join(self.BYTE_TO_UNICODE.get(b, '?') for b in word.encode('utf-8')) except UnicodeEncodeError: bpe_chars = ''.join( self.BYTE_TO_UNICODE.get(b, '?') for b in word.encode('utf-8', errors='replace') ) symbols = list(bpe_chars) while len(symbols) > 1: best_pair, best_rank = None, float('inf') for i in range(len(symbols) - 1): pair = (symbols[i], symbols[i + 1]) rank = self.MERGE_RANKS.get(pair, float('inf')) if rank < best_rank: best_rank = rank best_pair = pair if best_pair is None or best_rank == float('inf'): break new_symbols, i = [], 0 while i < len(symbols): if i < len(symbols) - 1 and (symbols[i], symbols[i + 1]) == best_pair: new_symbols.append(symbols[i] + symbols[i + 1]) i += 2 else: new_symbols.append(symbols[i]) i += 1 symbols = new_symbols return symbols def _tokenize(self, text: str) -> List[Tuple[int, bool, str]]: if not text or not text.strip(): return [] result = [] for word in text.strip().split(): symbols = self._bpe_encode_word(word) for idx, symbol in enumerate(symbols): tid = self.RAW_TO_ID.get(symbol, 1) is_final = (idx == len(symbols) - 1) decoded = self._decode_bpe_token(symbol) result.append((tid, is_final, decoded)) return result # ────────────────────────────────────────────────────────────────── # §4 Stage 3 — Feature Extraction # ────────────────────────────────────────────────────────────────── def _classify_token(self, tid: int) -> str: if tid in self.SPECIAL_IDS: return "special" elif tid in self.LATIN_IDS: return "latin" elif tid in self.PUNCT_IDS: return "punct" elif tid in self.BANGLA_IDS: return "bangla" else: return "other" # ────────────────────────────────────────────────────────────────── # §5 Stage 4 — Granular Scoring # ────────────────────────────────────────────────────────────────── def _score(self, tokens: List[Tuple[int, bool, str]]) -> dict: token_traces = [] weighted_sum = 0.0 signal_count = 0 suffix_hits = 0 effective_toks = 0 has_latin = False has_bangla = False strict_found = [] cumulative_raw = 0.0 for tid, is_word_final, decoded_str in tokens: tok_type = self._classify_token(tid) if tok_type in ("latin", "special", "punct"): if tok_type == "latin": has_latin = True token_traces.append({ "tid": tid, "decoded": decoded_str, "type": tok_type, "is_word_final": is_word_final, "action": "skipped", "base_weight": 0.0, "ig_value": 0.0, "blended_weight": 0.0, "position_boost": 1.0, "contribution": 0.0, "cumulative_raw": round(cumulative_raw, 6), }) continue effective_toks += 1 has_bangla = True w = self.WEIGHTS.get(tid, 0.0) if w < self.SIGNAL_THRESHOLD: token_traces.append({ "tid": tid, "decoded": decoded_str, "type": "neutral", "is_word_final": is_word_final, "action": "below_threshold", "base_weight": round(w, 6), "ig_value": 0.0, "blended_weight": 0.0, "position_boost": 1.0, "contribution": 0.0, "cumulative_raw": round(cumulative_raw, 6), }) continue signal_count += 1 ig = self.IG_WEIGHTS.get(tid, 0.0) w_blended = w + (self.IG_ALPHA * ig * w) boost = 1.0 boost_reason = "none" if is_word_final and tid in self.SUFFIX_IDS and suffix_hits < self.SUFFIX_STACK_CAP: boost = self.SUFFIX_AMP suffix_hits += 1 boost_reason = f"suffix_amp(hit#{suffix_hits})" elif is_word_final and tid in self.STRICT_IDS: boost = self.WORD_FINAL_MILD boost_reason = "word_final_mild" contribution = w_blended * boost weighted_sum += contribution cumulative_raw = weighted_sum if tid in self.STRICT_IDS: strict_found.append(tid) token_traces.append({ "tid": tid, "decoded": decoded_str, "type": "signal", "is_word_final": is_word_final, "action": "scored", "base_weight": round(w, 6), "ig_value": round(ig, 6), "blended_weight": round(w_blended, 6), "position_boost": boost, "boost_reason": boost_reason, "contribution": round(contribution, 6), "cumulative_raw": round(cumulative_raw, 6), }) raw_score = (weighted_sum / math.log(effective_toks + 1) if effective_toks > 0 else 0.0) z = max(-500.0, min(500.0, self.SIGMOID_K * (raw_score - self.SIGMOID_MID))) activated = 1.0 / (1.0 + math.exp(-z)) if signal_count < self.MIN_SIGNAL_TOKENS: final_score = 0.0 gate_reason = "min_signal_tokens" elif activated < self.DETECTION_THRESH: final_score = 0.0 gate_reason = "below_detection_threshold" else: final_score = activated gate_reason = "passed" signal_density = signal_count / max(effective_toks, 1) if signal_count >= self.CONF_MIN_SIGNALS and signal_density >= 0.15: confidence = "high" elif signal_count >= 1: confidence = "medium" else: confidence = "low" return { "weighted_sum": round(weighted_sum, 6), "effective_tokens": effective_toks, "log_denominator": round(math.log(effective_toks + 1), 6) if effective_toks > 0 else 0.0, "raw_score": round(raw_score, 6), "activated_score": round(activated, 6), "final_score": round(final_score, 6), "gate_reason": gate_reason, "signal_count": signal_count, "suffix_hits": suffix_hits, "signal_density": round(signal_density, 4), "confidence": confidence, "code_switch": has_latin and has_bangla, "strict_ids_found": strict_found[:10], "token_traces": token_traces, } # ────────────────────────────────────────────────────────────────── # §6 Stage 5 — Calibrated Output # ────────────────────────────────────────────────────────────────── def _calibrate(self, final_score: float, confidence: str, signal_count: int) -> dict: if confidence == "none" or signal_count == 0: return self._zero_display("no_signal") if final_score <= 0 or final_score < self.DISPLAY_FLOOR: return self._zero_display("below_floor") raw_percent = (100.0 if final_score >= self.SATURATION_POINT else ((final_score - self.DISPLAY_FLOOR) / (self.SATURATION_POINT - self.DISPLAY_FLOOR)) * 100.0) gamma = self.CONF_DAMPING.get(confidence, 1.0) adjusted_percent = max(0.0, min(100.0, raw_percent * gamma)) int_percent = max(0, min(100, int(round(adjusted_percent)))) byte_value = max(0, min(255, int(round(adjusted_percent * 2.55)))) category = "none" for cat, bounds in self.CAT_BOUNDARIES.items(): if bounds[0] <= int_percent <= bounds[1]: category = cat break return { "raw_percent": round(raw_percent, 2), "adjusted_percent": round(adjusted_percent, 2), "int_percent": int_percent, "byte_value": byte_value, "category": category, "confidence_damping": gamma, "suppression": None, } def _zero_display(self, reason: str) -> dict: return { "raw_percent": 0.0, "adjusted_percent": 0.0, "int_percent": 0, "byte_value": 0, "category": "none", "confidence_damping": 0.0, "suppression": reason, } # ────────────────────────────────────────────────────────────────── # §7 Public API # ────────────────────────────────────────────────────────────────── def infer(self, text: str, trace_mode: str = "none") -> dict: """ Run full inference pipeline. Args: text : Raw input string (any language, any length). trace_mode : "full" | "signals_only" | "none" (default, lowest latency) Returns: dict with int_percent, byte_value, category, confidence, code_switch, and optional trace. """ self._inference_count += 1 prep = self._preprocess(text) tokens = self._tokenize(prep["text"]) score_res = self._score(tokens) cal = self._calibrate(score_res["final_score"], score_res["confidence"], score_res["signal_count"]) output = { "input_text": text[:200], "int_percent": cal["int_percent"], "byte_value": cal["byte_value"], "category": cal["category"], "confidence": score_res["confidence"], "code_switch": score_res["code_switch"], } if trace_mode == "full": output["trace"] = { "preprocessing": prep["trace"], "token_count": len(tokens), "effective_tokens": score_res["effective_tokens"], "signal_count": score_res["signal_count"], "suffix_hits": score_res["suffix_hits"], "weighted_sum": score_res["weighted_sum"], "log_denominator": score_res["log_denominator"], "raw_score": score_res["raw_score"], "activated_score": score_res["activated_score"], "final_score": score_res["final_score"], "gate_reason": score_res["gate_reason"], "signal_density": score_res["signal_density"], "calibration": cal, "tokens": score_res["token_traces"], } elif trace_mode == "signals_only": output["trace"] = { "preprocessing": prep["trace"], "signal_count": score_res["signal_count"], "final_score": score_res["final_score"], "calibration": cal, "signal_tokens": [t for t in score_res["token_traces"] if t.get("action") == "scored"], } return output def infer_batch(self, texts: List[str], trace_mode: str = "none") -> List[dict]: """Batch inference.""" return [self.infer(t, trace_mode=trace_mode) for t in texts]