mic-system/beamforming.py

from __future__ import annotations

import numpy as np

SPEED_OF_SOUND = 343.0


def _fractional_delay(signal: np.ndarray, delay_samples: float) -> np.ndarray:
    """Apply fractional sample delay using linear interpolation."""
    if signal.size == 0:
        return signal.astype(np.float32, copy=False)

    x = np.arange(signal.size, dtype=np.float32)
    shifted_x = x - np.float32(delay_samples)
    delayed = np.interp(shifted_x, x, signal, left=0.0, right=0.0)
    return delayed.astype(np.float32)


def beamform_delay_and_sum(
    mic1_data: np.ndarray,
    mic2_data: np.ndarray,
    angle_deg: float,
    sample_rate: int,
    mic_spacing: float,
    speed_of_sound: float = SPEED_OF_SOUND,
) -> np.ndarray:
    """Delay-and-sum beamforming for two microphones arranged in a line."""
    if mic1_data.shape != mic2_data.shape:
        raise ValueError("mic1_data and mic2_data must have the same shape")

    mic1 = mic1_data.astype(np.float32, copy=False)
    mic2 = mic2_data.astype(np.float32, copy=False)

    angle_rad = np.deg2rad(float(angle_deg))
    delay_seconds = float(mic_spacing) * np.sin(angle_rad) / float(speed_of_sound)
    delay_samples = delay_seconds * float(sample_rate)

    aligned_mic1 = _fractional_delay(mic1, delay_samples / 2.0)
    aligned_mic2 = _fractional_delay(mic2, -delay_samples / 2.0)

    output = (aligned_mic1 + aligned_mic2) * 0.5
    return np.clip(output, -1.0, 1.0).astype(np.float32)