src/character_aligner.py

def needleman_wunsch_align(expected_seq, predicted_seq, match_score=1, mismatch_penalty=-1, gap_penalty=-1):
    """
    Needleman-Wunsch global alignment for two phoneme sequences.
    Returns a list of (expected, predicted) pairs with None for gaps.
    """
    n = len(expected_seq)
    m = len(predicted_seq)
    # Initialize DP table
    score = np.zeros((n+1, m+1), dtype=int)
    pointer = np.zeros((n+1, m+1), dtype=int)  # 0:diag, 1:up, 2:left

    for i in range(1, n+1):
        score[i, 0] = gap_penalty * i
        pointer[i, 0] = 1
    for j in range(1, m+1):
        score[0, j] = gap_penalty * j
        pointer[0, j] = 2

    for i in range(1, n+1):
        for j in range(1, m+1):
            match = score[i-1, j-1] + (match_score if expected_seq[i-1] == predicted_seq[j-1] else mismatch_penalty)
            delete = score[i-1, j] + gap_penalty
            insert = score[i, j-1] + gap_penalty
            best = max(match, delete, insert)
            score[i, j] = best
            if best == match:
                pointer[i, j] = 0
            elif best == delete:
                pointer[i, j] = 1
            else:
                pointer[i, j] = 2

    # Traceback
    i, j = n, m
    alignment = []
    while i > 0 or j > 0:
        if i > 0 and j > 0 and pointer[i, j] == 0:
            alignment.append((expected_seq[i-1], predicted_seq[j-1]))
            i -= 1
            j -= 1
        elif i > 0 and pointer[i, j] == 1:
            alignment.append((expected_seq[i-1], None))
            i -= 1
        else:
            alignment.append((None, predicted_seq[j-1]))
            j -= 1
    alignment.reverse()
    return alignment
"""
Production-Grade Character-Level Phoneme Alignment for Arabic Speech Therapy
Uses ASR transcription and confidence scores to create accurate phoneme-level feedback
Optimized for children's speech therapy - faster and more reliable than MFA

v2.0: Now includes CTC-based segmentation for frame-accurate timing
"""

import numpy as np
from typing import List, Dict, Optional, Any
import re
import torch

try:
    from .pronunciation_variants import is_accepted_variant, variant_metadata
except ImportError:
    from pronunciation_variants import is_accepted_variant, variant_metadata


# Arabic diacritics that should be removed for processing
ARABIC_DIACRITICS = re.compile(r'[\u064B-\u065F\u0670]')

def clean_arabic_text(text: str) -> str:
    """Remove diacritics, punctuation, and normalize Arabic text."""
    # Remove diacritics
    text = ARABIC_DIACRITICS.sub('', text)
    # Remove punctuation and non-Arabic characters (retain Arabic letters + spaces)
    # Arabic letters range: U+0621-U+064A (ء-ي), plus tatweel U+0640 (ـ)
    text = re.sub(r'[^\u0621-\u064A\u0640\s]', '', text)
    # Remove extra whitespace
    text = ' '.join(text.split())
    return text


def _build_phoneme_entry(
    symbol: str,
    is_correct: bool,
    confidence: float,
    error_type: Optional[str],
    duration: float,
    timestamp: float,
    substituted_with: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """Create a phoneme response entry with optional substitution metadata."""
    phoneme_entry = {
        'symbol': symbol,
        'expected': is_correct,
        'confidence': confidence,
        'errorType': error_type,
        'duration': duration,
        'timestamp': timestamp
    }

    if error_type == 'substitution' and substituted_with is not None:
        phoneme_entry['substitutedWith'] = substituted_with

    if metadata:
        phoneme_entry.update(metadata)

    return phoneme_entry


def align_phonemes_character_level(
    audio_array: np.ndarray,
    expected_text: str,
    transcribed_text: str,
    confidence_scores: np.ndarray,
    sr: int = 16000
) -> List[Dict]:
    """
    LEGACY: Equal-time distribution character-level alignment.
    
    NOTE: This is kept for backward compatibility. For production use,
    prefer align_phonemes_ctc() which uses frame-accurate CTC segmentation.
    
    This is production-ready for speech therapy applications:
    - Works with all audio quality (as long as ASR succeeds)
    - Fast processing (~2 seconds vs 15+ for MFA)
    - Handles mispronunciations gracefully
    - Provides accurate character-by-character feedback
    - Uses ASR confidence for reliability scoring
    
    Args:
        audio_array: Audio samples (numpy array)
        expected_text: Ground truth Arabic text
        transcribed_text: ASR transcription output
        confidence_scores: Frame-level ASR confidence (from wav2vec2)
        sr: Sample rate (default 16000)
        
    Returns:
        List of phoneme dictionaries with:
        - symbol: Arabic character
        - expected: True if correct, False if error
        - confidence: 0.0-1.0 based on ASR confidence
        - errorType: None, 'substitution', 'insertion', or 'deletion'
        - duration: Estimated duration in seconds (equal distribution)
        - timestamp: Start time in seconds
    """
    duration = len(audio_array) / sr
    
    # Clean both texts
    expected_clean = clean_arabic_text(expected_text)
    transcribed_clean = clean_arabic_text(transcribed_text)
    
    # Convert to character lists
    expected_chars = list(expected_clean.replace(' ', ''))  # Remove spaces for char-level
    transcribed_chars = list(transcribed_clean.replace(' ', ''))
    
    if not expected_chars:
        return []
    
    # Calculate timing
    # Distribute time equally across expected characters
    char_duration = duration / len(expected_chars) if len(expected_chars) > 0 else 0.1
    
    # Map ASR confidence to time segments
    hop_length = 512  # wav2vec2 default hop length
    frame_duration = hop_length / sr
    
    phonemes = []
    
    # Align expected vs transcribed using simple character matching
    # This handles substitutions, insertions, and deletions
    for i, expected_char in enumerate(expected_chars):
        start_time = i * char_duration
        end_time = (i + 1) * char_duration
        
        # Get ASR confidence for this time segment
        start_frame = int(start_time / frame_duration)
        end_frame = int(end_time / frame_duration)
        
        if start_frame < len(confidence_scores) and end_frame <= len(confidence_scores):
            segment_conf = confidence_scores[start_frame:end_frame]
            confidence = float(np.mean(segment_conf)) if len(segment_conf) > 0 else 0.85
        else:
            confidence = 0.85  # Default confidence
        
        # Determine if character matches transcription
        transcribed_char = transcribed_chars[i] if i < len(transcribed_chars) else None
        is_variant = is_accepted_variant(expected_char, transcribed_char)
        is_correct = (transcribed_char == expected_char) or is_variant
        metadata = (
            variant_metadata(expected_char, transcribed_char)
            if is_variant
            else None
        )
        
        # Determine error type
        error_type = None
        if not is_correct:
            if transcribed_char is None:
                error_type = 'deletion'  # Expected char not produced
                confidence = 0.0
            elif i < len(transcribed_chars):
                error_type = 'substitution'  # Wrong character produced
                confidence *= 0.5  # Reduce confidence for errors
            else:
                error_type = 'deletion'
                confidence = 0.0
        
        phonemes.append(_build_phoneme_entry(
            symbol=expected_char,
            is_correct=is_correct,
            confidence=max(0.0, min(1.0, confidence)),
            error_type=error_type,
            duration=char_duration,
            timestamp=start_time,
            substituted_with=transcribed_char if error_type == 'substitution' else None,
            metadata=metadata
        ))
    
    # Handle insertions (extra characters in transcription)
    if len(transcribed_chars) > len(expected_chars):
        for i in range(len(expected_chars), len(transcribed_chars)):
            extra_char = transcribed_chars[i]
            phonemes.append(_build_phoneme_entry(
                symbol=extra_char,
                is_correct=False,
                confidence=0.3,
                error_type='insertion',
                duration=0.05,
                timestamp=duration - 0.05
            ))
    
    return phonemes


def align_phonemes_ctc(
    audio_array: np.ndarray,
    expected_text: str,
    transcribed_text: str,
    logits: torch.Tensor,
    predicted_ids: torch.Tensor,
    vocab: Dict[str, int],
    sr: int = 16000
) -> List[Dict]:
    """
    CTC-based phoneme alignment using actual frame predictions from wav2vec2.
    More accurate than equal-time distribution - uses model's learned phoneme boundaries.
    
    Args:
        audio_array: Audio samples (numpy array)
        expected_text: Ground truth Arabic text
        transcribed_text: ASR transcription output
        logits: Raw CTC logits from wav2vec2 [1, time_steps, vocab_size]
        predicted_ids: Argmax of logits [1, time_steps]
        vocab: Tokenizer vocabulary {token: id}
        sr: Sample rate (default 16000)
        
    Returns:
        List of phoneme dictionaries with frame-accurate timing
    """
    duration = len(audio_array) / sr
    
    # Clean texts
    expected_clean = clean_arabic_text(expected_text)
    transcribed_clean = clean_arabic_text(transcribed_text)
    
    # Remove spaces for character-level analysis
    expected_chars = list(expected_clean.replace(' ', ''))
    transcribed_chars = list(transcribed_clean.replace(' ', ''))
    
    if not expected_chars:
        return []
    
    # Get CTC blank token (usually pad_token_id or 0)
    blank_id = vocab.get('[PAD]', vocab.get('<pad>', 0))
    
    # Extract frame-level predictions and confidence
    pred_ids = predicted_ids[0].cpu().numpy()  # Shape: [time_steps]
    probs = torch.nn.functional.softmax(logits[0], dim=-1).cpu().numpy()  # [time_steps, vocab]
    
    # Calculate frame duration
    num_frames = len(pred_ids)
    frame_duration = duration / num_frames
    
    # Step 1: Extract CTC segments (non-blank, non-repeated tokens)
    ctc_segments = []
    prev_token = None
    segment_start = 0
    
    for frame_idx, token_id in enumerate(pred_ids):
        # Skip blank tokens and repeated tokens (CTC collapse)
        if token_id == blank_id:
            if prev_token is not None:
                # End current segment
                ctc_segments.append({
                    'token_id': prev_token,
                    'start_frame': segment_start,
                    'end_frame': frame_idx,
                    'confidence': float(np.mean([probs[i, prev_token] for i in range(segment_start, frame_idx)]))
                })
                prev_token = None
        elif token_id != prev_token:
            if prev_token is not None:
                # End previous segment
                ctc_segments.append({
                    'token_id': prev_token,
                    'start_frame': segment_start,
                    'end_frame': frame_idx,
                    'confidence': float(np.mean([probs[i, prev_token] for i in range(segment_start, frame_idx)]))
                })
            # Start new segment
            prev_token = token_id
            segment_start = frame_idx
    
    # Handle last segment
    if prev_token is not None and prev_token != blank_id:
        ctc_segments.append({
            'token_id': prev_token,
            'start_frame': segment_start,
            'end_frame': num_frames,
            'confidence': float(np.mean([probs[i, prev_token] for i in range(segment_start, num_frames)]))
        })
    
    # Step 2: Map CTC segments to transcribed characters
    # CTC segments align with transcribed_chars
    segment_to_char = {}
    char_idx = 0
    for seg_idx, segment in enumerate(ctc_segments):
        if char_idx < len(transcribed_chars):
            segment_to_char[seg_idx] = char_idx
            char_idx += 1
    
    # Step 3: Align transcribed to expected characters
    phonemes = []
    
    # Use simple alignment: match position-wise with error detection
    for i, expected_char in enumerate(expected_chars):
        # Find corresponding CTC segment if available
        if i < len(ctc_segments):
            segment = ctc_segments[i]
            start_time = segment['start_frame'] * frame_duration
            end_time = segment['end_frame'] * frame_duration
            confidence = segment['confidence']
        else:
            # Fallback to equal distribution for missing segments
            char_duration = duration / len(expected_chars)
            start_time = i * char_duration
            end_time = (i + 1) * char_duration
            confidence = 0.5
        
        # Check if character matches
        transcribed_char = transcribed_chars[i] if i < len(transcribed_chars) else None
        is_variant = is_accepted_variant(expected_char, transcribed_char)
        is_correct = (transcribed_char == expected_char) or is_variant
        metadata = (
            variant_metadata(expected_char, transcribed_char)
            if is_variant
            else None
        )
        
        # Determine error type
        error_type = None
        if not is_correct:
            if transcribed_char is None:
                error_type = 'deletion'
                confidence = 0.0
            else:
                error_type = 'substitution'
                confidence *= 0.5
        
        phonemes.append(_build_phoneme_entry(
            symbol=expected_char,
            is_correct=is_correct,
            confidence=max(0.0, min(1.0, confidence)),
            error_type=error_type,
            duration=round(end_time - start_time, 3),
            timestamp=round(start_time, 3),
            substituted_with=transcribed_char if error_type == 'substitution' else None,
            metadata=metadata
        ))
    
    # Handle insertions (extra characters in transcription)
    if len(transcribed_chars) > len(expected_chars):
        for i in range(len(expected_chars), len(transcribed_chars)):
            if i < len(ctc_segments):
                segment = ctc_segments[i]
                timestamp = segment['start_frame'] * frame_duration
                duration_val = (segment['end_frame'] - segment['start_frame']) * frame_duration
            else:
                timestamp = duration - 0.05
                duration_val = 0.05
            
            phonemes.append(_build_phoneme_entry(
                symbol=transcribed_chars[i],
                is_correct=False,
                confidence=0.3,
                error_type='insertion',
                duration=round(duration_val, 3),
                timestamp=round(timestamp, 3)
            ))
    
    return phonemes


def calculate_pronunciation_accuracy(phonemes: List[Dict]) -> Dict[str, float]:
    """
    Calculate detailed pronunciation accuracy metrics.
    
    Returns:
        Dictionary with accuracy, error_rate, and confidence metrics
    """
    if not phonemes:
        return {
            'accuracy': 0.0,
            'error_rate': 1.0,
            'avg_confidence': 0.0,
            'correct_count': 0,
            'total_count': 0
        }
    
    total = len(phonemes)
    correct = sum(1 for p in phonemes if p['expected'])
    total_confidence = sum(p['confidence'] for p in phonemes)
    
    return {
        'accuracy': correct / total if total > 0 else 0.0,
        'error_rate': (total - correct) / total if total > 0 else 0.0,
        'avg_confidence': total_confidence / total if total > 0 else 0.0,
        'correct_count': correct,
        'total_count': total
    }