| |
| """ |
| 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 |
|
|
| |
| |
| |
|
|
| 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) |
|
|
| |
| 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"] |
|
|
| |
| 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"] |
|
|
| |
| 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"]) |
|
|
| |
| _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)} |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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 |
|
|
| |
| |
| |
|
|
| @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 |
|
|
| |
| |
| |
|
|
| def _preprocess(self, text: str) -> dict: |
| trace = { |
| "original_length": len(text), |
| "original_words": len(text.split()) if text.strip() else 0, |
| "modifications": [], |
| } |
|
|
| |
| text_nfc = unicodedata.normalize('NFC', text) |
| if text_nfc != text: |
| trace["modifications"].append("nfc_normalized") |
| text = text_nfc |
|
|
| |
| cleaned = self.RE_ZERO_WIDTH.sub('', text) |
| if cleaned != text: |
| trace["modifications"].append(f"zwc_removed:{len(text)-len(cleaned)}_chars") |
| text = cleaned |
|
|
| |
| cleaned = self.RE_EMOJI.sub('', text) |
| if cleaned != text: |
| trace["modifications"].append("emoji_removed") |
| text = cleaned |
|
|
| |
| cleaned = self.RE_MULTI_SPACE.sub(' ', text).strip() |
| if cleaned != text.strip(): |
| trace["modifications"].append("whitespace_collapsed") |
| text = cleaned |
|
|
| |
| 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} |
|
|
| |
| |
| |
|
|
| 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 |
|
|
| |
| |
| |
|
|
| 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" |
|
|
| |
| |
| |
|
|
| 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, |
| } |
|
|
| |
| |
| |
|
|
| 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, |
| } |
|
|
| |
| |
| |
|
|
| 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] |
|
|
