pytorch-trigger-word.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8e760e90",
   "metadata": {},
   "source": [
    "# import "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "id": "f9ff5ad6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch \n",
    "from torch import nn \n",
    "import os \n",
    "import matplotlib.pyplot as plt \n",
    "from torch.utils.data import Dataset , DataLoader\n",
    "import torchaudio\n",
    "import torchaudio.functional as F\n",
    "import torchaudio.transforms as T\n",
    "from torchaudio.utils import download_asset\n",
    "from IPython.display import Audio\n",
    "from tqdm.auto import tqdm\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "66e75fd6",
   "metadata": {},
   "source": [
    "# custom dataset \n",
    "### we preprocess the aduio file through some steps \n",
    "\n",
    "- load the audio file\n",
    "- pick the first audio channel \n",
    "- resample the wav \n",
    "- zero padding to unify the wav legnth \n",
    "- transform the wav into Spectrogram \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "id": "d78e569f",
   "metadata": {},
   "outputs": [],
   "source": [
    "class custom_ds(Dataset) : \n",
    "    def __init__(self,data_folder,label) : \n",
    "        super().__init__() \n",
    "        self.data_folder = data_folder \n",
    "        self.label = label \n",
    "        self.files = os.listdir(os.path.join(os.getcwd() , 'data' , 'recorded-data' , data_folder)) \n",
    "    def __len__(self) : \n",
    "        return len(self.files)\n",
    "    def __getitem__(self  , idx) : \n",
    "        wav, sample_rate = torchaudio.load(os.path.join(os.getcwd() , 'data' , 'recorded-data' , self.data_folder , self.files[idx] ) )\n",
    "        wav = wav[0 , :]\n",
    "        wav = F.resample(wav, sample_rate, 16000) \n",
    "        padding = 48000 - wav.shape[0] \n",
    "        zeros = torch.zeros([padding]) \n",
    "        wav = torch.cat([zeros , wav] , axis = 0 )\n",
    "\n",
    "        transform = torchaudio.transforms.Spectrogram(n_fft=320 )\n",
    "        spectrogram = transform(wav)\n",
    "        spectrogram = spectrogram.unsqueeze(0)\n",
    "        \n",
    "        return  spectrogram, self.label"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "id": "411a6747",
   "metadata": {},
   "outputs": [],
   "source": [
    "positive_ds = custom_ds('positive' ,1.)\n",
    "negative_ds = custom_ds('negative'  ,0.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "id": "97bdaeb3",
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset = torch.utils.data.ConcatDataset([positive_ds  ,negative_ds])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "id": "496803dc",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_loader = DataLoader(dataset, batch_size=8,\n",
    "                        shuffle=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "id": "98e19d27",
   "metadata": {},
   "outputs": [],
   "source": [
    "train_set, val_set = torch.utils.data.random_split(data_loader, [ 22,6 ])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "54061863",
   "metadata": {},
   "source": [
    "# the model \n",
    "### the model is basic down sample model \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "id": "63960eae",
   "metadata": {},
   "outputs": [],
   "source": [
    "class model(nn.Module) : \n",
    "    def __init__(self) : \n",
    "        super().__init__()\n",
    "        self.S = nn.Sequential(\n",
    "            nn.Conv2d(1, 16 , (3,3)) , \n",
    "            nn.ReLU() , \n",
    "            nn.MaxPool2d((2,2)) , \n",
    "            nn.Conv2d(16 , 16 , (3,3)) , \n",
    "            nn.ReLU() , \n",
    "            nn.Flatten() , \n",
    "            nn.Linear(181104 , 128 ), \n",
    "            nn.Dropout(.2),\n",
    "            nn.Linear(128 ,1) , \n",
    "            nn.Sigmoid()\n",
    "            \n",
    "        )\n",
    "    def forward(self , inputs ) : \n",
    "        return self.S(inputs ) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "id": "f01af492",
   "metadata": {},
   "outputs": [],
   "source": [
    "model = model()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ec98ad0b",
   "metadata": {},
   "source": [
    "# train step "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "id": "9f169127",
   "metadata": {},
   "outputs": [],
   "source": [
    "def train_step(model , dataloader , loss_fn , optimizer ) : \n",
    "    model.train()\n",
    "    train_loss = 0 \n",
    "    for batch_images , batch_labels in dataloader.dataset : \n",
    "        preds =  model(batch_images).reshape(-1) \n",
    "        loss = loss_fn(preds , batch_labels.type(torch.float)) \n",
    "        train_loss += loss.item()\n",
    "        optimizer.zero_grad()\n",
    "        loss.backward()\n",
    "        optimizer.step() \n",
    "        \n",
    "        \n",
    "    \n",
    "    train_loss = train_loss / len(dataloader)\n",
    "    return train_loss  \n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f1f56a28",
   "metadata": {},
   "source": [
    "# test step "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "id": "bb865ee5",
   "metadata": {},
   "outputs": [],
   "source": [
    "def test_step(model , dataloader , loss_fn) : \n",
    "    model.eval()\n",
    "    test_loss  = 0 \n",
    "    with torch.no_grad() : \n",
    "        for batch_images , batch_labels in dataloader.dataset : \n",
    "            preds = model(batch_images ).reshape(-1)\n",
    "            loss = loss_fn(preds , batch_labels.type(torch.float) )\n",
    "            test_loss += loss.item()\n",
    "            \n",
    "            \n",
    "    test_loss = test_loss / len(dataloader)\n",
    "    return test_loss "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "id": "98369e7a",
   "metadata": {},
   "outputs": [],
   "source": [
    "def train(model: torch.nn.Module, \n",
    "          train_dataloader: torch.utils.data.DataLoader, \n",
    "          test_dataloader: torch.utils.data.DataLoader, \n",
    "          optimizer: torch.optim.Optimizer,\n",
    "          loss_fn: torch.nn.Module = nn.CrossEntropyLoss(),\n",
    "          epochs: int = 5):\n",
    "    \n",
    "    # 2. Create empty results dictionary\n",
    "    results = {\"train_loss\": [],\n",
    "        \"test_loss\": [],\n",
    "    }\n",
    "    \n",
    "    # 3. Loop through training and testing steps for a number of epochs\n",
    "    for epoch in tqdm(range(epochs)):\n",
    "        train_loss = train_step(model=model,\n",
    "                                           dataloader=train_dataloader,\n",
    "                                           loss_fn=loss_fn,\n",
    "                                           optimizer=optimizer)\n",
    "        test_loss = test_step(model=model,\n",
    "            dataloader=test_dataloader,\n",
    "            loss_fn=loss_fn)\n",
    "        \n",
    "        # 4. Print out what's happening\n",
    "        print(\n",
    "            f\"Epoch: {epoch+1} | \"\n",
    "            f\"train_loss: {train_loss:.4f} | \"\n",
    "            f\"test_loss: {test_loss:.4f} | \"\n",
    "\n",
    "        )\n",
    "\n",
    "        # 5. Update results dictionary\n",
    "        results[\"train_loss\"].append(train_loss)\n",
    "        results[\"test_loss\"].append(test_loss)\n",
    "\n",
    "    # 6. Return the filled results at the end of the epochs\n",
    "    return results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "id": "0b64e3dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "optim = torch.optim.Adam(model.parameters() ,  lr = .0001)\n",
    "loss_fn = nn.BCELoss()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "id": "eb2d69dd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "5a1b5c62c0c94bd8803d399dcba6e403",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "  0%|          | 0/5 [00:00<?, ?it/s]"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch: 1 | train_loss: 0.0121 | train_acc: 1.2670 | test_loss: 0.0015 | test_acc: 4.6667\n",
      "Epoch: 2 | train_loss: 0.0006 | train_acc: 1.2727 | test_loss: 0.0006 | test_acc: 4.6667\n",
      "Epoch: 3 | train_loss: 0.0001 | train_acc: 1.2727 | test_loss: 0.0001 | test_acc: 4.6667\n",
      "Epoch: 4 | train_loss: 0.0000 | train_acc: 1.2727 | test_loss: 0.0001 | test_acc: 4.6667\n",
      "Epoch: 5 | train_loss: 0.0000 | train_acc: 1.2727 | test_loss: 0.0001 | test_acc: 4.6667\n"
     ]
    }
   ],
   "source": [
    "model_0_results = train(model=model, \n",
    "                        train_dataloader=train_set,\n",
    "                        test_dataloader=val_set,\n",
    "                        optimizer=optim,\n",
    "                        loss_fn=loss_fn, \n",
    "                        epochs=5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "id": "5d636bd4",
   "metadata": {},
   "outputs": [],
   "source": [
    "torch.save(model.state_dict(), 'torch.pth')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}