Spaces:
Sleeping
Sleeping
| 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() |