src/transforms.py

"""Telephony Audio Degradation Pipeline for ASVspoof 5.

Simulates realistic voice call channel artifacts:
1. PSTN Narrowband Bandpass Filtering (300 Hz - 3400 Hz).
2. 8 kHz Downsampling (Nyquist = 4000 Hz).
3. G.711 Mu-law 8-bit Companding and Quantization.
4. VoIP / RTP Packet Loss Simulation (frame erasure / burst drops).
"""

from typing import Optional, Tuple
import torch
import torch.nn as nn
import torchaudio.transforms as T
import torchaudio.functional as F


class TelephonyBandpassFilter(nn.Module):
    """Simulates the 300 Hz - 3400 Hz acoustic passband of standard PSTN telephone networks."""

    def __init__(self, sample_rate: int = 16000, low_freq: float = 300.0, high_freq: float = 3400.0):
        super().__init__()
        self.sample_rate = sample_rate
        self.low_freq = low_freq
        self.high_freq = high_freq

    def forward(self, waveform: torch.Tensor) -> torch.Tensor:
        """Apply highpass at 300 Hz and lowpass at 3400 Hz.
        
        waveform: Tensor of shape (..., num_samples)
        """
        # Highpass to remove low-frequency rumble / DC offset (< 300 Hz)
        filtered = F.highpass_biquad(waveform, sample_rate=self.sample_rate, cutoff_freq=self.low_freq)
        # Lowpass to limit to narrowband telephone bandwidth (< 3400 Hz)
        filtered = F.lowpass_biquad(filtered, sample_rate=self.sample_rate, cutoff_freq=self.high_freq)
        return filtered


class PacketLossSimulator(nn.Module):
    """Simulates VoIP / RTP burst packet loss with frame erasure and packet loss concealment."""

    def __init__(
        self,
        sample_rate: int = 8000,
        packet_duration_ms: float = 20.0,
        loss_rate: float = 0.05,
        max_burst_packets: int = 3,
        concealment: str = "zero"  # 'zero', 'noise', or 'attenuate'
    ):
        super().__init__()
        self.sample_rate = sample_rate
        self.packet_size = int(sample_rate * (packet_duration_ms / 1000.0))
        self.loss_rate = loss_rate
        self.max_burst_packets = max_burst_packets
        self.concealment = concealment

    def forward(self, waveform: torch.Tensor) -> torch.Tensor:
        """Apply packet drops on 20ms frames.
        
        waveform: Tensor of shape (..., num_samples)
        """
        if self.loss_rate <= 0.0 or not self.training:
            return waveform

        out = waveform.clone()
        num_samples = out.shape[-1]
        num_packets = num_samples // self.packet_size

        if num_packets == 0:
            return out

        i = 0
        while i < num_packets:
            if torch.rand(1).item() < self.loss_rate:
                burst_len = torch.randint(1, self.max_burst_packets + 1, (1,)).item()
                burst_len = min(burst_len, num_packets - i)
                start_sample = i * self.packet_size
                end_sample = min((i + burst_len) * self.packet_size, num_samples)

                if self.concealment == "zero":
                    out[..., start_sample:end_sample] = 0.0
                elif self.concealment == "noise":
                    comfort_noise = torch.randn_like(out[..., start_sample:end_sample]) * 0.001
                    out[..., start_sample:end_sample] = comfort_noise
                elif self.concealment == "attenuate":
                    out[..., start_sample:end_sample] *= 0.1

                i += burst_len
            else:
                i += 1

        return out


class TelephonyDegradation(nn.Module):
    """Complete telephony degradation pipeline.
    
    Transforms 16 kHz input speech into realistic 8 kHz narrowband telephony speech:
    1. Bandpass filter: 300 - 3400 Hz.
    2. Resampling: 16 kHz -> 8 kHz.
    3. G.711 Mu-law companding & 8-bit quantization.
    4. VoIP packet loss simulation (during training).
    5. Amplitude normalization.
    """

    def __init__(
        self,
        orig_sample_rate: int = 16000,
        target_sample_rate: int = 8000,
        apply_bandpass: bool = True,
        apply_mulaw: bool = True,
        packet_loss_rate: float = 0.05,
        packet_duration_ms: float = 20.0
    ):
        super().__init__()
        self.orig_sample_rate = orig_sample_rate
        self.target_sample_rate = target_sample_rate
        self.apply_bandpass = apply_bandpass
        self.apply_mulaw = apply_mulaw

        # 1. Bandpass (300 - 3400 Hz at 16 kHz)
        self.bandpass = TelephonyBandpassFilter(orig_sample_rate, 300.0, 3400.0)

        # 2. Resampler (16 kHz -> 8 kHz)
        self.resampler = T.Resample(orig_freq=orig_sample_rate, new_freq=target_sample_rate)

        # 3. G.711 Mu-law (8-bit, 256 quantization channels)
        self.mulaw_encoder = T.MuLawEncoding(quantization_channels=256)
        self.mulaw_decoder = T.MuLawDecoding(quantization_channels=256)

        # 4. Packet Loss Simulator (at 8 kHz)
        self.packet_loss = PacketLossSimulator(
            sample_rate=target_sample_rate,
            packet_duration_ms=packet_duration_ms,
            loss_rate=packet_loss_rate
        )

    def forward(self, waveform: torch.Tensor) -> torch.Tensor:
        """Apply telephony degradation pipeline.
        
        Args:
            waveform: Tensor of shape (channels, num_samples) or (batch, channels, num_samples)
                      Sampled at orig_sample_rate (16 kHz).
        Returns:
            degraded: Tensor at target_sample_rate (8 kHz) in [-1.0, 1.0].
        """
        # Ensure float32
        x = waveform.float()

        # Step 1: Bandpass filter (PSTN standard passband 300-3400 Hz)
        if self.apply_bandpass:
            x = self.bandpass(x)

        # Step 2: Resample to 8 kHz
        x = self.resampler(x)

        # Step 3: G.711 Mu-law companding & 8-bit non-linear quantization
        if self.apply_mulaw:
            # Clamp to [-1.0, 1.0] before mu-law companding
            x = torch.clamp(x, -1.0, 1.0)
            encoded = self.mulaw_encoder(x)
            x = self.mulaw_decoder(encoded)

        # Step 4: VoIP packet loss simulation
        x = self.packet_loss(x)

        # Step 5: Peak normalize to avoid clipping while maintaining dynamics
        max_val = torch.max(torch.abs(x), dim=-1, keepdim=True)[0]
        max_val = torch.clamp(max_val, min=1e-6)
        # Avoid amplifying pure silence
        scale = torch.where(max_val > 0.01, 0.95 / max_val, torch.ones_like(max_val))
        x = x * scale

        return x


class TandemTelephonyDegradation(nn.Module):
    """Dual-Codec (Tandem) Telephony Degradation Pipeline.
    
    Simulates real-world phone calls traversing multiple heterogeneous networks:
    1. Bandpass filter: 300 - 3400 Hz.
    2. Resampling: 16 kHz -> 8 kHz.
    3. Step 1 (Cellular Carrier): GSM-FR 6.10 (13 kbps, 8 kHz RPE-LTP).
    4. Step 2 (VoIP / Web Gateway): G.711 µ-law or Opus compression.
    5. VoIP packet loss simulation (burst drops on 20ms frames).
    6. Peak normalization.
    """

    def __init__(
        self,
        orig_sample_rate: int = 16000,
        target_sample_rate: int = 8000,
        cellular_codec: str = "GSM610",  # GSM-FR 6.10
        voip_codec: str = "ULAW",        # G.711 µ-law or 'OPUS'
        packet_loss_rate: float = 0.05,
        packet_duration_ms: float = 20.0
    ):
        super().__init__()
        self.orig_sample_rate = orig_sample_rate
        self.target_sample_rate = target_sample_rate
        self.cellular_codec = cellular_codec
        self.voip_codec = voip_codec.upper()

        self.bandpass = TelephonyBandpassFilter(orig_sample_rate, 300.0, 3400.0)
        self.resampler = T.Resample(orig_freq=orig_sample_rate, new_freq=target_sample_rate)
        self.packet_loss = PacketLossSimulator(
            sample_rate=target_sample_rate,
            packet_duration_ms=packet_duration_ms,
            loss_rate=packet_loss_rate
        )

    def _compress_gsm_single(self, audio_np: "np.ndarray") -> "np.ndarray":
        """In-memory GSM 6.10 Full Rate compression/decompression via libsndfile."""
        import io
        import soundfile as sf
        bio = io.BytesIO()
        sf.write(bio, audio_np, self.target_sample_rate, format="WAV", subtype="GSM610")
        bio.seek(0)
        decoded, _ = sf.read(bio, dtype="float32")
        return decoded

    def _compress_voip_single(self, audio_np: "np.ndarray") -> "np.ndarray":
        """In-memory VoIP compression (G.711 ULAW or OPUS) via libsndfile."""
        import io
        import soundfile as sf
        bio = io.BytesIO()
        if self.voip_codec == "OPUS":
            sf.write(bio, audio_np, self.target_sample_rate, format="OGG", subtype="OPUS")
        else:
            sf.write(bio, audio_np, self.target_sample_rate, format="WAV", subtype="ULAW")
        bio.seek(0)
        decoded, _ = sf.read(bio, dtype="float32")
        return decoded

    def forward(self, waveform: torch.Tensor) -> torch.Tensor:
        """Apply sequential tandem compression (Cellular GSM-FR -> VoIP ULAW/Opus).
        
        Args:
            waveform: Tensor of shape (..., num_samples) at 16 kHz.
        Returns:
            degraded: Tensor of shape (..., target_samples) at 8 kHz.
        """
        import numpy as np

        x = waveform.float()

        # Step 1: Bandpass filter (300 - 3400 Hz)
        x = self.bandpass(x)

        # Step 2: Downsample to 8 kHz
        x = self.resampler(x)

        # Flatten batch dimensions for in-memory codec passes
        orig_shape = x.shape
        x_flat = x.reshape(-1, orig_shape[-1])  # (B, N)
        target_len = orig_shape[-1]

        processed_batch = []
        for i in range(x_flat.shape[0]):
            sig = x_flat[i].detach().cpu().numpy()
            # Normalize to [-0.95, 0.95] to prevent codec clipping
            max_abs = np.max(np.abs(sig))
            if max_abs > 1e-4:
                sig = sig * (0.95 / max_abs)

            # Tandem Stage 1: Cellular GSM-FR 6.10
            gsm_out = self._compress_gsm_single(sig)
            # Ensure exact length alignment
            if len(gsm_out) > target_len:
                gsm_out = gsm_out[:target_len]
            elif len(gsm_out) < target_len:
                gsm_out = np.pad(gsm_out, (0, target_len - len(gsm_out)))

            # Tandem Stage 2: VoIP G.711 µ-law or Opus
            voip_out = self._compress_voip_single(gsm_out)
            if len(voip_out) > target_len:
                voip_out = voip_out[:target_len]
            elif len(voip_out) < target_len:
                voip_out = np.pad(voip_out, (0, target_len - len(voip_out)))

            processed_batch.append(voip_out)

        out_tensor = torch.from_numpy(np.stack(processed_batch)).to(device=x.device, dtype=x.dtype)
        out_tensor = out_tensor.reshape(*orig_shape[:-1], target_len)

        # Step 5: VoIP Packet loss simulation
        out_tensor = self.packet_loss(out_tensor)

        # Step 6: Peak normalization
        max_val = torch.max(torch.abs(out_tensor), dim=-1, keepdim=True)[0]
        max_val = torch.clamp(max_val, min=1e-6)
        scale = torch.where(max_val > 0.01, 0.95 / max_val, torch.ones_like(max_val))
        out_tensor = out_tensor * scale

        return out_tensor