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()