Spaces:
Sleeping
Sleeping
File size: 5,467 Bytes
9194e16 | 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 | """PyTorch Dataset & DataLoader for ASVspoof 5 with on-the-fly telephony degradation."""
from typing import Dict, Optional, Tuple, Union
import os
import torch
from torch.utils.data import Dataset, DataLoader
import soundfile as sf
import pandas as pd
from src.transforms import TelephonyDegradation, TandemTelephonyDegradation
def pad_or_crop_waveform(
waveform: torch.Tensor,
target_len: int,
is_train: bool = True
) -> torch.Tensor:
"""Pad or crop waveform tensor (1, N) to target_len samples."""
num_samples = waveform.shape[-1]
if num_samples == target_len:
return waveform
elif num_samples > target_len:
if is_train:
max_start = num_samples - target_len
start = torch.randint(0, max_start + 1, (1,)).item()
else:
start = (num_samples - target_len) // 2
return waveform[..., start:start + target_len]
else:
# Repeat or zero pad
repeat_factor = (target_len // num_samples) + 1
repeated = waveform.repeat(1, repeat_factor)
return repeated[..., :target_len]
class TelephonyASVDataset(Dataset):
"""ASVspoof 5 Dataset with on-the-fly telephony degradation.
Reads manifest CSV, loads 16 kHz raw audio, pads/crops to fixed duration,
and applies on-the-fly 8 kHz telephony degradation (supports single or dual tandem compression).
"""
def __init__(
self,
manifest_path: str,
duration_sec: float = 4.0,
is_train: bool = True,
apply_telephony: bool = True,
use_tandem: bool = True,
voip_codec: str = "ULAW",
packet_loss_rate: float = 0.05,
):
super().__init__()
if not os.path.exists(manifest_path):
raise FileNotFoundError(f"Manifest file not found: {manifest_path}")
self.df = pd.read_csv(manifest_path)
self.duration_sec = duration_sec
self.is_train = is_train
self.apply_telephony = apply_telephony
self.use_tandem = use_tandem
# 16 kHz target length in raw audio
self.target_raw_samples = int(16000 * duration_sec)
# On-the-fly degradation module: dual tandem (Cellular GSM -> VoIP ULAW/Opus) or standard
if use_tandem:
self.degradation = TandemTelephonyDegradation(
orig_sample_rate=16000,
target_sample_rate=8000,
cellular_codec="GSM610",
voip_codec=voip_codec,
packet_loss_rate=packet_loss_rate if is_train else 0.0,
)
else:
self.degradation = TelephonyDegradation(
orig_sample_rate=16000,
target_sample_rate=8000,
apply_bandpass=True,
apply_mulaw=True,
packet_loss_rate=packet_loss_rate if is_train else 0.0,
)
if is_train:
self.degradation.train()
else:
self.degradation.eval()
def __len__(self) -> int:
return len(self.df)
def __getitem__(self, idx: int) -> Dict[str, Union[torch.Tensor, str, int]]:
row = self.df.iloc[idx]
file_path = row["file_path"]
label = int(row["label"]) # 0: bonafide, 1: spoof
speaker_id = str(row["speaker_id"])
file_id = str(row["file_id"])
# 1. Load audio with soundfile (fast & robust)
audio_data, sr = sf.read(file_path)
if sr != 16000:
raise ValueError(f"Expected 16000 Hz audio, found {sr} Hz in {file_path}")
# Ensure tensor shape (1, num_samples)
waveform = torch.from_numpy(audio_data).float()
if waveform.ndim == 1:
waveform = waveform.unsqueeze(0)
elif waveform.ndim == 2 and waveform.shape[0] > 1:
# Multi-channel -> mono
waveform = waveform.mean(dim=0, keepdim=True)
# 2. Pad or crop to target raw duration (e.g. 4.0s @ 16kHz = 64,000 samples)
waveform = pad_or_crop_waveform(waveform, self.target_raw_samples, is_train=self.is_train)
# 3. Apply on-the-fly telephony degradation (returns 8kHz = 32,000 samples)
if self.apply_telephony:
with torch.no_grad():
processed_audio = self.degradation(waveform)
else:
processed_audio = waveform
return {
"audio": processed_audio.squeeze(0), # (num_samples,) e.g. 32000 @ 8kHz
"label": torch.tensor(label, dtype=torch.float32),
"speaker_id": speaker_id,
"file_id": file_id,
}
def get_dataloader(
manifest_path: str,
batch_size: int = 16,
is_train: bool = True,
duration_sec: float = 4.0,
apply_telephony: bool = True,
use_tandem: bool = True,
voip_codec: str = "ULAW",
packet_loss_rate: float = 0.05,
num_workers: int = 0,
pin_memory: bool = False,
) -> DataLoader:
"""Convenience factory for TelephonyASVDataset DataLoader."""
dataset = TelephonyASVDataset(
manifest_path=manifest_path,
duration_sec=duration_sec,
is_train=is_train,
apply_telephony=apply_telephony,
use_tandem=use_tandem,
voip_codec=voip_codec,
packet_loss_rate=packet_loss_rate,
)
loader = DataLoader(
dataset,
batch_size=batch_size,
shuffle=is_train,
num_workers=num_workers,
pin_memory=pin_memory,
drop_last=is_train,
)
return loader
|