{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":80640,"databundleVersionId":8654967,"sourceType":"competition"}],"dockerImageVersionId":30698,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## Library 설정 및 필요 함수 정의\n\n수정 필요 없음\n\n**find_wav_files**: 디렉토리 내에 있는 .wav 파일들을 리스트화하는 함수\n\n**set_seed**: 랜덤 시드 설정에 대한 함수","metadata":{}},{"cell_type":"code","source":"import os\nimport random\nfrom pathlib import Path\nimport numpy as np\nimport torch\nfrom torch import Tensor\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch.utils.data import ConcatDataset, DataLoader\nimport torchaudio\nimport logging\nfrom copy import deepcopy\nimport csv\nfrom typing import Callable, List, Optional, Tuple, Union\nfrom collections import OrderedDict\n\n\n### Utils\n\ndef find_wav_files(path_to_dir: Union[Path, str]) -> Optional[List[Path]]:\n    \"\"\"Find all wav files in the directory and its subtree.\n\n    Args:\n        path_to_dir: Path top directory.\n    Returns:\n        List containing Path objects or None (nothing found).\n    \"\"\"\n    paths = list(sorted(Path(path_to_dir).glob(\"**/*.wav\")))\n\n    if len(paths) == 0:\n        return None\n    return paths\n\n\ndef set_seed(seed: int):\n    \"\"\"Fix PRNG seed for reproducable experiments.\n    \"\"\"\n    random.seed(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    if torch.cuda.is_available():\n        torch.cuda.manual_seed(seed)\n        torch.cuda.manual_seed_all(seed)\n        torch.backends.cudnn.deterministic = True\n        torch.backends.cudnn.benchmark = False\n    os.environ[\"PYTHONHASHSEED\"] = str(seed)","metadata":{"execution":{"iopub.status.busy":"2024-05-26T02:39:37.277955Z","iopub.execute_input":"2024-05-26T02:39:37.278223Z","iopub.status.idle":"2024-05-26T02:39:41.628364Z","shell.execute_reply.started":"2024-05-26T02:39:37.278200Z","shell.execute_reply":"2024-05-26T02:39:41.627439Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## 데이터셋 관련 코드\n\n**AudioDataset**: 데이터셋 관련 코드로 .wav파일 리스트 수행하고 샘플레이트를 각 음성에 대해 동일하게 맞춰줌\n\n`__init__`: .wav파일 리스트 수행. 데이터셋이 real 음성 / fake 음성의 비율이 1:7이라 두 음성의 비율을 맞춰주기 위해 real 음성의 샘플 수를 7배 늘려줌 \n\n`__getitem__`: wav파일을 읽어서 tensor로 바꿔줌. 샘플레이트를 각 음성에 대해 동일하게 맞춰줌. tensor와 샘플레이트를 리턴\n\n**PadDataset**: 모든 음성의 길이를 4초로 맞춰줌. 음성이 4초보다 길 경우 0초~4초로 자르고, 4초보다 짧은 경우 반복 재생하여 4초로 맞춰줌. 미니 배치 단위로 처리하기 떄문에 미니 배치 내의 각 샘플들의 길이를 동일하게 맞춰주는 작업으로 볼 수 있음\n\n**load_dataset**: 학습 데이터셋 불러오는 함수\n\n**load_dataset_test**: 테스트 데이터셋 불러오는 함수","metadata":{}},{"cell_type":"code","source":"class AudioDataset(torch.utils.data.Dataset):\n    def __init__(\n            self,\n            directory_or_path_list: Union[Union[str, Path], List[Union[str, Path]]],\n            sample_rate: int = 16_000,\n            normalize: bool = True,\n            real: str = 'real',\n    ) -> None:\n        super().__init__()\n\n        self.sample_rate = sample_rate\n        self.normalize = normalize\n        if real == 'real':\n            directory_or_path_list = directory_or_path_list * 5\n\n        if isinstance(directory_or_path_list, list):\n            paths = directory_or_path_list\n        elif isinstance(directory_or_path_list, Path) \\\n                or isinstance(directory_or_path_list, str):\n            directory = Path(directory_or_path_list)\n            if not directory.exists():\n                raise IOError(f\"Directory does not exists: {self.directory}\")\n\n            paths = find_wav_files(directory)\n            if paths is None:\n                raise IOError(\n                    f\"Directory did not contain wav files: {self.directory}\")\n        else:\n            raise TypeError(\n                f\"Supplied unsupported type for argument directory_or_path_list {type(directory_or_path_list)}!\")\n\n\n        self._paths = paths\n\n    def __getitem__(self, index: int) -> Tuple[torch.Tensor, int]:\n        path = self._paths[index]\n\n        waveform, sample_rate = torchaudio.load(path, normalize=self.normalize)\n\n        if sample_rate != self.sample_rate:\n            transform = torchaudio.transforms.Resample(sample_rate, self.sample_rate)\n            waveform = transform(waveform)\n\n\n        return waveform, sample_rate\n\n    def __len__(self) -> int:\n        return len(self._paths)\n\n\nclass PadDataset(torch.utils.data.Dataset):\n\n    def __init__(self, dataset: torch.utils.data.Dataset, cut: int = 64600, label=None):\n        self.dataset = dataset\n        self.cut = cut  # max 4 sec (ASVSpoof default)\n        self.label = label\n\n    def __getitem__(self, index):\n        waveform, sample_rate = self.dataset[index]\n        waveform = waveform.squeeze(0)\n        waveform_len = waveform.shape[0]\n        if waveform_len >= self.cut:\n            if self.label is None:\n                return waveform[:self.cut], sample_rate\n            else:\n                return waveform[:self.cut], sample_rate, self.label\n        # need to pad\n        num_repeats = int(self.cut / waveform_len)+1\n        padded_waveform = torch.tile(waveform, (1, num_repeats))[\n            :, :self.cut][0]\n\n        if self.label is None:\n            return padded_waveform, sample_rate\n        else:\n            return padded_waveform, sample_rate, self.label\n\n    def __len__(self):\n        return len(self.dataset)\n\n\n\n\ndef load_dataset(\n        path: Union[Path, str],\n        pad: bool = False,\n        train: str = 'train',\n        real: str = 'real',\n        label: Optional[int] = None,\n) -> Tuple[torch.utils.data.Dataset]:\n\n    cur_path = \"{}/{}/{}\".format(path,train,real)\n    \n    paths = find_wav_files(cur_path)\n    if paths is None:\n        raise IOError(f\"Could not load files from {path}!\")\n\n    LOGGER.info(f\"Loading data from {path}...!\")\n\n    train_dataset = AudioDataset(\n        paths, real=real)\n    if pad:\n        train_dataset = PadDataset(train_dataset, label=label)\n\n    return train_dataset\n\n\n\n\ndef load_dataset_test(\n        path: Union[Path, str],\n        pad: bool = True,\n) -> Tuple[torch.utils.data.Dataset]:\n\n\n    paths = find_wav_files(path)\n    if paths is None:\n        raise IOError(f\"Could not load files from {path}!\")\n\n\n    test_dataset = AudioDataset(\n        paths, real='fake')\n    if pad:\n        test_dataset = PadDataset(test_dataset, label=0)\n\n    return test_dataset","metadata":{"execution":{"iopub.status.busy":"2024-05-26T02:39:41.630270Z","iopub.execute_input":"2024-05-26T02:39:41.630756Z","iopub.status.idle":"2024-05-26T02:39:41.653028Z","shell.execute_reply.started":"2024-05-26T02:39:41.630698Z","shell.execute_reply":"2024-05-26T02:39:41.651911Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## 모델 관련 코드\n\n**RawNet**: RawNet2 코드\n\n`__init__`: 구조 생성\n\n`__forward__`: 순전파 수행. 입력 음성신호 x에 대해서 fake/real에 대한 확률 값을 softmax()를 취하여 출력\n","metadata":{}},{"cell_type":"code","source":"# raw_Net2\n\nclass SincConv(nn.Module):\n    @staticmethod\n    def to_mel(hz):\n        return 2595 * np.log10(1 + hz / 700)\n\n    @staticmethod\n    def to_hz(mel):\n        return 700 * (10 ** (mel / 2595) - 1)\n\n\n    def __init__(self, device,out_channels, kernel_size,in_channels=1,sample_rate=16000,\n                 stride=1, padding=0, dilation=1, bias=False, groups=1):\n\n        super(SincConv,self).__init__()\n\n        if in_channels != 1:\n            \n            msg = \"SincConv only support one input channel (here, in_channels = {%i})\" % (in_channels)\n            raise ValueError(msg)\n        \n        self.out_channels = out_channels\n        self.kernel_size = kernel_size\n        self.sample_rate=sample_rate\n\n        # Forcing the filters to be odd (i.e, perfectly symmetrics)\n        if kernel_size%2==0:\n            self.kernel_size=self.kernel_size+1\n\n        self.device=device   \n        self.stride = stride\n        self.padding = padding\n        self.dilation = dilation\n        \n        if bias:\n            raise ValueError('SincConv does not support bias.')\n        if groups > 1:\n            raise ValueError('SincConv does not support groups.')\n        \n        \n        # initialize filterbanks using Mel scale\n        NFFT = 512\n        f=int(self.sample_rate/2)*np.linspace(0,1,int(NFFT/2)+1)\n        fmel=self.to_mel(f)   # Hz to mel conversion\n        fmelmax=np.max(fmel)\n        fmelmin=np.min(fmel)\n        filbandwidthsmel=np.linspace(fmelmin,fmelmax,self.out_channels+1)\n        filbandwidthsf=self.to_hz(filbandwidthsmel)  # Mel to Hz conversion\n        self.mel=filbandwidthsf\n        self.hsupp=torch.arange(-(self.kernel_size-1)/2, (self.kernel_size-1)/2+1)\n        self.band_pass=torch.zeros(self.out_channels,self.kernel_size)\n    \n       \n        \n    def forward(self,x):\n        for i in range(len(self.mel)-1):\n            fmin=self.mel[i]\n            fmax=self.mel[i+1]\n            hHigh=(2*fmax/self.sample_rate)*np.sinc(2*fmax*self.hsupp/self.sample_rate)\n            hLow=(2*fmin/self.sample_rate)*np.sinc(2*fmin*self.hsupp/self.sample_rate)\n            hideal=hHigh-hLow\n            \n            self.band_pass[i,:]=Tensor(np.hamming(self.kernel_size))*Tensor(hideal)\n        \n        band_pass_filter=self.band_pass.to(self.device)\n\n        self.filters = (band_pass_filter).view(self.out_channels, 1, self.kernel_size)\n        \n        return F.conv1d(x, self.filters, stride=self.stride,\n                        padding=self.padding, dilation=self.dilation,\n                         bias=None, groups=1)\n\n\n        \nclass Residual_block(nn.Module):\n    def __init__(self, nb_filts, first = False):\n        super(Residual_block, self).__init__()\n        self.first = first\n        \n        if not self.first:\n            self.bn1 = nn.BatchNorm1d(num_features = nb_filts[0])\n        \n        self.lrelu = nn.LeakyReLU(negative_slope=0.3)\n        \n        self.conv1 = nn.Conv1d(in_channels = nb_filts[0],\n\t\t\tout_channels = nb_filts[1],\n\t\t\tkernel_size = 3,\n\t\t\tpadding = 1,\n\t\t\tstride = 1)\n        \n        self.bn2 = nn.BatchNorm1d(num_features = nb_filts[1])\n        self.conv2 = nn.Conv1d(in_channels = nb_filts[1],\n\t\t\tout_channels = nb_filts[1],\n\t\t\tpadding = 1,\n\t\t\tkernel_size = 3,\n\t\t\tstride = 1)\n        \n        if nb_filts[0] != nb_filts[1]:\n            self.downsample = True\n            self.conv_downsample = nn.Conv1d(in_channels = nb_filts[0],\n\t\t\t\tout_channels = nb_filts[1],\n\t\t\t\tpadding = 0,\n\t\t\t\tkernel_size = 1,\n\t\t\t\tstride = 1)\n            \n        else:\n            self.downsample = False\n        self.mp = nn.MaxPool1d(3)\n        \n    def forward(self, x):\n        identity = x\n        if not self.first:\n            out = self.bn1(x)\n            out = self.lrelu(out)\n        else:\n            out = x\n            \n        out = self.conv1(x)\n        out = self.bn2(out)\n        out = self.lrelu(out)\n        out = self.conv2(out)\n        \n        if self.downsample:\n            identity = self.conv_downsample(identity)\n            \n        out += identity\n        out = self.mp(out)\n        return out\n\n\nclass RawNet(nn.Module):\n    def __init__(self, d_args, device):\n        super(RawNet, self).__init__()\n\n        \n        self.device=device\n\n        self.Sinc_conv=SincConv(device=self.device,\n\t\t\tout_channels = d_args['filts'][0],\n\t\t\tkernel_size = d_args['first_conv'],\n                        in_channels = d_args['in_channels']\n        )\n        \n        self.first_bn = nn.BatchNorm1d(num_features = d_args['filts'][0])\n        self.selu = nn.SELU(inplace=True)\n        self.block0 = Residual_block(nb_filts = d_args['filts'][1], first = True)\n        self.block1 = Residual_block(nb_filts = d_args['filts'][1])\n        self.block2 = Residual_block(nb_filts = d_args['filts'][2])\n        d_args['filts'][2][0] = d_args['filts'][2][1]\n        self.block3 = Residual_block(nb_filts = d_args['filts'][2])\n        self.block4 = Residual_block(nb_filts = d_args['filts'][2])\n        self.block5 = Residual_block(nb_filts = d_args['filts'][2])\n        self.avgpool = nn.AdaptiveAvgPool1d(1)\n\n        self.fc_attention0 = self._make_attention_fc(in_features = d_args['filts'][1][-1],\n            l_out_features = d_args['filts'][1][-1])\n        self.fc_attention1 = self._make_attention_fc(in_features = d_args['filts'][1][-1],\n            l_out_features = d_args['filts'][1][-1])\n        self.fc_attention2 = self._make_attention_fc(in_features = d_args['filts'][2][-1],\n            l_out_features = d_args['filts'][2][-1])\n        self.fc_attention3 = self._make_attention_fc(in_features = d_args['filts'][2][-1],\n            l_out_features = d_args['filts'][2][-1])\n        self.fc_attention4 = self._make_attention_fc(in_features = d_args['filts'][2][-1],\n            l_out_features = d_args['filts'][2][-1])\n        self.fc_attention5 = self._make_attention_fc(in_features = d_args['filts'][2][-1],\n            l_out_features = d_args['filts'][2][-1])\n\n        self.bn_before_gru = nn.BatchNorm1d(num_features = d_args['filts'][2][-1])\n        self.gru = nn.GRU(input_size = d_args['filts'][2][-1],\n\t\t\thidden_size = d_args['gru_node'],\n\t\t\tnum_layers = d_args['nb_gru_layer'],\n\t\t\tbatch_first = True)\n\n        \n        self.fc1_gru = nn.Linear(in_features = d_args['gru_node'],\n\t\t\tout_features = d_args['nb_fc_node'])\n       \n        self.fc2_gru = nn.Linear(in_features = d_args['nb_fc_node'],\n\t\t\tout_features = d_args['nb_classes'],bias=True)\n\t\t\t\n       \n        self.sig = nn.Sigmoid()\n        \n    def forward(self, x, y = None):\n        \n        \n        nb_samp = x.shape[0]\n        len_seq = x.shape[1]\n        x=x.view(nb_samp,1,len_seq)\n        \n        x = self.Sinc_conv(x)    \n        x = F.max_pool1d(torch.abs(x), 3)\n        x = self.first_bn(x)\n        x =  self.selu(x)\n        \n        x0 = self.block0(x)\n        y0 = self.avgpool(x0).view(x0.size(0), -1) # torch.Size([batch, filter])\n        y0 = self.fc_attention0(y0)\n        y0 = self.sig(y0).view(y0.size(0), y0.size(1), -1)  # torch.Size([batch, filter, 1])\n        x = x0 * y0 + y0  # (batch, filter, time) x (batch, filter, 1)\n        \n\n        x1 = self.block1(x)\n        y1 = self.avgpool(x1).view(x1.size(0), -1) # torch.Size([batch, filter])\n        y1 = self.fc_attention1(y1)\n        y1 = self.sig(y1).view(y1.size(0), y1.size(1), -1)  # torch.Size([batch, filter, 1])\n        x = x1 * y1 + y1 # (batch, filter, time) x (batch, filter, 1)\n\n        x2 = self.block2(x)\n        y2 = self.avgpool(x2).view(x2.size(0), -1) # torch.Size([batch, filter])\n        y2 = self.fc_attention2(y2)\n        y2 = self.sig(y2).view(y2.size(0), y2.size(1), -1)  # torch.Size([batch, filter, 1])\n        x = x2 * y2 + y2 # (batch, filter, time) x (batch, filter, 1)\n\n        x3 = self.block3(x)\n        y3 = self.avgpool(x3).view(x3.size(0), -1) # torch.Size([batch, filter])\n        y3 = self.fc_attention3(y3)\n        y3 = self.sig(y3).view(y3.size(0), y3.size(1), -1)  # torch.Size([batch, filter, 1])\n        x = x3 * y3 + y3 # (batch, filter, time) x (batch, filter, 1)\n\n        x4 = self.block4(x)\n        y4 = self.avgpool(x4).view(x4.size(0), -1) # torch.Size([batch, filter])\n        y4 = self.fc_attention4(y4)\n        y4 = self.sig(y4).view(y4.size(0), y4.size(1), -1)  # torch.Size([batch, filter, 1])\n        x = x4 * y4 + y4 # (batch, filter, time) x (batch, filter, 1)\n\n        x5 = self.block5(x)\n        y5 = self.avgpool(x5).view(x5.size(0), -1) # torch.Size([batch, filter])\n        y5 = self.fc_attention5(y5)\n        y5 = self.sig(y5).view(y5.size(0), y5.size(1), -1)  # torch.Size([batch, filter, 1])\n        x = x5 * y5 + y5 # (batch, filter, time) x (batch, filter, 1)\n\n        x = self.bn_before_gru(x)\n        x = self.selu(x)\n        x = x.permute(0, 2, 1)     #(batch, filt, time) >> (batch, time, filt)\n        self.gru.flatten_parameters()\n        x, _ = self.gru(x)\n        x = x[:,-1,:]\n        x = self.fc1_gru(x)\n        x = self.fc2_gru(x).softmax(dim=1)\n        output = x\n        return output\n\n\n    def _make_attention_fc(self, in_features, l_out_features):\n        l_fc = []\n        l_fc.append(nn.Linear(in_features = in_features,\n\t\t\t        out_features = l_out_features))\n        return nn.Sequential(*l_fc)\n","metadata":{"execution":{"iopub.status.busy":"2024-05-26T02:39:41.654544Z","iopub.execute_input":"2024-05-26T02:39:41.654985Z","iopub.status.idle":"2024-05-26T02:39:41.784856Z","shell.execute_reply.started":"2024-05-26T02:39:41.654954Z","shell.execute_reply":"2024-05-26T02:39:41.783718Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## 학습 관련 코드\n\n**GDTrainer**\n\n`train`\n\n학습, 검증(validation) 데이터로더 생성\n\n손실 함수는 교차 엔트로피(cross entropy) 이용\n\noptimizer는 Adam 이용\n\n각 epoch에 대한 학습 및 검증 관련 내용 포함: 각 epoch의 학습이 수행되면 검증 수행하여 각 에포크의 정확도 계산\n\n\n`test`\n\n검증 정확도가 가장 높은 모델을 불러와서 테스트 수행\n\n결과를 submission.csv에 저장","metadata":{}},{"cell_type":"code","source":"class Trainer(object):\n    def __init__(self,\n                 epochs: int = 20,\n                 batch_size: int = 32,\n                 device: str = \"cpu\",\n                 optimizer_fn: Callable = torch.optim.Adam,\n                 optimizer_kwargs: dict = {\"lr\": 1e-3},\n                 ) -> None:\n        self.epochs = epochs\n        self.batch_size = batch_size\n        self.device = device\n        self.optimizer_fn = optimizer_fn\n        self.optimizer_kwargs = optimizer_kwargs\n        self.epoch_test_losses: List[float] = []\n\n\n\nclass GDTrainer(Trainer):\n    def train(self,\n              dataset_train: torch.utils.data.Dataset,\n              dataset_validation: torch.utils.data.Dataset,\n              model: torch.nn.Module,\n              model_dir: str,\n              ):\n\n        train_loader = DataLoader(\n            dataset_train, batch_size=self.batch_size, shuffle=True, drop_last=True, num_workers=4)\n        validation_loader = DataLoader(\n            dataset_validation, batch_size=self.batch_size, drop_last=True, num_workers=4)\n\n        criterion = torch.nn.CrossEntropyLoss()\n        optim = self.optimizer_fn(model.parameters(), **self.optimizer_kwargs)\n\n        best_model = None\n        best_acc = 0\n        for epoch in range(self.epochs):\n            running_loss = 0\n            num_correct = 0.0\n            num_total = 0.0\n            model.train()\n\n            for i, (batch_x, _, batch_y) in enumerate(train_loader):\n                batch_size = batch_x.size(0)\n                num_total += batch_size\n\n                batch_x = batch_x.to(self.device)\n                batch_y = batch_y.to(self.device)\n\n                batch_out = model(batch_x)\n                batch_loss = criterion(batch_out, batch_y)\n\n                _, batch_pred = batch_out.max(dim=1)\n                num_correct += (batch_pred == batch_y).sum(dim=0).item()\n\n\n                running_loss += (batch_loss.item() * batch_size)\n                \n                if i % (train_loader.__len__() // 20) == 0:\n                    cur_loss = batch_loss\n                    LOGGER.info(f\"[{epoch:04d}] {i}/{train_loader.__len__()}: {cur_loss}\")\n\n                optim.zero_grad()\n                batch_loss.backward()\n                optim.step()\n\n            running_loss /= num_total\n            train_accuracy = (num_correct/num_total)*100\n            \n            \n            num_correct = 0.0\n            num_total = 0.0\n            model.eval()\n            for batch_x, _, batch_y in validation_loader:\n\n                batch_size = batch_x.size(0)\n                num_total += batch_size\n                batch_x = batch_x.to(self.device)\n                batch_y = batch_y.to(self.device)\n\n                batch_out = model(batch_x)\n\n                _, batch_pred = batch_out.max(dim=1)\n                num_correct += (batch_pred == batch_y).sum(dim=0).item()\n\n            valid_acc = 100 * (num_correct / num_total)\n\n            if best_model is None or valid_acc > best_acc:\n                best_acc = valid_acc\n                best_model = deepcopy(model.state_dict())\n                save_model(model, model_dir)\n\n            LOGGER.info(\n                f\"[{epoch:04d}]: {running_loss} - train acc: {train_accuracy} - valid_acc: {valid_acc}\")\n\n        model.load_state_dict(best_model)\n        return model\n    \n    def test(self,\n              dataset_test: torch.utils.data.Dataset,\n              model: torch.nn.Module,\n              ):\n        model.eval()\n        test_loader = DataLoader(\n            dataset_test, batch_size=1, drop_last=False)\n        \n        f = open('submission.csv', 'w', newline='')\n        wr = csv.writer(f)\n        wr.writerow(['Id', 'Predicted'])\n\n        for i, (batch_x, _, batch_y) in enumerate(test_loader):\n                batch_x = batch_x.to(self.device)\n                batch_y = batch_y.to(self.device)\n                batch_out = model(batch_x)\n\n                _, batch_pred = batch_out.max(dim=1)\n\n                wr.writerow([i+1, batch_pred[0].item()])\n        f.close()\n","metadata":{"execution":{"iopub.status.busy":"2024-05-26T02:39:41.787185Z","iopub.execute_input":"2024-05-26T02:39:41.787537Z","iopub.status.idle":"2024-05-26T02:39:41.810127Z","shell.execute_reply.started":"2024-05-26T02:39:41.787505Z","shell.execute_reply":"2024-05-26T02:39:41.809114Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## 메인 코드\n","metadata":{}},{"cell_type":"code","source":"LOGGER = logging.getLogger(__name__)\n\n## Trainer\nRAW_NET_CONFIG = {\n    \"nb_samp\": 64600,\n    \"first_conv\": 1024,   # no. of filter coefficients\n    \"in_channels\": 1,  # no. of filters channel in residual blocks\n    \"filts\": [20, [20, 20], [20, 128], [128, 128]],\n    \"blocks\": [2, 4],\n    \"nb_fc_node\": 1024,\n    \"gru_node\": 1024,\n    \"nb_gru_layer\": 3,\n    \"nb_classes\": 2,\n}\n\n\n\ndef init_logger(log_file):\n    LOGGER.setLevel(logging.INFO)\n\n    # create file handler\n    fh = logging.FileHandler(log_file)\n\n    # create console handler\n    ch = logging.StreamHandler()\n\n    # create formatter and add it to the handlers\n    formatter = logging.Formatter(\n        '%(asctime)s - %(levelname)s - %(message)s')\n    fh.setFormatter(formatter)\n    ch.setFormatter(formatter)\n    # add the handlers to the logger\n    LOGGER.addHandler(fh)\n    LOGGER.addHandler(ch)\n\n\ndef save_model(\n        model: torch.nn.Module,\n        model_dir: Union[Path, str],\n) -> None:\n    torch.save(model.state_dict(),\n               f\"{model_dir}/best.pth\")\n\n\ndef train_raw_net(\n        real_training_distribution: Union[Path, str],\n        fake_training_distributions: List[Union[Path, str]],\n        batch_size: int,\n        epochs: int,\n        device: str,\n        test_dir: str,\n        model_dir: Optional[str] = None,\n) -> None:\n\n    LOGGER.info(\"Loading data...\")\n\n    \n\n    # real data\n    real_dataset_train = load_dataset(\n        real_training_distribution,\n        pad=True,\n        train='train',\n        real='real',\n        label=1,\n    )\n\n    real_dataset_validation = load_dataset(\n        real_training_distribution,\n        pad=True,\n        train='validation',\n        real='real',\n        label=1,\n    )\n\n    LOGGER.info(f\"Training {fake_training_distributions}\")\n    # fake data\n    fake_dataset_train = load_dataset(\n            fake_training_distributions,\n            pad=True,\n            train='train',\n            real='fake',\n            label=0,\n        )\n\n    fake_dataset_validation = load_dataset(\n            fake_training_distributions,\n            pad=True,\n            train='validation',\n            real='fake',\n            label=0,\n        )\n    \n    # test\n    dataset_test = load_dataset_test(\n            test_dir,\n            pad=True,\n        )\n\n\n    current_model = RawNet(deepcopy(RAW_NET_CONFIG), device).to(device)\n    data_train = ConcatDataset([real_dataset_train, fake_dataset_train])\n    data_validation = ConcatDataset([real_dataset_validation, fake_dataset_validation])\n\n\n    LOGGER.info(\n            f\"Training rawnet model on {len(data_train)} audio files.\")\n    \n    current_model = GDTrainer(\n            device=device,\n            batch_size=batch_size,\n            epochs=epochs,\n            optimizer_kwargs={\n                \"lr\": 0.0001,\n                \"weight_decay\": 0.0001,\n            }\n        ).train(\n            dataset_train=data_train,\n            dataset_validation=data_validation,\n            model=current_model,\n            model_dir=model_dir,\n        )\n\n\n    LOGGER.info(\"Training is done!\")\n\n    #Test\n    GDTrainer(\n            device=device,\n            batch_size=batch_size,\n            epochs=epochs,\n            optimizer_kwargs={\n                \"lr\": 0.0001,\n                \"weight_decay\": 0.0001,\n            }\n        ).test(\n            dataset_test=dataset_test,\n            model=current_model,\n        )\n    LOGGER.info(\"Testing is done!\")\n\n\ndef main():\n    # fix all seeds\n    set_seed(42)\n\n    init_logger(\"experiments.log\")\n\n    LOGGER.setLevel(logging.DEBUG)\n\n    #device = \"cpu\"\n    device = \"cuda\" \n    \n    \n    db_dir = '/kaggle/input/2024-jbnu-competition-revised'\n    test_dir = '/kaggle/input/2024-jbnu-competition-revised/test'\n    batch_size = 128\n    #batch_size = 32\n    epochs = 10\n\n    model_dir_path = \"trained_models\"\n    model_dir = Path(model_dir_path)\n    if not model_dir.exists():\n        model_dir.mkdir(parents=True)\n\n    train_raw_net(real_training_distribution=db_dir, fake_training_distributions=db_dir, device=device, batch_size=batch_size, epochs=epochs, test_dir=test_dir, model_dir=model_dir)\n\nif __name__ == \"__main__\":\n    main()\n","metadata":{"execution":{"iopub.status.busy":"2024-05-26T02:39:41.811463Z","iopub.execute_input":"2024-05-26T02:39:41.811824Z"},"trusted":true},"execution_count":null,"outputs":[]}]}