Create core/tensor_bridge.py

core/tensor_bridge.py

@@ -0,0 +1,194 @@
+import numpy as np
+
+# ==========================================
+# 1. CONSTANTS (Must Match V7 Training Data)
+# ==========================================
+
+# The exact 64-phoneme interleaved list from AMLM V7
+PHONEME_LIST = [
+    # --- STOPS ---
+    'p', 'pʰ', 'b', 'bʰ',           
+    't', 'tʰ', 'd', 'dʰ',           
+    't̪', 't̪ʰ', 'd̪', 'd̪ʱ',           
+    'ʈ', 'ɖ',                       
+    'k', 'kʰ', 'g', 'ɡʰ',           
+    'q', 'ʔ',                       
+    # --- FRICATIVES ---
+    'f', 'v', 
+    's', 'z', 
+    'ʃ', 'ʒ', 'ʂ', 'ʐ',             
+    'x', 'ɣ', 'h', 
+    'θ', 'ð', 
+    # --- AFFRICATES ---
+    'tʃ', 'dʒ',
+    # --- NASALS ---
+    'm', 'mʰ', 'n', 'nʰ', 'ŋ', 'ɳ', 
+    # --- LIQUIDS ---
+    'l', 'lʰ', 'r', 'rʰ', 'ɽ', 'ɽʰ', 
+    # --- GLIDES ---
+    'w', 'j',
+    # --- VOWELS ---
+    'a', 'aː', 'i', 'iː', 'u', 'uː', 'e', 'eː', 'o', 'oː', 'ə', 'æ', 'ɔ', 'ɪ', 'ʊ'
+]
+
+# Map symbol to index (0-63)
+P2I = {p: i for i, p in enumerate(PHONEME_LIST)}
+
+# Error Column Indices (0-8)
+ERROR_MAP = {
+    "Sub": 0, "Substitution": 0,
+    "Front": 1, "Fronting": 1, "Velar Fronting": 1, "Palatal Fronting": 1,
+    "Stop": 2, "Stopping": 2,
+    "Glide": 3, "Gliding": 3,
+    "Clust": 4, "Cluster Reduction": 4,
+    "Epen": 5, "Epenthesis": 5,
+    "Block": 6, "Blocking": 6, "BlockBetweenWords": 6, "BlockWithinWord": 6,
+    "Prolong": 7, "Prolongation": 7, "ProlongationEvent": 7,
+    "Repeat": 8, "Repetition": 8, "RepetitionEvent": 8, "WordRepetition": 8, "SyllableRepetition": 8, "SoundRepetition": 8
+}
+
+POS_NAMES = ["Init", "Med", "Fin"]
+
+class TensorBridge:
+    def __init__(self):
+        self.num_channels = 3
+        self.num_phonemes = 64
+        self.num_errors = 9
+        # Cleaning map for backend IPA artifacts
+        self.ipa_cleaning_map = {
+            "g": "g", "ɡ": "g", # Unify various 'g' unicode chars
+            "r": "r", "ɾ": "r", # Tap vs Trill unification
+            "t_dental": "t̪", "d_dental": "d̪",
+            "t_retro": "ʈ", "d_retro": "ɖ"
+        }
+
+    def initialize_tensor(self):
+        """Returns a zeroed 3D tensor (List format for JSON compatibility)."""
+        # Shape: (3, 64, 9)
+        return np.zeros((3, 64, 9), dtype=np.float32).tolist()
+
+    def _unify_phoneme(self, symbol):
+        """Maps backend IPA (potentially messy) to V7 Tensor tokens."""
+        if not symbol: return None
+        
+        # 1. Direct Match
+        if symbol in P2I: return symbol
+        
+        # 2. Cleaning Map
+        if symbol in self.ipa_cleaning_map:
+            mapped = self.ipa_cleaning_map[symbol]
+            if mapped in P2I: return mapped
+
+        # 3. Fallback: Diacritic stripping (Try base character)
+        base = symbol[0]
+        if base in P2I: return base
+        
+        return None
+
+    def _determine_position(self, event_dict):
+        """
+        Determines position index (0=Init, 1=Med, 2=Fin).
+        Input: dict with keys 'position' (str) OR 'word' + 'index'.
+        """
+        # Explicit String from Backend
+        pos_str = event_dict.get('position', '').lower()
+        if 'init' in pos_str or 'start' in pos_str: return 0
+        if 'med' in pos_str or 'mid' in pos_str: return 1
+        if 'fin' in pos_str or 'end' in pos_str: return 2
+
+        # Implicit Calculation from Word Index
+        word = event_dict.get('word_ipa', '') or event_dict.get('word', '')
+        idx = event_dict.get('char_index', -1)
+        
+        if word and idx != -1:
+            if idx == 0: return 0
+            if idx == len(word) - 1: return 2
+            return 1 # Medial default
+            
+        return 0 # Default to Initial if unknown (safest assumption)
+
+    def parse_backend_output(self, analysis_data):
+        """
+        Converts Main Backend output (Test or Practice) into standardized Diagnostic Events.
+        
+        Output format: List of { 'phoneme_idx': int, 'error_idx': int, 'pos_idx': int }
+        """
+        events = []
+        
+        # Determine if this is a full AnalysisResponse or simple list
+        raw_list = analysis_data if isinstance(analysis_data, list) else analysis_data.get('errors', [])
+        
+        for item in raw_list:
+            # 1. Identify Error Type
+            proc_name = item.get('process') or item.get('type') or item.get('error_name')
+            if not proc_name: continue
+            
+            # Fuzzy match error name
+            err_idx = -1
+            for key, val in ERROR_MAP.items():
+                if key.lower() in proc_name.lower():
+                    err_idx = val
+                    break
+            if err_idx == -1: continue # Unknown error type
+
+            # 2. Identify Target Phoneme
+            # For SODA: 'expected' is the target. For Fluency: 'phone' or 'sound'.
+            target_char = item.get('expected') or item.get('phone') or item.get('target')
+            
+            # Clean and Map
+            clean_char = self._unify_phoneme(target_char)
+            if not clean_char: continue
+            
+            p_idx = P2I[clean_char]
+
+            # 3. Identify Position
+            pos_idx = self._determine_position(item)
+            
+            events.append({
+                'phoneme_idx': p_idx,
+                'error_idx': err_idx,
+                'pos_idx': pos_idx,
+                'raw': item
+            })
+            
+        return events
+
+    def update_tensor(self, current_tensor_blob, new_results, learning_rate=0.2):
+        """
+        Merges new errors into the patient tensor.
+        
+        Args:
+            current_tensor_blob: 3D List or None (from Firebase).
+            new_results: List of errors (from Test or Practice).
+            learning_rate: How much to increase severity per error (0.0 - 1.0).
+            
+        Returns:
+            Updated 3D List (JSON serializable).
+        """
+        # 1. Load or Initialize
+        if current_tensor_blob is None:
+            tensor = np.zeros((3, 64, 9), dtype=np.float32)
+        else:
+            tensor = np.array(current_tensor_blob, dtype=np.float32)
+            # Safety check shape
+            if tensor.shape != (3, 64, 9):
+                tensor = np.zeros((3, 64, 9), dtype=np.float32) # Reset if corrupt
+
+        # 2. Parse New Errors
+        events = self.parse_backend_output(new_results)
+        
+        # 3. Apply Updates
+        for ev in events:
+            c, h, w = ev['pos_idx'], ev['phoneme_idx'], ev['error_idx']
+            
+            # Logic: Increment severity.
+            # If it was 0, it becomes 0.2. If it was 0.8, it becomes 1.0.
+            current_val = tensor[c, h, w]
+            new_val = min(1.0, current_val + learning_rate)
+            tensor[c, h, w] = new_val
+            
+            # 4. Spillover Logic (Optional but smart)
+            # If Blocking happened on 'p' (Stop), slightly increase 'b' (Voiced Stop) too?
+            # For now, we rely on AMLM to fill these gaps, so we keep Bridge simple.
+
+        return tensor.tolist()