Spaces:
Runtime error
Runtime error
Computational_Consciousness_Engine / computational_consciousness_engine /input_mapping /text_mapper.py
| import string | |
| import numpy as np | |
| from typing import List, Dict, Set, Iterable | |
| import nltk | |
| from nltk.tokenize import word_tokenize | |
| from nltk.corpus import stopwords | |
| from nltk.stem import WordNetLemmatizer | |
| from sentence_transformers import SentenceTransformer | |
| class TextToChaosMapper: | |
| def __init__(self, num_shards: int = 1024): | |
| self.shard_to_word: Dict[int, str] = {} | |
| self.lemmatizer = WordNetLemmatizer() | |
| try: | |
| self.stop_words = set(stopwords.words("english")) | |
| except LookupError: | |
| self.stop_words = set() | |
| self.num_shards = max(1, int(num_shards)) | |
| # [NEW] Load the ML Embedding Model (Downloads ~80MB model automatically) | |
| print("Loading AI Semantic Embedding Model...") | |
| self.encoder = SentenceTransformer('all-MiniLM-L6-v2') | |
| # Fixed random projection matrix for stable LSH (Locality Sensitive Hashing) | |
| np.random.seed(42) | |
| self.projection_matrix = np.random.randn(384) | |
| def _clean_and_tokenize(self, text: str) -> List[str]: | |
| # [Keep this exactly as you have it right now] | |
| if not text: | |
| return [] | |
| text = text.lower() | |
| raw_tokens = word_tokenize(text) | |
| cleaned_tokens = [] | |
| for t in raw_tokens: | |
| t = t.strip(string.punctuation) | |
| t = t.translate(str.maketrans("", "", string.punctuation)) | |
| if not t or t in self.stop_words: | |
| continue | |
| base_word = self.lemmatizer.lemmatize(t) | |
| cleaned_tokens.append(base_word) | |
| return cleaned_tokens | |
| def _hash_token(self, token: str) -> int: | |
| """ | |
| [NEW] AI Semantic Locality-Sensitive Hashing. | |
| Instead of arbitrary SHA256, we calculate a 384-dimensional semantic | |
| meaning vector, and project it into an integer space. | |
| """ | |
| if not token: | |
| return 0 | |
| # 1. Calculate dense neural representation of the word | |
| embedding = self.encoder.encode(token) | |
| # 2. Project 384-dimensions down to a scalar using our fixed matrix | |
| semantic_scalar = np.dot(embedding, self.projection_matrix) | |
| # 3. Map into the shard integer space | |
| val = int(abs(semantic_scalar) * 1000000) | |
| return val % self.num_shards | |
| # [Keep map_text_to_shards, seed_chaos_pool, get_shard_word, text_to_shard_list exactly as they are] | |
| def map_text_to_shards(self, text: str) -> Dict[int, Set[str]]: | |
| """ | |
| Map cleaned tokens to shard indices. Returns a dict: shard -> set(tokens). | |
| Useful for building inverted indices or seeding chaos pools. | |
| """ | |
| tokens = self._clean_and_tokenize(text) | |
| shard_map: Dict[int, Set[str]] = {} | |
| for tok in tokens: | |
| shard = self._hash_token(tok) | |
| if shard not in shard_map: | |
| shard_map[shard] = set() | |
| shard_map[shard].add(tok) | |
| return shard_map | |
| def seed_chaos_pool(self, texts: Iterable[str]) -> None: | |
| """ | |
| Populate self.shard_to_word with a representative token for each shard. | |
| If multiple tokens map to the same shard, the first seen token wins. | |
| """ | |
| for text in texts: | |
| shard_map = self.map_text_to_shards(text) | |
| for shard, toks in shard_map.items(): | |
| if shard not in self.shard_to_word: | |
| # choose a deterministic representative (sorted) | |
| rep = sorted(toks)[0] | |
| self.shard_to_word[shard] = rep | |
| def get_shard_word(self, shard: int) -> str: | |
| """ | |
| Return the representative word for a shard, or empty string if none. | |
| """ | |
| return self.shard_to_word.get(shard, "") | |
| def text_to_shard_list(self, text: str) -> List[int]: | |
| """ | |
| Convenience: return sorted list of shard indices for a text. | |
| """ | |
| return sorted(self.map_text_to_shards(text).keys()) | |