Upload 28 files

src/__init__.py

@@ -0,0 +1,3 @@
+import logging
+
+logging.getLogger(__name__).addHandler(logging.NullHandler())

src/commons.py

@@ -0,0 +1,22 @@
+"""Utility file for src toolkit."""
+import os
+import random
+
+import numpy as np
+import torch
+
+WHISPER_MODEL_WEIGHTS_PATH = "src/models/assets/tiny_enc.en.pt"
+
+
+def set_seed(seed: int):
+    """Fix PRNG seed for reproducable experiments.
+    """
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed(seed)
+        torch.cuda.manual_seed_all(seed)
+        torch.backends.cudnn.deterministic = True
+        torch.backends.cudnn.benchmark = False
+    os.environ["PYTHONHASHSEED"] = str(seed)

src/datasets/KoAAD_dataset.py

@@ -0,0 +1,45 @@
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+import pandas as pd
+from pathlib import Path
+import os
+
+class KoAAD(SimpleAudioFakeDataset):                                            
+    def __init__(self, root_path, subset=None, **kwargs):
+        super().__init__(root_path, subset, **kwargs)
+        self.root_path = Path(f'{root_path}')
+        self.subset = subset
+        self.samples = self.load_samples()
+
+    def load_samples(self):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        folders_1 = list(self.root_path.glob("*"))
+        for f1 in folders_1:
+            if not os.path.isdir(f1):
+                continue
+        
+            if not f1.exists():
+                print(f"{path} 경로를 찾을 수 없습니다.")
+            
+            samples_list = list(f1.rglob("*.[wm][ap][v3]"))
+            if self.subset == 'train':
+                samples_list = samples_list[:int(len(samples_list)*0.7)]
+            else:
+                samples_list = samples_list[int(len(samples_list)*0.7):]
+            for sample in samples_list:
+                if os.path.exists(sample):
+                    samples["user_id"].append(None)
+                    samples["path"].append(sample)
+                    samples["sample_name"].append(sample.stem)
+                    samples["attack_type"].append("-")
+                    samples["label"].append("spoof")
+
+        print(f"KoAAD_{self.subset}:{len(samples['label'])}")
+        return pd.DataFrame(samples)
+

src/datasets/MAILABS_dataset.py

@@ -0,0 +1,50 @@
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+import pandas as pd
+from pathlib import Path
+import os
+
+class MAILABS(SimpleAudioFakeDataset):                                            
+    def __init__(self, root_path, subset=None, **kwargs):
+        super().__init__(root_path, subset, **kwargs)
+        self.root_path = Path(f'{root_path}')
+        self.subset = subset
+        self.samples = self.load_samples()
+
+    def load_samples(self):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+        split = [0.7, 0.3]
+
+        folders_1 = list(self.root_path.glob("en_US/by_book/*"))
+        for f1 in folders_1:
+            if not os.path.isdir(f1):
+                continue
+            folders_2 = list(f1.glob("*"))
+            for f2 in folders_2:
+                path = f1 / f2.name
+
+                if not path.exists():
+                    print(f"{path} 경로를 찾을 수 없습니다.")
+
+                samples_list = list(path.rglob("*.wav"))
+                if self.subset == 'train':
+                    samples_list = samples_list[:int(len(samples_list)*split[0])]
+                elif self.subset == 'test':
+                    samples_list = samples_list[int(len(samples_list)*(split[0])):]
+                for sample in samples_list:
+                    if sample.stem[0]==".":
+                        continue
+                    if os.path.exists(sample):
+                        samples["user_id"].append(None)
+                        samples["path"].append(sample)
+                        samples["sample_name"].append(sample.stem)
+                        samples["attack_type"].append("-")
+                        samples["label"].append("bonafide")
+        print(f"MAILABS_{self.subset}:{len(samples['label'])}")
+        return pd.DataFrame(samples)
+

src/datasets/MLAADv3_dataset.py

@@ -0,0 +1,47 @@
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+import pandas as pd
+from pathlib import Path
+
+class MLAADv3(SimpleAudioFakeDataset):
+    languages=['fr', 'et', 'ar', 'hu', 'bg', 'es', 'el', 'da', 'ga', 'ru', 'fi', 
+               'uk', 'pl', 'en', 'sw', 'mt', 'sk', 'ro', 'hi', 'cs', 'nl', 'it', 'de']
+                                            
+    def __init__(self, root_path, subset=None, **kwargs):
+        super().__init__(root_path, subset, **kwargs)
+        self.root_path = Path(f'{root_path}')
+        self.subset = subset
+        self.samples = self.load_samples()
+
+    def load_samples(self):
+        samples = {
+            "user_id": [],
+            "language" : [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        for lang in self.languages:
+            r_path = self.root_path / f"fake/{lang}"
+            folders = list(r_path.glob("*"))
+            for folder in folders:
+                path = r_path / folder.name
+
+                if not path.exists():
+                    print(f"{path} 경로를 찾을 수 없습니다.")
+                    continue
+                samples_list = list(path.rglob("*.wav"))
+                if self.subset == 'train':
+                    samples_list = samples_list[:int(len(samples_list)*0.7)]
+                else:
+                    samples_list = samples_list[int(len(samples_list)*0.7):]
+                for sample in samples_list:
+                    samples["user_id"].append(None)
+                    samples["language"].append(lang)
+                    samples["path"].append(sample)
+                    samples["sample_name"].append(sample.stem)
+                    samples["attack_type"].append("-")
+                    samples["label"].append("spoof")
+        print(f"__MLAADv3_{self.subset}:{len(samples['label'])}") 
+        return pd.DataFrame(samples)

src/datasets/aihub_dataset.py

@@ -0,0 +1,40 @@
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+import pandas as pd
+from pathlib import Path
+
+class AIHUB(SimpleAudioFakeDataset):                                            
+    def __init__(self, root_path, subset=None, **kwargs):
+        super().__init__(root_path, subset, **kwargs)
+        self.root_path = Path(f'{root_path}')
+        self.subset = subset
+        self.samples = self.load_samples()
+
+    def load_samples(self):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        path = self.root_path / ""
+        
+        # 해당 언어의 디렉토리가 존재하는지 확인
+        if not path.exists():
+            print(f"{path} 경로를 찾을 수 없습니다.")
+        
+        samples_list = list(path.rglob("*.wav"))
+        if self.subset == 'train':
+            samples_list = samples_list[:int(len(samples_list)*0.7)]
+        else:
+            samples_list = samples_list[int(len(samples_list)*0.7):]
+        for sample in samples_list:
+            samples["user_id"].append(None)
+            samples["path"].append(sample)
+            samples["sample_name"].append(sample.stem)
+            samples["attack_type"].append("-")
+            samples["label"].append("bonafide")
+        print(f"__AIHUB_{self.subset}:{len(samples['label'])}") 
+        return pd.DataFrame(samples)
+

src/datasets/asvspoof_dataset.py

@@ -0,0 +1,155 @@
+from pathlib import Path
+
+import pandas as pd
+if __name__ == "__main__":
+    import sys
+    sys.path.append(str(Path(__file__).parent.parent.parent.absolute()))
+
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+
+ASVSPOOF_SPLIT = {
+    "train": ['A01', 'A07', 'A08', 'A02', 'A09', 'A10', 'A03', 'A04', 'A05', 'A06', 'A11', 'A12', 'A13', 'A14', 'A15', 'A16', 'A17', 'A18', 'A19'],
+    "test":  ['A01', 'A07', 'A08', 'A02', 'A09', 'A10', 'A03', 'A04', 'A05', 'A06', 'A11', 'A12', 'A13', 'A14', 'A15', 'A16', 'A17', 'A18', 'A19'],
+    "val":   ['A01', 'A07', 'A08', 'A02', 'A09', 'A10', 'A03', 'A04', 'A05', 'A06', 'A11', 'A12', 'A13', 'A14', 'A15', 'A16', 'A17', 'A18', 'A19'],
+    "partition_ratio": [0.7, 0.15],
+    "seed": 45,
+}
+
+
+class ASVSpoofDataset(SimpleAudioFakeDataset):
+
+    protocol_folder_name = "ASVspoof2019_LA_cm_protocols"
+    subset_dir_prefix = "ASVspoof2019_LA_"
+    subsets = ("train", "dev", "eval")
+
+    def __init__(self, path, subset="train", transform=None):
+        super().__init__(subset, transform)
+        self.path = path
+
+        self.allowed_attacks = ASVSPOOF_SPLIT[subset]
+        self.partition_ratio = ASVSPOOF_SPLIT["partition_ratio"]
+        self.seed = ASVSPOOF_SPLIT["seed"]
+
+        self.samples = pd.DataFrame()
+
+        for subset in self.subsets:
+            subset_dir = Path(self.path) / f"{self.subset_dir_prefix}{subset}"
+            subset_protocol_path = self.get_protocol_path(subset)
+            subset_samples = self.read_protocol(subset_dir, subset_protocol_path)
+
+            self.samples = pd.concat([self.samples, subset_samples])
+
+        self.transform = transform
+
+    def get_protocol_path(self, subset):
+        paths = list((Path(self.path) / self.protocol_folder_name).glob("*.txt"))
+        for path in paths:
+            if subset in Path(path).stem:
+                return path
+
+    def read_protocol(self, subset_dir, protocol_path):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        real_samples = []
+        fake_samples = []
+        with open(protocol_path, "r") as file:
+            for line in file:
+                attack_type = line.strip().split(" ")[3]
+
+                if attack_type == "-":
+                    real_samples.append(line)
+                elif attack_type in self.allowed_attacks:
+                    fake_samples.append(line)
+
+                if attack_type not in self.allowed_attacks:
+                    continue
+
+        fake_samples = self.split_samples(fake_samples)
+        for line in fake_samples:
+            samples = self.add_line_to_samples(samples, line, subset_dir)
+
+        real_samples = self.split_samples(real_samples)
+        for line in real_samples:
+            samples = self.add_line_to_samples(samples, line, subset_dir)
+
+        return pd.DataFrame(samples)
+
+    @staticmethod
+    def add_line_to_samples(samples, line, subset_dir):
+        user_id, sample_name, _, attack_type, label = line.strip().split(" ")
+        samples["user_id"].append(user_id)
+        samples["sample_name"].append(sample_name)
+        samples["attack_type"].append(attack_type)
+        samples["label"].append(label)
+
+        assert (subset_dir / "flac" / f"{sample_name}.flac").exists()
+        samples["path"].append(subset_dir / "flac" / f"{sample_name}.flac")
+
+        return samples
+
+class ASVSpoof2019DatasetOriginal(ASVSpoofDataset):
+
+    subsets = {"train": "train", "test": "dev", "val": "eval"}
+
+    protocol_folder_name = "ASVspoof2019_LA_cm_protocols"
+    subset_dir_prefix = "ASVspoof2019_LA_"
+    subset_dirs_attacks = {
+        "train": ["A01", "A02", "A03", "A04", "A05", "A06"],
+        "dev":  ["A01", "A02", "A03", "A04", "A05", "A06"],
+        "eval": [
+            "A07", "A08", "A09", "A10", "A11",  "A12", "A13", "A14", "A15",
+            "A16", "A17", "A18", "A19"
+        ]
+    }
+
+
+    def __init__(self, path, fold_subset="train"):
+        """
+        Initialise object. Skip __init__ of ASVSpoofDataset doe to different 
+        logic, but follow SimpleAudioFakeDataset constructor.
+        """
+        super(ASVSpoofDataset, self).__init__(float('inf'), fold_subset)
+        self.path = path
+        subset = self.subsets[fold_subset]
+        self.allowed_attacks = self.subset_dirs_attacks[subset]
+        subset_dir = Path(self.path) / f"{self.subset_dir_prefix}{subset}"
+        subset_protocol_path = self.get_protocol_path(subset)
+        self.samples = self.read_protocol(subset_dir, subset_protocol_path)
+
+    def read_protocol(self, subset_dir, protocol_path):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        real_samples = []
+        fake_samples = []
+
+        with open(protocol_path, "r") as file:
+            for line in file:
+                attack_type = line.strip().split(" ")[3]
+                if attack_type == "-":
+                    real_samples.append(line)
+                elif attack_type in self.allowed_attacks:
+                    fake_samples.append(line)
+                else:
+                    raise ValueError(
+                        "Tried to load attack that shouldn't be here!"
+                    )
+
+        for line in fake_samples:
+            samples = self.add_line_to_samples(samples, line, subset_dir)
+        for line in real_samples:
+            samples = self.add_line_to_samples(samples, line, subset_dir)
+
+        return pd.DataFrame(samples)
+

src/datasets/base_dataset.py

@@ -0,0 +1,172 @@
+"""Base dataset classes."""
+import logging
+import math
+import random
+
+import numpy as np
+import pandas as pd
+import torch
+import torchaudio
+from torch.utils.data import Dataset
+from torch.utils.data.dataset import T_co
+
+
+LOGGER = logging.getLogger(__name__)
+
+SAMPLING_RATE = 16_000
+APPLY_NORMALIZATION = True
+APPLY_TRIMMING = True
+APPLY_PADDING = True
+FRAMES_NUMBER = 480_000  # <- originally 64_600
+
+
+SOX_SILENCE = [
+    # trim all silence that is longer than 0.2s and louder than 1% volume (relative to the file)
+    # from beginning and middle/end
+    ["silence", "1", "0.2", "1%", "-1", "0.2", "1%"],
+]
+
+
+class SimpleAudioFakeDataset(Dataset):
+    def __init__(
+        self,
+        subset,
+        transform=None,
+        return_label: bool = True,
+        return_meta: bool = False,
+    ):
+        self.transform = transform
+        self.samples = pd.DataFrame()
+
+        self.subset = subset
+        self.allowed_attacks = None
+        self.partition_ratio = None
+        self.seed = None
+        self.return_label = return_label
+        self.return_meta = return_meta
+
+    def split_samples(self, samples_list):
+        if isinstance(samples_list, pd.DataFrame):
+            samples_list = samples_list.sort_values(by=list(samples_list.columns))
+            samples_list = samples_list.sample(frac=1, random_state=self.seed)
+        else:
+            samples_list = sorted(samples_list)
+            random.seed(self.seed)
+            random.shuffle(samples_list)
+
+        p, s = self.partition_ratio
+        subsets = np.split(
+            samples_list, [int(p * len(samples_list)), int((p + s) * len(samples_list))]
+        )
+        return dict(zip(["train", "test", "val"], subsets))[self.subset]
+
+    def df2tuples(self):
+        tuple_samples = []
+        for i, elem in self.samples.iterrows():
+            tuple_samples.append(
+                (str(elem["path"]), elem["label"], elem["attack_type"])
+            )
+
+        self.samples = tuple_samples
+        return self.samples
+
+    def __getitem__(self, index) -> T_co:
+        if isinstance(self.samples, pd.DataFrame):
+            sample = self.samples.iloc[index]
+
+            path = str(sample["path"])
+            label = sample["label"]
+            attack_type = sample["attack_type"]
+            if type(attack_type) != str and math.isnan(attack_type):
+                attack_type = "N/A"
+        else:
+            path, label, attack_type = self.samples[index]
+
+        waveform, sample_rate = torchaudio.load(path, normalize=APPLY_NORMALIZATION)
+        real_sec_length = len(waveform[0]) / sample_rate
+
+        waveform, sample_rate = apply_preprocessing(waveform, sample_rate)
+
+        return_data = [waveform, sample_rate]
+        if self.return_label:
+            label = 1 if label == "bonafide" else 0
+            return_data.append(label)
+
+        if self.return_meta:
+            return_data.append(
+                (
+                    attack_type,
+                    path,
+                    self.subset,
+                    real_sec_length,
+                )
+            )
+        return return_data
+
+    def __len__(self):
+        return len(self.samples)
+
+
+def apply_preprocessing(
+    waveform,
+    sample_rate,
+):
+    if sample_rate != SAMPLING_RATE and SAMPLING_RATE != -1:
+        waveform, sample_rate = resample_wave(waveform, sample_rate, SAMPLING_RATE)
+
+    # Stereo to mono
+    if waveform.dim() > 1 and waveform.shape[0] > 1:
+        waveform = waveform[:1, ...]
+
+    # Trim too long utterances...
+    if APPLY_TRIMMING:
+        waveform, sample_rate = apply_trim(waveform, sample_rate)
+
+    # ... or pad too short ones.
+    if APPLY_PADDING:
+        waveform = apply_pad(waveform, FRAMES_NUMBER)
+
+    return waveform, sample_rate
+
+
+def resample_wave(waveform, sample_rate, target_sample_rate):
+    waveform, sample_rate = torchaudio.sox_effects.apply_effects_tensor(
+        waveform, sample_rate, [["rate", f"{target_sample_rate}"]]
+    )
+    return waveform, sample_rate
+
+
+def resample_file(path, target_sample_rate, normalize=True):
+    waveform, sample_rate = torchaudio.sox_effects.apply_effects_file(
+        path, [["rate", f"{target_sample_rate}"]], normalize=normalize
+    )
+
+    return waveform, sample_rate
+
+
+def apply_trim(waveform, sample_rate):
+    (
+        waveform_trimmed,
+        sample_rate_trimmed,
+    ) = torchaudio.sox_effects.apply_effects_tensor(waveform, sample_rate, SOX_SILENCE)
+
+    if waveform_trimmed.size()[1] > 0:
+        waveform = waveform_trimmed
+        sample_rate = sample_rate_trimmed
+
+    return waveform, sample_rate
+
+
+def apply_pad(waveform, cut):
+    """Pad wave by repeating signal until `cut` length is achieved."""
+    waveform = waveform.squeeze(0)
+    waveform_len = waveform.shape[0]
+
+    if waveform_len >= cut:
+        return waveform[:cut]
+
+    # need to pad
+    num_repeats = int(cut / waveform_len) + 1
+    padded_waveform = torch.tile(waveform, (1, num_repeats))[:, :cut][0]
+
+    return padded_waveform

src/datasets/deepfake_asvspoof_dataset.py

@@ -0,0 +1,86 @@
+import logging
+from pathlib import Path
+
+import pandas as pd
+
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+
+DF_ASVSPOOF_SPLIT = {
+    "partition_ratio": [0.7, 0.15],
+    "seed": 45
+}
+
+LOGGER = logging.getLogger()
+
+class DeepFakeASVSpoofDataset(SimpleAudioFakeDataset):
+
+    protocol_file_name = "keys/CM/trial_metadata.txt"
+    subset_dir_prefix = "ASVspoof2021_DF_eval"
+    subset_parts = ("part00", "part01", "part02", "part03")
+
+    def __init__(self, path, subset="train", transform=None):
+        super().__init__(subset, transform)
+        self.path = path
+
+        self.partition_ratio = DF_ASVSPOOF_SPLIT["partition_ratio"]
+        self.seed = DF_ASVSPOOF_SPLIT["seed"]
+
+        self.flac_paths = self.get_file_references()
+        self.samples = self.read_protocol()
+
+        self.transform = transform
+        LOGGER.info(f"Spoof: {len(self.samples[self.samples['label'] == 'spoof'])}")
+        LOGGER.info(f"Original: {len(self.samples[self.samples['label'] == 'bonafide'])}")
+
+    def get_file_references(self):
+        flac_paths = {}
+        for part in self.subset_parts:
+            path = Path(self.path) / f"{self.subset_dir_prefix}_{part}" / self.subset_dir_prefix / "flac"
+            flac_list = list(path.glob("*.flac"))
+
+            for path in flac_list:
+                flac_paths[path.stem] = path
+
+        return flac_paths
+
+    def read_protocol(self):
+        samples = {
+            "sample_name": [],
+            "label": [],
+            "path": [],
+            "attack_type": [],
+        }
+
+        real_samples = []
+        fake_samples = []
+        with open(Path(self.path) / self.protocol_file_name, "r") as file:
+            for line in file:
+                label = line.strip().split(" ")[5]
+
+                if label == "bonafide":
+                    real_samples.append(line)
+                elif label == "spoof":
+                    fake_samples.append(line)
+
+        fake_samples = self.split_samples(fake_samples)
+        for line in fake_samples:
+            samples = self.add_line_to_samples(samples, line)
+
+        real_samples = self.split_samples(real_samples)
+        for line in real_samples:
+            samples = self.add_line_to_samples(samples, line)
+
+        return pd.DataFrame(samples)
+
+    def add_line_to_samples(self, samples, line):
+        _, sample_name, _, _, _, label, _, _ = line.strip().split(" ")
+        samples["sample_name"].append(sample_name)
+        samples["label"].append(label)
+        samples["attack_type"].append(label)
+
+        sample_path = self.flac_paths[sample_name]
+        assert sample_path.exists()
+        samples["path"].append(sample_path)
+
+        return samples
+

src/datasets/detection_dataset.py

@@ -0,0 +1,156 @@
+import logging
+from typing import List, Optional
+
+
+import pandas as pd
+
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+from src.datasets.deepfake_asvspoof_dataset import DeepFakeASVSpoofDataset
+from src.datasets.fakeavceleb_dataset import FakeAVCelebDataset
+from src.datasets.wavefake_dataset import WaveFakeDataset
+from src.datasets.asvspoof_dataset import ASVSpoof2019DatasetOriginal
+from src.datasets.MLAADv3_dataset import MLAADv3
+from src.datasets.MAILABS_dataset import MAILABS
+from src.datasets.aihub_dataset import AIHUB
+from src.datasets.KoAAD_dataset import KoAAD
+
+
+LOGGER = logging.getLogger()
+
+
+class DetectionDataset(SimpleAudioFakeDataset):
+    def __init__(
+        self,
+        asvspoof_path=None,
+        wavefake_path=None,
+        fakeavceleb_path=None,
+        asvspoof2019_path=None,
+        MLAADv3_path=None,
+        MAILABS_path=None,
+        AIHUB_path=None,
+        KoAAD_path=None,
+        subset: str = "val",
+        transform=None,
+        oversample: bool = True,
+        undersample: bool = False,
+        return_label: bool = True,
+        reduced_number: Optional[int] = None,
+        return_meta: bool = False,
+    ):
+        super().__init__(
+            subset=subset,
+            transform=transform,
+            return_label=return_label,
+            return_meta=return_meta,
+        )
+        datasets = self._init_datasets(
+            asvspoof_path=asvspoof_path,
+            wavefake_path=wavefake_path,
+            fakeavceleb_path=fakeavceleb_path,
+            asvspoof2019_path=asvspoof2019_path,
+            MLAADv3_path=MLAADv3_path,
+            MAILABS_path=MAILABS_path,
+            AIHUB_path=AIHUB_path,
+            KoAAD_path=KoAAD_path,
+            subset=subset,
+        )
+        self.samples = pd.concat([ds.samples for ds in datasets], ignore_index=True)
+
+        if oversample:
+            self.oversample_dataset()
+        elif undersample:
+            self.undersample_dataset()
+
+        if reduced_number:
+            LOGGER.info(f"Using reduced number of samples - {reduced_number}!")
+            self.samples = self.samples.sample(
+                min(len(self.samples), reduced_number),
+                random_state=42,
+            )
+
+    def _init_datasets(
+        self,
+        subset: str,
+        asvspoof_path: Optional[str],
+        wavefake_path: Optional[str],
+        fakeavceleb_path: Optional[str],
+        asvspoof2019_path: Optional[str],
+        MLAADv3_path=Optional[str],
+        MAILABS_path=Optional[str],
+        AIHUB_path=Optional[str],
+        KoAAD_path=Optional[str],
+    ) -> List[SimpleAudioFakeDataset]:
+        datasets = []
+
+        if asvspoof_path is not None:
+            asvspoof_dataset = DeepFakeASVSpoofDataset(asvspoof_path, subset=subset)
+            datasets.append(asvspoof_dataset)
+
+        if wavefake_path is not None:
+            wavefake_dataset = WaveFakeDataset(wavefake_path, subset=subset)
+            datasets.append(wavefake_dataset)
+
+        if fakeavceleb_path is not None:
+            fakeavceleb_dataset = FakeAVCelebDataset(fakeavceleb_path, subset=subset)
+            datasets.append(fakeavceleb_dataset)
+
+        if asvspoof2019_path is not None:
+            la_dataset = ASVSpoof2019DatasetOriginal(
+                asvspoof2019_path, fold_subset=subset
+            )
+            datasets.append(la_dataset)
+
+        if MLAADv3_path is not None:
+            MLAADv3_dataset = MLAADv3(MLAADv3_path, subset=subset)
+            datasets.append(MLAADv3_dataset)
+
+        if MAILABS_path is not None:
+            MAILABS_dataset = MAILABS(MAILABS_path, subset=subset)
+            datasets.append(MAILABS_dataset)
+
+        if AIHUB_path is not None:
+            aihub_dataset = AIHUB(AIHUB_path, subset=subset)
+            datasets.append(aihub_dataset)
+
+        if KoAAD_path is not None:
+            KoAAD_dataset = KoAAD(KoAAD_path, subset=subset)
+            datasets.append(KoAAD_dataset)
+        return datasets
+
+    def oversample_dataset(self):
+        samples = self.samples.groupby(by=["label"])
+        bona_length = len(samples.groups["bonafide"])
+        spoof_length = len(samples.groups["spoof"])
+
+        diff_length = spoof_length - bona_length
+
+        if diff_length < 0:
+            raise NotImplementedError
+
+        if diff_length > 0:
+            bonafide = samples.get_group("bonafide").sample(diff_length, replace=True)
+            self.samples = pd.concat([self.samples, bonafide], ignore_index=True)
+
+    def undersample_dataset(self):
+        samples = self.samples.groupby(by=["label"])
+        bona_length = len(samples.groups["bonafide"])
+        spoof_length = len(samples.groups["spoof"])
+
+        if spoof_length < bona_length:
+            raise NotImplementedError
+
+        if spoof_length > bona_length:
+            spoofs = samples.get_group("spoof").sample(bona_length, replace=True)
+            self.samples = pd.concat(
+                [samples.get_group("bonafide"), spoofs], ignore_index=True
+            )
+
+    def get_bonafide_only(self):
+        samples = self.samples.groupby(by=["label"])
+        self.samples = samples.get_group("bonafide")
+        return self.samples
+
+    def get_spoof_only(self):
+        samples = self.samples.groupby(by=["label"])
+        self.samples = samples.get_group("spoof")
+        return self.samples

src/datasets/fakeavceleb_dataset.py

@@ -0,0 +1,94 @@
+from pathlib import Path
+
+import pandas as pd
+
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+
+FAKEAVCELEB_SPLIT = {
+    "train": ['faceswap-wav2lip', 'fsgan-wav2lip', 'wav2lip', 'rtvc'],
+    "test":  ['faceswap-wav2lip', 'fsgan-wav2lip', 'wav2lip', 'rtvc'],
+    "val":   ['faceswap-wav2lip', 'fsgan-wav2lip', 'wav2lip', 'rtvc'],
+    "partition_ratio": [0.7, 0.15],
+    "seed": 45
+}
+
+
+class FakeAVCelebDataset(SimpleAudioFakeDataset):
+
+    audio_folder = "FakeAVCeleb-audio"
+    audio_extension = ".mp3"
+    metadata_file = Path(audio_folder) / "meta_data.csv"
+    subsets = ("train", "dev", "eval")
+
+    def __init__(self, path, subset="train", transform=None):
+        super().__init__(subset, transform)
+        self.path = path
+
+        self.subset = subset
+        self.allowed_attacks = FAKEAVCELEB_SPLIT[subset]
+        self.partition_ratio = FAKEAVCELEB_SPLIT["partition_ratio"]
+        self.seed = FAKEAVCELEB_SPLIT["seed"]
+
+        self.metadata = self.get_metadata()
+
+        self.samples = pd.concat([self.get_fake_samples(), self.get_real_samples()], ignore_index=True)
+
+    def get_metadata(self):
+        md = pd.read_csv(Path(self.path) / self.metadata_file)
+        md["audio_type"] = md["type"].apply(lambda x: x.split("-")[-1])
+        return md
+
+    def get_fake_samples(self):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        for attack_name in self.allowed_attacks:
+            fake_samples = self.metadata[
+                (self.metadata["method"] == attack_name) & (self.metadata["audio_type"] == "FakeAudio")
+            ]
+
+            samples_list = fake_samples.iterrows()
+            samples_list = self.split_samples(samples_list)
+
+            for _, sample in samples_list:
+                samples["user_id"].append(sample["source"])
+                samples["sample_name"].append(Path(sample["filename"]).stem)
+                samples["attack_type"].append(sample["method"])
+                samples["label"].append("spoof")
+                samples["path"].append(self.get_file_path(sample))
+
+        return pd.DataFrame(samples)
+
+    def get_real_samples(self):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        samples_list = self.metadata[
+            (self.metadata["method"] == "real") & (self.metadata["audio_type"] == "RealAudio")
+        ]
+
+        samples_list = self.split_samples(samples_list)
+
+        for index, sample in samples_list.iterrows():
+            samples["user_id"].append(sample["source"])
+            samples["sample_name"].append(Path(sample["filename"]).stem)
+            samples["attack_type"].append("-")
+            samples["label"].append("bonafide")
+            samples["path"].append(self.get_file_path(sample))
+
+        return pd.DataFrame(samples)
+
+    def get_file_path(self, sample):
+        path = "/".join([self.audio_folder, *sample["path"].split("/")[1:]])
+        return Path(self.path) / path / Path(sample["filename"]).with_suffix(self.audio_extension)
+

src/datasets/in_the_wild_dataset.py

@@ -0,0 +1,62 @@
+import numpy as np
+import pandas as pd
+from pathlib import Path
+
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+
+
+class InTheWildDataset(SimpleAudioFakeDataset):
+
+    def __init__(
+        self,
+        path,
+        subset="train",
+        transform=None,
+        seed=None,
+        partition_ratio=(0.7, 0.15),
+        split_strategy="random"
+    ):
+        super().__init__(subset=subset, transform=transform)
+        self.path = path
+        self.read_samples()
+        self.partition_ratio = partition_ratio
+        self.seed = seed
+
+
+    def read_samples(self):
+        path = Path(self.path)
+        meta_path = path / "meta.csv"
+
+        self.samples = pd.read_csv(meta_path)
+        self.samples["path"] = self.samples["file"].apply(lambda n: str(path / n))
+        self.samples["file"] = self.samples["file"].apply(lambda n: Path(n).stem)
+        self.samples["label"] = self.samples["label"].map({"bona-fide": "bonafide", "spoof": "spoof"})
+        self.samples["attack_type"] = self.samples["label"].map({"bonafide": "-", "spoof": "X"})
+        self.samples.rename(columns={'file': 'sample_name', 'speaker': 'user_id'}, inplace=True)
+
+
+    def split_samples_per_speaker(self, samples):
+        speaker_list = pd.Series(samples["user_id"].unique())
+        speaker_list = speaker_list.sort_values()
+        speaker_list = speaker_list.sample(frac=1, random_state=self.seed)
+        speaker_list = list(speaker_list)
+
+        p, s = self.partition_ratio
+        subsets = np.split(speaker_list, [int(p * len(speaker_list)), int((p + s) * len(speaker_list))])
+        speaker_subset = dict(zip(['train', 'test', 'val'], subsets))[self.subset]
+        return self.samples[self.samples["user_id"].isin(speaker_subset)]
+
+
+if __name__ == "__main__":
+    dataset = InTheWildDataset(
+        path="../datasets/release_in_the_wild",
+        subset="val",
+        seed=242,
+        split_strategy="per_speaker"
+    )
+
+    print(len(dataset))
+    print(len(dataset.samples["user_id"].unique()))
+    print(dataset.samples["user_id"].unique())
+
+    print(dataset[0])

src/datasets/wavefake_dataset.py

@@ -0,0 +1,85 @@
+from pathlib import Path
+
+import pandas as pd
+
+from src.datasets.base_dataset import SimpleAudioFakeDataset
+
+WAVEFAKE_SPLIT = {
+    "train": ['multi_band_melgan', 'melgan_large', 'parallel_wavegan', 'waveglow', 'full_band_melgan', 'melgan', 'hifiGAN'],
+    "test":  ['multi_band_melgan', 'melgan_large', 'parallel_wavegan', 'waveglow', 'full_band_melgan', 'melgan', 'hifiGAN'],
+    "val":   ['multi_band_melgan', 'melgan_large', 'parallel_wavegan', 'waveglow', 'full_band_melgan', 'melgan', 'hifiGAN'],
+    "partition_ratio": [0.7, 0.15],
+    "seed": 45
+}
+
+
+class WaveFakeDataset(SimpleAudioFakeDataset):
+
+    fake_data_path = "generated_audio"
+    jsut_real_data_path = "real_audio/jsut_ver1.1/basic5000/wav"
+    ljspeech_real_data_path = "real_audio/LJSpeech-1.1/wavs"
+
+    def __init__(self, path, subset="train", transform=None):
+        super().__init__(subset, transform)
+        self.path = Path(path)
+
+        self.fold_subset = subset
+        self.allowed_attacks = WAVEFAKE_SPLIT[subset]
+        self.partition_ratio = WAVEFAKE_SPLIT["partition_ratio"]
+        self.seed = WAVEFAKE_SPLIT["seed"]
+
+        self.samples = pd.concat([self.get_fake_samples(), self.get_real_samples()], ignore_index=True)
+
+    def get_fake_samples(self):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        samples_list = list((self.path / self.fake_data_path).glob("*/*.wav"))
+        samples_list = self.filter_samples_by_attack(samples_list)
+        samples_list = self.split_samples(samples_list)
+
+        for sample in samples_list:
+            samples["user_id"].append(None)
+            samples["sample_name"].append("_".join(sample.stem.split("_")[:-1]))
+            samples["attack_type"].append(self.get_attack_from_path(sample))
+            samples["label"].append("spoof")
+            samples["path"].append(sample)
+
+        return pd.DataFrame(samples)
+
+    def filter_samples_by_attack(self, samples_list):
+        return [s for s in samples_list if self.get_attack_from_path(s) in self.allowed_attacks]
+
+    def get_real_samples(self):
+        samples = {
+            "user_id": [],
+            "sample_name": [],
+            "attack_type": [],
+            "label": [],
+            "path": []
+        }
+
+        samples_list = list((self.path / self.jsut_real_data_path).glob("*.wav"))
+        samples_list += list((self.path / self.ljspeech_real_data_path).glob("*.wav"))
+        samples_list = self.split_samples(samples_list)
+
+        for sample in samples_list:
+            samples["user_id"].append(None)
+            samples["sample_name"].append(sample.stem)
+            samples["attack_type"].append("-")
+            samples["label"].append("bonafide")
+            samples["path"].append(sample)
+
+        return pd.DataFrame(samples)
+
+    @staticmethod
+    def get_attack_from_path(path):
+        folder_name = path.parents[0].relative_to(path.parents[1])
+        return str(folder_name).split("_", maxsplit=1)[-1]
+
+

src/frontends.py

@@ -0,0 +1,72 @@
+from typing import List, Union, Callable
+
+import torch
+import torchaudio
+
+SAMPLING_RATE = 16_000
+win_length = 400  # int((25 / 1_000) * SAMPLING_RATE)
+hop_length = 160  # int((10 / 1_000) * SAMPLING_RATE)
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+
+MFCC_FN = torchaudio.transforms.MFCC(
+    sample_rate=SAMPLING_RATE,
+    n_mfcc=128,
+    melkwargs={
+        "n_fft": 512,
+        "win_length": win_length,
+        "hop_length": hop_length,
+    },
+).to(device)
+
+
+LFCC_FN = torchaudio.transforms.LFCC(
+    sample_rate=SAMPLING_RATE,
+    n_lfcc=128,
+    speckwargs={
+        "n_fft": 512,
+        "win_length": win_length,
+        "hop_length": hop_length,
+    },
+).to(device)
+
+MEL_SCALE_FN = torchaudio.transforms.MelScale(
+    n_mels=80,
+    n_stft=257,
+    sample_rate=SAMPLING_RATE,
+).to(device)
+
+delta_fn = torchaudio.transforms.ComputeDeltas(
+    win_length=400,
+    mode="replicate",
+)
+
+
+def get_frontend(
+    frontends: List[str],
+) -> Union[torchaudio.transforms.MFCC, torchaudio.transforms.LFCC, Callable,]:
+    if "mfcc" in frontends:
+        return prepare_mfcc_double_delta
+    elif "lfcc" in frontends:
+        return prepare_lfcc_double_delta
+    raise ValueError(f"{frontends} frontend is not supported!")
+
+
+def prepare_lfcc_double_delta(input):
+    if input.ndim < 4:
+        input = input.unsqueeze(1)  # (bs, 1, n_lfcc, frames)
+    x = LFCC_FN(input)
+    delta = delta_fn(x)
+    double_delta = delta_fn(delta)
+    x = torch.cat((x, delta, double_delta), 2)  # -> [bs, 1, 128 * 3, 1500]
+    return x[:, :, :, :512]  # (bs, n, n_lfcc * 3, frames)
+
+
+def prepare_mfcc_double_delta(input):
+    if input.ndim < 4:
+        input = input.unsqueeze(1)  # (bs, 1, n_lfcc, frames)
+    x = MFCC_FN(input)
+    delta = delta_fn(x)
+    double_delta = delta_fn(delta)
+    x = torch.cat((x, delta, double_delta), 2)  # -> [bs, 1, 128 * 3, 1500]
+    return x[:, :, :, :512]  # (bs, n, n_lfcc * 3, frames)

src/metrics.py

@@ -0,0 +1,15 @@
+from typing import Tuple
+
+import numpy as np
+from scipy.interpolate import interp1d
+from scipy.optimize import brentq
+from sklearn.metrics import roc_curve
+from sklearn.metrics import roc_curve
+
+
+def calculate_eer(y, y_score) -> Tuple[float, float, np.ndarray, np.ndarray]:
+    fpr, tpr, thresholds = roc_curve(y, -y_score)
+
+    eer = brentq(lambda x: 1.0 - x - interp1d(fpr, tpr)(x), 0.0, 1.0)
+    thresh = interp1d(fpr, thresholds)(eer)
+    return thresh, eer, fpr, tpr

src/trainer.py

@@ -0,0 +1,173 @@
+"""A generic training wrapper."""
+from copy import deepcopy
+import logging
+from typing import Callable, List, Optional
+
+import torch
+from torch.utils.data import DataLoader
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class Trainer:
+    def __init__(
+        self,
+        epochs: int = 20,
+        batch_size: int = 32,
+        device: str = "cpu",
+        optimizer_fn: Callable = torch.optim.Adam,
+        optimizer_kwargs: dict = {"lr": 1e-3},
+        use_scheduler: bool = False,
+    ) -> None:
+        self.epochs = epochs
+        self.batch_size = batch_size
+        self.device = device
+        self.optimizer_fn = optimizer_fn
+        self.optimizer_kwargs = optimizer_kwargs
+        self.epoch_test_losses: List[float] = []
+        self.use_scheduler = use_scheduler
+
+
+def forward_and_loss(model, criterion, batch_x, batch_y, **kwargs):
+    batch_out = model(batch_x)
+    batch_loss = criterion(batch_out, batch_y)
+    return batch_out, batch_loss
+
+
+class GDTrainer(Trainer):
+    def train(
+        self,
+        dataset: torch.utils.data.Dataset,
+        model: torch.nn.Module,
+        test_len: Optional[float] = None,
+        test_dataset: Optional[torch.utils.data.Dataset] = None,
+    ):
+        if test_dataset is not None:
+            train = dataset
+            test = test_dataset
+        else:
+            test_len = int(len(dataset) * test_len)
+            train_len = len(dataset) - test_len
+            lengths = [train_len, test_len]
+            train, test = torch.utils.data.random_split(dataset, lengths)
+
+        train_loader = DataLoader(
+            train,
+            batch_size=self.batch_size,
+            shuffle=True,
+            drop_last=True,
+            num_workers=6,
+        )
+        test_loader = DataLoader(
+            test,
+            batch_size=self.batch_size,
+            shuffle=True,
+            drop_last=True,
+            num_workers=6,
+        )
+
+        criterion = torch.nn.BCEWithLogitsLoss()
+        optim = self.optimizer_fn(model.parameters(), **self.optimizer_kwargs)
+
+        best_model = None
+        best_acc = 0
+
+        LOGGER.info(f"Starting training for {self.epochs} epochs!")
+
+        forward_and_loss_fn = forward_and_loss
+
+        if self.use_scheduler:
+            batches_per_epoch = len(train_loader) * 2  # every 2nd epoch
+            scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(
+                optimizer=optim,
+                T_0=batches_per_epoch,
+                T_mult=1,
+                eta_min=5e-6,
+                # verbose=True,
+            )
+        use_cuda = self.device != "cpu"
+
+        for epoch in range(self.epochs):
+            LOGGER.info(f"Epoch num: {epoch}")
+
+            running_loss = 0
+            num_correct = 0.0
+            num_total = 0.0
+            model.train()
+
+            for i, (batch_x, _, batch_y) in enumerate(train_loader):
+                batch_size = batch_x.size(0)
+                num_total += batch_size
+                batch_x = batch_x.to(self.device)
+
+                batch_y = batch_y.unsqueeze(1).type(torch.float32).to(self.device)
+
+                batch_out, batch_loss = forward_and_loss_fn(
+                    model, criterion, batch_x, batch_y, use_cuda=use_cuda
+                )
+                batch_pred = (torch.sigmoid(batch_out) + 0.5).int()
+                num_correct += (batch_pred == batch_y.int()).sum(dim=0).item()
+
+                running_loss += batch_loss.item() * batch_size
+
+                if i % 100 == 0:
+                    LOGGER.info(
+                        f"[{epoch:04d}][{i:05d}]: {running_loss / num_total} {num_correct/num_total*100}"
+                    )
+
+                optim.zero_grad()
+                batch_loss.backward()
+                optim.step()
+                if self.use_scheduler:
+                    scheduler.step()
+
+            running_loss /= num_total
+            train_accuracy = (num_correct / num_total) * 100
+
+            LOGGER.info(
+                f"Epoch [{epoch+1}/{self.epochs}]: train/loss: {running_loss}, train/accuracy: {train_accuracy}"
+            )
+
+            test_running_loss = 0.0
+            num_correct = 0.0
+            num_total = 0.0
+            model.eval()
+            eer_val = 0
+
+            for batch_x, _, batch_y in test_loader:
+                batch_size = batch_x.size(0)
+                num_total += batch_size
+                batch_x = batch_x.to(self.device)
+
+                with torch.no_grad():
+                    batch_pred = model(batch_x)
+
+                batch_y = batch_y.unsqueeze(1).type(torch.float32).to(self.device)
+                batch_loss = criterion(batch_pred, batch_y)
+
+                test_running_loss += batch_loss.item() * batch_size
+
+                batch_pred = torch.sigmoid(batch_pred)
+                batch_pred_label = (batch_pred + 0.5).int()
+                num_correct += (batch_pred_label == batch_y.int()).sum(dim=0).item()
+
+            if num_total == 0:
+                num_total = 1
+
+            test_running_loss /= num_total
+            test_acc = 100 * (num_correct / num_total)
+            LOGGER.info(
+                f"Epoch [{epoch+1}/{self.epochs}]: test/loss: {test_running_loss}, test/accuracy: {test_acc}, test/eer: {eer_val}"
+            )
+
+            if best_model is None or test_acc > best_acc:
+                best_acc = test_acc
+                best_model = deepcopy(model.state_dict())
+
+            LOGGER.info(
+                f"[{epoch:04d}]: {running_loss} - train acc: {train_accuracy} - test_acc: {test_acc}"
+            )
+
+        model.load_state_dict(best_model)
+        return model