pipeline_modules/spatial_processing.py

"""Spatial audio processing and Direction-of-Arrival (DoA) estimation."""

import logging
import numpy as np

log = logging.getLogger(__name__)


def estimate_direction_from_mixing_matrix(
    mixing_matrix_column: np.ndarray,
) -> float:
    """
    Estimate Direction of Arrival (DoA) from ICA mixing matrix column.
    
    Assumes 4-channel hearing aid microphone array:
    - Channel 0: Left Front (LF)
    - Channel 1: Left Rear (LR)
    - Channel 2: Right Front (RF)
    - Channel 3: Right Rear (RR)
    
    Args:
        mixing_matrix_column: Single column from ICA mixing matrix (length 4)
        
    Returns:
        Azimuth angle in degrees (0°=front, 90°=right, 180°=rear, 270°=left)
    """
    if len(mixing_matrix_column) != 4:
        log.warning(f"Expected 4 channel weights, got {len(mixing_matrix_column)}")
        return 0.0
    
    w = mixing_matrix_column
    
    # Left (0,1) vs Right (2,3) balance
    left_energy = (np.abs(w[0]) + np.abs(w[1])) / 2
    right_energy = (np.abs(w[2]) + np.abs(w[3])) / 2
    lr_ratio = (right_energy - left_energy) / (right_energy + left_energy + 1e-10)
    
    # Front (0,2) vs Back (1,3) balance
    front_energy = (np.abs(w[0]) + np.abs(w[2])) / 2
    back_energy = (np.abs(w[1]) + np.abs(w[3])) / 2
    fb_ratio = (front_energy - back_energy) / (front_energy + back_energy + 1e-10)
    
    # Convert to angle using arctan2
    azimuth = np.degrees(np.arctan2(lr_ratio, fb_ratio))
    azimuth = (azimuth + 360) % 360  # Normalize to [0, 360)
    
    return float(azimuth)


def estimate_direction_from_tdoa(
    tdoa_estimates: dict,
) -> float:
    """
    Estimate direction from Time Difference of Arrival (TDOA).
    
    Args:
        tdoa_estimates: Dictionary with keys like 'lr_tdoa', 'fb_tdoa'
                       containing time differences in samples
    
    Returns:
        Azimuth angle in degrees
    """
    # This is a placeholder for TDOA-based DoA estimation
    # Actual implementation depends on microphone geometry and TDOA calculation
    log.warning("TDOA-based DoA estimation not yet implemented")
    return 0.0