Spaces:
Sleeping
Sleeping
File size: 6,991 Bytes
2e78168 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 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() |